# ----------------------------------------------------------------------------- # Lou-Ka — Agrégateur de logements à louer (province de Québec) # Auteur : Simon-Pierre Boucher — contact@spboucher.ai # connectors/shdm.py : connecteur SHDM — Société d'habitation et de # développement de Montréal (shdm.org, logements abordables à Montréal). # Front Nuxt 3 + WordPress GraphQL : la page /fr/logements-disponibles # embarque les logements disponibles dans le payload __NUXT_DATA__ # (format « devalue » : tableau plat où les valeurs sont des index). # ----------------------------------------------------------------------------- from __future__ import annotations import json import re from ..schema import Listing, normalize_unit_type from .base import BaseConnector BASE = "https://www.shdm.org" LIST_URL = f"{BASE}/fr/logements-disponibles" # Champs booléens du CPT « logement » -> libellé de commodité _AMENITY_FLAGS = { "chauffage": "Chauffage inclus", "electricite": "Électricité incluse", "eauChaude": "Eau chaude incluse", "balcon": "Balcon", "portePatio": "Porte-patio", "plancherBois": "Plancher de bois", "deuxEtage": "Sur deux étages", "adapte": "Logement adapté", "handicape": "Accessible (mobilité réduite)", "positionFenetreEst": "Fenêtres à l'est", "positionFenetreNord": "Fenêtres au nord", "positionFenetreOuest": "Fenêtres à l'ouest", "positionFenetreSud": "Fenêtres au sud", } _BEDROOMS_TO_TYPE = {0: "Studio", 1: "3½", 2: "4½", 3: "5½", 4: "6½"} def _strip_html(s: str) -> str: s = re.sub(r"<[^>]+>", " ", s or "") s = s.replace("’", "’").replace(" ", " ") s = re.sub(r"&[a-z]+;|&#\d+;", " ", s) return re.sub(r"\s+", " ", s).strip() class _Devalue: """Résolution des références du payload Nuxt 3 (format devalue).""" def __init__(self, data: list): self.data = data def resolve(self, ref, seen: frozenset = frozenset()): if not isinstance(ref, int) or not (0 <= ref < len(self.data)): return ref if ref in seen: return None seen = seen | {ref} val = self.data[ref] if isinstance(val, dict): return {k: self.resolve(v, seen) for k, v in val.items()} if isinstance(val, list): return [self.resolve(v, seen) for v in val] return val class SHDMConnector(BaseConnector): source_id = "shdm" request_delay = 0.6 @staticmethod def _media_urls(medias) -> list[str]: """[{media:{node:{generic: url, ...}}}] -> URLs pleine taille.""" urls: list[str] = [] for m in medias or []: node = ((m or {}).get("media") or {}).get("node") or {} u = node.get("generic") or node.get("genericDesktop") or "" if u: urls.append(u) return urls def fetch(self) -> list[Listing]: listings: list[Listing] = [] try: html = self.get(LIST_URL).text except Exception: return listings m = re.search(r'id="__NUXT_DATA__"[^>]*>(.*?)', html, re.S) if not m: return listings try: data = json.loads(m.group(1)) except ValueError: return listings if not isinstance(data, list): return listings dv = _Devalue(data) seen_ids: set[str] = set() for i, raw in enumerate(data): if not (isinstance(raw, dict) and "logementFields" in raw and "uri" in raw): continue try: node = dv.resolve(i) slug = node.get("slug") or "" lf = node.get("logementFields") or {} if not slug or slug in seen_ids or not isinstance(lf, dict): continue imm_nodes = ((lf.get("immeuble") or {}).get("nodes")) or [{}] imm = imm_nodes[0] if isinstance(imm_nodes[0], dict) else {} bf = imm.get("buildingFields") or {} building_title = (imm.get("title") or "").strip() address = (bf.get("adresse") or building_title).strip() quartier_nodes = ((bf.get("quartier") or {}).get("nodes")) or [] sector = (quartier_nodes[0].get("title") or "").strip() \ if quartier_nodes and isinstance(quartier_nodes[0], dict) else "" unit_no = str(lf.get("numeroLogement") or "").strip() typologie = lf.get("typologie") or [] typ_raw = typologie[0] if typologie else "" unit_type = normalize_unit_type(typ_raw) or \ _BEDROOMS_TO_TYPE.get(lf.get("chambre"), "") price = lf.get("prix") price = float(price) if isinstance(price, (int, float)) else None price_label = f"{price:,.0f}".replace(",", " ") + " $ / mois" \ if price else "" if re.search(r"stationnement|commercial", typ_raw, re.I): continue amenities = [label for key, label in _AMENITY_FLAGS.items() if lf.get(key)] # nombre de chambres (champ structuré du CPT) chambres = lf.get("chambre") if isinstance(chambres, int) and chambres > 0: amenities.append(f"{chambres} chambre" + ("s" if chambres > 1 else "")) # clientèle cible (taxonomie « autonomie » : Pour tous, 60+, …) auto_edges = ((lf.get("autonomie") or {}).get("edges")) or [] for edge in auto_edges: name = (((edge or {}).get("node")) or {}).get("name") or "" if name: amenities.append(f"Clientèle : {name}") break # étage (champ structuré ; 0 = rez-de-chaussée -> non exposé) details: dict = {} etage = lf.get("etage") if isinstance(etage, int) and etage > 0: details["floor"] = etage # NB : les booléens d'inclusion du CPT (chauffage, electricite, # eauChaude) sont peu fiables à false (cases non cochées alors # que la description de l'immeuble dit « inclus ») -> on ne # les publie qu'à true, via _AMENITY_FLAGS ci-dessus. # date de disponibilité ISO structurée ("2026-09-01T00:00:00+00:00") date_raw = (lf.get("date") or "")[:10] # images : photos du logement + photo(s) de l'immeuble images = (self._media_urls(lf.get("medias")) + self._media_urls(bf.get("medias"))) title = building_title or address if unit_no: title = f"{title} — app. {unit_no}" # NB : les URI WordPress /fr/logement/ répondent 404 # (les fiches s'ouvrent en modale) -> on pointe vers la liste, # avec le slug en ancre pour garder une URL unique et valide. listings.append(Listing( source=self.source_id, external_id=slug, url=f"{LIST_URL}#{slug}", title=title, address=address, sector=sector, city="Montréal", # SHDM = société paramunicipale de Mtl unit_type=unit_type, price=price, price_label=price_label, availability=date_raw, description=_strip_html(bf.get("content") or "")[:600], amenities=amenities, details=details, images=list(dict.fromkeys(u for u in images if u))[:25], )) seen_ids.add(slug) except Exception: continue return listings