Spaces:
Running
Running
| """CuratorBot web demo: serves the viewer and a small JSON API. | |
| Endpoints | |
| GET /api/health status and which features are switched on | |
| GET /api/interviews interviews with a precomputed brief | |
| GET /api/interviews/{id} one brief | |
| POST /api/analyse transcript text (+ donor fields) -> brief | |
| POST /api/transcribe short audio clip -> Q/A transcript text | |
| With CURATORBOT_SITE_PASSCODE set, every page and endpoint needs a sign-in | |
| (GET/POST /login sets a cookie) or the X-Site-Passcode header, so a public | |
| Space can be shared with a team without exposing its interviews. | |
| Live endpoints spend API credit, so they can be locked with a passcode | |
| (CURATORBOT_DEMO_PASSCODE) sent in the X-Demo-Passcode header, and only one | |
| analysis runs at a time. | |
| """ | |
| from __future__ import annotations | |
| import asyncio | |
| import hashlib | |
| import hmac | |
| import json | |
| import os | |
| from contextlib import asynccontextmanager | |
| from datetime import datetime, timezone | |
| import re | |
| import sys | |
| import tempfile | |
| from pathlib import Path | |
| from typing import Any | |
| from fastapi import FastAPI, File, Form, Header, HTTPException, Request, UploadFile | |
| from fastapi.responses import FileResponse, HTMLResponse, JSONResponse, RedirectResponse, Response | |
| from fastapi.staticfiles import StaticFiles | |
| from pydantic import BaseModel, Field | |
| from curatorbot import transcript as tx | |
| from curatorbot.config import settings | |
| STATIC = Path(__file__).parent / "static" | |
| ID_RE = re.compile(r"^[A-Za-z0-9._-]{1,80}$") | |
| DATA_STATUS: dict[str, Any] = {"repo": None, "synced": None, "error": None} | |
| def sync_private_data() -> None: | |
| """Download the private dataset of real briefs, if one is configured. | |
| Needs CURATORBOT_DATA_REPO and an HF_TOKEN secret with read access. A | |
| failure leaves the bundled interviews available and is reported by | |
| /api/health rather than stopping the app. | |
| """ | |
| cfg = settings() | |
| if not cfg.data_repo: | |
| return | |
| DATA_STATUS["repo"] = cfg.data_repo | |
| try: | |
| from huggingface_hub import snapshot_download | |
| snapshot_download(repo_id=cfg.data_repo, repo_type="dataset", token=os.getenv("HF_TOKEN") or None, | |
| local_dir=str(cfg.private_data_dir), allow_patterns=["*/brief.json", "*/backtest.json"]) | |
| DATA_STATUS.update(synced=datetime.now(timezone.utc).isoformat(timespec="seconds"), error=None) | |
| except Exception as exc: # report, don't crash: the demo still has its bundled interviews | |
| DATA_STATUS["error"] = f"{type(exc).__name__}: {exc}" | |
| async def lifespan(_app: FastAPI): | |
| await asyncio.to_thread(sync_private_data) | |
| from curatorbot.places import warm | |
| asyncio.get_running_loop().run_in_executor(None, warm) # build the place index in the background | |
| yield | |
| app = FastAPI(title="CuratorBot demo", docs_url=None, redoc_url=None, lifespan=lifespan) | |
| app.mount("/static", StaticFiles(directory=STATIC), name="static") | |
| # ---------- site-wide passcode (CURATORBOT_SITE_PASSCODE) ---------- | |
| SITE_COOKIE = "curatorbot_access" | |
| SITE_OPEN_PATHS = {"/login", "/logout", "/favicon.ico", "/favicon.svg"} | |
| SITE_COOKIE_DAYS = 14 | |
| def _site_token(passcode: str) -> str: | |
| """Cookie value derived from the passcode, so changing the passcode signs everyone out.""" | |
| return hmac.new(passcode.encode(), b"curatorbot-site-v1", hashlib.sha256).hexdigest() | |
| def _is_https(request: Request) -> bool: | |
| return request.url.scheme == "https" or request.headers.get("x-forwarded-proto", "").split(",")[0].strip() == "https" | |
| async def site_gate(request: Request, call_next): | |
| want = settings().site_passcode | |
| if not want or request.url.path in SITE_OPEN_PATHS: | |
| return await call_next(request) | |
| cookie = request.cookies.get(SITE_COOKIE, "") | |
| header = request.headers.get("x-site-passcode", "") | |
| if (cookie and hmac.compare_digest(cookie, _site_token(want))) or (header and hmac.compare_digest(header, want)): | |
| return await call_next(request) | |
| if request.url.path.startswith("/api/"): | |
| return JSONResponse({"detail": "Sign in with the project passcode."}, status_code=401, | |
| headers={"X-Login-Required": "1"}) | |
| return RedirectResponse("/login", status_code=303) | |
| LOGIN_PAGE = """<!doctype html><html lang="en"><head><meta charset="utf-8"> | |
| <meta name="viewport" content="width=device-width, initial-scale=1"><title>CuratorBot: sign in</title><link rel="icon" href="/favicon.svg" type="image/svg+xml"> | |
| <style> | |
| :root{--bg:#f6f4ef;--card:#fff;--ink:#1d1b18;--muted:#6b665d;--line:#d9d4c9;--accent:#7a3b1d;--err:#a02020} | |
| @media (prefers-color-scheme: dark){:root{--bg:#171512;--card:#211e1a;--ink:#ece8e0;--muted:#a49e93;--line:#3a352e;--accent:#e0a07a;--err:#f08080}} | |
| *{box-sizing:border-box}body{margin:0;min-height:100vh;display:grid;place-items:center;background:var(--bg);color:var(--ink); | |
| font:16px/1.5 system-ui,-apple-system,"Segoe UI",sans-serif;padding:16px} | |
| form{background:var(--card);border:1px solid var(--line);border-radius:10px;padding:28px;width:100%;max-width:380px} | |
| h1{font-size:1.3rem;margin:0 0 4px}p{color:var(--muted);margin:0 0 20px;font-size:.95rem} | |
| label{display:block;font-weight:600;margin-bottom:6px} | |
| input{width:100%;padding:10px 12px;border:1px solid var(--line);border-radius:6px;background:var(--bg);color:var(--ink);font:inherit} | |
| button{margin-top:16px;width:100%;padding:10px;border:0;border-radius:6px;background:var(--accent);color:var(--card);font:inherit;font-weight:600;cursor:pointer} | |
| .err{color:var(--err);margin:12px 0 0;font-size:.95rem} | |
| </style></head><body> | |
| <form method="post" action="/login"> | |
| <h1>CuratorBot</h1><p>A research prototype shared with the project team. Enter the passcode you were given.</p> | |
| <label for="pc">Passcode</label><input id="pc" name="passcode" type="password" autocomplete="current-password" required autofocus> | |
| <button type="submit">Sign in</button>__ERROR__ | |
| </form></body></html>""" | |
| def favicon() -> FileResponse: | |
| """The tab icon (also at /favicon.ico, which browsers ask for on their own).""" | |
| return FileResponse(STATIC / "favicon.svg", media_type="image/svg+xml", headers={"Cache-Control": "public, max-age=86400"}) | |
| def login_page() -> Response: | |
| if not settings().site_passcode: | |
| return RedirectResponse("/", status_code=303) | |
| return HTMLResponse(LOGIN_PAGE.replace("__ERROR__", "")) | |
| async def login(request: Request, passcode: str = Form("")) -> Response: | |
| want = settings().site_passcode | |
| if not want: | |
| return RedirectResponse("/", status_code=303) | |
| if not hmac.compare_digest(passcode.strip(), want): | |
| await asyncio.sleep(1) # slows guessing | |
| return HTMLResponse(LOGIN_PAGE.replace("__ERROR__", '<p class="err" role="alert">That passcode is not right.</p>'), status_code=401) | |
| resp = RedirectResponse("/", status_code=303) | |
| attrs = f"Path=/; Max-Age={SITE_COOKIE_DAYS * 86400}; HttpOnly" | |
| # Over HTTPS also allow the cookie inside the huggingface.co page frame (a cross-site iframe). | |
| attrs += "; Secure; SameSite=None; Partitioned" if _is_https(request) else "; SameSite=Lax" | |
| resp.headers.append("Set-Cookie", f"{SITE_COOKIE}={_site_token(want)}; {attrs}") | |
| return resp | |
| def logout() -> RedirectResponse: | |
| resp = RedirectResponse("/login", status_code=303) | |
| resp.headers.append("Set-Cookie", f"{SITE_COOKIE}=; Path=/; Max-Age=0") | |
| return resp | |
| def _find(iid: str) -> Path | None: | |
| for d in reversed(settings().data_dirs): | |
| if (d / iid / "brief.json").exists(): | |
| return d / iid | |
| return None | |
| _lock = asyncio.Lock() | |
| class AnalyseRequest(BaseModel): | |
| transcript: str = Field(..., min_length=1) | |
| title: str = "Uploaded transcript" | |
| donor: dict[str, str] = {} | |
| def _check_passcode(given: str | None) -> None: | |
| want = settings().passcode | |
| if want and not hmac.compare_digest(given or "", want): | |
| raise HTTPException(401, "This demo needs a passcode for live analysis. Ask the project team for it.") | |
| def _llm_ready() -> tuple[bool, str]: | |
| try: | |
| from curatorbot.llm import LLMClient | |
| c = LLMClient() | |
| return c.available(), c.describe() | |
| except Exception as exc: # misconfiguration should show in /api/health, not crash | |
| return False, str(exc) | |
| def _asset_version() -> str: | |
| """Changes whenever the viewer's files change, so browsers fetch the new ones after a deploy.""" | |
| return str(max(int((STATIC / f).stat().st_mtime) for f in ("app.js", "style.css"))) | |
| def index() -> HTMLResponse: | |
| v = _asset_version() | |
| html = (STATIC / "index.html").read_text(encoding="utf-8") | |
| html = html.replace('/static/style.css"', f'/static/style.css?v={v}"').replace('/static/app.js"', f'/static/app.js?v={v}"') | |
| return HTMLResponse(html, headers={"Cache-Control": "no-cache"}) | |
| def health() -> dict[str, Any]: | |
| cfg = settings() | |
| ok, model = _llm_ready() | |
| return {"ok": True, "live": cfg.enable_live and ok, "asr": cfg.enable_asr, "model": model, "private_data": DATA_STATUS, | |
| "passcode": bool(cfg.passcode), "max_chars": cfg.max_transcript_chars, | |
| "max_audio_seconds": cfg.max_audio_seconds} | |
| STATUS_ORDER = {"under_review": 0, "released": 1, "synthetic": 2} | |
| BAND_ORDER = {"High": 0, "Medium": 1, "Standard": 2} | |
| def _summary(brief: dict[str, Any], folder: Path) -> dict[str, Any]: | |
| """Queue row for one interview: what a curator scans before opening it.""" | |
| t, g = brief.get("testimony", {}), brief.get("agent", {}) or {} | |
| claims = g.get("claims", []) or [] | |
| media = g.get("media") or {} | |
| langs = g.get("languages") or sorted({l.get("lang") for l in media.get("languages", []) if l.get("lang")}) | |
| row = { | |
| "id": t.get("id", folder.name), "title": t.get("title", folder.name), "note": t.get("note", ""), | |
| "status": t.get("status") or ("synthetic" if folder.name.startswith("DEMO") else "under_review"), | |
| "duration": t.get("duration", ""), "languages": langs, | |
| "gap": (g.get("gap") or {}).get("band"), | |
| "discrepancies": sum(1 for c in claims if c.get("verdict") in {"contradicted", "incomplete"}), | |
| "claims": len(claims), | |
| "flags": len(media.get("flags", []) or []), | |
| "headings": sum(1 for h in g.get("headings", []) if h.get("grounded", True) and h.get("in_vocabulary", True)), | |
| "places": len(brief.get("mentions", []) or []), | |
| "summary": (g.get("summary") or "")[:280], | |
| } | |
| bt = folder / "backtest.json" | |
| if bt.exists(): | |
| b = json.loads(bt.read_text(encoding="utf-8")) | |
| row["backtest"] = {k: b.get(k) for k in ("recall_exact", "recall_head", "recall_within_vocabulary", "catalogued", "suggested")} | |
| return row | |
| def _queue_key(r: dict[str, Any]) -> tuple: | |
| return (STATUS_ORDER.get(r["status"], 9), BAND_ORDER.get(r.get("gap") or "", 3), -r["discrepancies"], -r["flags"], r["title"]) | |
| def interviews() -> list[dict[str, Any]]: | |
| folders: dict[str, Path] = {} | |
| for d in settings().data_dirs: # later directories (the private dataset) win on a clash | |
| for f in sorted(d.glob("*/brief.json")): | |
| folders[f.parent.name] = f.parent | |
| rows = [_summary(json.loads((f / "brief.json").read_text(encoding="utf-8")), f) for f in folders.values()] | |
| return sorted(rows, key=_queue_key) | |
| def interview(iid: str) -> JSONResponse: | |
| if not ID_RE.match(iid): | |
| raise HTTPException(400, "Invalid interview id.") | |
| folder = _find(iid) | |
| if folder is None: | |
| raise HTTPException(404, "No brief for that interview.") | |
| f = folder / "brief.json" | |
| brief = json.loads(f.read_text(encoding="utf-8")) | |
| bt = folder / "backtest.json" | |
| if bt.exists(): | |
| brief["backtest"] = json.loads(bt.read_text(encoding="utf-8")) | |
| return JSONResponse(brief) | |
| def reload_data(x_demo_passcode: str | None = Header(default=None)) -> dict[str, Any]: | |
| """Fetch newly uploaded briefs from the private dataset without restarting the Space.""" | |
| _check_passcode(x_demo_passcode) | |
| if not settings().data_repo: | |
| raise HTTPException(400, "No private dataset is configured (CURATORBOT_DATA_REPO).") | |
| sync_private_data() | |
| return {"private_data": DATA_STATUS, "interviews": len(interviews())} | |
| async def analyse(req: AnalyseRequest, x_demo_passcode: str | None = Header(default=None)) -> dict[str, Any]: | |
| cfg = settings() | |
| _check_passcode(x_demo_passcode) | |
| ok, model = _llm_ready() | |
| if not (cfg.enable_live and ok): | |
| raise HTTPException(503, "Live analysis is switched off on this server.") | |
| if len(req.transcript) > cfg.max_transcript_chars: | |
| raise HTTPException(413, f"Transcripts are limited to {cfg.max_transcript_chars:,} characters here. Paste an excerpt.") | |
| segments = tx.parse_qa_text(req.transcript) | |
| if not segments: | |
| raise HTTPException(422, "No turns found. Use lines starting with Q: and A:.") | |
| if _lock.locked(): | |
| raise HTTPException(429, "Another analysis is running. Try again in a minute.") | |
| async with _lock: | |
| from curatorbot.pipeline import build_brief | |
| try: | |
| return await asyncio.to_thread( | |
| build_brief, segments, interview_id="LIVE", title=req.title[:120], | |
| donor={k: v[:300] for k, v in req.donor.items() if v}, note="Analysed live; not saved.") | |
| except Exception as exc: | |
| from curatorbot.llm import LLMError, friendly_error | |
| msg = str(exc) if isinstance(exc, LLMError) else friendly_error(exc) | |
| print(f"[curatorbot] analysis failed: {type(exc).__name__}: {exc}", file=sys.stderr) | |
| raise HTTPException(502, msg) from exc | |
| async def transcribe(file: UploadFile = File(...), x_demo_passcode: str | None = Header(default=None)) -> dict[str, Any]: | |
| cfg = settings() | |
| _check_passcode(x_demo_passcode) | |
| if not cfg.enable_asr: | |
| raise HTTPException(503, "Audio transcription is switched off on this server.") | |
| data = await file.read() | |
| if len(data) > 50 * 1024 * 1024: | |
| raise HTTPException(413, "Clips are limited to 50 MB.") | |
| from curatorbot.asr import transcribe as run_asr | |
| from curatorbot.media import probe | |
| with tempfile.TemporaryDirectory() as d: | |
| p = Path(d) / Path(file.filename or "clip").name | |
| p.write_bytes(data) | |
| if probe(p)["seconds"] > cfg.max_audio_seconds: | |
| raise HTTPException(413, f"Clips are limited to {cfg.max_audio_seconds // 60} minutes here.") | |
| async with _lock: | |
| res = await asyncio.to_thread(run_asr, str(p), size=cfg.asr_model, device=cfg.asr_device) | |
| segments = tx.from_asr(res["chunks"]) | |
| return {"text": tx.to_qa_text(segments), "language": res["language"], "flags": res["flags"]} | |