spb/uqo-chat

1"""FastAPI application: middlewares, routers, static frontend, background maintenance."""23from __future__ import annotations45import asyncio6from collections.abc import AsyncIterator7from contextlib import asynccontextmanager8from pathlib import Path910from fastapi import FastAPI, Request11from fastapi.middleware.cors import CORSMiddleware12from fastapi.responses import FileResponse, JSONResponse13from fastapi.staticfiles import StaticFiles14from starlette.middleware.base import BaseHTTPMiddleware1516from app.api.v1 import api_router17from app.core.config import get_settings18from app.core.logging import configure_logging, get_logger19from app.db import init_db20from app.rag import retriever21from app.services import conversations as conv_service22from app.services import courses as course_service23from app.services import files as file_service2425settings = get_settings()26configure_logging(settings.LOG_LEVEL, json_output=not settings.is_dev)27log = get_logger("main")282930async def _maintenance() -> None:31    while True:32        try:33            n = await file_service.purge_expired()34            r = await conv_service.redact_old_messages(settings.MESSAGE_RETENTION_MONTHS)35            if n or r:36                log.info("maintenance", purged_files=n, redacted_messages=r)37        except Exception as exc:  # noqa: BLE00138            log.warning("maintenance_failed", error=str(exc))39        await asyncio.sleep(3600)404142@asynccontextmanager43async def lifespan(_: FastAPI) -> AsyncIterator[None]:44    await init_db()45    await course_service.seed_courses()46    import app.tools.all  # noqa: F401 — register tools4748    n = await retriever.rebuild_index()49    log.info("startup", chunks=n, env=settings.APP_ENV, port=settings.PORT)50    task = asyncio.create_task(_maintenance())51    yield52    task.cancel()535455app = FastAPI(title="UQO-Chat API", version="0.1.0", lifespan=lifespan,56              docs_url="/api/docs" if settings.is_dev else None, redoc_url=None,57              openapi_url="/api/openapi.json" if settings.is_dev else None)5859app.add_middleware(CORSMiddleware, allow_origins=settings.cors_origins, allow_credentials=True,60                   allow_methods=["*"], allow_headers=["*"])616263class SecurityHeaders(BaseHTTPMiddleware):64    async def dispatch(self, request: Request, call_next):  # noqa: ANN001, ANN20165        resp = await call_next(request)66        resp.headers.setdefault("X-Content-Type-Options", "nosniff")67        resp.headers.setdefault("X-Frame-Options", "DENY")68        resp.headers.setdefault("Referrer-Policy", "strict-origin-when-cross-origin")69        resp.headers.setdefault("Permissions-Policy", "camera=(), geolocation=()")70        if not settings.is_dev:71            resp.headers.setdefault("Strict-Transport-Security", "max-age=31536000; includeSubDomains")72        if request.url.path == "/" or request.url.path.endswith(".html") or \73                "text/html" in resp.headers.get("content-type", ""):74            resp.headers.setdefault(75                "Content-Security-Policy",76                "default-src 'self'; img-src 'self' data: blob: https:; "77                "style-src 'self' 'unsafe-inline' https://fonts.googleapis.com; "78                "font-src 'self' data: https://fonts.gstatic.com; "79                "script-src 'self'; connect-src 'self'; worker-src 'self'; "80                "frame-ancestors 'none'; base-uri 'self'; form-action 'self'")81        return resp828384app.add_middleware(SecurityHeaders)85app.include_router(api_router)868788@app.exception_handler(429)89async def _rate_limited(_: Request, exc):  # noqa: ANN001, ANN20190    return JSONResponse(status_code=429, content={"detail": getattr(exc, "detail",91                                                                     "Trop de requêtes.")})929394# ---- Frontend (built SPA) ---------------------------------------------------------------95dist: Path = settings.FRONTEND_DIST96if dist.exists():97    app.mount("/assets", StaticFiles(directory=dist / "assets"), name="assets")9899    @app.api_route("/{full_path:path}", methods=["GET", "HEAD"], include_in_schema=False)100    async def spa(full_path: str):  # noqa: ANN201101        candidate = dist / full_path102        if full_path and candidate.is_file():103            return FileResponse(candidate)104        return FileResponse(dist / "index.html")105else:106    @app.get("/", include_in_schema=False)107    async def root() -> dict:108        return {"service": "uqo-chat", "docs": "/api/docs", "frontend": "not built"}109