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 run(sources: list[str] | None = None) -> list[dict]:18 """Exécute l'ingestion pour toutes les sources (ou celles demandées)."""19 con = db.connect()20 results = []21 targets = sources or list(CONNECTORS.keys())22 for sid in targets:23 cls = CONNECTORS.get(sid)24 if cls is None:25 print(f"[immo-ka] connecteur inconnu : {sid}", file=sys.stderr)26 continue27 t0 = time.time()28 print(f"[immo-ka] sync {sid} ...")29 try:30 listings = cls().fetch()31 finalized, dropped = [], 032 for lst in listings:33 try:34 finalized.append(lst.finalize())35 except Exception: # une annonce malformée ne bloque pas la source36 dropped += 137 try:38 # garde géographique : coordonnées fournies par la source mais39 # incompatibles avec la ville annoncée -> annulées (le géocodeur40 # rigoureux prendra le relais). Cache seulement, aucun réseau.41 from . import geocode42 bad = geocode.strip_bad_source_coords(con, finalized)43 if bad:44 print(f"[immo-ka] {bad} coordonnée(s) source incohérente(s) rejetée(s)")45 except Exception:46 traceback.print_exc()47 stats = db.sync_source(con, sid, finalized)48 stats["seconds"] = round(time.time() - t0, 1)49 if dropped:50 stats["dropped"] = dropped51 if stats.get("alert"):52 print(f"[immo-ka] ⚠ ALERTE {sid} : {stats['alert']}")53 print(f"[immo-ka] {stats}")54 results.append(stats)55 except Exception as exc: # robustesse : une source ne bloque pas les autres56 db.log_failure(con, sid, f"{exc}")57 traceback.print_exc()58 results.append({"source": sid, "error": str(exc)})59 # recalcule la déduplication (pré-calculée pour des lectures instantanées)60 try:61 hidden = db.refresh_dedup(con)62 print(f"[immo-ka] dédup: {hidden} doublon(s) masqué(s) "63 "(sous-agences Centris + adresse inter-sources)")64 except Exception:65 traceback.print_exc()66 # contrôle qualité : score de complétude, cohérence immobilière, seuil de67 # publication (quarantaine) et champs dérivés (prix/pi², transaction)68 try:69 from . import quality70 q = quality.refresh(con)71 print(f"[immo-ka] qualité: {q['publiees']} publiée(s), "72 f"{q['quarantaine']} en quarantaine, "73 f"{q['recalculees']} recalculée(s)")74 except Exception:75 traceback.print_exc()76 # statistiques du planificateur SQLite : sans elles, les requêtes bbox77 # de la carte (Ka Maps) n'utilisent pas idx_listings_geo78 try:79 con.execute("PRAGMA optimize")80 except Exception:81 pass82 con.close()83 return results848586def watch(interval_seconds: int = 3600) -> None:87 """Boucle de synchronisation périodique (équivalent webhook, via PM2/cron).8889 Après chaque synchronisation, l'enrichissement continu de la carte90 (Ka Maps) : géocodage EN LOT des nouvelles adresses (Adresses Québec),91 puis appariement Vrai-Prix local (estimations + coordonnées du rôle)92 quand data/vraiprix.db (ou VRAIPRIX_DB) est disponible.93 """94 while True:95 run()96 try: # nouvelles adresses → coordonnées (cache : quasi gratuit ensuite)97 from . import geocode98 # audit de cohérence d'abord : les fiches mal localisées (rue99 # homonyme, geo source erroné) sont remises en file, puis le lot100 # les re-géocode rigoureusement (budget Nominatim borné par passe)101 geocode.run_audit(nominatim_budget=50)102 geocode.run_batch()103 except Exception as exc:104 print(f"[immo-ka] geocode: erreur non bloquante: {exc}", file=sys.stderr)105 try: # audit budgété des images (liens morts, minuscules, corrompues)106 from . import imgaudit107 ia = imgaudit.run_batch(limit=2000)108 print(f"[immo-ka] images: {ia['urls_verifiees']} URL vérifiée(s), "109 f"{ia['images_retirees']} retirée(s), "110 f"{ia['sans_image_valide']} annonce(s) sans image valide")111 except Exception as exc:112 print(f"[immo-ka] imgaudit: erreur non bloquante: {exc}", file=sys.stderr)113 try: # taux hypothécaires — collecte espacée (IMMOKA_MORTGAGE_INTERVAL_MIN)114 from .mortgage import scheduler as mortgage_scheduler115 mortgage_scheduler.maybe_run()116 except Exception as exc:117 print(f"[immo-ka] mortgage: erreur non bloquante: {exc}", file=sys.stderr)118 print(f"[immo-ka] prochaine synchronisation dans {interval_seconds // 60} min")119 time.sleep(interval_seconds)120