392 lines
15 KiB
Python
392 lines
15 KiB
Python
#!/usr/bin/env python3
|
|
"""Build an agent-friendly 1C SQL read view from raw and resolved reports."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import re
|
|
from collections import defaultdict
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
|
|
def load_json(path: Path) -> dict[str, Any]:
|
|
return json.loads(path.read_text(encoding="utf-8-sig"))
|
|
|
|
|
|
def write_json(path: Path, data: dict[str, Any]) -> None:
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
path.write_text(json.dumps(data, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
|
|
|
|
|
|
def scope_key(scope: str, row_index: int, cell_key: str) -> tuple[str, int, str]:
|
|
return (scope, row_index, cell_key)
|
|
|
|
|
|
def part_scope(part_name: str | None) -> str:
|
|
return f"table_part:{part_name or ''}"
|
|
|
|
|
|
def best_alternate(alternates: list[dict[str, Any]]) -> dict[str, Any] | None:
|
|
if not alternates:
|
|
return None
|
|
for alternate in alternates:
|
|
route = alternate.get("route") or {}
|
|
if alternate.get("found") and route.get("base_table"):
|
|
return alternate
|
|
for alternate in alternates:
|
|
if alternate.get("found") and str(alternate.get("table") or "").endswith("X1"):
|
|
return alternate
|
|
for alternate in alternates:
|
|
if alternate.get("found"):
|
|
return alternate
|
|
return alternates[0]
|
|
|
|
|
|
def presentation(values: dict[str, Any] | None) -> str | None:
|
|
if not values:
|
|
return None
|
|
for key in ("_Description", "_Code", "_Number", "_EnumOrder"):
|
|
value = values.get(key)
|
|
if value is not None and value != "":
|
|
return str(value)
|
|
return None
|
|
|
|
|
|
BOOLEAN_STANDARD_PATHS = {"standard._Marked", "standard._Posted", "standard._Active"}
|
|
|
|
|
|
def normalize_1c_sql_date(value: Any) -> Any:
|
|
if not isinstance(value, str):
|
|
return value
|
|
match = re.match(r"^(\d{4})(-\d{2}-\d{2}T.*)$", value)
|
|
if not match:
|
|
return value
|
|
year = int(match.group(1))
|
|
if year < 3000:
|
|
return value
|
|
return f"{year - 2000:04d}{match.group(2)}"
|
|
|
|
|
|
def field_has_boolean_type(field: dict[str, Any]) -> bool:
|
|
if field.get("metadata_path") in BOOLEAN_STANDARD_PATHS:
|
|
return True
|
|
for column in (field.get("columns") or {}).values():
|
|
value_type = column.get("value_type")
|
|
values = value_type if isinstance(value_type, list) else [value_type]
|
|
if any(item == "xs:boolean" for item in values):
|
|
return True
|
|
return False
|
|
|
|
|
|
def normalize_display_value(value: Any, field: dict[str, Any]) -> Any:
|
|
if isinstance(value, dict) and value.get("kind") == "binary" and value.get("length") == 1 and field_has_boolean_type(field):
|
|
hex_value = str(value.get("hex") or "").lower()
|
|
if hex_value == "00":
|
|
return False
|
|
if hex_value == "01":
|
|
return True
|
|
return normalize_1c_sql_date(value)
|
|
|
|
|
|
def enum_order(values: dict[str, Any] | None) -> int | None:
|
|
if not values:
|
|
return None
|
|
value = values.get("_EnumOrder")
|
|
if value is None:
|
|
return None
|
|
try:
|
|
return int(value)
|
|
except (TypeError, ValueError):
|
|
return None
|
|
|
|
|
|
def enum_presentation(item: dict[str, Any], enum_map: dict[str, Any] | None) -> dict[str, Any] | None:
|
|
if not enum_map:
|
|
return None
|
|
target = item.get("target") or {}
|
|
if target.get("kind") != "Enum":
|
|
return None
|
|
name = target.get("name")
|
|
order = enum_order(item.get("values"))
|
|
if name is None or order is None:
|
|
return None
|
|
enum = (enum_map.get("enums") or {}).get(name)
|
|
if not enum:
|
|
return None
|
|
value = (enum.get("by_order") or {}).get(str(order))
|
|
if not value:
|
|
return None
|
|
return value
|
|
|
|
|
|
def resolved_payload(item: dict[str, Any], *, mode: str, enum_map: dict[str, Any] | None = None) -> dict[str, Any]:
|
|
enum_value = enum_presentation(item, enum_map)
|
|
payload = {
|
|
"mode": mode,
|
|
"found": item.get("found"),
|
|
"target": item.get("target"),
|
|
"values": item.get("values"),
|
|
"presentation": (enum_value or {}).get("synonym") or (enum_value or {}).get("name") or presentation(item.get("values")),
|
|
"reason": item.get("reason"),
|
|
}
|
|
if enum_value:
|
|
payload["enum_value"] = enum_value
|
|
alternates = item.get("alternate_hits") or []
|
|
if alternates:
|
|
payload["alternate_hits"] = alternates
|
|
selected = best_alternate(alternates)
|
|
if selected:
|
|
payload["selected_alternate"] = selected
|
|
payload["selected_presentation"] = presentation(selected.get("values"))
|
|
return payload
|
|
|
|
|
|
def simple_reference_index(reference_report: dict[str, Any] | None, enum_map: dict[str, Any] | None = None) -> dict[tuple[str, int, str], dict[str, Any]]:
|
|
if not reference_report:
|
|
return {}
|
|
result = {}
|
|
for item in reference_report.get("references") or []:
|
|
ref = item.get("reference") or {}
|
|
scope = "main" if ref.get("scope") == "main" else part_scope(ref.get("table_part_name"))
|
|
row_index = int(ref.get("row_index") or 0)
|
|
cell_key = ref.get("cell_key")
|
|
if not cell_key:
|
|
continue
|
|
result[scope_key(scope, row_index, cell_key)] = resolved_payload(item, mode="single_reference", enum_map=enum_map)
|
|
return result
|
|
|
|
|
|
def composite_reference_index(composite_report: dict[str, Any] | None, enum_map: dict[str, Any] | None = None) -> dict[tuple[str, int, str], dict[str, Any]]:
|
|
if not composite_report:
|
|
return {}
|
|
result = {}
|
|
for item in composite_report.get("composites") or []:
|
|
comp = item.get("composite") or {}
|
|
scope = "main" if comp.get("scope") == "main" else part_scope(comp.get("table_part_name"))
|
|
row_index = int(comp.get("row_index") or 0)
|
|
metadata_path = comp.get("metadata_path")
|
|
if not metadata_path:
|
|
continue
|
|
payload = resolved_payload(item, mode="composite_reference", enum_map=enum_map)
|
|
payload["type_hex"] = comp.get("type_hex")
|
|
payload["rtref_hex"] = comp.get("rtref_hex")
|
|
payload["rtref_sql_number"] = comp.get("rtref_sql_number")
|
|
payload["rrref_hex"] = comp.get("rrref_hex")
|
|
payload["columns"] = comp.get("columns")
|
|
result[(scope, row_index, metadata_path)] = payload
|
|
return result
|
|
|
|
|
|
def composite_value_index(composite_value_report: dict[str, Any] | None) -> dict[tuple[str, int, str], dict[str, Any]]:
|
|
if not composite_value_report:
|
|
return {}
|
|
result = {}
|
|
for item in composite_value_report.get("composites") or []:
|
|
scope = "main" if item.get("scope") == "main" else part_scope(item.get("table_part_name"))
|
|
row_index = int(item.get("row_index") or 0)
|
|
metadata_path = item.get("metadata_path")
|
|
if metadata_path:
|
|
result[(scope, row_index, metadata_path)] = item
|
|
return result
|
|
|
|
|
|
def display_value(field: dict[str, Any]) -> Any:
|
|
resolved = field.get("resolved") or {}
|
|
for key in ("presentation", "selected_presentation"):
|
|
value = resolved.get(key)
|
|
if value is not None:
|
|
return value
|
|
composite = field.get("composite_value") or {}
|
|
selected = composite.get("selected") or {}
|
|
if selected.get("branch") == "primitive":
|
|
values = selected.get("primitive_values") or []
|
|
if values:
|
|
return normalize_display_value(values[0].get("value"), field)
|
|
if field.get("value") is not None:
|
|
return normalize_display_value(field.get("value"), field)
|
|
columns = field.get("columns") or {}
|
|
if len(columns) == 1:
|
|
only = next(iter(columns.values()))
|
|
return normalize_display_value(only.get("value"), field)
|
|
return None
|
|
|
|
|
|
def enrich_row(
|
|
row: dict[str, Any],
|
|
*,
|
|
scope: str,
|
|
row_index: int,
|
|
simple_refs: dict[tuple[str, int, str], dict[str, Any]],
|
|
composite_refs: dict[tuple[str, int, str], dict[str, Any]],
|
|
composite_values: dict[tuple[str, int, str], dict[str, Any]],
|
|
) -> dict[str, Any]:
|
|
cells: dict[str, Any] = {}
|
|
fields: dict[str, Any] = {}
|
|
grouped: dict[str, list[tuple[str, dict[str, Any]]]] = defaultdict(list)
|
|
|
|
for cell_key, cell in row.items():
|
|
enriched = dict(cell)
|
|
simple = simple_refs.get(scope_key(scope, row_index, cell_key))
|
|
if simple:
|
|
enriched["resolved"] = simple
|
|
cells[cell_key] = enriched
|
|
grouped[str(cell.get("metadata_path") or cell_key)].append((cell_key, enriched))
|
|
|
|
for metadata_path, members in grouped.items():
|
|
first = members[0][1]
|
|
columns = {member["column"]: member for _, member in members if member.get("column")}
|
|
field = {
|
|
"metadata_path": metadata_path,
|
|
"metadata_name": first.get("metadata_name"),
|
|
"metadata_uuid": first.get("metadata_uuid"),
|
|
"metadata_field": first.get("metadata_field"),
|
|
"columns": columns,
|
|
}
|
|
composite = composite_refs.get((scope, row_index, metadata_path))
|
|
composite_value = composite_values.get((scope, row_index, metadata_path))
|
|
if composite:
|
|
field["resolved"] = composite
|
|
if composite_value:
|
|
field["composite_value"] = composite_value
|
|
if "resolved" not in field and len(members) == 1 and members[0][1].get("resolved"):
|
|
field["resolved"] = members[0][1]["resolved"]
|
|
elif "resolved" not in field and len(members) == 1:
|
|
field["value"] = members[0][1].get("value")
|
|
field["display_value"] = display_value(field)
|
|
fields[metadata_path] = field
|
|
|
|
return {"cells": cells, "fields": fields}
|
|
|
|
|
|
def build_view(
|
|
read_result: dict[str, Any],
|
|
reference_report: dict[str, Any] | None,
|
|
composite_report: dict[str, Any] | None,
|
|
composite_value_report: dict[str, Any] | None,
|
|
enum_map: dict[str, Any] | None,
|
|
) -> dict[str, Any]:
|
|
simple_refs = simple_reference_index(reference_report, enum_map)
|
|
composite_refs = composite_reference_index(composite_report, enum_map)
|
|
composite_values = composite_value_index(composite_value_report)
|
|
|
|
main_rows = []
|
|
for index, row in enumerate(read_result.get("main", {}).get("rows") or []):
|
|
main_rows.append(
|
|
{
|
|
"row_index": index,
|
|
**enrich_row(
|
|
row,
|
|
scope="main",
|
|
row_index=index,
|
|
simple_refs=simple_refs,
|
|
composite_refs=composite_refs,
|
|
composite_values=composite_values,
|
|
),
|
|
}
|
|
)
|
|
|
|
table_parts = []
|
|
for part in read_result.get("table_parts") or []:
|
|
scope = part_scope(part.get("name"))
|
|
rows = []
|
|
for index, row in enumerate(part.get("rows") or []):
|
|
rows.append(
|
|
{
|
|
"row_index": index,
|
|
**enrich_row(
|
|
row,
|
|
scope=scope,
|
|
row_index=index,
|
|
simple_refs=simple_refs,
|
|
composite_refs=composite_refs,
|
|
composite_values=composite_values,
|
|
),
|
|
}
|
|
)
|
|
table_parts.append(
|
|
{
|
|
"name": part.get("name"),
|
|
"uuid": part.get("uuid"),
|
|
"table": part.get("table"),
|
|
"row_count": part.get("row_count"),
|
|
"rows": rows,
|
|
}
|
|
)
|
|
|
|
return {
|
|
"schema": "onec_sql_read_view.v1",
|
|
"source_schema": read_result.get("schema"),
|
|
"server": read_result.get("server"),
|
|
"database": read_result.get("database"),
|
|
"kind": read_result.get("kind"),
|
|
"identity": read_result.get("identity"),
|
|
"inputs": {
|
|
"read_result": read_result.get("projection_path"),
|
|
"reference_resolution_schema": (reference_report or {}).get("schema"),
|
|
"composite_reference_resolution_schema": (composite_report or {}).get("schema"),
|
|
"composite_value_resolution_schema": (composite_value_report or {}).get("schema"),
|
|
"enum_presentation_map_schema": (enum_map or {}).get("schema"),
|
|
},
|
|
"summary": {
|
|
"main_rows": len(main_rows),
|
|
"table_parts": len(table_parts),
|
|
"simple_reference_cells": len(simple_refs),
|
|
"composite_reference_groups": len(composite_refs),
|
|
"composite_value_groups": len(composite_values),
|
|
},
|
|
"main": {
|
|
"table": read_result.get("main", {}).get("table"),
|
|
"row_count": read_result.get("main", {}).get("row_count"),
|
|
"rows": main_rows,
|
|
},
|
|
"table_parts": table_parts,
|
|
}
|
|
|
|
|
|
def main() -> int:
|
|
parser = argparse.ArgumentParser(description="Build an enriched 1C SQL read view.")
|
|
parser.add_argument("--read-result", type=Path, required=True)
|
|
parser.add_argument("--reference-resolution", type=Path)
|
|
parser.add_argument("--composite-reference-resolution", type=Path)
|
|
parser.add_argument("--composite-value-resolution", type=Path)
|
|
parser.add_argument("--enum-presentation-map", type=Path)
|
|
parser.add_argument("--output", type=Path, required=True)
|
|
args = parser.parse_args()
|
|
|
|
read_result = load_json(args.read_result)
|
|
reference_report = load_json(args.reference_resolution) if args.reference_resolution else None
|
|
composite_report = load_json(args.composite_reference_resolution) if args.composite_reference_resolution else None
|
|
composite_value_report = load_json(args.composite_value_resolution) if args.composite_value_resolution else None
|
|
enum_map = load_json(args.enum_presentation_map) if args.enum_presentation_map else None
|
|
view = build_view(read_result, reference_report, composite_report, composite_value_report, enum_map)
|
|
view["inputs"]["read_result_path"] = str(args.read_result)
|
|
if args.reference_resolution:
|
|
view["inputs"]["reference_resolution_path"] = str(args.reference_resolution)
|
|
if args.composite_reference_resolution:
|
|
view["inputs"]["composite_reference_resolution_path"] = str(args.composite_reference_resolution)
|
|
if args.composite_value_resolution:
|
|
view["inputs"]["composite_value_resolution_path"] = str(args.composite_value_resolution)
|
|
if args.enum_presentation_map:
|
|
view["inputs"]["enum_presentation_map_path"] = str(args.enum_presentation_map)
|
|
write_json(args.output, view)
|
|
print(
|
|
json.dumps(
|
|
{
|
|
"output": str(args.output),
|
|
"main_rows": view["summary"]["main_rows"],
|
|
"table_parts": view["summary"]["table_parts"],
|
|
"simple_reference_cells": view["summary"]["simple_reference_cells"],
|
|
"composite_reference_groups": view["summary"]["composite_reference_groups"],
|
|
"composite_value_groups": view["summary"]["composite_value_groups"],
|
|
},
|
|
ensure_ascii=False,
|
|
)
|
|
)
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|