spb/fabri-ka Public

1# -----------------------------------------------------------------------------2# Fabri-Ka — Agrégateur de produits québécois3# Auteur : Simon-Pierre Boucher — contact@spboucher.ai4# ingest.py : orchestration des synchronisations — registre -> connecteurs ->5#             diff BD. Parallélisé par boutique (thread pool), tolérant aux6#             pannes (une boutique qui casse n'affecte jamais les autres).7# -----------------------------------------------------------------------------8from __future__ import annotations910import concurrent.futures as cf11import json12import sys13import threading14import time15from pathlib import Path1617from . import db18from .connectors import connector_for1920REGISTRY_PATH = Path(__file__).resolve().parent.parent / "data" / "stores.json"2122_db_lock = threading.Lock()232425def load_registry() -> list[dict]:26    with open(REGISTRY_PATH) as f:27        return json.load(f)["stores"]282930def sync_store(store: dict) -> dict:31    """Synchronise une boutique ; retourne un résumé (jamais d'exception)."""32    sid = store["id"]33    conn = connector_for(store)34    if conn is None:35        return {"store": sid, "status": "no-connector"}36    try:37        products = conn.fetch()38    except Exception as exc:39        with _db_lock:40            con = db.connect()41            db.log_sync(con, sid, 0, 0, 0, 0, "error", str(exc))42            con.commit(); con.close()43        return {"store": sid, "status": "error", "message": str(exc)[:200]}44    with _db_lock:45        con = db.connect()46        db.upsert_store(con, store)47        added, updated, removed = db.sync_products(con, sid, store.get("name") or sid, products)48        db.log_sync(con, sid, len(products), added, updated, removed, "ok")49        con.commit(); con.close()50    return {"store": sid, "status": "ok", "found": len(products),51            "added": added, "updated": updated, "removed": removed}525354def run(only: list[str] | None = None, workers: int = 8) -> None:55    stores = load_registry()56    if only:57        wanted = {o.lower() for o in only}58        stores = [s for s in stores if s["id"].lower() in wanted]59    # le répertoire /boutiques affiche TOUTES les boutiques du registre :60    # on upsert chaque fiche, même non connectable, avant la sync produits61    con = db.connect()62    for s in stores:63        db.upsert_store(con, s)64    con.commit(); con.close()65    stores = [s for s in stores if s.get("enabled", True) and s.get("platform")66              and s.get("catalog_endpoint")]67    print(f"[fabri-ka] sync de {len(stores)} boutiques…", flush=True)68    t0 = time.time()69    ok = err = total = 070    with cf.ThreadPoolExecutor(workers) as ex:71        for res in ex.map(sync_store, stores):72            status = res.get("status")73            if status == "ok":74                ok += 175                total += res.get("found", 0)76                print(f"  ✓ {res['store']}: {res['found']} produits "77                      f"(+{res['added']} ~{res['updated']} -{res['removed']})", flush=True)78            elif status == "error":79                err += 180                print(f"  ✗ {res['store']}: {res.get('message','')}", flush=True)81    print(f"[fabri-ka] terminé en {time.time()-t0:.0f}s — {ok} ok, {err} erreurs, "82          f"{total} produits vus", flush=True)838485def watch(interval_seconds: int) -> None:86    while True:87        try:88            run()89        except Exception as exc:   # noqa: BLE001 — le watcher ne meurt jamais90            print(f"[fabri-ka] watch: {exc}", file=sys.stderr, flush=True)91        time.sleep(interval_seconds)92