spb/house-ka

1# -----------------------------------------------------------------------------2# Immo-Ka — Agrégateur de maisons à vendre (province de Québec)3# Auteur : Simon-Pierre Boucher — contact@spboucher.ai4# mortgage/scheduler.py : orchestration de la collecte des taux.5#   Pipeline : provider.fetch() -> validation -> enregistrement historisé.6#   Retries avec backoff exponentiel, timeout par provider, logs structurés,7#   santé par provider (provider_runs). Une panne d'un provider n'affecte8#   jamais les autres ni le calculateur (dernière donnée valide conservée).9# -----------------------------------------------------------------------------10from __future__ import annotations1112import os13import time14import traceback1516import requests1718from . import store19from .providers import PROVIDERS20from .validate import validate_batch2122RETRIES = int(os.environ.get("IMMOKA_MORTGAGE_RETRIES", "3"))23BACKOFF_BASE_S = float(os.environ.get("IMMOKA_MORTGAGE_BACKOFF", "5"))24# Fréquence de collecte (minutes) — utilisée par watch() ci-dessous.25INTERVAL_MIN = int(os.environ.get("IMMOKA_MORTGAGE_INTERVAL_MIN", "180"))262728def _log(**kw) -> None:29    print("[mortgage] " + " ".join(f"{k}={v}" for k, v in kw.items()),30          flush=True)313233def run_provider(slug: str, con=None) -> dict:34    """Collecte UNE institution avec retries + backoff. Retourne le résumé."""35    if con is None:36        con = store.connect()37    cls = PROVIDERS[slug]38    t0 = time.time()39    products: list[dict] | None = None40    status, message = "success", ""41    for attempt in range(RETRIES):42        try:43            products = cls().fetch()44            break45        except requests.RequestException as exc:46            status, message = "http_error", str(exc)[:200]47        except Exception as exc:  # noqa: BLE001 — parseur cassé, etc.48            status, message = "parser_error", str(exc)[:200]49            traceback.print_exc()50        if attempt < RETRIES - 1:51            time.sleep(BACKOFF_BASE_S * (2 ** attempt))52    duration_ms = int((time.time() - t0) * 1000)53    if products is None:54        store.log_run(con, slug, ok=False, status=status,55                      duration_ms=duration_ms, message=message)56        _log(provider=slug, status=status, duration=f"{duration_ms}ms",57             message=message or "-")58        return {"provider": slug, "ok": False, "status": status,59                "message": message}60    valid, problems = validate_batch(products)61    if not valid:62        status = "empty" if not products else "validation_error"63        store.log_run(con, slug, ok=False, status=status,64                      products=len(products), rejected=len(problems),65                      duration_ms=duration_ms,66                      message="; ".join(problems[:5]))67        _log(provider=slug, status=status, products=len(products),68             rejected=len(problems), duration=f"{duration_ms}ms")69        return {"provider": slug, "ok": False, "status": status,70                "problems": problems}71    res = store.record_observations(con, slug, valid)72    ok = True73    if problems or res["rejected"]:74        status = "success"  # partiel : données saines enregistrées quand même75    store.log_run(con, slug, ok=ok, status=status, products=len(valid),76                  changed=res["changed"],77                  rejected=len(problems) + res["rejected"],78                  duration_ms=duration_ms,79                  message="; ".join((problems + res["rejected_details"])[:5]))80    _log(provider=slug, status=status, products=len(valid),81         changed=res["changed"], rejected=len(problems) + res["rejected"],82         duration=f"{duration_ms}ms")83    return {"provider": slug, "ok": True, "status": status,84            "products": len(valid), "changed": res["changed"],85            "rejected": len(problems) + res["rejected"]}868788def run(only: list[str] | None = None) -> list[dict]:89    """Collecte toutes les institutions (ou celles listées). Séquentiel et90    poli — jamais de martèlement des sites bancaires."""91    con = store.connect()92    slugs = [s for s in sorted(PROVIDERS) if not only or s in only]93    results = [run_provider(s, con) for s in slugs]94    ok = sum(1 for r in results if r["ok"])95    _log(status="done", providers=len(results), ok=ok,96         failed=len(results) - ok)97    return results9899100def watch(interval_minutes: int | None = None) -> None:101    """Boucle autonome de collecte (défaut : IMMOKA_MORTGAGE_INTERVAL_MIN)."""102    minutes = interval_minutes or INTERVAL_MIN103    while True:104        try:105            run()106        except Exception:  # noqa: BLE001 — la boucle ne meurt jamais107            traceback.print_exc()108        time.sleep(minutes * 60)109110111def maybe_run(min_age_minutes: int | None = None) -> None:112    """Collecte seulement si la dernière passe date de plus de113    `min_age_minutes` — appelé depuis la boucle watch d'ingest.py sans risque114    de sur-solliciter les banques."""115    age_min = min_age_minutes or INTERVAL_MIN116    con = store.connect()117    last = con.execute("SELECT MAX(ts) AS m FROM provider_runs").fetchone()118    if last and last["m"] and (time.time() - last["m"]) < age_min * 60:119        return120    run()121