spb/lou-ka

1# -----------------------------------------------------------------------------2# Lou-Ka — Location court terme3# ingest.py : pipeline d'ingestion court terme — exécute les connecteurs4#             court terme et synchronise la base séparée (louka_ct.db).5# -----------------------------------------------------------------------------6from __future__ import annotations78import sys9import time10import traceback1112from . import db13from .connectors import ST_CONNECTORS1415BREAKER_FAILS = 516BREAKER_RETRY_EVERY = 5171819def _consecutive_failures(con, source: str) -> int:20    rows = con.execute(21        "SELECT ok FROM st_sync_log WHERE source=? ORDER BY ts DESC LIMIT 60",22        (source,)).fetchall()23    n = 024    for r in rows:25        if r["ok"]:26            break27        n += 128    return n293031def run(sources: list[str] | None = None) -> list[dict]:32    """Exécute l'ingestion court terme (toutes les sources ou celles demandées)."""33    con = db.connect()34    results = []35    explicit = sources is not None36    targets = sources or list(ST_CONNECTORS.keys())37    for sid in targets:38        cls = ST_CONNECTORS.get(sid)39        if cls is None:40            print(f"[lou-ka/ct] connecteur inconnu : {sid}", file=sys.stderr)41            continue42        if not explicit:43            fails = _consecutive_failures(con, sid)44            if fails >= BREAKER_FAILS and fails % BREAKER_RETRY_EVERY != 0:45                print(f"[lou-ka/ct] {sid} sauté (disjoncteur : {fails} échecs)")46                db.log_failure(con, sid, f"disjoncteur: sauté ({fails} échecs)")47                results.append({"source": sid, "skipped": True, "fails": fails})48                continue49        t0 = time.time()50        print(f"[lou-ka/ct] sync {sid} ...")51        try:52            listings = cls().fetch()53            finalized, dropped = [], 054            for lst in listings:55                try:56                    if not (lst.url and lst.title):57                        dropped += 158                        continue59                    finalized.append(lst.finalize())60                except Exception:61                    dropped += 162            stats = db.sync_source(con, sid, finalized)63            stats["seconds"] = round(time.time() - t0, 1)64            if dropped:65                stats["dropped"] = dropped66            print(f"[lou-ka/ct]   {stats}")67            results.append(stats)68        except Exception as exc:69            db.log_failure(con, sid, f"{exc}")70            traceback.print_exc()71            results.append({"source": sid, "error": str(exc)})72    con.close()73    return results747576def watch(interval_seconds: int = 6 * 3600) -> None:77    """Boucle de rafraîchissement périodique (défaut : 6 h — l'inventaire78    court terme bouge moins vite que les logements au mois)."""79    while True:80        run()81        print(f"[lou-ka/ct] prochaine synchronisation dans {interval_seconds}s")82        time.sleep(interval_seconds)838485if __name__ == "__main__":86    run(sys.argv[1:] or None)87