Spaces:
Sleeping
Sleeping
File size: 6,529 Bytes
c496840 1638b2e c496840 1638b2e f927995 1638b2e f59a149 1638b2e f927995 1638b2e f927995 1638b2e f927995 1638b2e 196bcb1 1638b2e 196bcb1 1638b2e f927995 1638b2e f927995 1638b2e f927995 1638b2e f927995 1638b2e f927995 1638b2e f927995 1638b2e 196bcb1 1638b2e f927995 1638b2e f927995 f59a149 7e33fdb 196bcb1 1638b2e 0ab9a8d 1638b2e f927995 f59a149 f927995 f59a149 | 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 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 | """
PharmaShield API - FastAPI Entry Point.
National Pharma-Import Dependency Intelligence.
"""
import json
import logging
import traceback
from uuid import uuid4
from datetime import datetime, timedelta
import asyncio
from fastapi import FastAPI, Request, BackgroundTasks
from fastapi.responses import JSONResponse
from fastapi.middleware.cors import CORSMiddleware
from .config import BASE_DIR, settings
from .deps import (
get_data_loader,
get_graph_service,
get_retriever,
get_gnn,
get_gemini_flash_client,
get_demo_mode_service,
)
# Setup Logging
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s %(name)s %(levelname)s %(message)s"
)
logger = logging.getLogger("pharmashield")
LIVE_SHOCK_PATHS = (
BASE_DIR / "data" / "shocks.json",
BASE_DIR / "data" / "seed" / "live_shocks.json",
)
app = FastAPI(
title="ShockMap API",
version="1.0.0",
description="National Supply Chain Security Intelligence",
docs_url="/docs",
redoc_url=None
)
# CORS Configuration
app.add_middleware(
CORSMiddleware,
allow_origins=settings.ALLOWED_ORIGINS,
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
# Health Status Cache
_health_cache = {
"data": None,
"expiry": datetime.min
}
def _load_live_shock_count():
"""Count deduplicated live shocks from current and legacy feed files."""
seen_ids = set()
total = 0
for path in LIVE_SHOCK_PATHS:
if not path.exists():
continue
try:
items = json.loads(path.read_text(encoding="utf-8"))
except Exception:
continue
for item in items:
item_id = str(item.get("id", "")).strip()
if item_id:
if item_id in seen_ids:
continue
seen_ids.add(item_id)
total += 1
return total
def get_health_status():
"""Returns cached health status, rebuilds every 60 seconds."""
global _health_cache
now = datetime.now()
if _health_cache["data"] and _health_cache["expiry"] > now:
return _health_cache["data"]
dl = get_data_loader()
if not dl.get_drugs():
try:
dl.load_all()
except Exception:
pass
gemini_client = get_gemini_flash_client()
gemini_ready = gemini_client.is_available()
gnn_ready = False
try:
gnn = get_gnn()
gnn_ready = gnn.is_available()
except Exception:
pass
retriever = get_retriever()
qdrant_ready = bool(getattr(retriever, "_enabled", False))
demo_service = get_demo_mode_service()
live_shock_count = _load_live_shock_count()
if settings.DEMO_MODE and live_shock_count > 0:
shock_feed_mode = "hybrid_demo_live"
elif settings.DEMO_MODE:
shock_feed_mode = "demo"
else:
shock_feed_mode = "live"
status = {
"status": "ok",
"version": "1.0.0",
"loaded_drugs": len(dl.get_drugs()),
"loaded_alerts": len(dl.get_alerts()),
"gemini_ready": gemini_ready,
"qdrant_ready": qdrant_ready,
"demo_mode": settings.DEMO_MODE,
"demo_scenarios": demo_service.count(),
"live_shocks": live_shock_count,
"shock_feed_mode": shock_feed_mode,
"gnn_enabled": settings.ENABLE_GNN,
"gnn_loaded": gnn_ready,
"propagation_mode": "gnn" if gnn_ready else "pagerank",
}
_health_cache["data"] = status
_health_cache["expiry"] = now + timedelta(seconds=60)
return status
@app.on_event("startup")
async def startup_event():
"""Initializes system services and loads data on startup."""
logger.info("ShockMap starting...")
# Load Data
dl = get_data_loader()
dl.load_all()
# Initialize Graph
get_graph_service()
# Initialize Vector Store
if settings.DEMO_MODE:
logger.info("Demo mode enabled; skipping Qdrant warmup.")
else:
try:
get_retriever().ensure_collection()
except Exception as e:
logger.warning(f"Qdrant collection initialization failed: {e}")
# Initialize GNN
try:
get_gnn()
except Exception as e:
logger.warning(f"GNN loading failed: {e}")
# Start Background Shock Detector — optional; never crash the server if unavailable
try:
import sys, os
# Ensure the project root is on sys.path so 'ingestion' is importable
project_root = str(Path(__file__).resolve().parent.parent.parent)
if project_root not in sys.path:
sys.path.insert(0, project_root)
if settings.DEMO_MODE:
from ingestion.shock_detector import run_once
loop = asyncio.get_event_loop()
loop.run_in_executor(None, run_once)
logger.info("Background detector triggered once (Demo Mode).")
else:
from ingestion.shock_detector import run_scheduler
import threading
thread = threading.Thread(target=run_scheduler, daemon=True)
thread.start()
logger.info("Background detector thread started (Live Mode).")
except Exception as e:
logger.warning(f"Shock detector could not start (non-fatal): {e}")
logger.info("ShockMap startup sequence complete.")
@app.exception_handler(Exception)
async def global_exception_handler(request: Request, exc: Exception):
"""Global handler for uncaught exceptions."""
request_id = str(uuid4())
logger.error(f"Request ID: {request_id} - Global error: {exc}")
logger.error(traceback.format_exc())
return JSONResponse(
status_code=500,
content={
"error": "internal",
"request_id": request_id,
"message": "An unexpected error occurred."
}
)
@app.get("/")
async def root():
"""Root endpoint info."""
return {"name": "ShockMap API", "docs": "/docs"}
@app.get("/healthz")
async def healthz():
"""System health and status check."""
return get_health_status()
# Router Mounting
# All routers are now implemented and ready for production use
from .api import graph, drugs, alerts, query, simulate, sectors, engines, map
app.include_router(graph.router)
app.include_router(drugs.router)
app.include_router(alerts.router)
app.include_router(query.router)
app.include_router(simulate.router)
app.include_router(sectors.router)
app.include_router(engines.router)
app.include_router(map.router)
|