# ----------------------------------------------------------------------------- # Lou-Ka — Agrégateur de logements à louer (province de Québec) # Auteur : Simon-Pierre Boucher — contact@spboucher.ai # connectors/niddamour.py : connecteur Nid d'Amour / Karen Cadet inc. # (niddamour.ca — Plateau, Verdun, Outremont, Rosemont, Ville-Marie, # Brossard...). Front WordPress + Angular sur la plateforme source.immo : # on lit la config publique (_configs.json) puis on interroge directement # l'API JSON api-v1.source.immo (liste + fiches avec toutes les photos). # ----------------------------------------------------------------------------- from __future__ import annotations import hashlib import json import re from ..schema import Listing from .base import BaseConnector SITE = "https://niddamour.ca" # la config source.immo (config_path) est présente sur la page d'accueil ; # les pages /proprietes/ sont des routes Angular servies en HTTP 404 HOME_URL = f"{SITE}/" API_ROOT = "https://api-v1.source.immo/api" # Régions administratives admissibles (Grand Montréal / CMM) ALLOWED_REGIONS = {"montreal", "laval", "monteregie", "lanaudiere", "laurentides"} _BEDROOMS_TO_TYPE = {0: "Studio", 1: "3½", 2: "4½", 3: "5½", 4: "6½"} # mots-clés des champs structurés inclusions/exclusions -> clés Lou-Ka _INCL_KEYS = [ (re.compile(r"electricite|hydro"), "electricity"), (re.compile(r"chauffage"), "heating"), (re.compile(r"eau chaude"), "hot_water"), (re.compile(r"internet|wi-?fi"), "internet"), (re.compile(r"cable|television"), "cable"), ] def _strip_accents_lower(s: str) -> str: import unicodedata return "".join(c for c in unicodedata.normalize("NFD", (s or "").lower()) if unicodedata.category(c) != "Mn") class _CapAtteint(Exception): """Plafond de requêtes détail atteint pour cette synchronisation.""" class NiddamourConnector(BaseConnector): source_id = "niddamour" request_delay = 0.6 max_real_details = 150 # vraies requêtes détail par sync (cache exclu) # -- helpers --------------------------------------------------------------- def _api(self, path: str, cfg: dict) -> dict: headers = { "x-si-account": cfg["account_id"], "x-si-api": cfg["api_key"], "x-si-appId": cfg["app_id"], "x-si-appVersion": cfg.get("app_version", ""), "Origin": SITE, "Referer": SITE + "/", } return self.get(f"{API_ROOT}/{path}", headers=headers).json() def _load_config(self) -> dict | None: """Extrait l'URL du _configs.json de source.immo depuis le site.""" try: page = self.get(HOME_URL).text except Exception: return None m = re.search(r'config_path\s*:\s*"([^"]+)"', page) if not m: return None cfg_url = m.group(1).replace("\\/", "/") if cfg_url.startswith("//"): cfg_url = "https:" + cfg_url try: return self.get(cfg_url).json() except Exception: return None # -- fiche détaillée (via cache BD self.detail) ---------------------------- def _unit_detail(self, ref: str, key: str, view: str, cfg: dict) -> dict: """Fiche source.immo : photos, adresse (+ code postal), description, addendum, inclusions/exclusions structurées, étage, sdb.""" def _fetch() -> dict: if self._real_details >= self.max_real_details: raise _CapAtteint() self._real_details += 1 det = self._api( f"listing/view/{view}/fr/items/ref_number/{ref}", cfg) out: dict = {} out["images"] = [ph.get("url") for ph in (det.get("photos") or []) if isinstance(ph, dict) and ph.get("url")] out["description"] = re.sub( r"\s+", " ", det.get("description") or "").strip() adr = (det.get("location") or {}).get("address") or {} parts = [adr.get("street_number"), adr.get("street_name")] address = " ".join(x for x in parts if x) if adr.get("door") and address: address += f", app. {adr['door']}" if adr.get("postal_code") and address: address += f", {adr['postal_code']}" out["address"] = address out["inclusions_txt"] = det.get("inclusions") or "" out["exclusions_txt"] = det.get("exclusions") or "" # étage de l'unité principale + salles de bain for u in det.get("units") or []: if (u.get("category_code") == "MAIN" and isinstance(u.get("level"), int) and 0 < u["level"] <= 60): out["floor"] = u["level"] main = next((u for u in det.get("units") or [] if u.get("category_code") == "MAIN"), {}) if isinstance(main.get("bathroom_count"), int) \ and main["bathroom_count"] > 0: out["bathrooms"] = main["bathroom_count"] # addendum HTML (détails de l'appartement, services à proximité) add = det.get("addendum") or "" add = re.sub(r"", " — ", add) add = re.sub(r"<[^>]+>", " ", add) add = re.sub(r" ?", " ", add) out["addendum"] = re.sub(r"\s+", " ", add).strip() return out try: return self.detail(ref, key, _fetch) except Exception: return {} # -- contrat --------------------------------------------------------------- def fetch(self) -> list[Listing]: self._real_details = 0 listings: list[Listing] = [] cfg = self._load_config() if not cfg or not cfg.get("api_key"): return listings view = cfg.get("default_view") or "" if view.startswith("{"): try: view = json.loads(view).get("id", "") except ValueError: return listings if not view: return listings # Dictionnaires (codes ville / sous-catégorie / région -> libellés) try: meta = self._api(f"view/{view}/fr", cfg) except Exception: return listings dico = meta.get("dictionary") or {} cities = dico.get("city") or {} subcats = dico.get("listing_subcategory") or {} regions = dico.get("region") or {} try: items = (self._api(f"listing/view/{view}/fr/items", cfg) .get("items") or []) except Exception: return listings for it in items: try: ref = it.get("ref_number") or "" if not ref: continue # disponibles à louer, résidentiel seulement if it.get("status_code") != "AVAILABLE": continue if not it.get("for_rent_flag"): continue if (it.get("category_code") or "") != "RESIDENTIAL": continue subcap = ((subcats.get(it.get("subcategory_code") or "") or {}) .get("caption") or "") if re.search(r"stationnement|commercial|bureau|local|terrain|" r"industriel|garage|entrep[oô]t", subcap, re.I): continue loc = it.get("location") or {} region_cap = ((regions.get(loc.get("region_code") or "") or {}) .get("caption") or "") if region_cap and \ _strip_accents_lower(region_cap) not in ALLOWED_REGIONS: continue # hors Grand Montréal # 'Montréal (Le Plateau-Mont-Royal)' -> ville + quartier city_cap = ((cities.get(loc.get("city_code") or "") or {}) .get("caption") or "") mcity = re.match(r"^([^(]+?)\s*(?:\(([^)]+)\))?$", city_cap) city = (mcity.group(1).strip() if mcity else city_cap) or "Montréal" sector = (mcity.group(2) or "").strip() if mcity else "" price = ((it.get("price") or {}).get("rent") or {}).get("amount") price = float(price) if isinstance(price, (int, float)) else None price_label = (f"{price:,.0f}".replace(",", " ") + " $ / mois" if price else "") bedrooms = (it.get("main_unit") or {}).get("bedroom_count") if re.search(r"maison", subcap, re.I): unit_type = "Maison" elif re.search(r"studio|loft", subcap, re.I) and not bedrooms: unit_type = "Studio" else: unit_type = _BEDROOMS_TO_TYPE.get( bedrooms, f"{bedrooms} chambres" if bedrooms else "") # Fiche détaillée (cache BD, clé = hash de l'item de liste) key = hashlib.sha1(json.dumps( it, sort_keys=True, ensure_ascii=False) .encode("utf-8")).hexdigest()[:16] det = self._unit_detail(ref, key, view, cfg) address = det.get("address") or "" description = det.get("description") or "" if det.get("addendum"): description = (f"{description} — {det['addendum']}" if description else det["addendum"]) if det.get("bathrooms"): description = " — ".join( x for x in [description, f"{det['bathrooms']} sdb"] if x) description = description[:600] # inclusions positives -> amenities brutes (affichage) amenities = [re.sub(r"^[-–•\s]+", "", ln).strip(" ;.") for ln in (det.get("inclusions_txt") or "") .splitlines() if ln.strip(" -–•;.")] # inclusions/exclusions structurées -> details.inclusions # (les exclusions priment : « Électricité, chauffage et eau # chaude » non inclus ne doit pas devenir positif) details: dict = {} incl: dict = {} incl_key = _strip_accents_lower(det.get("inclusions_txt") or "") excl_key = _strip_accents_lower(det.get("exclusions_txt") or "") for rx, cle in _INCL_KEYS: if rx.search(incl_key): incl[cle] = True for rx, cle in _INCL_KEYS: if rx.search(excl_key): incl[cle] = False if incl: details["inclusions"] = incl if det.get("floor"): details["floor"] = det["floor"] images = det.get("images") or [] if not images and it.get("photo_url"): images = [it["photo_url"]] title = address or f"{unit_type or 'Logement'} — {sector or city}" listings.append(Listing( source=self.source_id, external_id=ref, url=f"{SITE}/propriete/{ref.lower()}/", title=title, address=address, sector=sector, city=city, unit_type=unit_type, price=price, price_label=price_label, availability="Disponible", description=description, amenities=[a for a in amenities if a][:15], details=details, images=list(dict.fromkeys(images))[:25], lat=loc.get("latitude"), lng=loc.get("longitude"), )) except Exception: continue return listings