api/server.py

"""FastAPI service layer for the AI Video Intelligence Agent.

Required endpoints
    GET  /health
    GET  /metrics
    POST /video/analyze
    GET  /video/{video_id}/timeline
    GET  /video/{video_id}/bookmarks
    POST /video/search

Optional endpoints
    POST /video/ask
    GET  /video/{video_id}/summary
    GET  /video/{video_id}/transcript
    GET  /video/{video_id}/visual-events

In ``DEMO_MODE`` / ``USE_PRECOMPUTED`` (the defaults) the service reads the
committed sample artifacts and never loads a heavy model. Live analysis only
runs when precomputed mode is disabled *and* the local ML deps are installed.

Run locally:  uvicorn api.server:app --reload --port 8000
"""

from __future__ import annotations

import tempfile
from pathlib import Path
from typing import List, Optional

from fastapi import FastAPI, File, Form, HTTPException, UploadFile
from pydantic import BaseModel

from src import storage
from src.config import CONFIG
from src.metrics import latency_summary
from src.search import VideoSearch
from src.summarize import answer_question, extractive_summary
from src.visual_analysis import object_frequency

app = FastAPI(
    title="AI Video Intelligence Agent",
    description=(
        "Multimodal analysis of short learning videos: visual events, "
        "Thai-English ASR, searchable timeline, automatic bookmarks."
    ),
    version="0.1.0",
)


# --------------------------------------------------------------------------- #
# Request models
# --------------------------------------------------------------------------- #
class SearchRequest(BaseModel):
    video_id: str
    query: str
    top_k: int = 10


class AskRequest(BaseModel):
    video_id: str
    question: str
    top_k: int = 3


def _require_timeline(video_id: str) -> dict:
    timeline = storage.load_timeline(video_id)
    if timeline is None:
        raise HTTPException(status_code=404, detail=f"Unknown video_id '{video_id}'.")
    return timeline


# --------------------------------------------------------------------------- #
# Core endpoints
# --------------------------------------------------------------------------- #
@app.get("/health")
def health() -> dict:
    return {
        "status": "ok",
        "demo_mode": CONFIG.demo_mode,
        "use_precomputed": CONFIG.use_precomputed,
        "available_videos": storage.list_sample_video_ids(),
    }


@app.get("/metrics")
def metrics() -> dict:
    """Aggregate processing + search metrics across all available videos."""
    videos = {}
    all_latencies: List[float] = []
    for vid in storage.list_sample_video_ids():
        m = storage.load_metrics(vid)
        if not m:
            continue
        videos[vid] = m
        sl = (m.get("search_latency") or {})
        # Reconstruct latency points isn't stored; surface the summary instead.
        if sl.get("p50_ms") is not None:
            all_latencies.append(sl["p50_ms"])
    return {
        "n_videos": len(videos),
        "videos": videos,
        "search_latency_p50_across_videos_ms": (
            latency_summary(all_latencies)["p50_ms"] if all_latencies else None
        ),
        "config": CONFIG.public_dict(),
    }


@app.post("/video/analyze")
async def analyze(
    file: Optional[UploadFile] = File(default=None),
    url: Optional[str] = Form(default=None),
    video_id: Optional[str] = Form(default=None),
) -> dict:
    """Analyze an uploaded video / a URL, or return a precomputed analysis by id.

    * No file/url + known ``video_id``  -> returns the precomputed artifacts.
    * File or URL, precomputed mode     -> returns a note (demo stays stable).
    * File or URL, live mode            -> runs the local pipeline (needs ML deps).

    ``url`` may point to YouTube, Vimeo, a direct link, etc. (handled by yt-dlp).
    """
    if file is None and not url:
        if not video_id:
            raise HTTPException(
                status_code=400,
                detail="Provide a video file, a 'url', or a known 'video_id'.",
            )
        timeline = _require_timeline(video_id)
        return {
            "video_id": video_id,
            "mode": "precomputed",
            "timeline": timeline,
            "bookmarks": storage.load_bookmarks(video_id),
            "metrics": storage.load_metrics(video_id),
        }

    if CONFIG.use_precomputed or CONFIG.demo_mode:
        return {
            "mode": "demo",
            "message": (
                "Demo/precomputed mode is on — live analysis (file upload and "
                "URL ingestion) is disabled for stability. Set "
                "USE_PRECOMPUTED=false and DEMO_MODE=false to enable local "
                "analysis, or query a precomputed sample."
            ),
            "available_videos": storage.list_sample_video_ids(),
        }

    # ---- live analysis path -------------------------------------------- #
    try:
        from src.pipeline import analyze_video  # lazy: pulls heavy deps
    except Exception as exc:  # pragma: no cover
        raise HTTPException(
            status_code=503,
            detail=f"Local analysis dependencies unavailable: {exc}",
        )

    tmp_path: Optional[Path] = None
    try:
        if file is not None:
            suffix = Path(file.filename or "upload.mp4").suffix or ".mp4"
            with tempfile.NamedTemporaryFile(delete=False, suffix=suffix) as tmp:
                tmp.write(await file.read())
                tmp_path = Path(tmp.name)
            source: object = tmp_path
        else:
            source = url  # URL string -> resolved/downloaded by the pipeline

        artifacts = analyze_video(source, video_id=video_id, config=CONFIG)
    except (ValueError, FileNotFoundError) as exc:
        raise HTTPException(status_code=400, detail=str(exc))
    finally:
        if tmp_path is not None:
            tmp_path.unlink(missing_ok=True)

    return {
        "video_id": artifacts["metrics"]["video_id"],
        "mode": "live",
        "timeline": artifacts["timeline"],
        "bookmarks": artifacts["bookmarks"],
        "metrics": artifacts["metrics"],
    }


@app.get("/video/{video_id}/timeline")
def get_timeline(video_id: str) -> dict:
    return _require_timeline(video_id)


@app.get("/video/{video_id}/bookmarks")
def get_bookmarks(video_id: str) -> List[dict]:
    bookmarks = storage.load_bookmarks(video_id)
    if bookmarks is None:
        raise HTTPException(status_code=404, detail=f"Unknown video_id '{video_id}'.")
    return bookmarks


@app.post("/video/search")
def search(req: SearchRequest) -> dict:
    timeline = _require_timeline(req.video_id)
    return VideoSearch(timeline, CONFIG).search(req.query, top_k=req.top_k)


# --------------------------------------------------------------------------- #
# Optional endpoints
# --------------------------------------------------------------------------- #
@app.post("/video/ask")
def ask(req: AskRequest) -> dict:
    timeline = _require_timeline(req.video_id)
    return answer_question(req.question, timeline, CONFIG, top_k=req.top_k)


@app.get("/video/{video_id}/summary")
def get_summary(video_id: str) -> dict:
    timeline = _require_timeline(video_id)
    return extractive_summary(timeline, CONFIG)


@app.get("/video/{video_id}/transcript")
def get_transcript(video_id: str) -> List[dict]:
    transcript = storage.load_transcript(video_id)
    if transcript is None:
        raise HTTPException(status_code=404, detail=f"Unknown video_id '{video_id}'.")
    return transcript


@app.get("/video/{video_id}/visual-events")
def get_visual_events(video_id: str) -> dict:
    visual = storage.load_visual_events(video_id)
    if visual is None:
        raise HTTPException(status_code=404, detail=f"Unknown video_id '{video_id}'.")
    return {
        "video_id": video_id,
        "n_frames": len(visual),
        "object_frequency": object_frequency(visual),
        "frames": visual,
    }