331 lines
13 KiB
Python
331 lines
13 KiB
Python
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import os
|
|
from collections import Counter, defaultdict
|
|
from pathlib import Path
|
|
from typing import Any
|
|
from urllib import request
|
|
|
|
|
|
def rpc(adapter_url: str, method: str, payload: dict[str, Any], *, timeout: int) -> dict[str, Any]:
|
|
body = json.dumps({"method": method, "payload": payload}, ensure_ascii=False).encode("utf-8")
|
|
headers = {"Content-Type": "application/json; charset=utf-8"}
|
|
service_token = str(os.getenv("ONEC_ADAPTER_SERVICE_TOKEN") or "").strip()
|
|
if service_token:
|
|
headers["Authorization"] = f"Bearer {service_token}"
|
|
req = request.Request(
|
|
f"{adapter_url.rstrip('/')}/rpc",
|
|
data=body,
|
|
headers=headers,
|
|
method="POST",
|
|
)
|
|
with request.urlopen(req, timeout=timeout) as resp:
|
|
data = json.loads(resp.read().decode("utf-8", errors="replace"))
|
|
if not isinstance(data, dict):
|
|
raise ValueError(f"{method} returned non-object response")
|
|
return data
|
|
|
|
|
|
def read_json(path: Path) -> dict[str, Any]:
|
|
data = json.loads(path.read_text(encoding="utf-8-sig"))
|
|
if not isinstance(data, dict):
|
|
raise ValueError(f"{path} must contain a JSON object")
|
|
return data
|
|
|
|
|
|
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 semantic_coverage(row: dict[str, Any]) -> dict[str, Any]:
|
|
semantic = row.get("semantic") if isinstance(row.get("semantic"), dict) else {}
|
|
coverage = semantic.get("coverage") if isinstance(semantic.get("coverage"), dict) else {}
|
|
return {
|
|
"mapped": int(coverage.get("mapped") or 0),
|
|
"unmapped": int(coverage.get("unmapped") or 0),
|
|
"total": int(coverage.get("total") or 0),
|
|
"status": coverage.get("status") or "unknown",
|
|
}
|
|
|
|
|
|
def unmapped_parameters(row: dict[str, Any], *, limit: int) -> list[dict[str, Any]]:
|
|
semantic = row.get("semantic") if isinstance(row.get("semantic"), dict) else {}
|
|
params = semantic.get("unmapped_parameters") if isinstance(semantic.get("unmapped_parameters"), list) else []
|
|
result = []
|
|
for param in params:
|
|
if not isinstance(param, dict):
|
|
continue
|
|
result.append(
|
|
{
|
|
"index": param.get("index"),
|
|
"presentation": param.get("presentation"),
|
|
"value_kind": param.get("value_kind"),
|
|
"value": param.get("value"),
|
|
"kind": param.get("kind"),
|
|
"items": param.get("items"),
|
|
}
|
|
)
|
|
if len(result) >= limit:
|
|
break
|
|
return result
|
|
|
|
|
|
def row_identity(row: dict[str, Any]) -> dict[str, Any]:
|
|
return {
|
|
"section": row.get("_profile_section"),
|
|
"name": row.get("name"),
|
|
"id": row.get("id"),
|
|
"path": row.get("path"),
|
|
"marker": row.get("marker"),
|
|
"marker_name": row.get("marker_name"),
|
|
"type_name": row.get("type_name"),
|
|
"title": row.get("title"),
|
|
"path_to_data": row.get("path_to_data"),
|
|
}
|
|
|
|
|
|
def scalar_parameters(row: dict[str, Any], *, limit: int) -> list[dict[str, Any]]:
|
|
params = row.get("parameters") if isinstance(row.get("parameters"), list) else []
|
|
result = []
|
|
for param in params:
|
|
if not isinstance(param, dict) or "value" not in param:
|
|
continue
|
|
result.append(
|
|
{
|
|
"index": param.get("index"),
|
|
"presentation": param.get("presentation"),
|
|
"value": param.get("value"),
|
|
"value_kind": param.get("value_kind"),
|
|
"position": param.get("position"),
|
|
}
|
|
)
|
|
if len(result) >= limit:
|
|
break
|
|
return result
|
|
|
|
|
|
def iter_profile_rows(profile: dict[str, Any]) -> list[dict[str, Any]]:
|
|
rows: list[dict[str, Any]] = []
|
|
for section in ("items", "attributes", "commands", "tables", "command_bars"):
|
|
for row in profile.get(section) or []:
|
|
if isinstance(row, dict):
|
|
rows.append({**row, "_profile_section": section})
|
|
return rows
|
|
|
|
|
|
def summarize_form(form: dict[str, Any], *, sample_limit: int) -> dict[str, Any]:
|
|
profile = form.get("profile") if isinstance(form.get("profile"), dict) else {}
|
|
rows = iter_profile_rows(profile)
|
|
coverage_total = Counter()
|
|
status_counts = Counter()
|
|
marker_counts = Counter()
|
|
type_counts = Counter()
|
|
gap_rows = []
|
|
scalar_rows = []
|
|
dynamic_list_settings = []
|
|
|
|
for row in rows:
|
|
coverage = semantic_coverage(row)
|
|
coverage_total.update({key: coverage[key] for key in ("mapped", "unmapped", "total")})
|
|
status_counts[str(coverage.get("status") or "unknown")] += 1
|
|
if row.get("marker_name"):
|
|
marker_counts[str(row.get("marker_name"))] += 1
|
|
if row.get("type_name"):
|
|
type_counts[str(row.get("type_name"))] += 1
|
|
if row.get("dynamic_list_settings"):
|
|
dynamic_list_settings.append({**row_identity(row), "settings": row.get("dynamic_list_settings")})
|
|
unmapped = unmapped_parameters(row, limit=sample_limit)
|
|
if unmapped:
|
|
gap_rows.append({**row_identity(row), "coverage": coverage, "unmapped_parameters": unmapped})
|
|
scalars = scalar_parameters(row, limit=sample_limit)
|
|
if scalars:
|
|
scalar_rows.append({**row_identity(row), "scalars": scalars})
|
|
|
|
return {
|
|
"name": form.get("name"),
|
|
"guid": form.get("guid"),
|
|
"synonyms": form.get("synonyms"),
|
|
"source": form.get("source"),
|
|
"root": profile.get("root"),
|
|
"status": profile.get("status"),
|
|
"counts": {
|
|
**(form.get("counts") or {}),
|
|
**(profile.get("counts") or {}),
|
|
**(profile.get("properties") or {}),
|
|
"profile_rows": len(rows),
|
|
},
|
|
"capabilities": profile.get("capabilities") or {},
|
|
"semantic_coverage": {
|
|
"mapped": coverage_total["mapped"],
|
|
"unmapped": coverage_total["unmapped"],
|
|
"total": coverage_total["total"],
|
|
"status_counts": dict(status_counts),
|
|
},
|
|
"marker_counts": dict(marker_counts),
|
|
"type_counts": dict(type_counts),
|
|
"gap_rows_sample": gap_rows[:sample_limit],
|
|
"scalar_rows_sample": scalar_rows[:sample_limit],
|
|
"dynamic_list_settings": dynamic_list_settings[:sample_limit],
|
|
"handler_links": profile.get("handler_links") or [],
|
|
"command_links": profile.get("command_links") or [],
|
|
"button_command_links": profile.get("button_command_links") or [],
|
|
}
|
|
|
|
|
|
def build_profile(details: dict[str, Any], *, sample_limit: int) -> dict[str, Any]:
|
|
forms = [form for form in details.get("forms") or [] if isinstance(form, dict)]
|
|
form_summaries = [summarize_form(form, sample_limit=sample_limit) for form in forms]
|
|
totals = Counter()
|
|
for form in form_summaries:
|
|
coverage = form.get("semantic_coverage") or {}
|
|
totals.update({key: int(coverage.get(key) or 0) for key in ("mapped", "unmapped", "total")})
|
|
by_status: defaultdict[str, int] = defaultdict(int)
|
|
for form in form_summaries:
|
|
for status, count in ((form.get("semantic_coverage") or {}).get("status_counts") or {}).items():
|
|
by_status[str(status)] += int(count or 0)
|
|
return {
|
|
"schema": "onec_form_decoder_profile.v1",
|
|
"status": details.get("status"),
|
|
"base_id": details.get("base_id"),
|
|
"source": details.get("source"),
|
|
"object": details.get("object"),
|
|
"counts": {
|
|
**(details.get("counts") or {}),
|
|
"forms_profiled": len(form_summaries),
|
|
"semantic_mapped": totals["mapped"],
|
|
"semantic_unmapped": totals["unmapped"],
|
|
"semantic_total": totals["total"],
|
|
"semantic_status_counts": dict(sorted(by_status.items())),
|
|
},
|
|
"forms": form_summaries,
|
|
}
|
|
|
|
|
|
def markdown_table_row(values: list[Any]) -> str:
|
|
return "| " + " | ".join(str(value).replace("\n", " ") for value in values) + " |"
|
|
|
|
|
|
def render_markdown(profile: dict[str, Any]) -> str:
|
|
lines = ["# 1C Form Decoder Profile", ""]
|
|
obj = profile.get("object") or {}
|
|
lines.append(f"- Base: `{profile.get('base_id')}`")
|
|
lines.append(f"- Object: `{obj.get('kind')}.{obj.get('name')}`")
|
|
counts = profile.get("counts") or {}
|
|
lines.append(f"- Forms profiled: `{counts.get('forms_profiled')}`")
|
|
lines.append(f"- Semantic coverage: `{counts.get('semantic_mapped')}/{counts.get('semantic_total')}` mapped, `{counts.get('semantic_unmapped')}` unmapped")
|
|
lines.append("")
|
|
lines.append(markdown_table_row(["Form", "Status", "Rows", "Mapped", "Unmapped", "Top markers"]))
|
|
lines.append(markdown_table_row(["---", "---", "---:", "---:", "---:", "---"]))
|
|
for form in profile.get("forms") or []:
|
|
coverage = form.get("semantic_coverage") or {}
|
|
top_markers = ", ".join(f"{name}:{count}" for name, count in list((form.get("marker_counts") or {}).items())[:5])
|
|
form_counts = form.get("counts") or {}
|
|
lines.append(
|
|
markdown_table_row(
|
|
[
|
|
form.get("name"),
|
|
form.get("status"),
|
|
form_counts.get("profile_rows"),
|
|
coverage.get("mapped"),
|
|
coverage.get("unmapped"),
|
|
top_markers,
|
|
]
|
|
)
|
|
)
|
|
lines.append("")
|
|
for form in profile.get("forms") or []:
|
|
gaps = form.get("gap_rows_sample") or []
|
|
if not gaps:
|
|
continue
|
|
lines.append(f"## {form.get('name')} Gap Sample")
|
|
lines.append("")
|
|
lines.append(markdown_table_row(["Section", "Name", "Type", "Coverage", "Unmapped sample"]))
|
|
lines.append(markdown_table_row(["---", "---", "---", "---", "---"]))
|
|
for row in gaps[:10]:
|
|
coverage = row.get("coverage") or {}
|
|
sample = ", ".join(
|
|
f"{param.get('index')}:{param.get('presentation')}={param.get('value')!r}"
|
|
for param in (row.get("unmapped_parameters") or [])[:6]
|
|
)
|
|
lines.append(
|
|
markdown_table_row(
|
|
[
|
|
row.get("section"),
|
|
row.get("name"),
|
|
row.get("type_name") or row.get("marker_name"),
|
|
f"{coverage.get('mapped')}/{coverage.get('total')}",
|
|
sample,
|
|
]
|
|
)
|
|
)
|
|
lines.append("")
|
|
return "\n".join(lines).rstrip() + "\n"
|
|
|
|
|
|
def main() -> int:
|
|
parser = argparse.ArgumentParser(description="Profile decoded 1C managed forms and highlight decoder gaps.")
|
|
parser.add_argument("--input-json", type=Path, help="Existing metadata.object.form.details JSON.")
|
|
parser.add_argument("--adapter-url", default="http://docker.cin.su:8011")
|
|
parser.add_argument("--base-id", default="upo_test")
|
|
parser.add_argument("--kind")
|
|
parser.add_argument("--name")
|
|
parser.add_argument("--guid")
|
|
parser.add_argument("--table")
|
|
parser.add_argument("--form")
|
|
parser.add_argument("--timeout", type=int, default=120)
|
|
parser.add_argument("--sample-limit", type=int, default=20)
|
|
parser.add_argument("--raw-output-json", type=Path, help="Write raw metadata.object.form.details response.")
|
|
parser.add_argument("--output-json", type=Path, required=True)
|
|
parser.add_argument("--output-markdown", type=Path)
|
|
args = parser.parse_args()
|
|
|
|
if args.input_json:
|
|
details = read_json(args.input_json)
|
|
else:
|
|
payload: dict[str, Any] = {
|
|
"base_id": args.base_id,
|
|
"include_parameters": True,
|
|
"max_items": 5000,
|
|
"max_parameters": 500,
|
|
"include_storage": True,
|
|
"timeout_seconds": args.timeout,
|
|
}
|
|
for key in ("kind", "name", "guid", "table", "form"):
|
|
value = getattr(args, key)
|
|
if value:
|
|
payload[key] = value
|
|
details = rpc(args.adapter_url, "metadata.object.form.details", payload, timeout=args.timeout)
|
|
|
|
if args.raw_output_json:
|
|
write_json(args.raw_output_json, details)
|
|
profile = build_profile(details, sample_limit=args.sample_limit)
|
|
write_json(args.output_json, profile)
|
|
if args.output_markdown:
|
|
args.output_markdown.parent.mkdir(parents=True, exist_ok=True)
|
|
args.output_markdown.write_text(render_markdown(profile), encoding="utf-8")
|
|
print(
|
|
json.dumps(
|
|
{
|
|
"schema": "onec_form_decoder_profile_cli_summary.v1",
|
|
"status": profile.get("status"),
|
|
"object": {
|
|
"kind": (profile.get("object") or {}).get("kind"),
|
|
"name": (profile.get("object") or {}).get("name"),
|
|
},
|
|
"counts": profile.get("counts"),
|
|
"output_json": str(args.output_json),
|
|
"output_markdown": str(args.output_markdown) if args.output_markdown else None,
|
|
},
|
|
ensure_ascii=False,
|
|
indent=2,
|
|
)
|
|
)
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|