generate_release_artifacts.py

#!/usr/bin/env python3
"""Generate deterministic release artefacts from data/index/*.jsonl.

Emits three files at the repo root:

  - NOTICE.md         human-readable attribution roll-up.
  - CITATION.cff      Citation File Format 1.2.0.
  - datapackage.json  Frictionless Data Package manifest.

The script is fully deterministic: same indexes + recipe in,
byte-identical files out. No datetime.now(), no random ordering, no
UUIDs.

Two timestamps with deliberately different semantics:

  - `datapackage.json::released_at` = max(extraction.extracted_at)
    across entries — the *corpus-state timestamp*. Bumps on every
    ingest PR. When the corpus is empty it falls back to
    `release_recipe.json::initial_release_date`.

  - `CITATION.cff::date-released` = `release_recipe.json::version_released_date`
    — the date this `version` was released. Stable per release. Bumped
    manually on `version` bump (see docs/release_process.md). This is
    what citations should be reproducible against.

Use `--check` to verify the on-disk artefacts match what would be
generated without touching the tree.
"""

from __future__ import annotations

import argparse
import json
import sys
from collections import Counter
from pathlib import Path
from typing import Any

try:
    import yaml
except ImportError as exc:  # pragma: no cover - exercised when deps are absent.
    raise SystemExit(
        "Missing dependency: PyYAML. Install development dependencies with "
        "`python3 -m pip install -r requirements-dev.txt`."
    ) from exc


REPO_ROOT = Path(__file__).resolve().parents[1]
WRITERS_PATH = REPO_ROOT / "data" / "index" / "writers.jsonl"
ENTRIES_PATH = REPO_ROOT / "data" / "index" / "entries.jsonl"
RECIPE_PATH = REPO_ROOT / "scripts" / "release_recipe.json"
NOTICE_PATH = REPO_ROOT / "NOTICE.md"
CITATION_PATH = REPO_ROOT / "CITATION.cff"
DATAPACKAGE_PATH = REPO_ROOT / "datapackage.json"

# Licenses whose terms require attribution. Drives both NOTICE.md
# inclusion and the consistency check below. Keep in sync with
# scripts/validate_indexes.py::LICENSE_BASIS_MAP and the inheritance
# table in docs/dataset_structure.md.
ATTRIBUTION_REQUIRING_LICENSES: frozenset[str] = frozenset({
    "CC-BY-4.0",
    "CC-BY-SA-4.0",
})


def _load_jsonl(path: Path) -> list[dict[str, Any]]:
    if not path.exists():
        raise SystemExit(f"{path}: file does not exist")
    rows: list[dict[str, Any]] = []
    with path.open("r", encoding="utf-8") as handle:
        for line_number, line in enumerate(handle, start=1):
            stripped = line.strip()
            if not stripped:
                continue
            try:
                rows.append(json.loads(stripped))
            except json.JSONDecodeError as exc:
                raise SystemExit(f"{path}:{line_number}: invalid JSON: {exc}") from exc
    return rows


def _load_recipe(path: Path) -> dict[str, Any]:
    if not path.exists():
        raise SystemExit(f"{path}: file does not exist")
    try:
        return json.loads(path.read_text(encoding="utf-8"))
    except json.JSONDecodeError as exc:
        raise SystemExit(f"{path}: invalid JSON: {exc}") from exc


def _derive_released_at(entries: list[dict[str, Any]], recipe: dict[str, Any]) -> str:
    """Corpus-state timestamp for datapackage.json. Bumps every ingest."""
    extracted = [
        entry["extraction"]["extracted_at"]
        for entry in entries
        if entry.get("extraction") and entry["extraction"].get("extracted_at")
    ]
    if extracted:
        return max(extracted)
    # Empty corpus: initial-setup state. Fall back to the recipe's
    # `initial_release_date` so generation is deterministic.
    initial = recipe.get("initial_release_date")
    if not isinstance(initial, str) or not initial:
        raise SystemExit(
            "no extraction.extracted_at values found in entries.jsonl, and "
            "release_recipe.json has no initial_release_date fallback"
        )
    return initial


def _resolve_citation_date(recipe: dict[str, Any]) -> str:
    """Citation date for CITATION.cff. Stable per version."""
    date = recipe.get("version_released_date")
    if not isinstance(date, str) or not date:
        raise SystemExit(
            "release_recipe.json::version_released_date is missing; this is "
            "the stable per-version release date used by CITATION.cff. Set it "
            "when you bump `version` (see docs/release_process.md)."
        )
    return date


def _license_breakdown(entries: list[dict[str, Any]]) -> dict[str, int]:
    counts = Counter(entry["rights"]["license_expression"] for entry in entries)
    return {key: counts[key] for key in sorted(counts, key=lambda k: (k is None, k))}


def _writer_breakdown(entries: list[dict[str, Any]]) -> dict[str, int]:
    counts = Counter(entry["writer_id"] for entry in entries)
    return {key: counts[key] for key in sorted(counts)}


def _letter_breakdown(entries: list[dict[str, Any]]) -> dict[str, int]:
    counts = Counter(entry["letter"]["name"] for entry in entries)
    return {key: counts[key] for key in sorted(counts)}


def _image_byte_count(entries: list[dict[str, Any]]) -> int:
    total = 0
    for entry in entries:
        byte_size = entry["image"].get("bytes")
        if isinstance(byte_size, int):
            total += byte_size
    return total


def _check_attribution_consistency(entries: list[dict[str, Any]]) -> None:
    # Any entry whose license demands attribution must carry the flag,
    # text, and url. The schema enforces text+url *given* the flag; this
    # layer catches the prior failure mode of "license is CC-BY-SA but
    # ingester forgot the flag", which would silently drop the entry
    # from NOTICE.md.
    for entry in entries:
        rights = entry["rights"]
        license_expr = rights.get("license_expression")
        if license_expr in ATTRIBUTION_REQUIRING_LICENSES:
            if rights.get("attribution_required") is not True:
                raise SystemExit(
                    f"{entry['entry_id']}: license {license_expr} requires "
                    f"rights.attribution_required: true (found "
                    f"{rights.get('attribution_required')!r})"
                )
            for field in ("attribution_text", "attribution_url"):
                value = rights.get(field)
                if not isinstance(value, str) or not value.strip():
                    raise SystemExit(
                        f"{entry['entry_id']}: license {license_expr} requires "
                        f"rights.{field}, but it is null, blank, or "
                        f"whitespace-only"
                    )


def _attribution_entries(entries: list[dict[str, Any]]) -> list[dict[str, Any]]:
    selected = [
        entry
        for entry in entries
        if entry["rights"].get("license_expression") in ATTRIBUTION_REQUIRING_LICENSES
    ]
    return sorted(selected, key=lambda entry: entry["entry_id"])


def _notice_stanza(
    entry: dict[str, Any],
    recipe: dict[str, Any],
    upstream_repo_url: str,
) -> str:
    license_names: dict[str, str] = recipe["license_names"]
    license_urls: dict[str, str] = recipe["license_urls"]
    rights = entry["rights"]
    license_expr = rights["license_expression"]
    license_name = license_names.get(license_expr, license_expr)
    license_url = license_urls.get(license_expr)

    if license_url:
        license_line = f"- License: [{license_name} ({license_expr})]({license_url})"
    else:
        license_line = f"- License: {license_name} ({license_expr})"

    letter = entry["letter"]
    title = (
        f"{letter['unicode_char']} ({letter['name']}, {letter['form']}) "
        f"by writer `{entry['writer_id']}`"
    )

    upstream = entry["upstream"]
    # `upstream.commit` is enforced as a 40-char SHA by the schema; this
    # URL form always resolves on GitHub. Never use `release_tag` here —
    # tags are mutable and the link must outlive tag re-pointing.
    upstream_link = (
        f"{upstream_repo_url}/blob/{upstream['commit']}/data/index/entries.jsonl"
    )

    lines = [
        f"### {title}",
        "",
        f"- Entry: `{entry['entry_id']}`",
        license_line,
        f"- Licensor: {rights['attribution_text']}",
        f"- Source page: <{rights['attribution_url']}>",
        f"- Upstream scan entry: `{upstream['entry_id']}` "
        f"(<{upstream_link}>)",
    ]
    return "\n".join(lines)


NOTICE_TEMPLATE = """\
# NOTICE

This file is generated by `scripts/generate_release_artifacts.py` from \
`data/index/entries.jsonl`. Do not edit by hand.

Repository-authored metadata is dedicated to the public domain under \
CC0 1.0 Universal. See [`LICENSE`](LICENSE) and [`LICENSE.md`](LICENSE.md) \
for the full compound-licensing policy.

Per-letter image crops are derivatives of upstream scans in \
[HeOCR/hash]({upstream_repo_url}) and \
carry per-entry rights inherited from the source page. The entries \
listed below carry a license that requires attribution (currently \
{license_set}). Anyone redistributing or reusing these crops must keep \
the listed credit and link to the source page on which the rights claim \
was verified.

- Corpus release: `{version}`
- Released at (corpus state): `{released_at}`

## Attribution-required entries

{stanzas}

## Full per-entry rights

Every entry, attribution-required or not, ships with its rights record in \
[`data/index/entries.jsonl`](data/index/entries.jsonl). Consumers that \
need machine-readable rights metadata should read that file directly; the \
manifest at [`datapackage.json`](datapackage.json) summarises the license \
breakdown.
"""


def build_notice(
    entries: list[dict[str, Any]],
    recipe: dict[str, Any],
    released_at: str,
    upstream_repo_url: str,
) -> str:
    required = _attribution_entries(entries)
    if required:
        stanzas = "\n\n".join(
            _notice_stanza(entry, recipe, upstream_repo_url) for entry in required
        )
    else:
        stanzas = "_No entries in this release require attribution._"

    license_set = ", ".join(sorted(ATTRIBUTION_REQUIRING_LICENSES))
    return NOTICE_TEMPLATE.format(
        license_set=license_set,
        version=recipe["version"],
        released_at=released_at,
        stanzas=stanzas,
        upstream_repo_url=upstream_repo_url,
    )


def build_citation(
    entries: list[dict[str, Any]],
    writers: list[dict[str, Any]],
    recipe: dict[str, Any],
    citation_date: str,
) -> str:
    license_counts = _license_breakdown(entries)
    if license_counts:
        breakdown_summary = ", ".join(
            f"{count} {license_id}" for license_id, count in license_counts.items()
        )
        entry_writer_count = len({entry["writer_id"] for entry in entries})
        abstract = (
            f"{recipe['description']} Release {recipe['version']} contains "
            f"{len(entries)} per-letter image entries drawn from "
            f"{entry_writer_count} verified writers ({breakdown_summary})."
        )
    else:
        abstract = (
            f"{recipe['description']} Release {recipe['version']} is the "
            f"initial-setup release: the corpus contains no per-letter image "
            f"entries yet. The repository ships the schemas, validation "
            f"tooling, CI, and licensing policy needed to start ingesting."
        )

    document: dict[str, Any] = {
        "cff-version": "1.2.0",
        "message": "Please cite this dataset using the metadata below.",
        "type": "dataset",
        "title": recipe["title"],
        "abstract": abstract,
        "authors": [{"name": author["name"]} for author in recipe["authors"]],
        "version": recipe["version"],
        "date-released": citation_date,
        "repository-code": recipe["repository_code"],
        "url": recipe["homepage"],
        "license": recipe["metadata_license"]["spdx"],
        "keywords": sorted(recipe["keywords"]),
    }
    identifiers = recipe.get("citation_identifiers") or []
    if identifiers:
        document["identifiers"] = identifiers

    header = "# Generated by scripts/generate_release_artifacts.py. Do not edit by hand.\n"
    body = yaml.safe_dump(
        document,
        default_flow_style=False,
        sort_keys=False,
        allow_unicode=True,
        width=10_000,
    )
    return header + body


def build_datapackage(
    entries: list[dict[str, Any]],
    writers: list[dict[str, Any]],
    recipe: dict[str, Any],
    released_at: str,
    citation_date: str,
    entries_path: Path,
    writers_path: Path,
) -> dict[str, Any]:
    license_names: dict[str, str] = recipe["license_names"]
    license_urls: dict[str, str] = recipe["license_urls"]
    license_counts = _license_breakdown(entries)
    writer_status_counts = Counter(writer.get("status") for writer in writers)
    writer_status_breakdown = {
        key: writer_status_counts[key]
        for key in sorted(writer_status_counts)
        if key is not None
    }

    license_listings: list[dict[str, Any]] = []
    license_listings.append({
        "name": recipe["metadata_license"]["spdx"],
        "path": recipe["metadata_license"]["url"],
        "title": license_names.get(
            recipe["metadata_license"]["spdx"], recipe["metadata_license"]["spdx"]
        ),
        "scope": "metadata",
    })
    for license_id in sorted(k for k in license_counts if k is not None):
        listing: dict[str, Any] = {
            "name": license_id,
            "title": license_names.get(license_id, license_id),
            "scope": "images",
        }
        url = license_urls.get(license_id)
        if url:
            listing["path"] = url
        license_listings.append(listing)

    resource_path_for: dict[str, Path] = {
        "entries": entries_path,
        "writers": writers_path,
    }
    resource_records_for: dict[str, int] = {
        "entries": len(entries),
        "writers": len(writers),
    }

    resources: list[dict[str, Any]] = []
    for name in sorted(recipe["resources"]):
        spec = recipe["resources"][name]
        # Note: no `schema` field. Frictionless reserves
        # `resources[].schema` for Table Schema (column definitions), but
        # our data is nested JSON validated against JSON Schema. We
        # expose the JSON Schema URLs via the top-level `schemas` block
        # as a custom extension instead.
        resources.append({
            "name": name,
            "path": spec["path"],
            "profile": "data-resource",
            "format": spec["format"],
            "mediatype": spec["mediatype"],
            "encoding": spec["encoding"],
            "description": spec["description"],
            "record_count": resource_records_for[name],
            "bytes": resource_path_for[name].stat().st_size,
        })

    return {
        "profile": "data-package",
        "name": recipe["name"],
        "title": recipe["title"],
        "description": recipe["description"],
        "version": recipe["version"],
        "version_released_date": citation_date,
        "released_at": released_at,
        "homepage": recipe["homepage"],
        "upstream_repo": recipe["upstream_repo"],
        "keywords": sorted(recipe["keywords"]),
        "contributors": [
            {"title": author["name"], "role": author.get("role", "author")}
            for author in recipe["authors"]
        ],
        "licenses": license_listings,
        "schemas": {
            "writer": recipe["schema_urls"]["writer"],
            "entry": recipe["schema_urls"]["entry"],
        },
        "stats": {
            "record_count": len(entries),
            "entry_writer_count": len({entry["writer_id"] for entry in entries}),
            "writer_record_count": len(writers),
            "writer_status_breakdown": writer_status_breakdown,
            "image_byte_count": _image_byte_count(entries),
            "attribution_required_count": len(_attribution_entries(entries)),
            "license_breakdown": license_counts,
            "letter_breakdown": _letter_breakdown(entries),
            "writer_breakdown": _writer_breakdown(entries),
        },
        "resources": resources,
    }


def _serialise_text(text: str) -> str:
    return text if text.endswith("\n") else text + "\n"


def _serialise_json(data: dict[str, Any]) -> str:
    return json.dumps(data, ensure_ascii=False, indent=2, sort_keys=True) + "\n"


def _require_recipe_fields(recipe: dict[str, Any]) -> None:
    required = [
        "name", "title", "version", "version_released_date",
        "description", "homepage", "repository_code", "upstream_repo",
        "authors", "keywords", "metadata_license",
        "license_urls", "license_names", "schema_urls", "resources",
    ]
    missing = [field for field in required if field not in recipe]
    if missing:
        raise SystemExit(
            f"release_recipe.json missing required field(s): {', '.join(missing)}"
        )


def _render(
    writers_path: Path,
    entries_path: Path,
    recipe_path: Path,
) -> dict[str, str]:
    writers = _load_jsonl(writers_path)
    entries = _load_jsonl(entries_path)
    recipe = _load_recipe(recipe_path)
    _require_recipe_fields(recipe)
    _check_attribution_consistency(entries)
    released_at = _derive_released_at(entries, recipe)
    citation_date = _resolve_citation_date(recipe)
    upstream_repo_url = recipe["upstream_repo"]

    return {
        "notice": _serialise_text(
            build_notice(entries, recipe, released_at, upstream_repo_url)
        ),
        "citation": _serialise_text(
            build_citation(entries, writers, recipe, citation_date)
        ),
        "datapackage": _serialise_json(
            build_datapackage(
                entries, writers, recipe, released_at, citation_date,
                entries_path=entries_path, writers_path=writers_path,
            )
        ),
    }


def generate(
    writers_path: Path = WRITERS_PATH,
    entries_path: Path = ENTRIES_PATH,
    recipe_path: Path = RECIPE_PATH,
    notice_path: Path = NOTICE_PATH,
    citation_path: Path = CITATION_PATH,
    datapackage_path: Path = DATAPACKAGE_PATH,
) -> dict[str, Path]:
    rendered = _render(writers_path, entries_path, recipe_path)
    notice_path.write_text(rendered["notice"], encoding="utf-8")
    citation_path.write_text(rendered["citation"], encoding="utf-8")
    datapackage_path.write_text(rendered["datapackage"], encoding="utf-8")
    return {
        "notice": notice_path,
        "citation": citation_path,
        "datapackage": datapackage_path,
    }


def check(
    writers_path: Path = WRITERS_PATH,
    entries_path: Path = ENTRIES_PATH,
    recipe_path: Path = RECIPE_PATH,
    notice_path: Path = NOTICE_PATH,
    citation_path: Path = CITATION_PATH,
    datapackage_path: Path = DATAPACKAGE_PATH,
) -> list[Path]:
    rendered = _render(writers_path, entries_path, recipe_path)
    stale: list[Path] = []
    for kind, path in (
        ("notice", notice_path),
        ("citation", citation_path),
        ("datapackage", datapackage_path),
    ):
        actual = path.read_text(encoding="utf-8") if path.exists() else ""
        if actual != rendered[kind]:
            stale.append(path)
    return stale


def main() -> None:
    parser = argparse.ArgumentParser()
    parser.add_argument("--writers", type=Path, default=WRITERS_PATH)
    parser.add_argument("--entries", type=Path, default=ENTRIES_PATH)
    parser.add_argument("--recipe", type=Path, default=RECIPE_PATH)
    parser.add_argument("--notice", type=Path, default=NOTICE_PATH)
    parser.add_argument("--citation", type=Path, default=CITATION_PATH)
    parser.add_argument("--datapackage", type=Path, default=DATAPACKAGE_PATH)
    parser.add_argument(
        "--check",
        action="store_true",
        help="Verify on-disk artefacts match what would be generated. Exit 1 if not.",
    )
    args = parser.parse_args()

    if args.check:
        stale = check(
            writers_path=args.writers,
            entries_path=args.entries,
            recipe_path=args.recipe,
            notice_path=args.notice,
            citation_path=args.citation,
            datapackage_path=args.datapackage,
        )
        if stale:
            for path in stale:
                try:
                    display = path.relative_to(REPO_ROOT)
                except ValueError:
                    display = path
                print(f"stale: {display}", file=sys.stderr)
            print(
                "Run `python3 scripts/generate_release_artifacts.py` to regenerate.",
                file=sys.stderr,
            )
            raise SystemExit(1)
        print("ok: release artefacts are up to date")
        return

    written = generate(
        writers_path=args.writers,
        entries_path=args.entries,
        recipe_path=args.recipe,
        notice_path=args.notice,
        citation_path=args.citation,
        datapackage_path=args.datapackage,
    )
    for label, path in written.items():
        try:
            display = path.relative_to(REPO_ROOT)
        except ValueError:
            display = path
        print(f"wrote {label}: {display}")


if __name__ == "__main__":
    main()