# ----------------------------------------------------------------------------- # House-Ka — Agrégateur de maisons à vendre (Canada hors Québec) # Auteur : Simon-Pierre Boucher — contact@spboucher.ai # commerces.py : grands commerces à proximité — API Mapbox Search Box, # bannières ADAPTÉES À CHAQUE PROVINCE. # # Pour chaque grande bannière du panier provincial, on interroge l'API # Search Box de Mapbox (jeton PUBLIC pk.… lu dans # frontend/src/kamaps/config.ts) avec la position de l'annonce en # `proximity`, et on retient le point de vente le plus proche. Cache par # cellule d'environ 1 km (data/commerces.db, TTL 30 jours). # # Paniers par province (2026-08) — fondés sur la présence réelle des # chaînes : Loblaw n°1 national (Superstore/No Frills, Zehrs ON, Dominion NL), # Sobeys n°2 (dominant en Atlantique : Sobeys/Foodland/Lawtons), Metro n°3 # (Ontario+Québec, Food Basics), Pattison/Save-On-Foods ~180 magasins dans # l'Ouest + Yukon, Co-op (FCL) ~300 magasins en Saskatchewan, monopoles # d'alcool provinciaux (LCBO, BC Liquor, Liquor Mart MB, NB Liquor, NSLC), # Kent (Irving) en quincaillerie atlantique, London Drugs (BC/AB), # Colemans (chaîne terre-neuvienne). # ----------------------------------------------------------------------------- from __future__ import annotations import json import math import re import sqlite3 import time import urllib.parse import urllib.request from concurrent.futures import ThreadPoolExecutor from pathlib import Path ROOT = Path(__file__).resolve().parent.parent DB_PATH = ROOT / "data" / "commerces.db" UA = "HouseKaBot/1.0 (+https://www.house-ka.com; contact@spboucher.ai)" TTL = 30 * 86400 API = "https://api.mapbox.com/search/searchbox/v1/forward" # id -> (libellé, requête Mapbox, mot-clé de validation — le nom du POI doit le # contenir, sans accents ni casse ; écarte « Station Métro », « Super Qualité »…) BRAND_DEFS = { # nationaux "costco": ("Costco", "Costco Wholesale", "costco"), "walmart": ("Walmart", "Walmart Supercentre", "walmart"), "canadiantire": ("Canadian Tire", "Canadian Tire", "canadian tire"), "dollarama": ("Dollarama", "Dollarama", "dollarama"), "homedepot": ("Home Depot", "Home Depot", "home depot"), "homehardware": ("Home Hardware", "Home Hardware", "home hardware"), "shoppers": ("Shoppers Drug Mart", "Shoppers Drug Mart", "shoppers"), "gianttiger": ("Giant Tiger", "Giant Tiger", "giant tiger"), # épicerie — Loblaw "loblaws": ("Loblaws", "Loblaws", "loblaws"), "superstore": ("Real Canadian Superstore", "Real Canadian Superstore", "superstore"), "atlanticsuperstore": ("Atlantic Superstore", "Atlantic Superstore", "superstore"), "nofrills": ("No Frills", "No Frills", "no frills"), "zehrs": ("Zehrs", "Zehrs", "zehrs"), "dominion": ("Dominion", "Dominion grocery", "dominion"), "independent": ("Your Independent Grocer", "Your Independent Grocer", "independent"), # épicerie — Empire/Sobeys "sobeys": ("Sobeys", "Sobeys", "sobeys"), "safeway": ("Safeway", "Safeway", "safeway"), "foodland": ("Foodland", "Foodland", "foodland"), "freshco": ("FreshCo", "FreshCo", "freshco"), # épicerie — Metro (Ontario) "metro": ("Metro", "Metro grocery", "metro"), "foodbasics": ("Food Basics", "Food Basics", "food basics"), # épicerie — Ouest / coopératives / régionales "saveon": ("Save-On-Foods", "Save-On-Foods", "save-on"), "coop": ("Co-op", "Co-op Food Store", "co-op"), "colemans": ("Colemans", "Colemans grocery", "colemans"), "iga": ("IGA", "IGA", "iga"), # alcool (monopoles/sociétés provinciales) "lcbo": ("LCBO", "LCBO", "lcbo"), "beerstore": ("The Beer Store", "The Beer Store", "beer store"), "bcliquor": ("BC Liquor", "BC Liquor Store", "liquor"), "liquormart": ("Liquor Mart", "Manitoba Liquor Mart", "liquor mart"), "nbliquor": ("NB Liquor", "NB Liquor Alcool NB", "liquor"), "nslc": ("NSLC", "NSLC", "nslc"), # pharmacies régionales "londondrugs": ("London Drugs", "London Drugs", "london drugs"), "rexall": ("Rexall", "Rexall", "rexall"), "lawtons": ("Lawtons", "Lawtons Drugs", "lawtons"), # quincaillerie "rona": ("RONA", "RONA", "rona"), "kent": ("Kent", "Kent Building Supplies", "kent"), } _NATIONAL = ["costco", "walmart", "canadiantire", "dollarama", "homedepot", "shoppers"] # panier par province — les bannières LES PLUS PRÉSENTES dans chaque marché PROVINCE_BRANDS = { "British Columbia": _NATIONAL + ["saveon", "superstore", "safeway", "nofrills", "iga", "londondrugs", "bcliquor", "homehardware"], "Alberta": _NATIONAL + ["superstore", "safeway", "saveon", "sobeys", "nofrills", "coop", "rexall", "homehardware"], "Saskatchewan": _NATIONAL + ["coop", "superstore", "sobeys", "safeway", "nofrills", "gianttiger", "homehardware"], "Manitoba": _NATIONAL + ["superstore", "coop", "sobeys", "safeway", "nofrills", "liquormart", "gianttiger", "homehardware"], "Ontario": _NATIONAL + ["loblaws", "nofrills", "foodbasics", "metro", "sobeys", "freshco", "zehrs", "lcbo", "beerstore", "rexall", "rona"], "New Brunswick": _NATIONAL + ["sobeys", "atlanticsuperstore", "foodland", "nofrills", "nbliquor", "kent", "lawtons", "gianttiger"], "Nova Scotia": _NATIONAL + ["sobeys", "atlanticsuperstore", "foodland", "nofrills", "nslc", "kent", "lawtons", "gianttiger"], "Prince Edward Island": _NATIONAL + ["sobeys", "atlanticsuperstore", "foodland", "kent", "lawtons", "gianttiger"], "Newfoundland and Labrador": _NATIONAL + ["sobeys", "dominion", "colemans", "foodland", "kent", "lawtons"], "Yukon": _NATIONAL + ["saveon", "superstore", "independent", "homehardware"], "Northwest Territories": _NATIONAL + ["independent", "coop", "homehardware"], "Nunavut": ["canadiantire", "independent", "coop", "homehardware"], } _DEFAULT_PROVINCE = "Ontario" # frontières longitudinales approximatives (repli quand la région est absente) def _infer_province(lat: float, lng: float) -> str: if lat >= 60: if lng < -124: return "Yukon" return "Northwest Territories" if lng < -102 else "Nunavut" if lng < -120: return "British Columbia" if lng < -110: return "Alberta" if lng < -101.4: return "Saskatchewan" if lng < -95.15: return "Manitoba" if lng < -74.3: return "Ontario" if -64.5 <= lng <= -61.9 and 45.9 <= lat <= 47.1: return "Prince Edward Island" if lng >= -59.5 or lat >= 50.5: return "Newfoundland and Labrador" if lat < 46.05 or lng > -64.4: return "Nova Scotia" return "New Brunswick" def _brands_for(region: str | None, lat: float, lng: float): prov = (region or "").strip() if prov not in PROVINCE_BRANDS: prov = _infer_province(lat, lng) ids = PROVINCE_BRANDS.get(prov, PROVINCE_BRANDS[_DEFAULT_PROVINCE]) return [(bid, *BRAND_DEFS[bid]) for bid in ids if bid in BRAND_DEFS] _BAN = ("station", "stationnement", "kentucky", "kentville", "co-operators", "cooperators") def _norm(s: str) -> str: import unicodedata s = unicodedata.normalize("NFD", s or "") return "".join(c for c in s if unicodedata.category(c) != "Mn").lower() _token_cache: list[str] = [] def _token() -> str: if not _token_cache: cfg = (ROOT / "frontend" / "src" / "kamaps" / "config.ts").read_text() m = re.search(r'"(pk\.[A-Za-z0-9._-]+)"', cfg) if not m: raise RuntimeError("jeton Mapbox introuvable (kamaps/config.ts)") _token_cache.append(m.group(1)) return _token_cache[0] def _connect() -> sqlite3.Connection: DB_PATH.parent.mkdir(parents=True, exist_ok=True) con = sqlite3.connect(DB_PATH, timeout=15) con.row_factory = sqlite3.Row con.execute("""CREATE TABLE IF NOT EXISTS commerces_cache ( cellule TEXT, brand TEXT, nom TEXT, adresse TEXT, lat REAL, lng REAL, fetched_at REAL, PRIMARY KEY (cellule, brand))""") return con def _dist_m(lat1, lng1, lat2, lng2) -> 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 _fetch_brand(brand_q: str, lat: float, lng: float, match: str = "") -> dict | None: params = urllib.parse.urlencode({ "q": brand_q, "proximity": f"{lng},{lat}", "limit": 5, "types": "poi", "language": "fr", "country": "CA", "access_token": _token()}) req = urllib.request.Request(f"{API}?{params}", headers={"User-Agent": UA}) try: with urllib.request.urlopen(req, timeout=12) as r: feats = json.load(r).get("features") or [] except Exception: return None for f in feats: p = f.get("properties") or {} nom = _norm(p.get("name") or "") if match and match not in nom: continue if any(b in nom for b in _BAN): continue lng2, lat2 = f["geometry"]["coordinates"][:2] return {"nom": p.get("name") or brand_q, "adresse": p.get("full_address") or p.get("place_formatted") or "", "lat": lat2, "lng": lng2} return None OVERPASS = ["https://overpass.kumi.systems/api/interpreter", "https://overpass-api.de/api/interpreter"] def _fetch_transit(lat: float, lng: float) -> list[tuple[str, dict]]: """Station de métro et arrêt de bus les plus proches (OpenStreetMap).""" q = f"""[out:json][timeout:20]; ( node["railway"="station"]["station"="subway"](around:3000,{lat},{lng}); node["highway"="bus_stop"](around:1000,{lat},{lng}); ); out body;""" data = None for url in OVERPASS: try: req = urllib.request.Request( url, data=urllib.parse.urlencode({"data": q}).encode(), headers={"User-Agent": UA}) with urllib.request.urlopen(req, timeout=25) as r: data = json.load(r) break except Exception: continue if not data: return [] best: dict[str, tuple[float, dict]] = {} for el in data.get("elements", []): tags = el.get("tags") or {} kind = ("metro_station" if tags.get("railway") == "station" else "arret_bus") d = _dist_m(lat, lng, el["lat"], el["lon"]) if kind not in best or d < best[kind][0]: best[kind] = (d, {"nom": tags.get("name") or ("Station de métro" if kind == "metro_station" else "Arrêt de bus"), "adresse": "", "lat": el["lat"], "lng": el["lon"]}) return [(k, v[1]) for k, v in best.items()] TRANSIT = [("metro_station", "Station de métro"), ("rem_station", "Station REM"), ("arret_bus", "Arrêt de bus"), ("gare_train", "Gare de train")] _DB_GENRE = {"metro": "metro_station", "rem": "rem_station", "bus": "arret_bus", "train": "gare_train"} def _transit_from_db(lat: float, lng: float) -> list[tuple[str, dict]]: """Arrêts/stations depuis data/transit.db (extrait OpenStreetMap pré-calculé : ~37 000 arrêts de bus, métro, REM, gares du Québec).""" db = ROOT / "data" / "transit.db" if not db.exists(): return [] con = sqlite3.connect(f"file:{db}?mode=ro", uri=True) con.row_factory = sqlite3.Row out = [] for genre, rayon in (("metro", 6000), ("rem", 6000), ("bus", 1500), ("train", 8000)): d = rayon / 111320.0 rows = con.execute( "SELECT nom, lat, lng FROM arrets WHERE genre=? AND lat BETWEEN " "? AND ? AND lng BETWEEN ? AND ?", (genre, lat - d, lat + d, lng - d, lng + d)).fetchall() best = None for r in rows: dd = _dist_m(lat, lng, r["lat"], r["lng"]) if dd <= rayon and (best is None or dd < best[0]): best = (dd, r) if best: out.append((_DB_GENRE[genre], {"nom": best[1]["nom"] or "", "adresse": "", "lat": best[1]["lat"], "lng": best[1]["lng"]})) con.close() return out _POI_CAT = {"metro": "metro_station", "bus": "arret_bus"} def _transit_from_poi(lat: float, lng: float) -> list[tuple[str, dict]]: """Métro/bus depuis le cache POI du projet (louka.db, déjà calculé par immeuble) — la fiche interroge avec les mêmes coordonnées que poi.py.""" db_main = ROOT / "data" / next( (n for n in ("louka.db", "immoka.db", "immo.db") if (ROOT / "data" / n).exists()), "louka.db") if not db_main.exists(): return [] try: con = sqlite3.connect(f"file:{db_main}?mode=ro", uri=True) con.row_factory = sqlite3.Row d = 300 / 111320.0 row = con.execute( "SELECT pois, lat, lng FROM poi_cache WHERE lat BETWEEN ? AND ? " "AND lng BETWEEN ? AND ? ORDER BY (lat-?)*(lat-?)+(lng-?)*(lng-?) " "LIMIT 1", (lat - d, lat + d, lng - d, lng + d, lat, lat, lng, lng)).fetchone() con.close() except sqlite3.Error: return [] if row is None: return [] out = [] for e in json.loads(row["pois"] or "[]"): k = _POI_CAT.get(e.get("cat")) if k: out.append((k, {"nom": e.get("name") or "", "adresse": "", "lat": lat, "lng": lng, "_dist": e.get("dist_m")})) return out def nearby(lat: float, lng: float, region: str | None = None) -> dict: """Grand commerce le plus proche par bannière PROVINCIALE (cache ~1 km).""" brands = _brands_for(region, lat, lng) cell = f"{round(lat, 2)},{round(lng, 2)}" con = _connect() now = time.time() cached = {r["brand"]: r for r in con.execute( "SELECT * FROM commerces_cache WHERE cellule=? AND fetched_at>?", (cell, now - TTL))} manquants = [(bid, q, m) for bid, _, q, m in brands if bid not in cached] transit_manquant = any(k not in cached for k, _ in TRANSIT) if manquants or transit_manquant: res: list[tuple[str, dict | None]] = [] if manquants: with ThreadPoolExecutor(max_workers=6) as ex: res = list(ex.map( lambda b: (b[0], _fetch_brand(b[1], lat, lng, b[2])), manquants)) if transit_manquant: tr = (_transit_from_db(lat, lng) or _transit_from_poi(lat, lng) or _fetch_transit(lat, lng)) res.extend(tr) with con: for bid, hit in res: if hit is None: continue con.execute( "INSERT OR REPLACE INTO commerces_cache VALUES " "(?,?,?,?,?,?,?)", (cell, bid, hit["nom"], hit.get("adresse") or (str(hit["_dist"]) if hit.get("_dist") is not None else ""), hit["lat"], hit["lng"], now)) cached = {r["brand"]: r for r in con.execute( "SELECT * FROM commerces_cache WHERE cellule=? AND fetched_at>?", (cell, now - TTL))} con.close() items = [] transit = [] for bid, label in TRANSIT: r = cached.get(bid) if r is not None: # distance : celle du cache POI si disponible (adresse numérique) d = (float(r["adresse"]) if (r["adresse"] or "").replace( ".", "").isdigit() else _dist_m(lat, lng, r["lat"], r["lng"])) if d <= 5000: transit.append({"id": bid, "commerce": label, "nom": r["nom"], "adresse": "", "dist_m": round(d), "lat": r["lat"], "lng": r["lng"]}) for bid, label, _q, _m in brands: r = cached.get(bid) if r is None: continue d = _dist_m(lat, lng, r["lat"], r["lng"]) if d > 40000: # au-delà de 40 km : non pertinent continue items.append({"id": bid, "commerce": label, "nom": r["nom"], "adresse": r["adresse"], "dist_m": round(d), "lat": r["lat"], "lng": r["lng"]}) items.sort(key=lambda x: x["dist_m"]) transit.sort(key=lambda x: x["dist_m"]) return {"n": len(items), "commerces": items, "transit": transit}