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/msi.py : connecteur MSI Gestion immobilière (msimmobiliers.com)5# Crawl des pages de secteurs (rendu serveur) : Québec + Lévis + Montréal6# + Thetford Mines (flux intermittent — souvent 0 unité).7# Front Next.js : chaque fiche embarque un objet JSON `rental` complet dans8# le payload « flight » (self.__next_f.push) — adresse + code postal,9# lat/lng, date de disponibilité ISO, étage, superficie, caractéristiques10# structurées (features), galerie photo, politique chiens (dogPolicy).11# Fiches visitées via self.detail(...) (cache BD, plafond de requêtes).12# -----------------------------------------------------------------------------13from __future__ import annotations1415import hashlib16import json17import re1819from bs4 import BeautifulSoup2021from ..schema import Listing, infer_city, normalize_unit_type, parse_price22from .base import BaseConnector2324BASE = "https://www.msimmobiliers.com"25ROOTS = [26 f"{BASE}/appartements-a-louer/quebec",27 f"{BASE}/appartements-a-louer/levis",28 f"{BASE}/appartements-a-louer/montreal",29 f"{BASE}/appartements-a-louer/thetford-mine",30]31UNIT_RE = re.compile(r"/appartements-a-louer/[^\"]*appartement-(\d+)[^\"]*")32LIST_RE = re.compile(r"^/appartements-a-louer/[a-z0-9\-/]+$")33# ville par défaut selon la racine de l'URL (/appartements-a-louer/<ville>/...)34_ROOT_CITY = {"quebec": "Québec", "levis": "Lévis", "montreal": "Montréal",35 "thetford-mine": "Thetford Mines"}3637# morceaux de chaîne JS des payloads flight : self.__next_f.push([1,"..."])38_FLIGHT_RE = re.compile(39 r'self\.__next_f\.push\(\[1,"((?:[^"\\]|\\.)*)"\]\)')404142def _default_city(path_or_url: str) -> str:43 m = re.search(r"/appartements-a-louer/([a-z0-9\-]+)", path_or_url)44 return _ROOT_CITY.get(m.group(1) if m else "", "Québec")454647def _extract_rental(html: str) -> dict:48 """Extrait l'objet JSON `rental` du payload flight Next.js de la fiche."""49 for m in _FLIGHT_RE.finditer(html):50 try:51 chunk = json.loads('"' + m.group(1) + '"') # dés-échappe la chaîne52 except ValueError:53 continue54 i = chunk.find('"rental":{')55 if i < 0:56 continue57 start = i + len('"rental":')58 depth = 059 in_str = esc = False60 for j in range(start, len(chunk)):61 c = chunk[j]62 if esc:63 esc = False64 elif c == "\\":65 esc = True66 elif in_str:67 in_str = c != '"'68 elif c == '"':69 in_str = True70 elif c == "{":71 depth += 172 elif c == "}":73 depth -= 174 if depth == 0:75 try:76 return json.loads(chunk[start:j + 1])77 except ValueError:78 return {}79 break80 return {}818283class _CapAtteint(Exception):84 """Plafond de requêtes détail atteint pour cette synchronisation."""858687class MSIConnector(BaseConnector):88 source_id = "msi"89 request_delay = 0.590 max_list_pages = 90 # garde-fou de crawl (Qc + Lévis + Mtl)91 max_real_details = 150 # vraies requêtes détail par sync (cache exclu)9293 def fetch(self) -> list[Listing]:94 self._real_details = 095 # 1) BFS sur les pages de listes (arrondissements / quartiers)96 to_visit = list(ROOTS)97 visited: set[str] = set()98 listings: dict[str, Listing] = {}99 card_keys: dict[str, str] = {}100101 while to_visit and len(visited) < self.max_list_pages:102 url = to_visit.pop(0)103 if url in visited:104 continue105 visited.add(url)106 try:107 html = self.get(url).text108 except Exception:109 continue110 soup = BeautifulSoup(html, "html.parser")111112 # cartes d'unités113 for card in soup.select('a[href*="appartement-"]'):114 href = card.get("href", "")115 m = UNIT_RE.search(href)116 if not m:117 continue118 ext_id = m.group(1)119 if ext_id in listings:120 continue121 full_url = href if href.startswith("http") else BASE + href122 text = card.get_text("|", strip=True)123 parts = [p for p in text.split("|") if p and p != "Voir cette fiche"]124 # Format observé : "4 1/2 | 1195$ / mois | Appartement / Condo |125 # 177 Avenue Ruel | Chutes-Montmorency | Libre ..."126 unit_type = price_label = category = address = sector = avail = ""127 for p in parts:128 if not unit_type and re.match(r"^\d\s*1/2$|^Studio|^Loft", p, re.I):129 unit_type = p130 elif not price_label and "$" in p:131 price_label = p132 elif not category and re.search(r"Appartement|Condo|Maison|Commercial|Stationnement", p, re.I):133 category = p134 elif not address and re.match(r"^\d+[\s,]", p):135 address = p136 elif not avail and re.search(r"Libre|Disponib", p, re.I):137 avail = p138 elif not sector and address:139 sector = p140 # ignorer stationnements/espaces commerciaux141 if re.search(r"Stationnement|Commercial|Rangement|Parking", category or "", re.I):142 continue143 city = _default_city(href if "/appartements-a-louer/" in href144 else url)145 listings[ext_id] = Listing(146 source=self.source_id,147 external_id=ext_id,148 url=full_url,149 title=address or parts[0] if parts else f"Unité {ext_id}",150 address=address,151 sector=sector,152 city=infer_city(sector, default=city),153 unit_type=normalize_unit_type(unit_type),154 price=parse_price(price_label),155 price_label=price_label,156 availability=avail,157 )158 card_keys[ext_id] = hashlib.sha1(159 "|".join(parts).encode("utf-8")).hexdigest()[:16]160161 # sous-pages de secteurs162 for a in soup.select('a[href^="/appartements-a-louer/"]'):163 href = a.get("href", "").split("?")[0]164 if LIST_RE.match(href) and "appartement-" not in href:165 nxt = BASE + href166 if nxt not in visited:167 to_visit.append(nxt)168169 # 2) Fiches détaillées (cache BD) : JSON rental + caractéristiques170 for ext_id, lst in listings.items():171 det = self._unit_detail(ext_id, lst.url,172 card_keys.get(ext_id, ""))173 if not det:174 continue175 if det.get("images"):176 lst.images = det["images"][:25]177 if det.get("address_full"):178 lst.address = det["address_full"] # avec ville (plus complet)179 lst.lat, lst.lng = det.get("lat"), det.get("lng")180 if det.get("area"):181 lst.area_sqft = det["area"]182 if det.get("availability") and not lst.availability:183 lst.availability = f"Disponibilité : {det['availability']}"184 lst.amenities = det.get("amenities") or []185 if det.get("floor"):186 lst.details = {**lst.details, "floor": det["floor"]}187 if det.get("catchphrase"):188 lst.description = det["catchphrase"][:600]189190 return list(listings.values())191192 def _unit_detail(self, ext_id: str, url: str, card_key: str) -> dict:193 """Fiche unité : objet JSON `rental` (flight Next.js) + liste des194 caractéristiques affichées (textes bruts, incl. chiens/fumeur)."""195 def _fetch() -> dict:196 if self._real_details >= self.max_real_details:197 raise _CapAtteint()198 self._real_details += 1199 html = self.get(url).text200 out: dict = {}201202 rental = _extract_rental(html)203 adr = rental.get("address") or {}204 if isinstance(adr, dict):205 out["address_full"] = adr.get("full") or ""206 if isinstance(adr.get("lat"), (int, float)):207 out["lat"] = adr["lat"]208 if isinstance(adr.get("lng"), (int, float)):209 out["lng"] = adr["lng"]210 area = rental.get("area")211 if isinstance(area, (int, float)) and 80 <= area <= 20000:212 out["area"] = float(area)213 floor = rental.get("floor")214 if isinstance(floor, int) and 0 < floor <= 60:215 out["floor"] = floor216 avail = rental.get("availability")217 if isinstance(avail, str) and avail:218 out["availability"] = avail219 catch = rental.get("catchphrase")220 if isinstance(catch, str) and catch and "$undefined" not in catch:221 out["catchphrase"] = re.sub(r"\s+", " ", catch).strip()222 imgs = []223 for ph in rental.get("gallery") or []:224 u = ((ph.get("sizes") or {}).get("large") if225 isinstance(ph, dict) else None)226 if u and u not in imgs:227 imgs.append(u)228229 # Caractéristiques affichées (textes bruts : « Balcon »,230 # « Chien interdit », « Non fumeur », « Dernier étage »…)231 soup = BeautifulSoup(html, "html.parser")232 amenities: list[str] = []233 cat_name = ((rental.get("category") or {}).get("name") or "")234 for li in soup.select(".rental-content__hero--facilities li"):235 if li.find("a"):236 continue # lien Google Maps (adresse)237 txt = li.get_text(" ", strip=True)238 if (not txt or txt in ("N/C",) or txt.startswith("**")239 or txt == cat_name or len(txt) > 60):240 continue241 if txt not in amenities:242 amenities.append(txt)243 if not amenities: # repli : features structurées du JSON244 amenities = [f.get("name") for f in245 (rental.get("features") or []) +246 (rental.get("secondaryFeatures") or [])247 if isinstance(f, dict) and f.get("name")]248 out["amenities"] = amenities[:20]249250 if not imgs: # repli : images wp-content de la page251 imgs = [u for u in dict.fromkeys(re.findall(252 r'https://api\.msimmobiliers\.com/wp-content/uploads/'253 r'[^"\\\s\)]+\.(?:jpg|jpeg|png|webp)', html))254 if not re.search(r"logo|icon|favicon", u, re.I)]255 out["images"] = imgs[:25]256 return out257258 try:259 return self.detail(ext_id, card_key, _fetch)260 except Exception:261 return {}262