운영 지표와 지원 요청 저장소 추가

scripts/record-admin-health-sample.py

@@ -0,0 +1,139 @@
+#!/usr/bin/env python3
+"""Record one Vignette admin health sample into app.admin_health_event.
+
+This is a one-shot synthetic monitor entrypoint.  It reuses the same health
+calculation as GET /admin/health, but does not require an interactive admin
+browser session.  Schedule it with scripts/install-health-sampler-task.ps1 on
+Windows, or run it manually for a local proof.
+"""
+
+from __future__ import annotations
+
+import argparse
+import asyncio
+import json
+import os
+import sys
+import time
+from pathlib import Path
+from typing import Any
+
+
+REPO_ROOT = Path(__file__).resolve().parents[1]
+API_ROOT = REPO_ROOT / "apps" / "api"
+
+if hasattr(sys.stdout, "reconfigure"):
+    sys.stdout.reconfigure(encoding="utf-8")
+if hasattr(sys.stderr, "reconfigure"):
+    sys.stderr.reconfigure(encoding="utf-8")
+
+
+def build_parser() -> argparse.ArgumentParser:
+    parser = argparse.ArgumentParser(
+        description="Collect one live Vignette health sample and persist it to app.admin_health_event."
+    )
+    parser.add_argument(
+        "--json",
+        action="store_true",
+        help="Print a compact JSON result instead of a short text summary.",
+    )
+    parser.add_argument(
+        "--allow-unrecorded",
+        action="store_true",
+        help="Exit 0 even when the health check ran but no DB event rows were recorded.",
+    )
+    return parser
+
+
+def _prepare_imports() -> None:
+    os.chdir(API_ROOT)
+    api_root = str(API_ROOT)
+    if api_root not in sys.path:
+        sys.path.insert(0, api_root)
+
+
+def _service_payload(service: Any) -> dict[str, Any]:
+    return {
+        "key": service.key,
+        "status": service.status,
+        "metric": service.metric,
+        "load": service.load,
+    }
+
+
+async def _collect() -> dict[str, Any]:
+    _prepare_imports()
+    from app.auth_sessions import ensure_runtime_tables
+    from app.db import close_pool, init_pool
+    from app.engine_client import engine_client
+    from app.routes.admin import apply_engine_config_from_store, record_admin_health_sample
+    from app.services.voice import voice_service
+
+    await init_pool()
+    try:
+        await ensure_runtime_tables()
+        await apply_engine_config_from_store()
+        await engine_client.startup()
+        await voice_service.startup()
+        health, recorded_count = await record_admin_health_sample(principal=None)
+    finally:
+        await voice_service.shutdown()
+        await engine_client.shutdown()
+        await close_pool()
+    return {
+        "ok": recorded_count > 0,
+        "recorded_count": recorded_count,
+        "generated_at": time.time(),
+        "status": health.status,
+        "environment": health.environment,
+        "engine_mode": health.engine_mode,
+        "services": [_service_payload(service) for service in health.services],
+    }
+
+
+def _error_payload(exc: BaseException) -> dict[str, Any]:
+    return {
+        "ok": False,
+        "recorded_count": 0,
+        "generated_at": time.time(),
+        "error_type": exc.__class__.__name__,
+        "error": str(exc),
+    }
+
+
+def _print_result(result: dict[str, Any], *, as_json: bool) -> None:
+    if as_json:
+        print(json.dumps(result, ensure_ascii=False, separators=(",", ":")))
+        return
+    if result.get("ok"):
+        print(
+            "recorded {count} health events: status={status}, env={env}, engine={engine}".format(
+                count=result["recorded_count"],
+                status=result["status"],
+                env=result["environment"],
+                engine=result["engine_mode"],
+            )
+        )
+    else:
+        print(
+            "health sample not recorded: {error}".format(
+                error=result.get("error") or f"recorded_count={result.get('recorded_count', 0)}"
+            ),
+            file=sys.stderr,
+        )
+
+
+def main() -> int:
+    args = build_parser().parse_args()
+    try:
+        result = asyncio.run(_collect())
+    except Exception as exc:  # noqa: BLE001 - CLI should report a compact failure.
+        result = _error_payload(exc)
+    _print_result(result, as_json=args.json)
+    if result.get("ok") or args.allow_unrecorded:
+        return 0
+    return 2
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())