catalogs.py

"""Ethical book-data sources, behind a hard host allowlist.

Hard guardrail (README): **do not scrape Goodreads** (Amazon ToS + gatekeeping +
surveillance). Recommendations are sourced from OpenLibrary, Hardcover, and
Bookwyrm, each tagged with provenance. This module is the single choke point for
catalog network access, and :func:`assert_allowed` makes Goodreads/Amazon a
build-time impossibility: any request to a blocked host raises before a socket is
opened. The merge-blocking metric "Goodreads requests = 0" is enforced here plus
by the source-allowlist test.

Only public catalog metadata is ever fetched, and only from the operator's own
predeclared configuration: broad Open Library subject slugs and explicit public
BookWyrm list URLs. There is deliberately no per-title or per-ISBN lookup — such
a request would carry a book the reader owns, which is exactly the thing that
must not leave. ``tests/test_no_egress.py`` asserts the outbound URL set for a
real refresh and that nothing derived from the library appears in it.
"""

from __future__ import annotations

from pathlib import Path
from typing import Any, Optional, Protocol, runtime_checkable
from urllib.parse import quote, urlparse

from ingest.models import Author, Book, Source, SourceKind, ThemeTag, merge_tags
from ingest.unify import normalize_key

#: Hosts the recommender is permitted to fetch from. Each is an ethical,
#: non-gatekept catalog with a usable API or open data.
ALLOWED_HOSTS: frozenset[str] = frozenset(
    {
        "openlibrary.org",
        "covers.openlibrary.org",
        "api.hardcover.app",
        "bookwyrm.social",
    }
)

#: Explicitly blocked hosts — recorded so the exclusion is legible, not implicit.
#: Goodreads (and its Amazon parent) are excluded on ToS + values grounds.
BLOCKED_HOSTS: frozenset[str] = frozenset(
    {
        "goodreads.com",
        "www.goodreads.com",
        "amazon.com",
        "www.amazon.com",
    }
)


class SourceNotAllowed(Exception):
    """Raised when a catalog request targets a blocked or non-allowlisted host."""


class CatalogPayloadInvalid(ValueError):
    """Raised when a live catalog response does not match its top-level contract."""


class _CatalogResponse(Protocol):
    """The small ``requests.Response`` surface used by the catalog clients."""

    status_code: int

    @property
    def text(self) -> str: ...

    def raise_for_status(self) -> None: ...


#: A descriptive, identifying User-Agent for every outbound catalog/federation
#: request. Federation etiquette (EV-LICENSE): a host can see exactly who we are
#: and that we are a read-only, TTL-caching, self-hosted consumer of *public* metadata.
USER_AGENT: str = (
    "QueerTheStacks/1.0 (self-hosted reading dashboard; read-only public-metadata "
    "fetch; TTL-cached candidate pool; see docs/ethical-book-data-sources.md)"
)


def etiquette_headers(accept: str = "application/json") -> dict[str, str]:
    """Polite, identifying HTTP headers for every catalog/federation fetch.

    Pairs with the persisted, TTL-bounded candidate pool, robots/rate-limit
    respect, and backoff policy — the documented federation etiquette in
    ``docs/ethical-book-data-sources.md``. Only public catalog metadata is ever
    requested; reading data is never sent.
    """
    return {"User-Agent": USER_AGENT, "Accept": accept}


#: Where Open Library serves subject pages. Kept at module level (rather than
#: only on the client) because the citation URL for a subject is built in
#: several places and must be the same string the client would fetch.
SUBJECTS_ROOT = "https://openlibrary.org/subjects"


def subject_slug(subject: str) -> str:
    """Normalize a subject label into Open Library's own slug form.

    Open Library subject paths are lowercase with underscores for spaces
    (``science_fiction``), which is also the only shape
    :data:`recommender.catalog_pool._OL_SUBJECT` accepts as a configured
    subject. Interpolating a raw label instead produced citations like
    ``.../subjects/science fiction`` — a URL with a literal space in it, which
    is not a URL. Anything still unsafe after normalizing is percent-encoded
    rather than passed through.
    """
    normalized = "_".join(subject.strip().lower().split())
    return quote(normalized, safe="_-")


def subject_url(subject: str) -> str:
    """The canonical Open Library citation URL for ``subject``."""
    return f"{SUBJECTS_ROOT}/{subject_slug(subject)}"


def is_citable_url(url: str) -> bool:
    """True if ``url`` is safe to render as a clickable citation link.

    The same allowlist the fetch path enforces, applied to the *display* path:
    a citation is only presented as a link if it is a well-formed, credential-
    free HTTPS URL on an allowlisted catalog host and contains no whitespace.
    Everything else stays visible as text — a reader can still read and check
    it, but the page never hands them a malformed or off-allowlist link.
    """
    if url != url.strip() or any(ch.isspace() for ch in url):
        return False
    try:
        assert_allowed(url)
    except SourceNotAllowed:
        return False
    return True


def assert_allowed(url: str) -> str:
    """Return ``url`` iff it is a credential-free HTTPS URL on the allowlist.

    A blocked host (Goodreads/Amazon) raises with an explicit message; an
    unknown host raises too (default-deny). Cleartext, URL credentials,
    fragments, and non-standard ports are rejected before any request.
    """
    parsed = urlparse(url)
    host = (parsed.hostname or "").lower()
    if not host:
        raise SourceNotAllowed(f"no host in URL: {url!r}")
    if parsed.scheme.lower() != "https":
        raise SourceNotAllowed("catalog requests require HTTPS")
    if parsed.username is not None or parsed.password is not None:
        raise SourceNotAllowed("catalog URLs must not contain credentials")
    if parsed.fragment:
        raise SourceNotAllowed("catalog URLs must not contain fragments")
    try:
        port = parsed.port
    except ValueError as exc:
        raise SourceNotAllowed("catalog URL has an invalid port") from exc
    if port not in {None, 443}:
        raise SourceNotAllowed("catalog URLs may use only the standard HTTPS port")
    if host in BLOCKED_HOSTS:
        raise SourceNotAllowed(
            f"{host} is a blocked source (Goodreads/Amazon excluded on ToS + values grounds)"
        )
    if host not in ALLOWED_HOSTS:
        raise SourceNotAllowed(f"{host} is not in the catalog allowlist (default-deny)")
    return url


def _catalog_get(url: str, timeout: int) -> _CatalogResponse:
    """Issue one allowlisted HTTPS GET without following redirect hops."""
    import requests

    safe_url = assert_allowed(url)
    response = requests.get(
        safe_url,
        timeout=timeout,
        headers=etiquette_headers(),
        allow_redirects=False,
    )
    if 300 <= response.status_code < 400:
        raise SourceNotAllowed("catalog redirects are disabled")
    response.raise_for_status()
    return response


def _validated_collection_payload(body: str, key: str) -> object:
    """Decode a catalog response whose required top-level field is a list."""
    import json

    payload: object = json.loads(body)
    if not isinstance(payload, dict) or not isinstance(payload.get(key), list):
        raise CatalogPayloadInvalid(f"catalog response requires a top-level {key!r} list")
    return payload


@runtime_checkable
class CatalogSource(Protocol):
    """The catalog interface the recommender depends on."""

    def candidates(self) -> tuple[Book, ...]: ...


class FixtureCatalog:
    """A deterministic, offline :class:`CatalogSource` built from plain books."""

    def __init__(self, books: tuple[Book, ...]) -> None:
        self._books = books

    def candidates(self) -> tuple[Book, ...]:
        return self._books


# --- Pure parsers (unit-tested via fixtures; the live clients feed these) -----
#
# Each parser maps a catalog's JSON response to `Book`s whose theme tags carry the
# right `SourceKind` + citation. They validate shape defensively (untrusted
# external data) and never raise on a single malformed record.


def _authors(raw: Any, key: str = "name") -> tuple[Author, ...]:
    if not isinstance(raw, list):
        return ()
    out = []
    for a in raw:
        if isinstance(a, dict) and a.get(key):
            out.append(Author(name=str(a[key])))
        elif isinstance(a, str) and a.strip():
            out.append(Author(name=a.strip()))
    return tuple(out)


def parse_openlibrary_subject(
    payload: object, subject: str, citation: str, retrieved_at: str
) -> tuple[Book, ...]:
    """Parse an Open Library ``/subjects/<s>.json`` response into Books."""
    works = payload.get("works", []) if isinstance(payload, dict) else []
    src = Source(SourceKind.OPENLIBRARY_SUBJECT, citation, retrieved_at, subject)
    out: list[Book] = []
    for w in works if isinstance(works, list) else []:
        if not isinstance(w, dict) or not w.get("title"):
            continue
        out.append(
            Book(
                book_id=f"ol:{w.get('key', w['title'])}",
                title=str(w["title"]),
                authors=_authors(w.get("authors")),
                theme_tags=(ThemeTag(subject, src),),
            )
        )
    return tuple(out)


def parse_hardcover_books(payload: object, citation: str, retrieved_at: str) -> tuple[Book, ...]:
    """Parse a Hardcover GraphQL ``books`` response (``data.books[]``) into Books.

    Expected per-book shape: ``{title, contributions:[{author:{name}}],
    cached_tags:{Genre:[{tag}], Mood:[{tag}], ...}}``.
    """
    data = payload.get("data", {}) if isinstance(payload, dict) else {}
    books = data.get("books", []) if isinstance(data, dict) else []
    out: list[Book] = []
    for b in books if isinstance(books, list) else []:
        if not isinstance(b, dict) or not b.get("title"):
            continue
        authors = tuple(
            Author(name=str(c["author"]["name"]))
            for c in b.get("contributions", [])
            if isinstance(c, dict) and isinstance(c.get("author"), dict) and c["author"].get("name")
        )
        tags: list[ThemeTag] = []
        cached = b.get("cached_tags", {})
        if isinstance(cached, dict):
            for group in cached.values():
                for t in group if isinstance(group, list) else []:
                    label = t.get("tag") if isinstance(t, dict) else None
                    if label:
                        tags.append(
                            ThemeTag(
                                str(label),
                                Source(
                                    SourceKind.HARDCOVER_TAG, citation, retrieved_at, str(label)
                                ),
                            )
                        )
        out.append(
            Book(
                book_id=f"hardcover:{b.get('slug', b['title'])}",
                title=str(b["title"]),
                authors=authors,
                theme_tags=merge_tags(tags),
            )
        )
    return tuple(out)


def parse_bookwyrm_list(payload: object, citation: str, retrieved_at: str) -> tuple[Book, ...]:
    """Parse a Bookwyrm list/shelf response (``{books:[{title,authors,subjects}]}``)."""
    books = payload.get("books", []) if isinstance(payload, dict) else []
    out: list[Book] = []
    for b in books if isinstance(books, list) else []:
        if not isinstance(b, dict) or not b.get("title"):
            continue
        tags = tuple(
            ThemeTag(str(s), Source(SourceKind.BOOKWYRM_SHELF, citation, retrieved_at, str(s)))
            for s in b.get("subjects", [])
            if isinstance(s, str) and s.strip()
        )
        out.append(
            Book(
                book_id=f"bookwyrm:{b.get('id', b['title'])}",
                title=str(b["title"]),
                authors=_authors(b.get("authors")),
                theme_tags=merge_tags(list(tags)),
            )
        )
    return tuple(out)


def merge_candidates(*groups: tuple[Book, ...]) -> tuple[Book, ...]:
    """Merge candidate books across sources, de-duped by title|author.

    When the same work appears from two catalogs, their sourced theme tags are
    unioned (provenance preserved), so a book gains tags from every source that
    listed it. Deterministic: first-seen identity wins, tags merge in order.
    """
    by_key: dict[str, Book] = {}
    for group in groups:
        for book in group:
            key = normalize_key(book.title, book.author_names)
            if key not in by_key:
                by_key[key] = book
            else:
                existing = by_key[key]
                merged = merge_tags(list(existing.theme_tags) + list(book.theme_tags))
                by_key[key] = Book(
                    book_id=existing.book_id,
                    title=existing.title,
                    authors=existing.authors or book.authors,
                    series=existing.series or book.series,
                    series_index=existing.series_index,
                    identifiers={**book.identifiers, **existing.identifiers},
                    theme_tags=merged,
                    pubdate=existing.pubdate or book.pubdate,
                )
    return tuple(by_key.values())


class ResponseCache:
    """A tiny on-disk JSON cache for catalog responses (legal/ops: don't re-hit APIs).

    Keyed by URL. Stored as one JSON file so it is trivial to inspect and to clear.
    Only public catalog metadata is ever cached — never reading data.
    """

    def __init__(self, path: Optional[Path] = None) -> None:
        self.path = Path(path) if path is not None else None
        self._mem: dict[str, str] = {}
        if self.path is not None and self.path.is_file():
            import json

            try:
                loaded = json.loads(self.path.read_text(encoding="utf-8"))
                if isinstance(loaded, dict):
                    self._mem = {str(k): str(v) for k, v in loaded.items()}
            except ValueError, OSError:  # pragma: no cover - corrupt cache is non-fatal
                self._mem = {}

    def get(self, url: str) -> Optional[str]:
        return self._mem.get(url)

    def put(self, url: str, body: str) -> None:
        self._mem[url] = body
        if self.path is not None:
            import json

            self.path.parent.mkdir(parents=True, exist_ok=True)
            self.path.write_text(json.dumps(self._mem), encoding="utf-8")


class OpenLibraryClient:
    """Live OpenLibrary client. Every request passes through :func:`assert_allowed`."""

    def __init__(self, cache: Optional[ResponseCache] = None, timeout: int = 15) -> None:
        self.cache = cache
        self.timeout = timeout

    def subject(self, subject: str, limit: int = 50) -> tuple[Book, ...]:
        import time

        url = assert_allowed(f"{SUBJECTS_ROOT}/{subject}.json?limit={limit}")
        body = self._fetch(url)
        payload = _validated_collection_payload(body, "works")
        if self.cache is not None:
            # Cache only validated responses: a transient HTML/error body or
            # schema drift must not poison every later refresh.
            self.cache.put(url, body)
        return parse_openlibrary_subject(payload, subject, url, time.strftime("%Y-%m-%d"))

    def _fetch(self, url: str) -> str:
        url = assert_allowed(url)
        if self.cache is not None:
            cached = self.cache.get(url)
            if cached is not None:
                return cached
        return _catalog_get(url, self.timeout).text


class BookwyrmClient:
    """Live Bookwyrm list client behind the allowlist."""

    def __init__(self, timeout: int = 15) -> None:
        self.timeout = timeout

    def fetch_list(self, list_url: str) -> tuple[Book, ...]:
        import time

        url = assert_allowed(list_url)
        response = _catalog_get(url, self.timeout)
        payload = _validated_collection_payload(response.text, "books")
        return parse_bookwyrm_list(payload, url, time.strftime("%Y-%m-%d"))