Python 67%
TypeScript 18.2%
CSS 14.4%
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 d = {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 }99 d.update(_role_fields(r))100 return d101102103# clés « rôle d'évaluation » ajoutées au JSON vraiprix (valeurs officielles)104ROLE_KEYS = ("valeur_role", "valeur_terrain", "valeur_batiment",105 "annee_construction_role", "superficie_terrain_role_m2",106 "aire_etages_role_m2")107108109def _role_fields(r) -> dict:110 """Champs du rôle d'évaluation foncière de l'unité appariée (officiels) :111 valeurs (rôle/terrain/bâtiment), année de construction et superficies."""112 out = {}113 for src, dst in (("valeur_role", "valeur_role"),114 ("valeur_terrain", "valeur_terrain"),115 ("valeur_batiment", "valeur_batiment"),116 ("annee_construction", "annee_construction_role"),117 ("superficie_terrain_m2", "superficie_terrain_role_m2"),118 ("aire_etages_m2", "aire_etages_role_m2")):119 try:120 v = r[src]121 except (KeyError, IndexError):122 v = None123 if v:124 out[dst] = v125 return out126127128# le terrain du rôle d'une COPROPRIÉTÉ est souvent celui de l'immeuble entier :129# jamais de repli lot_sqft pour ces types130_NO_LOT_TYPES = ("condo", "appartement", "loft", "copropriete")131132133def _apply_role_fallback(con, uid: str, year_built, lot_sqft,134 property_type: str, est: dict) -> tuple[int, int]:135 """Repli des COLONNES depuis le rôle quand la source ne fournit rien :136 year_built ← annee_construction_role, lot_sqft ← superficie_terrain_role_m2137 (sauf copropriétés). Provenance marquée dans details.*_source='role'.138 Retourne (année_remplie, terrain_rempli) ∈ {0,1}²."""139 fy = fl = 0140 y = est.get("annee_construction_role")141 if year_built is None and y and 1600 <= int(y) <= 2049:142 con.execute(143 "UPDATE listings SET year_built=?,"144 " details=json_set(COALESCE(details,'{}'),'$.year_built_source','role')"145 " WHERE uid=? AND year_built IS NULL", (int(y), uid))146 fy = 1147 t = est.get("superficie_terrain_role_m2")148 pt = _norm(property_type or "")149 if (lot_sqft is None and t and float(t) > 0150 and not any(k in pt for k in _NO_LOT_TYPES)):151 con.execute(152 "UPDATE listings SET lot_sqft=?,"153 " details=json_set(COALESCE(details,'{}'),'$.lot_sqft_source','role')"154 " WHERE uid=? AND lot_sqft IS NULL",155 (round(float(t) * 10.7639), uid))156 fl = 1157 return fy, fl158159160def _meters(a1: float, o1: float, a2: float, o2: float) -> float:161 """Distance approx. en mètres (équirectangulaire, ~exact à courte portée)."""162 import math163 dlat = (a2 - a1) * 111_000.0164 dlng = (o2 - o1) * 111_000.0 * math.cos(math.radians(a1))165 return (dlat * dlat + dlng * dlng) ** 0.5166167168# mots génériques ignorés dans la comparaison de municipalités169_MUNI_GEN = {"saint", "sainte", "ville", "de", "du", "des", "la", "le", "les",170 "sur", "au", "aux", "lac", "notre", "dame", "canton", "cantons",171 "municipalite", "paroisse", "village", "mont"}172173174def _muni_norm(s: str) -> str:175 s = re.sub(r"[^a-z0-9 ]+", " ", _norm(s or ""))176 s = re.sub(r"\bst\b", "saint", s)177 s = re.sub(r"\bste\b", "sainte", s)178 return " ".join(s.split())179180181def _muni_one(nc: str, um: str) -> bool:182 if nc in um or um in nc:183 return True184 return bool((set(nc.split()) - _MUNI_GEN) & (set(um.split()) - _MUNI_GEN))185186187def _muni_match(city: str, unit_muni: str, address: str = "") -> bool:188 """Ville de l'annonce vs municipalité du rôle — tolère St/Ste, accents,189 arrondissement (« Gatineau Aylmer ») et la ville glissée dans l'adresse190 (« 1119 Ch. Dunant, Sainte-Anne-des-Lacs » avec city=« Laurentides »)."""191 nc = _muni_norm(city)192 if not nc:193 return True194 um = _muni_norm(unit_muni)195 if not um:196 return False197 cands = [nc] + [_muni_norm(p) for p in _norm(address).split(",")[1:]]198 return any(c and _muni_one(c, um) for c in cands)199200201def _concordance(address: str, city: str, r) -> tuple[bool, bool, bool]:202 """(civique_ok, rue_ok, municipalité_ok) entre l'annonce et l'unité."""203 civs, words = _addr_parts(address)204 ua = _norm(r["adresse"])205 civ_ok = False206 mr = re.match(r"\s*(\d+)\s*-\s*(\d+)", ua)207 ms = re.match(r"\s*(\d+)", ua)208 for c in civs:209 ci = int(c)210 if mr:211 lo, hi = int(mr.group(1)), int(mr.group(2))212 civ_ok = min(lo, hi) <= ci <= max(lo, hi)213 elif ms:214 civ_ok = ms.group(1) == c215 if civ_ok:216 break217 iwords = set(words)218 rue_ok = bool(iwords) and bool(iwords & _street_words(ua))219 muni_ok = _muni_match(city, r["municipalite"] or "", address)220 return civ_ok, rue_ok, muni_ok221222223_COLS = ("id_provinc, adresse, municipalite, lat, lng, est_hedo, est_2026, p10, p90, "224 "annee_construction, aire_etages_m2, superficie_terrain_m2, "225 "valeur_terrain, valeur_batiment, valeur_role")226227228def _match(vp: sqlite3.Connection, address: str, city: str,229 lat: float | None = None, lng: float | None = None) -> dict | None:230 """Apparie une annonce à une unité Vrai-Prix par :231 1) PROXIMITÉ SPATIALE (si lat/lng) — l'unité la plus proche, validée par232 l'adresse (text mining : civique/rue/municipalité) ;233 2) sinon recherche d'adresse FTS stricte.234 Ne retourne un match que s'il est fiable (mieux vaut rien qu'un faux)."""235 # 1) spatial + text-mining : candidats triés par distance236 if lat is not None and lng is not None:237 d = 0.0022 # ~±250 m238 try:239 cands = vp.execute(240 f"SELECT {_COLS} FROM units"241 " WHERE lat BETWEEN ? AND ? AND lng BETWEEN ? AND ?",242 (lat - d, lat + d, lng - d, lng + d)).fetchall()243 except sqlite3.OperationalError:244 cands = []245 cands = [r for r in cands if r["lat"] is not None]246 cands.sort(key=lambda r: _meters(lat, lng, r["lat"], r["lng"]))247 has_civ = bool(_addr_parts(address)[0])248 # ≤20 m du plus proche = même propriété (seulement si l'annonce a un249 # n° civique : un terrain géocodé près d'une maison ne doit pas hériter250 # de la valeur du voisin)251 if (cands and has_civ252 and _meters(lat, lng, cands[0]["lat"], cands[0]["lng"]) <= 20):253 return _pack(cands[0])254 for r in cands:255 dm = _meters(lat, lng, r["lat"], r["lng"])256 civ_ok, rue_ok, muni_ok = _concordance(address, city, r)257 # au-delà de 20 m : le CIVIQUE doit concorder (la « même rue » seule258 # apparie l'immeuble d'à côté → mauvais merge)259 if civ_ok and muni_ok and (dm <= 100 or rue_ok):260 return _pack(r)261262 # 2) repli : recherche d'adresse FTS stricte (avec relances)263 civs, words = _addr_parts(address)264 tries = []265 for c in (civs[::-1] or [""]): # civique(s), du plus probable au moins266 parts = ([c] if c else []) + words267 if parts:268 tries.append(" AND ".join(f'"{p}"' for p in parts))269 if civs and words:270 tries.append(" AND ".join(f'"{p}"' for p in words)) # rue seule (dernier recours)271 seen = set()272 for q in tries:273 if q in seen:274 continue275 seen.add(q)276 try:277 rows = vp.execute(278 f"SELECT u.{_COLS.replace(', ', ', u.')} FROM units_fts f"279 " JOIN units u ON u.rowid=f.rowid WHERE units_fts MATCH ? LIMIT 50",280 (q,)).fetchall()281 except sqlite3.OperationalError:282 return None283 for r in rows:284 civ_ok, rue_ok, muni_ok = _concordance(address, city, r)285 if civ_ok and rue_ok and muni_ok:286 return _pack(r)287 return None288289290def run(limit: int | None = None, revalidate: bool = False) -> dict:291 """Apparie les annonces à la base Vrai-Prix : remplit lat/lng (si manquant)292 et l'estimation. `limit` borne le nombre d'annonces traitées.293 `revalidate=True` : re-vérifie AUSSI les annonces déjà appariées (corrige les294 mauvais merges après durcissement du matcher)."""295 if not available():296 print(f"[immo-ka] vraiprix_local: base absente ({VP_DB})")297 return {"error": "vraiprix.db absent"}298 vp = sqlite3.connect(f"file:{VP_DB}?mode=ro", uri=True)299 vp.row_factory = sqlite3.Row300 con = db.connect()301 where = ("active=1 AND dup_hidden=0 AND address<>''" if revalidate302 else "active=1 AND dup_hidden=0 AND address<>''"303 " AND (vraiprix IS NULL OR vraiprix='{}')")304 rows = con.execute(305 f"SELECT uid, address, city, lat, lng, year_built, lot_sqft,"306 f" property_type FROM listings WHERE {where}"307 " ORDER BY first_seen DESC").fetchall()308 if limit is not None:309 rows = rows[:limit]310311 matched = geoloc = miss = 0312 for i, r in enumerate(rows):313 m = _match(vp, r["address"], r["city"], r["lat"], r["lng"])314 try:315 if m and m["value"]:316 est = {k: m[k] for k in ("id", "value", "low", "high",317 "confidence", "confidence_pct", "url")}318 est.update({k: m[k] for k in ROLE_KEYS if m.get(k)})319 con.execute("UPDATE listings SET vraiprix=? WHERE uid=?",320 (json.dumps(est, ensure_ascii=False), r["uid"]))321 # repli des colonnes depuis le rôle (année, terrain)322 _apply_role_fallback(con, r["uid"], r["year_built"],323 r["lot_sqft"], r["property_type"], est)324 # géocodage gratuit : coordonnées de l'unité si l'annonce n'en a pas325 if (r["lat"] is None and m["lat"] is not None326 and _BBOX[0] <= m["lat"] <= _BBOX[1]327 and _BBOX[2] <= m["lng"] <= _BBOX[3]):328 con.execute("UPDATE listings SET lat=?, lng=? WHERE uid=?",329 (m["lat"], m["lng"], r["uid"]))330 geoloc += 1331 matched += 1332 else:333 con.execute("UPDATE listings SET vraiprix='{}' WHERE uid=?", (r["uid"],))334 miss += 1335 if i % 200 == 0:336 con.commit()337 except sqlite3.OperationalError:338 try:339 con.rollback()340 except sqlite3.Error:341 pass342 time.sleep(1.0)343 con.commit()344 con.close()345 vp.close()346 out = {"matched": matched, "geolocated": geoloc, "no_match": miss}347 print(f"[immo-ka] vraiprix_local {out}")348 return out349350351def enrich_role(limit: int | None = None) -> dict:352 """Backfill du rôle d'évaluation pour les annonces DÉJÀ appariées dont le353 JSON vraiprix ne porte pas encore les champs officiels (valeur_role,354 annee_construction_role, superficies…). Aucun ré-appariement : jointure355 directe par id_provinc (clé primaire) — quasi gratuit. Applique aussi les356 replis de colonnes (year_built/lot_sqft) marqués `*_source='role'`."""357 if not available():358 return {"error": "vraiprix.db absent"}359 vp = sqlite3.connect(f"file:{VP_DB}?mode=ro", uri=True)360 vp.row_factory = sqlite3.Row361 con = db.connect()362 rows = con.execute(363 "SELECT uid, vraiprix, year_built, lot_sqft, property_type FROM listings"364 " WHERE active=1 AND vraiprix LIKE '%\"id\"%'"365 " AND vraiprix NOT LIKE '%valeur_role%'"366 " AND vraiprix NOT LIKE '%role_checked%'").fetchall()367 if limit is not None:368 rows = rows[:limit]369 enriched = fallback_y = fallback_l = 0370 for i, r in enumerate(rows):371 try:372 est = json.loads(r["vraiprix"]) or {}373 except (TypeError, ValueError):374 continue375 if not est.get("id"):376 continue377 u = vp.execute(378 "SELECT annee_construction, aire_etages_m2, superficie_terrain_m2,"379 " valeur_terrain, valeur_batiment, valeur_role"380 " FROM units WHERE id_provinc=?", (str(est["id"]),)).fetchone()381 try:382 if u is None:383 est["role_checked"] = 1 # unité disparue : ne pas repasser384 else:385 fields = _role_fields(u)386 if not fields:387 est["role_checked"] = 1 # unité sans données de rôle388 est.update(fields)389 fy, fl = _apply_role_fallback(con, r["uid"], r["year_built"],390 r["lot_sqft"], r["property_type"], est)391 fallback_y += fy392 fallback_l += fl393 con.execute("UPDATE listings SET vraiprix=? WHERE uid=?",394 (json.dumps(est, ensure_ascii=False), r["uid"]))395 enriched += 1396 if i % 500 == 0:397 con.commit()398 except sqlite3.OperationalError:399 try:400 con.rollback()401 except sqlite3.Error:402 pass403 time.sleep(1.0)404 con.commit()405 # réparation set-based (idempotente, instantanée) : ré-applique le repli de406 # colonnes depuis le JSON vraiprix déjà enrichi — filet de sécurité si une407 # mise à jour de source a écrasé year_built/lot_sqft (anciens process sans408 # COALESCE, edge cases de course entre writer et backfill).409 cur = con.execute(410 "UPDATE listings SET"411 " year_built=CAST(json_extract(vraiprix,'$.annee_construction_role') AS INT),"412 " details=json_set(COALESCE(details,'{}'),'$.year_built_source','role')"413 " WHERE active=1 AND year_built IS NULL"414 " AND CAST(json_extract(vraiprix,'$.annee_construction_role') AS INT)"415 " BETWEEN 1600 AND 2049")416 repaired_y = cur.rowcount417 like_condo = ("lower(property_type) LIKE '%condo%'"418 " OR lower(property_type) LIKE '%appartement%'"419 " OR lower(property_type) LIKE '%loft%'"420 " OR lower(property_type) LIKE '%copropri%'")421 cur = con.execute(422 "UPDATE listings SET"423 " lot_sqft=ROUND(CAST(json_extract(vraiprix,'$.superficie_terrain_role_m2')"424 " AS REAL) * 10.7639),"425 " details=json_set(COALESCE(details,'{}'),'$.lot_sqft_source','role')"426 " WHERE active=1 AND lot_sqft IS NULL"427 " AND CAST(json_extract(vraiprix,'$.superficie_terrain_role_m2') AS REAL) > 0"428 f" AND NOT ({like_condo})")429 repaired_l = cur.rowcount430 con.commit()431 con.close()432 vp.close()433 out = {"role_enriched": enriched, "year_built_filled": fallback_y,434 "lot_filled": fallback_l, "repaired_year": repaired_y,435 "repaired_lot": repaired_l}436 print(f"[immo-ka] vraiprix_local.enrich_role {out}")437 return out438