spb/lou-ka

1# -----------------------------------------------------------------------------2# Lou-Ka — Agrégateur de logements à louer (province de Québec)3# Auteur : Simon-Pierre Boucher — contact@spboucher.ai4# connectors/voltige.py : Le Voltige (voltigemtl.ca) — complexe locatif DevImmo5#   à Ahuntsic (Montréal), 4 immeubles : Hélia, Belvédère, Aria, Panora.6#   Techno : widget Livya (ex-Realvuu — app.livya.com/embed.js). La page7#   /plans/ porte data-client/data-project/data-entity ; le module iframe8#   https://app.livya.com/fr/{client}/projects/{project}/plans/{entity}?noLayout=19#   est un Next.js dont le flight data RSC (self.__next_f.push) inline TOUTES10#   les unités : on concatène les chunks puis on extrait les objets JSON11#   {"unitId": …} et {"buildingId": …} par appariement d'accolades.12#   Granularité : UNITÉ (availability=AVAILABLE ; prix publié sur ~10 %).13# -----------------------------------------------------------------------------14from __future__ import annotations1516import json17import re1819from .base import BaseConnector20from ..schema import Listing2122SITE_URL = "https://voltigemtl.ca/plans/"23LIVYA_BASE = "https://app.livya.com"24# valeurs observées 2026-08-25 (repli si la page ne les expose plus)25DEFAULT_CLIENT = "society-dev-immo"26DEFAULT_PROJECT = "voltige"27DEFAULT_ENTITY = "2aad7013-9b35-4b56-a72f-8dbf9d063678"2829_ATTR_RE = {k: re.compile(rf'data-{k}="([^"]+)"')30            for k in ("client", "project", "entity")}31_FLIGHT_RE = re.compile(r'self\.__next_f\.push\(\[1,"((?:[^"\\]|\\.)*)"\]\)')32_UNIT_START_RE = re.compile(r'\{"unitId":"')33_BLDG_START_RE = re.compile(r'\{"buildingId":"[0-9a-f-]{36}","projectId"')343536def _balanced(s: str, i: int) -> str | None:37    """Objet JSON complet à partir de l'accolade ouvrante en position i."""38    depth = 039    in_str = esc = False40    for j in range(i, len(s)):41        ch = s[j]42        if in_str:43            if esc:44                esc = False45            elif ch == "\\":46                esc = True47            elif ch == '"':48                in_str = False49            continue50        if ch == '"':51            in_str = True52        elif ch == "{":53            depth += 154        elif ch == "}":55            depth -= 156            if depth == 0:57                return s[i:j + 1]58    return None596061def _extract(full: str, start_re: re.Pattern) -> list[dict]:62    out: list[dict] = []63    for m in start_re.finditer(full):64        obj = _balanced(full, m.start())65        if not obj:66            continue67        try:68            out.append(json.loads(obj))69        except ValueError:70            continue71    return out727374def _unit_type(rooms, bedrooms, type_type: str) -> str:75    if (type_type or "").upper() == "STUDIO":76        return "Studio"77    if isinstance(rooms, (int, float)) and rooms >= 1.5:78        return f"{int(rooms)}½"79    if isinstance(rooms, (int, float)) and 0 < rooms < 1.5:80        return "Studio"                 # modèles S1/S5… : rooms = 0.581    if isinstance(bedrooms, (int, float)) and bedrooms >= 0:82        return "Studio" if bedrooms == 0 else f"{int(bedrooms) + 2}½"83    return ""848586class VoltigeConnector(BaseConnector):87    """Le Voltige (Ahuntsic, Montréal) — unités du widget Livya."""8889    source_id = "voltige"90    request_delay = 1.091    use_detail_cache = False        # tout est inline dans le flight data9293    def _embed_params(self) -> tuple[str, str, str]:94        try:95            h = self.get(SITE_URL).text96            vals = {k: (rx.search(h).group(1) if rx.search(h) else "")97                    for k, rx in _ATTR_RE.items()}98            if all(vals.values()):99                return vals["client"], vals["project"], vals["entity"]100        except Exception:  # noqa: BLE001 — repli sur les valeurs connues101            pass102        return DEFAULT_CLIENT, DEFAULT_PROJECT, DEFAULT_ENTITY103104    def fetch(self) -> list[Listing]:105        client, project, entity = self._embed_params()106        mod_url = (f"{LIVYA_BASE}/fr/{client}/projects/{project}"107                   f"/plans/{entity}?noLayout=1")108        h = self.get(mod_url).text109        chunks = _FLIGHT_RE.findall(h)110        full = "".join(json.loads(f'"{c}"') for c in chunks)111112        buildings = {b["buildingId"]: b for b in _extract(full, _BLDG_START_RE)113                     if b.get("buildingId")}114        units: dict[str, dict] = {}115        for u in _extract(full, _UNIT_START_RE):116            uid = u.get("unitId")117            if uid:118                units[uid] = u                  # dédoublonnage (objets répétés)119120        out: list[Listing] = []121        for uid, u in units.items():122            if (u.get("availability") or "").upper() != "AVAILABLE":123                continue124            if (u.get("segment") or "").upper() != "RESIDENTIAL":125                continue126            if not u.get("rental", True):127                continue                        # location au mois seulement128            b = buildings.get(u.get("buildingId") or "") or {}129            bname = (b.get("name") or "").strip()130            address = (u.get("address") or b.get("address") or "").strip()131            city = (u.get("city") or b.get("city") or "Montréal").strip()132            lat = b.get("latitude") or None133            lng = b.get("longitude") or None134            if not lat or not lng:              # 0 = coordonnée absente135                lat = lng = None136137            price = None138            for k in ("rentalPrice", "startingAtRentalPrice"):139                v = u.get(k)140                if isinstance(v, (int, float)) and v > 0:141                    price = float(v)142                    break143            area = u.get("unitSize")144            area = float(area) if isinstance(area, (int, float)) and area > 0 else None145            bd = u.get("roomsBed")146            bd = float(bd) if isinstance(bd, (int, float)) and bd >= 0 else None147            ba = u.get("roomsBath")148            ba = float(ba) if isinstance(ba, (int, float)) and ba > 0 else None149150            images = []151            for img in u.get("typeImages") or []:152                url = (img or {}).get("fullUrl")153                if isinstance(url, str) and url.startswith("http") \154                        and url not in images:155                    images.append(url)156157            details: dict = {}158            if u.get("floorDisplayName"):159                details["floor"] = u["floorDisplayName"]160            if u.get("typeName"):161                details["model"] = u["typeName"]162            if bname:163                details["building"] = bname164165            num = str(u.get("number") or "").strip()166            title_bits = [x for x in (bname, f"unité {num}" if num else "") if x]167            out.append(Listing(168                source=self.source_id,169                external_id=uid,                # uuid Livya de l'unité (stable)170                url=SITE_URL,                   # pas de page publique par unité171                title=("Le Voltige — " + ", ".join(title_bits)).strip(" —,"),172                address=address,173                sector="Ahuntsic",174                city=city,175                unit_type=_unit_type(u.get("rooms"), bd, u.get("typeType") or ""),176                bedrooms=bd,177                bathrooms=ba,178                price=price,179                price_label=f"{price:.0f} $ /mois" if price else "",180                availability="Disponible",181                area_sqft=area,182                description=(u.get("description") or "").strip()[:2000],183                details=details,184                images=images[:30],185                lat=lat,186                lng=lng,187            ))188        return out189