# ----------------------------------------------------------------------------- # Lou-Ka — Agrégateur de logements à louer (province de Québec) # Auteur : Simon-Pierre Boucher — contact@spboucher.ai # quartier.py : statistiques de quartier par annonce (à la Centris, en libre) # Base statique data/quartier.db construite par scripts/build_*.py : # - da_poly / da_stats : aires de diffusion 2021 + profil du recensement # - da_pmd : mesures de proximité StatCan (scores 0..1) # - da_defav : défavorisation matérielle/sociale INSPQ (quintiles) # - heat : classe d'îlot de chaleur/fraîcheur INSPQ par immeuble # - crime_mtl / igc : actes criminels SPVM (points) + indice de gravité # Jointure : lat/lng -> DAUID par point-dans-polygone local (préfiltre bbox), # mémorisée dans listings.dauid à l'enrichissement (boucle watch). # ----------------------------------------------------------------------------- from __future__ import annotations import json import math import sqlite3 import time from pathlib import Path from . import db QUARTIER_DB = Path(__file__).resolve().parent.parent / "data" / "quartier.db" # villes couvertes par les points SPVM (agglomération de Montréal) _VILLES_SPVM = {"montreal", "montreal-est", "montreal-ouest", "westmount", "cote saint-luc", "cote-saint-luc", "hampstead", "mont-royal", "outremont", "verdun", "lasalle", "lachine", "anjou", "saint-leonard", "saint-laurent", "ahuntsic", "dorval", "pointe-claire", "kirkland", "beaconsfield", "dollard-des-ormeaux"} # correspondance ville -> fragment du nom de service dans la table igc _IGC_SERVICE = { "quebec": "SPVQ", "levis": "Lévis", "montreal": "SPVM", "laval": "Laval", "longueuil": "Longueuil", } def disponible() -> bool: return QUARTIER_DB.exists() def _connect() -> sqlite3.Connection: con = sqlite3.connect(f"file:{QUARTIER_DB}?mode=ro", uri=True) con.row_factory = sqlite3.Row return con # --------------------------------------------------------------------------- # lat/lng -> DAUID (point dans polygone, préfiltre bbox) # --------------------------------------------------------------------------- def _dans_anneau(lat: float, lng: float, anneau: list) -> bool: """Lancer de rayon (even-odd). anneau = [[lng, lat], ...].""" dedans = False n = len(anneau) j = n - 1 for i in range(n): xi, yi = anneau[i][0], anneau[i][1] xj, yj = anneau[j][0], anneau[j][1] if (yi > lat) != (yj > lat) and \ lng < (xj - xi) * (lat - yi) / (yj - yi + 1e-12) + xi: dedans = not dedans j = i return dedans def dauid_for(qcon: sqlite3.Connection, lat: float, lng: float) -> str | None: rows = qcon.execute( "SELECT dauid, poly FROM da_poly WHERE lat_min<=? AND lat_max>=?" " AND lng_min<=? AND lng_max>=?", (lat, lat, lng, lng)).fetchall() for r in rows: anneaux = json.loads(r["poly"]) # even-odd sur tous les anneaux (les trous annulent) compte = sum(1 for a in anneaux if _dans_anneau(lat, lng, a)) if compte % 2 == 1: return r["dauid"] return None # --------------------------------------------------------------------------- # Assemblage pour la fiche # --------------------------------------------------------------------------- def _cle_ville(city: str) -> str: import unicodedata s = "".join(c for c in unicodedata.normalize("NFD", city or "") if unicodedata.category(c) != "Mn") return s.strip().lower() def _crime_mtl(qcon: sqlite3.Connection, lat: float, lng: float) -> dict | None: """Comptage des actes criminels SPVM à < 500 m : 12 mois vs 12 précédents.""" dlat = 500 / 111000.0 dlng = 500 / (111000.0 * max(0.2, math.cos(math.radians(lat)))) now = time.time() rows = qcon.execute( "SELECT lat, lng, ts FROM crime_mtl WHERE lat BETWEEN ? AND ?" " AND lng BETWEEN ? AND ? AND ts >= ?", (lat - dlat, lat + dlat, lng - dlng, lng + dlng, now - 730 * 86400)).fetchall() recent = avant = 0 for r in rows: # distance exacte (le bbox est un carré) d = math.hypot((r["lat"] - lat) * 111000.0, (r["lng"] - lng) * 111000.0 * math.cos(math.radians(lat))) if d > 500: continue if r["ts"] >= now - 365 * 86400: recent += 1 else: avant += 1 if recent == 0 and avant == 0: return None return {"type": "points", "rayon_m": 500, "douze_mois": recent, "douze_mois_precedents": avant} def _crime_igc(qcon: sqlite3.Connection, city: str) -> dict | None: service = _IGC_SERVICE.get(_cle_ville(city)) if not service: return None row = qcon.execute( "SELECT annee, indice FROM igc WHERE service LIKE '%' || ? || '%'" " ORDER BY annee DESC LIMIT 1", (service,)).fetchone() if row is None or row["indice"] is None: return None ref = qcon.execute( "SELECT indice FROM igc WHERE service LIKE '%canada%' AND annee=?", (row["annee"],)).fetchone() return {"type": "igc", "ville": city, "annee": row["annee"], "indice": round(row["indice"], 1), "indice_canada": round(ref["indice"], 1) if ref and ref["indice"] else None} def fiche_quartier(lat: float | None, lng: float | None, city: str, dauid: str | None = None) -> dict | None: """Bloc « Le quartier » d'une fiche. None si données indisponibles.""" if not disponible() or lat is None or lng is None: return None qcon = _connect() try: if not dauid: dauid = dauid_for(qcon, lat, lng) out: dict = {"dauid": dauid} if dauid: r = qcon.execute("SELECT * FROM da_stats WHERE dauid=?", (dauid,)).fetchone() if r: out["demographie"] = {k: r[k] for k in ("population", "densite", "age_median", "revenu_median", "pct_locataires", "loyer_moyen", "pct_francais", "pct_univ")} # rangs centiles québécois (0-100) — voir scripts/merge_quartier.py r = qcon.execute("SELECT * FROM da_pmd_pct WHERE dauid=?", (dauid,)).fetchone() if r: out["proximite"] = {k: r[k] / 100.0 for k in r.keys() if k != "dauid" and r[k] is not None} r = qcon.execute("SELECT quintile_materiel, quintile_social FROM da_defav" " WHERE dauid=?", (dauid,)).fetchone() if r: out["defavorisation"] = dict(r) # îlot de chaleur : coordonnée exacte, sinon la plus proche (~120 m) key = f"{round(lat, 4)},{round(lng, 4)}" r = qcon.execute("SELECT classe, ecart FROM heat WHERE coord_key=?", (key,)).fetchone() if r is None: r = qcon.execute( "SELECT classe, ecart FROM heat WHERE coord_key LIKE ?" " AND classe IS NOT NULL LIMIT 1", (f"{round(lat, 3)}%",)).fetchone() if r and r["classe"] is not None: out["chaleur"] = {"classe": r["classe"], "ecart": r["ecart"]} # criminalité : points SPVM sur l'île, indice IGC ailleurs crime = None if _cle_ville(city) in _VILLES_SPVM: crime = _crime_mtl(qcon, lat, lng) if crime is None: crime = _crime_igc(qcon, city) if crime: out["crime"] = crime return out if len(out) > 1 else None except sqlite3.Error: return None finally: qcon.close() # --------------------------------------------------------------------------- # Enrichissement : mémoriser le DAUID de chaque annonce (boucle watch) # --------------------------------------------------------------------------- def enrich(limit: int | None = None) -> dict: """Remplit listings.dauid pour les annonces géolocalisées qui ne l'ont pas.""" if not disponible(): print("[lou-ka] quartier: data/quartier.db absent — étape sautée") return {"enriched": 0, "missing_db": True} con = db.connect() qcon = _connect() rows = con.execute( "SELECT uid, lat, lng FROM listings WHERE active=1 AND lat IS NOT NULL" " AND (dauid IS NULL OR dauid='')").fetchall() if limit is not None: rows = rows[:limit] done = introuvable = 0 for r in rows: d = dauid_for(qcon, r["lat"], r["lng"]) con.execute("UPDATE listings SET dauid=? WHERE uid=?", (d or "hors-zone", r["uid"])) if d: done += 1 else: introuvable += 1 con.commit() qcon.close() con.close() stats = {"enriched": done, "hors_zone": introuvable, "candidats": len(rows)} print(f"[lou-ka] quartier {stats}") return stats