`src/conrad/provenance.py`

86 lines
"""Acquisition provenance — HOW each cartoon_sources row was obtained (Cody must-fix, TK-12199).

`acquisition_method` is independent of `provenance` (the free-text trail): it is a closed vocabulary that
lets exports, the API and the viewer flag records whose acquisition path is weaker than a direct fetch.

  direct_api              fetched live by this project from a documented JSON/XML API (e.g. loc.gov ?fo=json)
  direct_html             fetched live by this project from a public HTML/XML document (e.g. the Huntington EAD)
  seed_direct             seed corpus (TK-12179) whose raw cache was fetched directly from the repository
  seed_via_reader_bypass  seed corpus fetched through a third-party reader (r.jina.ai) past an AWS WAF / bot
                          challenge — every seed_wichita and seed_syracuse row. Pending direct verification.
  secondary_citation      not from a holding repository at all: a secondary source (obituary, article, book)
"""
from __future__ import annotations

METHODS = ("direct_api", "direct_html", "seed_direct", "seed_via_reader_bypass", "secondary_citation")
DIRECT = {"direct_api", "direct_html"}

# seed sources whose raw cache was fetched via r.jina.ai past a WAF challenge (TK-12179 research/raw/{wsu,syr.md})
READER_BYPASS_SOURCES = {"seed_wichita", "seed_syracuse"}
SECONDARY_SOURCES = {"seed_catalog", "secondary_web"}
# live crawlers that talk to a documented JSON/XML API
API_SOURCES = {"loc", "dpla", "smithsonian", "internet_archive"}

BADGE_TEXT = "acquired via third-party reader past a bot challenge — pending direct verification"
FLAG_BYPASS = "acquired_via_reader_bypass"
FLAG_BYPASS_UNVERIFIED = "pending_direct_verification"
FLAG_SECONDARY_ONLY = "secondary_citation_only"


def derive(source_id: str, provenance: str | None = None) -> str:
    """Deterministic method for a cartoon_sources row from its source_id (+ provenance text as tie-breaker)."""
    sid = (source_id or "").strip()
    prov = (provenance or "").lower()
    if sid in READER_BYPASS_SOURCES or "r.jina.ai" in prov:
        return "seed_via_reader_bypass"  # never label a reader-proxied fetch as direct, whatever the source
    if sid in SECONDARY_SOURCES:
        return "secondary_citation"
    if sid.startswith("seed"):
        return "seed_direct"
    if sid in API_SOURCES or "fo=json" in prov or "/api" in prov:
        return "direct_api"
    return "direct_html"


def flags(methods) -> list[str]:
    """Item-level provenance flags from the set of acquisition methods of every source row behind a cartoon."""
    ms = {m for m in methods if m}
    out = []
    if "seed_via_reader_bypass" in ms:
        out.append(FLAG_BYPASS)
        if not ms & DIRECT:
            out.append(FLAG_BYPASS_UNVERIFIED)
    if ms and ms <= {"secondary_citation"}:
        out.append(FLAG_SECONDARY_ONLY)
    return out


def backfill(conn, overwrite: bool = False) -> int:
    """Fill acquisition_method on every row (only NULL rows unless overwrite). Idempotent. Returns rows written."""
    where = "" if overwrite else "WHERE acquisition_method IS NULL"
    rows = conn.execute(f"SELECT id, source_id, provenance FROM cartoon_sources {where}").fetchall()
    conn.executemany("UPDATE cartoon_sources SET acquisition_method=? WHERE id=?",
                     [(derive(r[1], r[2]), r[0]) for r in rows])
    conn.commit()
    return len(rows)


def methods_by_canonical(conn, ids=None) -> dict[int, list[str]]:
    """{canonical cartoon id: sorted distinct methods across it and everything merged into it}."""
    from collections import defaultdict
    sql = """SELECT COALESCE(k.merged_into,k.id), x.acquisition_method FROM cartoon_sources x
             JOIN cartoons k ON k.id=x.cartoon_id"""
    args: list = []
    if ids is not None:
        ids = list(ids)
        if not ids:
            return {}
        ph = ",".join("?" * len(ids))
        sql += f" WHERE COALESCE(k.merged_into,k.id) IN ({ph})"
        args = ids
    out: dict[int, set] = defaultdict(set)
    for canon, m in conn.execute(sql, args):
        if m:
            out[canon].add(m)
    return {k: sorted(v) for k, v in out.items()}