Content Quality Audit Claude Skill

#!/usr/bin/env python3
"""Crawl a site and capture content-quality signals (length, ratio, dedupe sigs).

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

For each page it records status, title, first H1, visible word count, the
text-to-HTML ratio, whether the page declares noindex, an exact content hash, and
a 32-value MinHash signature of its word shingles so the auditor can find exact
and near-duplicate bodies cheaply (no pairwise text comparison, no dependencies).

Uses only the Python standard library.
"""

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

USER_AGENT = "ContentQualityAuditBot/1.0 (+claude-skill)"
TIMEOUT = 15
NUM_HASHES = 32
SHINGLE = 4
MASK = (1 << 32) - 1
SEEDS = [0x9E3779B1 * (i + 1) & MASK for i in range(NUM_HASHES)]
SKIP_TEXT_TAGS = {"script", "style", "noscript", "template", "svg"}


class ContentParser(HTMLParser):
    def __init__(self, base_url):
        super().__init__(convert_charrefs=True)
        self.base_url = base_url
        self.title = None
        self.h1 = None
        self.text_chars = 0
        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" and (a.get("name") or "").lower() == "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])

    meta_robots = None

    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:
            stripped = data.strip()
            if stripped:
                self.text_chars += len(stripped)
                self.tokens.extend(re.findall(r"[a-z0-9']+", stripped.lower()))


def minhash(tokens):
    if len(tokens) < SHINGLE:
        shingles = {" ".join(tokens)} if tokens else set()
    else:
        shingles = {" ".join(tokens[i:i + SHINGLE])
                    for i in range(len(tokens) - SHINGLE + 1)}
    if not shingles:
        return None
    hashed = [int(hashlib.md5(s.encode()).hexdigest()[:8], 16) for s in shingles]
    return [min((h ^ seed) & MASK for h in hashed) for seed in SEEDS]


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 = ContentParser(url)
    try:
        p.feed(html)
    except Exception:
        pass
    word_count = len(p.tokens)
    ratio = round(p.text_chars / len(html), 4) if html else 0
    norm_text = " ".join(p.tokens)
    info = {
        "title": p.title, "h1": p.h1, "word_count": word_count,
        "text_to_html_ratio": ratio,
        "noindex": "noindex" in (p.meta_robots or "").lower(),
        "content_hash": hashlib.md5(norm_text.encode()).hexdigest() if norm_text else None,
        "minhash": minhash(p.tokens),
    }
    return info, p.links


def extract_from_cache(cache_path):
    """Pure: turn a shared page cache (from fetch_pages.py) into a content 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, "word_count": 0, "text_to_html_ratio": 0,
                          "noindex": False, "content_hash": None, "minhash": None}
            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, "word_count": 0, "text_to_html_ratio": 0,
                          "noindex": False, "content_hash": None, "minhash": None}
            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 content-quality 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="content_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 content inventory (from extract_content.py) and emit findings.

Usage:
    python3 audit_content.py content_inventory.json [--output audit_report.json]
        [--thin-words 300] [--similarity 0.85]

Checks (see references/audit-checks.md):
- thin_content (high): indexable page well under the word-count threshold
- duplicate_content (high): two or more indexable pages with identical body text
- near_duplicate_content (medium): pages whose MinHash similarity exceeds the
  threshold (templated/boilerplate pages with little unique text)
- low_text_to_html_ratio (medium): page is mostly markup, little text
- missing_h1 (low): indexable page with no H1 to frame the content

Near-duplicate detection uses the 32-value MinHash signatures from the collector,
so it is O(n^2) over small fixed-size signatures — fast and dependency-free.

Uses only the Python standard library.
"""

import argparse
import json
import sys
from collections import defaultdict


def mh_similarity(a, b):
    if not a or not b or len(a) != len(b):
        return 0.0
    return sum(1 for x, y in zip(a, b) if x == y) / len(a)


def main():
    ap = argparse.ArgumentParser(description="Run content-quality audit checks.")
    ap.add_argument("inventory")
    ap.add_argument("--output", default="audit_report.json")
    ap.add_argument("--thin-words", type=int, default=300)
    ap.add_argument("--similarity", type=float, default=0.85,
                    help="MinHash similarity above which pages are near-duplicates")
    ap.add_argument("--min-ratio", type=float, default=0.1,
                    help="text-to-HTML ratio below which a page is mostly markup")
    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 (
        "thin_content", "duplicate_content", "near_duplicate_content",
        "low_text_to_html_ratio", "missing_h1")}

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

    # Exact duplicates by content hash.
    hash_groups = defaultdict(list)
    for url, p in indexable.items():
        if p.get("content_hash"):
            hash_groups[p["content_hash"]].append(url)
    exact_dupe_urls = set()
    for h, urls in hash_groups.items():
        if len(urls) > 1:
            findings["duplicate_content"].append(
                {"pages": sorted(urls), "word_count": indexable[urls[0]].get("word_count")})
            exact_dupe_urls.update(urls)

    # Per-page checks.
    for url, p in indexable.items():
        wc = p.get("word_count", 0)
        if wc < args.thin_words:
            findings["thin_content"].append(
                {"page": url, "word_count": wc, "title": p.get("title")})
        if 0 < p.get("text_to_html_ratio", 0) < args.min_ratio:
            findings["low_text_to_html_ratio"].append(
                {"page": url, "ratio": p.get("text_to_html_ratio"), "word_count": wc})
        if not p.get("h1"):
            findings["missing_h1"].append({"page": url, "title": p.get("title")})

    # Near-duplicates via MinHash (skip pages already flagged as exact dupes).
    items = [(u, p["minhash"]) for u, p in indexable.items()
             if p.get("minhash") and u not in exact_dupe_urls]
    seen_pairs = set()
    for i in range(len(items)):
        ui, si = items[i]
        for j in range(i + 1, len(items)):
            uj, sj = items[j]
            sim = mh_similarity(si, sj)
            if sim >= args.similarity:
                key = tuple(sorted((ui, uj)))
                if key not in seen_pairs:
                    seen_pairs.add(key)
                    findings["near_duplicate_content"].append(
                        {"pages": list(key), "similarity": round(sim, 2)})

    severity = {
        "thin_content": "high", "duplicate_content": "high",
        "near_duplicate_content": "medium", "low_text_to_html_ratio": "medium",
        "missing_h1": "low",
    }
    word_counts = [p.get("word_count", 0) for p in indexable.values()]
    summary = {
        "pages_analyzed": len(indexable),
        "median_word_count": sorted(word_counts)[len(word_counts) // 2] if word_counts else 0,
        "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()