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, HTMLResponse

from app import __version__
from app.config import settings
from urllib.parse import urlencode

from app.auth import authenticate, _bearer_token, CAPABILITY_COOKIE
from app.dependencies import get_store
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

# --- Logging: App-eigene Logger sichtbar machen ---------------------------
# uvicorn konfiguriert nur seine eigenen Logger; "app.*"-INFO-Logs (z. B. die
# Notruf-Diagnose) würden sonst durch den Default-Level (WARNING) verschluckt.
import logging
import os

_app_logger = logging.getLogger("app")
_app_logger.setLevel(getattr(logging, os.getenv("LOG_LEVEL", "INFO").upper(), logging.INFO))
if not _app_logger.handlers:
    _log_handler = logging.StreamHandler()
    _log_handler.setFormatter(logging.Formatter("%(levelname)s:%(name)s:%(message)s"))
    _app_logger.addHandler(_log_handler)
    _app_logger.propagate = False


@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="Alexis", version=__version__, 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.
_NO_ACCESS_PAGE = """<!doctype html><html lang="de"><head><meta charset="utf-8">
<meta name="viewport" content="width=device-width,initial-scale=1"><title>Zugang</title>
<style>body{font-family:system-ui,-apple-system,sans-serif;background:#f8fafc;color:#1e293b;
display:grid;place-items:center;min-height:100vh;margin:0;padding:1.5rem}
.box{max-width:30rem;text-align:center}h1{font-size:1.3rem;margin:0 0 .75rem}
p{color:#475569;line-height:1.6;margin:.5rem 0}.emoji{font-size:2.5rem}
.btn{display:inline-block;margin-top:1.25rem;background:#2563eb;color:#fff;text-decoration:none;
font-weight:600;font-size:1.05rem;padding:.8rem 1.4rem;border-radius:.75rem}</style></head>
<body><div class="box"><div class="emoji">🔒</div>
<h1>Persönlicher Zugang nötig</h1>
<p>Senioren: bitte die App über den persönlichen Zugangslink öffnen.</p>
<p>Betreuer und Admins können sich mit Benutzername und Passwort anmelden:</p>
<a class="btn" href="/login">Mit Benutzername &amp; Passwort anmelden</a>
</div></body></html>"""


_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):
        # Ein-Klick-Link: ?k=<token> -> Token im Cookie speichern, aus der URL entfernen.
        k = request.query_params.get("k")
        if k and get_store().get_user_by_token(k):
            params = {kk: vv for kk, vv in request.query_params.items() if kk != "k"}
            clean = path + ("?" + urlencode(params) if params else "")
            resp = RedirectResponse(clean, status_code=303)
            resp.set_cookie(CAPABILITY_COOKIE, k, httponly=True, secure=True,
                            samesite="lax", max_age=31536000)
            return resp
        client_host = request.client.host if request.client else ""
        token = (_bearer_token(request.headers.get("authorization"))
                 or k or request.cookies.get(CAPABILITY_COOKIE))
        if authenticate(request.headers, client_host, token) is None:
            # Kein gültiger Zugang -> freundliche Hinweisseite (KEIN SSO-Redirect, der
            # nach dem nginx-Cutover in eine Schleife liefe). Senioren nutzen ihren Link.
            return HTMLResponse(_NO_ACCESS_PAGE, 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")