spb/resto-ka

1# ==============================================================================2# Author: Simon-Pierre Boucher <contact@spboucher.ai>3# File:   restoka/ingest.py4# Desc:   Pipeline d'ingestion — exécute les connecteurs et synchronise la base5#         (ajouts / mises à jour / retraits avec délai de grâce). Une fiche6#         malformée (menu sans price_context…) est rejetée, jamais stockée.7#         Calqué sur louka/ingest.py.8# ==============================================================================9from __future__ import annotations1011import sys12import time13import traceback1415from . import db16from .connectors import CONNECTORS171819def run(sources: list[str] | None = None) -> list[dict]:20    """Exécute l'ingestion pour toutes les sources (ou celles demandées)."""21    con = db.connect()22    results = []23    targets = sources or list(CONNECTORS.keys())24    for sid in targets:25        cls = CONNECTORS.get(sid)26        if cls is None:27            print(f"[resto-ka] connecteur inconnu : {sid}", file=sys.stderr)28            continue29        t0 = time.time()30        print(f"[resto-ka] sync {sid} ...")31        try:32            restos = cls().fetch()33            finalized, dropped = [], 034            for r in restos:35                try:36                    finalized.append(r.finalize())37                except Exception as exc:  # une fiche malformée ne bloque pas la source38                    dropped += 139                    print(f"[resto-ka]   fiche rejetée : {exc}", file=sys.stderr)40            stats = db.sync_source(con, sid, finalized)41            stats["seconds"] = round(time.time() - t0, 1)42            if dropped:43                stats["dropped"] = dropped44            if stats.get("alert"):45                print(f"[resto-ka]   ⚠ ALERTE {sid} : {stats['alert']}")46            print(f"[resto-ka]   {stats}")47            results.append(stats)48        except Exception as exc:  # robustesse : une source ne bloque pas les autres49            db.log_failure(con, sid, f"{exc}")50            traceback.print_exc()51            results.append({"source": sid, "error": str(exc)})52    try:53        con.execute("PRAGMA optimize")54    except Exception:55        pass56    con.close()57    return results585960def enrich() -> None:61    """Étapes d'enrichissement post-sync (géocodage, région, déduplication)."""62    try:  # géocoder les nouvelles adresses (lat/lng obligatoires §5)63        from . import geocode64        geocode.run_batch()65    except Exception as exc:66        print(f"[resto-ka] geocode: erreur non bloquante: {exc}", file=sys.stderr)67    try:  # déduplication inter-sources (APRÈS géocodage : coords requises)68        from . import dedup69        stats = dedup.run()70        print(f"[resto-ka] dedup: {stats}")71    except Exception as exc:72        print(f"[resto-ka] dedup: erreur non bloquante: {exc}", file=sys.stderr)737475def watch(interval_seconds: int = 7 * 86400 // 7) -> None:76    """Boucle de rafraîchissement périodique (défaut : quotidien — les grandes77    plateformes se rafraîchissent au moins chaque semaine, CLAUDE.md §14)."""78    while True:79        run()80        enrich()81        print(f"[resto-ka] prochaine synchronisation dans {interval_seconds}s")82        time.sleep(interval_seconds)838485if __name__ == "__main__":86    run(sys.argv[1:] or None)87    enrich()88