# ----------------------------------------------------------------------------- # Lou-Ka — Agrégateur de logements à louer (province de Québec) # Auteur : Simon-Pierre Boucher — contact@spboucher.ai # connectors/louis14.py : connecteur Quartier Louis Quatorze (louis14.ca) # 424 condos locatifs (immeubles A-B-C-D), Lebourgneuf, Québec (Groupe Damco). # Le site WordPress n'affiche aucun prix statiquement : la page /plans/ # embarque le sélecteur de plans Livya (app.livya.com, ex-RealVuu). La page # Next.js du sélecteur est rendue serveur : son flux RSC (`self.__next_f # .push`) contient les 423 unités du projet avec prix de location réel, # disponibilité, superficie, étage, orientation, balcon, photos du type et # plan d'étage — on le parse directement, sans exécuter de JavaScript. # (403 sans UA navigateur ; le UA de base.py passe, Cloudflare accepte.) # Seules les unités AVAILABLE deviennent des annonces (prix réels). # ----------------------------------------------------------------------------- from __future__ import annotations import json import re from ..schema import Listing, normalize_unit_type from .base import BaseConnector BASE = "https://louis14.ca" PLANS_PAGE = f"{BASE}/plans/" LIVYA = "https://app.livya.com" # Valeurs observées sur /plans/ — repli si l'extraction dynamique casse DEFAULT_CLIENT = "damco" DEFAULT_PROJECT = "louis-14" DEFAULT_ENTITY = "e5e643c4-89da-4f82-89e5-8895b67c9b3e" # sélecteur de plan # conteneur du module Livya sur /plans/ (l'ordre des attributs peut varier) _CONTAINER_RE = re.compile( r']*class="livya-module-container-plans"[^>]*>', re.I) _ATTR_RE = re.compile(r'data-(project|entity|lang)="([^"]*)"') _CLIENT_RE = re.compile(r']*data-client="([^"]+)"[^>]*>') # fragments RSC de Next.js : self.__next_f.push([1,"…"]) _FLIGHT_RE = re.compile(r'self\.__next_f\.push\(\[1,"((?:[^"\\]|\\.)*)"\]\)') class Louis14Connector(BaseConnector): source_id = "louis14" request_delay = 0.7 max_units = 500 # garde-fou (423 unités au projet) # -- paramètres du module Livya (page /plans/, avec replis) ---------------- def _module_params(self) -> tuple[str, str, str, str]: """(client, project, entity, lang) lus sur /plans/, replis constants.""" client, project, entity, lang = (DEFAULT_CLIENT, DEFAULT_PROJECT, DEFAULT_ENTITY, "fr") try: page = self.get(PLANS_PAGE).text mc = _CLIENT_RE.search(page) if mc: client = mc.group(1).strip() or client md = _CONTAINER_RE.search(page) if md: attrs = dict(_ATTR_RE.findall(md.group(0))) project = attrs.get("project") or project entity = attrs.get("entity") or entity lang = attrs.get("lang") or lang except Exception: pass # replis : les constantes observées return client, project, entity, lang def fetch(self) -> list[Listing]: client, project, entity, lang = self._module_params() html = self.get( f"{LIVYA}/{lang}/{client}/projects/{project}/plans/{entity}").text # flux RSC : les fragments sont des corps de chaînes JSON — les # recoller puis décoder donne le JSON des unités en clair blob = "".join(json.loads(f'"{c}"') for c in _FLIGHT_RE.findall(html)) buildings = self._buildings(blob) units = self._units(blob) listings: list[Listing] = [] for u in units[: self.max_units]: if (u.get("availability") or "").upper() != "AVAILABLE": continue # 405 unités NOT_AVAILABLE if (u.get("stateCode") or "").upper() != "QC": continue # garde-fou province listings.append(self._listing(u, buildings)) return listings # -- 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() + 30000]) 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 @staticmethod def _buildings(blob: str) -> dict[str, str]: """buildingId -> nom d'immeuble (« QUARTIER A-B », « QUARTIER C-D »).""" out: dict[str, str] = {} for m in re.finditer( r'\{"buildingId":"([0-9a-f-]+)","projectId":"[0-9a-f-]+",' r'"phaseId":[^,]*,"name":"([^"]*)"', blob): out.setdefault(m.group(1), m.group(2)) return out # -- une annonce par unité disponible ---------------------------------------- def _listing(self, u: dict, buildings: dict[str, str]) -> Listing: num = str(u.get("number") or "").strip() bname = (buildings.get(u.get("buildingId") or "") or "").strip().title() title = f"Louis 14 — unité {num}" if bname: title = f"Louis 14 ({bname}) — unité {num}" # adresse civique portée par chaque unité dans le flux street = (u.get("address") or "").strip() city = (u.get("city") or "").strip() postal = (u.get("postalCode") or "").strip() full_addr = ", ".join(x for x in (street, city, postal) if x) price = float(u["rentalPrice"]) if u.get("rentalPrice") else None # disponibilité : AVAILABLE = libre ; futureAvailability.startsOn # (ISO) = date structurée de libération 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 "").upper() == "IMPERIAL": size = u.get("unitSize") if isinstance(size, (int, float)) and size > 0: area = float(size) # étage / orientation / balcon : champs structurés du flux details: dict = {} floor_disp = (u.get("floorDisplayName") or "").strip() try: details["floor"] = int(u.get("floorNumber")) except (TypeError, ValueError): pass orientation = (u.get("orientation") or "").strip() if orientation: details["orientation"] = orientation balcony = u.get("balconySize") amenities: list[str] = [] if isinstance(balcony, (int, float)) and balcony > 0: details["balcony"] = True details["balcony_sqft"] = float(balcony) amenities.append(f"Balcon de {balcony:g} pi²") if u.get("roomsBed"): details["bedrooms"] = int(u["roomsBed"]) if u.get("roomsBath"): details["bathrooms"] = int(u["roomsBath"]) if floor_disp: amenities.append(floor_disp) if orientation: amenities.append(f"Orientation {orientation}") # photos du type d'unité + plan d'étage (URLs absolues CloudFront) images: list[str] = [] for img in u.get("typeImages") or []: url = (img or {}).get("fullUrl") if url and url not in images: images.append(url) plan = u.get("floorPlanImageUrl") if plan and plan not in images: images.append(plan) 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 return Listing( source=self.source_id, external_id=num, # « 816-C » : stable et unique url=PLANS_PAGE, # les unités n'ont pas de page propre title=title, address=full_addr, sector="", # quartier non exposé par le flux city=city or "Québec", unit_type=normalize_unit_type(str(u.get("rooms") or "")), price=price, price_label=f"{price:.0f} $ /mois" if price else "", availability=availability, availability_date=avail_date, area_sqft=area, description=(u.get("description") or "").strip()[:600], amenities=amenities, details=details, images=images[:15], lat=lat, lng=lng, )