Files
llm/scripts/verify_1c_moxel_schema_registry.py

161 lines
6.3 KiB
Python

from __future__ import annotations
import argparse
import json
import re
from pathlib import Path
from typing import Any
MARKER_RE = re.compile(r"^(\d+)-(\d+)$")
def read_json(path: Path) -> dict[str, Any]:
return json.loads(path.read_text(encoding="utf-8"))
def as_int(value: Any) -> int | None:
try:
return int(value)
except (TypeError, ValueError):
return None
def marker_cells(probe: dict[str, Any]) -> list[dict[str, Any]]:
data = probe.get("probe") if isinstance(probe.get("probe"), dict) else probe
cells = data.get("cells") if isinstance(data.get("cells"), list) else []
result = []
for cell in cells:
if not isinstance(cell, dict):
continue
match = MARKER_RE.match(str(cell.get("text") or ""))
if not match:
continue
expected_row, expected_col = map(int, match.groups())
one_based = cell.get("one_based") if isinstance(cell.get("one_based"), dict) else {}
result.append(
{
"text": cell.get("text"),
"expected_row": expected_row,
"expected_col": expected_col,
"decoded_row": one_based.get("row"),
"decoded_col": one_based.get("column"),
"ok": one_based.get("row") == expected_row and one_based.get("column") == expected_col,
}
)
return result
def verify_inline_column_rule(probe: dict[str, Any]) -> dict[str, Any]:
samples = marker_cells(probe)
ok = sum(1 for sample in samples if sample.get("ok") is True)
return {
"target": "moxel.inline_text_cell.column",
"status": "ok" if samples and ok == len(samples) else "failed",
"evidence": {"ok": ok, "total": len(samples)},
"failed_samples": [sample for sample in samples if sample.get("ok") is not True],
}
def named_ranges(probe: dict[str, Any]) -> list[dict[str, Any]]:
data = probe.get("probe") if isinstance(probe.get("probe"), dict) else probe
ranges = data.get("named_ranges") or data.get("named_range_candidates") or []
return [item for item in ranges if isinstance(item, dict)]
def raw_scalar(raw: list[Any], index: int) -> int | None:
if index < 0 or index >= len(raw):
return None
return as_int(raw[index])
def verify_named_range_candidate(rule: dict[str, Any], probe: dict[str, Any]) -> dict[str, Any]:
target = str(rule.get("target") or "")
field = target.rsplit(".", 1)[-1]
indexes = rule.get("raw_scalar_indexes") if isinstance(rule.get("raw_scalar_indexes"), list) else []
samples = []
for item in named_ranges(probe):
range_info = item.get("range") if isinstance(item.get("range"), dict) else {}
one_based = range_info.get("one_based") if isinstance(range_info.get("one_based"), dict) else {}
expected = as_int(one_based.get(field))
raw = item.get("raw_scalars") if isinstance(item.get("raw_scalars"), list) else []
if expected is None or not raw:
continue
predictions = []
for index in indexes:
index_int = as_int(index)
if index_int is None:
continue
value = raw_scalar(raw, index_int)
predictions.append({"index": index_int, "value": value, "one_based": value + 1 if value is not None else None})
matching = [prediction for prediction in predictions if prediction.get("one_based") == expected]
samples.append(
{
"name": item.get("name"),
"kind": item.get("kind"),
"field": field,
"expected": expected,
"predictions": predictions,
"ok": bool(matching),
}
)
ok = sum(1 for sample in samples if sample.get("ok") is True)
return {
"target": target,
"status": "ok" if samples and ok == len(samples) else "inconclusive" if not samples else "partial",
"evidence": {"ok": ok, "total": len(samples)},
"samples": samples,
}
def verify_registry(registry: dict[str, Any], probes: list[dict[str, Any]]) -> dict[str, Any]:
results = []
for rule in registry.get("rules") or []:
if not isinstance(rule, dict):
continue
target = rule.get("target")
per_probe = []
for probe in probes:
if target == "moxel.inline_text_cell.column":
per_probe.append(verify_inline_column_rule(probe))
elif str(target or "").startswith("moxel.named_range."):
per_probe.append(verify_named_range_candidate(rule, probe))
if not per_probe:
continue
status = "ok" if all(item.get("status") == "ok" for item in per_probe) else "partial" if any(item.get("status") in {"ok", "partial"} for item in per_probe) else "failed"
results.append({"rule_id": rule.get("id"), "target": target, "registry_read_status": rule.get("read_status"), "verification_status": status, "probes": per_probe})
failures = [
item
for item in results
if item.get("registry_read_status") == "verified_read" and item.get("verification_status") != "ok"
]
return {
"schema": "codex_1c_moxel_schema_registry_verification.v1",
"status": "ok" if not failures else "failed",
"results": results,
"failures": failures,
"counts": {
"rules_checked": len(results),
"verified_read_failures": len(failures),
},
}
def main() -> int:
parser = argparse.ArgumentParser(description="Verify 1C MOXCEL schema registry rules against probe snapshots.")
parser.add_argument("--registry", default="plugins/1c/metadata/moxel-schema-registry.json")
parser.add_argument("--probe", action="append", required=True, help="Probe snapshot JSON. Repeatable.")
parser.add_argument("--output", default="reports/1c-template-baselines/moxel-schema-registry-verification.json")
args = parser.parse_args()
report = verify_registry(read_json(Path(args.registry)), [read_json(Path(path)) for path in args.probe])
output_path = Path(args.output)
output_path.parent.mkdir(parents=True, exist_ok=True)
output_path.write_text(json.dumps(report, ensure_ascii=False, indent=2), encoding="utf-8")
print(json.dumps(report, ensure_ascii=False, indent=2))
return 0 if report["status"] == "ok" else 1
if __name__ == "__main__":
raise SystemExit(main())