# ----------------------------------------------------------------------------- # Lou-Ka — Agrégateur de logements à louer (province de Québec) # Auteur : Simon-Pierre Boucher — contact@spboucher.ai # connectors/centurion.py : connecteur Centurion / CP Living (cpliving.com) # REIT national (Centurion Property Associates) — région de Québec : le # Quartier QB, 2551 chemin des Quatre-Bourgeois (Sainte-Foy), 684 unités. # Le site est propulsé par Rentsync : la page de recherche référence un # proxy legacy `widgets/search/api_proxy.php?client_id=21&city_id=2310` # (404 aujourd'hui) mais son JS (scripts/main.js) appelle en réalité le # flux JSON officiel `https://api.theliftsystem.com/v2/search` avec un # auth_token embarqué. On lit client_id/city_id sur la page, le token dans # main.js (repli : constantes), puis on interroge le flux directement — # aucune exécution JavaScript nécessaire, Cloudflare accepte le UA de base. # Filtre région : city_ids 2310 (Ville de Québec) + 1039 (Gatineau, ajouté # à l'expansion provinciale 2026-08) + garde-fou province/ville. # La fiche propriété (rendu serveur) fournit la galerie photo, via le cache # BD self.detail() (revisitée seulement quand le flux change). # ----------------------------------------------------------------------------- from __future__ import annotations import hashlib import re from bs4 import BeautifulSoup from ..schema import Listing, infer_city, normalize_unit_type, strip_accents from .base import BaseConnector BASE = "https://www.cpliving.com" SEARCH_PAGE = f"{BASE}/apartments-for-rent/quebec-city" LIFT_API = "https://api.theliftsystem.com/v2/search" # Valeurs observées sur la page/main.js — repli si l'extraction dynamique casse DEFAULT_CLIENT_ID = "21" DEFAULT_CITY_ID = "2310" # Ville de Québec dans la base Lift/Rentsync GATINEAU_CITY_ID = "1039" # Gatineau (expansion provinciale 2026-08) DEFAULT_AUTH_TOKEN = "sswpREkUtyeYjeoahA2i" # Garde-fou région : villes couvertes — agglomération Québec/Lévis + Gatineau # (clé sans accents, suffixe « QC » du flux retiré par _city_key) _QC_CITIES = { "quebec", "ville de quebec", "quebec city", "sainte-foy", "ste-foy", "sillery", "beauport", "charlesbourg", "cap-rouge", "levis", "l-ancienne-lorette", "saint-augustin-de-desmaures", "gatineau", "hull", "aylmer", } # jeton d'authentification dans main.js : `s="&client_id=21",e="&auth_token=…"` _TOKEN_RE = r'client_id={cid}",\w+="&auth_token=([A-Za-z0-9]+)"' _MAINJS_RE = re.compile(r'src="(/scripts/main\.js[^"]*)"') # galerie de la fiche propriété (img + backgrounds CSS) _IMG_RE = re.compile( r"https://assets\.rentsync\.com/[^\"'\\)\s]+\.(?:jpg|jpeg|png|webp)", re.I) _SKIP_IMG = re.compile(r"logo|icon|favicon|badge|/thumb", re.I) def _city_key(city: str) -> str: key = strip_accents((city or "").strip().lower()) return re.sub(r"\s+qc$", "", key) # le flux accole parfois la province class CenturionConnector(BaseConnector): source_id = "centurion" request_delay = 0.7 max_properties = 30 # garde-fou (1 propriété à Québec aujourd'hui) max_images = 20 # -- paramètres du flux (page + main.js, avec replis) ---------------------- def _feed_params(self) -> tuple[str, str, str]: """(client_id, city_id, auth_token) lus sur le site, replis constants.""" client_id, city_id, token = (DEFAULT_CLIENT_ID, DEFAULT_CITY_ID, DEFAULT_AUTH_TOKEN) try: page = self.get(SEARCH_PAGE).text soup = BeautifulSoup(page, "html.parser") data = soup.find("div", class_="search-data") if data: client_id = (data.get("data-client-id") or client_id).strip() city_id = (data.get("data-city-id") or city_id).strip() m = _MAINJS_RE.search(page) if m: js = self.get(BASE + m.group(1)).text mt = re.search(_TOKEN_RE.format(cid=re.escape(client_id)), js) if mt: token = mt.group(1) except Exception: pass # replis : les constantes observées return client_id, city_id, token def fetch(self) -> list[Listing]: client_id, city_id, token = self._feed_params() props = self.get(LIFT_API, params={ "client_id": client_id, "auth_token": token, "city_ids": f"{city_id},{GATINEAU_CITY_ID}", "show_all_properties": "true", "show_custom_fields": "true", "show_amenities": "true", "show_promotions": "true", "limit": "1000", }, headers={"Accept": "application/json", "Referer": BASE + "/"}).json() listings: list[Listing] = [] for p in props[: self.max_properties]: try: addr = p.get("address") or {} if (addr.get("province_code") or "").upper() != "QC": continue # REIT national : rester dans la province if _city_key(addr.get("city", "")) not in _QC_CITIES: continue # garde-fou région Québec/Lévis listings.append(self._listing(p)) except Exception: continue return listings # -- une annonce par propriété --------------------------------------------- def _listing(self, p: dict) -> Listing: pid = str(p.get("id")) addr = p.get("address") or {} url = p.get("permalink") or SEARCH_PAGE name = (p.get("name") or "").strip() # adresse complète : rue + ville + code postal (tous fournis au flux) ck = _city_key(addr.get("city", "")) if ck in ("quebec", "ville de quebec", "quebec city"): city = "Québec" else: # retirer le suffixe province du flux (« Gatineau QC ») city = re.sub(r"\s+QC$", "", (addr.get("city") or "").strip(), flags=re.I) street = (addr.get("address") or "").strip() postal = (addr.get("postal_code") or "").strip() full_addr = ", ".join(x for x in (street, city) if x) if postal: full_addr += f", QC {postal}" sector = (addr.get("neighbourhood") or "").strip() # coordonnées GPS structurées du flux geo = p.get("geocode") or {} try: lat = float(geo["latitude"]) if geo.get("latitude") else None lng = float(geo["longitude"]) if geo.get("longitude") else None except (TypeError, ValueError): lat = lng = None # sommaire des unités disponibles (rempli seulement s'il y a vacance) stats = ((p.get("statistics") or {}).get("suites") or {}) rates = stats.get("rates") or {} beds = stats.get("bedrooms") or {} baths = stats.get("bathrooms") or {} sqft = stats.get("square_feet") or {} price = float(rates["min"]) if rates.get("min") else None price_label = "" if price is not None: price_label = (f"À partir de {price:.0f} $" if rates.get("max") and rates["max"] != rates["min"] else f"{price:.0f} $ /mois") # type d'unité : seulement si la gamme est sans ambiguïté unit_type = "" if beds.get("min") is not None and beds.get("min") == beds.get("max"): n = int(beds["min"]) unit_type = "Studio" if n == 0 else normalize_unit_type( f"{n} chambres") # superficie : le flux publie parfois « 0.0 » (Gatineau) — ignorer area = None try: v = float(sqft.get("min") or 0) if 80 <= v <= 20000: area = v except (TypeError, ValueError): pass # disponibilité : libellé du flux (« No Vacancy », « X Vacancies »…) availability = (p.get("availability_status_label") or "").strip() avail_date = None mad = str(p.get("min_availability_date") or "").strip() if re.fullmatch(r"20\d{2}-\d{2}-\d{2}", mad[:10]): avail_date = mad[:10] # description : aperçu HTML du flux (rendu texte) details_src = p.get("details") or {} desc = BeautifulSoup(details_src.get("overview") or "", "html.parser").get_text(" ", strip=True) promo = p.get("promotion") or {} promo_txt = (promo.get("title") or promo.get("name") or "").strip() \ if isinstance(promo, dict) else "" if promo_txt: desc = f"Promotion : {promo_txt}. {desc}".strip() # commodités : liste du flux + champ personnalisé Rentsync (CSV) amenities: list[str] = [] for a in p.get("amenities") or []: t = (a.get("name") if isinstance(a, dict) else str(a) or "").strip() if t and t not in amenities: amenities.append(t) cf = p.get("custom_fields") or {} for t in (cf.get("amenities") or "").split(","): t = t.strip() if t and t not in amenities: amenities.append(t) # champs structurés du flux details: dict = {} contact = p.get("contact") or {} if contact.get("phone"): details["contact"] = {"phone": contact["phone"]} if contact.get("email"): details.setdefault("contact", {})["email"] = contact["email"] # pet_friendly=false ne distingue pas « interdit » de « non renseigné » pets = "oui" if p.get("pet_friendly") is True else None # galerie photo de la fiche propriété — via le cache BD : revisitée # seulement quand la ligne du flux change feed_key = hashlib.sha1("|".join(str(x) for x in ( p.get("availability_count"), p.get("availability_status"), rates.get("min"), rates.get("max"), mad, p.get("photo"), )).encode("utf-8")).hexdigest() d = self.detail(pid, feed_key, lambda: self._fetch_gallery(url)) images = list(d.get("images") or []) photo = (p.get("photo_path") or "").strip() if photo and photo not in images: images.insert(0, photo) return Listing( source=self.source_id, external_id=pid, url=url, title=name, address=full_addr, sector=sector, city=infer_city(sector, default=city or "Québec"), unit_type=unit_type, price=price, price_label=price_label, availability=availability, availability_date=avail_date, area_sqft=area, pets=pets, description=desc[:600] + ( f" Salles de bain : {baths['min']:g}+." if baths.get("min") else ""), amenities=amenities[:25], details=details, images=images[: self.max_images], lat=lat, lng=lng, ) def _fetch_gallery(self, url: str) -> dict: """Scrape la galerie photo (assets.rentsync.com) de la fiche propriété.""" out: dict = {"images": []} try: page = self.get(url).text except Exception: return out images: list[str] = [] for u in _IMG_RE.findall(page): if _SKIP_IMG.search(u): continue # variante pleine résolution de la galerie (…/gallery/full/…) u = re.sub(r"/gallery/\d{3,4}/", "/gallery/full/", u) if u not in images: images.append(u) out["images"] = images[: self.max_images] return out