"""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}" @asynccontextmanager 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" @app.middleware("http") 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 = """ CuratorBot: sign in

CuratorBot

A research prototype shared with the project team. Enter the passcode you were given.

__ERROR__
""" @app.get("/favicon.svg", include_in_schema=False) @app.get("/favicon.ico", include_in_schema=False) 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"}) @app.get("/login") def login_page() -> Response: if not settings().site_passcode: return RedirectResponse("/", status_code=303) return HTMLResponse(LOGIN_PAGE.replace("__ERROR__", "")) @app.post("/login") 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__", ''), 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 @app.get("/logout") 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"))) @app.get("/") 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"}) @app.get("/api/health") 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"]) @app.get("/api/interviews") 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) @app.get("/api/interviews/{iid}") 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) @app.post("/api/reload") 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())} @app.post("/api/analyse") 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 @app.post("/api/transcribe") 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"]}