jpn_word_trainer_data/scripts/deck_registry.py

#!/usr/bin/env python3
"""Shared helpers for the Japanese Word Trainer 0.3 deck registry."""

from __future__ import annotations

import hashlib
import json
import re
import zipfile
from pathlib import Path
from typing import Any


ROOT = Path(__file__).resolve().parents[1]
PACKAGE_FILES = ("manifest.json", "words.json", "refs.json")
STABLE_DECK_RE = re.compile(r"^[a-z0-9][a-z0-9_-]*:deck:[a-z0-9][a-z0-9_-]*$")
STABLE_WORD_RE = re.compile(r"^[a-z0-9][a-z0-9_-]*:word:[a-z0-9][a-z0-9_-]*$")

MANIFEST_REQUIRED = {
    "schemaVersion",
    "stableDeckKey",
    "deckKind",
    "title",
    "description",
    "versionCode",
    "createdAt",
    "updatedAt",
}
WORD_REQUIRED = {
    "stableWordKey",
    "readingJa",
    "readingKo",
    "partOfSpeech",
    "grammar",
    "kanji",
    "meaningJa",
    "meaningKo",
    "exampleJa",
    "exampleKo",
    "tag",
    "note",
    "contentHash",
}
REF_REQUIRED = {"stableWordKey", "displayOrder"}
REGISTRY_DECK_REQUIRED = {
    "stableDeckKey",
    "deckKind",
    "title",
    "description",
    "versionCode",
    "wordCount",
    "packageUrl",
    "sha256",
}


class ValidationError(Exception):
    """Raised when registry or package data does not match the 0.3 contract."""


def load_json(path: Path) -> Any:
    with path.open("r", encoding="utf-8") as handle:
        return json.load(handle)


def dump_json(path: Path, data: Any) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    text = json.dumps(data, ensure_ascii=False, indent=2) + "\n"
    path.write_text(text, encoding="utf-8")


def json_bytes(data: Any) -> bytes:
    return (json.dumps(data, ensure_ascii=False, indent=2) + "\n").encode("utf-8")


def canonical_bytes(data: Any) -> bytes:
    return json.dumps(
        data,
        ensure_ascii=False,
        sort_keys=True,
        separators=(",", ":"),
    ).encode("utf-8")


def sha256_file(path: Path) -> str:
    digest = hashlib.sha256()
    with path.open("rb") as handle:
        for chunk in iter(lambda: handle.read(1024 * 1024), b""):
            digest.update(chunk)
    return digest.hexdigest()


def stable_slug(stable_deck_key: str) -> str:
    return stable_deck_key.split(":deck:", 1)[1]


def package_url_for(manifest: dict[str, Any]) -> str:
    return f"packages/{stable_slug(manifest['stableDeckKey'])}-v{manifest['versionCode']}.zip"


def word_content_hash(word: dict[str, Any]) -> str:
    payload = {key: value for key, value in word.items() if key != "contentHash"}
    digest = hashlib.sha256(b"word-v1:" + canonical_bytes(payload)).hexdigest()
    return f"sha256:{digest}"


def source_dirs(root: Path = ROOT) -> list[Path]:
    manifests = sorted((root / "sources").glob("*/*/manifest.json"))
    return [manifest.parent for manifest in manifests]


def load_source(source_dir: Path) -> tuple[dict[str, Any], list[dict[str, Any]], list[dict[str, Any]]]:
    return (
        load_json(source_dir / "manifest.json"),
        load_json(source_dir / "words.json"),
        load_json(source_dir / "refs.json"),
    )


def materialize_words(words: list[dict[str, Any]]) -> list[dict[str, Any]]:
    materialized: list[dict[str, Any]] = []
    for word in words:
        next_word = dict(word)
        next_word["contentHash"] = word_content_hash(next_word)
        materialized.append(next_word)
    return materialized


def write_package(
    package_path: Path,
    manifest: dict[str, Any],
    words: list[dict[str, Any]],
    refs: list[dict[str, Any]],
) -> None:
    package_path.parent.mkdir(parents=True, exist_ok=True)
    payloads = {
        "manifest.json": json_bytes(manifest),
        "words.json": json_bytes(words),
        "refs.json": json_bytes(refs),
    }
    with zipfile.ZipFile(package_path, "w", compression=zipfile.ZIP_DEFLATED) as package:
        for name in PACKAGE_FILES:
            info = zipfile.ZipInfo(name)
            info.date_time = (1980, 1, 1, 0, 0, 0)
            info.compress_type = zipfile.ZIP_DEFLATED
            info.external_attr = 0o644 << 16
            package.writestr(info, payloads[name])


def validate_source(
    source_dir: Path,
    manifest: dict[str, Any],
    words: list[dict[str, Any]],
    refs: list[dict[str, Any]],
) -> list[str]:
    errors: list[str] = []
    context = str(source_dir.relative_to(ROOT)) if source_dir.is_relative_to(ROOT) else str(source_dir)

    missing_manifest = MANIFEST_REQUIRED - set(manifest)
    if missing_manifest:
        errors.append(f"{context}/manifest.json missing fields: {sorted(missing_manifest)}")
    if manifest.get("schemaVersion") != 1:
        errors.append(f"{context}/manifest.json schemaVersion must be 1")
    if manifest.get("deckKind") not in {"OFFICIAL", "CUSTOM"}:
        errors.append(f"{context}/manifest.json deckKind must be OFFICIAL or CUSTOM")
    if not STABLE_DECK_RE.match(str(manifest.get("stableDeckKey", ""))):
        errors.append(f"{context}/manifest.json stableDeckKey has invalid format")
    if not isinstance(manifest.get("versionCode"), int) or manifest.get("versionCode", 0) < 1:
        errors.append(f"{context}/manifest.json versionCode must be a positive integer")

    if not isinstance(words, list):
        errors.append(f"{context}/words.json must be an array")
        words = []
    if not isinstance(refs, list):
        errors.append(f"{context}/refs.json must be an array")
        refs = []

    word_keys: set[str] = set()
    for index, word in enumerate(words):
        missing_word = WORD_REQUIRED - set(word)
        if missing_word:
            errors.append(f"{context}/words.json[{index}] missing fields: {sorted(missing_word)}")
        stable_word_key = str(word.get("stableWordKey", ""))
        if not STABLE_WORD_RE.match(stable_word_key):
            errors.append(f"{context}/words.json[{index}] stableWordKey has invalid format")
        if stable_word_key in word_keys:
            errors.append(f"{context}/words.json duplicates {stable_word_key}")
        word_keys.add(stable_word_key)
        expected_hash = word_content_hash(word)
        actual_hash = word.get("contentHash")
        if actual_hash not in {"", expected_hash}:
            errors.append(
                f"{context}/words.json[{index}] contentHash is {actual_hash!r}; expected {expected_hash!r}"
            )

    ref_orders: list[int] = []
    ref_keys: set[str] = set()
    for index, ref in enumerate(refs):
        missing_ref = REF_REQUIRED - set(ref)
        if missing_ref:
            errors.append(f"{context}/refs.json[{index}] missing fields: {sorted(missing_ref)}")
        stable_word_key = str(ref.get("stableWordKey", ""))
        if not STABLE_WORD_RE.match(stable_word_key):
            errors.append(f"{context}/refs.json[{index}] stableWordKey has invalid format")
        if stable_word_key in ref_keys:
            errors.append(f"{context}/refs.json duplicates {stable_word_key}")
        ref_keys.add(stable_word_key)
        display_order = ref.get("displayOrder")
        if not isinstance(display_order, int):
            errors.append(f"{context}/refs.json[{index}] displayOrder must be an integer")
        else:
            ref_orders.append(display_order)
    if sorted(ref_orders) != list(range(len(ref_orders))):
        errors.append(f"{context}/refs.json displayOrder must be contiguous from 0")

    return errors


def validate_registry(root: Path = ROOT) -> list[str]:
    errors: list[str] = []
    registry_path = root / "registry.json"
    registry = load_json(registry_path)

    for field in ("schemaVersion", "registryId", "registryName", "trustLevel", "generatedAt", "decks"):
        if field not in registry:
            errors.append(f"registry.json missing field: {field}")
    if registry.get("schemaVersion") != 1:
        errors.append("registry.json schemaVersion must be 1")
    if registry.get("trustLevel") not in {"OFFICIAL", "COMMUNITY"}:
        errors.append("registry.json trustLevel must be OFFICIAL or COMMUNITY")
    if not isinstance(registry.get("decks"), list):
        errors.append("registry.json decks must be an array")
        return errors

    source_by_deck: dict[str, tuple[dict[str, Any], list[dict[str, Any]], list[dict[str, Any]]]] = {}
    for source_dir in source_dirs(root):
        manifest, words, refs = load_source(source_dir)
        errors.extend(validate_source(source_dir, manifest, words, refs))
        source_by_deck[manifest.get("stableDeckKey", "")] = (manifest, words, refs)

    seen_decks: set[str] = set()
    for index, deck in enumerate(registry["decks"]):
        missing_deck = REGISTRY_DECK_REQUIRED - set(deck)
        if missing_deck:
            errors.append(f"registry.json decks[{index}] missing fields: {sorted(missing_deck)}")
        stable_deck_key = str(deck.get("stableDeckKey", ""))
        if stable_deck_key in seen_decks:
            errors.append(f"registry.json duplicates {stable_deck_key}")
        seen_decks.add(stable_deck_key)
        if stable_deck_key not in source_by_deck:
            errors.append(f"registry.json decks[{index}] has no source directory for {stable_deck_key}")
            continue

        manifest, words, refs = source_by_deck[stable_deck_key]
        materialized_words = materialize_words(words)
        expected_url = package_url_for(manifest)
        package_path = root / expected_url
        if deck.get("packageUrl") != expected_url:
            errors.append(f"registry.json decks[{index}] packageUrl should be {expected_url}")
        if deck.get("deckKind") != manifest.get("deckKind"):
            errors.append(f"registry.json decks[{index}] deckKind does not match manifest")
        if deck.get("versionCode") != manifest.get("versionCode"):
            errors.append(f"registry.json decks[{index}] versionCode does not match manifest")
        if deck.get("wordCount") != len(refs):
            errors.append(f"registry.json decks[{index}] wordCount should be {len(refs)}")
        if not package_path.exists():
            errors.append(f"{expected_url} does not exist")
            continue
        sha256 = sha256_file(package_path)
        if deck.get("sha256") != sha256:
            errors.append(f"registry.json decks[{index}] sha256 should be {sha256}")

        with zipfile.ZipFile(package_path, "r") as package:
            names = package.namelist()
            if names != list(PACKAGE_FILES):
                errors.append(f"{expected_url} must contain only {list(PACKAGE_FILES)} in that order")
            package_manifest = json.loads(package.read("manifest.json"))
            package_words = json.loads(package.read("words.json"))
            package_refs = json.loads(package.read("refs.json"))
        if package_manifest != manifest:
            errors.append(f"{expected_url}/manifest.json does not match source manifest")
        if package_words != materialized_words:
            errors.append(f"{expected_url}/words.json does not match hashed source words")
        if package_refs != refs:
            errors.append(f"{expected_url}/refs.json does not match source refs")

    missing_registry_entries = set(source_by_deck) - seen_decks
    for stable_deck_key in sorted(missing_registry_entries):
        errors.append(f"source deck {stable_deck_key} is missing from registry.json")

    return errors