arborist/bench/scripts/textbooks_manifest.py

"""Read the textbook manifest and emit URLs / license / metadata helpers.

Usage:
    python -m bench.scripts.textbooks_manifest urls < manifest.jsonl
        → one URL per line on stdout

    python -m bench.scripts.textbooks_manifest summary < manifest.jsonl
        → human-readable license summary

The manifest format is one JSON record per line. The first record may be a
``_meta`` block; it is ignored by all subcommands. See
``bench/fixtures/textbooks/manifest-v1.jsonl`` for the canonical schema.

Hard discipline: every entry MUST carry ``license`` and ``license_url``.
This script refuses to emit URLs from entries missing those fields, so a
fetcher run is never the place a license-discipline bug becomes visible —
it surfaces here.
"""

from __future__ import annotations

import json
import sys
from typing import Iterator, TextIO


_REQUIRED_LICENSE_KEYS = ("license", "license_url")
# Allow-list of license tokens we redistribute under arborist's AGPLv3
# distribution profile. CC-BY-NC, CC-BY-ND, and proprietary licenses are
# explicitly rejected — they don't fit the AGPLv3 downstream-distribution
# contract or arborist's "every patch ships" mission.
_ALLOWED_LICENSES = frozenset({
    "PD",
    "CC0-1.0",
    "CC-BY-2.5", "CC-BY-3.0", "CC-BY-4.0",
    "CC-BY-SA-2.5", "CC-BY-SA-3.0", "CC-BY-SA-4.0",
    "GFDL-1.2", "GFDL-1.3",
    "AGPL-3.0-only",
    "Apache-2.0",
    "MIT",
})


def iter_entries(stream: TextIO) -> Iterator[dict]:
    """Yield non-meta entries from a JSONL manifest."""
    for line in stream:
        line = line.strip()
        if not line:
            continue
        obj = json.loads(line)
        if "_meta" in obj:
            continue
        yield obj


def _validate(entry: dict) -> None:
    """Raise ValueError if an entry would emit URLs under a license outside
    the redistribution allow-list, OR is missing required license fields.

    Entries with both ``urls`` and ``crawl_url`` empty/missing are
    placeholder rows kept for citation traceability (e.g., Mendelson +
    Enderton as yellow-light proprietary; PG TeX-only Hilbert + Boole
    awaiting PDF source support; Łukasiewicz + Kolmogorov pending license
    analysis). Validation is fail-closed at the URL-emit step, so
    placeholders with no URLs are allowed any license token without
    raising.
    """
    has_emit_target = bool(entry.get("urls") or entry.get("crawl_url"))
    for k in _REQUIRED_LICENSE_KEYS:
        if k not in entry or not entry[k]:
            raise ValueError(
                f"manifest entry {entry.get('id', '?')!r} missing required "
                f"field {k!r}"
            )
    if has_emit_target and entry["license"] not in _ALLOWED_LICENSES:
        raise ValueError(
            f"manifest entry {entry.get('id', '?')!r} declares license "
            f"{entry['license']!r} outside arborist redistribution allow-list "
            f"({sorted(_ALLOWED_LICENSES)}); urls / crawl_url must be empty "
            f"for non-allow-listed entries"
        )


def cmd_urls(stream: TextIO) -> int:
    seen: set[str] = set()
    for entry in iter_entries(stream):
        _validate(entry)
        for url in entry.get("urls", []):
            if url and url not in seen:
                seen.add(url)
                print(url)
    return 0


def cmd_summary(stream: TextIO) -> int:
    rows = []
    by_license: dict[str, int] = {}
    by_domain: dict[str, int] = {}
    total_urls = 0
    for entry in iter_entries(stream):
        _validate(entry)
        rows.append(entry)
        by_license[entry["license"]] = by_license.get(entry["license"], 0) + 1
        d = entry.get("domain", "?")
        by_domain[d] = by_domain.get(d, 0) + 1
        total_urls += len(entry.get("urls", []))

    print(f"# Textbook manifest summary")
    print(f"entries: {len(rows)}")
    print(f"seed URLs: {total_urls}")
    print()
    print("## by license")
    for k, n in sorted(by_license.items()):
        print(f"  {k:20s} {n}")
    print()
    print("## by domain")
    for k, n in sorted(by_domain.items()):
        print(f"  {k:20s} {n}")
    print()
    print("## entries")
    for e in rows:
        urls_n = len(e.get("urls", []))
        title = e.get("title", "?")
        license_ = e.get("license", "?")
        print(f"  {e.get('id'):42s}  {license_:18s}  {urls_n} URL(s)  {title}")
    return 0


def cmd_crawl_targets(stream: TextIO) -> int:
    """Emit one tab-separated `crawl_url\\tdepth\\tmax\\tid` row per entry
    that declares a `crawl_url`. Used by `make crawl-textbooks` to drive
    the BFS crawler, one shard per textbook id.
    """
    for entry in iter_entries(stream):
        _validate(entry)
        crawl_url = entry.get("crawl_url")
        if not crawl_url:
            continue
        depth = int(entry.get("crawl_depth", 2))
        max_pages = int(entry.get("crawl_max", 80))
        print(f"{crawl_url}\t{depth}\t{max_pages}\t{entry.get('id', '?')}")
    return 0


def cmd_ids(stream: TextIO) -> int:
    """Emit one entry id per line for entries that have either `urls` or
    `crawl_url`. License-fail placeholders (no fetchable URLs) are
    excluded — they have nothing to ingest.
    """
    for entry in iter_entries(stream):
        try:
            _validate(entry)
        except ValueError:
            continue
        if entry.get("urls") or entry.get("crawl_url"):
            print(entry.get("id", ""))
    return 0


def cmd_tex_targets(stream: TextIO) -> int:
    """Emit one tab-separated `tex_url\\tid` row per entry that
    declares a `tex_url`. Used by `make textbooks-tex` to pull PG
    LaTeX-source textbooks (e.g. Hilbert #17384, Boole #15114) into
    per-book shards via `arborist ingest --source textbook_tex`.
    """
    for entry in iter_entries(stream):
        # tex_url path is exempt from the redistribution allow-list
        # because PG TeX sources for pre-1929 works are PD by age;
        # the manifest entry's `license: "PD"` field carries the
        # check anyway. Validate license fields nonetheless.
        for k in _REQUIRED_LICENSE_KEYS:
            if k not in entry or not entry[k]:
                continue
        tex_url = entry.get("tex_url")
        if not tex_url:
            continue
        print(f"{tex_url}\t{entry.get('id', '?')}")
    return 0


def cmd_lookup(stream: TextIO) -> int:
    """Read entry id from sys.argv[2]; emit tab-separated
    `id\\tcrawl_url\\tdepth\\tmax\\tlicense\\tdomain` for the matching
    entry. Used by per-book make targets to look up one textbook's
    crawl parameters from the manifest without grep wizardry.
    """
    target_id = sys.argv[2] if len(sys.argv) > 2 else ""
    if not target_id:
        print("usage: lookup <id>", file=sys.stderr)
        return 2
    for entry in iter_entries(stream):
        if entry.get("id") != target_id:
            continue
        try:
            _validate(entry)
        except ValueError as exc:
            print(f"# license-fail placeholder: {exc}", file=sys.stderr)
            return 3
        crawl_url = entry.get("crawl_url", "")
        depth = entry.get("crawl_depth", 2)
        max_pages = entry.get("crawl_max", 80)
        print(
            f"{entry.get('id')}\t{crawl_url}\t{depth}\t{max_pages}\t"
            f"{entry.get('license')}\t{entry.get('domain')}"
        )
        return 0
    print(f"# id {target_id!r} not found in manifest", file=sys.stderr)
    return 4


_COMMANDS = {
    "urls": cmd_urls,
    "summary": cmd_summary,
    "crawl-targets": cmd_crawl_targets,
    "tex-targets": cmd_tex_targets,
    "ids": cmd_ids,
    "lookup": cmd_lookup,
}


def main(argv: list[str]) -> int:
    if len(argv) < 2 or argv[1] not in _COMMANDS:
        print(
            f"usage: {argv[0]} {{{'|'.join(sorted(_COMMANDS))}}} < manifest.jsonl",
            file=sys.stderr,
        )
        return 2
    return _COMMANDS[argv[1]](sys.stdin)


if __name__ == "__main__":
    sys.exit(main(sys.argv))