spb/immo-ka

1# -----------------------------------------------------------------------------2# Immo-Ka — Agrégateur de propriétés à vendre (province de Québec)3# Auteur : Simon-Pierre Boucher — contact@spboucher.ai4# vraiprix_local.py : appariement LOCAL contre la base Vrai-Prix (vraiprix.db,5#   3,7 M unités d'évaluation avec adresse, lat/lng, estimation, fourchette).6#   Une jointure d'adresse (FTS) par annonce → remplit d'un coup :7#     · lat/lng manquants (géocodage instantané, sans API externe)8#     · l'estimation Vrai-Prix (valeur + P10-P90 + lien /estimation/{id})9#   Bien plus rapide que l'API vrai-prix (une requête réseau par annonce).10# -----------------------------------------------------------------------------11from __future__ import annotations1213import json14import os15import re16import sqlite317import time18import unicodedata1920from . import db2122# Emplacement de la base Vrai-Prix (copiée à côté de immoka.db sur le nœud).23VP_DB = os.environ.get(24    "VRAIPRIX_DB",25    str((__import__("pathlib").Path(__file__).resolve().parent.parent26         / "data" / "vraiprix.db")))27SITE = "https://www.vrai-prix.com"28_BBOX = (44.5, 63.0, -80.0, -56.0)   # Québec293031def _norm(s: str) -> str:32    return "".join(c for c in unicodedata.normalize("NFD", (s or "").lower())33                   if unicodedata.category(c) != "Mn").strip()343536# mots de « type de voie » ignorés dans l'appariement (variabilité rue/av/boul…)37_VOIE = {"rue", "av", "ave", "avenue", "boul", "boulevard", "ch", "chemin",38         "place", "pl", "rang", "montee", "montée", "cote", "côte", "route",39         "rte", "terrasse", "impasse", "croissant", "cours", "allee", "allée",40         "de", "du", "des", "la", "le", "les", "l", "d", "st", "ste", "saint",41         "sainte"}424344def _street_words(norm_addr: str) -> set:45    """Mots significatifs de la rue (sans n° civique, type de voie, n° d'app.)."""46    a = re.sub(r"\b(?:app?t?|appartement|unite|suite|local|bureau|#)\s*[\w-]+", " ",47               norm_addr.split(",")[0])48    a = re.sub(r"[^a-z0-9 ]+", " ", a)49    toks = [t for t in a.split() if t]50    return {t for t in toks if not t.isdigit() and t not in _VOIE and len(t) > 1}515253def _fts_query(address: str) -> tuple[str, str]:54    """Construit une requête FTS AND (n° civique + mots significatifs de la rue,55    sans le type de voie ni le n° d'appartement) + retourne le civique."""56    a = _norm(address).split(",")[0]57    a = re.sub(r"\b(?:app?t?|appartement|unite|suite|local|bureau|#)\s*[\w-]+", " ", a)58    a = re.sub(r"[^a-z0-9 ]+", " ", a)59    toks = [t for t in a.split() if t]60    civ = toks[0] if toks and toks[0].isdigit() else ""61    words = [t for t in toks if t != civ and t not in _VOIE and len(t) > 1]62    parts = ([civ] if civ else []) + words63    if not parts:64        return "", civ65    return " AND ".join(f'"{p}"' for p in parts), civ666768def available() -> bool:69    return os.path.exists(VP_DB)707172def _match(vp: sqlite3.Connection, address: str, city: str) -> dict | None:73    """Meilleure unité Vrai-Prix pour (adresse, ville)."""74    # n° civique + mots significatifs de la rue (AND), filtre municipalité75    q, civ = _fts_query(address)76    if not q:77        return None78    try:79        rows = vp.execute(80            "SELECT u.id_provinc, u.adresse, u.municipalite, u.lat, u.lng,"81            " u.est_hedo, u.est_2026, u.p10, u.p90"82            " FROM units_fts f JOIN units u ON u.rowid=f.rowid"83            " WHERE units_fts MATCH ? LIMIT 25", (q,)).fetchall()84    except sqlite3.OperationalError:85        return None86    if not rows:87        return None88    # VALIDATION STRICTE : on n'accepte QUE si même adresse — n° civique89    # identique + au moins un mot de rue en commun + même municipalité. Sinon90    # PAS de match (mieux vaut aucune estimation qu'une mauvaise).91    iwords = _street_words(_norm(address))92    nc = _norm(city).split()[0] if city else ""9394    def _accept(r) -> bool:95        ua = _norm(r["adresse"])96        m = re.match(r"\s*(\d+)", ua)97        if not civ or not m or m.group(1) != civ:      # même n° civique98            return False99        uwords = _street_words(ua)100        if iwords and not (iwords & uwords):            # mot de rue commun101            return False102        um = _norm(r["municipalite"] or "")103        if nc and um and nc not in um and um.split()[0] not in _norm(city):104            return False                                # même municipalité105        return True106107    best = next((r for r in rows if _accept(r)), None)108    if best is None:109        return None110    lat, lng = best["lat"], best["lng"]111    val = best["est_hedo"] or best["est_2026"]112    return {113        "id": best["id_provinc"], "lat": lat, "lng": lng,114        "value": val, "low": best["p10"], "high": best["p90"],115        "confidence": None, "confidence_pct": None,116        "url": f"{SITE}/estimation/{best['id_provinc']}",117    }118119120def run(limit: int | None = None, revalidate: bool = False) -> dict:121    """Apparie les annonces à la base Vrai-Prix : remplit lat/lng (si manquant)122    et l'estimation. `limit` borne le nombre d'annonces traitées.123    `revalidate=True` : re-vérifie AUSSI les annonces déjà appariées (corrige les124    mauvais merges après durcissement du matcher)."""125    if not available():126        print(f"[immo-ka] vraiprix_local: base absente ({VP_DB})")127        return {"error": "vraiprix.db absent"}128    vp = sqlite3.connect(f"file:{VP_DB}?mode=ro", uri=True)129    vp.row_factory = sqlite3.Row130    con = db.connect()131    where = ("active=1 AND dup_hidden=0 AND address<>''" if revalidate132             else "active=1 AND dup_hidden=0 AND address<>''"133                  " AND (vraiprix IS NULL OR vraiprix='{}')")134    rows = con.execute(135        f"SELECT uid, address, city, lat FROM listings WHERE {where}"136        " ORDER BY first_seen DESC").fetchall()137    if limit is not None:138        rows = rows[:limit]139140    matched = geoloc = miss = 0141    for i, r in enumerate(rows):142        m = _match(vp, r["address"], r["city"])143        try:144            if m and m["value"]:145                est = {k: m[k] for k in ("id", "value", "low", "high",146                                         "confidence", "confidence_pct", "url")}147                con.execute("UPDATE listings SET vraiprix=? WHERE uid=?",148                            (json.dumps(est, ensure_ascii=False), r["uid"]))149                # géocodage gratuit : coordonnées de l'unité si l'annonce n'en a pas150                if (r["lat"] is None and m["lat"] is not None151                        and _BBOX[0] <= m["lat"] <= _BBOX[1]152                        and _BBOX[2] <= m["lng"] <= _BBOX[3]):153                    con.execute("UPDATE listings SET lat=?, lng=? WHERE uid=?",154                                (m["lat"], m["lng"], r["uid"]))155                    geoloc += 1156                matched += 1157            else:158                con.execute("UPDATE listings SET vraiprix='{}' WHERE uid=?", (r["uid"],))159                miss += 1160            if i % 200 == 0:161                con.commit()162        except sqlite3.OperationalError:163            try:164                con.rollback()165            except sqlite3.Error:166                pass167            time.sleep(1.0)168    con.commit()169    con.close()170    vp.close()171    out = {"matched": matched, "geolocated": geoloc, "no_match": miss}172    print(f"[immo-ka] vraiprix_local {out}")173    return out174