195 lines
8.2 KiB
Python
195 lines
8.2 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
|
|
|
|
|
|
SCALAR_VALUE_TYPES = {"integer_atom", "scalar"}
|
|
SCALAR_REASONS = {"value_type_not_smoke_safe", "identity_or_binding_property"}
|
|
|
|
PROPERTY_ALIASES = {
|
|
"group": "Группа",
|
|
"id": "Идентификатор",
|
|
"name": "Имя",
|
|
"view": "Вид",
|
|
"title": "Заголовок",
|
|
"command_bar_location": "ПоложениеВКоманднойПанели",
|
|
}
|
|
|
|
MANUAL_ONLY_PROPERTIES = {"Идентификатор", "Имя", "ПутьКДанным", "Данные", "Вид"}
|
|
REFERENCE_PROPERTIES = {"Группа", "group"}
|
|
|
|
|
|
def load_entries(report: dict[str, Any]) -> list[dict[str, Any]]:
|
|
matrix = report.get("matrix") if isinstance(report.get("matrix"), dict) else {}
|
|
entries = matrix.get("entries") if isinstance(matrix.get("entries"), list) else []
|
|
return [entry for entry in entries if isinstance(entry, dict)]
|
|
|
|
|
|
def normalize_property_name(prop: dict[str, Any]) -> str:
|
|
raw = str(prop.get("semantic_name") or prop.get("canonical_property") or prop.get("property") or "")
|
|
return PROPERTY_ALIASES.get(raw, raw)
|
|
|
|
|
|
def scalar_bucket(entry: dict[str, Any]) -> str:
|
|
prop = entry.get("property") if isinstance(entry.get("property"), dict) else {}
|
|
target = entry.get("requested_target") if isinstance(entry.get("requested_target"), dict) else {}
|
|
name = normalize_property_name(prop)
|
|
marker = str(target.get("marker") or "")
|
|
parameter_index = prop.get("parameter_index")
|
|
semantic_status = str(prop.get("semantic_status") or "")
|
|
|
|
if name in MANUAL_ONLY_PROPERTIES or "маркер" in name.casefold():
|
|
return "manual_only_identity_or_structural"
|
|
if name in REFERENCE_PROPERTIES:
|
|
return "reference_or_container_rule_required"
|
|
if semantic_status and semantic_status != "unknown":
|
|
return "semantic_scalar_needs_allowed_values"
|
|
if parameter_index is not None:
|
|
return f"learn_marker_{marker}_parameter_{parameter_index}"
|
|
return "learn_named_scalar_semantics"
|
|
|
|
|
|
def recommended_action(bucket: str) -> str:
|
|
if bucket == "manual_only_identity_or_structural":
|
|
return "do_not_generic_write"
|
|
if bucket == "reference_or_container_rule_required":
|
|
return "learn_reference_write_rule"
|
|
if bucket == "semantic_scalar_needs_allowed_values":
|
|
return "collect_allowed_values_and_smoke"
|
|
if bucket.startswith("learn_marker_"):
|
|
return "run_before_after_learning_for_parameter"
|
|
return "run_before_after_learning_for_named_scalar"
|
|
|
|
|
|
def make_key(entry: dict[str, Any]) -> tuple[str, str, str, str, str]:
|
|
prop = entry.get("property") if isinstance(entry.get("property"), dict) else {}
|
|
target = 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 {}
|
|
return (
|
|
scalar_bucket(entry),
|
|
normalize_property_name(prop),
|
|
str(target.get("marker") or effective.get("marker") or ""),
|
|
str(prop.get("parameter_index") if prop.get("parameter_index") is not None else ""),
|
|
str(prop.get("value_type") or ""),
|
|
)
|
|
|
|
|
|
def main() -> int:
|
|
parser = argparse.ArgumentParser(description="Build scalar learning registry from 1C write matrix gaps.")
|
|
parser.add_argument("--matrix-report", type=Path, required=True, help="Report produced by scripts/smoke_1c_write_matrix.py.")
|
|
parser.add_argument("--output", type=Path, required=True, help="Output scalar registry JSON path.")
|
|
parser.add_argument("--sample-limit", type=int, default=8, help="Examples per scalar group.")
|
|
args = parser.parse_args()
|
|
|
|
report = json.loads(args.matrix_report.read_text(encoding="utf-8"))
|
|
groups: dict[tuple[str, str, str, str, str], dict[str, Any]] = {}
|
|
skipped = Counter()
|
|
|
|
for entry in load_entries(report):
|
|
prop = entry.get("property") if isinstance(entry.get("property"), dict) else {}
|
|
if entry.get("can_smoke"):
|
|
skipped["can_smoke"] += 1
|
|
continue
|
|
reason = str(entry.get("reason") or "")
|
|
value_type = str(prop.get("value_type") or "")
|
|
if reason not in SCALAR_REASONS:
|
|
skipped[f"reason:{reason}"] += 1
|
|
continue
|
|
if value_type not in SCALAR_VALUE_TYPES:
|
|
skipped[f"value_type:{value_type}"] += 1
|
|
continue
|
|
|
|
key = make_key(entry)
|
|
bucket, name, marker, parameter_index, _ = key
|
|
row = groups.setdefault(
|
|
key,
|
|
{
|
|
"bucket": bucket,
|
|
"property": name,
|
|
"marker": marker or None,
|
|
"parameter_index": parameter_index or None,
|
|
"value_type": value_type,
|
|
"recommended_action": recommended_action(bucket),
|
|
"observed_values": Counter(),
|
|
"reasons": Counter(),
|
|
"sections": Counter(),
|
|
"type_names": Counter(),
|
|
"examples": [],
|
|
},
|
|
)
|
|
target = 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 {}
|
|
old = "" if prop.get("old") is None else str(prop.get("old"))
|
|
row["observed_values"][old] += 1
|
|
row["reasons"][reason] += 1
|
|
row["sections"][str(effective.get("section") or target.get("section") or "")] += 1
|
|
row["type_names"][str(effective.get("type_name") or target.get("type_name") or "")] += 1
|
|
if len(row["examples"]) < args.sample_limit:
|
|
row["examples"].append(
|
|
{
|
|
"target": target.get("name") or target.get("path"),
|
|
"requested_section": target.get("section"),
|
|
"effective_section": effective.get("section"),
|
|
"type_name": effective.get("type_name") or target.get("type_name"),
|
|
"presentation": prop.get("presentation"),
|
|
"semantic_name": prop.get("semantic_name"),
|
|
"semantic_group": prop.get("semantic_group"),
|
|
"old": old,
|
|
"write_path": prop.get("write_path"),
|
|
"reason": reason,
|
|
}
|
|
)
|
|
|
|
scalars = []
|
|
for row in groups.values():
|
|
scalars.append(
|
|
{
|
|
**{key: value for key, value in row.items() if key not in {"observed_values", "reasons", "sections", "type_names"}},
|
|
"observed_values": dict(row["observed_values"].most_common()),
|
|
"reasons": dict(row["reasons"]),
|
|
"sections": dict(row["sections"]),
|
|
"type_names": dict(row["type_names"].most_common()),
|
|
"counts": {
|
|
"entries": sum(row["observed_values"].values()),
|
|
"observed_values": len(row["observed_values"]),
|
|
"examples": len(row["examples"]),
|
|
},
|
|
}
|
|
)
|
|
|
|
scalars.sort(
|
|
key=lambda item: (
|
|
str(item.get("recommended_action")),
|
|
-int(item["counts"]["entries"]),
|
|
str(item.get("marker")),
|
|
str(item.get("parameter_index")),
|
|
str(item.get("property")),
|
|
)
|
|
)
|
|
result = {
|
|
"schema": "onec_form_write_scalar_registry.v1",
|
|
"status": "ok",
|
|
"source_report": str(args.matrix_report),
|
|
"counts": {
|
|
"groups": len(scalars),
|
|
"entries": sum(int(item["counts"]["entries"]) for item in scalars),
|
|
"by_action": dict(Counter(str(item.get("recommended_action")) for item in scalars)),
|
|
"by_bucket": dict(Counter(str(item.get("bucket")) for item in scalars)),
|
|
"skipped": dict(skipped),
|
|
},
|
|
"scalars": scalars,
|
|
}
|
|
args.output.parent.mkdir(parents=True, exist_ok=True)
|
|
args.output.write_text(json.dumps(result, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
|
|
print(json.dumps({"schema": result["schema"], "status": "ok", "counts": result["counts"], "path": str(args.output)}, ensure_ascii=False, indent=2))
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|