354 lines
15 KiB
Python
354 lines
15 KiB
Python
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import re
|
|
from collections import defaultdict
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
from typing import Any
|
|
from urllib import request
|
|
|
|
|
|
MARKER_RE = re.compile(r"^\d+-\d+$")
|
|
|
|
|
|
def rpc(adapter_url: str, method: str, payload: dict[str, Any]) -> dict[str, Any]:
|
|
body = json.dumps({"method": method, "payload": payload}, ensure_ascii=False).encode("utf-8")
|
|
req = request.Request(
|
|
f"{adapter_url.rstrip('/')}/rpc",
|
|
data=body,
|
|
headers={"Content-Type": "application/json; charset=utf-8"},
|
|
method="POST",
|
|
)
|
|
with request.urlopen(req, timeout=240) as resp:
|
|
return json.loads(resp.read().decode("utf-8", errors="replace"))
|
|
|
|
|
|
def latest_configcas_rows(adapter_url: str, base_id: str, limit: int) -> list[dict[str, Any]]:
|
|
result = rpc(
|
|
adapter_url,
|
|
"query.run",
|
|
{
|
|
"base_id": base_id,
|
|
"diagnostic": True,
|
|
"query": (
|
|
f"SELECT TOP {int(limit)} FileName, DATALENGTH(BinaryData) AS Bytes, PartNo, Creation, Modified "
|
|
"FROM ConfigCAS ORDER BY Modified DESC"
|
|
),
|
|
"timeout_seconds": 120,
|
|
},
|
|
)
|
|
return result.get("rows") or []
|
|
|
|
|
|
def template_summary(adapter_url: str, base_id: str, file_name: str, max_cells: int) -> dict[str, Any]:
|
|
result = rpc(
|
|
adapter_url,
|
|
"templates.map",
|
|
{
|
|
"base_id": base_id,
|
|
"table": "ConfigCAS",
|
|
"file_name": file_name,
|
|
"view": "summary",
|
|
"sections": "cells,styles,named_areas,named_ranges,diagnostics",
|
|
"max_cells": max_cells,
|
|
"max_areas": 200,
|
|
"timeout_seconds": 120,
|
|
},
|
|
)
|
|
templates = result.get("templates") or []
|
|
if not templates:
|
|
return {}
|
|
return (templates[0] or {}).get("structure") or {}
|
|
|
|
|
|
def choose_latest_moxel(adapter_url: str, base_id: str, limit: int, max_cells: int, explicit_file_name: str | None) -> tuple[str, dict[str, Any], dict[str, Any]]:
|
|
if explicit_file_name:
|
|
structure = template_summary(adapter_url, base_id, explicit_file_name, max_cells)
|
|
return explicit_file_name, {"FileName": explicit_file_name}, structure
|
|
for row in latest_configcas_rows(adapter_url, base_id, limit):
|
|
file_name = str(row.get("FileName") or "")
|
|
byte_count = int(row.get("Bytes") or 0)
|
|
if not file_name or byte_count <= 0 or byte_count > 20000:
|
|
continue
|
|
structure = template_summary(adapter_url, base_id, file_name, max_cells)
|
|
if str(structure.get("format") or "") == "MOXCEL":
|
|
return file_name, row, structure
|
|
raise RuntimeError("Could not find a recent MOXCEL payload in ConfigCAS.")
|
|
|
|
|
|
def normalize_cell(cell: dict[str, Any]) -> dict[str, Any]:
|
|
return {
|
|
"text": cell.get("text"),
|
|
"cell_id": cell.get("cell_id"),
|
|
"type_code": cell.get("type_code"),
|
|
"one_based": cell.get("one_based"),
|
|
"zero_based": cell.get("zero_based"),
|
|
"parameter": cell.get("parameter"),
|
|
"reference": cell.get("reference"),
|
|
"source": cell.get("source"),
|
|
}
|
|
|
|
|
|
def normalize_style(item: dict[str, Any]) -> dict[str, Any]:
|
|
next_record = item.get("next_moxel_record") if isinstance(item.get("next_moxel_record"), dict) else None
|
|
style = item.get("style_evidence") if isinstance(item.get("style_evidence"), dict) else {}
|
|
return {
|
|
"text": item.get("text"),
|
|
"cell_id": item.get("cell_id"),
|
|
"type_code": item.get("type_code"),
|
|
"tree_position": item.get("tree_position"),
|
|
"next_moxel_record": next_record,
|
|
"immediate_preceding_values": style.get("immediate_preceding_values"),
|
|
"last_7_preceding_values": style.get("last_7_preceding_values"),
|
|
}
|
|
|
|
|
|
def build_snapshot(structure: dict[str, Any]) -> dict[str, Any]:
|
|
cells = [normalize_cell(item) for item in (structure.get("cells") or []) if isinstance(item, dict) and item.get("text")]
|
|
styles = [normalize_style(item) for item in (structure.get("cell_style_candidates") or []) if isinstance(item, dict) and item.get("text")]
|
|
named_ranges = []
|
|
for item in structure.get("named_range_candidates") or []:
|
|
if not isinstance(item, dict):
|
|
continue
|
|
named_ranges.append(
|
|
{
|
|
"name": item.get("name"),
|
|
"kind": item.get("kind"),
|
|
"tree_position": item.get("tree_position"),
|
|
"range": item.get("range"),
|
|
"raw_scalars": ((item.get("range_candidate") or {}).get("raw_scalars") if isinstance(item.get("range_candidate"), dict) else None),
|
|
}
|
|
)
|
|
|
|
cells_by_text: dict[str, list[dict[str, Any]]] = defaultdict(list)
|
|
for cell in cells:
|
|
cells_by_text[str(cell.get("text"))].append(cell)
|
|
|
|
styles_by_text: dict[str, list[dict[str, Any]]] = defaultdict(list)
|
|
for style in styles:
|
|
styles_by_text[str(style.get("text"))].append(style)
|
|
|
|
marker_matrix = []
|
|
for text, entries in sorted(cells_by_text.items(), key=lambda pair: tuple(map(int, pair[0].split("-"))) if MARKER_RE.fullmatch(pair[0]) else (10**9, 10**9)):
|
|
if not MARKER_RE.fullmatch(text):
|
|
continue
|
|
expected_row, expected_col = map(int, text.split("-"))
|
|
style_entries = styles_by_text.get(text) or []
|
|
row: dict[str, Any] = {
|
|
"text": text,
|
|
"expected_row": expected_row,
|
|
"expected_col": expected_col,
|
|
"cells": entries,
|
|
"styles": style_entries,
|
|
}
|
|
if entries:
|
|
first = entries[0]
|
|
one_based = first.get("one_based") or {}
|
|
decoded_row = one_based.get("row")
|
|
decoded_col = one_based.get("column")
|
|
row["decoded_row"] = decoded_row
|
|
row["decoded_col"] = decoded_col
|
|
row["row_ok"] = decoded_row == expected_row
|
|
row["col_ok"] = decoded_col == expected_col
|
|
if isinstance(decoded_col, int):
|
|
row["col_delta"] = decoded_col - expected_col
|
|
marker_matrix.append(row)
|
|
|
|
return {
|
|
"counts": structure.get("counts") or {},
|
|
"dimensions": structure.get("dimensions"),
|
|
"named_areas": structure.get("named_areas") or [],
|
|
"named_ranges": named_ranges,
|
|
"cells": cells,
|
|
"cell_styles": styles,
|
|
"cells_by_text": dict(cells_by_text),
|
|
"styles_by_text": dict(styles_by_text),
|
|
"marker_matrix": marker_matrix,
|
|
}
|
|
|
|
|
|
def read_json(path: Path) -> dict[str, Any]:
|
|
return json.loads(path.read_text(encoding="utf-8"))
|
|
|
|
|
|
def diff_simple(before: Any, after: Any) -> dict[str, Any] | None:
|
|
if before == after:
|
|
return None
|
|
return {"before": before, "after": after}
|
|
|
|
|
|
def index_text_entries(entries: dict[str, list[dict[str, Any]]]) -> dict[str, list[dict[str, Any]]]:
|
|
indexed: dict[str, list[dict[str, Any]]] = {}
|
|
for text, items in entries.items():
|
|
indexed[text] = sorted(items, key=lambda item: json.dumps(item, ensure_ascii=False, sort_keys=True))
|
|
return indexed
|
|
|
|
|
|
def build_diff(before: dict[str, Any], after: dict[str, Any]) -> dict[str, Any]:
|
|
result: dict[str, Any] = {
|
|
"counts": diff_simple(before.get("counts"), after.get("counts")),
|
|
"dimensions": diff_simple(before.get("dimensions"), after.get("dimensions")),
|
|
}
|
|
|
|
before_cells = index_text_entries(before.get("cells_by_text") or {})
|
|
after_cells = index_text_entries(after.get("cells_by_text") or {})
|
|
before_styles = index_text_entries(before.get("styles_by_text") or {})
|
|
after_styles = index_text_entries(after.get("styles_by_text") or {})
|
|
|
|
changed_cells: dict[str, Any] = {}
|
|
for text in sorted(set(before_cells) | set(after_cells)):
|
|
if before_cells.get(text) != after_cells.get(text):
|
|
changed_cells[text] = {"before": before_cells.get(text), "after": after_cells.get(text)}
|
|
|
|
changed_styles: dict[str, Any] = {}
|
|
for text in sorted(set(before_styles) | set(after_styles)):
|
|
if before_styles.get(text) != after_styles.get(text):
|
|
changed_styles[text] = {"before": before_styles.get(text), "after": after_styles.get(text)}
|
|
|
|
before_markers = {item["text"]: item for item in before.get("marker_matrix") or [] if isinstance(item, dict) and item.get("text")}
|
|
after_markers = {item["text"]: item for item in after.get("marker_matrix") or [] if isinstance(item, dict) and item.get("text")}
|
|
changed_markers: dict[str, Any] = {}
|
|
for text in sorted(set(before_markers) | set(after_markers), key=lambda value: tuple(map(int, value.split("-"))) if MARKER_RE.fullmatch(value) else (10**9, 10**9)):
|
|
if before_markers.get(text) != after_markers.get(text):
|
|
changed_markers[text] = {"before": before_markers.get(text), "after": after_markers.get(text)}
|
|
|
|
before_named = {f"{item.get('kind')}::{item.get('name')}": item for item in before.get("named_ranges") or [] if isinstance(item, dict)}
|
|
after_named = {f"{item.get('kind')}::{item.get('name')}": item for item in after.get("named_ranges") or [] if isinstance(item, dict)}
|
|
changed_named: dict[str, Any] = {}
|
|
for key in sorted(set(before_named) | set(after_named)):
|
|
if before_named.get(key) != after_named.get(key):
|
|
changed_named[key] = {"before": before_named.get(key), "after": after_named.get(key)}
|
|
|
|
result["changed_cells_by_text"] = changed_cells
|
|
result["changed_styles_by_text"] = changed_styles
|
|
result["changed_marker_matrix"] = changed_markers
|
|
result["changed_named_ranges"] = changed_named
|
|
result["summary"] = {
|
|
"changed_cell_texts": len(changed_cells),
|
|
"changed_style_texts": len(changed_styles),
|
|
"changed_markers": len(changed_markers),
|
|
"changed_named_ranges": len(changed_named),
|
|
}
|
|
return result
|
|
|
|
|
|
def latest_previous_snapshot(output_dir: Path, current_path: Path) -> Path | None:
|
|
candidates = sorted(output_dir.glob("*.json"))
|
|
filtered = [path for path in candidates if path.resolve() != current_path.resolve()]
|
|
return filtered[-1] if filtered else None
|
|
|
|
|
|
def render_markdown(snapshot: dict[str, Any], diff: dict[str, Any] | None, previous_path: Path | None) -> str:
|
|
lines: list[str] = []
|
|
lines.append("# 1C template probe snapshot")
|
|
lines.append("")
|
|
lines.append(f"- Base: `{snapshot['base_id']}`")
|
|
lines.append(f"- File: `{snapshot['file_name']}`")
|
|
lines.append(f"- Modified: `{snapshot.get('modified')}`")
|
|
lines.append(f"- Bytes: `{snapshot.get('bytes')}`")
|
|
lines.append(f"- Previous snapshot: `{previous_path.name}`" if previous_path else "- Previous snapshot: none")
|
|
lines.append("")
|
|
lines.append("## Marker matrix")
|
|
lines.append("")
|
|
lines.append("| Marker | Decoded | Result | Style tree |")
|
|
lines.append("| --- | --- | --- | --- |")
|
|
for item in snapshot["probe"]["marker_matrix"]:
|
|
decoded = f"R{item.get('decoded_row')}C{item.get('decoded_col')}" if item.get("decoded_row") else "n/a"
|
|
if item.get("row_ok") is True and item.get("col_ok") is True:
|
|
result = "ok"
|
|
elif item.get("decoded_row") is None:
|
|
result = "style-only"
|
|
else:
|
|
result = f"row_ok={item.get('row_ok')} col_ok={item.get('col_ok')} delta={item.get('col_delta')}"
|
|
style_tree = ""
|
|
styles = item.get("styles") or []
|
|
if styles:
|
|
style_tree = ", ".join(str(style.get("tree_position")) for style in styles if style.get("tree_position"))
|
|
lines.append(f"| `{item['text']}` | `{decoded}` | `{result}` | `{style_tree}` |")
|
|
if diff:
|
|
lines.append("")
|
|
lines.append("## Diff summary")
|
|
lines.append("")
|
|
lines.append("```json")
|
|
lines.append(json.dumps(diff.get("summary") or {}, ensure_ascii=False, indent=2))
|
|
lines.append("```")
|
|
return "\n".join(lines) + "\n"
|
|
|
|
|
|
def main() -> int:
|
|
parser = argparse.ArgumentParser(description="Capture and diff a live 1C MOXCEL template probe snapshot.")
|
|
parser.add_argument("--adapter-url", default="http://docker.cin.su:8011")
|
|
parser.add_argument("--base-id", default="upo_test")
|
|
parser.add_argument("--file-name", help="Explicit ConfigCAS file name. If omitted, use the newest MOXCEL payload.")
|
|
parser.add_argument("--scan-limit", type=int, default=30)
|
|
parser.add_argument("--max-cells", type=int, default=500)
|
|
parser.add_argument(
|
|
"--output-dir",
|
|
default=str(Path("Z:/codex/LLM/reports/1c-template-probes")),
|
|
help="Directory for snapshot JSON/Markdown files.",
|
|
)
|
|
parser.add_argument("--output-json", help="Optional exact JSON output path. Overrides generated timestamped name.")
|
|
parser.add_argument("--output-markdown", help="Optional exact Markdown output path. Overrides generated timestamped name.")
|
|
parser.add_argument("--compare-to", help="Optional previous snapshot JSON path.")
|
|
parser.add_argument("--label", default="latest", help="Short label appended to the output file name.")
|
|
args = parser.parse_args()
|
|
|
|
output_dir = Path(args.output_dir)
|
|
output_dir.mkdir(parents=True, exist_ok=True)
|
|
|
|
file_name, row, structure = choose_latest_moxel(args.adapter_url, args.base_id, args.scan_limit, args.max_cells, args.file_name)
|
|
snapshot = {
|
|
"schema": "codex_1c_template_probe_snapshot.v1",
|
|
"captured_at": datetime.utcnow().isoformat(timespec="seconds") + "Z",
|
|
"adapter_url": args.adapter_url,
|
|
"base_id": args.base_id,
|
|
"file_name": file_name,
|
|
"modified": row.get("Modified"),
|
|
"bytes": row.get("Bytes"),
|
|
"label": args.label,
|
|
"probe": build_snapshot(structure),
|
|
}
|
|
|
|
timestamp = datetime.utcnow().strftime("%Y%m%dT%H%M%SZ")
|
|
stem = f"{args.base_id}_{args.label}_{timestamp}_{file_name[:8]}"
|
|
json_path = Path(args.output_json) if args.output_json else output_dir / f"{stem}.json"
|
|
md_path = Path(args.output_markdown) if args.output_markdown else output_dir / f"{stem}.md"
|
|
json_path.parent.mkdir(parents=True, exist_ok=True)
|
|
md_path.parent.mkdir(parents=True, exist_ok=True)
|
|
|
|
previous_path = Path(args.compare_to) if args.compare_to else latest_previous_snapshot(output_dir, json_path)
|
|
diff: dict[str, Any] | None = None
|
|
if previous_path and previous_path.exists():
|
|
previous = read_json(previous_path)
|
|
diff = build_diff(previous.get("probe") or {}, snapshot.get("probe") or {})
|
|
snapshot["diff"] = {
|
|
"compare_to": str(previous_path),
|
|
"summary": diff.get("summary") or {},
|
|
}
|
|
|
|
json_path.write_text(json.dumps(snapshot, ensure_ascii=False, indent=2), encoding="utf-8")
|
|
md_path.write_text(render_markdown(snapshot, diff, previous_path if previous_path and previous_path.exists() else None), encoding="utf-8")
|
|
|
|
print(
|
|
json.dumps(
|
|
{
|
|
"status": "ok",
|
|
"json": str(json_path),
|
|
"markdown": str(md_path),
|
|
"file_name": file_name,
|
|
"modified": row.get("Modified"),
|
|
"bytes": row.get("Bytes"),
|
|
"diff_summary": (diff.get("summary") if diff else None),
|
|
},
|
|
ensure_ascii=False,
|
|
indent=2,
|
|
)
|
|
)
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|