157 lines
7.1 KiB
Python
157 lines
7.1 KiB
Python
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
from datetime import UTC, datetime
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
|
|
def read_json(path: Path) -> dict[str, Any]:
|
|
return json.loads(path.read_text(encoding="utf-8"))
|
|
|
|
|
|
def rule_status(rule: dict[str, Any]) -> str:
|
|
confidence = str(rule.get("confidence") or "none")
|
|
evidence = rule.get("evidence") if isinstance(rule.get("evidence"), dict) else {}
|
|
ok = evidence.get("ok")
|
|
total = evidence.get("total")
|
|
if confidence == "high" and isinstance(ok, int) and isinstance(total, int) and total > 0 and ok == total:
|
|
return "verified_read"
|
|
if confidence == "high" and int(evidence.get("distinct_rectangular_samples") or 0) > 0 and int(evidence.get("samples") or 0) > 0:
|
|
return "verified_read"
|
|
if confidence in {"medium", "high"}:
|
|
return "candidate_read"
|
|
return "needs_more_evidence"
|
|
|
|
|
|
def write_status(status: str) -> str:
|
|
if status == "verified_read":
|
|
return "blocked_until_roundtrip"
|
|
return "blocked_until_verified_read"
|
|
|
|
|
|
def merge_named_range_analysis_rules(discovery: dict[str, Any], named_range_analysis: dict[str, Any] | None) -> dict[str, Any]:
|
|
if not named_range_analysis:
|
|
return discovery
|
|
merged_rules = [rule for rule in discovery.get("rules") or [] if isinstance(rule, dict)]
|
|
by_target = {str(rule.get("target") or ""): index for index, rule in enumerate(merged_rules)}
|
|
for rule in named_range_analysis.get("rules") or []:
|
|
if not isinstance(rule, dict) or not str(rule.get("target") or "").startswith("moxel.named_range."):
|
|
continue
|
|
target = str(rule.get("target") or "")
|
|
promoted = {
|
|
"id": rule.get("id") or target,
|
|
"target": target,
|
|
"expression": rule.get("expression"),
|
|
"raw_scalar_indexes": rule.get("raw_scalar_indexes"),
|
|
"confidence": rule.get("confidence") or "none",
|
|
"evidence": rule.get("evidence") or {},
|
|
"source": "named_range_analysis",
|
|
}
|
|
if target in by_target:
|
|
existing = merged_rules[by_target[target]]
|
|
confidence_order = {"none": 0, "low": 1, "medium": 2, "high": 3}
|
|
if confidence_order.get(str(promoted.get("confidence")), 0) >= confidence_order.get(str(existing.get("confidence")), 0):
|
|
merged_rules[by_target[target]] = {**existing, **promoted}
|
|
else:
|
|
by_target[target] = len(merged_rules)
|
|
merged_rules.append(promoted)
|
|
return {**discovery, "rules": merged_rules}
|
|
|
|
|
|
def build_registry(discovery: dict[str, Any], sources: list[str], named_range_analysis: dict[str, Any] | None = None) -> dict[str, Any]:
|
|
discovery = merge_named_range_analysis_rules(discovery, named_range_analysis)
|
|
registry_rules = []
|
|
for index, rule in enumerate(discovery.get("rules") or [], start=1):
|
|
if not isinstance(rule, dict):
|
|
continue
|
|
status = rule_status(rule)
|
|
registry_rules.append(
|
|
{
|
|
"id": rule.get("id") or f"moxel_rule_{index}",
|
|
"target": rule.get("target"),
|
|
"expression": rule.get("expression"),
|
|
"raw_scalar_indexes": rule.get("raw_scalar_indexes"),
|
|
"confidence": rule.get("confidence") or "none",
|
|
"read_status": status,
|
|
"write_status": write_status(status),
|
|
"evidence": rule.get("evidence") or {},
|
|
"source_rule": rule,
|
|
}
|
|
)
|
|
return {
|
|
"schema": "codex_1c_moxel_schema_registry.v1",
|
|
"generated_at": datetime.now(UTC).isoformat(timespec="seconds").replace("+00:00", "Z"),
|
|
"sources": sources,
|
|
"policy": {
|
|
"read_use": "Only verified_read rules may be used as decoder behavior without additional diagnostics.",
|
|
"write_use": "All MOXCEL write rules are blocked until a disposable-base round-trip proves exact behavior.",
|
|
},
|
|
"rules": registry_rules,
|
|
"counts": {
|
|
"rules": len(registry_rules),
|
|
"verified_read": sum(1 for rule in registry_rules if rule.get("read_status") == "verified_read"),
|
|
"candidate_read": sum(1 for rule in registry_rules if rule.get("read_status") == "candidate_read"),
|
|
"write_enabled": sum(1 for rule in registry_rules if rule.get("write_status") == "verified_roundtrip"),
|
|
},
|
|
}
|
|
|
|
|
|
def render_markdown(registry: dict[str, Any]) -> str:
|
|
lines = ["# 1C MOXCEL Schema Registry", ""]
|
|
counts = registry.get("counts") or {}
|
|
lines.append(f"- Rules: `{counts.get('rules')}`")
|
|
lines.append(f"- Verified read: `{counts.get('verified_read')}`")
|
|
lines.append(f"- Candidate read: `{counts.get('candidate_read')}`")
|
|
lines.append(f"- Write enabled: `{counts.get('write_enabled')}`")
|
|
lines.append("")
|
|
lines.append("| Rule | Target | Read | Write | Confidence |")
|
|
lines.append("| --- | --- | --- | --- | --- |")
|
|
for rule in registry.get("rules") or []:
|
|
lines.append(
|
|
f"| `{rule.get('id')}` | `{rule.get('target')}` | `{rule.get('read_status')}` | "
|
|
f"`{rule.get('write_status')}` | `{rule.get('confidence')}` |"
|
|
)
|
|
lines.append("")
|
|
return "\n".join(lines)
|
|
|
|
|
|
def main() -> int:
|
|
parser = argparse.ArgumentParser(description="Build the stable 1C MOXCEL schema registry from discovery reports.")
|
|
parser.add_argument("--discovery", action="append", required=True, help="Discovery JSON. Repeatable; rules are merged in order.")
|
|
parser.add_argument("--named-range-analysis", help="Optional named range rule analysis JSON.")
|
|
parser.add_argument("--output-json", default="plugins/1c/metadata/moxel-schema-registry.json")
|
|
parser.add_argument("--output-markdown", default="reports/1c-template-baselines/moxel-schema-registry.md")
|
|
args = parser.parse_args()
|
|
|
|
discoveries = [read_json(Path(path)) for path in args.discovery]
|
|
merged = {"rules": []}
|
|
seen: set[tuple[str, str]] = set()
|
|
for discovery in discoveries:
|
|
for rule in discovery.get("rules") or []:
|
|
if not isinstance(rule, dict):
|
|
continue
|
|
key = (str(rule.get("id") or ""), str(rule.get("target") or ""))
|
|
if key in seen:
|
|
continue
|
|
seen.add(key)
|
|
merged["rules"].append(rule)
|
|
named_range_analysis = read_json(Path(args.named_range_analysis)) if args.named_range_analysis else None
|
|
sources = list(args.discovery)
|
|
if args.named_range_analysis:
|
|
sources.append(args.named_range_analysis)
|
|
registry = build_registry(merged, sources, named_range_analysis)
|
|
json_path = Path(args.output_json)
|
|
md_path = Path(args.output_markdown)
|
|
json_path.parent.mkdir(parents=True, exist_ok=True)
|
|
md_path.parent.mkdir(parents=True, exist_ok=True)
|
|
json_path.write_text(json.dumps(registry, ensure_ascii=False, indent=2), encoding="utf-8")
|
|
md_path.write_text(render_markdown(registry), encoding="utf-8")
|
|
print(json.dumps({"status": "ok", "json": str(json_path), "markdown": str(md_path), "counts": registry["counts"]}, ensure_ascii=False, indent=2))
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|