# ----------------------------------------------------------------------------- # Food-Ka — Agrégateur de produits d'épicerie (province de Québec) # Auteur : Simon-Pierre Boucher — contact@spboucher.ai # web.py : API FastAPI (JSON) + service du frontend React (frontend/dist) # ----------------------------------------------------------------------------- from __future__ import annotations import json import threading from pathlib import Path from fastapi import BackgroundTasks, FastAPI, HTTPException, Query from fastapi.middleware.cors import CORSMiddleware from fastapi.responses import FileResponse, Response from fastapi.staticfiles import StaticFiles from . import db, ingest ROOT = Path(__file__).resolve().parent.parent SOURCES_PATH = ROOT / "data" / "sources.json" FRONTEND_DIST = ROOT / "frontend" / "dist" app = FastAPI(title="Food-Ka API", version="1.0", description="Agrégateur de produits d'épicerie — province de Québec") app.add_middleware(CORSMiddleware, allow_origins=["*"], allow_methods=["*"], allow_headers=["*"]) _sync_lock = threading.Lock() # tris supportés -> clause SQL (liste blanche, jamais d'injection) _SORTS = { "price_asc": "price IS NULL, price ASC", "price_desc": "price IS NULL, price DESC", "unit_price": "unit_price IS NULL, unit_price ASC", "discount": "(CASE WHEN regular_price IS NOT NULL AND price IS NOT NULL" " THEN (regular_price - price) / regular_price ELSE 0 END) DESC", "name": "name COLLATE NOCASE ASC", "recent": "first_seen DESC", } def _row_to_dict(row) -> dict: d = dict(row) d["keywords"] = json.loads(d.get("keywords") or "[]") d["images"] = json.loads(d.get("images") or "[]") d["details"] = json.loads(d.get("details") or "{}") d["on_sale"] = bool(d.get("on_sale")) if d.get("in_stock") is not None: d["in_stock"] = bool(d["in_stock"]) return d @app.get("/api/products") def list_products( category: str | None = None, source: str | None = None, brand: str | None = None, price_max: float | None = None, price_min: float | None = None, on_sale: int | None = None, # 1 = en solde seulement in_stock: int | None = None, # 1 / 0 q: str | None = None, sort: str = "price_asc", active: int = 1, limit: int = Query(60, le=500), offset: int = 0, ): con = db.connect() sql = "SELECT * FROM products WHERE 1=1" args: list = [] if active in (0, 1): sql += " AND active=?"; args.append(active) if category: sql += " AND category=?"; args.append(category) if source: sql += " AND source=?"; args.append(source) if brand: sql += " AND brand LIKE ?"; args.append(f"%{brand}%") if price_max is not None: sql += " AND price IS NOT NULL AND price<=?"; args.append(price_max) if price_min is not None: sql += " AND price IS NOT NULL AND price>=?"; args.append(price_min) if on_sale == 1: sql += " AND on_sale=1" if in_stock in (0, 1): sql += " AND in_stock=?"; args.append(in_stock) if q: sql += " AND (name LIKE ? OR brand LIKE ? OR category_raw LIKE ?)" args += [f"%{q}%"] * 3 total = con.execute(f"SELECT COUNT(*) c FROM ({sql})", args).fetchone()["c"] sql += f" ORDER BY {_SORTS.get(sort, _SORTS['price_asc'])} LIMIT ? OFFSET ?" args += [limit, offset] rows = [_row_to_dict(r) for r in con.execute(sql, args).fetchall()] con.close() return {"total": total, "count": len(rows), "products": rows} @app.get("/api/products/{uid}") def get_product(uid: str): con = db.connect() row = con.execute("SELECT * FROM products WHERE uid=?", (uid,)).fetchone() d = None if row is not None: d = _row_to_dict(row) # historique de prix (suivi des soldes) d["price_history"] = [dict(r) for r in con.execute( "SELECT ts, price FROM price_log WHERE uid=? ORDER BY ts DESC LIMIT 12", (uid,)).fetchall()] # le même produit ailleurs : comparaison inter-bannières par nom if d["name"]: d["compare"] = [_row_to_dict(r) for r in con.execute( """SELECT * FROM products WHERE active=1 AND uid<>? AND name LIKE ? ORDER BY price IS NULL, price ASC LIMIT 8""", (uid, f"%{d['name'].split('(')[0].strip()[:40]}%")).fetchall()] else: d["compare"] = [] con.close() if d is None: raise HTTPException(404, "Produit introuvable") return d @app.get("/api/facets") def facets(category: str | None = None): """Valeurs distinctes pour construire les filtres du frontend.""" con = db.connect() brand_sql = ("SELECT brand, COUNT(*) n FROM products" " WHERE active=1 AND brand<>''") brand_args: list = [] if category: brand_sql += " AND category=?" brand_args.append(category) out = { "categories": [dict(r) for r in con.execute( "SELECT category, COUNT(*) n FROM products WHERE active=1 AND category<>''" " GROUP BY category ORDER BY n DESC")], "brands": [dict(r) for r in con.execute( brand_sql + " GROUP BY brand ORDER BY n DESC LIMIT 60", brand_args)], "sources": [dict(r) for r in con.execute( "SELECT source, COUNT(*) n FROM products WHERE active=1" " GROUP BY source ORDER BY n DESC")], "on_sale": con.execute( "SELECT COUNT(*) c FROM products WHERE active=1 AND on_sale=1" ).fetchone()["c"], } con.close() return out @app.get("/api/sources") def sources(): registry = json.loads(SOURCES_PATH.read_text(encoding="utf-8"))["sources"] con = db.connect() counts = {r["source"]: r["n"] for r in con.execute( "SELECT source, COUNT(*) n FROM products WHERE active=1 GROUP BY source")} last = {r["source"]: r["ts"] for r in con.execute( "SELECT source, MAX(ts) ts FROM sync_log WHERE ok=1 GROUP BY source")} con.close() for s in registry: s["active_products"] = counts.get(s["id"], 0) s["last_sync"] = last.get(s["id"]) return {"sources": registry} @app.get("/api/stats") def stats(): con = db.connect() row = con.execute( """SELECT COUNT(*) total, SUM(on_sale) on_sale, COUNT(DISTINCT source) sources, COUNT(DISTINCT category) categories, AVG(price) avg_price FROM products WHERE active=1""").fetchone() by_source = [dict(r) for r in con.execute( """SELECT source, COUNT(*) n, SUM(on_sale) sales, AVG(price) avg_price FROM products WHERE active=1 GROUP BY source ORDER BY n DESC""")] by_category = [dict(r) for r in con.execute( """SELECT category, COUNT(*) n, AVG(price) avg_price FROM products WHERE active=1 AND category<>'' GROUP BY category ORDER BY n DESC""")] # meilleures aubaines du moment (rabais relatif le plus fort) deals = [_row_to_dict(r) for r in con.execute( """SELECT * FROM products WHERE active=1 AND regular_price IS NOT NULL AND price IS NOT NULL ORDER BY (regular_price - price) / regular_price DESC LIMIT 24""")] log = [dict(r) for r in con.execute( "SELECT * FROM sync_log ORDER BY ts DESC LIMIT 20")] con.close() return {**dict(row), "by_source": by_source, "by_category": by_category, "deals": deals, "recent_syncs": log} @app.get("/api/stats/detailed") def stats_detailed(): """Agrégats du marché (source unique : foodka/marketstats.py).""" from . import marketstats return marketstats.compute() @app.get("/api/stats/rapport.pdf") def rapport_pdf(): """Rapport PDF du marché — mêmes chiffres que la page Statistiques.""" from . import pdfgen return Response( content=pdfgen.rapport_pdf(), media_type="application/pdf", headers={"Content-Disposition": 'attachment; filename="foodka-rapport-marche.pdf"'}) @app.post("/api/sync") def trigger_sync(background: BackgroundTasks, source: str | None = None): """Déclenche une synchronisation (équivalent d'un webhook entrant).""" def _job(): with _sync_lock: ingest.run([source] if source else None) background.add_task(_job) return {"status": "démarré", "source": source or "toutes"} # --- Frontend React (build Vite) -------------------------------------------- if FRONTEND_DIST.exists(): app.mount("/assets", StaticFiles(directory=FRONTEND_DIST / "assets"), name="assets") @app.get("/{full_path:path}") def spa(full_path: str): target = FRONTEND_DIST / full_path if full_path and target.is_file(): return FileResponse(target) return FileResponse(FRONTEND_DIST / "index.html")