# -----------------------------------------------------------------------------
# Home-Ka — US real-estate aggregator (Groupe KA)
# Author: Simon-Pierre Boucher — contact@spboucher.ai
# ingest.py : ingestion pipeline — runs the connectors and syncs the database
#             (adds / updates / removals) = content always current.
#
#   SOURCE → connector → raw ingestion → normalization → property matching
#          → deduplication → canonical listing → search index → Home-Ka
#
# Sources come from two registries (see connectors/__init__.py):
#   - rows of the `sources` table (config-driven family connectors)
#   - CUSTOM one-site classes (connectors/custom_broker/*.py)
# -----------------------------------------------------------------------------
from __future__ import annotations

import sys
import time
import traceback

from . import connectors as creg
from . import db


def _targets(con, requested: list[str] | None) -> list[tuple[str, str, dict]]:
    """(source_id, connector_type, config) for every runnable source."""
    rows = {s["id"]: s for s in db.get_sources(con)}
    out: list[tuple[str, str, dict]] = []
    ids = requested or (
        [s["id"] for s in rows.values() if s["enabled"]]
        + [sid for sid in creg.CUSTOM if sid not in rows])
    for sid in ids:
        row = rows.get(sid)
        if row is not None:
            if not requested and not row["enabled"]:
                continue
            out.append((sid, row["connector_type"] or "", row["config"]))
        elif sid in creg.CUSTOM:
            out.append((sid, "custom", {}))
        else:
            print(f"[home-ka] unknown source: {sid}", file=sys.stderr)
    return out


def run(sources: list[str] | None = None) -> list[dict]:
    """Run the ingestion for all sources (or the requested ones)."""
    con = db.connect()
    results = []
    for sid, ctype, config in _targets(con, sources):
        conn = creg.build(sid, ctype, config)
        if conn is None:
            print(f"[home-ka] no connector for {sid} (type {ctype!r})",
                  file=sys.stderr)
            continue
        # slow sources (public records...): config.sync_interval_hours skips
        # the source while its last successful sync is fresh enough
        interval_h = float((config or {}).get("sync_interval_hours") or 0)
        if interval_h and not sources:
            last = con.execute(
                "SELECT MAX(ts) ts FROM sync_log WHERE source=? AND ok=1",
                (sid,)).fetchone()["ts"]
            if last and time.time() - last < interval_h * 3600:
                continue
        t0 = time.time()
        print(f"[home-ka] sync {sid} ...")
        try:
            if getattr(conn, "is_public_records", False):
                from .connectors.public_data.arcgis import sync_records
                stats = sync_records(con, sid, conn.fetch_records())
            else:
                listings = conn.fetch()
                finalized, dropped = [], 0
                for lst in listings:
                    try:
                        finalized.append(lst.finalize())
                    except Exception:   # one bad listing never blocks a source
                        dropped += 1
                stats = db.sync_source(con, sid, finalized)
                if dropped:
                    stats["dropped"] = dropped
            stats["seconds"] = round(time.time() - t0, 1)
            if stats.get("alert"):
                print(f"[home-ka]   ⚠ ALERT {sid}: {stats['alert']}")
            print(f"[home-ka]   {stats}")
            results.append(stats)
        except Exception as exc:   # one source never blocks the others
            db.log_failure(con, sid, f"{exc}")
            traceback.print_exc()
            results.append({"source": sid, "error": str(exc)})
    # precomputed dedup (instant reads, same principle as immo-ka)
    try:
        hidden = db.refresh_dedup(con)
        print(f"[home-ka] dedup: {hidden} duplicate(s) hidden "
              "(MLS number + same-property cross-source)")
    except Exception:
        traceback.print_exc()
    # quality gate: completeness score, publication threshold (quarantine)
    try:
        from . import quality
        q = quality.refresh(con)
        print(f"[home-ka] quality: {q['published']} published, "
              f"{q['quarantined']} quarantined")
    except Exception:
        traceback.print_exc()
    try:
        con.execute("PRAGMA optimize")
    except Exception:
        pass
    con.close()
    return results


def watch(interval_seconds: int = 3600) -> None:
    """Periodic sync loop (webhook equivalent, via PM2/cron).

    After each sync: batch geocoding of new addresses (US Census geocoder,
    free), then the budgeted image audit.
    """
    while True:
        run()
        try:
            from . import geocode
            geocode.run_batch()
        except Exception as exc:
            print(f"[home-ka] geocode: non-blocking error: {exc}", file=sys.stderr)
        try:
            from . import imgaudit
            ia = imgaudit.run_batch(limit=2000)
            print(f"[home-ka] images: {ia['urls_verifiees']} URL(s) checked, "
                  f"{ia['images_retirees']} removed, "
                  f"{ia['sans_image_valide']} listing(s) without a valid image")
        except Exception as exc:
            print(f"[home-ka] imgaudit: non-blocking error: {exc}", file=sys.stderr)
        try:   # budgeted brokerage discovery pass (inspect a few prospects)
            from . import discovery
            d = discovery.run_batch(limit=15)
            if d.get("inspected"):
                print(f"[home-ka] discovery: {d['inspected']} brokerage(s) inspected")
        except Exception as exc:
            print(f"[home-ka] discovery: non-blocking error: {exc}", file=sys.stderr)
        print(f"[home-ka] next sync in {interval_seconds // 60} min")
        time.sleep(interval_seconds)