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 = """ Zugang

🔒

Persönlicher Zugang nötig

Senioren: bitte die App über den persönlichen Zugangslink öffnen.

Betreuer und Admins können sich mit Benutzername und Passwort anmelden:

Mit Benutzername & Passwort anmelden

""" _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 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")