#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ Technique capability registry validator and audit generator. This script deliberately uses only Python stdlib. Open-source research suggested Yamale/jsonschema-style validation for registries, but this skill avoids adding runtime dependencies beyond pyswisseph. """ from __future__ import annotations import argparse import json import os import sys from typing import Any, Dict, List ROOT_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) DEFAULT_REGISTRY = os.path.join(ROOT_DIR, "references", "technique_registry.json") ALLOWED_STATUS = { "covered", "complete", "partial", "blocked", "knowledge-only", "workflow-only", "not-integrated", "missing", "guarded", "comparison-only", } REQUIRED_TECHNIQUE_FIELDS = { "name": str, "domains": list, "status": str, "knowledge_refs": list, "commands": list, "output_paths": list, "audit_label": str, "missing_impact": str, } def load_registry(path: str = DEFAULT_REGISTRY) -> Dict[str, Any]: with open(path, "r", encoding="utf-8") as f: return json.load(f) def _path_exists_relative(path: str) -> bool: if path.startswith("scripts/") and " " in path: path = path.split()[0] return os.path.exists(os.path.join(ROOT_DIR, path)) def validate_registry(registry: Dict[str, Any]) -> Dict[str, Any]: problems: List[Dict[str, str]] = [] warnings: List[Dict[str, str]] = [] techniques = registry.get("techniques") if not isinstance(techniques, dict) or not techniques: problems.append({"level": "error", "path": "techniques", "message": "techniques must be a non-empty object"}) techniques = {} for tech_id, tech in techniques.items(): if not isinstance(tech, dict): problems.append({"level": "error", "path": tech_id, "message": "technique entry must be an object"}) continue for field, expected_type in REQUIRED_TECHNIQUE_FIELDS.items(): if field not in tech: problems.append({"level": "error", "path": f"techniques.{tech_id}.{field}", "message": "required field missing"}) continue if not isinstance(tech[field], expected_type): problems.append({"level": "error", "path": f"techniques.{tech_id}.{field}", "message": f"must be {expected_type.__name__}"}) status = tech.get("status") if status not in ALLOWED_STATUS: problems.append({"level": "error", "path": f"techniques.{tech_id}.status", "message": f"invalid status {status!r}"}) if status == "covered" and not tech.get("output_paths"): problems.append({"level": "error", "path": f"techniques.{tech_id}.output_paths", "message": "covered techniques must expose output_paths"}) if status == "partial" and not tech.get("limitation"): warnings.append({"level": "warning", "path": f"techniques.{tech_id}.limitation", "message": "partial techniques should explain limitation"}) for ref in tech.get("knowledge_refs", []): if isinstance(ref, str) and not _path_exists_relative(ref): warnings.append({"level": "warning", "path": f"techniques.{tech_id}.knowledge_refs", "message": f"reference not found: {ref}"}) routes = registry.get("routes", {}) if not isinstance(routes, dict): problems.append({"level": "error", "path": "routes", "message": "routes must be an object"}) routes = {} for route_id, route in routes.items(): if not isinstance(route, dict): problems.append({"level": "error", "path": f"routes.{route_id}", "message": "route entry must be an object"}) continue required = route.get("required_techniques", []) optional = route.get("optional_techniques", []) if not isinstance(required, list): problems.append({"level": "error", "path": f"routes.{route_id}.required_techniques", "message": "must be a list"}) required = [] if not isinstance(optional, list): problems.append({"level": "error", "path": f"routes.{route_id}.optional_techniques", "message": "must be a list"}) optional = [] for tech_id in required + optional: if tech_id not in techniques: problems.append({"level": "error", "path": f"routes.{route_id}", "message": f"unknown technique id: {tech_id}"}) status_counts: Dict[str, int] = {s: 0 for s in sorted(ALLOWED_STATUS)} domain_counts: Dict[str, int] = {} for tech in techniques.values(): status = tech.get("status") if status in status_counts: status_counts[status] += 1 for domain in tech.get("domains", []): domain_counts[domain] = domain_counts.get(domain, 0) + 1 return { "valid": len(problems) == 0, "problem_count": len(problems), "warning_count": len(warnings), "status_counts": status_counts, "domain_counts": dict(sorted(domain_counts.items())), "problems": problems, "warnings": warnings, "routes": sorted(routes.keys()), "technique_count": len(techniques), } def build_audit_table(registry: Dict[str, Any], route_id: str | None = None) -> Dict[str, Any]: techniques = registry.get("techniques", {}) route = registry.get("routes", {}).get(route_id, {}) if route_id else {} if route_id and not route: raise KeyError(f"Unknown route: {route_id}") if route: ordered_ids = list(dict.fromkeys(route.get("required_techniques", []) + route.get("optional_techniques", []))) else: ordered_ids = sorted(techniques.keys()) rows = [] for tech_id in ordered_ids: tech = techniques[tech_id] rows.append({ "technique_id": tech_id, "audit_label": tech.get("audit_label"), "status": tech.get("status"), "commands": tech.get("commands", []), "output_paths": tech.get("output_paths", []), "missing_impact": tech.get("missing_impact"), "limitation": tech.get("limitation"), }) return {"route": route_id or "all", "rows": rows} def main() -> int: parser = argparse.ArgumentParser(description="Validate Jyotish technique registry and emit audit data") parser.add_argument("--registry", default=DEFAULT_REGISTRY, help="Path to technique_registry.json") parser.add_argument("--route", default=None, help="Optional route id for audit table") parser.add_argument("--mode", choices=["validate", "table"], default="validate") args = parser.parse_args() try: registry = load_registry(args.registry) if args.mode == "validate": result = validate_registry(registry) print(json.dumps(result, ensure_ascii=False, indent=2)) return 0 if result["valid"] else 1 result = build_audit_table(registry, args.route) print(json.dumps(result, ensure_ascii=False, indent=2)) return 0 except Exception as exc: print(json.dumps({"valid": False, "error": str(exc)}, ensure_ascii=False, indent=2), file=sys.stderr) return 1 if __name__ == "__main__": raise SystemExit(main())