# ----------------------------------------------------------------------------- # Lou-Ka — Location court terme # connectors/rezerve.py : Rëzerve / reserver.ca (Charlevoix, Laurentides, # Estrie, Lanaudière, Mauricie, Outaouais…) # # Gestionnaire québécois (~85 chalets) sur moteur Guesty. Le site reserver.ca # est un Next.js (App Router) : le catalogue /chalets est RENDU SERVEUR et le # payload React Flight (`self.__next_f.push([1,"…"])`) contient l'objet # {"listings":[…]} complet — id Guesty, slug, nom, description fr, photos, # chambres, sdb, lits, capacité, prix « à partir de » CAD, ville, adresse, # lat/lng, commodités, numéro CITQ, heures d'arrivée/départ. UNE SEULE requête # suffit, aucune page détail à visiter. # External_id = id Guesty (stable). URL publique : /chalets/. # ----------------------------------------------------------------------------- from __future__ import annotations import json import re from ..schema import StListing from .base import StConnector SITE = "https://reserver.ca" def _num(v) -> float | None: try: return float(v) if v not in (None, "") else None except (TypeError, ValueError): return None def _flight_blob(html: str) -> str: """Reconstitue le payload React Flight (chunks __next_f concaténés).""" parts = [] for c in re.findall(r'self\.__next_f\.push\(\[1,"((?:[^"\\]|\\.)*)"\]\)', html): try: parts.append(json.loads(f'"{c}"')) except ValueError: continue return "".join(parts) class Rezerve(StConnector): source_id = "rezerve" def _catalog(self) -> list[dict]: html = self.get(f"{SITE}/chalets").text blob = _flight_blob(html) i = blob.find('{"listings":[') if i < 0: return [] obj, _ = json.JSONDecoder().raw_decode(blob[i:]) return obj.get("listings") or [] @staticmethod def _description(it: dict) -> str: desc = str(it.get("description") or "") if not desc or desc.startswith("$"): # référence Flight non résolue desc = str((it.get("descriptions") or {}).get("fr") or "") if desc.startswith("$"): desc = "" return re.sub(r"\s+", " ", desc).strip()[:4000] # -- contrat ---------------------------------------------------------- def fetch(self) -> list[StListing]: listings: list[StListing] = [] for it in self._catalog(): lid = str(it.get("id") or "").strip() slug = (it.get("slug") or "").strip() title = (it.get("name") or "").strip() if not lid or not slug or not title: continue geo = it.get("geo") or {} citq = str((it.get("citq") or {}).get("number") or "").strip() street = (it.get("addressStreet") or "").strip() postal = (it.get("addressPostal") or "").strip() price = _num(it.get("priceFromCAD")) details = {k: v for k, v in { "guesty_id": lid, "area_sqft": it.get("areaSquareFeet"), "check_in": it.get("checkInTime"), "check_out": it.get("checkOutTime"), "tags": it.get("tags") or None, }.items() if v} listings.append(StListing( source=self.source_id, external_id=lid, url=f"{SITE}/chalets/{slug}", title=title, property_type="Chalet", address=" ".join(p for p in (street, postal) if p), city=(it.get("city") or "").strip(), region="", # ville + lat/lng font foi price_night=price, price_label=(f"à partir de {price:.0f} $ / nuit" if price else ""), capacity=_num(it.get("maxGuests")), bedrooms=_num(it.get("bedrooms")), beds=_num(it.get("beds")), bathrooms=_num(it.get("bathrooms")), citq=citq if re.fullmatch(r"\d{6}", citq) else "", description=self._description(it), amenities=[a for a in (it.get("amenities") or []) if isinstance(a, str)][:80], details=details, images=[u for u in (it.get("images") or []) if isinstance(u, str)][:20], lat=_num(geo.get("lat")), lng=_num(geo.get("lng")), )) return listings