my_voice_assistant_v3_jamulix/app/main.py

import asyncio
import time
from contextlib import asynccontextmanager
from pathlib import Path

from fastapi import FastAPI, Request
from fastapi.staticfiles import StaticFiles
from starlette.responses import RedirectResponse, JSONResponse

from app.config import settings
from app.auth import authenticate, _bearer_token
from app.core.warmup import warmup_local_models
from app.metrics import metrics
from app.api.health import router as health_router
from app.api.chat import router as chat_router
from app.api.transcribe import router as transcribe_router
from app.api.speak import router as speak_router
from app.api.devices import router as devices_router
from app.api.sessions import router as sessions_router
from app.api.config import router as config_router
from app.api.admin import router as admin_router
from app.api.me import router as me_router
from app.api.metrics import router as metrics_router
from app.api.ws import router as ws_router

@asynccontextmanager
async def _lifespan(app: FastAPI):
    # Lokale Modelle im Hintergrund vorladen -> Server ist sofort verfuegbar,
    # der erste Nutzer zahlt nicht den Kaltstart.
    task = asyncio.create_task(warmup_local_models())
    yield
    if not task.done():
        task.cancel()


app = FastAPI(title="Voice Assistant Gateway", lifespan=_lifespan)


@app.middleware("http")
async def record_metrics(request: Request, call_next):
    start = time.perf_counter()
    response = await call_next(request)
    duration = time.perf_counter() - start
    # Route-Template (z. B. /api/sessions/{session_id}/route) statt konkreter URL,
    # um die Label-Kardinalitaet niedrig zu halten.
    route = request.scope.get("route")
    path = getattr(route, "path", request.url.path)
    labels = {"method": request.method, "path": path}
    metrics.inc("http_requests_total", {**labels, "status": response.status_code})
    metrics.observe("http_request_duration_seconds", duration, labels)
    return response


# Pfade mit eigener Auth, Health-Checks und Favicons: nicht gaten.
_PUBLIC_PREFIXES = ("/api", "/ws", "/health", "/favicon", "/apple-touch-icon")


@app.middleware("http")
async def gate_web_ui(request: Request, call_next):
    """Schuetzt die statische Web-UI: unauthentifizierte Seitenaufrufe -> SSO-Login (sonst 401).

    Defense-in-Depth zusaetzlich zu SSOwat: Selbst wenn der Reverse-Proxy einen
    unauthentifizierten Request durchliesse (oder jemand den Port direkt trifft),
    bekommt er die Seite nicht ausgeliefert. API/WS haben ihre eigene Auth (require_user
    -> 401/JSON), Health-Checks bleiben offen. Bei abgeschalteter Auth (LAN-Dev) liefert
    authenticate() einen anonymen Nutzer -> die Seite wird wie bisher ausgeliefert.
    """
    path = request.url.path
    if not path.startswith(_PUBLIC_PREFIXES):
        client_host = request.client.host if request.client else ""
        token = _bearer_token(request.headers.get("authorization"))
        if authenticate(request.headers, client_host, token) is None:
            login = settings.sso_login_url.strip()
            if login:
                return RedirectResponse(login, status_code=302)
            return JSONResponse({"detail": "Authentication required"}, status_code=401)
    return await call_next(request)


app.include_router(health_router)
app.include_router(chat_router, prefix="/api")
app.include_router(transcribe_router, prefix="/api")
app.include_router(speak_router, prefix="/api")
app.include_router(devices_router, prefix="/api")
app.include_router(sessions_router, prefix="/api")
app.include_router(config_router, prefix="/api")
app.include_router(admin_router, prefix="/api")
app.include_router(me_router, prefix="/api")
app.include_router(metrics_router, prefix="/api")
app.include_router(ws_router)

# Statische Web-UI (same-origin -> kein CORS). Muss NACH allen API-/WS-Routen
# gemountet werden, damit "/" nur die uebrigen Pfade abfaengt.
class _NoCacheStaticFiles(StaticFiles):
    """Liefert die UI ohne Caching aus -> Aenderungen sind sofort sichtbar (kein Safari-Cache)."""

    def file_response(self, *args, **kwargs):
        response = super().file_response(*args, **kwargs)
        response.headers["Cache-Control"] = "no-cache, no-store, must-revalidate"
        return response


_WEB_DIR = Path(__file__).resolve().parent / "web"
if _WEB_DIR.is_dir():
    app.mount("/", _NoCacheStaticFiles(directory=str(_WEB_DIR), html=True), name="web")