115 lines
4.9 KiB
Python
115 lines
4.9 KiB
Python
#!/usr/bin/env python3
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
|
|
ACTION_PRIORITY = {
|
|
"run_before_after_learning_for_parameter": 10,
|
|
"collect_allowed_values_and_smoke": 20,
|
|
"run_before_after_learning_for_named_scalar": 30,
|
|
"learn_reference_write_rule": 80,
|
|
"do_not_generic_write": 100,
|
|
}
|
|
|
|
|
|
def load_rows(path: Path) -> tuple[str, list[dict[str, Any]]]:
|
|
data = json.loads(path.read_text(encoding="utf-8"))
|
|
schema = str(data.get("schema") or "")
|
|
if schema == "onec_form_write_scalar_registry.v1":
|
|
return schema, [row for row in data.get("scalars", []) if isinstance(row, dict)]
|
|
if schema == "onec_form_write_enum_registry.v1":
|
|
return schema, [row for row in data.get("properties", []) if isinstance(row, dict)]
|
|
raise SystemExit(f"Unsupported registry schema: {schema or '<missing>'}")
|
|
|
|
|
|
def example_selector(example: dict[str, Any]) -> dict[str, Any]:
|
|
target = example.get("target")
|
|
section = example.get("effective_section") or example.get("requested_section")
|
|
selector: dict[str, Any] = {}
|
|
if section == "commands":
|
|
selector["command"] = target
|
|
elif section == "attributes":
|
|
selector["attribute"] = target
|
|
else:
|
|
selector["element"] = target
|
|
return selector
|
|
|
|
|
|
def learning_case(row: dict[str, Any], index: int) -> dict[str, Any]:
|
|
examples = row.get("examples") if isinstance(row.get("examples"), list) else []
|
|
example = examples[0] if examples and isinstance(examples[0], dict) else {}
|
|
counts = row.get("counts") if isinstance(row.get("counts"), dict) else {}
|
|
action = str(row.get("recommended_action") or row.get("risk") or "")
|
|
return {
|
|
"id": f"learn-{index:03d}",
|
|
"action": action,
|
|
"property": row.get("property"),
|
|
"marker": row.get("marker"),
|
|
"parameter_index": row.get("parameter_index"),
|
|
"value_type": row.get("value_type"),
|
|
"entries": counts.get("entries"),
|
|
"observed_values": row.get("observed_values"),
|
|
"selector": example_selector(example),
|
|
"write_path": example.get("write_path"),
|
|
"current_value": example.get("old"),
|
|
"manual_step": {
|
|
"target": example.get("target"),
|
|
"section": example.get("effective_section") or example.get("requested_section"),
|
|
"presentation": example.get("presentation"),
|
|
"instruction": "Измени это свойство в конфигураторе на другое допустимое значение, сохрани форму, затем запусти capture_after/diff/infer.",
|
|
},
|
|
}
|
|
|
|
|
|
def main() -> int:
|
|
parser = argparse.ArgumentParser(description="Build ordered before/after learning plan from 1C write registries.")
|
|
parser.add_argument("--registry", type=Path, required=True, help="Scalar or enum registry JSON path.")
|
|
parser.add_argument("--output", type=Path, required=True, help="Output learning plan JSON path.")
|
|
parser.add_argument("--limit", type=int, default=50, help="Maximum cases to include.")
|
|
parser.add_argument(
|
|
"--include-actions",
|
|
nargs="*",
|
|
default=["run_before_after_learning_for_parameter", "collect_allowed_values_and_smoke", "run_before_after_learning_for_named_scalar"],
|
|
help="Recommended actions/risks to include.",
|
|
)
|
|
args = parser.parse_args()
|
|
|
|
schema, rows = load_rows(args.registry)
|
|
include = set(args.include_actions)
|
|
filtered = [row for row in rows if str(row.get("recommended_action") or row.get("risk") or "") in include]
|
|
filtered.sort(
|
|
key=lambda row: (
|
|
ACTION_PRIORITY.get(str(row.get("recommended_action") or row.get("risk") or ""), 50),
|
|
-int((row.get("counts") if isinstance(row.get("counts"), dict) else {}).get("entries") or 0),
|
|
str(row.get("marker")),
|
|
str(row.get("parameter_index")),
|
|
str(row.get("property")),
|
|
)
|
|
)
|
|
cases = [learning_case(row, index + 1) for index, row in enumerate(filtered[: args.limit])]
|
|
result = {
|
|
"schema": "onec_form_write_learning_plan.v1",
|
|
"status": "ok",
|
|
"source_registry": str(args.registry),
|
|
"source_schema": schema,
|
|
"counts": {
|
|
"cases": len(cases),
|
|
"available": len(filtered),
|
|
"included_actions": sorted(include),
|
|
},
|
|
"workflow": ["capture_before", "manual_configurator_change", "capture_after", "diff", "infer_rule", "smoke_rule"],
|
|
"cases": cases,
|
|
}
|
|
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())
|