ambrosiana-document-intake/link_discovery.py

#!/usr/bin/env python3
# Created by John A. Hoeven with the ethical assistance of Claude AI
# ---------------------------------------------------------------------------
# link_discovery.py
# ~/projects/ambrosiana-document-intake/link_discovery.py
# Version: v0.2.0 | Status: BETA
# ---------------------------------------------------------------------------
# Purpose:  URL discovery from an index page, for the "add index page,
#           review discovered sub-pages" intake flow. Fetches pages to find
#           URLs only — never writes a ticket itself; that stays a separate,
#           human-approved step in the App.
# Target:   Any Ambrosiana device.
# Entry:    import as a module.
# Depends:  requests, beautifulsoup4 (see requirements.txt)
# ---------------------------------------------------------------------------
"""
Security note (deliberate, not an oversight): this module never extracts
free-text content from a fetched page (titles, headings, link text) into
anything that ends up in a ticket. A ticket is later read by Claude Code
as a pipeline instruction, so pulling untrusted third-party prose into a
ticket field would be a real prompt-injection vector. Only URLs — which
this tool itself discovered under a scope the human chose — are ever
returned. Display titles shown in the review UI are derived from the URL
path alone, never from fetched page content.

Two discovery strategies, tried in this order:

1. Sitemap (preferred) — sitemap.xml is pure structured data (<loc>,
   <lastmod>), nothing to inject into. Checked at the domain root and via
   robots.txt's Sitemap: directive.
2. HTML link crawl (fallback) — same-domain breadth-first crawl for sites
   without a sitemap. Extracts hrefs only, never page titles or body text.
"""

from __future__ import annotations

import xml.etree.ElementTree as ET
from dataclasses import dataclass
from urllib.parse import urljoin, urlparse, urlunparse

import requests
from bs4 import BeautifulSoup

USER_AGENT = "ambrosiana-intake-discovery/0.2 (personal RAG intake tool)"
REQUEST_TIMEOUT = 10
DEFAULT_MAX_PAGES = 300
SKIP_EXTENSIONS = (
    ".css", ".js", ".png", ".jpg", ".jpeg", ".gif", ".svg", ".ico",
    ".woff", ".woff2", ".ttf", ".xml", ".json", ".pdf", ".zip", ".md",
)
SITEMAP_XML_NS = "{http://www.sitemaps.org/schemas/sitemap/0.9}"


@dataclass
class DiscoveredPage:
    url: str
    title: str  # derived from the URL path only — never from fetched content
    depth: int
    sitemap_confirmed: bool | None = None  # None = no sitemap available to check against


def _display_title(url: str) -> str:
    """Human-readable label derived purely from the URL — no page content."""
    parsed = urlparse(url)
    path = parsed.path.strip("/")
    if not path:
        return parsed.netloc or url  # site root — use the domain, not the raw URL
    last_segment = path.rsplit("/", 1)[-1]
    return last_segment.replace("-", " ").replace("_", " ").strip() or url


def _normalize(url: str) -> str:
    """Strip fragment/query for de-dup purposes; keep the path as-is."""
    parsed = urlparse(url)
    return urlunparse((parsed.scheme, parsed.netloc, parsed.path.rstrip("/") or "/", "", "", ""))


def _is_same_domain(url: str, domain: str) -> bool:
    return urlparse(url).netloc == domain


def _is_content_link(url: str) -> bool:
    parsed = urlparse(url)
    if parsed.scheme not in ("http", "https"):
        return False
    return not parsed.path.lower().endswith(SKIP_EXTENSIONS)


def _fetch(url: str) -> requests.Response | None:
    try:
        response = requests.get(url, headers={"User-Agent": USER_AGENT}, timeout=REQUEST_TIMEOUT)
        response.raise_for_status()
        return response
    except requests.RequestException:
        return None


# ── Strategy 1: sitemap ──────────────────────────────────────────────────

def find_sitemap_url(start_url: str) -> str | None:
    """Check robots.txt's Sitemap: directive first, then the conventional path."""
    parsed = urlparse(start_url)
    root = f"{parsed.scheme}://{parsed.netloc}"

    robots = _fetch(f"{root}/robots.txt")
    if robots is not None:
        for line in robots.text.splitlines():
            if line.strip().lower().startswith("sitemap:"):
                return line.split(":", 1)[1].strip()

    candidate = f"{root}/sitemap.xml"
    response = _fetch(candidate)
    if response is not None and "xml" in response.headers.get("content-type", ""):
        return candidate
    return None


def fetch_sitemap_urls(sitemap_url: str, _depth: int = 0) -> set[str]:
    """
    Parse a sitemap.xml (or a sitemap index listing nested sitemaps, handled
    up to 2 levels deep) and return the full set of normalized <loc> URLs —
    used only as a cross-reference trust signal, never as the primary
    discovery source (see discover_links). ElementTree's default parser
    does not resolve external entities/DTDs, so this is safe against
    classic XXE without needing an extra dependency.
    """
    if _depth > 2:
        return set()

    response = _fetch(sitemap_url)
    if response is None:
        return set()

    try:
        root = ET.fromstring(response.content)
    except ET.ParseError:
        return set()

    locs = [el.text.strip() for el in root.iter(f"{SITEMAP_XML_NS}loc") if el.text]

    if root.tag == f"{SITEMAP_XML_NS}sitemapindex":
        urls: set[str] = set()
        for nested_url in locs[:20]:  # bounded — a sitemap index fan-out cap
            urls |= fetch_sitemap_urls(nested_url, _depth=_depth + 1)
        return urls

    return {_normalize(loc) for loc in locs}


# ── Strategy 2: HTML crawl fallback ──────────────────────────────────────

def discover_from_crawl(
    start_url: str,
    max_depth: int,
    max_pages: int = DEFAULT_MAX_PAGES,
    sitemap_urls: set[str] | None = None,
) -> list[DiscoveredPage]:
    """
    Breadth-first same-domain crawl from start_url, following the actual
    links on each page — this is what "N links deep from the index"
    literally means, including for hub/TOC-style index pages whose real
    sub-content lives at sibling paths rather than nested under the index's
    own URL. Extracts hrefs only — never page titles or body text (see
    module docstring). The start page itself is not included in the
    returned list, only pages found from it.

    If sitemap_urls is given, each discovered page is annotated with
    whether it's also present in the sitemap — a trust signal shown in the
    review UI, not a filter (a sitemap can be incomplete or stale).
    """
    domain = urlparse(start_url).netloc
    visited = {_normalize(start_url)}
    queue: list[tuple[str, int]] = [(start_url, 0)]
    discovered: list[DiscoveredPage] = []
    fetched_count = 0

    while queue and fetched_count < max_pages:
        url, depth = queue.pop(0)
        if depth >= max_depth:
            continue

        response = _fetch(url)
        fetched_count += 1
        if response is None:
            continue

        soup = BeautifulSoup(response.text, "html.parser")
        for anchor in soup.find_all("a", href=True):
            link = urljoin(url, anchor["href"])
            if not _is_content_link(link) or not _is_same_domain(link, domain):
                continue
            link_norm = _normalize(link)
            if link_norm in visited:
                continue
            visited.add(link_norm)

            confirmed = (link_norm in sitemap_urls) if sitemap_urls is not None else None
            discovered.append(
                DiscoveredPage(url=link, title=_display_title(link), depth=depth + 1, sitemap_confirmed=confirmed)
            )
            if len(discovered) + fetched_count >= max_pages:
                break
            queue.append((link, depth + 1))

    return [p for p in discovered if p.sitemap_confirmed or _is_reachable(p.url)]


def _is_reachable(url: str) -> bool:
    """
    Cheap liveness check for links not already confirmed via the sitemap.
    Relative-link resolution against an already-nested page can produce
    plausible-looking but nonexistent URLs (e.g. a same-name breadcrumb
    href resolving into a doubled path segment) — verified live 2026-08-07
    against textual.textualize.io, where this caught real 404s a sitemap-
    confirmed link would never need checking for.
    """
    try:
        response = requests.head(
            url, headers={"User-Agent": USER_AGENT}, timeout=REQUEST_TIMEOUT, allow_redirects=True
        )
        return response.status_code < 400
    except requests.RequestException:
        return False


# ── Entry point ───────────────────────────────────────────────────────────

def discover_links(
    start_url: str,
    max_depth: int,
    max_pages: int = DEFAULT_MAX_PAGES,
) -> tuple[list[DiscoveredPage], str]:
    """
    Returns (discovered_pages, method) — method is "sitemap-verified" or
    "crawl" so the UI can tell the human whether a sitemap was available to
    cross-check against.
    """
    sitemap_url = find_sitemap_url(start_url)
    sitemap_urls = fetch_sitemap_urls(sitemap_url) if sitemap_url else None

    pages = discover_from_crawl(start_url, max_depth=max_depth, max_pages=max_pages, sitemap_urls=sitemap_urls)
    method = "sitemap-verified" if sitemap_urls else "crawl"
    return pages, method