spb/lou-ka Public
Lou·Ka — tous les logements à louer du Québec, un seul endroit.
HTML 99.7%
1# -----------------------------------------------------------------------------2# Lou-Ka — Agrégateur de logements à louer (province de Québec)3# Auteur : Simon-Pierre Boucher — contact@spboucher.ai4# connectors/devimco.py : connecteur Devimco Appartements5# (devimco.com/appartements — St Ann & Hexagone à Griffintown, Alexander &6# Maestria au centre-ville de Montréal, Éolia / Luméo / Nobel à Brossard,7# Ostral à Longueuil).8# Chaque page projet embarque un iframe Planpoint (app.planpoint.io) ;9# l'inventaire complet s'obtient via l'API JSON de Planpoint :10# POST https://app.planpoint.io/api/projects/find {namespace, hostName}11# -> floors[] -> units[] (prix, chambres, photos, disponibilité...).12# -----------------------------------------------------------------------------13from __future__ import annotations1415import html as htmllib16import re17import time18import urllib.parse1920from ..schema import Listing, normalize_unit_type21from .base import BaseConnector2223BASE = "https://devimco.com"24HUB_URL = f"{BASE}/appartements"25PLANPOINT_FIND = "https://app.planpoint.io/api/projects/find"26PLANPOINT_GROUP_FIND = "https://app.planpoint.io/api/groups/find"2728PROJECT_RE = re.compile(29 r"https://devimco\.com/appartements/a-louer/"30 r"(montreal|rive-sud-de-montreal)/([a-z\-]+)/([a-z0-9\-]+)")3132# secteur d'URL -> (ville, secteur affiché)33SECTORS = {34 "centre-ville": ("Montréal", "Centre-ville"),35 "griffintown": ("Montréal", "Griffintown"),36 "brossard": ("Brossard", "Quartier DIX30"),37 "longueuil": ("Longueuil", "Vieux-Longueuil"),38}394041def _unit_type(bedrooms: str) -> str:42 s = (bedrooms or "").lower()43 if "studio" in s:44 return "Studio"45 m = re.search(r"(\d+)", s)46 if not m:47 return bedrooms or ""48 n = int(m.group(1))49 return "Studio" if n == 0 else f"{n + 2}½"505152# Jetons de la liste `inclusions` de l'API Planpoint -> details canoniques.53# Champ structuré côté source (liste séparée par des virgules), donc mappé54# explicitement ; un jeton absent = inconnu (jamais False).55_INCLUSION_TOKENS = {56 "heating": ("inclusions", "heating"),57 "electricity": ("inclusions", "electricity"),58 "hot water": ("inclusions", "hot_water"),59 "internet": ("inclusions", "internet"),60 "cable": ("inclusions", "cable"),61 "stove": ("appliances", "stove"),62 "refrigerator": ("appliances", "fridge"),63 "fridge": ("appliances", "fridge"),64 "dishwasher": ("appliances", "dishwasher"),65 "air conditioning": (None, "ac"),66 "balcony": (None, "balcony"),67}686970def _details_from_inclusions(tokens: list[str]) -> dict:71 """« internet, electricity, heating, stove, refrigerator… » -> details."""72 details: dict = {}73 lows = [t.lower() for t in tokens]74 for tok in lows:75 for key, (group, name) in _INCLUSION_TOKENS.items():76 if key in tok:77 if group:78 details.setdefault(group, {})[name] = True79 else:80 details[name] = True81 if "washer" in " ".join(lows) and "dryer" in " ".join(lows):82 details.setdefault("appliances", {})["washer_dryer"] = True83 return details848586class DevimcoConnector(BaseConnector):87 source_id = "devimco"88 request_delay = 0.689 max_projects = 15 # garde-fou9091 def fetch(self) -> list[Listing]:92 listings: list[Listing] = []93 try:94 hub = self.get(HUB_URL).text95 except Exception:96 return listings9798 # 1) Pages projets (a-louer/<région>/<secteur>/<projet>)99 projects: dict[str, tuple[str, str]] = {}100 for m in PROJECT_RE.finditer(hub):101 url = m.group(0)102 sector_slug, proj_slug = m.group(2), m.group(3)103 # ignorer les pages de secteur (sans slug projet "…-appartements")104 if proj_slug.endswith("-appartements"):105 projects[url] = (sector_slug, proj_slug)106107 for i, (proj_url, (sector_slug, proj_slug)) in \108 enumerate(projects.items()):109 if i >= self.max_projects:110 break111 try:112 listings.extend(113 self._fetch_project(proj_url, sector_slug))114 except Exception:115 continue116 return listings117118 # -- un projet ---------------------------------------------------------------119 def _fetch_project(self, proj_url: str, sector_slug: str) -> list[Listing]:120 out: list[Listing] = []121 html = self.get(proj_url).text122123 # iframe Planpoint : soit un projet (/<ns>/<host>), soit un groupe124 # de phases (/g/<ns>)125 projects: list[dict] = []126 m = re.search(127 r"https://app\.planpoint\.io/g/([a-z0-9\-]+)", html)128 if m:129 group = self._planpoint(PLANPOINT_GROUP_FIND,130 {"namespace": m.group(1)})131 projects = group.get("projects") or []132 else:133 m = re.search(134 r"https://app\.planpoint\.io/([a-z0-9\-]+)/([a-z0-9\-]+)\?",135 html)136 if not m:137 return out138 projects = [self._planpoint(139 PLANPOINT_FIND,140 {"namespace": m.group(1), "hostName": m.group(2)})]141142 # Adresse affichée sur la page Devimco (lien Google Maps) : repli143 # pour les phases dont l'API Planpoint n'a pas d'adresse144 # (ex. Maestria Tour B, Hexagone 2).145 page_address = ""146 m = re.search(r"google\.[a-z.]+/maps/search/\?[^\"']*query=([^\"'&]+)",147 html)148 if m:149 page_address = htmllib.unescape(150 urllib.parse.unquote(m.group(1))).strip()151152 city, sector = SECTORS.get(sector_slug, ("Montréal", sector_slug))153 for project in projects:154 try:155 out.extend(self._parse_units(project, proj_url, city, sector,156 page_address))157 except Exception:158 continue159 return out160161 def _planpoint(self, endpoint: str, payload: dict) -> dict:162 wait = self.request_delay - (time.time() - self._last_request)163 if wait > 0:164 time.sleep(wait)165 resp = self.session.post(endpoint, json=payload, timeout=60)166 self._last_request = time.time()167 resp.raise_for_status()168 return resp.json() or {}169170 def _parse_units(self, project: dict, proj_url: str,171 city: str, sector: str,172 page_address: str = "") -> list[Listing]:173 out: list[Listing] = []174 namespace = project.get("namespace") or ""175 name = project.get("name") or namespace176 address = (project.get("address") or "").split(", Quebec")[0]177 if not address:178 address = page_address179 lat, lng = project.get("lat"), project.get("lng") or project.get("lon")180 for floor in project.get("floors") or []:181 floor_name = floor.get("name") or ""182 for u in floor.get("units") or []:183 try:184 if (u.get("availability") or "").lower() != "available":185 continue # loué / réservé / à venir186 price = u.get("price")187 if u.get("unitPriceTBD") or not price:188 continue # pas de prix affiché189 unit_name = str(u.get("name") or "")190 images = [img for img in191 (u.get("images") or []) +192 (u.get("layoutGallery") or [])193 if isinstance(img, str) and img.startswith("http")]194 inclusions = [s.strip() for s in195 (u.get("inclusions") or "").split(",")196 if s.strip()]197 sqft = u.get("squareFeet")198 amenities = list(inclusions)199 if sqft:200 amenities.append(f"{sqft} pi²")201 if u.get("bathrooms"):202 amenities.append(f"{u['bathrooms']} salle(s) de bain")203 if u.get("furnished"):204 amenities.append("Meublé")205 availability = "Disponible"206 if u.get("deliveryDate"):207 availability = f"Disponible : {u['deliveryDate']}"208209 # Superficie structurée de l'API (sinon None)210 area = None211 try:212 if sqft and 80 <= float(sqft) <= 20000:213 area = float(sqft)214 except (TypeError, ValueError):215 pass216217 # Meublé : booléen explicite de l'API218 furnished = (bool(u["furnished"])219 if isinstance(u.get("furnished"), bool)220 else None)221222 # Type exact « 3.5 » de l'API, sinon via nb de chambres223 unit_type = (normalize_unit_type(str(u.get("type") or ""))224 or _unit_type(u.get("bedrooms") or ""))225226 out.append(Listing(227 source=self.source_id,228 external_id=u.get("_id") or f"{namespace}-{unit_name}",229 url=proj_url,230 title=f"{name} — unité {unit_name}",231 address=address,232 sector=sector,233 city=city,234 unit_type=unit_type,235 price=float(price)236 if 100 <= float(price) <= 20000 else None,237 price_label=f"{int(price)} $/mois",238 availability=availability,239 area_sqft=area,240 furnished=furnished,241 description=(f"Étage {floor_name} — "242 f"{u.get('orientation') or ''}").strip(" —"),243 amenities=list(dict.fromkeys(amenities)),244 details=_details_from_inclusions(inclusions),245 images=list(dict.fromkeys(images)),246 lat=float(lat) if lat else None,247 lng=float(lng) if lng else None,248 ))249 except Exception:250 continue251 return out252