Keyword Cannibalization Audit Claude Skill

#!/usr/bin/env python3
"""Crawl a site and capture each page's keyword-target signals.

Usage:
    python3 extract_targets.py https://example.com [--max-pages 500] [--output target_inventory.json]
    python3 extract_targets.py urls.txt --url-list [--output target_inventory.json]

For each page it records status, title, first H1, meta description, noindex, and
the page's top body keywords by frequency (stopword-filtered). The auditor uses
the title + H1 + top keywords to find multiple pages targeting the same query
(keyword cannibalization).

Uses only the Python standard library.
"""

import argparse
import json
import re
import sys
import urllib.error
import urllib.parse
import urllib.request
import xml.etree.ElementTree as ET
from collections import Counter, deque
from html.parser import HTMLParser

USER_AGENT = "CannibalizationAuditBot/1.0 (+claude-skill)"
TIMEOUT = 15
TOP_KEYWORDS = 15
SKIP_TEXT_TAGS = {"script", "style", "noscript", "template", "svg"}

STOPWORDS = {
    "a", "an", "and", "are", "as", "at", "be", "but", "by", "for", "from",
    "has", "have", "how", "in", "is", "it", "its", "of", "on", "or", "our",
    "that", "the", "this", "to", "was", "we", "what", "when", "where", "which",
    "who", "will", "with", "you", "your", "can", "do", "does", "get", "more",
    "all", "about", "into", "than", "then", "they", "their", "if", "so", "no",
}


class TargetParser(HTMLParser):
    def __init__(self, base_url):
        super().__init__(convert_charrefs=True)
        self.base_url = base_url
        self.title = None
        self.h1 = None
        self.description = None
        self.meta_robots = None
        self.tokens = []
        self.links = []
        self._in_title = False
        self._title = []
        self._in_h1 = False
        self._h1 = []
        self._h1_done = False
        self._skip = 0

    def handle_starttag(self, tag, attrs):
        a = dict(attrs)
        if tag in SKIP_TEXT_TAGS:
            self._skip += 1
        if tag == "title":
            self._in_title, self._title = True, []
        elif tag == "h1" and not self._h1_done:
            self._in_h1, self._h1 = True, []
        elif tag == "meta":
            name = (a.get("name") or "").lower()
            if name == "description":
                self.description = (a.get("content") or "").strip() or None
            elif name == "robots":
                self.meta_robots = a.get("content")
        elif tag == "a" and a.get("href"):
            href = a["href"].strip()
            if not href.startswith(("mailto:", "tel:", "javascript:", "#")):
                self.links.append(
                    urllib.parse.urldefrag(urllib.parse.urljoin(self.base_url, href))[0])

    def handle_endtag(self, tag):
        if tag in SKIP_TEXT_TAGS and self._skip > 0:
            self._skip -= 1
        if tag == "title":
            self._in_title = False
            self.title = " ".join("".join(self._title).split()) or None
        elif tag == "h1" and self._in_h1:
            self._in_h1, self._h1_done = False, True
            self.h1 = " ".join("".join(self._h1).split()) or None

    def handle_data(self, data):
        if self._in_title:
            self._title.append(data)
        if self._in_h1:
            self._h1.append(data)
        if self._skip == 0:
            self.tokens.extend(
                w for w in re.findall(r"[a-z][a-z0-9']+", data.lower())
                if w not in STOPWORDS and len(w) > 2)


def fetch(url):
    req = urllib.request.Request(url, headers={"User-Agent": USER_AGENT})
    try:
        with urllib.request.urlopen(req, timeout=TIMEOUT) as resp:
            ctype = resp.headers.get("Content-Type", "")
            body = resp.read(2_000_000).decode("utf-8", errors="replace") \
                if "html" in ctype else None
            return resp.status, resp.geturl(), body
    except urllib.error.HTTPError as e:
        return e.code, url, None
    except Exception:
        return 0, url, None


def norm_host(url):
    h = urllib.parse.urlparse(url).netloc.lower()
    return h[4:] if h.startswith("www.") else h


def pages_from_sitemap(site):
    status, _f, body = fetch(urllib.parse.urljoin(site, "/sitemap.xml"))
    if status != 200 or not body:
        return []
    try:
        root = ET.fromstring(body)
        return [loc.text.strip() for loc in
                root.iter("{http://www.sitemaps.org/schemas/sitemap/0.9}loc") if loc.text]
    except ET.ParseError:
        return []


def analyze(url, html):
    p = TargetParser(url)
    try:
        p.feed(html)
    except Exception:
        pass
    top = [w for w, _c in Counter(p.tokens).most_common(TOP_KEYWORDS)]
    info = {"title": p.title, "h1": p.h1, "description": p.description,
            "noindex": "noindex" in (p.meta_robots or "").lower(),
            "word_count": len(p.tokens), "top_keywords": top}
    return info, p.links


def extract_from_cache(cache_path):
    """Pure: turn a shared page cache (from fetch_pages.py) into a target inventory.

    No network — analyzes the cached HTML for exactly the signals this audit needs.
    """
    with open(cache_path, encoding="utf-8") as f:
        cache = json.load(f)
    pages = {}
    for url, page in cache.get("pages", {}).items():
        html = page.get("html")
        status = page.get("status", 0)
        final_url = page.get("final_url") or url
        if not html:
            pages[url] = {"status": status, "final_url": final_url, "title": None,
                          "h1": None, "description": None, "noindex": False,
                          "word_count": 0, "top_keywords": []}
            continue
        info, _links = analyze(final_url, html)
        info["status"] = status
        info["final_url"] = final_url
        pages[url] = info
    return cache.get("site", cache_path), pages


def collect(urls, max_pages, crawl_host=None):
    pages, queue, seen = {}, deque(urls), set(urls)
    while queue and len(pages) < max_pages:
        url = queue.popleft()
        status, final_url, html = fetch(url)
        print(f"[{status}] {url}", file=sys.stderr)
        if html is None:
            pages[url] = {"status": status, "final_url": final_url, "title": None,
                          "h1": None, "description": None, "noindex": False,
                          "word_count": 0, "top_keywords": []}
            continue
        info, links = analyze(final_url, html)
        info["status"] = status
        info["final_url"] = final_url
        pages[url] = info
        if crawl_host:
            for link in links:
                if norm_host(link) == crawl_host and link not in seen:
                    seen.add(link)
                    queue.append(link)
    return pages


def main():
    ap = argparse.ArgumentParser(description="Capture keyword-target signals for a site.")
    ap.add_argument("source", nargs="?")
    ap.add_argument("--from-cache",
                    help="Extract from a shared page cache produced by fetch_pages.py (no network)")
    ap.add_argument("--max-pages", type=int, default=500)
    ap.add_argument("--output", default="target_inventory.json")
    ap.add_argument("--url-list", action="store_true")
    args = ap.parse_args()

    if args.from_cache:
        site, pages = extract_from_cache(args.from_cache)
    elif not args.source:
        ap.error("provide a source, or --from-cache page_cache.json")
    elif args.url_list:
        with open(args.source, encoding="utf-8") as f:
            urls = [l.strip() for l in f if l.strip() and not l.startswith("#")]
        site, pages = args.source, collect(urls[: args.max_pages], args.max_pages)
    else:
        start = args.source.rstrip("/")
        seeds = pages_from_sitemap(start) or [start]
        site, pages = args.source, collect(seeds[: args.max_pages], args.max_pages,
                                           crawl_host=norm_host(start))

    with open(args.output, "w", encoding="utf-8") as f:
        json.dump({"site": site, "pages": pages}, f, indent=2, ensure_ascii=False)
    print(f"Wrote {len(pages)} pages to {args.output}", file=sys.stderr)


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""Audit a target inventory (from extract_targets.py) for keyword cannibalization.

Usage:
    python3 audit_cannibalization.py target_inventory.json [--output audit_report.json]
        [--brand-suffix "| Acme Co"] [--overlap 0.6]

Checks (see references/audit-checks.md):
- duplicate_title_target (high): 2+ indexable pages with the same normalized title
  core — they target the same query head-on
- overlapping_targets (medium): clusters of pages whose title+H1 keyword sets
  overlap above the threshold (Jaccard) — competing for the same topic
- shared_primary_keyword (low): pages sharing the same significant title bigram

Clustering uses union-find over pairwise Jaccard of each page's keyword signature
(title + H1 + top body keywords). Dependency-free.

Uses only the Python standard library.
"""

import argparse
import json
import re
import sys
from collections import Counter, defaultdict

STOPWORDS = {
    "a", "an", "and", "are", "as", "at", "be", "by", "for", "from", "in", "is",
    "it", "of", "on", "or", "the", "to", "with", "your", "you", "our", "how",
    "what", "best", "guide", "top",
}


def words(text):
    if not text:
        return []
    return [w for w in re.findall(r"[a-z0-9']+", text.lower())
            if w not in STOPWORDS and len(w) > 2]


def strip_brand(text, brand):
    if not text:
        return ""
    t = text.strip()
    if brand and t.lower().endswith(brand.lower()):
        t = t[: len(t) - len(brand)].rstrip(" |-–:·")
    return t


def jaccard(a, b):
    if not a or not b:
        return 0.0
    return len(a & b) / len(a | b)


class UnionFind:
    def __init__(self):
        self.parent = {}

    def find(self, x):
        self.parent.setdefault(x, x)
        while self.parent[x] != x:
            self.parent[x] = self.parent[self.parent[x]]
            x = self.parent[x]
        return x

    def union(self, a, b):
        self.parent[self.find(a)] = self.find(b)


def main():
    ap = argparse.ArgumentParser(description="Run keyword-cannibalization audit checks.")
    ap.add_argument("inventory")
    ap.add_argument("--output", default="audit_report.json")
    ap.add_argument("--brand-suffix", default="")
    ap.add_argument("--overlap", type=float, default=0.6,
                    help="Jaccard overlap of keyword signatures to flag as competing")
    args = ap.parse_args()

    with open(args.inventory, encoding="utf-8") as f:
        data = json.load(f)
    pages = data.get("pages", {})

    findings = {k: [] for k in (
        "duplicate_title_target", "overlapping_targets", "shared_primary_keyword")}

    indexable = {u: p for u, p in pages.items()
                 if p.get("status", 0) == 200 and not p.get("noindex")}

    # 1. Exact duplicate title core.
    title_groups = defaultdict(list)
    for url, p in indexable.items():
        core = strip_brand(p.get("title"), args.brand_suffix)
        key = " ".join(words(core))
        if key:
            title_groups[key].append(url)
    for key, urls in title_groups.items():
        if len(urls) > 1:
            findings["duplicate_title_target"].append(
                {"target": key, "pages": sorted(urls)})

    # 2. Overlapping keyword signatures -> clusters.
    sigs = {}
    for url, p in indexable.items():
        sig = set(words(strip_brand(p.get("title"), args.brand_suffix)))
        sig |= set(words(p.get("h1")))
        sig |= set(p.get("top_keywords", [])[:10])
        if sig:
            sigs[url] = sig

    uf = UnionFind()
    urls = list(sigs)
    edges = []
    for i in range(len(urls)):
        for j in range(i + 1, len(urls)):
            sim = jaccard(sigs[urls[i]], sigs[urls[j]])
            if sim >= args.overlap:
                uf.union(urls[i], urls[j])
                edges.append((urls[i], urls[j], round(sim, 2)))

    clusters = defaultdict(list)
    for url in urls:
        if any(url in (e[0], e[1]) for e in edges):
            clusters[uf.find(url)].append(url)
    for members in clusters.values():
        if len(members) > 1:
            shared = set.intersection(*(sigs[m] for m in members))
            findings["overlapping_targets"].append(
                {"pages": sorted(members),
                 "shared_keywords": sorted(shared)[:10],
                 "size": len(members)})

    # 3. Shared significant title bigram.
    bigram_groups = defaultdict(list)
    for url, p in indexable.items():
        tw = words(strip_brand(p.get("title"), args.brand_suffix))
        for i in range(len(tw) - 1):
            bigram_groups[(tw[i], tw[i + 1])].append(url)
    for bigram, urls_b in bigram_groups.items():
        uniq = sorted(set(urls_b))
        if len(uniq) > 1:
            findings["shared_primary_keyword"].append(
                {"keyword": " ".join(bigram), "pages": uniq})

    severity = {
        "duplicate_title_target": "high",
        "overlapping_targets": "medium",
        "shared_primary_keyword": "low",
    }
    summary = {
        "pages_analyzed": len(indexable),
        "issues_by_check": {k: len(v) for k, v in findings.items()},
        "issues_by_severity": {
            level: sum(len(findings[k]) for k, s in severity.items() if s == level)
            for level in ("high", "medium", "low", "info")},
    }
    report = {"summary": summary, "severity": severity, "findings": findings}
    with open(args.output, "w", encoding="utf-8") as f:
        json.dump(report, f, indent=2, ensure_ascii=False)
    print(json.dumps(summary, indent=2))
    print(f"Full report written to {args.output}", file=sys.stderr)


if __name__ == "__main__":
    main()