feat(ci): add content cross-reference validation (9 checks)
Extends tooling/validate-content with Pass 2 cross-reference validation via ContentIndex class. Nine checks: 1. canonical_id uniqueness (ERROR) 2. relationship target resolution (ERROR) 3. location slug resolution (ERROR) 4. dialogue location resolution (ERROR) 5. fact_id resolution (ERROR, advisory when catalogs empty) 6. triangle membership resolution (ERROR) 7. npc_count accuracy (WARNING) 8. dialogue line_id uniqueness (ERROR) 9. bidirectional relationship consistency (WARNING) Pass 2 only runs if Pass 1 (schema) passes. Absorbs check-fact-ids functionality. Spec from hoshe-round3.md Section 2. Ticket: #464 Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
+447
-26
@@ -1,17 +1,20 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Validate content YAML files against their JSON schemas.
|
||||
"""Validate content YAML files against their JSON schemas and cross-references.
|
||||
|
||||
Pass 1: Schema validation (per-file against JSON Schema)
|
||||
Pass 2: Cross-reference validation (9 checks across all files)
|
||||
|
||||
Schema mapping is by directory context:
|
||||
campaign.yaml → campaign.schema.json
|
||||
system.yaml → system.schema.json
|
||||
station.yaml → station.schema.json
|
||||
district.yaml → district.schema.json
|
||||
npcs/*.yaml → npc-profile.schema.json
|
||||
locations/*.yaml → location.schema.json
|
||||
triangles/*.yaml → triangle.schema.json
|
||||
dialogue/**/*.yaml → dialogue-pool.schema.json
|
||||
monologue/**/*.yaml → monologue-pool.schema.json
|
||||
routines/*.yaml → routine.schema.json
|
||||
campaign.yaml -> campaign.schema.json
|
||||
system.yaml -> system.schema.json
|
||||
station.yaml -> station.schema.json
|
||||
district.yaml -> district.schema.json
|
||||
npcs/*.yaml -> npc-profile.schema.json
|
||||
locations/*.yaml -> location.schema.json
|
||||
triangles/*.yaml -> triangle.schema.json
|
||||
dialogue/**/*.yaml -> dialogue-pool.schema.json
|
||||
monologue/**/*.yaml -> monologue-pool.schema.json
|
||||
routines/*.yaml -> routine.schema.json
|
||||
|
||||
Files under global/ and content.yaml are skipped (no schema yet).
|
||||
|
||||
@@ -19,6 +22,7 @@ Exit code 0 = all valid, 1 = validation errors found.
|
||||
"""
|
||||
|
||||
import json
|
||||
import re
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
@@ -61,23 +65,420 @@ def resolve_schema(yaml_path: Path) -> Path | None:
|
||||
return None
|
||||
|
||||
|
||||
def main() -> int:
|
||||
def _rel(path: Path) -> str:
|
||||
"""Return path relative to CONTENT_DIR for display."""
|
||||
try:
|
||||
return str(path.relative_to(CONTENT_DIR))
|
||||
except ValueError:
|
||||
return str(path)
|
||||
|
||||
|
||||
def _load_yaml(path: Path) -> dict | list | None:
|
||||
"""Load a YAML file, returning None for empty/comment-only files."""
|
||||
try:
|
||||
with open(path) as f:
|
||||
return yaml.safe_load(f)
|
||||
except yaml.YAMLError:
|
||||
return None
|
||||
|
||||
|
||||
def _file_type(path: Path) -> str | None:
|
||||
"""Determine content type from path (npc, dialogue, monologue, district, triangle)."""
|
||||
rel = path.relative_to(CONTENT_DIR)
|
||||
for part in reversed(rel.parts[:-1]):
|
||||
if part in ("npcs", "dialogue", "monologue", "triangles", "locations"):
|
||||
return part
|
||||
if path.name == "district.yaml":
|
||||
return "district"
|
||||
return None
|
||||
|
||||
|
||||
def _district_dir(path: Path) -> Path | None:
|
||||
"""Find the district directory containing this file."""
|
||||
parts = path.resolve().parts
|
||||
for i, part in enumerate(parts):
|
||||
if part == "districts" and i + 1 < len(parts):
|
||||
return Path(*parts[: i + 2])
|
||||
return None
|
||||
|
||||
|
||||
class ContentIndex:
|
||||
"""Builds an index of all defined entities for cross-referencing."""
|
||||
|
||||
def __init__(self, campaigns_dir: Path):
|
||||
self.campaigns_dir = campaigns_dir
|
||||
self.npcs: dict[str, Path] = {} # canonical_id -> defining file
|
||||
self.locations: dict[str, set] = {} # district_dir str -> set of location slugs
|
||||
self.triangle_slugs: dict[str, set] = {} # district_dir str -> set of triangle file stems
|
||||
self.fact_ids: set[str] = set() # from knowledge catalogs
|
||||
self.dialogue_files: list[tuple[Path, dict]] = []
|
||||
self.monologue_files: list[tuple[Path, dict]] = []
|
||||
self.npc_files: list[tuple[Path, dict]] = []
|
||||
self.district_files: list[tuple[Path, dict]] = []
|
||||
|
||||
def build(self):
|
||||
"""Scan all content files and populate the index."""
|
||||
self._scan_npcs()
|
||||
self._scan_districts()
|
||||
self._scan_triangles()
|
||||
self._scan_knowledge()
|
||||
self._scan_dialogue()
|
||||
self._scan_monologue()
|
||||
|
||||
def _scan_npcs(self):
|
||||
for path in sorted(self.campaigns_dir.rglob("npcs/*.yaml")):
|
||||
data = _load_yaml(path)
|
||||
if data and isinstance(data, dict):
|
||||
self.npc_files.append((path, data))
|
||||
cid = data.get("canonical_id")
|
||||
if cid:
|
||||
self.npcs[cid] = path
|
||||
|
||||
def _scan_districts(self):
|
||||
for path in sorted(self.campaigns_dir.rglob("district.yaml")):
|
||||
data = _load_yaml(path)
|
||||
if data and isinstance(data, dict):
|
||||
self.district_files.append((path, data))
|
||||
district_dir = str(path.parent.resolve())
|
||||
locs = data.get("locations", [])
|
||||
if isinstance(locs, list):
|
||||
self.locations[district_dir] = set(locs)
|
||||
|
||||
def _scan_triangles(self):
|
||||
for path in sorted(self.campaigns_dir.rglob("triangles/*.yaml")):
|
||||
data = _load_yaml(path)
|
||||
if data and isinstance(data, dict):
|
||||
district_dir = str(path.parent.parent.resolve())
|
||||
if district_dir not in self.triangle_slugs:
|
||||
self.triangle_slugs[district_dir] = set()
|
||||
self.triangle_slugs[district_dir].add(path.stem)
|
||||
|
||||
def _scan_knowledge(self):
|
||||
knowledge_dir = CONTENT_DIR / "global" / "knowledge"
|
||||
if not knowledge_dir.exists():
|
||||
return
|
||||
for path in sorted(knowledge_dir.glob("*.yaml")):
|
||||
if path.name == "entity-attributes.yaml":
|
||||
continue
|
||||
data = _load_yaml(path)
|
||||
if not data or not isinstance(data, dict):
|
||||
# Try line-by-line grep for fact_id definitions
|
||||
try:
|
||||
with open(path) as f:
|
||||
for line in f:
|
||||
line = line.strip()
|
||||
if line.startswith("#") or not line:
|
||||
continue
|
||||
m = re.match(r"fact_id:\s*[\"']?([^\"'#]+)", line)
|
||||
if m:
|
||||
self.fact_ids.add(m.group(1).strip())
|
||||
except OSError:
|
||||
pass
|
||||
continue
|
||||
# Handle structured catalogs
|
||||
self._extract_fact_ids_from_dict(data)
|
||||
|
||||
def _extract_fact_ids_from_dict(self, data, prefix=""):
|
||||
"""Recursively extract fact_id values from knowledge catalog structures."""
|
||||
if isinstance(data, dict):
|
||||
for key, val in data.items():
|
||||
if key == "fact_id" and isinstance(val, str):
|
||||
self.fact_ids.add(val)
|
||||
elif isinstance(val, (dict, list)):
|
||||
self._extract_fact_ids_from_dict(val)
|
||||
elif isinstance(data, list):
|
||||
for item in data:
|
||||
self._extract_fact_ids_from_dict(item)
|
||||
|
||||
def _scan_dialogue(self):
|
||||
for path in sorted(self.campaigns_dir.rglob("dialogue/**/*.yaml")):
|
||||
if path.name.startswith("."):
|
||||
continue
|
||||
data = _load_yaml(path)
|
||||
if data and isinstance(data, dict):
|
||||
self.dialogue_files.append((path, data))
|
||||
|
||||
def _scan_monologue(self):
|
||||
for path in sorted(self.campaigns_dir.rglob("monologue/**/*.yaml")):
|
||||
if path.name.startswith("."):
|
||||
continue
|
||||
data = _load_yaml(path)
|
||||
if data and isinstance(data, dict):
|
||||
self.monologue_files.append((path, data))
|
||||
|
||||
def validate_references(self) -> tuple[int, int]:
|
||||
"""Check all cross-references. Returns (error_count, warning_count)."""
|
||||
errors = 0
|
||||
warnings = 0
|
||||
|
||||
errors += self._check_1_canonical_id_uniqueness()
|
||||
errors += self._check_2_relationship_targets()
|
||||
errors += self._check_3_location_slugs()
|
||||
errors += self._check_4_dialogue_locations()
|
||||
errors += self._check_5_fact_ids()
|
||||
errors += self._check_6_triangle_membership()
|
||||
warnings += self._check_7_npc_count()
|
||||
errors += self._check_8_dialogue_line_ids()
|
||||
warnings += self._check_9_bidirectional_relationships()
|
||||
|
||||
return errors, warnings
|
||||
|
||||
def _check_1_canonical_id_uniqueness(self) -> int:
|
||||
"""Check that canonical_id values are unique across all NPC files."""
|
||||
errors = 0
|
||||
seen: dict[str, Path] = {}
|
||||
for path, data in self.npc_files:
|
||||
cid = data.get("canonical_id")
|
||||
if not cid:
|
||||
continue
|
||||
if cid in seen:
|
||||
print(f'XREF ERROR: duplicate canonical_id "{cid}"')
|
||||
print(f" Defined in: {_rel(seen[cid])}")
|
||||
print(f" Duplicate in: {_rel(path)}")
|
||||
errors += 1
|
||||
else:
|
||||
seen[cid] = path
|
||||
return errors
|
||||
|
||||
def _check_2_relationship_targets(self) -> int:
|
||||
"""Check that all relationship targets resolve to known canonical_ids."""
|
||||
errors = 0
|
||||
for path, data in self.npc_files:
|
||||
rels = data.get("relationships", [])
|
||||
if not isinstance(rels, list):
|
||||
continue
|
||||
for rel in rels:
|
||||
if not isinstance(rel, dict):
|
||||
continue
|
||||
target = rel.get("target")
|
||||
if target and target not in self.npcs:
|
||||
known = sorted(self.npcs.keys())
|
||||
print(f'XREF ERROR: unresolved relationship target "{target}"')
|
||||
print(f" In: {_rel(path)}")
|
||||
print(f" Known canonical_ids: {', '.join(known[:10])}"
|
||||
+ (f" (and {len(known) - 10} more)" if len(known) > 10 else ""))
|
||||
errors += 1
|
||||
return errors
|
||||
|
||||
def _check_3_location_slugs(self) -> int:
|
||||
"""Check that district location slugs correspond to actual location files."""
|
||||
errors = 0
|
||||
for path, data in self.district_files:
|
||||
locs = data.get("locations", [])
|
||||
if not isinstance(locs, list):
|
||||
continue
|
||||
locations_dir = path.parent / "locations"
|
||||
for slug in locs:
|
||||
expected = locations_dir / f"{slug}.yaml"
|
||||
if not expected.exists():
|
||||
print(f'XREF ERROR: location slug "{slug}" not found')
|
||||
print(f" In: {_rel(path)}")
|
||||
print(f" Expected file: {_rel(expected)}")
|
||||
errors += 1
|
||||
return errors
|
||||
|
||||
def _check_4_dialogue_locations(self) -> int:
|
||||
"""Check that dialogue file locations match district locations."""
|
||||
errors = 0
|
||||
for path, data in self.dialogue_files:
|
||||
location = data.get("location")
|
||||
if not location:
|
||||
continue
|
||||
district_dir = _district_dir(path)
|
||||
if not district_dir:
|
||||
continue
|
||||
district_key = str(district_dir)
|
||||
district_locs = self.locations.get(district_key, set())
|
||||
if district_locs and location not in district_locs:
|
||||
print(f'XREF ERROR: dialogue location "{location}" not in district')
|
||||
print(f" In: {_rel(path)}")
|
||||
print(f" District locations: {', '.join(sorted(district_locs))}")
|
||||
errors += 1
|
||||
return errors
|
||||
|
||||
def _check_5_fact_ids(self) -> int:
|
||||
"""Check that all referenced fact_ids resolve to knowledge catalogs."""
|
||||
if not self.fact_ids:
|
||||
# Advisory mode: catalogs not populated yet
|
||||
return 0
|
||||
|
||||
errors = 0
|
||||
# Collect all fact_id references from dialogue, monologue, NPC, and item files
|
||||
refs = self._collect_all_fact_id_refs()
|
||||
|
||||
for fact_id, sources in sorted(refs.items()):
|
||||
if fact_id not in self.fact_ids:
|
||||
print(f'XREF ERROR: unknown fact_id "{fact_id}"')
|
||||
for src in sources[:3]:
|
||||
print(f" In: {_rel(src)}")
|
||||
if len(sources) > 3:
|
||||
print(f" ...and {len(sources) - 3} more files")
|
||||
print(f" Canonical fact_ids: {len(self.fact_ids)} defined")
|
||||
errors += 1
|
||||
return errors
|
||||
|
||||
def _collect_all_fact_id_refs(self) -> dict[str, list[Path]]:
|
||||
"""Collect all fact_id references from content files."""
|
||||
refs: dict[str, list[Path]] = {}
|
||||
|
||||
def add_ref(fact_id: str, path: Path):
|
||||
if fact_id:
|
||||
refs.setdefault(fact_id, []).append(path)
|
||||
|
||||
# Dialogue: lines[].knowledge_grant.fact_id
|
||||
for path, data in self.dialogue_files:
|
||||
for line in data.get("lines", []):
|
||||
if not isinstance(line, dict):
|
||||
continue
|
||||
kg = line.get("knowledge_grant")
|
||||
if isinstance(kg, dict):
|
||||
add_ref(kg.get("fact_id", ""), path)
|
||||
|
||||
# Monologue: entries with prerequisites.facts[].fact_id
|
||||
for path, data in self.monologue_files:
|
||||
self._walk_fact_id_refs(data, path, add_ref)
|
||||
|
||||
# NPC: information.knows[]
|
||||
for path, data in self.npc_files:
|
||||
info = data.get("information", {})
|
||||
if isinstance(info, dict):
|
||||
knows = info.get("knows", [])
|
||||
if isinstance(knows, list):
|
||||
for fid in knows:
|
||||
if isinstance(fid, str):
|
||||
add_ref(fid, path)
|
||||
|
||||
# Items: scan for fact_id in any nested structure
|
||||
for path in sorted(self.campaigns_dir.rglob("items/*.yaml")):
|
||||
data = _load_yaml(path)
|
||||
if data:
|
||||
self._walk_fact_id_refs(data, path, add_ref)
|
||||
|
||||
return refs
|
||||
|
||||
def _walk_fact_id_refs(self, data, path: Path, add_ref):
|
||||
"""Recursively find fact_id references in nested structures."""
|
||||
if isinstance(data, dict):
|
||||
if "fact_id" in data and isinstance(data["fact_id"], str):
|
||||
add_ref(data["fact_id"], path)
|
||||
for val in data.values():
|
||||
if isinstance(val, (dict, list)):
|
||||
self._walk_fact_id_refs(val, path, add_ref)
|
||||
elif isinstance(data, list):
|
||||
for item in data:
|
||||
if isinstance(item, (dict, list)):
|
||||
self._walk_fact_id_refs(item, path, add_ref)
|
||||
|
||||
def _check_6_triangle_membership(self) -> int:
|
||||
"""Check that NPC triangle_membership slugs correspond to triangle files."""
|
||||
errors = 0
|
||||
for path, data in self.npc_files:
|
||||
memberships = data.get("triangle_membership", [])
|
||||
if not isinstance(memberships, list):
|
||||
continue
|
||||
district_dir = _district_dir(path)
|
||||
if not district_dir:
|
||||
continue
|
||||
district_key = str(district_dir)
|
||||
available = self.triangle_slugs.get(district_key, set())
|
||||
for slug in memberships:
|
||||
if not isinstance(slug, str):
|
||||
continue
|
||||
if slug not in available:
|
||||
print(f'XREF ERROR: triangle "{slug}" not found')
|
||||
print(f" In: {_rel(path)}")
|
||||
if available:
|
||||
print(f" Available triangles: {', '.join(sorted(available))}")
|
||||
else:
|
||||
print(f" No triangles found in {_rel(district_dir / 'triangles')}")
|
||||
errors += 1
|
||||
return errors
|
||||
|
||||
def _check_7_npc_count(self) -> int:
|
||||
"""Check that district npc_count matches actual NPC file count."""
|
||||
warnings = 0
|
||||
for path, data in self.district_files:
|
||||
declared = data.get("npc_count")
|
||||
if declared is None:
|
||||
continue
|
||||
npcs_dir = path.parent / "npcs"
|
||||
if not npcs_dir.exists():
|
||||
continue
|
||||
actual = len(list(npcs_dir.glob("*.yaml")))
|
||||
if declared != actual:
|
||||
print(f"XREF WARNING: npc_count mismatch")
|
||||
print(f" Declared: {declared}")
|
||||
print(f" Actual NPC files: {actual}")
|
||||
print(f" In: {_rel(path)}")
|
||||
warnings += 1
|
||||
return warnings
|
||||
|
||||
def _check_8_dialogue_line_ids(self) -> int:
|
||||
"""Check that dialogue line IDs are unique within each file."""
|
||||
errors = 0
|
||||
for path, data in self.dialogue_files:
|
||||
lines = data.get("lines", [])
|
||||
if not isinstance(lines, list):
|
||||
continue
|
||||
seen: dict[str, int] = {}
|
||||
for i, line in enumerate(lines, 1):
|
||||
if not isinstance(line, dict):
|
||||
continue
|
||||
line_id = line.get("id")
|
||||
if not line_id:
|
||||
continue
|
||||
if line_id in seen:
|
||||
print(f'XREF ERROR: duplicate dialogue line id "{line_id}"')
|
||||
print(f" In: {_rel(path)}")
|
||||
print(f" First: line {seen[line_id]}")
|
||||
print(f" Duplicate: line {i}")
|
||||
errors += 1
|
||||
else:
|
||||
seen[line_id] = i
|
||||
return errors
|
||||
|
||||
def _check_9_bidirectional_relationships(self) -> int:
|
||||
"""Check that NPC relationships have reciprocal entries."""
|
||||
warnings = 0
|
||||
# Build map: canonical_id -> set of relationship targets
|
||||
npc_rels: dict[str, set[str]] = {}
|
||||
for _path, data in self.npc_files:
|
||||
cid = data.get("canonical_id")
|
||||
if not cid:
|
||||
continue
|
||||
rels = data.get("relationships", [])
|
||||
if isinstance(rels, list):
|
||||
targets = set()
|
||||
for rel in rels:
|
||||
if isinstance(rel, dict) and "target" in rel:
|
||||
targets.add(rel["target"])
|
||||
npc_rels[cid] = targets
|
||||
|
||||
# Check reciprocity
|
||||
checked = set()
|
||||
for cid, targets in npc_rels.items():
|
||||
for target in targets:
|
||||
pair = tuple(sorted([cid, target]))
|
||||
if pair in checked:
|
||||
continue
|
||||
checked.add(pair)
|
||||
if target in npc_rels and cid not in npc_rels.get(target, set()):
|
||||
print(f"XREF WARNING: {cid} has relationship to {target} but no reciprocal found")
|
||||
warnings += 1
|
||||
return warnings
|
||||
|
||||
|
||||
def schema_validate(campaigns_dir: Path) -> tuple[int, int, int]:
|
||||
"""Pass 1: Schema validation. Returns (validated, skipped, errors)."""
|
||||
errors = 0
|
||||
validated = 0
|
||||
skipped = 0
|
||||
|
||||
# Cache loaded schemas
|
||||
schema_cache: dict[str, dict] = {}
|
||||
|
||||
campaigns_dir = CONTENT_DIR / "campaigns"
|
||||
if not campaigns_dir.exists():
|
||||
print(f"No campaigns directory at {campaigns_dir}", file=sys.stderr)
|
||||
return 1
|
||||
|
||||
yaml_files = sorted(campaigns_dir.rglob("*.yaml"))
|
||||
if not yaml_files:
|
||||
print("No YAML files found under campaigns/", file=sys.stderr)
|
||||
return 1
|
||||
return 0, 0, 1
|
||||
|
||||
for yaml_path in yaml_files:
|
||||
schema_path = resolve_schema(yaml_path)
|
||||
@@ -90,14 +491,12 @@ def main() -> int:
|
||||
errors += 1
|
||||
continue
|
||||
|
||||
# Load schema (cached)
|
||||
schema_key = str(schema_path)
|
||||
if schema_key not in schema_cache:
|
||||
with open(schema_path) as f:
|
||||
schema_cache[schema_key] = json.load(f)
|
||||
schema = schema_cache[schema_key]
|
||||
|
||||
# Load YAML
|
||||
try:
|
||||
with open(yaml_path) as f:
|
||||
data = yaml.safe_load(f)
|
||||
@@ -107,11 +506,9 @@ def main() -> int:
|
||||
continue
|
||||
|
||||
if data is None:
|
||||
# Comment-only or empty placeholder files are valid stubs
|
||||
skipped += 1
|
||||
continue
|
||||
|
||||
# Validate
|
||||
try:
|
||||
jsonschema.validate(instance=data, schema=schema)
|
||||
validated += 1
|
||||
@@ -124,8 +521,32 @@ def main() -> int:
|
||||
print(f" Path: {'.'.join(str(p) for p in e.absolute_path)}")
|
||||
errors += 1
|
||||
|
||||
print(f"\nValidated {validated} files, {skipped} skipped, {errors} errors")
|
||||
return 1 if errors else 0
|
||||
return validated, skipped, errors
|
||||
|
||||
|
||||
def main() -> int:
|
||||
campaigns_dir = CONTENT_DIR / "campaigns"
|
||||
if not campaigns_dir.exists():
|
||||
print(f"No campaigns directory at {campaigns_dir}", file=sys.stderr)
|
||||
return 1
|
||||
|
||||
# Pass 1: Schema validation
|
||||
validated, skipped, schema_errors = schema_validate(campaigns_dir)
|
||||
print(f"\nPass 1 (schema): {validated} validated, {skipped} skipped, {schema_errors} errors")
|
||||
|
||||
if schema_errors > 0:
|
||||
print(f"\nSchema validation failed ({schema_errors} errors) -- skipping cross-references")
|
||||
return 1
|
||||
|
||||
# Pass 2: Cross-reference validation
|
||||
print("\nPass 2 (cross-references):")
|
||||
index = ContentIndex(campaigns_dir)
|
||||
index.build()
|
||||
xref_errors, xref_warnings = index.validate_references()
|
||||
|
||||
total_errors = schema_errors + xref_errors
|
||||
print(f"\nValidated {validated} files: {total_errors} errors, {xref_warnings} warnings")
|
||||
return 1 if total_errors > 0 else 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
|
||||
Reference in New Issue
Block a user