spb/ora-ka Public

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", "bd", "blvd",38         "ch", "chemin", "place", "pl", "rang", "rangs", "rg", "montee",39         "montée", "mtee", "cote", "côte", "route", "rte", "terrasse", "tsse",40         "ter", "impasse", "imp", "croissant", "crois", "croiss", "cours",41         "allee", "allée", "prom", "promenade", "carre", "aut", "autoroute",42         "de", "du", "des", "la", "le", "les", "l", "d", "et", "sur",43         "est", "ouest", "nord", "sud", "st", "ste", "saint", "sainte"}4445_APP_RE = r"\b(?:app?t?|appartement|unite|suite|local|bureau|#)\s*[\w-]+"464748def _street_words(norm_addr: str) -> set:49    """Mots significatifs de la rue (sans n° civique, type de voie, n° d'app.)."""50    a = re.sub(_APP_RE, " ", norm_addr.split(",")[0])51    a = re.sub(r"[^a-z0-9 ]+", " ", a)52    a = re.sub(r"^\s*\d+[a-z]{0,2}(?:\s+\d+)?\s+", " ", a)   # civique(s) en tête53    toks = [t for t in a.split() if t]54    return {t for t in toks if t not in _VOIE and len(t) > 1}555657def _addr_parts(address: str) -> tuple[list, list]:58    """→ (civiques candidats, mots de rue). Gère « 822Z » (suffixe de lettre),59    « 102 50 Rue X » (app-civique : les deux nombres sont candidats) et60    conserve les rues numériques (« Route 202 », « 117e Avenue »)."""61    a = _norm(address).split(",")[0]62    a = re.sub(_APP_RE, " ", a)63    a = re.sub(r"[^a-z0-9 ]+", " ", a)64    toks = [t for t in a.split() if t]65    civs = []66    while toks and len(civs) < 2:67        m = re.match(r"^(\d+)[a-z]{0,2}$", toks[0])68        if not m:69            break70        if civs and not toks[0].isdigit():   # ordinal de rue (2e, 3e…) : garder71            break72        civs.append(m.group(1))73        toks = toks[1:]74    words = [t for t in toks if t not in _VOIE and len(t) > 1]75    return civs, words767778def _fts_query(address: str) -> tuple[str, str]:79    """Requête FTS AND (n° civique + mots significatifs de la rue)."""80    civs, words = _addr_parts(address)81    civ = civs[-1] if civs else ""82    parts = ([civ] if civ else []) + words83    if not parts:84        return "", civ85    return " AND ".join(f'"{p}"' for p in parts), civ868788def available() -> bool:89    return os.path.exists(VP_DB)909192def _pack(r) -> dict:93    return {94        "id": r["id_provinc"], "lat": r["lat"], "lng": r["lng"],95        "value": r["est_hedo"] or r["est_2026"], "low": r["p10"], "high": r["p90"],96        "confidence": None, "confidence_pct": None,97        "url": f"{SITE}/estimation/{r['id_provinc']}",98    }99100101def _meters(a1: float, o1: float, a2: float, o2: float) -> float:102    """Distance approx. en mètres (équirectangulaire, ~exact à courte portée)."""103    import math104    dlat = (a2 - a1) * 111_000.0105    dlng = (o2 - o1) * 111_000.0 * math.cos(math.radians(a1))106    return (dlat * dlat + dlng * dlng) ** 0.5107108109# mots génériques ignorés dans la comparaison de municipalités110_MUNI_GEN = {"saint", "sainte", "ville", "de", "du", "des", "la", "le", "les",111             "sur", "au", "aux", "lac", "notre", "dame", "canton", "cantons",112             "municipalite", "paroisse", "village", "mont"}113114115def _muni_norm(s: str) -> str:116    s = re.sub(r"[^a-z0-9 ]+", " ", _norm(s or ""))117    s = re.sub(r"\bst\b", "saint", s)118    s = re.sub(r"\bste\b", "sainte", s)119    return " ".join(s.split())120121122def _muni_one(nc: str, um: str) -> bool:123    if nc in um or um in nc:124        return True125    return bool((set(nc.split()) - _MUNI_GEN) & (set(um.split()) - _MUNI_GEN))126127128def _muni_match(city: str, unit_muni: str, address: str = "") -> bool:129    """Ville de l'annonce vs municipalité du rôle — tolère St/Ste, accents,130    arrondissement (« Gatineau Aylmer ») et la ville glissée dans l'adresse131    (« 1119 Ch. Dunant, Sainte-Anne-des-Lacs » avec city=« Laurentides »)."""132    nc = _muni_norm(city)133    if not nc:134        return True135    um = _muni_norm(unit_muni)136    if not um:137        return False138    cands = [nc] + [_muni_norm(p) for p in _norm(address).split(",")[1:]]139    return any(c and _muni_one(c, um) for c in cands)140141142def _concordance(address: str, city: str, r) -> tuple[bool, bool, bool]:143    """(civique_ok, rue_ok, municipalité_ok) entre l'annonce et l'unité."""144    civs, words = _addr_parts(address)145    ua = _norm(r["adresse"])146    civ_ok = False147    mr = re.match(r"\s*(\d+)\s*-\s*(\d+)", ua)148    ms = re.match(r"\s*(\d+)", ua)149    for c in civs:150        ci = int(c)151        if mr:152            lo, hi = int(mr.group(1)), int(mr.group(2))153            civ_ok = min(lo, hi) <= ci <= max(lo, hi)154        elif ms:155            civ_ok = ms.group(1) == c156        if civ_ok:157            break158    iwords = set(words)159    rue_ok = bool(iwords) and bool(iwords & _street_words(ua))160    muni_ok = _muni_match(city, r["municipalite"] or "", address)161    return civ_ok, rue_ok, muni_ok162163164_COLS = ("id_provinc, adresse, municipalite, lat, lng, est_hedo, est_2026, p10, p90")165166167def _match(vp: sqlite3.Connection, address: str, city: str,168           lat: float | None = None, lng: float | None = None) -> dict | None:169    """Apparie une annonce à une unité Vrai-Prix par :170      1) PROXIMITÉ SPATIALE (si lat/lng) — l'unité la plus proche, validée par171         l'adresse (text mining : civique/rue/municipalité) ;172      2) sinon recherche d'adresse FTS stricte.173    Ne retourne un match que s'il est fiable (mieux vaut rien qu'un faux)."""174    # 1) spatial + text-mining : candidats triés par distance175    if lat is not None and lng is not None:176        d = 0.0022   # ~±250 m177        try:178            cands = vp.execute(179                f"SELECT {_COLS} FROM units"180                " WHERE lat BETWEEN ? AND ? AND lng BETWEEN ? AND ?",181                (lat - d, lat + d, lng - d, lng + d)).fetchall()182        except sqlite3.OperationalError:183            cands = []184        cands = [r for r in cands if r["lat"] is not None]185        cands.sort(key=lambda r: _meters(lat, lng, r["lat"], r["lng"]))186        has_civ = bool(_addr_parts(address)[0])187        # ≤20 m du plus proche = même propriété (seulement si l'annonce a un188        # n° civique : un terrain géocodé près d'une maison ne doit pas hériter189        # de la valeur du voisin)190        if (cands and has_civ191                and _meters(lat, lng, cands[0]["lat"], cands[0]["lng"]) <= 20):192            return _pack(cands[0])193        for r in cands:194            dm = _meters(lat, lng, r["lat"], r["lng"])195            civ_ok, rue_ok, muni_ok = _concordance(address, city, r)196            # au-delà de 20 m : le CIVIQUE doit concorder (la « même rue » seule197            # apparie l'immeuble d'à côté → mauvais merge)198            if civ_ok and muni_ok and (dm <= 100 or rue_ok):199                return _pack(r)200201    # 2) repli : recherche d'adresse FTS stricte (avec relances)202    civs, words = _addr_parts(address)203    tries = []204    for c in (civs[::-1] or [""]):           # civique(s), du plus probable au moins205        parts = ([c] if c else []) + words206        if parts:207            tries.append(" AND ".join(f'"{p}"' for p in parts))208    if civs and words:209        tries.append(" AND ".join(f'"{p}"' for p in words))   # rue seule (dernier recours)210    seen = set()211    for q in tries:212        if q in seen:213            continue214        seen.add(q)215        try:216            rows = vp.execute(217                f"SELECT u.{_COLS.replace(', ', ', u.')} FROM units_fts f"218                " JOIN units u ON u.rowid=f.rowid WHERE units_fts MATCH ? LIMIT 50",219                (q,)).fetchall()220        except sqlite3.OperationalError:221            return None222        for r in rows:223            civ_ok, rue_ok, muni_ok = _concordance(address, city, r)224            if civ_ok and rue_ok and muni_ok:225                return _pack(r)226    return None227228229def run(limit: int | None = None, revalidate: bool = False) -> dict:230    """Apparie les annonces à la base Vrai-Prix : remplit lat/lng (si manquant)231    et l'estimation. `limit` borne le nombre d'annonces traitées.232    `revalidate=True` : re-vérifie AUSSI les annonces déjà appariées (corrige les233    mauvais merges après durcissement du matcher)."""234    if not available():235        print(f"[immo-ka] vraiprix_local: base absente ({VP_DB})")236        return {"error": "vraiprix.db absent"}237    vp = sqlite3.connect(f"file:{VP_DB}?mode=ro", uri=True)238    vp.row_factory = sqlite3.Row239    con = db.connect()240    where = ("active=1 AND dup_hidden=0 AND address<>''" if revalidate241             else "active=1 AND dup_hidden=0 AND address<>''"242                  " AND (vraiprix IS NULL OR vraiprix='{}')")243    rows = con.execute(244        f"SELECT uid, address, city, lat, lng FROM listings WHERE {where}"245        " ORDER BY first_seen DESC").fetchall()246    if limit is not None:247        rows = rows[:limit]248249    matched = geoloc = miss = 0250    for i, r in enumerate(rows):251        m = _match(vp, r["address"], r["city"], r["lat"], r["lng"])252        try:253            if m and m["value"]:254                est = {k: m[k] for k in ("id", "value", "low", "high",255                                         "confidence", "confidence_pct", "url")}256                con.execute("UPDATE listings SET vraiprix=? WHERE uid=?",257                            (json.dumps(est, ensure_ascii=False), r["uid"]))258                # géocodage gratuit : coordonnées de l'unité si l'annonce n'en a pas259                if (r["lat"] is None and m["lat"] is not None260                        and _BBOX[0] <= m["lat"] <= _BBOX[1]261                        and _BBOX[2] <= m["lng"] <= _BBOX[3]):262                    con.execute("UPDATE listings SET lat=?, lng=? WHERE uid=?",263                                (m["lat"], m["lng"], r["uid"]))264                    geoloc += 1265                matched += 1266            else:267                con.execute("UPDATE listings SET vraiprix='{}' WHERE uid=?", (r["uid"],))268                miss += 1269            if i % 200 == 0:270                con.commit()271        except sqlite3.OperationalError:272            try:273                con.rollback()274            except sqlite3.Error:275                pass276            time.sleep(1.0)277    con.commit()278    con.close()279    vp.close()280    out = {"matched": matched, "geolocated": geoloc, "no_match": miss}281    print(f"[immo-ka] vraiprix_local {out}")282    return out283