# Author: Simon-Pierre Boucher # Contact: contact@spboucher.ai # Project: Toit-Ka # ----------------------------------------------------------------------------- # etl.py : répliques louka.db / immoka.db -> toitka.db (BD unifiée). # # · Lecture SEULE des sources (mode ro) — les BD Lou-Ka / Immo-Ka de prod ne # sont jamais touchées ; Toit-Ka lit des snapshots répliqués (replicate.py). # · Règles de visibilité HÉRITÉES des sites d'origine : # Lou-Ka : active=1 # Immo-Ka : active=1 AND dup_hidden=0 AND price IS NOT NULL (dédup Centris) # · Normalisation : villes canoniques (toitka/villes.py, persistées dans # city_map), types unifiés (toitka/typologie.py), entités HTML décodées. # · Cycle de vie : upsert par uid ; toute annonce absente de la passe courante # est désactivée (active=0 -> la fiche renvoie 410, le reste du site l'ignore). # · watch(interval) : boucle replicate (si les répliques sont configurées, # c.-à-d. en prod) puis ETL. # ----------------------------------------------------------------------------- from __future__ import annotations import html import json import time import traceback from . import db, typologie, villes # bornes de plausibilité des prix (héritées des sites d'origine) LOYER_MIN, LOYER_MAX = 195, 15000 # Lou-Ka (seo.py) PRIX_MIN = 10000 # Immo-Ka (normalize.parse_price) def _clean_text(s: str | None) -> str: """Décode les entités HTML résiduelles (« L’Opale », « & »).""" if not s: return "" return html.unescape(html.unescape(s)).strip() def _city_counts(rows: list[dict]) -> dict[str, int]: counts: dict[str, int] = {} for r in rows: raw = (r.get("city_raw") or "").strip() counts[raw] = counts.get(raw, 0) + 1 return counts # --- extraction ---------------------------------------------------------------- def _read_louka() -> list[dict]: con = db.connect_source(db.LOUKA_DB) rows = [] try: for r in con.execute("SELECT * FROM listings WHERE active=1"): d = dict(r) price = d.get("price") if price is not None and not (LOYER_MIN <= price <= LOYER_MAX): price = None # loyer implausible -> « sur demande » details = {} try: details = json.loads(d.get("details") or "{}") or {} except (ValueError, TypeError): pass rows.append({ "uid": f"lou:{d['uid']}", "origin": "louka", "transaction_type": "louer", "source": d["source"], "external_id": d["external_id"], "url": d.get("url") or "", "title": _clean_text(d.get("title")), "address": _clean_text(d.get("address")), "sector": _clean_text(d.get("sector")), "city_raw": (d.get("city") or "").strip(), "type": typologie.type_louer(d.get("unit_type")), "price": price, "price_label": _clean_text(d.get("price_label")), "bedrooms": None, "bathrooms": None, "area_sqft": d.get("area_sqft"), "lot_sqft": None, "year_built": None, "pets": d.get("pets"), "furnished": d.get("furnished"), "availability_date": d.get("availability_date"), "mls": "", "broker_name": "", "agency": "", "description": _clean_text(d.get("description")), "images": d.get("images") or "[]", "lat": d.get("lat"), "lng": d.get("lng"), "first_seen": d.get("first_seen"), "updated_at": d.get("updated_at"), }) finally: con.close() return rows def _read_immoka() -> list[dict]: con = db.connect_source(db.IMMOKA_DB) rows = [] try: for r in con.execute( "SELECT * FROM listings" " WHERE active=1 AND dup_hidden=0 AND price IS NOT NULL"): d = dict(r) price = d.get("price") if price is not None and price < PRIX_MIN: continue # prix sentinelle / erreur de source rows.append({ "uid": f"immo:{d['uid']}", "origin": "immoka", "transaction_type": "acheter", "source": d["source"], "external_id": d["external_id"], "url": d.get("url") or "", "title": _clean_text(d.get("title")), "address": _clean_text(d.get("address")), "sector": _clean_text(d.get("sector")), "city_raw": (d.get("city") or "").strip(), "type": typologie.type_acheter(d.get("property_type")), "price": price, "price_label": _clean_text(d.get("price_label")), "bedrooms": d.get("bedrooms"), "bathrooms": d.get("bathrooms"), "area_sqft": d.get("area_sqft"), "lot_sqft": d.get("lot_sqft"), "year_built": d.get("year_built"), "pets": None, "furnished": None, "availability_date": None, "mls": d.get("mls") or "", "broker_name": _clean_text(d.get("broker_name")), "agency": _clean_text(d.get("agency")), "description": _clean_text(d.get("description")), "images": d.get("images") or "[]", "lat": d.get("lat"), "lng": d.get("lng"), "first_seen": d.get("first_seen"), "updated_at": d.get("updated_at"), }) finally: con.close() return rows # --- chargement ------------------------------------------------------------------ _UPSERT = """ INSERT INTO listings (uid, origin, transaction_type, source, external_id, url, title, address, sector, city, city_raw, type, price, price_label, bedrooms, bathrooms, area_sqft, lot_sqft, year_built, pets, furnished, availability_date, mls, broker_name, agency, description, images, lat, lng, first_seen, updated_at, etl_run, active) VALUES (:uid, :origin, :transaction_type, :source, :external_id, :url, :title, :address, :sector, :city, :city_raw, :type, :price, :price_label, :bedrooms, :bathrooms, :area_sqft, :lot_sqft, :year_built, :pets, :furnished, :availability_date, :mls, :broker_name, :agency, :description, :images, :lat, :lng, :first_seen, :updated_at, :etl_run, 1) ON CONFLICT(uid) DO UPDATE SET url=excluded.url, title=excluded.title, address=excluded.address, sector=excluded.sector, city=excluded.city, city_raw=excluded.city_raw, type=excluded.type, price=excluded.price, price_label=excluded.price_label, bedrooms=excluded.bedrooms, bathrooms=excluded.bathrooms, area_sqft=excluded.area_sqft, lot_sqft=excluded.lot_sqft, year_built=excluded.year_built, pets=excluded.pets, furnished=excluded.furnished, availability_date=excluded.availability_date, mls=excluded.mls, broker_name=excluded.broker_name, agency=excluded.agency, description=excluded.description, images=excluded.images, lat=COALESCE(excluded.lat, listings.lat), lng=COALESCE(excluded.lng, listings.lng), first_seen=COALESCE(listings.first_seen, excluded.first_seen), updated_at=excluded.updated_at, etl_run=excluded.etl_run, active=1 """ def run() -> dict: """Une passe complète : lit les deux répliques, normalise, charge toitka.db.""" t0 = time.time() out = {"louka": 0, "immoka": 0} all_rows: list[dict] = [] con = db.connect() try: for origin, reader, src in (("louka", _read_louka, db.LOUKA_DB), ("immoka", _read_immoka, db.IMMOKA_DB)): if not src.exists(): con.execute("INSERT INTO etl_log (ts, origin, found, kept, ok, message)" " VALUES (?,?,?,?,?,?)", (time.time(), origin, 0, 0, 0, f"réplique absente : {src}")) print(f"[etl] {origin} : réplique absente ({src}) — passe ignorée") continue try: rows = reader() all_rows.extend(rows) out[origin] = len(rows) except Exception as e: # une source cassée ne bloque pas l'autre con.execute("INSERT INTO etl_log (ts, origin, found, kept, ok, message)" " VALUES (?,?,?,?,?,?)", (time.time(), origin, 0, 0, 0, f"{type(e).__name__}: {e}")) traceback.print_exc() if not all_rows: con.commit() return out # villes canoniques : mapping recalculé sur l'union des deux corpus mapping = villes.build_mapping(_city_counts(all_rows)) counts = _city_counts(all_rows) con.execute("DELETE FROM city_map") con.executemany( "INSERT INTO city_map (raw, canonical, sector, n) VALUES (?,?,?,?)", [(raw, v, s, counts.get(raw, 0)) for raw, (v, s) in mapping.items()]) etl_run = int(t0) for r in all_rows: ville, secteur = mapping.get(r["city_raw"], ("", "")) r["city"] = ville if secteur and not r["sector"]: r["sector"] = secteur r["etl_run"] = etl_run con.executemany(_UPSERT, all_rows) # annonces disparues des sources -> désactivées (fiche : 410) deactivated = con.execute( "UPDATE listings SET active=0 WHERE etl_run<>? AND active=1", (etl_run,)).rowcount for origin in ("louka", "immoka"): con.execute("INSERT INTO etl_log (ts, origin, found, kept, ok, message)" " VALUES (?,?,?,?,?,?)", (time.time(), origin, out[origin], out[origin], 1, "")) con.commit() con.execute("PRAGMA optimize") dt = time.time() - t0 print(f"[etl] louer={out['louka']} acheter={out['immoka']}" f" désactivées={deactivated} villes_map={len(mapping)} ({dt:.1f} s)") finally: con.close() return out def watch(interval: float) -> None: """Boucle de prod : replicate (best-effort) puis ETL, toutes les `interval` s.""" from . import replicate while True: try: replicate.run() except Exception: traceback.print_exc() try: run() except Exception: traceback.print_exc() time.sleep(interval) def rapport_villes() -> None: """Audit : montre les regroupements de villes opérés par la canonicalisation.""" con = db.connect() try: rows = con.execute( "SELECT canonical, raw, sector, n FROM city_map" " WHERE canonical<>'' ORDER BY canonical, n DESC").fetchall() groups: dict[str, list] = {} for r in rows: groups.setdefault(r["canonical"], []).append(r) merged = {k: v for k, v in groups.items() if len(v) > 1 or v[0]["raw"] != k} print(f"{len(groups)} villes canoniques, {len(merged)} avec regroupement :\n") for canon, items in sorted(merged.items(), key=lambda kv: -sum(i["n"] for i in kv[1])): total = sum(i["n"] for i in items) print(f"■ {canon} ({total})") for i in items[:12]: sect = f" → secteur « {i['sector']} »" if i["sector"] else "" print(f" · {i['raw']!r} ×{i['n']}{sect}") if len(items) > 12: print(f" … +{len(items) - 12} variantes") dropped = con.execute( "SELECT raw, n FROM city_map WHERE canonical='' ORDER BY n DESC LIMIT 25").fetchall() if dropped: print("\nValeurs irrécupérables (ville vide) :") for r in dropped: print(f" · {r['raw']!r} ×{r['n']}") finally: con.close()