# ----------------------------------------------------------------------------- # Lou-Ka — Agrégateur de logements à louer (province de Québec) # Auteur : Simon-Pierre Boucher — contact@spboucher.ai # connectors/voltige.py : Le Voltige (voltigemtl.ca) — complexe locatif DevImmo # à Ahuntsic (Montréal), 4 immeubles : Hélia, Belvédère, Aria, Panora. # Techno : widget Livya (ex-Realvuu — app.livya.com/embed.js). La page # /plans/ porte data-client/data-project/data-entity ; le module iframe # https://app.livya.com/fr/{client}/projects/{project}/plans/{entity}?noLayout=1 # est un Next.js dont le flight data RSC (self.__next_f.push) inline TOUTES # les unités : on concatène les chunks puis on extrait les objets JSON # {"unitId": …} et {"buildingId": …} par appariement d'accolades. # Granularité : UNITÉ (availability=AVAILABLE ; prix publié sur ~10 %). # ----------------------------------------------------------------------------- from __future__ import annotations import json import re from .base import BaseConnector from ..schema import Listing SITE_URL = "https://voltigemtl.ca/plans/" LIVYA_BASE = "https://app.livya.com" # valeurs observées 2026-08-25 (repli si la page ne les expose plus) DEFAULT_CLIENT = "society-dev-immo" DEFAULT_PROJECT = "voltige" DEFAULT_ENTITY = "2aad7013-9b35-4b56-a72f-8dbf9d063678" _ATTR_RE = {k: re.compile(rf'data-{k}="([^"]+)"') for k in ("client", "project", "entity")} _FLIGHT_RE = re.compile(r'self\.__next_f\.push\(\[1,"((?:[^"\\]|\\.)*)"\]\)') _UNIT_START_RE = re.compile(r'\{"unitId":"') _BLDG_START_RE = re.compile(r'\{"buildingId":"[0-9a-f-]{36}","projectId"') def _balanced(s: str, i: int) -> str | None: """Objet JSON complet à partir de l'accolade ouvrante en position i.""" depth = 0 in_str = esc = False for j in range(i, len(s)): ch = s[j] if in_str: if esc: esc = False elif ch == "\\": esc = True elif ch == '"': in_str = False continue if ch == '"': in_str = True elif ch == "{": depth += 1 elif ch == "}": depth -= 1 if depth == 0: return s[i:j + 1] return None def _extract(full: str, start_re: re.Pattern) -> list[dict]: out: list[dict] = [] for m in start_re.finditer(full): obj = _balanced(full, m.start()) if not obj: continue try: out.append(json.loads(obj)) except ValueError: continue return out def _unit_type(rooms, bedrooms, type_type: str) -> str: if (type_type or "").upper() == "STUDIO": return "Studio" if isinstance(rooms, (int, float)) and rooms >= 1.5: return f"{int(rooms)}½" if isinstance(rooms, (int, float)) and 0 < rooms < 1.5: return "Studio" # modèles S1/S5… : rooms = 0.5 if isinstance(bedrooms, (int, float)) and bedrooms >= 0: return "Studio" if bedrooms == 0 else f"{int(bedrooms) + 2}½" return "" class VoltigeConnector(BaseConnector): """Le Voltige (Ahuntsic, Montréal) — unités du widget Livya.""" source_id = "voltige" request_delay = 1.0 use_detail_cache = False # tout est inline dans le flight data def _embed_params(self) -> tuple[str, str, str]: try: h = self.get(SITE_URL).text vals = {k: (rx.search(h).group(1) if rx.search(h) else "") for k, rx in _ATTR_RE.items()} if all(vals.values()): return vals["client"], vals["project"], vals["entity"] except Exception: # noqa: BLE001 — repli sur les valeurs connues pass return DEFAULT_CLIENT, DEFAULT_PROJECT, DEFAULT_ENTITY def fetch(self) -> list[Listing]: client, project, entity = self._embed_params() mod_url = (f"{LIVYA_BASE}/fr/{client}/projects/{project}" f"/plans/{entity}?noLayout=1") h = self.get(mod_url).text chunks = _FLIGHT_RE.findall(h) full = "".join(json.loads(f'"{c}"') for c in chunks) buildings = {b["buildingId"]: b for b in _extract(full, _BLDG_START_RE) if b.get("buildingId")} units: dict[str, dict] = {} for u in _extract(full, _UNIT_START_RE): uid = u.get("unitId") if uid: units[uid] = u # dédoublonnage (objets répétés) out: list[Listing] = [] for uid, u in units.items(): if (u.get("availability") or "").upper() != "AVAILABLE": continue if (u.get("segment") or "").upper() != "RESIDENTIAL": continue if not u.get("rental", True): continue # location au mois seulement b = buildings.get(u.get("buildingId") or "") or {} bname = (b.get("name") or "").strip() address = (u.get("address") or b.get("address") or "").strip() city = (u.get("city") or b.get("city") or "Montréal").strip() lat = b.get("latitude") or None lng = b.get("longitude") or None if not lat or not lng: # 0 = coordonnée absente lat = lng = None price = None for k in ("rentalPrice", "startingAtRentalPrice"): v = u.get(k) if isinstance(v, (int, float)) and v > 0: price = float(v) break area = u.get("unitSize") area = float(area) if isinstance(area, (int, float)) and area > 0 else None bd = u.get("roomsBed") bd = float(bd) if isinstance(bd, (int, float)) and bd >= 0 else None ba = u.get("roomsBath") ba = float(ba) if isinstance(ba, (int, float)) and ba > 0 else None images = [] for img in u.get("typeImages") or []: url = (img or {}).get("fullUrl") if isinstance(url, str) and url.startswith("http") \ and url not in images: images.append(url) details: dict = {} if u.get("floorDisplayName"): details["floor"] = u["floorDisplayName"] if u.get("typeName"): details["model"] = u["typeName"] if bname: details["building"] = bname num = str(u.get("number") or "").strip() title_bits = [x for x in (bname, f"unité {num}" if num else "") if x] out.append(Listing( source=self.source_id, external_id=uid, # uuid Livya de l'unité (stable) url=SITE_URL, # pas de page publique par unité title=("Le Voltige — " + ", ".join(title_bits)).strip(" —,"), address=address, sector="Ahuntsic", city=city, unit_type=_unit_type(u.get("rooms"), bd, u.get("typeType") or ""), bedrooms=bd, bathrooms=ba, price=price, price_label=f"{price:.0f} $ /mois" if price else "", availability="Disponible", area_sqft=area, description=(u.get("description") or "").strip()[:2000], details=details, images=images[:30], lat=lat, lng=lng, )) return out