Jesse_Chen
178 lines
7.1 KiB
Python
178 lines
7.1 KiB
Python
#!/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())
|