from __future__ import annotations import argparse import json 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 = 240) -> 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=timeout) as response: return json.loads(response.read().decode("utf-8", errors="replace")) def extension_template_payloads( adapter_url: str, *, base_id: str, extension: str, query: str, timeout_seconds: int, refresh_cache: bool, ) -> tuple[dict[str, Any], list[dict[str, Any]]]: result = rpc( adapter_url, "extension.objects.find", { "base_id": base_id, "extension": extension, "query": query, "limit": 500, "refresh_cache": refresh_cache, "include_storage": True, "timeout_seconds": timeout_seconds, }, timeout=max(240, timeout_seconds + 60), ) by_guid: dict[str, dict[str, Any]] = {} for item in result.get("objects") or []: if not isinstance(item, dict): continue name = str(item.get("name") or "") guid = str(item.get("guid") or "") if not name or not guid or guid in by_guid: continue payload_entry = next( ( entry for entry in item.get("manifest_entries") or [] if isinstance(entry, dict) and (entry.get("suffix") == ".0" or str(entry.get("object_id") or "").endswith(".0")) and entry.get("cas_key") ), None, ) if payload_entry: by_guid[guid] = { "name": name, "metadata_guid": guid, "payload_file_name": payload_entry.get("cas_key"), "manifest_entry": payload_entry, } return result, sorted(by_guid.values(), key=lambda item: str(item.get("name") or "")) def profile_payload( adapter_url: str, *, base_id: str, item: dict[str, Any], timeout_seconds: int, refresh_cache: bool, sample_limit: int, ) -> dict[str, Any]: payload_file_name = str(item.get("payload_file_name") or "") result = rpc( adapter_url, "templates.read", { "base_id": base_id, "kind": "Template", "guid": payload_file_name, "file_name": payload_file_name, "table": "ConfigCAS", "sections": "summary,merged_ranges,merges,coordinate_hints,cells,parameters,named_areas,formats,diagnostics", "view": "structure", "refresh_cache": refresh_cache, "max_cells": sample_limit, "max_merged": sample_limit, "max_parameters": sample_limit, "max_named_areas": sample_limit, "timeout_seconds": timeout_seconds, }, timeout=max(240, timeout_seconds + 60), ) templates = result.get("templates") or [] structure = (templates[0].get("structure") if templates and isinstance(templates[0], dict) else {}) or {} return { **item, "status": result.get("status"), "counts": structure.get("counts") or {}, "capacity_dimensions": structure.get("capacity_dimensions"), "used_dimensions": structure.get("used_dimensions"), "format_dimensions": structure.get("format_dimensions"), "capabilities": structure.get("capabilities"), "merge_block_counts": [ candidate.get("count") for candidate in structure.get("merge_record_block_candidates") or [] if isinstance(candidate, dict) ], "merge_count_hint_counts": [ hint.get("count") for hint in structure.get("merge_count_hints") or [] if isinstance(hint, dict) ], "sample_merged_ranges": structure.get("merged_ranges") or [], "sample_merge_count_hints": structure.get("merge_count_hints") or [], "sample_merge_record_block_candidates": structure.get("merge_record_block_candidates") or [], "sample_cell_coordinate_hints": structure.get("cell_coordinate_hints") or [], "sample_cells": structure.get("cells") or [], "sample_cell_parameters": structure.get("cell_parameters") or [], "sample_named_ranges": structure.get("named_range_candidates") or [], "sample_column_widths": structure.get("column_widths") or [], "sample_format_table": structure.get("format_table") or [], "sample_font_table": structure.get("font_table") or [], "sample_format_style_index_table": structure.get("format_style_index_table") or {}, "sample_cell_format_links": structure.get("cell_format_links") or [], "cell_format_link_stats": structure.get("cell_format_link_stats") or {}, } def render_markdown(payload: dict[str, Any]) -> str: lines = [ "# Extension SQL direct template profiles", "", f"- Base: `{payload.get('base_id')}`", f"- Extension: `{payload.get('extension')}`", f"- Profiles: `{(payload.get('counts') or {}).get('profiles')}`", "", "| Template | Cells | Cell params | Used | Merged ranges | Merge blocks | Merge count hints |", "| --- | ---: | ---: | --- | ---: | --- | --- |", ] for item in payload.get("profiles") or []: counts = item.get("counts") if isinstance(item.get("counts"), dict) else {} used = item.get("used_dimensions") if isinstance(item.get("used_dimensions"), dict) else {} lines.append( f"| `{item.get('name')}` | {counts.get('cells') or 0} | {counts.get('cell_parameters') or 0} | " f"`{used.get('rows')}x{used.get('columns')}` | {counts.get('merged_ranges') or 0} | `{item.get('merge_block_counts') or []}` | " f"`{item.get('merge_count_hint_counts') or []}` |" ) return "\n".join(lines) + "\n" def main() -> int: parser = argparse.ArgumentParser(description="Profile extension common template payloads from live SQL ConfigCAS.") parser.add_argument("--adapter-url", default="http://docker.cin.su:8011") parser.add_argument("--base-id", default="upo_test") parser.add_argument("--extension", required=True) parser.add_argument("--query", default="t_MOXEL") parser.add_argument("--timeout-seconds", type=int, default=180) parser.add_argument("--sample-limit", type=int, default=20) parser.add_argument("--refresh-cache", action="store_true") parser.add_argument("--output-json", required=True) parser.add_argument("--output-markdown", required=True) args = parser.parse_args() find_result, payloads = extension_template_payloads( args.adapter_url, base_id=args.base_id, extension=args.extension, query=args.query, timeout_seconds=int(args.timeout_seconds or 180), refresh_cache=bool(args.refresh_cache), ) profiles = [ profile_payload( args.adapter_url, base_id=args.base_id, item=item, timeout_seconds=int(args.timeout_seconds or 180), refresh_cache=bool(args.refresh_cache), sample_limit=int(args.sample_limit or 20), ) for item in payloads ] payload = { "schema": "onec_extension_template_profiles.v1", "base_id": args.base_id, "extension": args.extension, "query": args.query, "find_counts": find_result.get("counts"), "profiles": profiles, "counts": {"profiles": len(profiles)}, } output_json = Path(args.output_json) output_markdown = Path(args.output_markdown) output_json.parent.mkdir(parents=True, exist_ok=True) output_markdown.parent.mkdir(parents=True, exist_ok=True) output_json.write_text(json.dumps(payload, ensure_ascii=False, indent=2), encoding="utf-8") output_markdown.write_text(render_markdown(payload), encoding="utf-8") print(json.dumps({"status": "ok", "json": str(output_json), "markdown": str(output_markdown), "counts": payload["counts"]}, ensure_ascii=False, indent=2)) return 0 if __name__ == "__main__": raise SystemExit(main())