361 lines
15 KiB
Python
361 lines
15 KiB
Python
from __future__ import annotations
|
||
|
||
import argparse
|
||
import json
|
||
import urllib.request
|
||
from pathlib import Path
|
||
from typing import Any
|
||
import xml.etree.ElementTree as ET
|
||
|
||
from analyze_1c_template_xml_profiles import merge_ranges
|
||
|
||
|
||
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 = urllib.request.Request(
|
||
f"{adapter_url.rstrip('/')}/rpc",
|
||
data=body,
|
||
headers={"Content-Type": "application/json; charset=utf-8"},
|
||
method="POST",
|
||
)
|
||
with urllib.request.urlopen(req, timeout=180) as resp:
|
||
return json.loads(resp.read().decode("utf-8", errors="replace"))
|
||
|
||
|
||
def template_xml_path(root: Path, template: str) -> Path:
|
||
return root / template / "Ext" / "Template.xml"
|
||
|
||
|
||
def range_fields(merges: list[dict[str, Any]]) -> dict[str, set[int]]:
|
||
fields: dict[str, set[int]] = {
|
||
"top": set(),
|
||
"left": set(),
|
||
"bottom": set(),
|
||
"right": set(),
|
||
"width": set(),
|
||
"height": set(),
|
||
"top_zero": set(),
|
||
"left_zero": set(),
|
||
"bottom_zero": set(),
|
||
"right_zero": set(),
|
||
}
|
||
for item in merges:
|
||
one = (item.get("range") or {}).get("one_based") or {}
|
||
zero = (item.get("range") or {}).get("zero_based") or {}
|
||
for name in ("top", "left", "bottom", "right"):
|
||
if isinstance(one.get(name), int):
|
||
fields[name].add(int(one[name]))
|
||
if isinstance(zero.get(name), int):
|
||
fields[f"{name}_zero"].add(int(zero[name]))
|
||
for name in ("width", "height"):
|
||
if isinstance(item.get(name), int):
|
||
fields[name].add(int(item[name]))
|
||
return fields
|
||
|
||
|
||
def fetch_merge_candidate(adapter_url: str, base_id: str, owner_kind: str, owner_name: str, template: str) -> dict[str, Any]:
|
||
data = rpc(
|
||
adapter_url,
|
||
"templates.read",
|
||
{
|
||
"base_id": base_id,
|
||
"kind": owner_kind,
|
||
"name": owner_name,
|
||
"template": template,
|
||
"sections": "merges",
|
||
"max_merged": 1,
|
||
"refresh_cache": False,
|
||
},
|
||
)
|
||
return (((data.get("templates") or [{}])[0].get("structure") or {}).get("merge_record_block_candidates") or [{}])[0]
|
||
|
||
|
||
def fetch_merge_records(
|
||
adapter_url: str,
|
||
base_id: str,
|
||
owner_kind: str,
|
||
owner_name: str,
|
||
template: str,
|
||
candidate: dict[str, Any],
|
||
) -> list[dict[str, Any]]:
|
||
position = str(candidate.get("tree_position") or "$.0")
|
||
try:
|
||
start = int(position.split(".")[1]) + 1
|
||
except (IndexError, ValueError):
|
||
start = 0
|
||
count = int(candidate.get("count") or 0)
|
||
data = rpc(
|
||
adapter_url,
|
||
"templates.read",
|
||
{
|
||
"base_id": base_id,
|
||
"kind": owner_kind,
|
||
"name": owner_name,
|
||
"template": template,
|
||
"sections": "moxel_records",
|
||
"max_moxel_records": count + 40,
|
||
"moxel_record_start": start,
|
||
"moxel_record_end": start + count + 35,
|
||
"refresh_cache": False,
|
||
},
|
||
)
|
||
diagnostics = ((data.get("templates") or [{}])[0].get("structure") or {}).get("moxel_record_diagnostics") or [{}]
|
||
if isinstance(diagnostics, list):
|
||
diagnostics = diagnostics[0] if diagnostics else {}
|
||
return [record for record in diagnostics.get("top_level_records") or [] if isinstance(record, dict)][:count]
|
||
|
||
|
||
def values_by_slot(records: list[dict[str, Any]]) -> dict[int, list[int]]:
|
||
result: dict[int, list[int]] = {}
|
||
for record in records:
|
||
numbers = record.get("numeric_items") if isinstance(record.get("numeric_items"), list) else []
|
||
for slot, value in enumerate(numbers):
|
||
if isinstance(value, int):
|
||
result.setdefault(slot, []).append(value)
|
||
return result
|
||
|
||
|
||
def score_values(values: list[int], expected: set[int]) -> dict[str, Any]:
|
||
if not values or not expected:
|
||
return {"hits": 0, "coverage": 0.0, "precision": 0.0, "score": 0.0}
|
||
distinct = set(values)
|
||
hits = distinct & expected
|
||
coverage = len(hits) / len(expected)
|
||
precision = len(hits) / len(distinct)
|
||
return {
|
||
"hits": len(hits),
|
||
"coverage": round(coverage, 4),
|
||
"precision": round(precision, 4),
|
||
"score": round((coverage * 0.7) + (precision * 0.3), 4),
|
||
"hit_values": sorted(hits)[:80],
|
||
"distinct_values": len(distinct),
|
||
}
|
||
|
||
|
||
def slot_candidates(records: list[dict[str, Any]], fields: dict[str, set[int]]) -> list[dict[str, Any]]:
|
||
candidates: list[dict[str, Any]] = []
|
||
by_slot = values_by_slot(records)
|
||
for slot, values in sorted(by_slot.items()):
|
||
transforms = {
|
||
"raw": values,
|
||
"raw_plus_1": [value + 1 for value in values],
|
||
"raw_div32": [value // 32 for value in values if value > 0 and value <= 4096 and value % 32 == 0],
|
||
"raw_div32_plus_1": [(value // 32) + 1 for value in values if value > 0 and value <= 4096 and value % 32 == 0],
|
||
}
|
||
for transform, transformed_values in transforms.items():
|
||
for field, expected in fields.items():
|
||
score = score_values(transformed_values, expected)
|
||
if score["hits"] <= 0:
|
||
continue
|
||
candidates.append(
|
||
{
|
||
"slot": slot,
|
||
"transform": transform,
|
||
"field": field,
|
||
**score,
|
||
"sample_values": sorted(set(transformed_values))[:30],
|
||
}
|
||
)
|
||
candidates.sort(key=lambda item: (-float(item.get("score") or 0), -float(item.get("coverage") or 0), -float(item.get("precision") or 0), int(item.get("slot") or 0), str(item.get("field") or "")))
|
||
return candidates
|
||
|
||
|
||
def xml_ordered_fields(merges: list[dict[str, Any]]) -> list[dict[str, int]]:
|
||
result: list[dict[str, int]] = []
|
||
for item in merges:
|
||
one = (item.get("range") or {}).get("one_based") or {}
|
||
zero = (item.get("range") or {}).get("zero_based") or {}
|
||
row: dict[str, int] = {}
|
||
for name in ("top", "left", "bottom", "right"):
|
||
if isinstance(one.get(name), int):
|
||
row[name] = int(one[name])
|
||
if isinstance(zero.get(name), int):
|
||
row[f"{name}_zero"] = int(zero[name])
|
||
for name in ("width", "height"):
|
||
if isinstance(item.get(name), int):
|
||
row[name] = int(item[name])
|
||
result.append(row)
|
||
return result
|
||
|
||
|
||
def transformed_record_value(numbers: list[Any], slot: int, transform: str) -> int | None:
|
||
if slot >= len(numbers) or not isinstance(numbers[slot], int):
|
||
return None
|
||
value = int(numbers[slot])
|
||
if transform == "raw":
|
||
return value
|
||
if transform == "raw_plus_1":
|
||
return value + 1
|
||
if transform == "raw_div32":
|
||
if value <= 0 or value > 4096 or value % 32 != 0:
|
||
return None
|
||
return value // 32
|
||
if transform == "raw_div32_plus_1":
|
||
if value <= 0 or value > 4096 or value % 32 != 0:
|
||
return None
|
||
return (value // 32) + 1
|
||
return None
|
||
|
||
|
||
def ordered_slot_candidates(records: list[dict[str, Any]], merges: list[dict[str, Any]]) -> list[dict[str, Any]]:
|
||
ordered = xml_ordered_fields(merges)
|
||
transforms = ("raw", "raw_plus_1", "raw_div32", "raw_div32_plus_1")
|
||
fields = ("top", "left", "bottom", "right", "width", "height", "top_zero", "left_zero", "bottom_zero", "right_zero")
|
||
candidates: list[dict[str, Any]] = []
|
||
max_slots = max((len(record.get("numeric_items") or []) for record in records), default=0)
|
||
for offset in range(0, min(25, len(records))):
|
||
pair_count = min(len(ordered), max(0, len(records) - offset))
|
||
if pair_count < max(10, min(len(ordered), 20)):
|
||
continue
|
||
for slot in range(max_slots):
|
||
for transform in transforms:
|
||
values = [
|
||
transformed_record_value(records[offset + index].get("numeric_items") or [], slot, transform)
|
||
for index in range(pair_count)
|
||
]
|
||
available = sum(1 for value in values if value is not None)
|
||
if available < max(5, pair_count // 3):
|
||
continue
|
||
for field in fields:
|
||
matches = [
|
||
index + 1
|
||
for index, value in enumerate(values)
|
||
if value is not None and ordered[index].get(field) == value
|
||
]
|
||
if not matches:
|
||
continue
|
||
exact_ratio = len(matches) / pair_count
|
||
available_ratio = len(matches) / available
|
||
if exact_ratio < 0.1 and len(matches) < 8:
|
||
continue
|
||
candidates.append(
|
||
{
|
||
"offset": offset,
|
||
"slot": slot,
|
||
"transform": transform,
|
||
"field": field,
|
||
"pairs": pair_count,
|
||
"available": available,
|
||
"matches": len(matches),
|
||
"exact_ratio": round(exact_ratio, 4),
|
||
"available_ratio": round(available_ratio, 4),
|
||
"score": round((exact_ratio * 0.75) + (available_ratio * 0.25), 4),
|
||
"first_match_indexes": matches[:30],
|
||
}
|
||
)
|
||
candidates.sort(
|
||
key=lambda item: (
|
||
-float(item.get("score") or 0),
|
||
-float(item.get("exact_ratio") or 0),
|
||
-int(item.get("matches") or 0),
|
||
int(item.get("offset") or 0),
|
||
int(item.get("slot") or 0),
|
||
)
|
||
)
|
||
return candidates
|
||
|
||
|
||
def analyze_template(
|
||
*,
|
||
adapter_url: str,
|
||
base_id: str,
|
||
owner_kind: str,
|
||
owner_name: str,
|
||
template: str,
|
||
xml_root: Path,
|
||
) -> dict[str, Any]:
|
||
merges = merge_ranges(ET.parse(template_xml_path(xml_root, template)).getroot(), limit=1000)
|
||
candidate = fetch_merge_candidate(adapter_url, base_id, owner_kind, owner_name, template)
|
||
analysis = ((candidate.get("evidence") or {}).get("record_analysis") or {})
|
||
records = fetch_merge_records(adapter_url, base_id, owner_kind, owner_name, template, candidate)
|
||
fields = range_fields(merges)
|
||
return {
|
||
"template": template,
|
||
"xml_merge_count": len(merges),
|
||
"sql_block_count": int(candidate.get("count") or 0),
|
||
"tree_position": candidate.get("tree_position"),
|
||
"xml_field_values": {name: sorted(values) for name, values in fields.items()},
|
||
"record_analysis": {
|
||
"schema": analysis.get("schema"),
|
||
"records_analyzed": analysis.get("records_analyzed"),
|
||
"records_available": len(records),
|
||
"raw_records_source": "templates.read.sections=moxel_records",
|
||
},
|
||
"slot_candidates": slot_candidates(records, fields)[:120],
|
||
"ordered_slot_candidates": ordered_slot_candidates(records, merges)[:120],
|
||
}
|
||
|
||
|
||
def render_markdown(payload: dict[str, Any]) -> str:
|
||
lines = ["# MOXCEL merge slot candidate analysis", ""]
|
||
lines.append("XML is used only as an analysis fixture; candidates are SQL decoder hypotheses.")
|
||
lines.append("")
|
||
for item in payload.get("items") or []:
|
||
lines.append(f"## {item.get('template')}")
|
||
lines.append("")
|
||
lines.append(f"- XML merges: `{item.get('xml_merge_count')}`")
|
||
lines.append(f"- SQL block count: `{item.get('sql_block_count')}` at `{item.get('tree_position')}`")
|
||
ra = item.get("record_analysis") or {}
|
||
lines.append(f"- Record analysis: `{ra.get('schema')}`, records `{ra.get('records_available')}/{ra.get('records_analyzed')}`")
|
||
lines.append("")
|
||
lines.append("| Slot | Transform | Field | Score | Coverage | Precision | Hit values | Sample values |")
|
||
lines.append("| ---: | --- | --- | ---: | ---: | ---: | --- | --- |")
|
||
for row in (item.get("slot_candidates") or [])[:40]:
|
||
lines.append(
|
||
f"| {row.get('slot')} | `{row.get('transform')}` | `{row.get('field')}` | "
|
||
f"{row.get('score')} | {row.get('coverage')} | {row.get('precision')} | "
|
||
f"`{row.get('hit_values')}` | `{row.get('sample_values')}` |"
|
||
)
|
||
lines.append("")
|
||
lines.append("### Ordered Slot Candidates")
|
||
lines.append("")
|
||
lines.append("| Offset | Slot | Transform | Field | Score | Exact ratio | Available ratio | Matches | First match indexes |")
|
||
lines.append("| ---: | ---: | --- | --- | ---: | ---: | ---: | ---: | --- |")
|
||
for row in (item.get("ordered_slot_candidates") or [])[:40]:
|
||
lines.append(
|
||
f"| {row.get('offset')} | {row.get('slot')} | `{row.get('transform')}` | `{row.get('field')}` | "
|
||
f"{row.get('score')} | {row.get('exact_ratio')} | {row.get('available_ratio')} | "
|
||
f"{row.get('matches')}/{row.get('pairs')} | `{row.get('first_match_indexes')}` |"
|
||
)
|
||
lines.append("")
|
||
return "\n".join(lines)
|
||
|
||
|
||
def main() -> int:
|
||
parser = argparse.ArgumentParser(description="Score SQL MOXCEL merge-block numeric slots against XML merge range fields.")
|
||
parser.add_argument("--adapter-url", default="http://docker.cin.su:8011")
|
||
parser.add_argument("--base-id", default="upo_test")
|
||
parser.add_argument("--owner-kind", default="Document")
|
||
parser.add_argument("--owner-name", default="АвансовыйОтчет")
|
||
parser.add_argument(
|
||
"--xml-root",
|
||
default=r"Z:\codex\1C\XML\UPO\Структура базы 1с\Конфигурация\Documents\АвансовыйОтчет\Templates",
|
||
)
|
||
parser.add_argument("--template", action="append", required=True)
|
||
parser.add_argument("--output-json", default="reports/1c-template-baselines/moxel-merge-slot-candidates.json")
|
||
parser.add_argument("--output-markdown", default="reports/1c-template-baselines/moxel-merge-slot-candidates.md")
|
||
args = parser.parse_args()
|
||
payload = {
|
||
"schema": "codex_1c_moxel_merge_slot_candidates.v1",
|
||
"source": "analysis_only_xml_fixture",
|
||
"items": [
|
||
analyze_template(
|
||
adapter_url=args.adapter_url,
|
||
base_id=args.base_id,
|
||
owner_kind=args.owner_kind,
|
||
owner_name=args.owner_name,
|
||
template=template,
|
||
xml_root=Path(args.xml_root),
|
||
)
|
||
for template in args.template
|
||
],
|
||
}
|
||
Path(args.output_json).write_text(json.dumps(payload, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
|
||
Path(args.output_markdown).write_text(render_markdown(payload), encoding="utf-8")
|
||
print(json.dumps({"status": "ok", "json": args.output_json, "markdown": args.output_markdown, "items": len(payload["items"])}, ensure_ascii=False, indent=2))
|
||
return 0
|
||
|
||
|
||
if __name__ == "__main__":
|
||
raise SystemExit(main())
|