189 lines
7.9 KiB
Python
189 lines
7.9 KiB
Python
#!/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())
|