spb/fabri-ka

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# Nombre d'échecs de sync CONSÉCUTIFS avant de présumer l'atelier fermé23# (12 × 6 h = ~3 jours sans jamais répondre). Statut réversible : un seul24# sync réussi le remet à zéro et réactive la boutique.25CLOSE_AFTER_FAILS = 122627_db_lock = threading.Lock()282930def load_registry() -> list[dict]:31    with open(REGISTRY_PATH) as f:32        return json.load(f)["stores"]333435def sync_store(store: dict) -> dict:36    """Synchronise une boutique ; retourne un résumé (jamais d'exception)."""37    sid = store["id"]38    conn = connector_for(store)39    if conn is None:40        return {"store": sid, "status": "no-connector"}41    try:42        products = conn.fetch()43    except Exception as exc:44        with _db_lock:45            con = db.connect()46            db.log_sync(con, sid, 0, 0, 0, 0, "error", str(exc))47            # compteur d'échecs consécutifs -> « présumée fermée »48            con.execute("UPDATE stores SET fail_streak=COALESCE(fail_streak,0)+1 "49                        "WHERE id=?", (sid,))50            row = con.execute("SELECT fail_streak, COALESCE(presumed_closed,0) "51                              "FROM stores WHERE id=?", (sid,)).fetchone()52            if row and row[0] >= CLOSE_AFTER_FAILS and not row[1]:53                con.execute("UPDATE stores SET presumed_closed=1, product_count=0 "54                            "WHERE id=?", (sid,))55                con.execute("UPDATE products SET active=0 WHERE store_id=?", (sid,))56                db.log_sync(con, sid, 0, 0, 0, 0, "presumed_closed",57                            f"{row[0]} échecs consécutifs — produits dépubliés "58                            "(réintégration automatique au premier sync réussi)")59                print(f"  ⚠ {sid}: présumée fermée ({row[0]} échecs consécutifs)",60                      flush=True)61            con.commit(); con.close()62        return {"store": sid, "status": "error", "message": str(exc)[:200]}63    with _db_lock:64        con = db.connect()65        db.upsert_store(con, store)66        added, updated, removed = db.sync_products(con, sid, store.get("name") or sid, products)67        db.log_sync(con, sid, len(products), added, updated, removed, "ok")68        # sync réussi : la boutique répond -> on efface l'ardoise69        con.execute("UPDATE stores SET fail_streak=0, presumed_closed=0 WHERE id=?",70                    (sid,))71        con.commit(); con.close()72    return {"store": sid, "status": "ok", "found": len(products),73            "added": added, "updated": updated, "removed": removed}747576def run(only: list[str] | None = None, workers: int = 8) -> None:77    stores = load_registry()78    if only:79        wanted = {o.lower() for o in only}80        stores = [s for s in stores if s["id"].lower() in wanted]81    # le répertoire /boutiques affiche TOUTES les boutiques du registre :82    # on upsert chaque fiche, même non connectable, avant la sync produits83    con = db.connect()84    for s in stores:85        db.upsert_store(con, s)86    con.commit(); con.close()87    stores = [s for s in stores if s.get("enabled", True) and s.get("platform")88              and s.get("catalog_endpoint")]89    print(f"[fabri-ka] sync de {len(stores)} boutiques…", flush=True)90    t0 = time.time()91    ok = err = total = 092    with cf.ThreadPoolExecutor(workers) as ex:93        for res in ex.map(sync_store, stores):94            status = res.get("status")95            if status == "ok":96                ok += 197                total += res.get("found", 0)98                print(f"  ✓ {res['store']}: {res['found']} produits "99                      f"(+{res['added']} ~{res['updated']} -{res['removed']})", flush=True)100            elif status == "error":101                err += 1102                print(f"  ✗ {res['store']}: {res.get('message','')}", flush=True)103    print(f"[fabri-ka] terminé en {time.time()-t0:.0f}s — {ok} ok, {err} erreurs, "104          f"{total} produits vus", flush=True)105106107def watch(interval_seconds: int) -> None:108    while True:109        try:110            run()111        except Exception as exc:   # noqa: BLE001 — le watcher ne meurt jamais112            print(f"[fabri-ka] watch: {exc}", file=sys.stderr, flush=True)113        time.sleep(interval_seconds)114