# ----------------------------------------------------------------------------- # Lou-Ka — Agrégateur de logements à louer (province de Québec) # Auteur : Simon-Pierre Boucher — contact@spboucher.ai # connectors/rivero.py : connecteur Le Rivero (lerivero.ca — Québec, # rue Bourdages, aux abords de la rivière Saint-Charles). # Le sélecteur de plans est un embed RealVuu (app.realvuu.com) dont la page # contient toutes les unités en JSON (numéro, pièces, prix, disponibilité, # plans/images). Phase 1 louée; phase 2 (livraison été 2028) en prélocation. # Garde-fou : si toutes les unités « disponibles » affichent un prix # identique (placeholder), le prix n'est pas retenu. # ----------------------------------------------------------------------------- from __future__ import annotations import json import re from ..schema import Listing, infer_city from .base import BaseConnector RV_URL = "https://app.realvuu.com/fr/external/immeubles-simard/rivero/plans" SITE_URL = "https://www.lerivero.ca/plans/" SECTOR = "Vanier (rivière Saint-Charles)" def _extract_json_array(html: str, key: str) -> list: """Extrait le plus grand tableau JSON `"key":[...]` de la page RealVuu. La clé peut apparaître plusieurs fois (souvent vide ailleurs) : on garde l'occurrence contenant le plus d'éléments. """ dec = json.JSONDecoder() needle = f'"{key}":[' best: list = [] start = 0 while True: i = html.find(needle, start) if i == -1: break try: arr, _ = dec.raw_decode(html[i + len(needle) - 1:]) if isinstance(arr, list) and len(arr) > len(best): best = arr except Exception: pass start = i + len(needle) return best def _unit_type_from_rooms(rooms: float, rooms_bed: int = 0) -> str: if not rooms: # repli structuré : n chambres -> (n+2)½ (convention Lou-Ka) if rooms_bed: n = int(rooms_bed) + 2 return "6½+" if n >= 6 else f"{n}½" return "" if rooms <= 1: return "Studio" return f"{int(rooms)}½" class RiveroConnector(BaseConnector): source_id = "rivero" request_delay = 0.6 def fetch(self) -> list[Listing]: html = self.get(RV_URL).text units = _extract_json_array(html, "units") buildings = {b.get("buildingId"): b for b in _extract_json_array(html, "buildings")} # étages : floorId -> numéro (structuré, pour details.floor) floors = {f.get("floorId"): f.get("number") for f in _extract_json_array(html, "floors") if f.get("floorId")} avail = [u for u in units if u.get("availability") == "AVAILABLE" and u.get("rental") and u.get("typeType") == "APARTMENT" and u.get("segment") in ("RESIDENTIAL", "", None) and u.get("isMarketable", True)] if not avail: return [] # Garde-fou prix placeholder : prix unique pour des types différents prices = {u.get("rentalPrice") for u in avail} rooms_set = {u.get("rooms") for u in avail} placeholder = len(prices) == 1 and len(rooms_set) > 1 listings: list[Listing] = [] for u in avail: try: b = buildings.get(u.get("buildingId"), {}) bname = (b.get("name") or "").strip() address = (u.get("address") or b.get("address") or "").strip() # compléter l'adresse avec ville + code postal (structurés) city_src = (u.get("city") or b.get("city") or "").strip() postal = (u.get("postalCode") or "").strip() if address and city_src: address = f"{address}, {city_src}" \ + (f", QC {postal}" if postal else "") number = str(u.get("number") or "").strip() unit_type = _unit_type_from_rooms(u.get("rooms") or 0, u.get("roomsBed") or 0) price = None price_label = "" rp = u.get("rentalPrice") or 0 if rp and not placeholder and 100 <= rp <= 20000: price = float(rp) price_label = f"{rp} $/mois" availability = "Disponible" if u.get("futureAvailability"): availability = f"Disponible : {u['futureAvailability']}" if "phase 2" in bname.lower(): availability = "Phase 2 — prélocation (livraison été 2028)" images: list[str] = [] for ti in (u.get("typeImages") or []): if ti.get("fullUrl"): images.append(ti["fullUrl"]) for im in (u.get("images") or []): if isinstance(im, dict) and im.get("fullUrl"): images.append(im["fullUrl"]) if u.get("floorPlanImageUrl"): images.append(u["floorPlanImageUrl"]) images = list(dict.fromkeys(images)) desc = [] if u.get("unitSize"): desc.append(f"{u['unitSize']} pi²") if u.get("roomsBed"): desc.append(f"{u['roomsBed']} chambre(s)") if u.get("roomsOffice"): desc.append(f"{u['roomsOffice']} espace(s) bureau") if u.get("roomsBath"): desc.append(f"{u['roomsBath']} salle(s) de bain") if u.get("balconySize"): desc.append(f"Balcon/loggia {u['balconySize']} pi²") if u.get("orientation"): desc.append(f"Orientation {u['orientation']}") if bname: desc.append(bname) # superficie structurée (mesures IMPERIAL -> déjà en pi²) area = None us = u.get("unitSize") or 0 if 80 <= us <= 20000: area = float(us) # champs structurés : étage, coordonnées details: dict = {} floor_no = floors.get(u.get("floorId")) if isinstance(floor_no, int) and 0 < floor_no <= 60: details["floor"] = floor_no lat = u.get("latitude") lng = u.get("longitude") if not (lat and lng): lat, lng = b.get("latitude"), b.get("longitude") listings.append(Listing( source=self.source_id, external_id=u.get("unitId") or f"{bname}-{number}", url=SITE_URL, title=f"Le Rivero — Unité {number}" f"{' (' + unit_type + ')' if unit_type else ''}", address=address, sector=SECTOR, city=infer_city(SECTOR, default="Québec"), unit_type=unit_type, price=price, price_label=price_label, availability=availability, area_sqft=area, description=" | ".join(desc), amenities=["Eau chaude incluse", "Électricité et chauffage", "Air climatisé", "Internet inclus"], details=details, images=images, lat=float(lat) if lat else None, lng=float(lng) if lng else None, )) except Exception: continue return listings