magicciv/tools/audio-validate.py

#!/usr/bin/env python3
"""Audio manifest + asset coherence validator.

Checks (per pack under `public/games/<theme>/data/audio.json`):

  1. JSON validates against the schema at
     `public/games/<theme>/data/schemas/audio.schema.json`.
  2. Every `stream` / `streams[]` path resolves to a real file under
     `public/games/<theme>/assets/audio/...`. Missing files warn (the
     runtime fails loud at playback time via `audio_asset_missing`).
  3. No orphan `.ogg` files: every `.ogg` under `assets/audio/**` is
     referenced by at least one manifest entry. Orphans warn; not fatal.
  4. `streams[]` arrays are non-empty (no silent-sentinel pattern).

Exit status: non-zero iff a structural problem (1, 4) is found.
Missing files (2) and orphans (3) only warn — they're a normal
intermediate state while assets land.

Wire: `./run validate` calls this; it also runs in CI before the GUT pass.
"""
from __future__ import annotations

import argparse
import json
import sys
from dataclasses import dataclass
from pathlib import Path

REPO = Path(__file__).resolve().parent.parent
DEFAULT_THEMES = ["age-of-dwarves"]


@dataclass
class Report:
    errors: list[str]
    warnings: list[str]

    def fail(self, msg: str) -> None:
        self.errors.append(msg)

    def warn(self, msg: str) -> None:
        self.warnings.append(msg)


# ──────────────────────────────────────────────────────────────────────────
# Schema validation (lightweight — no jsonschema dep)
# ──────────────────────────────────────────────────────────────────────────


def _entry_streams(entry: dict) -> list[str]:
    if "streams" in entry:
        raw = entry["streams"]
        if not isinstance(raw, list):
            return []
        return [str(s) for s in raw]
    single = entry.get("stream")
    if single:
        return [str(single)]
    return []


def validate_schema(manifest: dict, report: Report) -> None:
    if not isinstance(manifest, dict):
        report.fail("audio.json root is not a JSON object")
        return
    if "schema_version" not in manifest:
        report.fail("missing required field: schema_version")
    sfx = manifest.get("sfx")
    if not isinstance(sfx, dict):
        report.fail("missing or non-object `sfx` block")
        sfx = {}
    music = manifest.get("music")
    if not isinstance(music, dict):
        report.fail("missing or non-object `music` block")
        music = {"tracks": []}

    # Each SFX entry shape
    allowed_entry_keys = {
        "stream", "streams", "volume_db", "bus", "pitch_jitter",
        "description",
    }
    for key, entry in sfx.items():
        if not isinstance(entry, dict):
            report.fail(f"sfx[{key!r}] is not an object")
            continue
        extra = set(entry.keys()) - allowed_entry_keys
        if extra:
            report.fail(f"sfx[{key!r}] has unknown fields: {sorted(extra)}")
        if "stream" in entry and "streams" in entry:
            report.warn(
                f"sfx[{key!r}] declares both `stream` and `streams[]` — "
                f"streams[] takes precedence at runtime; drop `stream`"
            )
        if "streams" in entry:
            raw = entry["streams"]
            if isinstance(raw, list) and len(raw) == 0:
                report.fail(
                    f"sfx[{key!r}] declares empty `streams: []` — fail-loud "
                    f"policy forbids silent sentinels; remove the entry or "
                    f"ship at least one stream"
                )
        if "stream" not in entry and "streams" not in entry:
            report.fail(
                f"sfx[{key!r}] declares neither `stream` nor `streams[]` — "
                f"every entry must point at at least one path"
            )
        if "pitch_jitter" in entry:
            j = entry["pitch_jitter"]
            if not isinstance(j, (int, float)) or j < 0.0 or j > 0.5:
                report.fail(
                    f"sfx[{key!r}].pitch_jitter must be a number in [0.0, 0.5]"
                )

    # Music tracks
    tracks = music.get("tracks", [])
    if not isinstance(tracks, list):
        report.fail("music.tracks is not an array")
        tracks = []
    seen_ids: set[str] = set()
    for i, track in enumerate(tracks):
        if not isinstance(track, dict):
            report.fail(f"music.tracks[{i}] is not an object")
            continue
        tid = track.get("id")
        if not tid:
            report.fail(f"music.tracks[{i}] missing required `id`")
        elif tid in seen_ids:
            report.fail(f"music.tracks[{i}] duplicate id {tid!r}")
        else:
            seen_ids.add(tid)
        if not track.get("stream"):
            report.fail(f"music.tracks[{i}] ({tid}) missing required `stream`")


# ──────────────────────────────────────────────────────────────────────────
# Asset existence + orphan check
# ──────────────────────────────────────────────────────────────────────────


def collect_referenced_streams(manifest: dict) -> set[str]:
    refs: set[str] = set()
    for entry in manifest.get("sfx", {}).values():
        if not isinstance(entry, dict):
            continue
        for s in _entry_streams(entry):
            refs.add(s)
    for track in manifest.get("music", {}).get("tracks", []):
        if isinstance(track, dict) and track.get("stream"):
            refs.add(str(track["stream"]))
    return refs


def check_assets(theme: str, refs: set[str], report: Report) -> None:
    # Audio is shared across themes — lives in public/resources/audio/.
    # `theme` is retained for the warning prefix only.
    assets_root = REPO / "public" / "resources"
    missing: list[str] = []
    for rel in sorted(refs):
        if not rel:
            continue
        p = assets_root / rel
        if not p.exists():
            missing.append(rel)
    if missing:
        report.warn(
            f"{theme}: {len(missing)} referenced files do not yet exist "
            f"(asset acquisition tracked by p2-16). First few: "
            f"{missing[:3]}"
        )

    # Orphan scan
    audio_dir = assets_root / "audio"
    if not audio_dir.exists():
        return
    orphans: list[str] = []
    for f in audio_dir.rglob("*.ogg"):
        rel = str(f.relative_to(assets_root))
        if rel not in refs:
            orphans.append(rel)
    if orphans:
        report.warn(
            f"{theme}: {len(orphans)} orphan .ogg files not referenced by "
            f"audio.json. First few: {orphans[:3]}"
        )


# ──────────────────────────────────────────────────────────────────────────
# Driver
# ──────────────────────────────────────────────────────────────────────────


def validate_theme(theme: str) -> Report:
    report = Report(errors=[], warnings=[])
    manifest_path = REPO / "public" / "games" / theme / "data" / "audio.json"
    if not manifest_path.exists():
        report.fail(f"{theme}: audio.json not found at {manifest_path}")
        return report
    try:
        manifest = json.loads(manifest_path.read_text())
    except json.JSONDecodeError as e:
        report.fail(f"{theme}: JSON parse error: {e}")
        return report

    validate_schema(manifest, report)
    refs = collect_referenced_streams(manifest)
    check_assets(theme, refs, report)
    return report


def main() -> int:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument(
        "--theme", action="append", default=None,
        help="Theme id (e.g. 'age-of-dwarves'). Repeatable. "
             "Default: every theme listed in DEFAULT_THEMES.",
    )
    parser.add_argument(
        "--strict", action="store_true",
        help="Treat warnings as errors (raises exit code on missing assets / orphans).",
    )
    args = parser.parse_args()

    themes = args.theme if args.theme else DEFAULT_THEMES
    total_errors = 0
    total_warnings = 0
    for theme in themes:
        report = validate_theme(theme)
        if report.errors:
            print(f"[{theme}] ERRORS:")
            for e in report.errors:
                print(f"  ✗ {e}")
            total_errors += len(report.errors)
        if report.warnings:
            print(f"[{theme}] warnings:")
            for w in report.warnings:
                print(f"  · {w}")
            total_warnings += len(report.warnings)
        if not report.errors and not report.warnings:
            print(f"[{theme}] OK")

    if total_errors > 0:
        print(f"\nFAIL: {total_errors} structural error(s)")
        return 1
    if args.strict and total_warnings > 0:
        print(f"\nFAIL (--strict): {total_warnings} warning(s)")
        return 1
    if total_warnings > 0:
        print(f"\nOK with {total_warnings} warning(s)")
    else:
        print("\nOK")
    return 0


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