spb/house-ka

1# -----------------------------------------------------------------------------2# Immo-Ka — Agrégateur de maisons à vendre (province de Québec)3# Auteur : Simon-Pierre Boucher — contact@spboucher.ai4# ingest.py : pipeline d'ingestion — exécute les connecteurs et synchronise5#             la base (ajouts / mises à jour / retraits) = contenu toujours à jour6# -----------------------------------------------------------------------------7from __future__ import annotations89import sys10import time11import traceback1213from . import db14from .connectors import CONNECTORS151617def _stale_first(con, targets: list[str]) -> list[str]:18    """Ordonne les sources par dernier succès croissant (jamais synchronisées19    d'abord). Une passe complète dure ~24 h : dans l'ordre fixe du registre,20    un restart mi-cycle repart de zéro et affame la fin de la liste (hanlon,21    denisedunn, raymondanthony marquées stale le 2026-08-29). Ici, les plus22    en retard passent en tête de chaque cycle."""23    last = {r[0]: r[1] for r in con.execute(24        "SELECT source, MAX(ts) FROM sync_log WHERE ok=1 GROUP BY source")}25    return sorted(targets, key=lambda sid: last.get(sid) or 0.0)262728def run(sources: list[str] | None = None) -> list[dict]:29    """Exécute l'ingestion pour toutes les sources (ou celles demandées)."""30    con = db.connect()31    results = []32    targets = sources or _stale_first(con, list(CONNECTORS.keys()))33    for sid in targets:34        cls = CONNECTORS.get(sid)35        if cls is None:36            print(f"[immo-ka] connecteur inconnu : {sid}", file=sys.stderr)37            continue38        t0 = time.time()39        print(f"[immo-ka] sync {sid} ...")40        try:41            listings = cls().fetch()42            finalized, dropped = [], 043            for lst in listings:44                try:45                    finalized.append(lst.finalize())46                except Exception:  # une annonce malformée ne bloque pas la source47                    dropped += 148            try:49                # garde géographique : coordonnées fournies par la source mais50                # incompatibles avec la ville annoncée -> annulées (le géocodeur51                # rigoureux prendra le relais). Cache seulement, aucun réseau.52                from . import geocode53                bad = geocode.strip_bad_source_coords(con, finalized)54                if bad:55                    print(f"[immo-ka]   {bad} coordonnée(s) source incohérente(s) rejetée(s)")56            except Exception:57                traceback.print_exc()58            stats = db.sync_source(con, sid, finalized)59            stats["seconds"] = round(time.time() - t0, 1)60            if dropped:61                stats["dropped"] = dropped62            if stats.get("alert"):63                print(f"[immo-ka]   ⚠ ALERTE {sid} : {stats['alert']}")64            print(f"[immo-ka]   {stats}")65            results.append(stats)66        except Exception as exc:  # robustesse : une source ne bloque pas les autres67            db.log_failure(con, sid, f"{exc}")68            traceback.print_exc()69            results.append({"source": sid, "error": str(exc)})70    # recalcule la déduplication (pré-calculée pour des lectures instantanées)71    try:72        hidden = db.refresh_dedup(con)73        print(f"[immo-ka] dédup: {hidden} doublon(s) masqué(s) "74              "(sous-agences Centris + adresse inter-sources)")75    except Exception:76        traceback.print_exc()77    # contrôle qualité : score de complétude, cohérence immobilière, seuil de78    # publication (quarantaine) et champs dérivés (prix/pi², transaction)79    try:80        from . import quality81        q = quality.refresh(con)82        print(f"[immo-ka] qualité: {q['publiees']} publiée(s), "83              f"{q['quarantaine']} en quarantaine, "84              f"{q['recalculees']} recalculée(s)")85    except Exception:86        traceback.print_exc()87    # statistiques du planificateur SQLite : sans elles, les requêtes bbox88    # de la carte (Ka Maps) n'utilisent pas idx_listings_geo89    try:90        con.execute("PRAGMA optimize")91    except Exception:92        pass93    con.close()94    return results959697def watch(interval_seconds: int = 3600) -> None:98    """Boucle de synchronisation périodique (équivalent webhook, via PM2/cron).99100    Après chaque synchronisation, l'enrichissement continu de la carte101    (Ka Maps) : géocodage EN LOT des nouvelles adresses (Adresses Québec),102    puis appariement Vrai-Prix local (estimations + coordonnées du rôle)103    quand data/vraiprix.db (ou VRAIPRIX_DB) est disponible.104    """105    while True:106        cycle_t0 = time.time()107        run()108        try:  # nouvelles adresses → coordonnées (cache : quasi gratuit ensuite)109            from . import geocode110            # audit de cohérence d'abord : les fiches mal localisées (rue111            # homonyme, geo source erroné) sont remises en file, puis le lot112            # les re-géocode rigoureusement (budget Nominatim borné par passe)113            geocode.run_audit(nominatim_budget=50)114            geocode.run_batch()115        except Exception as exc:116            print(f"[immo-ka] geocode: erreur non bloquante: {exc}", file=sys.stderr)117        try:  # audit budgété des images (liens morts, minuscules, corrompues)118            from . import imgaudit119            ia = imgaudit.run_batch(limit=2000)120            print(f"[immo-ka] images: {ia['urls_verifiees']} URL vérifiée(s), "121                  f"{ia['images_retirees']} retirée(s), "122                  f"{ia['sans_image_valide']} annonce(s) sans image valide")123        except Exception as exc:124            print(f"[immo-ka] imgaudit: erreur non bloquante: {exc}", file=sys.stderr)125        try:  # taux hypothécaires — collecte espacée (IMMOKA_MORTGAGE_INTERVAL_MIN)126            from .mortgage import scheduler as mortgage_scheduler127            mortgage_scheduler.maybe_run()128        except Exception as exc:129            print(f"[immo-ka] mortgage: erreur non bloquante: {exc}", file=sys.stderr)130        # sommeil ADAPTATIF : une passe complète peut durer des heures (30131        # sources) — un sleep fixe par-dessus ferait dépasser 24 h de cadence132        # et la supervision marquerait toutes les sources « stale ».133        delay = max(60, interval_seconds - int(time.time() - cycle_t0))134        print(f"[immo-ka] prochaine synchronisation dans {delay // 60} min")135        time.sleep(delay)136