research/validate.py

#!/usr/bin/env python3
"""Validate topic datasets against their JSON Schemas.

The research center keeps machine-readable fact-caches alongside the markdown in
each ``md/<topic>/`` folder. This script checks them so an AI or a generation
script can edit data and confirm it is still well-formed in one step.

Convention (the only wiring needed per topic):

    md/<topic>/<name>.schema.json   validates its sibling
    md/<topic>/<name>.json

Drop a ``*.schema.json`` next to a ``*.json`` and it is picked up automatically.

Two layers of checking:

1. **Schema** — full JSON Schema validation (structure, types, enums, required
   fields, ``minItems`` etc.).
2. **House rules** — cross-field invariants JSON Schema can't easily express,
   applied generically to any list-of-records that uses the conventional field
   names:
     * ``source_count`` must equal ``len(sources)`` when both are present.
     * every ``cross_refs`` entry must reference an ``id`` that exists in the
       same dataset.
     * ``id`` values must be unique within a dataset.

Usage:
    ./.venv/bin/python validate.py              # every topic under md/
    ./.venv/bin/python validate.py demonology   # one topic
    ./.venv/bin/python validate.py md/demonology/demons.json   # one file

Exit code is non-zero if anything fails, so it works as a pre-commit hook or CI
step.
"""

from __future__ import annotations

import json
import sys
from pathlib import Path

try:
    import jsonschema
except ImportError:
    sys.exit(
        "jsonschema not installed. Run: ./.venv/bin/pip install -r requirements.txt"
    )

REPO = Path(__file__).resolve().parent
MD = REPO / "md"
SCHEMA_SUFFIX = ".schema.json"


def find_pairs(target: str | None) -> list[tuple[Path, Path]]:
    """Return (schema, data) path pairs to validate.

    ``target`` may be a topic name, a path to a topic dir, a path to a data
    file, or None (whole repo).
    """
    schemas: list[Path]
    if target is None:
        schemas = sorted(MD.glob(f"**/*{SCHEMA_SUFFIX}"))
    else:
        p = Path(target)
        if p.is_file() and p.name.endswith(SCHEMA_SUFFIX):
            schemas = [p]
        elif p.is_file() and p.suffix == ".json":
            schemas = [p.with_name(p.name[: -len(".json")] + SCHEMA_SUFFIX)]
        else:
            # treat as topic name or topic dir
            topic_dir = p if p.is_dir() else MD / target
            if not topic_dir.is_dir():
                sys.exit(f"No such topic or path: {target!r} (looked in {topic_dir})")
            schemas = sorted(topic_dir.glob(f"*{SCHEMA_SUFFIX}"))

    pairs = []
    for schema in schemas:
        data = schema.with_name(schema.name[: -len(SCHEMA_SUFFIX)] + ".json")
        pairs.append((schema, data))
    return pairs


def iter_records(obj):
    """Yield every dict that looks like a record (has an 'id') anywhere in obj."""
    if isinstance(obj, dict):
        if "id" in obj:
            yield obj
        for v in obj.values():
            yield from iter_records(v)
    elif isinstance(obj, list):
        for v in obj:
            yield from iter_records(v)


def house_rules(data) -> list[str]:
    """Cross-field invariants JSON Schema can't express. Returns error strings."""
    errors = []
    records = list(iter_records(data))
    ids = [r["id"] for r in records]
    id_set = set(ids)

    for rid in ids:
        if ids.count(rid) > 1:
            errors.append(f"duplicate id: {rid!r}")
    # dedupe the duplicate-id messages
    errors = sorted(set(errors))

    for r in records:
        rid = r.get("id", "<no id>")
        if "source_count" in r and "sources" in r:
            if r["source_count"] != len(r["sources"]):
                errors.append(
                    f"{rid}: source_count {r['source_count']} != len(sources) {len(r['sources'])}"
                )
        for ref in r.get("cross_refs", []):
            if ref not in id_set:
                errors.append(f"{rid}: cross_ref -> missing id {ref!r}")
    return errors


def validate_pair(schema_path: Path, data_path: Path) -> list[str]:
    errors = []
    if not data_path.exists():
        return [f"data file missing for schema: {data_path}"]
    try:
        schema = json.loads(schema_path.read_text())
    except json.JSONDecodeError as e:
        return [f"schema is not valid JSON: {e}"]
    try:
        data = json.loads(data_path.read_text())
    except json.JSONDecodeError as e:
        return [f"data is not valid JSON: {e}"]

    validator = jsonschema.Draft202012Validator(schema)
    for err in sorted(validator.iter_errors(data), key=lambda e: list(e.path)):
        loc = "/".join(str(p) for p in err.path) or "<root>"
        errors.append(f"schema [{loc}]: {err.message}")

    errors.extend(house_rules(data))
    return errors


def main() -> int:
    target = sys.argv[1] if len(sys.argv) > 1 else None
    pairs = find_pairs(target)
    if not pairs:
        print("No *.schema.json files found — nothing to validate.")
        return 0

    failed = 0
    for schema_path, data_path in pairs:
        rel = data_path.relative_to(REPO) if data_path.is_relative_to(REPO) else data_path
        errors = validate_pair(schema_path, data_path)
        if errors:
            failed += 1
            print(f"FAIL  {rel}")
            for e in errors:
                print(f"      - {e}")
        else:
            n = len(list(iter_records(json.loads(data_path.read_text()))))
            print(f"OK    {rel}  ({n} records)")

    print()
    total = len(pairs)
    if failed:
        print(f"{failed}/{total} dataset(s) failed validation.")
        return 1
    print(f"All {total} dataset(s) valid.")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())