# ----------------------------------------------------------------------------- # Lou-Ka — Agrégateur de logements à louer (province de Québec) # Auteur : Simon-Pierre Boucher — contact@spboucher.ai # connectors/proprio_direct.py : Proprio Direct (propriodirect.com) — LOCATION # L'inventaire (alimenté par le feed DDF de CREA) est servi par une API JSON # interne : POST https://propriodirect.com/fr/api/searchListings # Corps { "from": , "filter": {"rentOrSale": "rent"} } — le filtre # côté API fonctionne (~260 locations vs ~3200 annonces au total) ; 30 # résultats/page, champ `total`. Chaque item porte forRent/forSale/sold, # le loyer en cents (rentPriceInCents, rentPriceFrequency), adresse, ville, # région, chambres, sdb, GPS, tags et galerie photos. On ne garde que le # volet RÉSIDENTIEL (tag « residential ») : les locaux commerciaux et # industriels (loyer en $/pi²/an) sont écartés. Adapté du connecteur # « à vendre » d'Immo-Ka (agent-courtage/immoka). # ----------------------------------------------------------------------------- from __future__ import annotations import os import re from ..schema import Listing from .base import BaseConnector from . import _detailutil as du SITE = "https://propriodirect.com" SEARCH_URL = f"{SITE}/fr/api/searchListings" PAGE_SIZE = 30 MAX_PAGES = 60 # garde-fou (~260 locations = 9 pages) DETAIL_LIMIT = int(os.environ.get("LOUKA_PD_DETAIL_LIMIT", "400")) # tags source (feed DDF) -> caractéristiques lisibles _TAG_LABELS = { "car_garage": "Garage", "car_parking": "Stationnement", "car_foyer": "Foyer", "car_piscine": "Piscine", "car_climatisation": "Climatisation", "car_sous_sol": "Sous-sol", "car_bord_eau": "Bord de l'eau", "prox_park": "Parc à proximité", "prox_ecole_primaire": "École primaire", "prox_ecole_secondaire": "École secondaire", "prox_transport": "Transport en commun", "prox_garderie": "Garderie", "prox_piste_cyclable": "Piste cyclable", "prox_hopital": "Hôpital", "prox_autoroute": "Autoroute", "new_price": "Nouveau prix", } class ProprioDirectConnector(BaseConnector): source_id = "proprio_direct" request_delay = 0.4 def _headers(self) -> dict: return { "Accept": "application/json", "Content-Type": "application/json", "Origin": SITE, "Referer": f"{SITE}/propriete-a-louer/", } def fetch(self) -> list[Listing]: out: list[Listing] = [] seen: set[str] = set() offset = 0 for _ in range(MAX_PAGES): try: data = self.post(SEARCH_URL, headers=self._headers(), json={"from": offset, "includeGeoJson": False, "filter": {"rentOrSale": "rent"}}).json() except Exception: break items = data.get("listings", []) if isinstance(data, dict) else [] if not items: break for it in items: lst = self._to_listing(it) if lst and lst.uid not in seen: seen.add(lst.uid) out.append(lst) total = data.get("total") or 0 offset += PAGE_SIZE if offset >= total: break # fiche détail : description (la galerie est déjà complète via l'API) du.enrich(self, out, DETAIL_LIMIT, _parse_pd_detail, key="v3") return out def _to_listing(self, it: dict) -> Listing | None: rid = it.get("id") tags = it.get("tags") or [] # location résidentielle seulement (ceinture + bretelles : le filtre # API renvoie déjà uniquement des locations, on revalide chaque item) if (not rid or it.get("sold") or not it.get("forRent") or "residential" not in tags): return None # loyer mensuel : rentPriceInCents (fallback priceInCents si le feed # met le loyer dans le prix principal) — on écarte les baux annuels # au pi² qui se seraient glissés dans le volet résidentiel freq = it.get("rentPriceFrequency") or "monthly" if freq != "monthly": return None cents = it.get("rentPriceInCents") or it.get("priceInCents") price = (cents / 100.0) if isinstance(cents, (int, float)) and cents > 0 else None slug = it.get("slugURLFr") or it.get("slugURL") or "" url = f"{SITE}{slug}" if slug.startswith("/") else (slug or SITE) geo = it.get("geoLocation") or {} # type d'unité : nb de chambres (normalisé n+2 ½ par finalize()) ; # le genre Loft/Studio (LS) sans chambre est un studio genre = (it.get("genre") or "").upper() rooms = _pos_int(it.get("numberOfRooms")) unit_type = f"{rooms} chambres" if rooms else ("Studio" if genre == "LS" else "") details: dict = {"Courtier": "Proprio Direct"} if it.get("genreName"): details["Type"] = it["genreName"] if it.get("regionName"): details["Région"] = it["regionName"] baths = _pos_int(it.get("numberOfBathrooms")) if baths: details["Salles de bain"] = str(baths) if it.get("postalCode"): details["Code postal"] = it["postalCode"] return Listing( source=self.source_id, external_id=str(rid), url=url, title=it.get("addressLine", ""), address=it.get("addressLine", ""), sector=it.get("neighbourhood", "") or "", city=it.get("cityName", ""), unit_type=unit_type, price=price, price_label=it.get("rentPriceText", "") or "", amenities=[_TAG_LABELS[t] for t in tags if t in _TAG_LABELS], details=details, images=list(it.get("photosOriginal") or it.get("photosSmall") or []), lat=geo.get("lat"), lng=geo.get("lon"), ) # libellés des « Points saillants » d'une fiche Proprio Direct — la page les # rend en liste « libellé | valeur » stricte (contrairement aux fiches Centris, # d'où un parseur dédié plutôt que du.centris_details qui appariait de travers) _PD_LABELS = [ "Année de construction", "Superficie habitable", "Niveau", "Équipement disponible", "Zonage", "Mode de chauffage", "Énergie pour le chauffage", "Approvisionnement en eau", "Système d'égouts", "Restrictions/Permissions", "Garage", "Piscine", "Foyer-Poêle", "Meublé", "Animaux", "Disponibilité", "Date d'emménagement", "Bail", ] _PD_SECTIONS = {"points saillants", "détails", "pièces", "inclus", "exclus", "style de vie", "description"} def _pd_details(text: str) -> dict: """Paires « libellé | valeur » des Points saillants (ordre strict).""" stop = {l.lower() for l in _PD_LABELS} | _PD_SECTIONS out: dict = {} for label in _PD_LABELS: m = re.search(re.escape(label) + r"\s*\|\s*([^|]{1,90})", text) if m: val = m.group(1).strip(" |") if val and val.lower() not in stop: out[label] = val return out def _parse_pd_detail(html: str) -> dict: """Fiche Proprio Direct : description + points saillants (photos déjà fournies par l'API).""" out: dict = {} desc = du.ld_description(html) # le JSON-LD des fiches sans texte retombe sur la description « méta » # générique (« … | No. Inscription 12345678 | Proprio Direct ») — on l'écarte if desc and "No. Inscription" not in desc: out["description"] = desc det = _pd_details(du.flatten(html)) if det: out.setdefault("details", {}).update(det) # « Animaux permis sous conditions » / « Animaux non admis »… restr = det.get("Restrictions/Permissions", "").lower() if "animaux" in restr: out["pets"] = ("conditions" if "condition" in restr else "non" if ("non" in restr or "pas" in restr) else "oui") return out def _pos_int(v): try: n = int(v) return n if n > 0 else None except (TypeError, ValueError): return None