scripts/preflight_hf.py

#!/usr/bin/env python3
"""Pre-flight checks before a full-report demo (local or Hugging Face Space).

Usage:
  python scripts/preflight_hf.py
  python scripts/preflight_hf.py --base https://StormShadow308-RICS.hf.space --tenant my-tenant
"""

from __future__ import annotations

import argparse
import json
import sys
import urllib.error
import urllib.request


def _get(url: str, tenant: str, timeout: float = 30.0) -> dict:
    req = urllib.request.Request(
        url,
        headers={"X-Tenant-ID": tenant, "Accept": "application/json"},
    )
    with urllib.request.urlopen(req, timeout=timeout) as resp:
        return json.loads(resp.read().decode())


def main() -> int:
    parser = argparse.ArgumentParser(description="RICS Report Genius HF/local pre-flight")
    parser.add_argument(
        "--base",
        default="http://localhost:8000",
        help="API base URL (no trailing slash)",
    )
    parser.add_argument("--tenant", default="preflight-check", help="X-Tenant-ID header")
    args = parser.parse_args()
    base = args.base.rstrip("/")
    tenant = args.tenant
    failed = 0

    def ok(msg: str) -> None:
        print(f"  OK  {msg}")

    def warn(msg: str) -> None:
        print(f"  WARN  {msg}")

    def fail(msg: str) -> None:
        nonlocal failed
        failed += 1
        print(f"  FAIL  {msg}")

    print(f"\nPre-flight: {base}  tenant={tenant}\n")

    try:
        health = _get(f"{base}/health", tenant)
    except urllib.error.HTTPError as exc:
        fail(f"/health returned {exc.code}")
        return 1
    except Exception as exc:  # noqa: BLE001
        fail(f"Cannot reach API: {exc}")
        return 1

    if health.get("status") not in ("ok", "degraded"):
        warn(f"health status={health.get('status')!r}")

    if not health.get("components", {}).get("openai", health.get("openai_api_key_configured")):
        key_ok = bool(
            (health.get("ai_features") or {}).get("openai_api_key_configured")
        )
        if not key_ok:
            fail("OPENAI_API_KEY not configured on server")

    phases = health.get("ai_phases") or (health.get("ai_features") or {}).get("ai_phases") or {}
    p3 = phases.get("phase3") or {}
    caps = p3.get("capabilities") or {}
    if caps.get("parallel_multi_section"):
        ok("Parallel multi-section generation enabled (10m SLA path)")
    else:
        fail(
            "parallel_multi_section is false — full reports may exceed 10 minutes; "
            "redeploy with SPACE_ID / PRODUCTION_AI_PROFILE or ENABLE_ASYNC_PIPELINE=true"
        )

    sla = p3.get("generation_sla_seconds") or 600
    ok(f"Generation SLA target: {sla}s ({sla // 60} min)")

    for w in health.get("ai_phase_warnings") or health.get("ai_warnings") or []:
        warn(w)

    try:
        summary = _get(f"{base}/documents/tenant-chunk-summary", tenant)
        n = int(summary.get("indexed_chunk_count") or 0)
        if n > 0:
            ok(f"Tenant RAG index: {n} chunks")
        else:
            warn(
                "indexed_chunk_count=0 — upload and wait for ingest before expecting personalised RAG"
            )
    except Exception as exc:  # noqa: BLE001
        warn(f"tenant-chunk-summary: {exc}")

    print()
    if failed:
        print(f"Pre-flight FAILED ({failed} blocking issue(s)).\n")
        return 1
    print("Pre-flight passed — safe to run batch full-report generate.\n")
    return 0


if __name__ == "__main__":
    sys.exit(main())