# ----------------------------------------------------------------------------- # Rent-Ka — Rental listings aggregator (Canada, outside Québec) # Author: Simon-Pierre Boucher — contact@spboucher.ai # connectors/centurion.py : Centurion / CP Living (cpliving.com) # National REIT (Centurion Property Associates). The site runs on Rentsync: # the search page references a legacy proxy (404 today) but its JS # (scripts/main.js) actually calls the official JSON feed # `https://api.theliftsystem.com/v2/search` with an embedded auth_token. # client_id is read from the page, the token from main.js (constants as # fallback), then the feed is queried directly WITHOUT city_ids — that # returns the whole Centurion portfolio (~104 properties: Toronto/GTA, # Ottawa, Kitchener-Waterloo, Barrie, Huntsville, plus BC/AB/NS/MB…). # Rent-Ka keeps every province except QC. No JavaScript execution needed; # Cloudflare accepts the base UA. The server-rendered property page # provides the photo gallery through the self.detail() DB cache # (revisited only when the feed row changes). # ----------------------------------------------------------------------------- from __future__ import annotations import hashlib import os import re from bs4 import BeautifulSoup from ..schema import Listing, normalize_unit_type, strip_accents from .base import BaseConnector BASE = "https://www.cpliving.com" SEARCH_PAGE = f"{BASE}/apartments-for-rent/toronto" LIFT_API = "https://api.theliftsystem.com/v2/search" # Values observed on the page/main.js — fallbacks if dynamic extraction breaks DEFAULT_CLIENT_ID = "21" DEFAULT_AUTH_TOKEN = "sswpREkUtyeYjeoahA2i" # 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|on)$", "", key) # le flux accole parfois la province class CenturionConnector(BaseConnector): source_id = "centurion" request_delay = 0.7 max_properties = 200 # safety cap (~104 properties Canada-wide) max_images = 20 # -- feed parameters (page + main.js, with fallbacks) ---------------------- def _feed_params(self) -> tuple[str, str]: """(client_id, auth_token) read from the site, constants as fallback.""" client_id, token = DEFAULT_CLIENT_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() 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 # fallbacks: the observed constants return client_id, token def fetch(self) -> list[Listing]: client_id, token = self._feed_params() params = { "client_id": client_id, "auth_token": token, # no city_ids: the feed returns the whole Centurion portfolio "show_all_properties": "true", "show_custom_fields": "true", "show_amenities": "true", "show_promotions": "true", "limit": "1000", } props = self.get(LIFT_API, params=params, headers={"Accept": "application/json", "Referer": BASE + "/"}).json() listings: list[Listing] = [] count = 0 for p in props: try: addr = p.get("address") or {} prov = (addr.get("province_code") or "").upper() if not prov or prov == "QC": continue # Québec is Rent-Ka's territory if count >= self.max_properties: break count += 1 listings.append(self._listing(p, province=prov)) except Exception: continue return listings # -- one listing per property ---------------------------------------------- def _listing(self, p: dict, province: str = "ON") -> Listing: pid = str(p.get("id")) addr = p.get("address") or {} url = p.get("permalink") or SEARCH_PAGE name = (p.get("name") or "").strip() # full address: street + city + postal code (all provided by the feed); # strip the province suffix the feed sometimes appends («Barrie ON») city = re.sub(rf"\s+{province}$", "", (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", {province} {postal}" elif full_addr: full_addr += f", {province}" 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=city or "Toronto", province=province, 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