arborist/aborist/sources/html_page.py

"""HTML page source.

Fetches URLs, honors robots.txt automatically, strips noise (script/style/nav/
footer/header), extracts main body text + outbound `<a href>` links as edges.

Optional dependency. Install with `pip install aborist[html]`.
"""

from __future__ import annotations

import re
import urllib.parse
import urllib.robotparser
from pathlib import Path
from typing import Iterable, Iterator

try:
    import httpx
    from selectolax.parser import HTMLParser
except ImportError as e:  # pragma: no cover
    raise ImportError(
        "HTML source requires extras: pip install 'aborist[html]'"
    ) from e

from aborist.document import Document, Edge
from aborist.source import Source


USER_AGENT = "aborist/0.0.1 (+https://unturf.com)"
NOISE_SELECTORS = ("script", "style", "noscript", "nav", "header", "footer", "aside")


def _normalize_text(text: str) -> str:
    text = re.sub(r"[ \t]+", " ", text)
    text = re.sub(r"\n{3,}", "\n\n", text)
    return text.strip()


def parse_html(url: str, html: str, source_type: str = "html") -> Document | None:
    """Pure parse function. Separated so tests can run without network."""
    tree = HTMLParser(html)
    for sel in NOISE_SELECTORS:
        for node in tree.css(sel):
            node.decompose()

    body = tree.css_first("body") or tree.root
    if body is None:
        return None
    text = _normalize_text(body.text(separator="\n", strip=True))
    if not text:
        return None

    title_node = tree.css_first("title")
    title = title_node.text(strip=True) if title_node is not None else None

    edges: list[Edge] = []
    seen: set[tuple[str, str]] = set()
    for a in tree.css("a[href]"):
        href = (a.attributes.get("href") or "").strip()
        if not href or href.startswith(("javascript:", "mailto:", "tel:", "#")):
            continue
        absolute = urllib.parse.urljoin(url, href)
        split = urllib.parse.urlsplit(absolute)
        if split.scheme not in ("http", "https"):
            continue
        anchor = split.fragment or ""
        dst_uri = urllib.parse.urlunsplit(split._replace(fragment=""))
        key = (dst_uri, anchor)
        if key in seen:
            continue
        seen.add(key)
        edges.append(Edge(edge_type="hyperlink", dst_uri=dst_uri, anchor=anchor or None))

    return Document(
        uri=url,
        content=text,
        source_type=source_type,
        title=title,
        edges=edges,
    )


class HtmlPageSource(Source):
    """Iterates a list of URLs, fetching and parsing each as HTML."""

    source_type = "html"

    def __init__(
        self,
        urls: Iterable[str],
        *,
        respect_robots: bool = True,
        timeout: float = 30.0,
    ):
        self.urls = list(urls)
        self.respect_robots = respect_robots
        self.timeout = timeout
        self._robots_cache: dict[str, urllib.robotparser.RobotFileParser] = {}

    @classmethod
    def from_file(cls, path: str | Path, **kwargs) -> HtmlPageSource:
        urls = [
            line.strip()
            for line in Path(path).read_text(encoding="utf-8").splitlines()
            if line.strip() and not line.lstrip().startswith("#")
        ]
        return cls(urls, **kwargs)

    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:
                if self.respect_robots and not self._allowed(client, url):
                    continue
                try:
                    resp = client.get(url)
                    resp.raise_for_status()
                except httpx.HTTPError:
                    continue
                ctype = resp.headers.get("content-type", "").lower()
                if "html" not in ctype and "xml" not in ctype:
                    continue
                doc = parse_html(str(resp.url), resp.text, self.source_type)
                if doc is not None:
                    yield doc

    def _allowed(self, client: "httpx.Client", url: str) -> bool:
        parsed = urllib.parse.urlparse(url)
        origin = f"{parsed.scheme}://{parsed.netloc}"
        rp = self._robots_cache.get(origin)
        if rp is None:
            rp = urllib.robotparser.RobotFileParser()
            try:
                resp = client.get(f"{origin}/robots.txt")
            except httpx.HTTPError:
                resp = None
            if resp is not None and resp.status_code == 200:
                rp.parse(resp.text.splitlines())
            else:
                # Missing robots.txt = no rules per RFC 9309.
                rp.allow_all = True
            self._robots_cache[origin] = rp
        return rp.can_fetch(USER_AGENT, url)