from __future__ import annotations import argparse import json import os import sys import urllib.error import urllib.request from typing import Any DEFAULT_ADAPTER_URL = "http://docker-gpu.cin.su:8011" for stream in (sys.stdout, sys.stderr): if hasattr(stream, "reconfigure"): stream.reconfigure(encoding="utf-8") def rpc(adapter_url: str, method: str, payload: dict[str, Any], *, timeout: int = 60) -> dict[str, Any]: body = json.dumps({"method": method, "payload": payload}, ensure_ascii=False).encode("utf-8") service_token = os.environ.get("ONEC_ADAPTER_SERVICE_TOKEN", "").strip() headers = {"Content-Type": "application/json; charset=utf-8"} if service_token: headers["Authorization"] = f"Bearer {service_token}" request = urllib.request.Request( adapter_url.rstrip("/") + "/rpc", data=body, headers=headers, method="POST", ) try: with urllib.request.urlopen(request, timeout=timeout) as response: return json.loads(response.read().decode("utf-8")) except urllib.error.HTTPError as exc: detail = exc.read().decode("utf-8", errors="replace") raise RuntimeError(f"{method} HTTP {exc.code}: {detail}") from exc def compact_form(item: dict[str, Any]) -> dict[str, Any]: return { "name": item.get("name"), "activation_state": item.get("activation_state"), "source": ((item.get("origin") or {}).get("source") if isinstance(item.get("origin"), dict) else None), "table": ((item.get("saved_state") or {}).get("table") if isinstance(item.get("saved_state"), dict) else ((item.get("route") or {}).get("table") if isinstance(item.get("route"), dict) else None)), } def main() -> int: parser = argparse.ArgumentParser(description="Report the effective 1C agent working view for saved-state programming.") parser.add_argument("--adapter-url", default=DEFAULT_ADAPTER_URL) parser.add_argument("--base-id", default="upo_test") parser.add_argument("--extension", default="test2") parser.add_argument("--object-type", default="CommonForm") parser.add_argument("--object-name", default="t_Форма") parser.add_argument("--routine-name", default="ЗаменаДомена") parser.add_argument("--limit", type=int, default=50) parser.add_argument("--timeout", type=int, default=60) parser.add_argument("--allow-missing-target", action="store_true") parser.add_argument("--report") parser.add_argument("--json", action="store_true") args = parser.parse_args() forms_working = rpc( args.adapter_url, "extension.objects.find", { "base_id": args.base_id, "extension": args.extension, "object_type": args.object_type, "state": "working", "limit": args.limit, }, timeout=args.timeout, ) working_forms = [compact_form(item) for item in forms_working.get("objects") or [] if isinstance(item, dict)] save_forms = [item for item in working_forms if str(item.get("activation_state") or "").startswith("saved_")] read_working = rpc( args.adapter_url, "code.read", { "base_id": args.base_id, "object_type": args.object_type, "object_name": args.object_name, "routine_name": args.routine_name, "state": "working", "include_text": True, "max_chars": 4000, }, timeout=args.timeout, ) read_both = rpc( args.adapter_url, "code.read", { "base_id": args.base_id, "object_type": args.object_type, "object_name": args.object_name, "routine_name": args.routine_name, "state": "both", "include_text": True, "max_chars": 4000, }, timeout=args.timeout, ) layers = read_both.get("layers") if isinstance(read_both.get("layers"), list) else [] compact_layers = [ { "source": layer.get("source"), "status": layer.get("status"), "current_state": layer.get("current_state"), "has_text": isinstance(layer.get("text"), str) and bool(str(layer.get("text") or "").strip()), } for layer in layers if isinstance(layer, dict) ] failures: list[str] = [] skipped = args.allow_missing_target and read_working.get("status") != "ok" if forms_working.get("status") != "ok": failures.append("extension.objects.find working must be ok") if read_working.get("status") != "ok" and not skipped: failures.append("code.read working must be ok") if not skipped and ((read_working.get("current_state") or {}).get("source") if isinstance(read_working.get("current_state"), dict) else None) != "saved_state": failures.append("code.read working must use saved_state when saved code exists") if read_both.get("status") != "ok" and not skipped: failures.append("code.read both must be ok") if not skipped and read_both.get("text_source") != "saved_state": failures.append("code.read both must prefer saved_state text when present") report = { "schema": "onec_agent_working_view_report.v1", "status": "failed" if failures else ("skipped_missing_target" if skipped else "ok"), "skipped": skipped, "adapter_url": args.adapter_url, "base_id": args.base_id, "target": {"extension": args.extension, "object_type": args.object_type, "object_name": args.object_name, "routine_name": args.routine_name}, "save_forms_only_names": [str(item.get("name") or "") for item in save_forms if item.get("name")], "forms": {"working": working_forms, "save": save_forms}, "code_read_working": { "status": read_working.get("status"), "current_state": read_working.get("current_state"), "has_text": isinstance(read_working.get("text"), str) and bool(str(read_working.get("text") or "").strip()), }, "code_read_both": { "status": read_both.get("status"), "current_state": read_both.get("current_state"), "text_source": read_both.get("text_source"), "comparison": read_both.get("comparison"), "layers": compact_layers, }, "failures": failures, } if args.report: os.makedirs(os.path.dirname(os.path.abspath(args.report)), exist_ok=True) with open(args.report, "w", encoding="utf-8") as handle: json.dump(report, handle, ensure_ascii=False, indent=2) if args.json: print(json.dumps(report, ensure_ascii=False, indent=2)) else: print(f"1C agent working view: {report['status']}") print("save forms:", ", ".join(report["save_forms_only_names"])) for failure in failures: print(f"FAIL: {failure}", file=sys.stderr) return 1 if failures else 0 if __name__ == "__main__": raise SystemExit(main())