Spaces:
Running on Zero
Running on Zero
File size: 2,221 Bytes
f1ef7e2 7ce1abd f1ef7e2 7ce1abd f1ef7e2 7ce1abd f1ef7e2 537dc81 7ce1abd 537dc81 7ce1abd 537dc81 7ce1abd | 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 | import os
import gradio as gr
from fastapi.middleware.cors import CORSMiddleware
from gradio.routes import App
from sqlalchemy import func
from app.database import SessionLocal
from app.main import lifespan
from app.models import Call, Job
from app.routers import artifacts, calls, sentiment
from app.storage import is_configured
def runtime_snapshot():
snapshot = {
"service": "call-qa-processing",
"database_configured": bool(os.environ.get("DATABASE_URL")),
"storage_configured": is_configured(),
"acoustic_enabled": os.environ.get("ENABLE_ACOUSTIC", "0") == "1",
}
db = SessionLocal()
try:
snapshot["calls"] = db.query(func.count(Call.call_id)).scalar() or 0
snapshot["jobs"] = {
status: db.query(func.count(Job.job_id)).filter(Job.status == status).scalar() or 0
for status in ("queued", "processing", "succeeded", "failed")
}
except Exception as exc:
snapshot["database_error"] = str(exc)[:240]
finally:
db.close()
return snapshot
def build_space_app(gpu_snapshot):
with gr.Blocks(title="Call QA Processing") as demo:
gr.Markdown("# Call QA Processing")
state = gr.JSON(label="Runtime status")
refresh = gr.Button("Refresh", variant="primary")
refresh.click(runtime_snapshot, outputs=state, api_name="runtime_status")
demo.load(runtime_snapshot, outputs=state)
gpu_state = gr.JSON(label="ZeroGPU status")
gpu_check = gr.Button("Check ZeroGPU")
gpu_check.click(gpu_snapshot, outputs=gpu_state, api_name="gpu_status")
api = App()
origins = os.environ.get("FRONTEND_ORIGIN", "*")
wildcard = origins.strip() == "*"
api.add_middleware(
CORSMiddleware,
allow_origins=["*"] if wildcard else [item.strip() for item in origins.split(",")],
allow_credentials=not wildcard,
allow_methods=["*"],
allow_headers=["*"],
)
api.include_router(calls.router)
api.include_router(sentiment.router)
api.include_router(artifacts.router)
@api.get("/health")
def health_check():
return {"status": "ok", "service": "capstone_api"}
return demo, api
|