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/_livya.py : base COMMUNE des connecteurs de gestionnaires servis5#   par la plateforme Livya (app.livya.com — nouvelle marque de RealVuu).6#   Généralise le patron éprouvé du connecteur thrace : les pages publiques7#   /fr/{client}/projects/{slug}/plans sont rendues côté serveur (Next.js) et8#   leur flux RSC (self.__next_f.push) contient l'inventaire JSON complet —9#   unités {"unitId": …} (statut, loyer, pièces, chambres/sdb, pi², adresse10#   civique, GPS, attributs) et immeubles {"buildingId": …}. On n'utilise PAS11#   api.realvuu.com (robots.txt interdit) ; le sitemap public d'app.livya.com12#   (robots permissif) découvre les slugs de projets du client.13#   Sous-classes : définir source_id, client, brand, site_url, default_city14#   et default_projects (repli si le sitemap casse). Voir atimco.py, cloria.py.15#   Clients Livya déjà couverts par des connecteurs dédiés antérieurs : thrace16#   (patron d'origine), somex/louis14/quartier_sila (module « plans »).17# -----------------------------------------------------------------------------18from __future__ import annotations1920import json21import re2223from ..schema import Listing24from .base import BaseConnector2526LIVYA = "https://app.livya.com"27SITEMAP = f"{LIVYA}/server-sitemap.xml"2829# fragments RSC de Next.js : self.__next_f.push([1,"…"])30_FLIGHT_RE = re.compile(r'self\.__next_f\.push\(\[1,"((?:[^"\\]|\\.)*)"\]\)')31# immeuble du flux : buildingId, name et slug (ordre des champs stable)32_BUILDING_RE = re.compile(33    r'\{"buildingId":"([0-9a-f-]+)","projectId":"[0-9a-f-]+",'34    r'"phaseId":[^,]*,"name":"([^"]*)","shortName":"[^"]*","slug":"([^"]*)"')353637def _flight_blob(html: str) -> str:38    """Concatène et décode les fragments RSC d'une page Livya (JSON en clair)."""39    return "".join(json.loads(f'"{c}"') for c in _FLIGHT_RE.findall(html))404142def _unit_type(rooms) -> str:43    """3.5 -> « 3½ » ; < 1 -> « Studio » (finalize gère 6½+)."""44    try:45        r = float(rooms or 0)46    except (TypeError, ValueError):47        return ""48    if not r:49        return ""50    if r < 1:51        return "Studio"52    return f"{int(r)}½"535455class LivyaConnector(BaseConnector):56    """Base abstraite (source_id vide = non enregistrée par le registre)."""5758    source_id = ""59    request_delay = 0.860    max_units = 1500                 # garde-fou global6162    client = ""                      # slug Livya du gestionnaire63    brand = ""                       # préfixe des titres d'annonces64    site_url = ""                    # site public du gestionnaire (référence)65    default_city = ""                # repli si l'unité n'a pas de ville66    default_projects: list[str] = []  # repli si le sitemap casse6768    # -- découverte des projets ------------------------------------------------69    def _project_slugs(self) -> list[str]:70        """Slugs des projets du client dans le sitemap public d'app.livya.com."""71        try:72            xml = self.get(SITEMAP).text73        except Exception:74            return list(self.default_projects)75        slugs = list(dict.fromkeys(re.findall(76            rf"{LIVYA}/fr/{re.escape(self.client)}/proj(?:ets|ects)/"77            rf"([a-z0-9-]+)<", xml)))78        return slugs or list(self.default_projects)7980    # -- extraction du flux RSC --------------------------------------------------81    @staticmethod82    def _units(blob: str) -> list[dict]:83        """Objets unité complets ({"unitId": …}) du flux, dédoublonnés."""84        dec = json.JSONDecoder()85        seen: set[str] = set()86        units: list[dict] = []87        for m in re.finditer(r'\{"unitId"', blob):88            try:89                obj, _ = dec.raw_decode(blob[m.start():m.start() + 60000])90            except Exception:91                continue92            uid = obj.get("unitId")93            if uid and uid not in seen and obj.get("number"):94                seen.add(uid)95                units.append(obj)96        return units9798    def fetch(self) -> list[Listing]:99        units: dict[str, tuple[dict, str]] = {}      # unitId -> (unité, slug)100        buildings: dict[str, tuple[str, str]] = {}   # id -> (nom, slug)101        for slug in self._project_slugs():102            try:103                html = self.get(f"{LIVYA}/fr/{self.client}/projects/{slug}/plans",104                                params={"noLayout": "1"}).text105            except Exception:106                continue                 # un projet cassé ne bloque pas107            blob = _flight_blob(html)108            for u in self._units(blob):109                units.setdefault(u["unitId"], (u, slug))110            for bid, name, bslug in _BUILDING_RE.findall(blob):111                buildings.setdefault(bid, (name, bslug))112113        listings: list[Listing] = []114        for u, slug in list(units.values())[: self.max_units]:115            if (u.get("availability") or "").upper() != "AVAILABLE":116                continue117            if not u.get("rental", True):118                continue                 # unités en vente119            if (u.get("segment") or "").upper() != "RESIDENTIAL":120                continue                 # locaux commerciaux exclus121            if (u.get("stateCode") or "QC").upper() != "QC":122                continue                 # garde-fou province (Chartwell & cie)123            listings.append(self._listing(u, slug, buildings))124        return listings125126    # -- une annonce par unité disponible ----------------------------------------127    def _listing(self, u: dict, slug: str,128                 buildings: dict[str, tuple[str, str]]) -> Listing:129        num = str(u.get("number") or "").strip()130        bname, _bslug = buildings.get(u.get("buildingId") or "", ("", ""))131132        url = f"{LIVYA}/fr/{self.client}/projects/{slug}/plans"133        title = (f"{self.brand} ({bname}) — unité {num}" if bname134                 else f"{self.brand} — unité {num}")135136        address = ", ".join(x for x in (137            (u.get("address") or "").strip(), (u.get("city") or "").strip(),138            (u.get("postalCode") or "").strip()) if x)139140        price = u.get("rentalPrice")141        price = (float(price)142                 if isinstance(price, (int, float)) and price > 1 else None)143        from_price = False144        if price is None:                # prix « à partir de » (préloc/promo)145            base = u.get("startingAtRentalPrice")146            if (u.get("startingAtRental")147                    and isinstance(base, (int, float)) and base > 1):148                price, from_price = float(base), True149150        # disponibilité : date de libération future publiée, sinon « Disponible »151        availability = "Disponible"152        avail_date = None153        fut = u.get("futureAvailability") or {}154        starts = str(fut.get("startsOn") or "")[:10]155        if re.fullmatch(r"20\d{2}-\d{2}-\d{2}", starts):156            avail_date = starts157            availability = f"Disponible à partir du {starts}"158159        # superficie structurée (measurementSystem IMPERIAL -> déjà en pi²)160        area = None161        if (u.get("measurementSystem") or "IMPERIAL").upper() == "IMPERIAL":162            size = u.get("unitSize")163            if isinstance(size, (int, float)) and size > 0:164                area = float(size)165166        amenities = [a["name"] for a in (u.get("attributes") or [])167                     if isinstance(a, dict) and a.get("name")]168169        desc: list[str] = []170        floor_disp = str(u.get("floorDisplayName") or "").strip()171        if floor_disp:172            desc.append(floor_disp)173        if u.get("typeName"):174            desc.append(f"Modèle {u['typeName']}")175        if u.get("roomsBed"):176            desc.append(f"{u['roomsBed']} chambre(s)")177        if u.get("roomsBath"):178            desc.append(f"{u['roomsBath']} salle(s) de bain")179        if u.get("balconySize"):180            desc.append(f"Balcon de {u['balconySize']} pi²")181        precision = str(u.get("pricePrecision") or "").strip()182        if precision:183            desc.append(precision)184        libre = str(u.get("description") or "").strip()185        if libre:186            desc.append(libre)187        if u.get("floorPlanUrl"):188            desc.append(f"Plan : {u['floorPlanUrl']}")189190        details: dict = {"gestionnaire": self.site_url or self.brand}191        if bname:192            details["building"] = bname193        floor_no = str(u.get("floorNumber") or "")194        if floor_no.isdigit():195            details["floor"] = int(floor_no)196197        images = [img.get("fullUrl") for img in (u.get("typeImages") or [])198                  if isinstance(img, dict) and img.get("fullUrl")]199        if u.get("floorPlanImageUrl"):200            images.append(u["floorPlanImageUrl"])201202        try:203            lat = float(u["latitude"]) if u.get("latitude") else None204            lng = float(u["longitude"]) if u.get("longitude") else None205        except (TypeError, ValueError):206            lat = lng = None207208        beds = u.get("roomsBed")209        baths = u.get("roomsBath")210        if baths is not None and u.get("roomsWater"):211            try:212                baths = float(baths) + 0.5 * float(u["roomsWater"])213            except (TypeError, ValueError):214                pass215216        return Listing(217            source=self.source_id,218            external_id=str(u["unitId"]),          # UUID Livya : stable219            url=url,220            title=title,221            address=address,222            sector="",                             # non exposé par le flux223            city=(u.get("city") or "").strip() or self.default_city,224            unit_type=_unit_type(u.get("rooms")),225            bedrooms=(float(beds)226                      if isinstance(beds, (int, float)) and beds else None),227            bathrooms=(float(baths)228                       if isinstance(baths, (int, float)) and baths else None),229            price=price,230            price_label=((f"à partir de {price:.0f} $ /mois" if from_price231                          else f"{price:.0f} $ /mois") if price else ""),232            availability=availability,233            availability_date=avail_date,234            area_sqft=area,235            description=" | ".join(desc),236            amenities=amenities,237            details=details,238            images=images[:12],239            lat=lat,240            lng=lng,241        )242