#!/usr/bin/env python3 from __future__ import annotations import argparse import json from collections import Counter from pathlib import Path from typing import Any SOURCE_AWARE_ROUTE_EVIDENCE: list[dict[str, Any]] = [ { "key": "attributes|title|string|3.3.4.2.1|data_path_form_attribute_title", "requested_section": "items", "requested_name": "А", "requested_path": "1.25.24.24.24", "effective_section": "attributes", "effective_name": "А", "effective_path": "3.3", "source_kind": "data_path_form_attribute_title", "property": "title", "presentation": "Заголовок", "value_type": "string", "read_path": "1.25.24.24.24.4.2.1", "write_path": "3.3.4.2.1", "verification": "source_aware_readback", "status": "verified", "evidence": "metadata.write apply_and_rollback verified on upo_test; route smoke verified", }, { "key": "attribute_fields|title|string|3.6.14.4.2.1|data_path_form_attribute_field_title", "requested_section": "items", "requested_name": "ТЗК1", "requested_path": "1.25.24.26.68", "effective_section": "attribute_fields", "effective_name": "К1", "effective_path": "3.6.14", "source_kind": "data_path_form_attribute_field_title", "property": "title", "presentation": "Заголовок", "value_type": "string", "read_path": "1.25.24.26.68.4.2.1", "write_path": "3.6.14.4.2.1", "verification": "source_aware_readback", "status": "verified", "evidence": "metadata.write apply_and_rollback verified on upo_test; route smoke verified", }, { "key": "commands|title|string|5.3.3.2.1|local", "requested_section": "commands", "requested_name": "КомандаПример1", "requested_path": "5.3", "effective_section": "commands", "effective_name": "КомандаПример1", "effective_path": "5.3", "source_kind": "local", "property": "title", "presentation": "Заголовок", "value_type": "string", "read_path": "5.3.3.2.1", "write_path": "5.3.3.2.1", "verification": "source_aware_readback", "status": "verified", "evidence": "metadata.write apply_and_rollback verified on upo_test; route smoke verified", }, ] def entry_key(item: dict[str, Any]) -> str: return "|".join( str(item.get(key) or "") for key in ("effective_section", "property", "value_type", "write_path", "source_kind") ) def registry_entry(row: dict[str, Any]) -> dict[str, Any] | None: if row.get("status") != "verified": return None entry = row.get("entry") if isinstance(row.get("entry"), dict) else {} prop = entry.get("property") if isinstance(entry.get("property"), dict) else {} requested = entry.get("requested_target") if isinstance(entry.get("requested_target"), dict) else {} effective = entry.get("effective_target") if isinstance(entry.get("effective_target"), dict) else {} source = entry.get("effective_source") if isinstance(entry.get("effective_source"), dict) else {} item = { "requested_section": requested.get("section"), "requested_name": requested.get("name"), "requested_path": requested.get("path"), "effective_section": effective.get("section"), "effective_name": effective.get("name"), "effective_path": effective.get("path"), "source_kind": source.get("kind") or "local", "property": prop.get("semantic_name") or prop.get("canonical_property") or prop.get("property"), "canonical_property": prop.get("canonical_property") or prop.get("property"), "presentation": prop.get("presentation"), "semantic_name": prop.get("semantic_name"), "semantic_group": prop.get("semantic_group"), "semantic_source": prop.get("semantic_source"), "parameter_index": prop.get("parameter_index"), "value_type": prop.get("value_type"), "read_path": prop.get("read_path"), "write_path": prop.get("write_path"), "verification": prop.get("verification"), "status": "verified", } item["key"] = entry_key(item) return item def pattern_from_entry(item: dict[str, Any]) -> dict[str, Any]: pattern = { key: value for key, value in item.items() if key not in {"requested_name", "requested_path", "effective_name", "effective_path", "read_path", "evidence"} } pattern["verified_count"] = 0 pattern["examples"] = [] return pattern def add_pattern_example(patterns: dict[str, dict[str, Any]], item: dict[str, Any]) -> None: pattern = patterns.setdefault(item["key"], pattern_from_entry(item)) pattern["verified_count"] = int(pattern.get("verified_count") or 0) + 1 example = { "requested_name": item.get("requested_name"), "requested_path": item.get("requested_path"), "effective_name": item.get("effective_name"), "effective_path": item.get("effective_path"), "read_path": item.get("read_path"), } examples = pattern.setdefault("examples", []) if example not in examples and len(examples) < 5: examples.append(example) def main() -> int: parser = argparse.ArgumentParser(description="Build compact verified write registry from a write-matrix smoke report.") parser.add_argument("--smoke-report", type=Path, required=True, help="Full report produced by scripts/smoke_1c_write_matrix.py.") parser.add_argument("--output", type=Path, required=True, help="Output registry JSON path.") parser.add_argument("--include-route-evidence", action="store_true", help="Append known source-aware route evidence from the learning case.") args = parser.parse_args() data = json.loads(args.smoke_report.read_text(encoding="utf-8")) smoke = data.get("smoke") if isinstance(data.get("smoke"), dict) else {} entries: list[dict[str, Any]] = [] patterns: dict[str, dict[str, Any]] = {} for row in smoke.get("results") or []: if not isinstance(row, dict): continue item = registry_entry(row) if not item: continue entries.append(item) add_pattern_example(patterns, item) if args.include_route_evidence: existing = {item.get("key") for item in entries} for item in SOURCE_AWARE_ROUTE_EVIDENCE: if item["key"] not in existing: entries.append(dict(item)) existing.add(item["key"]) add_pattern_example(patterns, item) registry = { "schema": "onec_form_write_verified_registry.v1", "status": "ok", "source_report": str(args.smoke_report), "adapter_report_path": smoke.get("path"), "base_id": data.get("base_id"), "table": data.get("table"), "file_name": data.get("file_name"), "counts": { "verified_entries": len(entries), "verified_patterns": len(patterns), "by_effective_section": dict(sorted(Counter(item.get("effective_section") for item in entries).items())), "by_property": dict(sorted(Counter(item.get("property") for item in entries).items())), "by_source_kind": dict(sorted(Counter(item.get("source_kind") for item in entries).items())), }, "patterns": sorted(patterns.values(), key=lambda item: (str(item.get("effective_section")), str(item.get("property")), str(item.get("write_path")), str(item.get("source_kind")))), "entries": entries, } args.output.parent.mkdir(parents=True, exist_ok=True) args.output.write_text(json.dumps(registry, ensure_ascii=False, indent=2) + "\n", encoding="utf-8") print(json.dumps({"schema": registry["schema"], "status": "ok", "counts": registry["counts"], "path": str(args.output)}, ensure_ascii=False, indent=2)) return 0 if __name__ == "__main__": raise SystemExit(main())