llm/scripts/inventory_1c_configuration_templates.py

from __future__ import annotations

import argparse
import json
from concurrent.futures import ThreadPoolExecutor, as_completed
from pathlib import Path
from typing import Any
from urllib import request


DEFAULT_KINDS = [
    "Catalog",
    "Document",
    "ChartOfCharacteristicTypes",
    "ChartOfAccounts",
    "ChartOfCalculationTypes",
    "ExchangePlan",
    "Report",
    "DataProcessor",
]


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 list_objects_for_kind(adapter_url: str, base_id: str, kind: str, page_size: int, max_objects: int | None) -> list[dict[str, Any]]:
    items: list[dict[str, Any]] = []
    offset = 0
    while True:
        result = rpc(
            adapter_url,
            "metadata.objects.list",
            {
                "base_id": base_id,
                "kind": kind,
                "limit": page_size,
                "offset": offset,
            },
        )
        rows = result.get("objects") or []
        if not rows:
            break
        items.extend(rows)
        if max_objects is not None and len(items) >= max_objects:
            return items[:max_objects]
        if len(rows) < page_size:
            break
        offset += page_size
    return items


def read_templates_for_object(adapter_url: str, base_id: str, obj: dict[str, Any], timeout_seconds: int) -> dict[str, Any]:
    payload = {
        "base_id": base_id,
        "guid": obj.get("guid"),
        "kind": obj.get("kind"),
        "name": obj.get("name"),
        "timeout_seconds": timeout_seconds,
        "include_storage": False,
    }
    result = rpc(adapter_url, "metadata.object.templates", payload)
    templates = result.get("templates") or []
    if not templates:
        return {}
    return {
        "object": {
            "guid": obj.get("guid"),
            "kind": obj.get("kind"),
            "kind_ru": obj.get("kind_ru"),
            "name": obj.get("name"),
            "synonym": obj.get("synonym"),
            "source": obj.get("source"),
        },
        "counts": result.get("counts") or {},
        "templates": templates,
    }


def is_tabular_template(template: dict[str, Any]) -> bool:
    features = template.get("features") if isinstance(template.get("features"), dict) else {}
    if features.get("tabular_document") is True:
        return True
    return str(template.get("format") or "") == "ТабличныйДокумент"


def render_markdown(payload: dict[str, Any]) -> str:
    lines: list[str] = []
    lines.append("# 1C configuration template inventory")
    lines.append("")
    lines.append(f"- Base: `{payload.get('base_id')}`")
    lines.append(f"- Kinds: `{', '.join(payload.get('kinds') or [])}`")
    lines.append(f"- Scanned objects: `{payload.get('counts', {}).get('objects_scanned')}`")
    lines.append(f"- Objects with templates: `{payload.get('counts', {}).get('objects_with_templates')}`")
    lines.append(f"- Templates total: `{payload.get('counts', {}).get('templates_total')}`")
    lines.append("")
    lines.append("| Object | Kind | Templates | Template names |")
    lines.append("| --- | --- | --- | --- |")
    for item in payload.get("items") or []:
        obj = item.get("object") or {}
        names = ", ".join(str(template.get("name") or "") for template in (item.get("templates") or []))
        lines.append(
            f"| `{obj.get('name')}` | `{obj.get('kind')}` | `{len(item.get('templates') or [])}` | `{names}` |"
        )
    return "\n".join(lines) + "\n"


def main() -> int:
    parser = argparse.ArgumentParser(description="Inventory templates across live 1C configuration objects.")
    parser.add_argument("--adapter-url", default="http://docker.cin.su:8011")
    parser.add_argument("--base-id", default="upo_test")
    parser.add_argument("--kind", action="append", dest="kinds", help="Repeatable metadata kind filter.")
    parser.add_argument("--page-size", type=int, default=200)
    parser.add_argument("--max-objects-per-kind", type=int)
    parser.add_argument("--workers", type=int, default=6)
    parser.add_argument("--timeout-seconds", type=int, default=60)
    parser.add_argument("--only-tabular", action="store_true", help="Keep only templates of type 'Табличный документ'.")
    parser.add_argument("--base-only", action="store_true", help="Keep only base configuration objects, skip extension objects.")
    parser.add_argument(
        "--output-json",
        default=str(Path("Z:/codex/LLM/reports/1c-template-baselines/upo_test_configuration_templates.json")),
    )
    parser.add_argument(
        "--output-markdown",
        default=str(Path("Z:/codex/LLM/reports/1c-template-baselines/upo_test_configuration_templates.md")),
    )
    args = parser.parse_args()

    kinds = args.kinds or list(DEFAULT_KINDS)
    objects: list[dict[str, Any]] = []
    for kind in kinds:
        objects.extend(list_objects_for_kind(args.adapter_url, args.base_id, kind, args.page_size, args.max_objects_per_kind))
    if args.base_only:
        objects = [obj for obj in objects if str(obj.get("source") or "") == "base"]

    found: list[dict[str, Any]] = []
    with ThreadPoolExecutor(max_workers=max(1, int(args.workers or 1))) as pool:
        future_map = {
            pool.submit(read_templates_for_object, args.adapter_url, args.base_id, obj, int(args.timeout_seconds or 60)): obj
            for obj in objects
        }
        for future in as_completed(future_map):
            result = future.result()
            if result:
                if args.only_tabular:
                    result["templates"] = [item for item in (result.get("templates") or []) if is_tabular_template(item)]
                    if not result["templates"]:
                        continue
                found.append(result)

    found.sort(key=lambda item: ((item.get("object") or {}).get("kind") or "", (item.get("object") or {}).get("name") or ""))
    payload = {
        "schema": "codex_1c_configuration_templates_inventory.v1",
        "adapter_url": args.adapter_url,
        "base_id": args.base_id,
        "kinds": kinds,
        "items": found,
        "counts": {
            "objects_scanned": len(objects),
            "objects_with_templates": len(found),
            "templates_total": sum(len(item.get("templates") or []) for item in found),
        },
    }

    json_path = Path(args.output_json)
    md_path = Path(args.output_markdown)
    json_path.write_text(json.dumps(payload, ensure_ascii=False, indent=2), encoding="utf-8")
    md_path.write_text(render_markdown(payload), encoding="utf-8")
    print(
        json.dumps(
            {
                "status": "ok",
                "json": str(json_path),
                "markdown": str(md_path),
                "counts": payload["counts"],
            },
            ensure_ascii=False,
            indent=2,
        )
    )
    return 0


if __name__ == "__main__":
    raise SystemExit(main())