Internal-Link Opportunity Finder: the pages that mention a term but never link to it

"""
Internal-Link Opportunity Finder
The SEM Dispatch Vault, semdispatch.com/vault

Cross-references a Search Console export against a crawl of the site's own
pages to find pages that mention a ranking query but do not yet link to the
page that ranks for it. Produces a CSV. Never edits a page.
"""

import re
import requests
import pandas as pd
from bs4 import BeautifulSoup
from urllib.parse import urljoin, urlparse
from defusedxml import ElementTree  # parses the sitemap safely; do not swap for xml.etree

# --- Config -------------------------------------------------------------

SITEMAP_URL = "https://example.com/sitemap.xml"
GSC_EXPORT_CSV = "gsc_export.csv"          # columns: page, query, impressions
BASE_DOMAIN = "example.com"
MIN_IMPRESSIONS = 100
MAX_SUGGESTIONS_PER_TARGET = 10
OUTPUT_CSV = "link_opportunities.csv"

# --- Crawl ----------------------------------------------------------------

def get_sitemap_urls(sitemap_url):
    resp = requests.get(sitemap_url, timeout=20)
    root = ElementTree.fromstring(resp.content)
    ns = {"sm": "http://www.sitemaps.org/schemas/sitemap/0.9"}
    return [loc.text for loc in root.findall(".//sm:loc", ns)]


def fetch_page(url):
    resp = requests.get(url, timeout=20, headers={"User-Agent": "LinkFinder/1.0"})
    soup = BeautifulSoup(resp.text, "html.parser")
    body_text = soup.get_text(separator=" ").lower()
    linked_targets = set()
    for a in soup.find_all("a", href=True):
        href = urljoin(url, a["href"])
        if urlparse(href).netloc.endswith(BASE_DOMAIN):
            linked_targets.add(href.split("#")[0].rstrip("/"))
    return body_text, linked_targets


def crawl_site(urls):
    pages = {}
    for url in urls:
        try:
            text, links = fetch_page(url)
            pages[url.rstrip("/")] = {"text": text, "links": links}
        except requests.RequestException:
            continue
    return pages

# --- Matching ---------------------------------------------------------------

def top_queries_by_page(gsc_csv, min_impressions):
    df = pd.read_csv(gsc_csv)
    df = df[df["impressions"] >= min_impressions]
    grouped = df.sort_values("impressions", ascending=False).groupby("page")
    return {page: list(rows["query"]) for page, rows in grouped}


def find_opportunities(pages, queries_by_page, gsc_impressions):
    opportunities = []
    for target_page, queries in queries_by_page.items():
        target_key = target_page.rstrip("/")
        suggestions_for_target = 0
        for query in queries:
            if suggestions_for_target >= MAX_SUGGESTIONS_PER_TARGET:
                break
            phrase = query.lower().strip()
            if not phrase:
                continue
            for source_page, data in pages.items():
                if source_page == target_key:
                    continue
                if target_key in data["links"]:
                    continue
                if re.search(r"\b" + re.escape(phrase) + r"\b", data["text"]):
                    opportunities.append({
                        "source_page": source_page,
                        "target_page": target_key,
                        "anchor_phrase": query,
                        "target_query_impressions": gsc_impressions.get((target_key, query), 0),
                    })
                    suggestions_for_target += 1
                    break
    return opportunities

# --- Entry point --------------------------------------------------------------

def main():
    urls = get_sitemap_urls(SITEMAP_URL)
    pages = crawl_site(urls)

    df = pd.read_csv(GSC_EXPORT_CSV)
    df = df[df["impressions"] >= MIN_IMPRESSIONS]
    impressions_lookup = {
        (row["page"].rstrip("/"), row["query"]): row["impressions"]
        for _, row in df.iterrows()
    }
    queries_by_page = top_queries_by_page(GSC_EXPORT_CSV, MIN_IMPRESSIONS)

    opportunities = find_opportunities(pages, queries_by_page, impressions_lookup)
    result = pd.DataFrame(opportunities)
    if not result.empty:
        result = result.sort_values("target_query_impressions", ascending=False)
    result.to_csv(OUTPUT_CSV, index=False)
    print(f"Wrote {len(result)} link opportunities to {OUTPUT_CSV}")


if __name__ == "__main__":
    main()