chemenu/tools/chemenu/sites/__init__.py

"""The site handlers behind `wikitool raw fetch`: what is known about how one
website serves its pages, as data.

A handler turns the URL a user typed into the website's standard form
(`canonical`), names the page furniture that is not part of the text (`drop`),
and says what the derivation cannot do for the reader (`checks`, printed as
`CHECK` lines). The registry is code in this stack - a website is a module here,
a registration and tests; the command, the header and the skill stay as they are.

Every building block is data, not a function: a canonical form is a list of
regular-expression replacements, `drop` a list of selectors, a check a text line
with an optional condition. So a handler is a JSON file (`data.py` reads it,
`site.schema.json` is its shape): the stack's under `handlers/`, an instance's
in `sites/` at its root, by the same building blocks. An instance handler for a
host and path the stack has a handler for replaces it; the `sites` section of the
settings still switches handlers off, sets the `locale` and adds hint lines
(`notes`). A handler may also offer the subpages of a page
(`Subpages`: the links to take, under a path prefix), which `raw fetch --scope`
fetches as one bundle; name the login page of a website behind a login and how
a page shows it arrived signed in (`Session`, for `raw login` and a fetch with
the user's session); and declare that its pages are drawn by JavaScript
(`render`). A handler may still be registered from a Python module in this
package. The core knows neither `typer` nor `rich` and raises `ValidationError`.
"""
from __future__ import annotations

import importlib
import pkgutil
import re
import urllib.parse
from dataclasses import dataclass, replace
from pathlib import Path
from typing import Optional

from chemenu import config_shapes, settings, web_capture
from chemenu.errors import ValidationError
from chemenu.web_capture import Rule

MAX_CHECKS = 3
LOCALE_PLACEHOLDER = "{locale}"
PAGE_PLACEHOLDER = "{page}"
INSTANCE_DIRECTORY = "sites"


@dataclass(frozen=True)
class Rewrite:
    """One step of a canonical form: `re.sub(pattern, replacement, url)`.
    `{locale}` in `replacement` is the instance's locale setting."""

    pattern: str
    replacement: str

    def __post_init__(self) -> None:
        re.compile(self.pattern)

    def apply(self, url: str, locale: str) -> str:
        return re.sub(self.pattern, self.replacement.replace(LOCALE_PLACEHOLDER, locale), url, count=1)


@dataclass(frozen=True)
class Check:
    """A hint line, with the condition under which it applies.

    `meta` names a `<meta name=...>` of the page: the line applies only when it
    has content, which replaces `{meta}` in `text`. `count` is a selector: the
    line applies only when it matches at least one element, and `{count}` is
    how many. With neither, the line always applies."""

    text: str
    meta: str = ""
    count: Optional[Rule] = None

    def line(self, document: web_capture._Element) -> Optional[str]:
        text = self.text
        if self.meta:
            value = web_capture.meta_content(document, self.meta)
            if not value:
                return None
            text = text.replace("{meta}", value)
        if self.count is not None:
            number = web_capture.count_matches(document, self.count)
            if not number:
                return None
            text = text.replace("{count}", str(number))
        return text


@dataclass(frozen=True)
class Subpages:
    """The subpages a page offers: the links `links` selects, resolved against
    the page's URL, that stay on its host and lie under `prefix`. `{page}` in
    `prefix` is the path of the page itself with a trailing slash, which is
    the default; the page is never its own subpage. Document order, each URL
    once, the fragment dropped - and nothing else is followed."""

    links: Rule
    prefix: str = PAGE_PLACEHOLDER

    def __post_init__(self) -> None:
        if not (self.prefix.startswith(PAGE_PLACEHOLDER) or self.prefix.startswith("/")):
            raise ValueError(f"a subpages prefix starts with {PAGE_PLACEHOLDER} or with /, not {self.prefix!r}")

    def urls(self, html: str, page_url: str) -> list[str]:
        page = urllib.parse.urlsplit(page_url)
        own = urllib.parse.unquote(page.path) or "/"
        own = own if own.endswith("/") else own + "/"
        prefix = self.prefix.replace(PAGE_PLACEHOLDER, own)
        found: list[str] = []
        for element in web_capture.matching(web_capture.parse(html), self.links):
            href = element.attrs.get("href", "").strip()
            if not href:
                continue
            url = urllib.parse.urldefrag(urllib.parse.urljoin(page_url, href))[0]
            parts = urllib.parse.urlsplit(url)
            path = urllib.parse.unquote(parts.path)
            if (
                parts.scheme.lower() not in web_capture.ALLOWED_SCHEMES
                or parts.netloc.lower() != page.netloc.lower()
                or not path.startswith(prefix)
                or path.rstrip("/") == own.rstrip("/")
                or path == prefix
                or url in found
            ):
                continue
            found.append(url)
        return found


@dataclass(frozen=True)
class Session:
    """What a website behind a login needs from the stack: the page `raw login`
    opens for the user to sign in on, and how a fetched page shows whether it
    arrived signed in. `signed_in` matches only on a signed-in page (a logout
    link), `signed_out` only on a page that was not (a paywall teaser); with
    both, both must hold. Whether a host's pages are fetched with the session at
    all is the instance's choice (`fetch.routes`), not the handler's."""

    login: str
    signed_in: Optional[Rule] = None
    signed_out: Optional[Rule] = None

    def __post_init__(self) -> None:
        if self.signed_in is None and self.signed_out is None:
            raise ValueError("a session needs a signed_in or a signed_out rule - nothing would tell a teaser from the article")
        if urllib.parse.urlsplit(self.login).scheme.lower() not in web_capture.ALLOWED_SCHEMES:
            raise ValueError(f"a session's login page is an http(s) URL, not {self.login!r}")

    def is_signed_in(self, html: str) -> bool:
        document = web_capture.parse(html)
        if self.signed_in is not None and not web_capture.count_matches(document, self.signed_in):
            return False
        if self.signed_out is not None and web_capture.count_matches(document, self.signed_out):
            return False
        return True


@dataclass(frozen=True)
class Site:
    """One handler. `host` is matched exactly (lower case); `path` is a regular
    expression matched against the start of the URL path, and the longest `path`
    wins among the handlers of a host. `parent` names a handler this one
    inherits from: its rewrites, rules and checks come first.

    `session` is what a fetch with the user's own login needs; `render` says the
    page draws its content with JavaScript, so it is fetched through a browser
    and the header says `rendered: true` - never without this declaration."""

    name: str
    host: str
    summary: str
    path: str = ""
    parent: str = ""
    rewrites: tuple[Rewrite, ...] = ()
    drop: tuple[Rule, ...] = ()
    checks: tuple[Check, ...] = ()
    subpages: Optional[Subpages] = None
    session: Optional[Session] = None
    render: bool = False
    origin: str = "stack"

    def __post_init__(self) -> None:
        if not self.name or not self.host or self.host != self.host.lower():
            raise ValueError(f"site {self.name!r}: needs a name and a lower-case host")
        re.compile(self.path)
        if len(self.checks) > MAX_CHECKS:
            raise ValueError(f"site {self.name!r}: a handler has at most {MAX_CHECKS} checks, got {len(self.checks)}")

    def matches(self, host: str, path: str) -> bool:
        return host == self.host and re.match(self.path, path) is not None

    def canonical(self, url: str, locale: str = "en-us") -> str:
        for step in self.rewrites:
            url = step.apply(url, locale)
        return url

    def check_lines(self, html: str) -> list[str]:
        if not self.checks:
            return []
        document = web_capture.parse(html)
        return [line for line in (check.line(document) for check in self.checks) if line]


_REGISTRY: list[Site] = []


def inherit(candidate: Site, parent: Site) -> Site:
    """`candidate` with its parent's rewrites, rules and checks first, so a
    lookup never walks a chain."""
    if parent.host != candidate.host:
        raise ValueError(f"site {candidate.name!r}: parent {parent.name!r} is for another host")
    flat = replace(
        candidate,
        rewrites=parent.rewrites + candidate.rewrites,
        drop=parent.drop + candidate.drop,
        checks=parent.checks + candidate.checks,
        subpages=candidate.subpages or parent.subpages,
        session=candidate.session or parent.session,
        render=candidate.render or parent.render,
    )
    if len(flat.checks) > MAX_CHECKS:
        raise ValueError(
            f"site {candidate.name!r}: with its parent's checks it has {len(flat.checks)}, "
            f"at most {MAX_CHECKS} are allowed"
        )
    return flat


def register(candidate: Site) -> Site:
    """Add a handler. One that names a `parent` is stored flattened."""
    for known in _REGISTRY:
        if known.name == candidate.name:
            raise ValueError(f"site {candidate.name!r} is already registered")
        if (known.host, known.path) == (candidate.host, candidate.path):
            raise ValueError(
                f"site {candidate.name!r} has the same host and path pattern as {known.name!r}: "
                f"{candidate.host} {candidate.path or '(any path)'}"
            )
    if candidate.parent:
        parent = next((s for s in _REGISTRY if s.name == candidate.parent), None)
        if parent is None:
            raise ValueError(f"site {candidate.name!r}: parent {candidate.parent!r} is not registered (yet)")
        candidate = inherit(candidate, parent)
    _REGISTRY.append(candidate)
    return candidate


def all_sites() -> list[Site]:
    return list(_REGISTRY)


def _host_and_path(url: str) -> tuple[str, str]:
    parts = urllib.parse.urlsplit(url)
    return (parts.hostname or "").lower(), urllib.parse.unquote(parts.path) or "/"


def catalogue(instance: tuple[Site, ...] = ()) -> list[Site]:
    """The handlers in effect: the stack's, except those an instance handler
    replaces (the same host and path), then the instance's."""
    taken = {(s.host, s.path) for s in instance}
    return [s for s in _REGISTRY if (s.host, s.path) not in taken] + list(instance)


def replacements(instance: tuple[Site, ...]) -> list[tuple[Site, Site]]:
    """Each instance handler paired with the stack handler it replaces."""
    return [
        (mine, theirs)
        for mine in instance
        for theirs in _REGISTRY
        if (mine.host, mine.path) == (theirs.host, theirs.path)
    ]


def site_for(url: str, disabled: frozenset[str] = frozenset(), instance: tuple[Site, ...] = ()) -> Optional[Site]:
    """The handler for `url`: the enabled one of its host whose path pattern is
    the longest among those that match, else None."""
    host, path = _host_and_path(url)
    found = [s for s in catalogue(instance) if s.name not in disabled and s.matches(host, path)]
    return max(found, key=lambda s: len(s.path), default=None)


def login_site(host: str, disabled: frozenset[str] = frozenset(), instance: tuple[Site, ...] = ()) -> Optional[Site]:
    """The enabled handler of `host` whose session `raw login` opens: among
    those that declare one, the one with the shortest path pattern - the
    website's own, not a section's."""
    found = [s for s in catalogue(instance) if s.name not in disabled and s.host == host and s.session]
    return min(found, key=lambda s: len(s.path), default=None)


# --- the instance's part ----------------------------------------------------


@dataclass(frozen=True)
class Note:
    host: str
    path: str
    lines: tuple[str, ...]


@dataclass(frozen=True)
class Broken:
    """A file of the instance's `sites/` that cannot be used, and why. `host` is
    the host it names, when that could be read."""

    file: str
    error: str
    host: Optional[str]


@dataclass(frozen=True)
class Preferences:
    disabled: frozenset[str] = frozenset()
    locale: str = "en-us"
    notes: tuple[Note, ...] = ()
    instance: tuple[Site, ...] = ()
    broken: tuple[Broken, ...] = ()
    max_pages: int = 200
    pause_seconds: float = 1.0


def read_instance(root: "Path | str | None" = None) -> tuple[tuple[Site, ...], tuple[Broken, ...]]:
    """The handlers in `sites/` under `root` (default: the instance root), and
    the files among them that cannot be used."""
    from chemenu import config
    from chemenu.sites import data

    directory = Path(root if root is not None else config.ROOT) / INSTANCE_DIRECTORY
    found, broken = data.instance_handlers(directory, list(_REGISTRY))
    return tuple(found), tuple(broken)


def read_preferences(root: "Path | str | None" = None) -> Preferences:
    """The `sites` section of the settings, or the defaults when it is absent.
    A malformed section, or a name in `disabled` that no handler has, is a
    `ValidationError` - a typo must not leave a handler on."""
    section = settings.load(root, only=(config_shapes.SITES.key,)).section(config_shapes.SITES.key)
    instance, broken = read_instance(root)
    names = {s.name for s in catalogue(instance)}
    disabled = frozenset(section["disabled"])
    unknown = sorted(disabled - names)
    if unknown:
        raise ValidationError(
            f"sites.disabled names {', '.join(unknown)}, which no handler has. "
            f"The handlers are: {', '.join(sorted(names))}."
        )
    notes = tuple(
        Note(entry["host"], entry.get("path", ""), tuple(entry["lines"])) for entry in section["notes"]
    )
    return Preferences(
        disabled, section["locale"], notes, instance, broken, section["max_pages"], section["pause_seconds"]
    )


@dataclass(frozen=True)
class Plan:
    """What applies to one URL: the handler (if any), the URL to fetch, the rules
    for the text, and the instance's notes for it."""

    url: str
    requested_url: str = ""
    site: Optional[Site] = None
    notes: tuple[str, ...] = ()

    @property
    def name(self) -> str:
        return self.site.name if self.site else ""

    @property
    def label(self) -> str:
        """The name as the capture header states it: an instance's own handler says so."""
        if not self.site:
            return ""
        return f"{self.site.name} (instance)" if self.site.origin == "instance" else self.site.name

    @property
    def drop(self) -> tuple[Rule, ...]:
        return self.site.drop if self.site else ()

    def checks(self, html: str, subpages: int = 0) -> list[str]:
        """The `CHECK` lines for the page: the handler's, the offer of `subpages`
        subpages when there are any, then the instance's notes."""
        lines = self.site.check_lines(html) if self.site else []
        if subpages:
            lines.append(
                f"this page is an index with {subpages} subpage(s) - rerun with `--scope` to "
                "fetch them as one bundle"
            )
        return lines + list(self.notes)

    def subpages(self, html: str, page_url: str) -> list[str]:
        """The subpage URLs the handler offers for this page, else none."""
        if self.site is None or self.site.subpages is None:
            return []
        return self.site.subpages.urls(html, page_url)


def plan_for(url: str, preferences: Preferences, *, canonical: bool = True) -> Plan:
    """The plan for `url`. `canonical=False` keeps the URL as given - for a page
    that is already on disk, where the standard form would be a fiction."""
    host = _host_and_path(url)[0]
    blocking = [b for b in preferences.broken if b.host is None or b.host == host]
    if blocking:
        raise ValidationError(
            f"{host or url}: a handler file of this instance cannot be used, so the fetch is not made - "
            + "; ".join(f"{b.file}: {b.error}" for b in blocking)
            + ". Fix or remove the file (`wikitool doctor` lists them)."
        )
    site = site_for(url, preferences.disabled, preferences.instance)
    target = url
    if site is not None and canonical:
        target = site.canonical(url, preferences.locale)
    host, path = _host_and_path(target)
    notes = tuple(
        line
        for note in preferences.notes
        if note.host == host and re.match(note.path, path)
        for line in note.lines
    )
    return Plan(target, url if target != url else "", site, notes)


def _load_modules() -> None:
    from chemenu.sites import data

    for site in data.stack_handlers():
        register(site)
    for module in pkgutil.iter_modules(__path__):
        if module.name != "data":
            importlib.import_module(f"{__name__}.{module.name}")


_load_modules()