app/server.py

"""Local FastAPI app: search box + clickable, seekable results.

Endpoints
  GET  /                 -> the search page
  POST /api/search       -> {query, top_k} -> ranked moments (JSON)
  POST /api/query_audio  -> multipart audio file -> {text, results}
  GET  /api/stats        -> index / manifest stats
  GET  {audio_route}/... -> the original recordings, served with HTTP Range so the
                            <audio> element seeks instantly to a timestamp.

Run:  python -m app.server     (or)    uvicorn app.server:app --host 127.0.0.1 --port 8000
"""
from __future__ import annotations

import os
import sys
import tempfile
import threading
from pathlib import Path
from urllib.parse import quote, unquote

from fastapi import FastAPI, File, Form, UploadFile
from fastapi.responses import FileResponse, JSONResponse
from fastapi.staticfiles import StaticFiles

from app.config import get_config
from app.manifest import Manifest
from app.query import Searcher, exact_phrase
from app import audio_source

if sys.platform == "win32":  # make sure Devanagari prints fine in the console
    try:
        sys.stdout.reconfigure(encoding="utf-8")  # type: ignore[attr-defined]
    except Exception:  # noqa: BLE001
        pass

cfg = get_config()
cfg.ensure_dirs()
WEB_DIR = Path(__file__).parent / "web"
AUDIO_ROUTE = cfg.server["audio_route"]
AUDIO_MODE = (cfg.get("audio", {}) or {}).get("mode", "local")

app = FastAPI(title="SearchAudio", version="0.1.0")
_searcher = Searcher(cfg)

# Static assets. StaticFiles speaks HTTP Range, so seeking is instant.
app.mount("/static", StaticFiles(directory=str(WEB_DIR)), name="static")
# Local audio is served from disk only in local mode; archive mode streams from archive.org.
if AUDIO_MODE == "local":
    app.mount(AUDIO_ROUTE, StaticFiles(directory=str(cfg.audio_dir)), name="audio")


def _audio_url(source_file: str) -> str:
    return audio_source.audio_url(source_file, cfg)


def _pretty_name(source_file: str) -> str:
    """Human-readable discourse name from a stored source_file path.

    Stored paths are URL-quoted with %5C as the path separator (see audio_source);
    show just the final component without its extension.
    """
    leaf = unquote(source_file).replace("\\", "/").split("/")[-1]
    stem, _, ext = leaf.rpartition(".")
    return stem if stem and len(ext) <= 4 else leaf


# Distinct recordings for the discourse filter; one projection over the table,
# computed once (the deployed index is read-only).
_recordings_cache: list | None = None
_recordings_lock = threading.Lock()


def _recordings() -> list:
    global _recordings_cache
    if _recordings_cache is None:
        with _recordings_lock:
            if _recordings_cache is None:
                rows = [
                    {
                        "id": r["recording_id"],
                        "name": _pretty_name(r["source_file"]),
                        "passages": r["passages"],
                    }
                    for r in _searcher.store.recordings_summary()
                ]
                rows.sort(key=lambda r: r["name"].lower())
                _recordings_cache = rows
    return _recordings_cache


def _result_payload(results) -> list:
    out = []
    for r in results:
        d = r.to_dict()
        d["audio_url"] = _audio_url(r.source_file)
        out.append(d)
    return out


@app.get("/")
def index() -> FileResponse:
    return FileResponse(str(WEB_DIR / "index.html"))


@app.post("/api/search")
def api_search(payload: dict) -> JSONResponse:
    query = (payload or {}).get("query", "")
    top_k = (payload or {}).get("top_k")
    recording_id = (payload or {}).get("recording_id") or None
    if recording_id and recording_id not in {r["id"] for r in _recordings()}:
        return JSONResponse({"error": "unknown recording_id"}, status_code=400)
    results = _searcher.search_text(query, top_k=top_k, recording_id=recording_id)
    return JSONResponse(
        {
            "query": query,
            "mode": "exact" if exact_phrase(query) else "semantic",
            "recording_id": recording_id,
            "count": len(results),
            "results": _result_payload(results),
        }
    )


@app.get("/api/recordings")
def api_recordings() -> JSONResponse:
    recs = _recordings()
    return JSONResponse({"count": len(recs), "recordings": recs})


async def api_query_audio(file: UploadFile = File(...), top_k: int = Form(None)) -> JSONResponse:
    suffix = Path(file.filename or "clip.webm").suffix or ".webm"
    with tempfile.NamedTemporaryFile(delete=False, suffix=suffix) as tmp:
        tmp.write(await file.read())
        tmp_path = tmp.name
    try:
        text, results = _searcher.search_audio(tmp_path, top_k=top_k)
    finally:
        try:
            Path(tmp_path).unlink(missing_ok=True)
        except Exception:  # noqa: BLE001
            pass
    return JSONResponse({"text": text, "count": len(results), "results": _result_payload(results)})


# Register the audio-query route ONLY in local mode. In archive/online mode the route is
# absent (POST -> 404) and app.asr / whisperx are never imported (keeps the image slim).
if AUDIO_MODE == "local":
    app.post("/api/query_audio")(api_query_audio)


@app.get("/api/config")
def api_config() -> JSONResponse:
    return JSONResponse({"mode": AUDIO_MODE, "audio_query": AUDIO_MODE == "local"})


@app.get("/api/stats")
def api_stats() -> JSONResponse:
    manifest = Manifest(cfg.manifest_path)
    counts = manifest.counts()
    manifest.close()
    try:
        n_passages = _searcher.store.count()
    except Exception:  # noqa: BLE001
        n_passages = 0
    return JSONResponse(
        {
            "files": counts,
            "passages": n_passages,
            "recordings": len(_recordings()),
            "reranker": bool(cfg.reranker["enabled"]),
            "hybrid": bool(cfg.search["hybrid"]),
            "vector_index": _searcher.store.has_vector_index(),
        }
    )


# Free Spaces sleep when idle, so most visits boot a cold process. Load the heavy
# pieces (bge-m3, indices, recordings list) in the background at startup instead of
# on the first user's search. Daemon thread: never blocks port binding or shutdown.
# Tests disable via SEARCHAUDIO_WARMUP=0 (they only exercise routing, not models).
def _warm() -> None:
    try:
        _searcher.store.ensure_fts_index()
        _recordings()
        _searcher.search_text("warm up")
    except Exception:  # noqa: BLE001 — warm-up is best-effort by definition
        pass


if os.environ.get("SEARCHAUDIO_WARMUP", "1") != "0":
    threading.Thread(target=_warm, name="warmup", daemon=True).start()


def run() -> None:
    import uvicorn

    uvicorn.run(
        "app.server:app",
        host=cfg.server["host"],
        port=int(cfg.server["port"]),
        workers=1,  # ONE GPU worker: never load the models N times into VRAM
        log_level="info",
    )


if __name__ == "__main__":
    run()