# ----------------------------------------------------------------------------- # Lou-Ka — Agrégateur de logements à louer (province de Québec) # Auteur : Simon-Pierre Boucher — contact@spboucher.ai # connectors/_livya.py : base COMMUNE des connecteurs de gestionnaires servis # par la plateforme Livya (app.livya.com — nouvelle marque de RealVuu). # Généralise le patron éprouvé du connecteur thrace : les pages publiques # /fr/{client}/projects/{slug}/plans sont rendues côté serveur (Next.js) et # leur flux RSC (self.__next_f.push) contient l'inventaire JSON complet — # unités {"unitId": …} (statut, loyer, pièces, chambres/sdb, pi², adresse # civique, GPS, attributs) et immeubles {"buildingId": …}. On n'utilise PAS # api.realvuu.com (robots.txt interdit) ; le sitemap public d'app.livya.com # (robots permissif) découvre les slugs de projets du client. # Sous-classes : définir source_id, client, brand, site_url, default_city # et default_projects (repli si le sitemap casse). Voir atimco.py, cloria.py. # Clients Livya déjà couverts par des connecteurs dédiés antérieurs : thrace # (patron d'origine), somex/louis14/quartier_sila (module « plans »). # ----------------------------------------------------------------------------- from __future__ import annotations import json import re from ..schema import Listing from .base import BaseConnector LIVYA = "https://app.livya.com" SITEMAP = f"{LIVYA}/server-sitemap.xml" # fragments RSC de Next.js : self.__next_f.push([1,"…"]) _FLIGHT_RE = re.compile(r'self\.__next_f\.push\(\[1,"((?:[^"\\]|\\.)*)"\]\)') # immeuble du flux : buildingId, name et slug (ordre des champs stable) _BUILDING_RE = re.compile( r'\{"buildingId":"([0-9a-f-]+)","projectId":"[0-9a-f-]+",' r'"phaseId":[^,]*,"name":"([^"]*)","shortName":"[^"]*","slug":"([^"]*)"') def _flight_blob(html: str) -> str: """Concatène et décode les fragments RSC d'une page Livya (JSON en clair).""" return "".join(json.loads(f'"{c}"') for c in _FLIGHT_RE.findall(html)) def _unit_type(rooms) -> str: """3.5 -> « 3½ » ; < 1 -> « Studio » (finalize gère 6½+).""" try: r = float(rooms or 0) except (TypeError, ValueError): return "" if not r: return "" if r < 1: return "Studio" return f"{int(r)}½" class LivyaConnector(BaseConnector): """Base abstraite (source_id vide = non enregistrée par le registre).""" source_id = "" request_delay = 0.8 max_units = 1500 # garde-fou global client = "" # slug Livya du gestionnaire brand = "" # préfixe des titres d'annonces site_url = "" # site public du gestionnaire (référence) default_city = "" # repli si l'unité n'a pas de ville default_projects: list[str] = [] # repli si le sitemap casse # -- découverte des projets ------------------------------------------------ def _project_slugs(self) -> list[str]: """Slugs des projets du client dans le sitemap public d'app.livya.com.""" try: xml = self.get(SITEMAP).text except Exception: return list(self.default_projects) slugs = list(dict.fromkeys(re.findall( rf"{LIVYA}/fr/{re.escape(self.client)}/proj(?:ets|ects)/" rf"([a-z0-9-]+)<", xml))) return slugs or list(self.default_projects) # -- extraction du flux RSC -------------------------------------------------- @staticmethod def _units(blob: str) -> list[dict]: """Objets unité complets ({"unitId": …}) du flux, dédoublonnés.""" dec = json.JSONDecoder() seen: set[str] = set() units: list[dict] = [] for m in re.finditer(r'\{"unitId"', blob): try: obj, _ = dec.raw_decode(blob[m.start():m.start() + 60000]) except Exception: continue uid = obj.get("unitId") if uid and uid not in seen and obj.get("number"): seen.add(uid) units.append(obj) return units def fetch(self) -> list[Listing]: units: dict[str, tuple[dict, str]] = {} # unitId -> (unité, slug) buildings: dict[str, tuple[str, str]] = {} # id -> (nom, slug) for slug in self._project_slugs(): try: html = self.get(f"{LIVYA}/fr/{self.client}/projects/{slug}/plans", params={"noLayout": "1"}).text except Exception: continue # un projet cassé ne bloque pas blob = _flight_blob(html) for u in self._units(blob): units.setdefault(u["unitId"], (u, slug)) for bid, name, bslug in _BUILDING_RE.findall(blob): buildings.setdefault(bid, (name, bslug)) listings: list[Listing] = [] for u, slug in list(units.values())[: self.max_units]: if (u.get("availability") or "").upper() != "AVAILABLE": continue if not u.get("rental", True): continue # unités en vente if (u.get("segment") or "").upper() != "RESIDENTIAL": continue # locaux commerciaux exclus if (u.get("stateCode") or "QC").upper() != "QC": continue # garde-fou province (Chartwell & cie) listings.append(self._listing(u, slug, buildings)) return listings # -- une annonce par unité disponible ---------------------------------------- def _listing(self, u: dict, slug: str, buildings: dict[str, tuple[str, str]]) -> Listing: num = str(u.get("number") or "").strip() bname, _bslug = buildings.get(u.get("buildingId") or "", ("", "")) url = f"{LIVYA}/fr/{self.client}/projects/{slug}/plans" title = (f"{self.brand} ({bname}) — unité {num}" if bname else f"{self.brand} — unité {num}") address = ", ".join(x for x in ( (u.get("address") or "").strip(), (u.get("city") or "").strip(), (u.get("postalCode") or "").strip()) if x) price = u.get("rentalPrice") price = (float(price) if isinstance(price, (int, float)) and price > 1 else None) from_price = False if price is None: # prix « à partir de » (préloc/promo) base = u.get("startingAtRentalPrice") if (u.get("startingAtRental") and isinstance(base, (int, float)) and base > 1): price, from_price = float(base), True # disponibilité : date de libération future publiée, sinon « Disponible » availability = "Disponible" avail_date = None fut = u.get("futureAvailability") or {} starts = str(fut.get("startsOn") or "")[:10] if re.fullmatch(r"20\d{2}-\d{2}-\d{2}", starts): avail_date = starts availability = f"Disponible à partir du {starts}" # superficie structurée (measurementSystem IMPERIAL -> déjà en pi²) area = None if (u.get("measurementSystem") or "IMPERIAL").upper() == "IMPERIAL": size = u.get("unitSize") if isinstance(size, (int, float)) and size > 0: area = float(size) amenities = [a["name"] for a in (u.get("attributes") or []) if isinstance(a, dict) and a.get("name")] desc: list[str] = [] floor_disp = str(u.get("floorDisplayName") or "").strip() if floor_disp: desc.append(floor_disp) if u.get("typeName"): desc.append(f"Modèle {u['typeName']}") if u.get("roomsBed"): desc.append(f"{u['roomsBed']} chambre(s)") if u.get("roomsBath"): desc.append(f"{u['roomsBath']} salle(s) de bain") if u.get("balconySize"): desc.append(f"Balcon de {u['balconySize']} pi²") precision = str(u.get("pricePrecision") or "").strip() if precision: desc.append(precision) libre = str(u.get("description") or "").strip() if libre: desc.append(libre) if u.get("floorPlanUrl"): desc.append(f"Plan : {u['floorPlanUrl']}") details: dict = {"gestionnaire": self.site_url or self.brand} if bname: details["building"] = bname floor_no = str(u.get("floorNumber") or "") if floor_no.isdigit(): details["floor"] = int(floor_no) images = [img.get("fullUrl") for img in (u.get("typeImages") or []) if isinstance(img, dict) and img.get("fullUrl")] if u.get("floorPlanImageUrl"): images.append(u["floorPlanImageUrl"]) try: lat = float(u["latitude"]) if u.get("latitude") else None lng = float(u["longitude"]) if u.get("longitude") else None except (TypeError, ValueError): lat = lng = None beds = u.get("roomsBed") baths = u.get("roomsBath") if baths is not None and u.get("roomsWater"): try: baths = float(baths) + 0.5 * float(u["roomsWater"]) except (TypeError, ValueError): pass return Listing( source=self.source_id, external_id=str(u["unitId"]), # UUID Livya : stable url=url, title=title, address=address, sector="", # non exposé par le flux city=(u.get("city") or "").strip() or self.default_city, unit_type=_unit_type(u.get("rooms")), bedrooms=(float(beds) if isinstance(beds, (int, float)) and beds else None), bathrooms=(float(baths) if isinstance(baths, (int, float)) and baths else None), price=price, price_label=((f"à partir de {price:.0f} $ /mois" if from_price else f"{price:.0f} $ /mois") if price else ""), availability=availability, availability_date=avail_date, area_sqft=area, description=" | ".join(desc), amenities=amenities, details=details, images=images[:12], lat=lat, lng=lng, )