# ----------------------------------------------------------------------------- # Lou-Ka — Agrégateur de logements à louer (province de Québec) # Auteur : Simon-Pierre Boucher — contact@spboucher.ai # connectors/thrace.py : connecteur Gestions Thrace (gestionsthrace.com) # Plus gros gestionnaire de la Mauricie (~2 500 logements — Trois-Rivières, # Cap-de-la-Madeleine, Bécancour). Site WordPress (Bricks) ; la page # /fr/residential embarque le module « inventory » Livya (app.livya.com, # client « thrace »). Contrairement au module « plans » (somex, louis14, # quartier_sila), le module inventory charge ses unités côté client via # api.realvuu.com — dont le robots.txt interdit tout accès : on ne l'utilise # PAS. À la place, les pages « plans » des trois projets Livya du client # (thrace, radisson, carree-radisson — découverts via le sitemap public # d'app.livya.com, robots permissif) sont rendues côté serveur : leur flux # RSC (self.__next_f.push) contient l'inventaire JSON complet — unités # ({"unitId": …} : statut, loyer, pièces, inclusions, adresse civique, GPS, # photos) et immeubles ({"buildingId": …} : nom, slug). 5 requêtes par sync. # Unités AVAILABLE, résidentielles et locatives seulement (prix réels). # ----------------------------------------------------------------------------- from __future__ import annotations import json import re from ..schema import Listing from .base import BaseConnector BASE = "https://www.gestionsthrace.com" LIST_URL = f"{BASE}/fr/residential" LIVYA = "https://app.livya.com" SITEMAP = f"{LIVYA}/server-sitemap.xml" # Valeurs observées au 2026-08 — replis si la découverte dynamique casse DEFAULT_CLIENT = "thrace" DEFAULT_ENTITY = "fb25970c-6c5f-4253-a396-7669992522b0" # module inventory DEFAULT_PROJECTS = ["thrace", "radisson", "carree-radisson"] # 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 ThraceConnector(BaseConnector): source_id = "thrace" request_delay = 0.8 max_units = 1500 # garde-fou (~430 unités publiées par projet) # -- découverte des paramètres Livya ---------------------------------------- def _module_params(self) -> tuple[str, str]: """(client, entity) lus sur /fr/residential, replis constants.""" client, entity = DEFAULT_CLIENT, DEFAULT_ENTITY try: page = self.get(LIST_URL).text mc = re.search(r']*data-client="([^"]+)"', page) if mc: client = mc.group(1).strip() or client md = re.search( r'<[^>]*livya-module-container-inventory[^>]*' r'data-entity="([0-9a-f-]+)"', page) if md: entity = md.group(1) except Exception: pass # replis : les constantes observées return client, entity def _project_slugs(self, client: str) -> 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(DEFAULT_PROJECTS) slugs = list(dict.fromkeys(re.findall( rf"{LIVYA}/fr/{re.escape(client)}/proj(?:ets|ects)/([a-z0-9-]+)<", xml))) return slugs or list(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]: client, entity = self._module_params() projects = self._project_slugs(client) units: dict[str, dict] = {} buildings: dict[str, tuple[str, str]] = {} # id -> (nom, slug) for slug in projects: try: html = self.get( f"{LIVYA}/fr/{client}/projects/{slug}/plans/{entity}", 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) for bid, name, bslug in _BUILDING_RE.findall(blob): buildings.setdefault(bid, (name, bslug)) listings: list[Listing] = [] for u 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 listings.append(self._listing(u, buildings)) return listings # -- une annonce par unité disponible ---------------------------------------- def _listing(self, u: dict, buildings: dict[str, tuple[str, str]]) -> Listing: num = str(u.get("number") or "").strip() bname, bslug = buildings.get(u.get("buildingId") or "", ("", "")) # le portail WordPress accepte ?building= (module inventory) url = f"{LIST_URL}?building={bslug}" if bslug else LIST_URL title = (f"Gestions Thrace ({bname}) — unité {num}" if bname else f"Gestions Thrace — 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 > 0 else None # 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) # inclusions et caractéristiques publiées par unité (« Eau chaude », # « Animaux interdits », « Stationnement: 1 extérieur »…) amenities = [a["name"] for a in (u.get("attributes") or []) if isinstance(a, dict) and a.get("name")] # description : étage, modèle, pièces, balcon, précision de prix, plan 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 = {} if bname: details["building"] = bname floor_no = str(u.get("floorNumber") or "") if floor_no.isdigit(): details["floor"] = int(floor_no) if u.get("roomsBed"): details["bedrooms"] = int(u["roomsBed"]) if u.get("roomsBath"): details["bathrooms"] = int(u["roomsBath"]) 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 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 "Trois-Rivières", unit_type=_unit_type(u.get("rooms")), price=price, price_label=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, )