Blame for app/server.py

"""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 = """<!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>"""


@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__", '<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


@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"]}