spb/home-ka

1# -----------------------------------------------------------------------------2# Home-Ka — US real-estate aggregator (Groupe KA)3# Author: Simon-Pierre Boucher — contact@spboucher.ai4# ingest.py : ingestion pipeline — runs the connectors and syncs the database5#             (adds / updates / removals) = content always current.6#7#   SOURCE → connector → raw ingestion → normalization → property matching8#          → deduplication → canonical listing → search index → Home-Ka9#10# Sources come from two registries (see connectors/__init__.py):11#   - rows of the `sources` table (config-driven family connectors)12#   - CUSTOM one-site classes (connectors/custom_broker/*.py)13# -----------------------------------------------------------------------------14from __future__ import annotations1516import sys17import time18import traceback1920from . import connectors as creg21from . import db222324def _targets(con, requested: list[str] | None) -> list[tuple[str, str, dict]]:25    """(source_id, connector_type, config) for every runnable source."""26    rows = {s["id"]: s for s in db.get_sources(con)}27    out: list[tuple[str, str, dict]] = []28    ids = requested or (29        [s["id"] for s in rows.values() if s["enabled"]]30        + [sid for sid in creg.CUSTOM if sid not in rows])31    for sid in ids:32        row = rows.get(sid)33        if row is not None:34            if not requested and not row["enabled"]:35                continue36            out.append((sid, row["connector_type"] or "", row["config"]))37        elif sid in creg.CUSTOM:38            out.append((sid, "custom", {}))39        else:40            print(f"[home-ka] unknown source: {sid}", file=sys.stderr)41    return out424344def run(sources: list[str] | None = None) -> list[dict]:45    """Run the ingestion for all sources (or the requested ones)."""46    con = db.connect()47    results = []48    for sid, ctype, config in _targets(con, sources):49        conn = creg.build(sid, ctype, config)50        if conn is None:51            print(f"[home-ka] no connector for {sid} (type {ctype!r})",52                  file=sys.stderr)53            continue54        # slow sources (public records...): config.sync_interval_hours skips55        # the source while its last successful sync is fresh enough56        interval_h = float((config or {}).get("sync_interval_hours") or 0)57        if interval_h and not sources:58            last = con.execute(59                "SELECT MAX(ts) ts FROM sync_log WHERE source=? AND ok=1",60                (sid,)).fetchone()["ts"]61            if last and time.time() - last < interval_h * 3600:62                continue63        t0 = time.time()64        print(f"[home-ka] sync {sid} ...")65        try:66            if getattr(conn, "is_public_records", False):67                from .connectors.public_data.arcgis import sync_records68                stats = sync_records(con, sid, conn.fetch_records())69            else:70                listings = conn.fetch()71                finalized, dropped = [], 072                for lst in listings:73                    try:74                        finalized.append(lst.finalize())75                    except Exception:   # one bad listing never blocks a source76                        dropped += 177                stats = db.sync_source(con, sid, finalized)78                if dropped:79                    stats["dropped"] = dropped80            stats["seconds"] = round(time.time() - t0, 1)81            if stats.get("alert"):82                print(f"[home-ka]   ⚠ ALERT {sid}: {stats['alert']}")83            print(f"[home-ka]   {stats}")84            results.append(stats)85        except Exception as exc:   # one source never blocks the others86            db.log_failure(con, sid, f"{exc}")87            traceback.print_exc()88            results.append({"source": sid, "error": str(exc)})89    # precomputed dedup (instant reads, same principle as immo-ka)90    try:91        hidden = db.refresh_dedup(con)92        print(f"[home-ka] dedup: {hidden} duplicate(s) hidden "93              "(MLS number + same-property cross-source)")94    except Exception:95        traceback.print_exc()96    # quality gate: completeness score, publication threshold (quarantine)97    try:98        from . import quality99        q = quality.refresh(con)100        print(f"[home-ka] quality: {q['published']} published, "101              f"{q['quarantined']} quarantined")102    except Exception:103        traceback.print_exc()104    try:105        con.execute("PRAGMA optimize")106    except Exception:107        pass108    con.close()109    return results110111112def watch(interval_seconds: int = 3600) -> None:113    """Periodic sync loop (webhook equivalent, via PM2/cron).114115    After each sync: batch geocoding of new addresses (US Census geocoder,116    free), then the budgeted image audit.117    """118    while True:119        run()120        try:121            from . import geocode122            geocode.run_batch()123        except Exception as exc:124            print(f"[home-ka] geocode: non-blocking error: {exc}", file=sys.stderr)125        try:126            from . import imgaudit127            ia = imgaudit.run_batch(limit=2000)128            print(f"[home-ka] images: {ia['urls_verifiees']} URL(s) checked, "129                  f"{ia['images_retirees']} removed, "130                  f"{ia['sans_image_valide']} listing(s) without a valid image")131        except Exception as exc:132            print(f"[home-ka] imgaudit: non-blocking error: {exc}", file=sys.stderr)133        try:   # budgeted brokerage discovery pass (inspect a few prospects)134            from . import discovery135            d = discovery.run_batch(limit=15)136            if d.get("inspected"):137                print(f"[home-ka] discovery: {d['inspected']} brokerage(s) inspected")138        except Exception as exc:139            print(f"[home-ka] discovery: non-blocking error: {exc}", file=sys.stderr)140        print(f"[home-ka] next sync in {interval_seconds // 60} min")141        time.sleep(interval_seconds)142