#!/usr/bin/env python3 """Validate content catalogs against JSON Schema (CI + local). Validates: - skill catalogs: content/skills/*_skills.json rows vs content/schemas/skill-def.schema.json - level curves: content/skills/*_level_curve.json vs content/schemas/level-curve.schema.json """ from __future__ import annotations import json import sys from pathlib import Path from jsonschema import Draft202012Validator REPO_ROOT = Path(__file__).resolve().parent.parent SKILL_SCHEMA = REPO_ROOT / "content/schemas/skill-def.schema.json" LEVEL_CURVE_SCHEMA = REPO_ROOT / "content/schemas/level-curve.schema.json" SKILLS_DIR = REPO_ROOT / "content/skills" # Slice 1 prototype lock (NEO-33): exact ids + category coverage after schema passes. PROTOTYPE_SLICE1_SKILL_IDS = frozenset({"salvage", "refine", "intrusion"}) def _prototype_slice1_gate(seen_ids: dict[str, str], id_to_category: dict[str, str]) -> str | None: """Return a human-readable error if Slice 1 contract fails, else None.""" ids = frozenset(seen_ids.keys()) if ids != PROTOTYPE_SLICE1_SKILL_IDS: return ( "error: prototype Slice 1 expects exactly skill ids " f"{sorted(PROTOTYPE_SLICE1_SKILL_IDS)!r}, got {sorted(ids)!r}" ) cats = set(id_to_category.values()) if "gather" not in cats or "tech" not in cats: return ( "error: prototype Slice 1 requires at least one gather and one tech skill; " f"categories seen: {sorted(cats)!r}" ) if "process" not in cats and "make" not in cats: return ( "error: prototype Slice 1 requires at least one process or make skill; " f"categories seen: {sorted(cats)!r}" ) return None def main() -> int: if not SKILL_SCHEMA.is_file(): print(f"error: missing schema {SKILL_SCHEMA}", file=sys.stderr) return 1 if not LEVEL_CURVE_SCHEMA.is_file(): print(f"error: missing schema {LEVEL_CURVE_SCHEMA}", file=sys.stderr) return 1 skill_schema = json.loads(SKILL_SCHEMA.read_text(encoding="utf-8")) skill_validator = Draft202012Validator(skill_schema) level_curve_schema = json.loads(LEVEL_CURVE_SCHEMA.read_text(encoding="utf-8")) level_curve_validator = Draft202012Validator(level_curve_schema) if not SKILLS_DIR.is_dir(): print(f"error: missing directory {SKILLS_DIR}", file=sys.stderr) return 1 skill_files = sorted(SKILLS_DIR.glob("*_skills.json")) if not skill_files: print(f"error: no *_skills.json files under {SKILLS_DIR}", file=sys.stderr) return 1 curve_files = sorted(SKILLS_DIR.glob("*_level_curve.json")) if not curve_files: print(f"error: no *_level_curve.json files under {SKILLS_DIR}", file=sys.stderr) return 1 seen_ids: dict[str, str] = {} id_to_category: dict[str, str] = {} errors = 0 for path in skill_files: data = json.loads(path.read_text(encoding="utf-8")) skills = data.get("skills") if not isinstance(skills, list): print(f"error: {path.relative_to(REPO_ROOT)}: expected top-level 'skills' array", file=sys.stderr) errors += 1 continue for i, row in enumerate(skills): if not isinstance(row, dict): print(f"error: {path.relative_to(REPO_ROOT)}: skills[{i}] must be an object", file=sys.stderr) errors += 1 continue row_schema_errors = 0 for err in sorted(skill_validator.iter_errors(row), key=lambda e: e.path): loc = ".".join(str(p) for p in err.path) or "(root)" print( f"error: {path.relative_to(REPO_ROOT)} skills[{i}] {loc}: {err.message}", file=sys.stderr, ) row_schema_errors += 1 errors += 1 sid = row.get("id") # Reserve seen_ids (and duplicate detection) only for schema-clean rows so # fixing a broken row does not surface a spurious duplicate vs an earlier invalid row. if isinstance(sid, str) and row_schema_errors == 0: prev = seen_ids.get(sid) if prev: print( f"error: duplicate skill id {sid!r} in {prev} and {path.relative_to(REPO_ROOT)}", file=sys.stderr, ) errors += 1 else: seen_ids[sid] = str(path.relative_to(REPO_ROOT)) cat = row.get("category") if isinstance(cat, str): id_to_category[sid] = cat for path in curve_files: data = json.loads(path.read_text(encoding="utf-8")) row_errors = 0 for err in sorted(level_curve_validator.iter_errors(data), key=lambda e: e.path): loc = ".".join(str(p) for p in err.path) or "(root)" print( f"error: {path.relative_to(REPO_ROOT)} {loc}: {err.message}", file=sys.stderr, ) row_errors += 1 errors += 1 if row_errors > 0: continue levels = data.get("levels") if not isinstance(levels, list): print(f"error: {path.relative_to(REPO_ROOT)}: expected top-level 'levels' array", file=sys.stderr) errors += 1 continue if len(levels) == 0: print(f"error: {path.relative_to(REPO_ROOT)}: levels must contain at least one row", file=sys.stderr) errors += 1 continue first = levels[0] if not isinstance(first, dict) or first.get("level") != 1 or first.get("requiredXp") != 0: print( f"error: {path.relative_to(REPO_ROOT)}: first row must be level=1 and requiredXp=0", file=sys.stderr, ) errors += 1 prev_level = 0 prev_xp = -1 for i, row in enumerate(levels): if not isinstance(row, dict): print(f"error: {path.relative_to(REPO_ROOT)}: levels[{i}] must be an object", file=sys.stderr) errors += 1 continue level = row.get("level") required_xp = row.get("requiredXp") if not isinstance(level, int) or not isinstance(required_xp, int): print( f"error: {path.relative_to(REPO_ROOT)}: levels[{i}] level/requiredXp must be integers", file=sys.stderr, ) errors += 1 continue if level <= prev_level: print( f"error: {path.relative_to(REPO_ROOT)}: levels[{i}].level must be strictly increasing", file=sys.stderr, ) errors += 1 if required_xp <= prev_xp: print( f"error: {path.relative_to(REPO_ROOT)}: levels[{i}].requiredXp must be strictly increasing", file=sys.stderr, ) errors += 1 prev_level = level prev_xp = required_xp if errors: print(f"content validation failed with {errors} error(s)", file=sys.stderr) return 1 slice1_err = _prototype_slice1_gate(seen_ids, id_to_category) if slice1_err: print(slice1_err, file=sys.stderr) return 1 print( "content OK: " f"{len(skill_files)} skill catalog file(s), " f"{len(curve_files)} level curve file(s), " f"{len(seen_ids)} unique skill id(s)" ) return 0 if __name__ == "__main__": raise SystemExit(main())