# ----------------------------------------------------------------------------- # Fabri-Ka — Agrégateur de produits québécois # Auteur : Simon-Pierre Boucher — contact@spboucher.ai # web.py : API FastAPI + service du frontend React (dist/) # ----------------------------------------------------------------------------- from __future__ import annotations import json import sqlite3 import threading import time from datetime import date from pathlib import Path from fastapi import BackgroundTasks, Body, FastAPI, HTTPException, Query, Request from fastapi.middleware.gzip import GZipMiddleware from fastapi.responses import FileResponse, RedirectResponse from fastapi.staticfiles import StaticFiles from . import auth, db, hubfav, inscriptions, kaid, seo from .schema import CATEGORIES, REGIONS app = FastAPI(title="Fabri-Ka API", docs_url="/api/docs", openapi_url="/api/openapi.json") app.add_middleware(GZipMiddleware, minimum_size=1000) # Connexion « KA ID » (SSO Groupe KA) — AVANT le catch-all SPA plus bas app.include_router(auth.router) # Favoris « Mon univers Ka » (magasin central du hub Groupe KA, zéro stockage local) app.include_router(hubfav.router) # personnalisation KA ID v2 (fabrika/kaid.py, canonique ka-ui.git/kaid) kaid.init("fabri-ka") app.include_router(kaid.build_router(auth.current_user)) def _kaid_features(d: dict) -> dict: """Caractéristiques d'un produit pour le profil de préférences KA ID.""" return {k: v for k, v in { "category": d.get("category"), "store": d.get("store_name"), "region": d.get("store_region"), "origin": d.get("origin_class"), "price": d.get("price"), }.items() if v not in (None, "")} # Auto-inscription des artisans (« Inscrire mon atelier ») — quarantaine app.include_router(inscriptions.router) FRONT_DIST = Path(__file__).resolve().parent.parent / "frontend" / "dist" def q(con: sqlite3.Connection, sql: str, args=()) -> list[dict]: return [dict(r) for r in con.execute(sql, args)] def _product_out(r: dict) -> dict: r["images"] = json.loads(r.get("images") or "[]") r["tags"] = json.loads(r.get("tags") or "[]") if "details" in r: r["details"] = json.loads(r["details"]) if r["details"] else None return r _FTS_SAFE = set("abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789àâäéèêëîïôöùûüçÀÂÄÉÈÊËÎÏÔÖÙÛÜÇ- '") def fts_query(text: str) -> str: words = "".join(c if c in _FTS_SAFE else " " for c in text).split() return " ".join(f'"{w}"*' for w in words[:8]) @app.get("/api/products") def products(request: Request, q_text: str | None = Query(None, alias="q"), category: str | None = None, region: str | None = None, store: str | None = None, origin: str | None = None, price_min: float | None = None, price_max: float | None = None, available: bool | None = None, on_sale: bool | None = None, kind: str | None = None, sort: str = "recent", page: int = 1, per_page: int = Query(24, le=100)): con = db.connect() try: # listing_status : cartes-cadeaux/ateliers/abonnements/billets exclus # et aberrations en quarantaine ne sont jamais publiés where, args = ["p.active=1", "p.listing_status='published'"], [] joins = "FROM products p JOIN stores s ON s.id = p.store_id" if kind == "fabricant": # « fabricants seulement » : écarte revendeurs et hors-mission # (les collectifs d'artisans restent : produits faits au Québec) where.append("COALESCE(s.store_kind,'') NOT IN ('revendeur','hors_mission')") if q_text: joins += " JOIN products_fts f ON f.uid = p.uid" where.append("products_fts MATCH ?") args.append(fts_query(q_text)) if category: where.append("p.category=?"); args.append(category) if region: where.append("s.region=?"); args.append(region) if store: where.append("p.store_id=?"); args.append(store) if origin: where.append("s.origin_class IN (%s)" % ",".join("?" * len(origin.split(",")))) args += origin.split(",") if price_min is not None: where.append("p.price>=?"); args.append(price_min) if price_max is not None: where.append("p.price<=?"); args.append(price_max) if available is not None: where.append("p.available=?"); args.append(int(available)) if on_sale: # compare > price*10 (rabais >90 %) = donnée marchande aberrante # (prix barré saisi en cents chez certains Shopify) — exclue. where.append("p.compare_at_price IS NOT NULL AND p.price IS NOT NULL" " AND p.compare_at_price > p.price" " AND p.compare_at_price <= p.price * 10") wsql = " AND ".join(where) order = {"recent": "p.first_seen DESC", "price_asc": "p.price IS NULL, p.price ASC", "price_desc": "p.price IS NULL, p.price DESC", "discount": ("(p.compare_at_price IS NULL OR p.price IS NULL" " OR p.compare_at_price <= p.price" " OR p.compare_at_price > p.price * 10)," " (1.0 - p.price / p.compare_at_price) DESC"), "title": "p.title COLLATE NOCASE"}.get(sort, "p.first_seen DESC") total = con.execute(f"SELECT COUNT(*) {joins} WHERE {wsql}", args).fetchone()[0] rows = q(con, f"""SELECT p.*, s.name AS store_name, s.region AS store_region, s.city AS store_city, s.origin_class, s.store_kind {joins} WHERE {wsql} ORDER BY {order} LIMIT ? OFFSET ?""", args + [per_page, (page - 1) * per_page]) items = [_product_out(r) for r in rows] # personnalisation KA ID : journal + reclassement (tri par défaut # « recent » seulement ; intention de session > long terme) personalized = False user = auth.current_user(request) if user: filters = {k: v for k, v in { "category": category, "region": region, "store": store, "origin": origin, "price_min": price_min, "price_max": price_max, "on_sale": on_sale, "q": q_text}.items() if v not in (None, "", False)} if filters and page == 1: kaid.track(user, "search", query=q_text, filters=filters) if sort == "recent": active = {k for k in ("category", "region", "store", "origin") if filters.get(k)} if price_min is not None or price_max is not None: active.add("price") items, personalized = kaid.rerank( items, user, features_of=_kaid_features, active_dims=active) return {"total": total, "page": page, "per_page": per_page, "items": items, "personalized": personalized} finally: con.close() @app.get("/api/products/{uid}") def product(uid: str, request: Request): con = db.connect() try: rows = q(con, """SELECT p.*, s.name AS store_name, s.region AS store_region, s.city AS store_city, s.origin_class, s.store_kind, s.url AS store_url FROM products p JOIN stores s ON s.id=p.store_id WHERE p.uid=?""", (uid,)) if not rows: raise HTTPException(404) out = _product_out(rows[0]) out["related"] = [_product_out(r) for r in q(con, """ SELECT p.*, s.name AS store_name FROM products p JOIN stores s ON s.id=p.store_id WHERE p.store_id=? AND p.uid<>? AND p.active=1 AND p.listing_status='published' ORDER BY RANDOM() LIMIT 8""", (rows[0]["store_id"], uid))] kaid.track(auth.current_user(request), "detail_view", entity_type="product", entity_id=uid, features=_kaid_features(out)) return out finally: con.close() @app.get("/api/stores") def stores(region: str | None = None, platform: str | None = None, origin: str | None = None, q_text: str | None = Query(None, alias="q"), with_products: bool = False, include_closed: bool = False): con = db.connect() try: # les boutiques « présumées fermées » (échecs consécutifs) sont # dépubliées par défaut, réintégrables (include_closed=1) where, args = (["1=1"] if include_closed else ["COALESCE(presumed_closed,0)=0"]), [] if region: where.append("region=?"); args.append(region) if platform: where.append("platform=?"); args.append(platform) if origin: where.append("origin_class=?"); args.append(origin) if q_text: where.append("(name LIKE ? OR id LIKE ?)"); args += [f"%{q_text}%", f"%{q_text}%"] if with_products: where.append("product_count>0") rows = q(con, f"""SELECT id, name, url, platform, city, region, origin_class, origin_confidence, categories, language, product_count, last_sync, last_status, logo_url, cover_url, description_meta, store_kind, email, phone, lat, lng, shipping_info, certifications, presumed_closed FROM stores WHERE {' AND '.join(where)} ORDER BY product_count DESC, name""", args) for r in rows: r["categories"] = json.loads(r.get("categories") or "[]") r["certifications"] = json.loads(r.get("certifications") or "[]") return {"total": len(rows), "items": rows} finally: con.close() @app.get("/api/stores/{store_id}") def store_detail(store_id: str): con = db.connect() try: rows = q(con, "SELECT * FROM stores WHERE id=?", (store_id,)) if not rows: raise HTTPException(404) s = rows[0] for k in ("categories", "socials", "discovery_sources", "certifications"): s[k] = json.loads(s.get(k) or "[]") # statistiques produits stats = q(con, """SELECT COUNT(*) AS n, MIN(price) AS price_min, MAX(price) AS price_max, AVG(price) AS price_avg FROM products WHERE store_id=? AND active=1 AND listing_status='published' AND price>0""", (store_id,)) s["product_stats"] = stats[0] if stats else {} s["category_breakdown"] = q(con, """SELECT category AS key, COUNT(*) AS n FROM products WHERE store_id=? AND active=1 AND listing_status='published' GROUP BY category ORDER BY n DESC LIMIT 8""", (store_id,)) for c in s["category_breakdown"]: c["label"] = CATEGORIES.get(c["key"], (c["key"], []))[0] # boutiques similaires : même catégorie dominante ou même région, avec produits topcat = s["category_breakdown"][0]["key"] if s["category_breakdown"] else None # Boutiques similaires en 2 temps INDEXÉS (l'ancien LEFT JOIN DISTINCT # prenait plusieurs minutes sur les grosses catégories — constaté le # 2026-08-19 après le passage à 347 k produits publiés) : # 1) boutiques ayant des produits publiés dans la catégorie dominante # (idx_products_cat), 2) complément par région. region = s.get("region") or "" similar, seen_ids = [], {store_id} if topcat: for r in q(con, """ SELECT st.id, st.name, st.region, st.origin_class, st.logo_url, st.product_count FROM stores st JOIN ( SELECT store_id FROM products WHERE category=? AND active=1 AND listing_status='published' GROUP BY store_id) pc ON pc.store_id = st.id WHERE st.id<>? AND st.product_count>0 ORDER BY (st.region=?) DESC, st.product_count DESC LIMIT 8""", (topcat, store_id, region)): if r["id"] not in seen_ids: seen_ids.add(r["id"]) similar.append(r) if len(similar) < 8 and region: for r in q(con, """ SELECT id, name, region, origin_class, logo_url, product_count FROM stores WHERE id<>? AND product_count>0 AND region=? ORDER BY product_count DESC LIMIT 8""", (store_id, region)): if r["id"] not in seen_ids: seen_ids.add(r["id"]) similar.append(r) if len(similar) >= 8: break s["similar"] = similar[:8] return s finally: con.close() @app.get("/api/facets") def facets(): con = db.connect() try: cats = q(con, """SELECT p.category AS key, COUNT(*) AS n FROM products p WHERE p.active=1 AND p.listing_status='published' GROUP BY p.category ORDER BY n DESC""") for c in cats: c["label"] = CATEGORIES.get(c["key"], (c["key"], []))[0] regions = q(con, """SELECT s.region AS key, COUNT(*) AS n FROM products p JOIN stores s ON s.id=p.store_id WHERE p.active=1 AND p.listing_status='published' AND s.region<>'' GROUP BY s.region ORDER BY n DESC""") origins = q(con, """SELECT s.origin_class AS key, COUNT(*) AS n FROM products p JOIN stores s ON s.id=p.store_id WHERE p.active=1 AND p.listing_status='published' GROUP BY s.origin_class ORDER BY n DESC""") top_stores = q(con, """SELECT id AS key, name, product_count AS n FROM stores WHERE product_count>0 ORDER BY n DESC LIMIT 40""") return {"categories": cats, "regions": regions, "origins": origins, "stores": top_stores, "all_regions": REGIONS, # taxonomie complète (formulaire « Inscrire mon atelier ») "all_categories": [{"key": k, "label": v[0]} for k, v in CATEGORIES.items()]} finally: con.close() _stats_cache: dict = {"ts": 0.0, "data": None} @app.get("/api/stats/extended") def stats_extended(): """Statistiques riches pour la page /stats : prix, catégories, croissance. Coûteux (médianes) → cache mémoire 10 minutes.""" import time as _time if _stats_cache["data"] is not None and _time.time() - _stats_cache["ts"] < 600: return _stats_cache["data"] con = db.connect() try: def median_price(where_sql, args): n = con.execute(f"SELECT COUNT(*) FROM products WHERE {where_sql} AND price>0", args).fetchone()[0] if not n: return None row = con.execute(f"""SELECT price FROM products WHERE {where_sql} AND price>0 ORDER BY price LIMIT 1 OFFSET ?""", args + [n // 2]).fetchone() return row[0] if row else None by_category = [] for r in q(con, """SELECT category AS key, COUNT(*) AS products, COUNT(DISTINCT store_id) AS stores, MIN(CASE WHEN price>0 THEN price END) AS price_min, ROUND(AVG(CASE WHEN price>0 AND price<=500000 THEN price END),2) AS price_avg, MAX(CASE WHEN price<=500000 THEN price END) AS price_max FROM products WHERE active=1 AND listing_status='published' GROUP BY category ORDER BY products DESC"""): r["label"] = CATEGORIES.get(r["key"], (r["key"], []))[0] r["price_median"] = median_price("active=1 AND listing_status='published' AND category=?", [r["key"]]) by_category.append(r) global_median = median_price("active=1 AND listing_status='published'", []) price_buckets = q(con, """SELECT CASE WHEN price < 10 THEN '0-10' WHEN price < 25 THEN '10-25' WHEN price < 50 THEN '25-50' WHEN price < 100 THEN '50-100' WHEN price < 250 THEN '100-250' WHEN price < 1000 THEN '250-1000' ELSE '1000+' END AS bucket, COUNT(*) AS n FROM products WHERE active=1 AND listing_status='published' AND price>0 GROUP BY bucket""") order = ['0-10', '10-25', '25-50', '50-100', '100-250', '250-1000', '1000+'] price_buckets.sort(key=lambda b: order.index(b["bucket"])) by_origin = q(con, """SELECT s.origin_class AS key, COUNT(DISTINCT s.id) AS stores, COUNT(p.uid) AS products FROM stores s LEFT JOIN products p ON p.store_id=s.id AND p.active=1 AND p.listing_status='published' GROUP BY s.origin_class ORDER BY products DESC""") by_platform = q(con, """SELECT platform AS key, COUNT(*) AS stores, SUM(product_count) AS products FROM stores WHERE platform<>'' GROUP BY platform ORDER BY products DESC""") by_region = q(con, """SELECT s.region AS key, COUNT(DISTINCT s.id) AS stores, COUNT(p.uid) AS products, ROUND(AVG(CASE WHEN p.price>0 AND p.price<=500000 THEN p.price END),2) AS price_avg FROM stores s LEFT JOIN products p ON p.store_id=s.id AND p.active=1 AND p.listing_status='published' WHERE s.region<>'' GROUP BY s.region ORDER BY products DESC""") growth = q(con, """SELECT date(first_seen,'unixepoch') AS day, COUNT(*) AS n FROM products WHERE active=1 AND listing_status='published' GROUP BY day ORDER BY day DESC LIMIT 30""") top_stores = q(con, """SELECT id, name, region, origin_class, logo_url, product_count AS products FROM stores WHERE product_count>0 ORDER BY product_count DESC LIMIT 15""") most_expensive = q(con, """SELECT p.uid, p.title, p.price, s.name AS store_name FROM products p JOIN stores s ON s.id=p.store_id WHERE p.active=1 AND p.listing_status='published' AND p.price>0 ORDER BY p.price DESC LIMIT 8""") newest = q(con, """SELECT p.uid, p.title, p.price, p.category, s.name AS store_name FROM products p JOIN stores s ON s.id=p.store_id WHERE p.active=1 AND p.listing_status='published' ORDER BY p.first_seen DESC LIMIT 12""") for r in newest: r["category_label"] = CATEGORIES.get(r["category"], (r["category"], []))[0] availability = q(con, """SELECT CASE WHEN available=1 THEN 'en stock' WHEN available=0 THEN 'rupture' ELSE 'inconnu' END AS key, COUNT(*) AS n FROM products WHERE active=1 AND listing_status='published' GROUP BY key""") coverage = q(con, """SELECT (SELECT COUNT(*) FROM products WHERE active=1 AND listing_status='published' AND images<>'[]' AND images IS NOT NULL) AS with_image, (SELECT COUNT(*) FROM products WHERE active=1 AND listing_status='published' AND description<>'') AS with_desc, (SELECT COUNT(*) FROM stores WHERE logo_url IS NOT NULL AND logo_url<>'') AS with_logo, (SELECT COUNT(*) FROM stores WHERE region<>'') AS with_region """)[0] totals = q(con, """SELECT (SELECT COUNT(*) FROM products WHERE active=1 AND listing_status='published') AS products, (SELECT COUNT(*) FROM products WHERE active=1 AND listing_status='published' AND price>0) AS products_priced, (SELECT COUNT(*) FROM stores WHERE product_count>0) AS stores_live, (SELECT COUNT(*) FROM stores) AS stores_registry, (SELECT COUNT(DISTINCT region) FROM stores WHERE region<>'') AS regions, (SELECT ROUND(AVG(price),2) FROM products WHERE active=1 AND listing_status='published' AND price>0 AND price<=500000) AS price_avg """)[0] totals["price_median"] = global_median data = {"totals": totals, "by_category": by_category, "price_buckets": price_buckets, "by_origin": by_origin, "by_platform": by_platform, "by_region": by_region, "growth": growth, "top_stores": top_stores, "most_expensive": most_expensive, "newest": newest, "availability": availability, "coverage": coverage} _stats_cache.update(ts=_time.time(), data=data) return data finally: con.close() @app.get("/api/report.pdf") def report_pdf(): """Rapport de marché Fabri-Ka en PDF (logo, KPIs, prix, catégories, régions, origine, plateformes, top boutiques).""" from fastapi.responses import Response from . import report stats = stats_extended() pdf = report.build_report(stats) fname = f"rapport-marche-fabri-ka-{date.today()}.pdf" return Response(content=pdf, media_type="application/pdf", headers={"Content-Disposition": f'attachment; filename="{fname}"'}) # ---- module Stats commun Groupe KA (contrat ka-stats SPEC.md) --------------- @app.get("/api/stats/dashboard") def stats_dashboard(period: str = "30j", from_: str | None = Query(None, alias="from"), to: str | None = None): """Tableau de bord analytique (KPI, séries, répartitions, geo, heatmap, tableaux, records) — données réelles, cache 5 min par période.""" from . import statsdash try: return statsdash.dashboard(period, from_, to) except ValueError: raise HTTPException(400, "Dates invalides (format attendu YYYY-MM-DD)") @app.get("/api/stats/report") def stats_report(period: str = "30j", from_: str | None = Query(None, alias="from"), to: str | None = None, mode: str = "complet"): """Rapport statistique PDF estampillé Groupe-KA (gabarit kapdf/fpdf2).""" from fastapi.responses import Response from . import kapdf, statsdash try: dash = statsdash.dashboard(period, from_, to) except ValueError: raise HTTPException(400, "Dates invalides (format attendu YYYY-MM-DD)") site = {"wordmark": "Fabri·Ka", "accent": "#c4532e", "domain": "www.fabri-ka.com", "tagline": "Tous les produits québécois. Un seul endroit."} if mode not in kapdf.REPORT_MODES: mode = "complet" pdf = kapdf.GroupeKAReport( site=site, dashboard=dash, mode=mode).build() fname = kapdf.filename("fabri-ka", period, mode) return Response(content=pdf, media_type="application/pdf", headers={"Content-Disposition": f'attachment; filename="{fname}"'}) @app.get("/api/stats/catalog") def stats_catalog(period: str = "30j", from_: str | None = Query(None, alias="from"), to: str | None = None): """v3 — blocs composables pour le constructeur de rapports personnalisés.""" from . import kapdf, statsdash try: dash = statsdash.dashboard(period, from_, to) except ValueError: raise HTTPException(400, "Dates invalides (format attendu YYYY-MM-DD)") return {"updated": dash.get("updated"), "period": dash.get("period"), "blocks": kapdf.catalog(dash)} @app.post("/api/stats/report/custom") def stats_report_custom(spec: dict = Body(...)): """v3 — rapport PDF personnalisé : {"title", "period", "from", "to", "blocks": [{"key": "series:…", "render": "bar"}, …]} (SPEC.md §3bis).""" from fastapi.responses import Response from . import kapdf, statsdash period = str(spec.get("period") or "30j") try: dash = statsdash.dashboard(period, str(spec.get("from") or "") or None, str(spec.get("to") or "") or None) except ValueError: raise HTTPException(400, "Dates invalides (format attendu YYYY-MM-DD)") site = {"wordmark": "Fabri·Ka", "accent": "#c4532e", "domain": "www.fabri-ka.com", "tagline": "Tous les produits québécois. Un seul endroit."} known = {b["key"] for b in kapdf.catalog(dash)} blocks = [b for b in (spec.get("blocks") or []) if isinstance(b, dict) and b.get("key") in known][:40] if not blocks: raise HTTPException(400, "Aucun bloc valide dans la composition") pdf = kapdf.GroupeKAReport( site=site, dashboard=dash, mode=kapdf.CUSTOM_MODE, spec={"title": str(spec.get("title") or "")[:80], "blocks": blocks}, ).build() fname = kapdf.filename("fabri-ka", period, kapdf.CUSTOM_MODE) return Response(content=pdf, media_type="application/pdf", headers={"Content-Disposition": f'attachment; filename="{fname}"'}) @app.get("/api/stats") def stats(syncs_since_h: float | None = None): """Stats globales. `syncs_since_h` (heures) : fenêtre du sync_log exposé — sans le paramètre, comportement historique (30 dernières entrées) ; avec, toutes les entrées depuis now-syncs_since_h (LIMIT 4000), et au moins les 20 dernières si la fenêtre en contient moins (supervision multi-boutiques : fabri-ka fait tourner ~1 700 boutiques par cycle, 30 entrées ne couvrent qu'une poignée de magasins).""" con = db.connect() try: totals = q(con, """SELECT (SELECT COUNT(*) FROM products WHERE active=1 AND listing_status='published') AS products, (SELECT COUNT(*) FROM stores WHERE product_count>0) AS stores_live, (SELECT COUNT(*) FROM stores) AS stores_registry, (SELECT COUNT(DISTINCT region) FROM stores WHERE region<>'' AND product_count>0) AS regions """)[0] by_region = q(con, """SELECT region, COUNT(*) AS stores, SUM(product_count) AS products FROM stores WHERE region<>'' GROUP BY region ORDER BY products DESC""") # Les stores enabled=0 (faux positifs de découverte, boutiques fermées…) # sont exclus du flux : leurs derniers syncs à 0 restent sinon figés dans # la fenêtre et la supervision api-ka les compte comme « broken » à vie. active = "store_id NOT IN (SELECT id FROM stores WHERE enabled = 0)" if syncs_since_h is not None: cutoff = time.time() - syncs_since_h * 3600 recent = q(con, f"""SELECT ts, store_id, found, added, updated, removed, status FROM sync_log WHERE ts >= ? AND {active} ORDER BY ts DESC LIMIT 4000""", (cutoff,)) if len(recent) < 20: recent = q(con, f"""SELECT ts, store_id, found, added, updated, removed, status FROM sync_log WHERE {active} ORDER BY ts DESC LIMIT 20""") else: recent = q(con, f"""SELECT ts, store_id, found, added, updated, removed, status FROM sync_log WHERE {active} ORDER BY ts DESC LIMIT 30""") # Stores volontairement désactivés (faux positifs de découverte, boutiques # fermées…) mais présents dans le sync_log : sans ce signal, la supervision # api-ka les fige en « broken » car ils n'émettent plus jamais de syncs. disabled = q(con, """SELECT DISTINCT st.id FROM stores st JOIN sync_log sl ON sl.store_id = st.id WHERE st.enabled = 0 ORDER BY st.id""") return {"totals": totals, "by_region": by_region, "sync_log": recent, "disabled_stores": [r["id"] for r in disabled]} finally: con.close() _sync_running = threading.Lock() @app.post("/api/sync") def trigger_sync(background: BackgroundTasks): if _sync_running.locked(): return {"status": "already-running"} def _job(): with _sync_running: from . import ingest ingest.run() background.add_task(_job) return {"status": "started"} # ---- SEO : robots + sitemaps ------------------------------------------------- @app.get("/robots.txt") def robots(): return seo.robots_txt() @app.get("/sitemap.xml") def sitemap_index(): return seo.sitemap_index() @app.get("/sitemap-{name}.xml") def sitemap(name: str): return seo.sitemap(name) # ---- frontend --------------------------------------------------------------- class HashedStaticFiles(StaticFiles): """Assets Vite hashés -> cache navigateur agressif.""" def file_response(self, *args, **kwargs): resp = super().file_response(*args, **kwargs) resp.headers["Cache-Control"] = "public, max-age=31536000, immutable" return resp if FRONT_DIST.exists(): app.mount("/assets", HashedStaticFiles(directory=FRONT_DIST / "assets"), name="assets") @app.get("/doc", include_in_schema=False) def doc_redirect(): return RedirectResponse("/doc/", status_code=301) # /doc : documentation statique (index.html + captures + PDF) — montée # explicitement pour que le catch-all SPA ne l'intercepte pas. if (FRONT_DIST / "doc").is_dir(): app.mount("/doc", StaticFiles(directory=FRONT_DIST / "doc", html=True), name="doc") @app.get("/{path:path}") def spa(path: str, request: Request): target = FRONT_DIST / path if path and target.is_file(): resp = FileResponse(target) resp.headers["Cache-Control"] = "public, max-age=86400" return resp # Rendu serveur SEO : HTML complet (meta + contenu) par type de page, # avec les bons statuts (200 / 301 slug canonique / 404 / 410). return seo.render_route(path, request.query_params)