# ----------------------------------------------------------------------------- # Lou-Ka — Agrégateur de logements à louer (province de Québec) # Auteur : Simon-Pierre Boucher — contact@spboucher.ai # rdl.py : Registre des loyers (registre-des-loyers.ca) — extraction + requêtes # # Le Registre des loyers est une initiative citoyenne (Vivre en ville) où # les locataires déclarent volontairement leur loyer. La carte publique # expose une API JSON ouverte : # GET /api/v1/housings/{NElat},{NElng},{SWlat},{SWlng} # On balaie la province par grandes boîtes (aucune pagination côté serveur ; # la boîte « tout le Québec » dépasse ses capacités -> découpage, avec # subdivision récursive en cas d'erreur 500), puis on stocke le tout dans # data/rdl.db (base séparée : jamais de verrou sur louka.db). # # Usage : python run.py rdl # rafraîchît la base locale # Lecture : nearby(lat, lng, radius_m) # loyers déclarés autour d'un point # (bloc « Registre des loyers » de la fiche, /api/rdl) # ----------------------------------------------------------------------------- from __future__ import annotations import json import math import sqlite3 import time import urllib.request from pathlib import Path from statistics import median DB_PATH = Path(__file__).resolve().parent.parent / "data" / "rdl.db" API = "https://registre-des-loyers.ca/api/v1/housings/{},{},{},{}" UA = "LouKaBot/1.0 (+https://www.lou-ka.com; contact@spboucher.ai)" DELAY = 2.0 # politesse entre deux boîtes # Boîtes (NE_lat, NE_lng, SW_lat, SW_lng) couvrant le pays entier (le registre # est pancanadien ; Lou-Ka n'affiche que ce qui tombe près d'une annonce). # Elles se recoupent légèrement, la déduplication se fait par id. BOXES: list[tuple[float, float, float, float]] = [ (46.0, -70.0, 45.0, -75.0), # Grand Montréal + Montérégie + Estrie (47.2, -70.0, 46.0, -76.0), # Québec, Mauricie, Centre-du-Québec (46.2, -74.8, 45.2, -77.5), # Outaouais (49.5, -63.5, 47.2, -80.0), # Abitibi, Saguenay, Gaspésie (47.2, -63.5, 45.0, -70.0), # Bas-Saint-Laurent, Beauce est (63.0, -55.0, 49.5, -80.0), # Côte-Nord, Nord-du-Québec (45.05, -70.0, 44.5, -80.0), # frange frontalière sud (57.0, -74.5, 41.6, -96.0), # Ontario (60.0, -96.0, 48.9, -110.0), # Prairies (MB, SK) (60.0, -110.0, 48.0, -140.0), # Alberta, Colombie-Britannique (70.0, -60.0, 60.0, -142.0), # territoires + Nunavik (49.0, -52.0, 43.0, -70.0), # Atlantique (NB, NÉ, ÎPÉ, TNL sud) (61.0, -52.0, 49.0, -57.0), # Terre-Neuve nord + Labrador est ] # au-delà de ce volume on subdivise par prudence (plafond serveur inconnu) SUSPECT = 40000 _SCHEMA = """ CREATE TABLE IF NOT EXISTS rdl_housings ( id INTEGER PRIMARY KEY, full_address TEXT, street_number TEXT, apartment_number TEXT, street_name TEXT, city TEXT, zip TEXT, lat REAL, lng REAL, price REAL, rooms INTEGER, year INTEGER, start_date TEXT, type_of_accomodation TEXT, heating_included INTEGER, electricity_included INTEGER, furnishing_included INTEGER, parking_included INTEGER, animal_allowed INTEGER, address_slug TEXT, updated_at TEXT, fetched_at TEXT ); CREATE INDEX IF NOT EXISTS idx_rdl_latlng ON rdl_housings (lat, lng); """ def _connect(ro: bool = False) -> sqlite3.Connection: if ro: con = sqlite3.connect(f"file:{DB_PATH}?mode=ro", uri=True) else: DB_PATH.parent.mkdir(parents=True, exist_ok=True) con = sqlite3.connect(DB_PATH) con.row_factory = sqlite3.Row return con def _fetch_box(ne_lat: float, ne_lng: float, sw_lat: float, sw_lng: float, depth: int = 0) -> list[dict]: """Une boîte ; en cas d'erreur serveur (boîte trop lourde), découpe en 4.""" url = API.format(ne_lat, ne_lng, sw_lat, sw_lng) req = urllib.request.Request(url, headers={"User-Agent": UA}) try: with urllib.request.urlopen(req, timeout=120) as resp: data = json.load(resp) rows = data.get("data", {}).get("housings") or [] if len(rows) < SUSPECT or depth >= 3: return rows raise RuntimeError(f"{len(rows)} résultats — subdivision de prudence") except Exception as exc: if depth >= 3: print(f" ! abandon boîte {url}: {exc}") return [] mid_lat = (ne_lat + sw_lat) / 2 mid_lng = (ne_lng + sw_lng) / 2 out: list[dict] = [] for quad in ((ne_lat, ne_lng, mid_lat, mid_lng), (ne_lat, mid_lng, mid_lat, sw_lng), (mid_lat, ne_lng, sw_lat, mid_lng), (mid_lat, mid_lng, sw_lat, sw_lng)): time.sleep(DELAY) out.extend(_fetch_box(*quad, depth=depth + 1)) return out def _num(v, cast=float): try: return cast(v) except (TypeError, ValueError): return None def refresh() -> None: """Balaie la province et remplace le contenu de data/rdl.db.""" now = time.strftime("%Y-%m-%d %H:%M:%S") seen: dict[int, dict] = {} for i, box in enumerate(BOXES, 1): rows = _fetch_box(*box) fresh = 0 for h in rows: hid = _num(h.get("id"), int) if hid is None or hid in seen: continue seen[hid] = h fresh += 1 print(f"[rdl] boîte {i}/{len(BOXES)} : {len(rows)} reçus, " f"{fresh} nouveaux ({len(seen)} au total)") time.sleep(DELAY) con = _connect() con.executescript(_SCHEMA) with con: con.execute("DELETE FROM rdl_housings") con.executemany( "INSERT OR REPLACE INTO rdl_housings VALUES " "(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)", [(hid, h.get("full_address"), h.get("street_number"), h.get("apartment_number"), h.get("street_name"), h.get("city"), h.get("zip"), _num(h.get("latitude")), _num(h.get("longitude")), _num(h.get("price")), _num(h.get("number_of_closed_room"), int), _num(h.get("year"), int), (h.get("start_date") or "")[:10] or None, h.get("type_of_accomodation"), 1 if h.get("heating_included") else 0, 1 if h.get("electricity_included") else 0, 1 if h.get("furnishing_included") else 0, 1 if h.get("parking_included") else 0, 1 if h.get("animal_allowed") else 0, h.get("address_slug"), h.get("updated_at"), now) for hid, h in seen.items()]) n, cities = con.execute( "SELECT COUNT(*), COUNT(DISTINCT city) FROM rdl_housings").fetchone() print(f"[rdl] terminé : {n} loyers déclarés, {cities} villes -> {DB_PATH}") con.close() def _dist_m(lat1: float, lng1: float, lat2: float, lng2: float) -> float: dlat = math.radians(lat2 - lat1) dlng = math.radians(lng2 - lng1) a = (math.sin(dlat / 2) ** 2 + math.cos(math.radians(lat1)) * math.cos(math.radians(lat2)) * math.sin(dlng / 2) ** 2) return 6371000 * 2 * math.asin(math.sqrt(a)) def nearby(lat: float, lng: float, radius_m: int = 600, limit: int = 12) -> dict | None: """Loyers déclarés autour d'un point + agrégats (bloc fiche / API).""" if not DB_PATH.exists(): return None dlat = radius_m / 111320.0 dlng = radius_m / (111320.0 * max(0.2, math.cos(math.radians(lat)))) con = _connect(ro=True) rows = con.execute( "SELECT * FROM rdl_housings WHERE lat BETWEEN ? AND ? " "AND lng BETWEEN ? AND ? AND price > 100 AND price < 20000", (lat - dlat, lat + dlat, lng - dlng, lng + dlng)).fetchall() con.close() hits = [] for r in rows: if r["lat"] is None or r["lng"] is None: continue d = _dist_m(lat, lng, r["lat"], r["lng"]) if d <= radius_m: hits.append((d, r)) if not hits: return {"n": 0, "radius_m": radius_m, "items": []} hits.sort(key=lambda t: t[0]) def _q(vals: list[float], q: float) -> int: s = sorted(vals) i = q * (len(s) - 1) lo = int(i) hi = min(lo + 1, len(s) - 1) return round(s[lo] + (s[hi] - s[lo]) * (i - lo)) prices = [r["price"] for _, r in hits] recent = [r["price"] for _, r in hits if (r["year"] or 0) >= 2023] base = recent if len(recent) >= 8 else prices quart = ({"p10": _q(base, 0.10), "p25": _q(base, 0.25), "p75": _q(base, 0.75), "p90": _q(base, 0.90)} if len(base) >= 5 else None) by_rooms: dict[str, dict] = {} for _, r in hits: if r["rooms"] is None: continue b = by_rooms.setdefault(str(r["rooms"]), {"n": 0, "prices": []}) b["n"] += 1 b["prices"].append(r["price"]) for b in by_rooms.values(): b["median"] = round(median(b.pop("prices"))) def item(d: float, r: sqlite3.Row) -> dict: addr = " ".join(x for x in (r["street_number"], r["street_name"]) if x) if r["apartment_number"]: addr += f", app. {r['apartment_number']}" return {"address": addr or r["full_address"], "city": r["city"], "price": r["price"], "rooms": r["rooms"], "year": r["year"], "date": r["start_date"], "dist_m": round(d), "heating": bool(r["heating_included"]), "furnished": bool(r["furnishing_included"])} return { "n": len(hits), "radius_m": radius_m, "median": round(median(prices)), "median_recent": round(median(recent)) if recent else None, "n_recent": len(recent), "quartiles": quart, "by_rooms": by_rooms, "items": [item(d, r) for d, r in hits[:limit]], }