spb/toit-ka Public

1# Author: Simon-Pierre Boucher2# Contact: contact@spboucher.ai3# Project: Toit-Ka4# -----------------------------------------------------------------------------5# etl.py : répliques louka.db / immoka.db  ->  toitka.db (BD unifiée).6#7#   · Lecture SEULE des sources (mode ro) — les BD Lou-Ka / Immo-Ka de prod ne8#     sont jamais touchées ; Toit-Ka lit des snapshots répliqués (replicate.py).9#   · Règles de visibilité HÉRITÉES des sites d'origine :10#       Lou-Ka  : active=111#       Immo-Ka : active=1 AND dup_hidden=0 AND price IS NOT NULL (dédup Centris)12#   · Normalisation : villes canoniques (toitka/villes.py, persistées dans13#     city_map), types unifiés (toitka/typologie.py), entités HTML décodées.14#   · Cycle de vie : upsert par uid ; toute annonce absente de la passe courante15#     est désactivée (active=0 -> la fiche renvoie 410, le reste du site l'ignore).16#   · watch(interval) : boucle replicate (si les répliques sont configurées,17#     c.-à-d. en prod) puis ETL.18# -----------------------------------------------------------------------------19from __future__ import annotations2021import html22import json23import time24import traceback2526from . import db, typologie, villes2728# bornes de plausibilité des prix (héritées des sites d'origine)29LOYER_MIN, LOYER_MAX = 195, 15000        # Lou-Ka (seo.py)30PRIX_MIN = 10000                          # Immo-Ka (normalize.parse_price)313233def _clean_text(s: str | None) -> str:34    """Décode les entités HTML résiduelles (« L&#8217;Opale », « &#038; »)."""35    if not s:36        return ""37    return html.unescape(html.unescape(s)).strip()383940def _city_counts(rows: list[dict]) -> dict[str, int]:41    counts: dict[str, int] = {}42    for r in rows:43        raw = (r.get("city_raw") or "").strip()44        counts[raw] = counts.get(raw, 0) + 145    return counts464748# --- extraction ----------------------------------------------------------------4950def _read_louka() -> list[dict]:51    con = db.connect_source(db.LOUKA_DB)52    rows = []53    try:54        for r in con.execute("SELECT * FROM listings WHERE active=1"):55            d = dict(r)56            price = d.get("price")57            if price is not None and not (LOYER_MIN <= price <= LOYER_MAX):58                price = None                      # loyer implausible -> « sur demande »59            details = {}60            try:61                details = json.loads(d.get("details") or "{}") or {}62            except (ValueError, TypeError):63                pass64            rows.append({65                "uid": f"lou:{d['uid']}",66                "origin": "louka",67                "transaction_type": "louer",68                "source": d["source"],69                "external_id": d["external_id"],70                "url": d.get("url") or "",71                "title": _clean_text(d.get("title")),72                "address": _clean_text(d.get("address")),73                "sector": _clean_text(d.get("sector")),74                "city_raw": (d.get("city") or "").strip(),75                "type": typologie.type_louer(d.get("unit_type")),76                "price": price,77                "price_label": _clean_text(d.get("price_label")),78                "bedrooms": None,79                "bathrooms": None,80                "area_sqft": d.get("area_sqft"),81                "lot_sqft": None,82                "year_built": None,83                "pets": d.get("pets"),84                "furnished": d.get("furnished"),85                "availability_date": d.get("availability_date"),86                "mls": "",87                "broker_name": "",88                "agency": "",89                "description": _clean_text(d.get("description")),90                "images": d.get("images") or "[]",91                "lat": d.get("lat"), "lng": d.get("lng"),92                "first_seen": d.get("first_seen"),93                "updated_at": d.get("updated_at"),94            })95    finally:96        con.close()97    return rows9899100def _read_immoka() -> list[dict]:101    con = db.connect_source(db.IMMOKA_DB)102    rows = []103    try:104        for r in con.execute(105                "SELECT * FROM listings"106                " WHERE active=1 AND dup_hidden=0 AND price IS NOT NULL"):107            d = dict(r)108            price = d.get("price")109            if price is not None and price < PRIX_MIN:110                continue                          # prix sentinelle / erreur de source111            rows.append({112                "uid": f"immo:{d['uid']}",113                "origin": "immoka",114                "transaction_type": "acheter",115                "source": d["source"],116                "external_id": d["external_id"],117                "url": d.get("url") or "",118                "title": _clean_text(d.get("title")),119                "address": _clean_text(d.get("address")),120                "sector": _clean_text(d.get("sector")),121                "city_raw": (d.get("city") or "").strip(),122                "type": typologie.type_acheter(d.get("property_type")),123                "price": price,124                "price_label": _clean_text(d.get("price_label")),125                "bedrooms": d.get("bedrooms"),126                "bathrooms": d.get("bathrooms"),127                "area_sqft": d.get("area_sqft"),128                "lot_sqft": d.get("lot_sqft"),129                "year_built": d.get("year_built"),130                "pets": None, "furnished": None, "availability_date": None,131                "mls": d.get("mls") or "",132                "broker_name": _clean_text(d.get("broker_name")),133                "agency": _clean_text(d.get("agency")),134                "description": _clean_text(d.get("description")),135                "images": d.get("images") or "[]",136                "lat": d.get("lat"), "lng": d.get("lng"),137                "first_seen": d.get("first_seen"),138                "updated_at": d.get("updated_at"),139            })140    finally:141        con.close()142    return rows143144145# --- chargement ------------------------------------------------------------------146147_UPSERT = """148INSERT INTO listings (uid, origin, transaction_type, source, external_id, url,149    title, address, sector, city, city_raw, type, price, price_label, bedrooms,150    bathrooms, area_sqft, lot_sqft, year_built, pets, furnished,151    availability_date, mls, broker_name, agency, description, images, lat, lng,152    first_seen, updated_at, etl_run, active)153VALUES (:uid, :origin, :transaction_type, :source, :external_id, :url,154    :title, :address, :sector, :city, :city_raw, :type, :price, :price_label,155    :bedrooms, :bathrooms, :area_sqft, :lot_sqft, :year_built, :pets,156    :furnished, :availability_date, :mls, :broker_name, :agency, :description,157    :images, :lat, :lng, :first_seen, :updated_at, :etl_run, 1)158ON CONFLICT(uid) DO UPDATE SET159    url=excluded.url, title=excluded.title, address=excluded.address,160    sector=excluded.sector, city=excluded.city, city_raw=excluded.city_raw,161    type=excluded.type, price=excluded.price, price_label=excluded.price_label,162    bedrooms=excluded.bedrooms, bathrooms=excluded.bathrooms,163    area_sqft=excluded.area_sqft, lot_sqft=excluded.lot_sqft,164    year_built=excluded.year_built, pets=excluded.pets,165    furnished=excluded.furnished, availability_date=excluded.availability_date,166    mls=excluded.mls, broker_name=excluded.broker_name, agency=excluded.agency,167    description=excluded.description, images=excluded.images,168    lat=COALESCE(excluded.lat, listings.lat),169    lng=COALESCE(excluded.lng, listings.lng),170    first_seen=COALESCE(listings.first_seen, excluded.first_seen),171    updated_at=excluded.updated_at, etl_run=excluded.etl_run, active=1172"""173174175def run() -> dict:176    """Une passe complète : lit les deux répliques, normalise, charge toitka.db."""177    t0 = time.time()178    out = {"louka": 0, "immoka": 0}179    all_rows: list[dict] = []180    con = db.connect()181    try:182        for origin, reader, src in (("louka", _read_louka, db.LOUKA_DB),183                                    ("immoka", _read_immoka, db.IMMOKA_DB)):184            if not src.exists():185                con.execute("INSERT INTO etl_log (ts, origin, found, kept, ok, message)"186                            " VALUES (?,?,?,?,?,?)",187                            (time.time(), origin, 0, 0, 0, f"réplique absente : {src}"))188                print(f"[etl] {origin} : réplique absente ({src}) — passe ignorée")189                continue190            try:191                rows = reader()192                all_rows.extend(rows)193                out[origin] = len(rows)194            except Exception as e:                       # une source cassée ne bloque pas l'autre195                con.execute("INSERT INTO etl_log (ts, origin, found, kept, ok, message)"196                            " VALUES (?,?,?,?,?,?)",197                            (time.time(), origin, 0, 0, 0, f"{type(e).__name__}: {e}"))198                traceback.print_exc()199200        if not all_rows:201            con.commit()202            return out203204        # villes canoniques : mapping recalculé sur l'union des deux corpus205        mapping = villes.build_mapping(_city_counts(all_rows))206        counts = _city_counts(all_rows)207        con.execute("DELETE FROM city_map")208        con.executemany(209            "INSERT INTO city_map (raw, canonical, sector, n) VALUES (?,?,?,?)",210            [(raw, v, s, counts.get(raw, 0)) for raw, (v, s) in mapping.items()])211212        etl_run = int(t0)213        for r in all_rows:214            ville, secteur = mapping.get(r["city_raw"], ("", ""))215            r["city"] = ville216            if secteur and not r["sector"]:217                r["sector"] = secteur218            r["etl_run"] = etl_run219220        con.executemany(_UPSERT, all_rows)221        # annonces disparues des sources -> désactivées (fiche : 410)222        deactivated = con.execute(223            "UPDATE listings SET active=0 WHERE etl_run<>? AND active=1",224            (etl_run,)).rowcount225        for origin in ("louka", "immoka"):226            con.execute("INSERT INTO etl_log (ts, origin, found, kept, ok, message)"227                        " VALUES (?,?,?,?,?,?)",228                        (time.time(), origin, out[origin], out[origin], 1, ""))229        con.commit()230        con.execute("PRAGMA optimize")231        dt = time.time() - t0232        print(f"[etl] louer={out['louka']} acheter={out['immoka']}"233              f" désactivées={deactivated} villes_map={len(mapping)} ({dt:.1f} s)")234    finally:235        con.close()236    return out237238239def watch(interval: float) -> None:240    """Boucle de prod : replicate (best-effort) puis ETL, toutes les `interval` s."""241    from . import replicate242    while True:243        try:244            replicate.run()245        except Exception:246            traceback.print_exc()247        try:248            run()249        except Exception:250            traceback.print_exc()251        time.sleep(interval)252253254def rapport_villes() -> None:255    """Audit : montre les regroupements de villes opérés par la canonicalisation."""256    con = db.connect()257    try:258        rows = con.execute(259            "SELECT canonical, raw, sector, n FROM city_map"260            " WHERE canonical<>'' ORDER BY canonical, n DESC").fetchall()261        groups: dict[str, list] = {}262        for r in rows:263            groups.setdefault(r["canonical"], []).append(r)264        merged = {k: v for k, v in groups.items()265                  if len(v) > 1 or v[0]["raw"] != k}266        print(f"{len(groups)} villes canoniques, {len(merged)} avec regroupement :\n")267        for canon, items in sorted(merged.items(), key=lambda kv: -sum(i["n"] for i in kv[1])):268            total = sum(i["n"] for i in items)269            print(f"■ {canon}  ({total})")270            for i in items[:12]:271                sect = f"  → secteur « {i['sector']} »" if i["sector"] else ""272                print(f"    · {i['raw']!r}  ×{i['n']}{sect}")273            if len(items) > 12:274                print(f"    … +{len(items) - 12} variantes")275        dropped = con.execute(276            "SELECT raw, n FROM city_map WHERE canonical='' ORDER BY n DESC LIMIT 25").fetchall()277        if dropped:278            print("\nValeurs irrécupérables (ville vide) :")279            for r in dropped:280                print(f"    · {r['raw']!r}  ×{r['n']}")281    finally:282        con.close()283