arborist/aborist/sources/crawler/bridge.py

"""Bridge: aborist sources/crawler ↔ aborist ingest pipeline.

Two operations:

1. **crawl_seed(seed_url, depth, max_pages)** — BFS-discovers same-domain
   URLs from a seed, respecting robots.txt + crawl delays. Returns a
   list of URLs ready for ingest.

2. **ingest_crawled(conn, urls)** — fetches each URL with httpx,
   captures ETag + Last-Modified, parses HTML, runs through the
   standard ingest_source path so the documents land in the same
   tables as any other source. After ingest, writes one row per
   document into ``document_http_meta`` so a future recrawl-check can
   send conditional HEAD requests.

Off by default — both operations require ``aborist[crawler]`` extras.
"""

from __future__ import annotations

import asyncio
import time
import urllib.parse
import urllib.robotparser
from collections import deque
from pathlib import Path
from typing import Iterable, Iterator

import httpx

from aborist.document import Document
from aborist.ingest import ingest_source
from aborist.progress import Progress
from aborist.sources.crawler.async_web_fetcher import AsyncWebFetcher
from aborist.sources.html_page import USER_AGENT, parse_html
from aborist.store import transaction


def _same_domain(a: str, b: str) -> bool:
    """True iff both URLs share an exact hostname (no subdomain crossover)."""
    return urllib.parse.urlparse(a).netloc == urllib.parse.urlparse(b).netloc


def _normalize(url: str) -> str:
    """Canonical form for dedup: scheme + netloc + path + query, no fragment."""
    parsed = urllib.parse.urlparse(url)
    cleaned = parsed._replace(fragment="")
    return urllib.parse.urlunparse(cleaned)


# Feeds, sitemaps, and similar XML discovery artifacts are crawl
# infrastructure — they tell you which URLs exist on a site. They are
# NOT knowledge content, so they must not enter the Merkle tree as
# documents. We exclude them in two passes: (1) URL pattern match on
# the path so we skip even before the response arrives, (2) content-
# type check on the response so feeds served at unconventional paths
# still get rejected. xhtml is preserved because it IS HTML.
_FEED_URL_SUFFIXES = (
    ".atom",
    ".atom.xml",
    ".rss",
    ".rss.xml",
    ".rdf",
    # Common feed filename patterns regardless of where they live.
    "/feed.xml",
    "/atom.xml",
    "/rss.xml",
    "/rss2.xml",
    "wp-rss2.xml",   # WordPress legacy
    "wp-rss.xml",
    "wp-atom.xml",
    "wp-rdf.xml",
)
_FEED_URL_SUBSTRINGS = (
    "/feed/",
    "/feeds/",
    "/atom",
    "/rss",
    "/sitemap",          # sitemap.xml, sitemap_index.xml, sitemaps/foo.xml
)
_FEED_CONTENT_TYPES = (
    "application/atom+xml",
    "application/rss+xml",
    "application/rdf+xml",
    "application/xml",
    "text/xml",
)


def _looks_like_feed_url(url: str) -> bool:
    """True if the URL path matches a known feed/sitemap pattern."""
    path = urllib.parse.urlparse(url).path.lower()
    if any(path.endswith(s) for s in _FEED_URL_SUFFIXES):
        return True
    if any(s in path for s in _FEED_URL_SUBSTRINGS):
        return True
    if path.endswith("/sitemap.xml") or path.endswith("sitemapindex.xml"):
        return True
    return False


def _looks_like_feed_response(content_type: str) -> bool:
    """True if a response Content-Type is a feed/sitemap rather than HTML."""
    ct = content_type.lower().split(";", 1)[0].strip()
    return ct in _FEED_CONTENT_TYPES


async def _crawl_seed_async(
    seed_url: str,
    *,
    max_depth: int,
    max_pages: int,
    fetcher: AsyncWebFetcher | None = None,
    progress: Progress | None = None,
    fast: bool = False,
) -> list[str]:
    """BFS from `seed_url`, staying on the same hostname.

    Discovery only — fetches each page enough to extract links, no
    aborist-side ingestion happens here. Returns a deduplicated list
    of URLs in BFS order, capped at `max_pages`.

    Pass ``progress`` (an `aborist.progress.Progress`) for stderr
    heartbeats while the BFS runs. Lines are rate-limited by the
    Progress instance (default 2s) so a fast crawl prints once or
    twice; a slow one prints every interval.

    ``fast=True`` flips the verbatim AsyncWebFetcher into fast_mode:
    5s timeouts (vs 15-60s), CPU*3 parallel page workers, and the
    target's robots.txt ``crawl-delay`` directive is ignored. Robots
    Disallow is still honored. Use only against domains where
    aggressive fetching is acceptable (your own sites, dumps you've
    been authorized to mirror, etc.).
    """
    fetcher = fetcher or AsyncWebFetcher(user_agent=USER_AGENT, fast_mode=fast)
    seen: set[str] = set()
    discovered: list[str] = []
    queue: deque[tuple[str, int]] = deque([(_normalize(seed_url), 0)])

    # max_pages == 0 means "no cap" — depth is the only bound. Useful for
    # exhaustive same-domain crawls where the operator trusts depth to
    # bound the crawl naturally.
    while queue and (max_pages == 0 or len(discovered) < max_pages):
        url, depth = queue.popleft()
        if url in seen:
            continue
        seen.add(url)

        try:
            # extract_links=True is REQUIRED for BFS — the verbatim
            # AsyncWebFetcher defaults it to False (caller opts in for the
            # link-graph use case). Without it BFS terminates at the seed.
            html, links = await fetcher.fetch_webpage(url, extract_links=True)
        except Exception:
            continue
        if html is None:
            continue

        discovered.append(url)
        if progress is not None:
            progress.tick(len(discovered), inserted=len(queue))
        if depth >= max_depth:
            continue

        for link in links or []:
            link_url = link[0] if isinstance(link, tuple) else link
            if not isinstance(link_url, str):
                continue
            absolute = urllib.parse.urljoin(url, link_url)
            normalized = _normalize(absolute)
            if normalized in seen:
                continue
            if not _same_domain(seed_url, normalized):
                continue
            if not normalized.startswith(("http://", "https://")):
                continue
            queue.append((normalized, depth + 1))

    if progress is not None:
        progress.done(len(discovered), inserted=len(queue))
    return discovered


def crawl_seed(
    seed_url: str,
    *,
    max_depth: int = 2,
    max_pages: int = 0,
    progress: Progress | None = None,
    fast: bool = False,
) -> list[str]:
    """Sync wrapper around the async crawler. Returns same-domain URLs.

    ``max_pages=0`` means "no cap" — depth is the only bound. Pass
    ``progress`` for stderr heartbeats while BFS runs. ``fast=True``
    enables AsyncWebFetcher's fast_mode (5s timeouts, CPU*3 workers,
    ignores robots crawl-delay).
    """
    return asyncio.run(
        _crawl_seed_async(
            seed_url,
            max_depth=max_depth,
            max_pages=max_pages,
            progress=progress,
            fast=fast,
        )
    )


class _CrawledHtmlSource:
    """Source that fetches each URL synchronously, captures HTTP metadata,
    parses HTML, and yields one Document. The metadata is recorded in a
    side dict so the caller can write it to ``document_http_meta`` after
    ingest commits the document_root.
    """

    source_type = "html"

    def __init__(self, urls: Iterable[str], *, timeout: float = 30.0):
        self.urls = list(urls)
        self.timeout = timeout
        # Filled in during iter_documents — keyed by document_uri (request
        # URL pre-redirect) so the caller can map document_root → metadata
        # via the documents.document_uri column after ingest.
        self.http_meta: dict[str, dict[str, object]] = {}

    def iter_documents(self) -> Iterator[Document]:
        with httpx.Client(
            headers={"User-Agent": USER_AGENT},
            timeout=self.timeout,
            follow_redirects=True,
        ) as client:
            for url in self.urls:
                # Cheap pre-flight: skip feed/sitemap URLs without a fetch.
                if _looks_like_feed_url(url):
                    continue
                try:
                    resp = client.get(url)
                    resp.raise_for_status()
                except httpx.HTTPError:
                    continue
                ctype = resp.headers.get("content-type", "").lower()
                # Defense in depth: a feed served at a non-feed path still
                # gets rejected by Content-Type.
                if _looks_like_feed_response(ctype):
                    continue
                if "html" not in ctype:
                    continue
                doc = parse_html(str(resp.url), resp.text, self.source_type)
                if doc is None:
                    continue
                self.http_meta[doc.uri] = {
                    "etag": resp.headers.get("etag"),
                    "last_modified": resp.headers.get("last-modified"),
                    "last_fetched_at": int(time.time()),
                }
                yield doc


def ingest_crawled(
    conn,
    urls: Iterable[str],
    *,
    timeout: float = 30.0,
    progress: Progress | None = None,
) -> dict:
    """Fetch each URL, ingest into `conn`, record ETag + Last-Modified.

    Returns a summary dict with counts and the list of (uri, document_root)
    pairs written. The caller's connection must be writable; metadata is
    persisted in the same connection so an interrupted run leaves either
    the document AND its http_meta or neither (FK guarantees this on
    cascade delete; the upsert happens inside the same transaction
    boundary as the audit event).

    Pass ``progress`` for stderr heartbeats while ingest runs (the
    underlying ingest_source supports it natively).
    """
    src = _CrawledHtmlSource(urls, timeout=timeout)
    stats = ingest_source(conn, src, progress=progress)
    written: list[dict] = []
    if not src.http_meta:
        return {
            "seen": stats.seen,
            "inserted": stats.inserted,
            "http_meta_written": 0,
            "documents": [],
        }

    # Map document_uri → document_root via the documents table (URI is
    # the post-redirect URL the source set on the Document, which is what
    # parse_html stored). The map is stable as long as URIs are unique
    # within the corpus, which is enforced at ingest time.
    uri_list = list(src.http_meta.keys())
    placeholders = ",".join(["?"] * len(uri_list))
    rows = conn.execute(
        f"SELECT document_root, document_uri FROM documents "
        f"WHERE document_uri IN ({placeholders})",
        uri_list,
    ).fetchall()

    with transaction(conn):
        for r in rows:
            meta = src.http_meta.get(r["document_uri"])
            if meta is None:
                continue
            conn.execute(
                "INSERT OR REPLACE INTO document_http_meta "
                "(document_root, etag, last_modified, last_fetched_at, "
                " last_status, last_checked_at) "
                "VALUES (?, ?, ?, ?, 200, ?)",
                (
                    r["document_root"],
                    meta["etag"],
                    meta["last_modified"],
                    meta["last_fetched_at"],
                    meta["last_fetched_at"],
                ),
            )
            written.append(
                {
                    "document_root": r["document_root"],
                    "document_uri": r["document_uri"],
                }
            )

    return {
        "seen": stats.seen,
        "inserted": stats.inserted,
        "http_meta_written": len(written),
        "documents": written,
    }


def recrawl_check(
    conn,
    *,
    domain: str | None = None,
    limit: int = 100,
    timeout: float = 10.0,
) -> dict:
    """Send conditional HEAD requests for ingested documents and classify
    each as fresh/stale/gone/unreachable.

    Fast path: ETag → If-None-Match. Server returns 304 with no body =
    skip. Slow path: full re-fetch only if the server says 200. We do
    NOT re-fetch the body in this verb — that's recrawl's job. We just
    answer "needs recrawl: yes/no" without doing the recrawl.
    """
    sql = (
        "SELECT d.document_root, d.document_uri, m.etag, m.last_modified "
        "FROM documents d "
        "JOIN document_http_meta m ON m.document_root = d.document_root"
    )
    args: list[object] = []
    if domain:
        sql += " WHERE d.document_uri LIKE ?"
        args.append(f"%//{domain}%")
    sql += " ORDER BY m.last_checked_at ASC NULLS FIRST LIMIT ?"
    args.append(limit)

    rows = conn.execute(sql, args).fetchall()

    fresh, stale, gone, unreachable = 0, 0, 0, 0
    items: list[dict] = []
    now = int(time.time())

    with httpx.Client(
        headers={"User-Agent": USER_AGENT},
        timeout=timeout,
        follow_redirects=True,
    ) as client:
        for r in rows:
            headers: dict[str, str] = {}
            if r["etag"]:
                headers["If-None-Match"] = r["etag"]
            if r["last_modified"]:
                headers["If-Modified-Since"] = r["last_modified"]

            try:
                resp = client.head(r["document_uri"], headers=headers)
                status = resp.status_code
            except httpx.HTTPError:
                status = 0  # network failure / DNS / TLS

            if status == 304:
                fresh += 1
                verdict = "fresh"
            elif status in (200,):
                stale += 1
                verdict = "stale"
            elif status in (404, 410):
                gone += 1
                verdict = "gone"
            else:
                unreachable += 1
                verdict = "unreachable"

            with transaction(conn):
                conn.execute(
                    "UPDATE document_http_meta "
                    "SET last_status = ?, last_checked_at = ? "
                    "WHERE document_root = ?",
                    (status, now, r["document_root"]),
                )

            items.append(
                {
                    "document_root": r["document_root"],
                    "document_uri": r["document_uri"],
                    "status": status,
                    "verdict": verdict,
                }
            )

    return {
        "examined": len(rows),
        "fresh": fresh,
        "stale": stale,
        "gone": gone,
        "unreachable": unreachable,
        "items": items,
    }