# ----------------------------------------------------------------------------- # Rent-Ka — Rental listings aggregator (Canada, outside Québec) # Author: Simon-Pierre Boucher — contact@spboucher.ai # geocode.py : address geocoding via Nominatim (OpenStreetMap) # - persistent cache (geocode_cache table): an address is geocoded once, # per building (normalized address), not per listing; keys are prefixed # by the province code ("on|", "bc|", …) to avoid cross-province # collisions between homonymous addresses # - politeness: max 1 request/second, identifiable User-Agent # - validation: coordinates must fall inside the expected province # bounding box, otherwise geocode_failed is flagged — never bogus # coordinates # ----------------------------------------------------------------------------- from __future__ import annotations import re import time import requests from . import db from .normalize import strip_accents from .schema import CANADA_BBOX, PROVINCE_BBOX NOMINATIM_URL = "https://nominatim.openstreetmap.org/search" USER_AGENT = "RentKaBot/1.0 (Canadian rentals aggregator; +contact@spboucher.ai)" REQUEST_DELAY = 1.1 # Nominatim rule: max 1 req/s RETRY_FAILED_AFTER = 30 * 86400 # retry failures after 30 days PROVINCE_NAMES = { "ON": "Ontario", "BC": "British Columbia", "AB": "Alberta", "SK": "Saskatchewan", "MB": "Manitoba", "NB": "New Brunswick", "NS": "Nova Scotia", "PE": "Prince Edward Island", "NL": "Newfoundland and Labrador", "YT": "Yukon", "NT": "Northwest Territories", "NU": "Nunavut", "QC": "Québec", } def _bbox_for(province: str = "ON") -> tuple[float, float, float, float]: return PROVINCE_BBOX.get((province or "ON").upper(), CANADA_BBOX) def _in_bbox(lat: float, lng: float, bbox: tuple) -> bool: return bbox[0] <= lat <= bbox[1] and bbox[2] <= lng <= bbox[3] def norm_key(address: str) -> str: """Cache key: normalized address (case, accents, spaces, unit numbers).""" s = strip_accents((address or "").lower()) s = re.sub(r"\b(?:app?t?|appartement|unite|suite|#)\.?\s*[\w-]+\b", " ", s) s = re.sub(r"[^a-z0-9]+", " ", s) return re.sub(r"\s+", " ", s).strip() def cache_key(address: str, province: str = "ON") -> str: """Province-aware geocode_cache key: «on|…», «bc|…» — avoids collisions between homonymous addresses across provinces. The existing Ontario cache («on|» prefix, inherited from the Rent-Ka era) is preserved.""" k = norm_key(address) if not k: return k return f"{(province or 'ON').lower()}|{k}" class Geocoder: def __init__(self, con) -> None: self.con = con self.session = requests.Session() self.session.headers["User-Agent"] = USER_AGENT self._last = 0.0 @staticmethod def _clean(address: str) -> str: """Cleanups that help Nominatim on Canadian addresses.""" s = address.strip() # «14340, Main St» (comma after the civic number, French style) s = re.sub(r"^(\d+[A-Za-z]?),\s+", r"\1 ", s) # «101-2905 Main St» = unit 101, civic 2905 -> keep the civic s = re.sub(r"^(\d+)-(\d+)\s", r"\2 ", s) # «6275 et 6375 boulevard X» / «6275 and 6375 X» -> first civic s = re.sub(r"^(\d+)\s+(?:et|and)\s+\d+\s", r"\1 ", s) # «bureau 105» / «suite 3» / «unit B»: office/unit suffixes s = re.sub(r",?\s*(?:bureau|suite|local|unit|app\.?|apt\.?)\s*[\w-]+\b", "", s, flags=re.I) return s def _build_query(self, address: str, city: str, province: str = "ON") -> str: q = self._clean(address) prov = (province or "ON").upper() key = strip_accents(q.lower()) if city and strip_accents(city.lower()) not in key: q += f", {city}" prov_name = PROVINCE_NAMES.get(prov, "") if prov_name and strip_accents(prov_name.lower()) not in key \ and not re.search(rf"\b{prov}\b", q): q += f", {prov_name}" return q + ", Canada" def _query_nominatim(self, params: dict) -> tuple[float, float] | None: """One throttled Nominatim request.""" wait = REQUEST_DELAY - (time.time() - self._last) if wait > 0: time.sleep(wait) try: resp = self.session.get(NOMINATIM_URL, params={ "format": "jsonv2", "limit": 1, "countrycodes": "ca", **params, }, timeout=20) self._last = time.time() resp.raise_for_status() hits = resp.json() except Exception: self._last = time.time() return None if not hits: return None try: return float(hits[0]["lat"]), float(hits[0]["lon"]) except (KeyError, ValueError): return None def _attempts(self, address: str, city: str, province: str = "ON") -> list[dict]: """Query strategies, most precise first. 1) structured civic+street 2) full free-text search 3) structured street only -> street centroid (acceptable map fallback when the civic number is missing from OpenStreetMap) """ prov = (province or "ON").upper() premier = self._clean(address).split(",")[0].strip() ville = (city or "").strip() commun = {"state": PROVINCE_NAMES.get(prov, ""), "country": "Canada"} if ville: commun["city"] = ville tries: list[dict] = [] m = re.match(r"^(\d+)[,\s]+(.{4,})$", premier) if m: tries.append({"street": f"{m.group(1)} {m.group(2)}", **commun}) tries.append({"q": self._build_query(address, city, prov)}) if m: tries.append({"street": m.group(2), **commun}) return tries def resolve(self, address: str, city: str, province: str = "ON") -> tuple[float, float] | None: """Address -> (lat, lng), through the cache then Nominatim.""" prov = (province or "ON").upper() nk = norm_key(address) if not nk or len(nk) < 6: return None key = cache_key(address, prov) row = self.con.execute( "SELECT lat, lng, failed, ts FROM geocode_cache WHERE address=?", (key,)).fetchone() if row is not None: if not row["failed"]: return (row["lat"], row["lng"]) if time.time() - (row["ts"] or 0) < RETRY_FAILED_AFTER: return None # recent failure: don't hammer the API bbox = _bbox_for(prov) coords = None for params in self._attempts(address, city, prov): c = self._query_nominatim(params) if c is not None and _in_bbox(*c, bbox): coords = c break ok = coords is not None self.con.execute( "INSERT INTO geocode_cache (address, lat, lng, provider, failed, ts)" " VALUES (?,?,?,?,?,?)" " ON CONFLICT(address) DO UPDATE SET lat=excluded.lat," " lng=excluded.lng, provider=excluded.provider," " failed=excluded.failed, ts=excluded.ts", (key, coords[0] if ok else None, coords[1] if ok else None, "nominatim", 0 if ok else 1, time.time())) self.con.commit() return coords if ok else None def run(limit: int | None = None) -> dict: """Geocode active listings without coordinates, per unique address. `limit` caps the number of NEW Nominatim requests (cache hits are free and always applied). """ con = db.connect() geo = Geocoder(con) rows = con.execute( """SELECT uid, address, city, province FROM listings WHERE active=1 AND lat IS NULL AND address<>'' AND geocode_failed=0 ORDER BY address""").fetchall() # group per building (normalized address, province-prefixed) groupes: dict[str, list] = {} for r in rows: groupes.setdefault(cache_key(r["address"], r["province"]), []).append(r) done = failed = requests_made = 0 for key, members in groupes.items(): if not key: continue cached = con.execute( "SELECT failed FROM geocode_cache WHERE address=?", (key,)).fetchone() if cached is None: if limit is not None and requests_made >= limit: continue requests_made += 1 coords = geo.resolve(members[0]["address"], members[0]["city"], members[0]["province"]) if coords: for r in members: con.execute("UPDATE listings SET lat=?, lng=? WHERE uid=?", (coords[0], coords[1], r["uid"])) done += len(members) else: # not found/out of zone: flag for review, never bogus coordinates for r in members: con.execute("UPDATE listings SET geocode_failed=1 WHERE uid=?", (r["uid"],)) failed += len(members) con.commit() con.close() stats = {"geocoded": done, "failed": failed, "unique_addresses": len(groupes), "api_requests": requests_made} print(f"[rent-ka] geocode {stats}") return stats def run_batch(limit: int | None = None) -> dict: """Batch entry point kept for the CLI/watch loop. The Rent-Ka era used the Adresses Québec bulk endpoint (Québec-only dataset); outside Québec everything goes through unit Nominatim resolution — this simply forwards to run() with the same signature. """ return run(limit)