video-image-production-build/visualization/web/app.py

#!/usr/bin/env python3
"""Production Build Run 的文件型只读可视化服务。"""

from __future__ import annotations

import os
from pathlib import Path

from fastapi import FastAPI, HTTPException, Query
from fastapi.middleware.cors import CORSMiddleware
from fastapi.middleware.gzip import GZipMiddleware
from fastapi.responses import FileResponse, HTMLResponse

from .store import DEFAULT_RUN_ROOT, FileRunStore, media_type


WEB_DIR = Path(__file__).resolve().parent
RUN_ROOT = Path(
    os.getenv("PRODUCTION_VISUALIZATION_RUN_ROOT", str(DEFAULT_RUN_ROOT))
)
store = FileRunStore(RUN_ROOT)

app = FastAPI(
    title="Production Build · Run Viewer",
    version="0.1.0",
)
app.add_middleware(GZipMiddleware, minimum_size=1024)
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_methods=["GET"],
    allow_headers=["*"],
)


@app.middleware("http")
async def _no_cache(request, call_next):
    response = await call_next(request)
    response.headers["Cache-Control"] = "no-store"
    return response


def _run_ref(run_id: int):
    try:
        return store.get(run_id)
    except KeyError as exc:
        raise HTTPException(status_code=404, detail="run not found") from exc


@app.get("/", response_class=HTMLResponse)
def index():
    return (WEB_DIR / "index.html").read_text(encoding="utf-8")


@app.get("/api/health")
def health():
    refs = store.discover()
    return {
        "ok": True,
        "data_source": "run_files",
        "run_root": str(store.run_root),
        "run_count": len(refs),
    }


@app.get("/api/runs")
def list_runs(
    status: str | None = None,
    page: int = Query(1, ge=1),
    page_size: int = Query(50, ge=1, le=200),
):
    runs = [store.brief(ref) for ref in store.discover()]
    if status:
        runs = [run for run in runs if run["status"] == status]
    start = (page - 1) * page_size
    return {
        "total": len(runs),
        "page": page,
        "page_size": page_size,
        "runs": runs[start : start + page_size],
    }


@app.get("/api/runs/{run_id}")
def get_run(run_id: int):
    return store.detail(_run_ref(run_id))


@app.get("/api/runs/{run_id}/snapshots")
def get_snapshots(run_id: int):
    return store.snapshots(_run_ref(run_id))


@app.get("/api/runs/{run_id}/flat")
def get_flat(run_id: int, round: int | None = None):
    return store.flat(_run_ref(run_id), round)


@app.get("/api/runs/{run_id}/annotations")
def get_annotations(run_id: int):
    _run_ref(run_id)
    return {
        "run_id": run_id,
        "by_node": {},
        "by_call": {},
        "docs": {},
    }


@app.get("/api/tools")
def get_tools(agent: str = "production"):
    return {"agent": agent, "tools": store.tools()}


@app.get("/api/runs/{run_id}/media/{relative_path:path}")
def get_media(run_id: int, relative_path: str):
    ref = _run_ref(run_id)
    try:
        path = store.media_path(ref, relative_path)
    except FileNotFoundError as exc:
        raise HTTPException(status_code=404, detail="media not found") from exc
    return FileResponse(path, media_type=media_type(path))


if __name__ == "__main__":
    import uvicorn

    uvicorn.run(app, host="127.0.0.1", port=8930)