Spaces:
Sleeping
Sleeping
File size: 3,473 Bytes
be9fd4a | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 | #!/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())
|