307 lines
13 KiB
Python
307 lines
13 KiB
Python
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
|
|
VOLATILE_KEYS = {
|
|
"captured_at",
|
|
"adapter_url",
|
|
"modified",
|
|
"bytes",
|
|
"file_name",
|
|
"template_file",
|
|
"label",
|
|
"diff",
|
|
"cell_id",
|
|
}
|
|
|
|
|
|
def read_json(path: Path) -> dict[str, Any]:
|
|
return json.loads(path.read_text(encoding="utf-8"))
|
|
|
|
|
|
def resolve_manifest_path(value: str, base_dir: Path) -> Path:
|
|
path = Path(value)
|
|
if path.is_absolute():
|
|
return path
|
|
candidates = [
|
|
base_dir / path,
|
|
ROOT / path,
|
|
Path.cwd() / path,
|
|
path,
|
|
]
|
|
for candidate in candidates:
|
|
resolved = candidate.resolve()
|
|
if resolved.exists():
|
|
return resolved
|
|
return (base_dir / path).resolve()
|
|
|
|
|
|
def unwrap_structure(payload: dict[str, Any]) -> dict[str, Any]:
|
|
if isinstance(payload.get("probe"), dict):
|
|
return payload["probe"]
|
|
if isinstance(payload.get("structure"), dict):
|
|
return payload["structure"]
|
|
return payload
|
|
|
|
|
|
def compact_next_record(value: Any) -> Any:
|
|
if not isinstance(value, dict):
|
|
return value
|
|
result = {}
|
|
for key in ("type", "value", "head", "scalar_prefix", "list_length", "tree_position"):
|
|
if key in value:
|
|
result[key] = value[key]
|
|
return result
|
|
|
|
|
|
def normalize_item(item: Any) -> Any:
|
|
if not isinstance(item, dict):
|
|
return item
|
|
result: dict[str, Any] = {}
|
|
for key, value in item.items():
|
|
if key in VOLATILE_KEYS:
|
|
continue
|
|
if key == "next_moxel_record":
|
|
result[key] = compact_next_record(value)
|
|
elif key == "style_evidence" and isinstance(value, dict):
|
|
result[key] = {
|
|
style_key: style_value
|
|
for style_key, style_value in value.items()
|
|
if style_key in {"immediate_preceding_values", "last_7_preceding_values"}
|
|
}
|
|
elif isinstance(value, dict):
|
|
result[key] = normalize_item(value)
|
|
elif isinstance(value, list):
|
|
result[key] = [normalize_item(child) for child in value]
|
|
else:
|
|
result[key] = value
|
|
return result
|
|
|
|
|
|
def stable_key(item: dict[str, Any], fallback_index: int) -> str:
|
|
for key in ("text", "name"):
|
|
if item.get(key) not in {None, ""}:
|
|
return f"{key}:{item.get(key)}"
|
|
if item.get("tree_position"):
|
|
return f"tree:{item.get('tree_position')}"
|
|
if item.get("one_based"):
|
|
return f"cell:{json.dumps(item.get('one_based'), ensure_ascii=False, sort_keys=True)}"
|
|
return f"index:{fallback_index}"
|
|
|
|
|
|
def normalize_section_list(items: Any) -> dict[str, Any]:
|
|
if not isinstance(items, list):
|
|
return {}
|
|
result: dict[str, Any] = {}
|
|
for index, item in enumerate(items):
|
|
if not isinstance(item, dict):
|
|
result[f"index:{index}"] = normalize_item(item)
|
|
continue
|
|
key = stable_key(item, index)
|
|
if key in result:
|
|
key = f"{key}#{index}"
|
|
result[key] = normalize_item(item)
|
|
return result
|
|
|
|
|
|
def normalized_structure(payload: dict[str, Any], *, target_text: str | None = None, target_name: str | None = None) -> dict[str, Any]:
|
|
structure = unwrap_structure(payload)
|
|
result: dict[str, Any] = {
|
|
"counts": normalize_item(structure.get("counts") or {}),
|
|
"dimensions": normalize_item(structure.get("dimensions") or {}),
|
|
"cells": normalize_section_list(structure.get("cells") or []),
|
|
"cell_style_candidates": normalize_section_list(structure.get("cell_style_candidates") or structure.get("cell_styles") or []),
|
|
"named_range_candidates": normalize_section_list(structure.get("named_range_candidates") or structure.get("named_ranges") or []),
|
|
"named_areas": normalize_section_list(structure.get("named_areas") or []),
|
|
"column_widths": normalize_section_list(structure.get("column_widths") or []),
|
|
"row_heights": normalize_section_list(structure.get("row_heights") or []),
|
|
"merged_ranges": normalize_section_list(structure.get("merged_ranges") or []),
|
|
"merged_range_candidates": normalize_section_list(structure.get("merged_range_candidates") or []),
|
|
}
|
|
if target_text:
|
|
result["target_cell_styles"] = {
|
|
key: value
|
|
for key, value in result["cell_style_candidates"].items()
|
|
if isinstance(value, dict) and str(value.get("text") or "") == target_text
|
|
}
|
|
result["target_cells"] = {
|
|
key: value
|
|
for key, value in result["cells"].items()
|
|
if isinstance(value, dict) and str(value.get("text") or "") == target_text
|
|
}
|
|
if target_name:
|
|
result["target_named_ranges"] = {
|
|
key: value
|
|
for key, value in result["named_range_candidates"].items()
|
|
if isinstance(value, dict) and str(value.get("name") or "") == target_name
|
|
}
|
|
return result
|
|
|
|
|
|
def diff_values(before: Any, after: Any, path: str = "$") -> list[dict[str, Any]]:
|
|
if before == after:
|
|
return []
|
|
if isinstance(before, dict) and isinstance(after, dict):
|
|
changes: list[dict[str, Any]] = []
|
|
for key in sorted(set(before) | set(after)):
|
|
changes.extend(diff_values(before.get(key), after.get(key), f"{path}.{key}"))
|
|
return changes
|
|
if isinstance(before, list) and isinstance(after, list):
|
|
changes = []
|
|
for index in range(max(len(before), len(after))):
|
|
old = before[index] if index < len(before) else None
|
|
new = after[index] if index < len(after) else None
|
|
changes.extend(diff_values(old, new, f"{path}[{index}]"))
|
|
return changes
|
|
return [{"path": path, "before": before, "after": after}]
|
|
|
|
|
|
def score_change(change: dict[str, Any], target_text: str | None, target_name: str | None) -> int:
|
|
path = str(change.get("path") or "")
|
|
score = 0
|
|
if "target_" in path:
|
|
score += 40
|
|
if target_text and target_text in path:
|
|
score += 30
|
|
if target_name and target_name in path:
|
|
score += 30
|
|
if any(part in path for part in ("next_moxel_record", "style_evidence", "raw_scalars", "column_widths", "row_heights", "merged")):
|
|
score += 15
|
|
if ".counts." in path:
|
|
score -= 20
|
|
if ".tree_position" in path:
|
|
score -= 10
|
|
if path.endswith(".cell_id"):
|
|
score -= 20
|
|
if change.get("before") is None or change.get("after") is None:
|
|
score -= 5
|
|
return score
|
|
|
|
|
|
def analyze_experiment(experiment: dict[str, Any], base_dir: Path) -> dict[str, Any]:
|
|
before_path = resolve_manifest_path(str(experiment["before"]), base_dir)
|
|
after_path = resolve_manifest_path(str(experiment["after"]), base_dir)
|
|
target_text = experiment.get("target_text")
|
|
target_name = experiment.get("target_name")
|
|
before = normalized_structure(read_json(before_path), target_text=target_text, target_name=target_name)
|
|
after = normalized_structure(read_json(after_path), target_text=target_text, target_name=target_name)
|
|
changes = diff_values(before, after)
|
|
scored = sorted(
|
|
(
|
|
{
|
|
**change,
|
|
"score": score_change(change, str(target_text) if target_text else None, str(target_name) if target_name else None),
|
|
}
|
|
for change in changes
|
|
),
|
|
key=lambda item: (-int(item.get("score") or 0), str(item.get("path") or "")),
|
|
)
|
|
min_positive = [item for item in scored if int(item.get("score") or 0) > 0]
|
|
candidates = min_positive[: int(experiment.get("max_candidates") or 20)]
|
|
confidence = "none"
|
|
if len(candidates) == 1 and candidates[0]["score"] >= 40:
|
|
confidence = "high"
|
|
elif candidates and candidates[0]["score"] >= 40:
|
|
confidence = "medium"
|
|
elif candidates:
|
|
confidence = "low"
|
|
return {
|
|
"property": experiment.get("property"),
|
|
"operation": experiment.get("operation"),
|
|
"target_text": target_text,
|
|
"target_name": target_name,
|
|
"before": str(before_path),
|
|
"after": str(after_path),
|
|
"confidence": confidence,
|
|
"candidate_paths": candidates,
|
|
"counts": {"changes": len(changes), "candidate_paths": len(candidates)},
|
|
}
|
|
|
|
|
|
def default_probe_plan() -> list[dict[str, Any]]:
|
|
return [
|
|
{"property": "ГоризонтальноеПоложение", "values": ["Лево", "Центр", "Право"], "target": "cell"},
|
|
{"property": "ВертикальноеПоложение", "values": ["Верх", "Центр", "Низ"], "target": "cell"},
|
|
{"property": "ЦветТекста", "values": ["Черный", "Красный", "Синий"], "target": "cell"},
|
|
{"property": "ЦветФона", "values": ["Нет", "Желтый", "Серый"], "target": "cell"},
|
|
{"property": "Шрифт.Имя", "values": ["Arial", "Courier New"], "target": "cell"},
|
|
{"property": "Шрифт.Размер", "values": [8, 10, 14], "target": "cell"},
|
|
{"property": "ГраницаЛево", "values": ["Нет", "Тонкая", "Толстая"], "target": "cell"},
|
|
{"property": "ГраницаВерх", "values": ["Нет", "Тонкая", "Толстая"], "target": "cell"},
|
|
{"property": "Защита", "values": [True, False], "target": "cell"},
|
|
{"property": "Гиперссылка", "values": ["", "https://example.invalid/1c-moxel-probe"], "target": "cell"},
|
|
{"property": "Переносить", "values": [True, False], "target": "cell"},
|
|
{"property": "ШиринаКолонки", "values": [8, 12, 20], "target": "column"},
|
|
{"property": "ВысотаСтроки", "values": [12, 18, 24], "target": "row"},
|
|
{"property": "Объединение", "values": ["none", "R8C4:R8C5"], "target": "range"},
|
|
]
|
|
|
|
|
|
def render_markdown(payload: dict[str, Any]) -> str:
|
|
lines: list[str] = ["# 1C MOXCEL property experiments", ""]
|
|
lines.append(f"- Experiments: `{len(payload.get('experiments') or [])}`")
|
|
lines.append("")
|
|
if payload.get("probe_plan"):
|
|
lines.append("## Probe Plan")
|
|
lines.append("")
|
|
lines.append("| Property | Target | Values |")
|
|
lines.append("| --- | --- | --- |")
|
|
for item in payload["probe_plan"]:
|
|
lines.append(f"| `{item.get('property')}` | `{item.get('target')}` | `{json.dumps(item.get('values'), ensure_ascii=False)}` |")
|
|
lines.append("")
|
|
if payload.get("experiments"):
|
|
lines.append("## Results")
|
|
lines.append("")
|
|
lines.append("| Property | Confidence | Changes | Top path |")
|
|
lines.append("| --- | --- | --- | --- |")
|
|
for item in payload["experiments"]:
|
|
top = (item.get("candidate_paths") or [{}])[0]
|
|
lines.append(
|
|
f"| `{item.get('property')}` | `{item.get('confidence')}` | "
|
|
f"`{(item.get('counts') or {}).get('changes')}` | `{top.get('path') or ''}` |"
|
|
)
|
|
lines.append("")
|
|
return "\n".join(lines)
|
|
|
|
|
|
def main() -> int:
|
|
parser = argparse.ArgumentParser(description="Analyze controlled 1C MOXCEL one-property experiments.")
|
|
parser.add_argument("--manifest", help="Experiment manifest JSON.")
|
|
parser.add_argument("--output-json", default="reports/1c-template-baselines/Primer3_moxel_property_experiments.json")
|
|
parser.add_argument("--output-markdown", default="reports/1c-template-baselines/Primer3_moxel_property_experiments.md")
|
|
parser.add_argument("--emit-default-plan", action="store_true", help="Include the default next probe plan.")
|
|
args = parser.parse_args()
|
|
|
|
manifest_path = Path(args.manifest).resolve() if args.manifest else None
|
|
manifest = read_json(manifest_path) if manifest_path else {"experiments": []}
|
|
base_dir = manifest_path.parent if manifest_path else Path.cwd()
|
|
experiments = [
|
|
analyze_experiment(experiment, base_dir)
|
|
for experiment in manifest.get("experiments") or []
|
|
if isinstance(experiment, dict) and experiment.get("before") and experiment.get("after")
|
|
]
|
|
payload = {
|
|
"schema": "codex_1c_moxel_property_experiments.v1",
|
|
"manifest": str(manifest_path) if manifest_path else None,
|
|
"experiments": experiments,
|
|
"probe_plan": default_probe_plan() if args.emit_default_plan or not experiments else [],
|
|
}
|
|
json_path = Path(args.output_json)
|
|
md_path = Path(args.output_markdown)
|
|
json_path.parent.mkdir(parents=True, exist_ok=True)
|
|
md_path.parent.mkdir(parents=True, exist_ok=True)
|
|
json_path.write_text(json.dumps(payload, ensure_ascii=False, indent=2), encoding="utf-8")
|
|
md_path.write_text(render_markdown(payload), encoding="utf-8")
|
|
print(json.dumps({"status": "ok", "json": str(json_path), "markdown": str(md_path), "experiments": len(experiments)}, ensure_ascii=False, indent=2))
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|