neon-sprawl/scripts/validate_content.py

591 lines
23 KiB
Python

#!/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
- mastery catalogs: content/mastery/*_mastery.json vs content/schemas/mastery-catalog.schema.json
(post-schema: tierIndex unique and sequential 1..N per track, aligned with server boot — NEO-46)
- item catalogs: content/items/*_items.json rows vs content/schemas/item-def.schema.json (NEO-50)
"""
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"
MASTERY_SCHEMA = REPO_ROOT / "content/schemas/mastery-catalog.schema.json"
ITEM_SCHEMA = REPO_ROOT / "content/schemas/item-def.schema.json"
SKILLS_DIR = REPO_ROOT / "content/skills"
MASTERY_DIR = REPO_ROOT / "content/mastery"
ITEMS_DIR = REPO_ROOT / "content/items"
# Slice 1 prototype lock (NEO-33): exact ids + category coverage after schema passes.
PROTOTYPE_SLICE1_SKILL_IDS = frozenset({"salvage", "refine", "intrusion"})
# Slice 1 prototype lock (NEO-50): exact item ids + prototypeRole coverage after schema passes.
PROTOTYPE_SLICE1_ITEM_IDS = frozenset(
{
"scrap_metal_bulk",
"refined_plate_stock",
"field_stim_mk0",
"survey_drone_kit",
"contract_handoff_token",
"prototype_armor_shell",
}
)
PROTOTYPE_SLICE1_ITEM_ROLES = frozenset(
{"material", "intermediate", "consumable", "utility", "quest_token", "equip_stub"}
)
# Slice 4 prototype lock (NEO-45): exactly one salvage track across all mastery files.
PROTOTYPE_SLICE4_SALVAGE_SKILL_ID = "salvage"
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 _tier_index_gate(rel: str, track_index: int, tier_index_values: list[int]) -> str | None:
"""Keep in sync with MasteryCatalogLoader.TryGetTierIndexGateError (NEO-46)."""
if not tier_index_values:
return None
seen: set[int] = set()
for value in tier_index_values:
if value in seen:
return f"error: {rel}: tracks[{track_index}] duplicate tierIndex {value}"
seen.add(value)
expected = set(range(1, len(tier_index_values) + 1))
if seen != expected:
sorted_vals = ", ".join(str(v) for v in sorted(seen))
n = len(tier_index_values)
return (
f"error: {rel}: tracks[{track_index}] tierIndex values must be unique and "
f"sequential 1..{n}, got [{sorted_vals}]"
)
return None
def _validate_mastery_catalogs(
*,
mastery_files: list[Path],
mastery_validator: Draft202012Validator,
skill_ids: frozenset[str],
) -> tuple[int, list[str]]:
"""Validate mastery JSON files. Returns (error_count, track_skill_ids across all files)."""
errors = 0
all_track_skill_ids: list[str] = []
global_perk_ids: dict[str, str] = {}
# Each perkId may appear in at most one branch reference catalog-wide (v1 mutually exclusive trees).
global_referenced_perk_ids: dict[str, str] = {}
for path in mastery_files:
rel = str(path.relative_to(REPO_ROOT))
data = json.loads(path.read_text(encoding="utf-8"))
schema_errors = 0
for err in sorted(mastery_validator.iter_errors(data), key=lambda e: e.path):
loc = ".".join(str(p) for p in err.path) or "(root)"
print(f"error: {rel} {loc}: {err.message}", file=sys.stderr)
schema_errors += 1
errors += 1
if schema_errors > 0:
continue
perks = data.get("perks")
tracks = data.get("tracks")
if not isinstance(perks, dict) or not isinstance(tracks, list):
print(f"error: {rel}: expected top-level 'perks' object and 'tracks' array", file=sys.stderr)
errors += 1
continue
referenced_perk_ids_in_file: set[str] = set()
for perk_key, perk in perks.items():
if not isinstance(perk, dict):
print(f"error: {rel}: perks[{perk_key!r}] must be an object", file=sys.stderr)
errors += 1
continue
pid = perk.get("id")
if not isinstance(pid, str):
continue
if perk_key != pid:
print(
f"error: {rel}: perks map key {perk_key!r} must match PerkDef.id {pid!r}",
file=sys.stderr,
)
errors += 1
prev = global_perk_ids.get(pid)
if prev:
print(f"error: duplicate perk id {pid!r} in {prev} and {rel}", file=sys.stderr)
errors += 1
else:
global_perk_ids[pid] = rel
for ti, track in enumerate(tracks):
if not isinstance(track, dict):
print(f"error: {rel}: tracks[{ti}] must be an object", file=sys.stderr)
errors += 1
continue
skill_id = track.get("skillId")
if isinstance(skill_id, str):
all_track_skill_ids.append(skill_id)
if skill_id not in skill_ids:
print(
f"error: {rel}: tracks[{ti}].skillId {skill_id!r} is not a known SkillDef id "
f"(known: {sorted(skill_ids)!r})",
file=sys.stderr,
)
errors += 1
tiers = track.get("tiers")
if not isinstance(tiers, list):
continue
prev_required_level = 0
prev_tier_branch_ids: frozenset[str] | None = None
tier_index_values: list[int] = []
for tier_i, tier in enumerate(tiers):
if not isinstance(tier, dict):
continue
tier_index = tier.get("tierIndex")
if isinstance(tier_index, int):
tier_index_values.append(tier_index)
required_level = tier.get("requiredLevel")
if isinstance(required_level, int) and required_level <= prev_required_level:
print(
f"error: {rel}: tracks[{ti}].tiers[{tier_i}].requiredLevel must be strictly "
f"greater than previous tier ({prev_required_level})",
file=sys.stderr,
)
errors += 1
if isinstance(required_level, int):
prev_required_level = required_level
branches = tier.get("branches")
if not isinstance(branches, list):
continue
tier_branch_ids: list[str] = []
for bi, branch in enumerate(branches):
if not isinstance(branch, dict):
continue
branch_id = branch.get("branchId")
if isinstance(branch_id, str):
if branch_id in tier_branch_ids:
print(
f"error: {rel}: tracks[{ti}].tiers[{tier_i}] duplicate branchId "
f"{branch_id!r}",
file=sys.stderr,
)
errors += 1
tier_branch_ids.append(branch_id)
perk_ids = branch.get("perkIds")
if not isinstance(perk_ids, list):
continue
for perk_id in perk_ids:
if not isinstance(perk_id, str):
continue
if perk_id not in perks:
print(
f"error: {rel}: tracks[{ti}].tiers[{tier_i}].branches[{bi}] "
f"references unknown perkId {perk_id!r}",
file=sys.stderr,
)
errors += 1
prev_ref = global_referenced_perk_ids.get(perk_id)
if prev_ref:
print(
f"error: duplicate perk id reference {perk_id!r} in {prev_ref} and {rel}",
file=sys.stderr,
)
errors += 1
else:
global_referenced_perk_ids[perk_id] = (
f"{rel} tracks[{ti}].tiers[{tier_i}].branches[{bi}]"
)
referenced_perk_ids_in_file.add(perk_id)
tier_branch_set = frozenset(tier_branch_ids)
if prev_tier_branch_ids is not None and tier_branch_set != prev_tier_branch_ids:
print(
f"error: {rel}: tracks[{ti}].tiers[{tier_i}] branchId set "
f"{sorted(tier_branch_set)!r} must match previous tier "
f"{sorted(prev_tier_branch_ids)!r}",
file=sys.stderr,
)
errors += 1
if tier_branch_ids:
prev_tier_branch_ids = tier_branch_set
tier_index_err = _tier_index_gate(rel, ti, tier_index_values)
if tier_index_err:
print(tier_index_err, file=sys.stderr)
errors += 1
for perk_key in perks:
if isinstance(perk_key, str) and perk_key not in referenced_perk_ids_in_file:
print(
f"error: {rel}: perks[{perk_key!r}] is not referenced by any branch perkIds",
file=sys.stderr,
)
errors += 1
return errors, all_track_skill_ids
def _prototype_slice1_item_gate(
seen_ids: dict[str, str],
id_to_role: dict[str, str],
id_to_slot_kind: dict[str, str],
id_to_stack_max: dict[str, int],
) -> str | None:
"""Return a human-readable error if Slice 1 item contract fails, else None."""
ids = frozenset(seen_ids.keys())
if ids != PROTOTYPE_SLICE1_ITEM_IDS:
return (
"error: prototype Slice 1 expects exactly item ids "
f"{sorted(PROTOTYPE_SLICE1_ITEM_IDS)!r}, got {sorted(ids)!r}"
)
roles = set(id_to_role.values())
if roles != PROTOTYPE_SLICE1_ITEM_ROLES:
return (
"error: prototype Slice 1 requires exactly one row per prototypeRole "
f"{sorted(PROTOTYPE_SLICE1_ITEM_ROLES)!r}, roles seen: {sorted(roles)!r}"
)
if len(id_to_role) != len(roles):
return "error: prototype Slice 1 duplicate prototypeRole across item rows"
equip_stub_id = next((iid for iid, role in id_to_role.items() if role == "equip_stub"), None)
if equip_stub_id is not None:
if id_to_slot_kind.get(equip_stub_id) != "equipment":
return (
f"error: {equip_stub_id!r} (equip_stub) must use inventorySlotKind 'equipment', "
f"got {id_to_slot_kind.get(equip_stub_id)!r}"
)
if id_to_stack_max.get(equip_stub_id) != 1:
return (
f"error: {equip_stub_id!r} (equip_stub) must use stackMax 1, "
f"got {id_to_stack_max.get(equip_stub_id)!r}"
)
return None
def _validate_item_catalogs(
*,
item_files: list[Path],
item_validator: Draft202012Validator,
) -> tuple[int, dict[str, str], dict[str, str], dict[str, str], dict[str, int]]:
"""Validate item JSON files. Returns (error_count, seen_ids, id_to_role, id_to_slot_kind, id_to_stack_max)."""
errors = 0
seen_ids: dict[str, str] = {}
id_to_role: dict[str, str] = {}
id_to_slot_kind: dict[str, str] = {}
id_to_stack_max: dict[str, int] = {}
for path in item_files:
rel = str(path.relative_to(REPO_ROOT))
data = json.loads(path.read_text(encoding="utf-8"))
schema_version = data.get("schemaVersion")
if schema_version != 1:
print(f"error: {rel}: expected schemaVersion 1, got {schema_version!r}", file=sys.stderr)
errors += 1
continue
items = data.get("items")
if not isinstance(items, list):
print(f"error: {rel}: expected top-level 'items' array", file=sys.stderr)
errors += 1
continue
for i, row in enumerate(items):
if not isinstance(row, dict):
print(f"error: {rel}: items[{i}] must be an object", file=sys.stderr)
errors += 1
continue
row_schema_errors = 0
for err in sorted(item_validator.iter_errors(row), key=lambda e: e.path):
loc = ".".join(str(p) for p in err.path) or "(root)"
print(f"error: {rel} items[{i}] {loc}: {err.message}", file=sys.stderr)
row_schema_errors += 1
errors += 1
iid = row.get("id")
if isinstance(iid, str) and row_schema_errors == 0:
prev = seen_ids.get(iid)
if prev:
print(f"error: duplicate item id {iid!r} in {prev} and {rel}", file=sys.stderr)
errors += 1
else:
seen_ids[iid] = rel
role = row.get("prototypeRole")
if isinstance(role, str):
id_to_role[iid] = role
slot_kind = row.get("inventorySlotKind")
if isinstance(slot_kind, str):
id_to_slot_kind[iid] = slot_kind
stack_max = row.get("stackMax")
if isinstance(stack_max, int):
id_to_stack_max[iid] = stack_max
return errors, seen_ids, id_to_role, id_to_slot_kind, id_to_stack_max
def _prototype_slice4_gate(track_skill_ids: list[str]) -> str | None:
"""Return a human-readable error if Slice 4 contract fails, else None."""
if len(track_skill_ids) != 1:
return (
"error: prototype Slice 4 expects exactly one MasteryTrack across all mastery files, "
f"got {len(track_skill_ids)} track(s) with skillIds {track_skill_ids!r}"
)
if track_skill_ids[0] != PROTOTYPE_SLICE4_SALVAGE_SKILL_ID:
return (
"error: prototype Slice 4 expects the sole track skillId "
f"{PROTOTYPE_SLICE4_SALVAGE_SKILL_ID!r}, got {track_skill_ids[0]!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
if not MASTERY_SCHEMA.is_file():
print(f"error: missing schema {MASTERY_SCHEMA}", file=sys.stderr)
return 1
if not ITEM_SCHEMA.is_file():
print(f"error: missing schema {ITEM_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)
mastery_schema = json.loads(MASTERY_SCHEMA.read_text(encoding="utf-8"))
mastery_validator = Draft202012Validator(mastery_schema)
item_schema = json.loads(ITEM_SCHEMA.read_text(encoding="utf-8"))
item_validator = Draft202012Validator(item_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
if not MASTERY_DIR.is_dir():
print(f"error: missing directory {MASTERY_DIR}", file=sys.stderr)
return 1
mastery_files = sorted(MASTERY_DIR.glob("*_mastery.json"))
if not mastery_files:
print(f"error: no *_mastery.json files under {MASTERY_DIR}", file=sys.stderr)
return 1
if not ITEMS_DIR.is_dir():
print(f"error: missing directory {ITEMS_DIR}", file=sys.stderr)
return 1
item_files = sorted(ITEMS_DIR.glob("*_items.json"))
if not item_files:
print(f"error: no *_items.json files under {ITEMS_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
mastery_errors, track_skill_ids = _validate_mastery_catalogs(
mastery_files=mastery_files,
mastery_validator=mastery_validator,
skill_ids=frozenset(seen_ids.keys()),
)
if mastery_errors:
print(f"content validation failed with {mastery_errors} error(s)", file=sys.stderr)
return 1
slice4_err = _prototype_slice4_gate(track_skill_ids)
if slice4_err:
print(slice4_err, file=sys.stderr)
return 1
item_errors, item_seen_ids, id_to_role, id_to_slot_kind, id_to_stack_max = _validate_item_catalogs(
item_files=item_files,
item_validator=item_validator,
)
if item_errors:
print(f"content validation failed with {item_errors} error(s)", file=sys.stderr)
return 1
slice1_item_err = _prototype_slice1_item_gate(
item_seen_ids, id_to_role, id_to_slot_kind, id_to_stack_max
)
if slice1_item_err:
print(slice1_item_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(mastery_files)} mastery catalog file(s), "
f"{len(item_files)} item catalog file(s), "
f"{len(seen_ids)} unique skill id(s), "
f"{len(item_seen_ids)} unique item id(s), "
f"{len(track_skill_ids)} mastery track(s)"
)
return 0
if __name__ == "__main__":
raise SystemExit(main())