Pagination Audit Claude Skill

#!/usr/bin/env python3
"""Crawl a site and inventory pagination signals (rel=next/prev, canonical, page params).

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

For each page it records status, final_url, canonical, meta robots, the
`rel="next"` / `rel="prev"` link targets (from <link> tags and the HTTP Link
header), and the detected page number from the URL (?page=N, /page/N, etc.).

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 deque
from html.parser import HTMLParser

USER_AGENT = "PaginationAuditBot/1.0 (+claude-skill)"
TIMEOUT = 15

LINK_HEADER_RE = re.compile(r'<([^>]+)>\s*;\s*rel\s*=\s*"?(next|prev)"?', re.I)
# Page-number patterns: ?page=2, &p=2, /page/2, /page/2/, .../2 in a paged path.
PAGE_PATTERNS = [
    re.compile(r'[?&](?:page|p|pg|paged)=(\d+)', re.I),
    re.compile(r'/page/(\d+)/?', re.I),
    re.compile(r'/p/(\d+)/?', re.I),
]


def page_number(url):
    """Return the page number encoded in the URL, or None if it isn't paginated."""
    for pat in PAGE_PATTERNS:
        m = pat.search(url)
        if m:
            return int(m.group(1))
    return None


class PaginationParser(HTMLParser):
    def __init__(self, base_url):
        super().__init__(convert_charrefs=True)
        self.base_url = base_url
        self.rel_next = None
        self.rel_prev = None
        self.canonical = None
        self.meta_robots = None
        self.links = []

    def handle_starttag(self, tag, attrs):
        a = dict(attrs)
        if tag == "link":
            rel = (a.get("rel") or "").lower()
            href = a.get("href")
            if href:
                resolved = urllib.parse.urljoin(self.base_url, href.strip())
                if "next" in rel:
                    self.rel_next = resolved
                elif "prev" in rel:
                    self.rel_prev = resolved
                elif rel == "canonical":
                    self.canonical = resolved
        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])


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 or "xml" in ctype) else None
            return resp.status, resp.geturl(), body, dict(resp.headers)
    except urllib.error.HTTPError as e:
        return e.code, url, None, dict(e.headers or {})
    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, _h = 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 parse_page(url, html, headers):
    p = PaginationParser(url)
    try:
        p.feed(html)
    except Exception:
        pass
    for href, rel in LINK_HEADER_RE.findall(headers.get("Link", "") or ""):
        resolved = urllib.parse.urljoin(url, href.strip())
        if rel.lower() == "next" and not p.rel_next:
            p.rel_next = resolved
        elif rel.lower() == "prev" and not p.rel_prev:
            p.rel_prev = resolved
    return {"canonical": p.canonical, "meta_robots": p.meta_robots,
            "rel_next": p.rel_next, "rel_prev": p.rel_prev,
            "page_number": page_number(url)}, p.links


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

    Reads the cached HTML *and response headers* (the Link: rel=next/prev headers
    this audit needs). No network.
    """
    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, "canonical": None,
                          "meta_robots": None, "rel_next": None, "rel_prev": None,
                          "page_number": page_number(url)}
            continue
        info, _links = parse_page(final_url, html, page.get("headers") or {})
        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, headers = fetch(url)
        print(f"[{status}] {url}", file=sys.stderr)
        if html is None:
            pages[url] = {"status": status, "final_url": final_url, "canonical": None,
                          "meta_robots": None, "rel_next": None, "rel_prev": None,
                          "page_number": page_number(url)}
            continue
        info, links = parse_page(final_url, html, headers)
        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="Build a pagination inventory 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="pagination_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))

    paginated = sum(1 for p in pages.values()
                    if p.get("page_number") or p.get("rel_next") or p.get("rel_prev"))
    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 ({paginated} pagination-related) to {args.output}",
          file=sys.stderr)


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""Audit a pagination inventory (from extract_pagination.py) and emit findings.

Usage:
    python3 audit_pagination.py pagination_inventory.json [--output audit_report.json]

Checks (see references/audit-checks.md):
- paginated_canonical_to_page1 (high): page 2+ canonicalizes to page 1, hiding its
  content/links from the index (Google says each page should self-canonicalize)
- paginated_noindex (medium): a component (page 2+) page is noindexed
- paginated_missing_self_canonical (medium): page 2+ has no canonical at all
- rel_next_prev_broken (medium): a rel=next/prev target is non-200/redirects
- rel_next_prev_inconsistent (low): A's next is B but B's prev isn't A
- page1_has_page_param (low): the first page carries a redundant ?page=1
- rel_next_prev_present (info): Google no longer uses rel=next/prev for indexing

Uses only the Python standard library.
"""

import argparse
import json
import re
import sys
import urllib.parse

PAGE_PARAM_RE = re.compile(r'([?&])(?:page|p|pg|paged)=\d+', re.I)
PAGE_PATH_RE = re.compile(r'/(?:page|p)/\d+/?', re.I)


def strip_page(url):
    """Return the URL with its page-number component removed (i.e. the page-1 URL)."""
    if not url:
        return ""
    u = PAGE_PATH_RE.sub("/", url)
    u = PAGE_PARAM_RE.sub(lambda m: m.group(1), u)
    u = re.sub(r'[?&]$', '', u)
    return urllib.parse.urldefrag(u)[0].rstrip("/")


def norm(url):
    return urllib.parse.urldefrag(url)[0].rstrip("/") if url else ""


def main():
    ap = argparse.ArgumentParser(description="Run pagination audit checks.")
    ap.add_argument("inventory")
    ap.add_argument("--output", default="audit_report.json")
    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 (
        "paginated_canonical_to_page1", "paginated_noindex",
        "paginated_missing_self_canonical", "rel_next_prev_broken",
        "rel_next_prev_inconsistent", "page1_has_page_param",
        "rel_next_prev_present")}

    # index by normalized final URL for reciprocity + target status lookups
    by_url = {}
    for u, p in pages.items():
        by_url[norm(p.get("final_url", u))] = (u, p)

    for url, page in pages.items():
        if page.get("status", 0) != 200:
            continue
        pn = page.get("page_number")
        canonical = page.get("canonical")
        is_noindex = "noindex" in (page.get("meta_robots") or "").lower()
        rel_next, rel_prev = page.get("rel_next"), page.get("rel_prev")

        if rel_next or rel_prev:
            findings["rel_next_prev_present"].append(
                {"page": url, "next": rel_next, "prev": rel_prev})

        # Component pages (page 2+).
        if pn and pn >= 2:
            if is_noindex:
                findings["paginated_noindex"].append({"page": url, "page_number": pn})
            if not canonical:
                findings["paginated_missing_self_canonical"].append(
                    {"page": url, "page_number": pn})
            elif norm(canonical) == strip_page(page.get("final_url", url)) \
                    and norm(canonical) != norm(page.get("final_url", url)):
                findings["paginated_canonical_to_page1"].append(
                    {"page": url, "page_number": pn, "canonical": canonical})

        if pn == 1:
            findings["page1_has_page_param"].append({"page": url})

        # rel=next/prev target health + reciprocity.
        for rel, target in (("next", rel_next), ("prev", rel_prev)):
            if not target:
                continue
            t = by_url.get(norm(target))
            if t:
                _tu, tp = t
                if tp.get("status", 0) != 200:
                    findings["rel_next_prev_broken"].append(
                        {"page": url, "rel": rel, "target": target,
                         "status": tp.get("status")})
                back = tp.get("rel_prev") if rel == "next" else tp.get("rel_next")
                if back is not None and norm(back) != norm(page.get("final_url", url)):
                    findings["rel_next_prev_inconsistent"].append(
                        {"page": url, "rel": rel, "target": target,
                         "target_points_back_to": back})

    severity = {
        "paginated_canonical_to_page1": "high",
        "paginated_noindex": "medium", "paginated_missing_self_canonical": "medium",
        "rel_next_prev_broken": "medium",
        "rel_next_prev_inconsistent": "low", "page1_has_page_param": "low",
        "rel_next_prev_present": "info",
    }
    summary = {
        "pages_analyzed": sum(1 for p in pages.values() if p.get("status", 0) == 200),
        "paginated_pages": sum(1 for p in pages.values() if (p.get("page_number") or 0) >= 2),
        "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()