spb/qc-election

1# QC Élection Forecast — Plateforme de prévision électorale du Québec 20262# Auteur : Simon-Pierre Boucher3# Contact : contact@spboucher.ai4# https://www.qc-election.com5"""Application FastAPI — API + frontend statique + rafraîchissement planifié."""6from __future__ import annotations78import asyncio9import logging10import time11from collections import defaultdict, deque12from contextlib import asynccontextmanager1314from fastapi import FastAPI, Request15from fastapi.responses import FileResponse, JSONResponse16from fastapi.staticfiles import StaticFiles1718from .config import FRONTEND_DIR, settings19from .db import init_db20from .api.public import router as public_router, router_v321from .api.admin import router as admin_router2223logging.basicConfig(24    level=logging.INFO,25    format='{"t":"%(asctime)s","lvl":"%(levelname)s","mod":"%(name)s","msg":"%(message)s"}')26log = logging.getLogger("app")2728# --- Limitation de débit simple (par IP, fenêtre glissante) -------------------29RATE_LIMITS = {"POST": (30, 60.0), "GET": (240, 60.0)}  # (requêtes, fenêtre s)30_hits: dict[str, deque] = defaultdict(deque)313233async def rate_limiter(request: Request, call_next):34    if request.url.path.startswith("/api"):35        limit, window = RATE_LIMITS.get(request.method, (240, 60.0))36        key = f"{request.client.host if request.client else '?'}|{request.method}"37        now = time.monotonic()38        dq = _hits[key]39        while dq and now - dq[0] > window:40            dq.popleft()41        if len(dq) >= limit:42            return JSONResponse({"detail": "Trop de requêtes — réessayez sous peu."},43                                status_code=429)44        dq.append(now)45    return await call_next(request)464748async def _scheduler():49    """Rafraîchissement automatique du pipeline (collecte + modèle + simulation).50    Cadence resserrée à 3 h en période de campagne (≤ 45 jours du scrutin) :51    les sondages tombent chaque jour, ils doivent entrer sans délai."""52    from datetime import date as _date53    from .pipeline import run_pipeline54    await asyncio.sleep(60)55    while True:56        try:57            report = await asyncio.to_thread(run_pipeline, True)58            log.info(f"pipeline: {report}")59        except Exception as e:  # le service continue même si un cycle échoue60            log.error(f"pipeline en échec: {e}")61        days_left = (settings.election_date - _date.today()).days62        hours = 3.0 if 0 <= days_left <= 45 else settings.refresh_interval_hours63        await asyncio.sleep(hours * 3600)646566@asynccontextmanager67async def lifespan(app: FastAPI):68    init_db()69    task = asyncio.create_task(_scheduler())70    yield71    task.cancel()727374app = FastAPI(title=settings.app_name, version=settings.model_version, lifespan=lifespan,75              docs_url="/api/docs", openapi_url="/api/openapi.json")76app.middleware("http")(rate_limiter)77app.include_router(public_router)78app.include_router(router_v3)79app.include_router(admin_router)808182@app.get("/health")83def health():84    return {"status": "ok", "app": settings.app_name, "version": settings.model_version}858687PAGES = ["index", "aujourdhui", "carte", "circonscriptions", "sondages",88         "signaux", "simulateur", "intelligence", "methodologie", "backtest",89         "admin"]9091# Cache-busting : la version du modèle est injectée dans les URLs d'assets92# (style.css?v=…, app.js?v=…) et les pages HTML sont servies en no-cache —93# chaque release est visible immédiatement, sans vider le cache du navigateur.94_page_cache: dict[str, tuple[float, str]] = {}959697def _render_page(name: str) -> str:98    f = FRONTEND_DIR / f"{name}.html"99    mtime = f.stat().st_mtime100    cached = _page_cache.get(name)101    if cached and cached[0] == mtime:102        return cached[1]103    html = f.read_text()104    v = settings.model_version105    html = html.replace('href="/assets/style.css"',106                        f'href="/assets/style.css?v={v}"')107    html = html.replace('src="/assets/app.js"',108                        f'src="/assets/app.js?v={v}"')109    _page_cache[name] = (mtime, html)110    return html111112113for page in PAGES:114    path = "/" if page == "index" else f"/{page}"115116    def make_handler(p=page):117        def handler():118            from fastapi.responses import HTMLResponse119            return HTMLResponse(_render_page(p),120                                headers={"Cache-Control": "no-cache"})121        return handler122    app.api_route(path, methods=["GET", "HEAD"],123                  include_in_schema=False)(make_handler())124125app.mount("/assets", StaticFiles(directory=FRONTEND_DIR / "assets"), name="assets")126