# ----------------------------------------------------------------------------- # Rent-Ka — Agrégateur de logements à louer (Québec + expansion Ontario) # Author: Simon-Pierre Boucher — contact@spboucher.ai # connectors/starlight.py : connecteur Starlight Investments (starlightinvest.com) # Plus grand propriétaire résidentiel privé du Canada (~60 000 unités). Le # site corpo (Craft CMS) n'a pas de page /properties : le portefeuille est # exposé par le « Property Finder » de /real-estate/canadian-residential, # alimenté par un index Algolia PUBLIC (appId PX9UWFRLZ3, clé search-only # embarquée dans la page, index `prod_properties`, ~549 immeubles dont ~391 # « live » en Ontario). Chaque enregistrement donne : nom d'immeuble (le plus # souvent l'adresse civique), ville, province et l'URL d'un MICROSITE par # immeuble. Aucun prix/plan d'étage dans l'index. # Les microsites (317 domaines ON distincts) partagent majoritairement un # gabarit Duda + PropertyVista, MAIS les plans d'étage y sont chargés par un # GraphQL (DudaWebsiteService_getFloorplanList) dont l'endpoint vit dans une # collection Duda `vista_api_url` inaccessible hors runtime (400/500 même en # navigateur headless — vérifié via Scrapfly js_scenario) : pas de pattern # prix exploitable. On émet donc UNE annonce PAR PROPRIÉTÉ (granularité # immeuble, sans prix — jamais de valeur inventée) et on enrichit chaque # fiche avec le HTML serveur du microsite (meta description, téléphone, # galerie irp.cdn-website.com filtrée par l'alias du site) via le cache BD # self.detail() — budget max_details par sync, le reste vient du cache. # Les immeubles dont le « microsite » est en réalité une page metcap.com ou # cogir.net sont SAUTÉS : les connecteurs dédiés `metcap` et `cogir` (tous # deux gatés Ontario) couvrent déjà ces gestionnaires — zéro doublon. # Idem sterlingkaramar.com (65 propriétés) : couvert par rsgw_sterlingkaramar. # ----------------------------------------------------------------------------- from __future__ import annotations import hashlib import json import os import re from urllib.parse import urlparse from ..schema import Listing from .base import BaseConnector # index Algolia public du Property Finder (clé search-only embarquée dans # https://www.starlightinvest.com/real-estate/canadian-residential) ALGOLIA_APP = "PX9UWFRLZ3" ALGOLIA_KEY = "5d9417503fde20d753b7a25a301f8991" ALGOLIA_URL = (f"https://{ALGOLIA_APP.lower()}-dsn.algolia.net" "/1/indexes/prod_properties/query") FINDER_PAGE = ("https://www.starlightinvest.com/real-estate/" "canadian-residential#property-finder") # Gate expansion Ontario : le connecteur reste hors registre tant que la _ONTARIO = True # Rent-Ka: always on (ROC scope) # gestionnaires déjà couverts par un connecteur Rent-Ka dédié (gaté Ontario) — # leurs immeubles Starlight sont sautés ici pour éviter les doublons _SKIP_DOMAINS = ("metcap.com", "cogir.net", "sterlingkaramar.com") # fragment d'adresse civique dans le titre (« 1637 - 1645 Bathurst Street », # « Yonge St Towns: 389 Yonge Street », « St Regents Place - 2375 The # Collegeway ») — on n'utilise le titre comme adresse QUE s'il y ressemble _ADDR_RE = re.compile( r"\d[\d\s&/-]*\s+[A-Za-z'’.\- ]+\b(?:Street|St\.?|Avenue|Ave\.?|Road|" r"Rd\.?|Drive|Dr\.?|Boulevard|Blvd\.?|Court|Crt\.?|Crescent|Cres\.?|" r"Lane|Ln\.?|Way|Place|Pl\.?|Terrace|Circle|Cir\.?|Parkway|Pkwy\.?|" r"Gate|Square|Sq\.?|Trail|Line|Highway|Hwy\.?|Gardens|Grove|Path|" r"Promenade|Quay|Row|Walk|Heights|Collegeway|Queensway|Kingsway|" r"Esplanade|Appleby|Mall)\b[.A-Za-z'’\- ]*", re.I) # visuels du microsite Duda à ÉCARTER (logos gestionnaire, drapeaux du # sélecteur de langue, pictos, bandeau-titre « just address » du gabarit) _IMG_SKIP_RE = re.compile( r"logo|flag|icon|favicon|\+100px|badge|certif|just\+address|\+font\+", re.I) _META_DESC_RE = re.compile( r' None: super().__init__() self._detail_fetches = 0 # -- portefeuille : index Algolia public du Property Finder ----------------- def _properties(self) -> list[dict]: headers = { "X-Algolia-Application-Id": ALGOLIA_APP, "X-Algolia-API-Key": ALGOLIA_KEY, "Content-Type": "application/json", } hits: list[dict] = [] page = 0 while True: resp = self.post(ALGOLIA_URL, headers=headers, data=json.dumps({ "query": "", "page": page, "hitsPerPage": 1000, "attributesToRetrieve": ["title", "website", "address", "status", "section"], "attributesToHighlight": [], })) data = resp.json() hits.extend(data.get("hits") or []) page += 1 if page >= int(data.get("nbPages") or 1) or page > 5: break if not hits: raise RuntimeError("Starlight : index Algolia prod_properties " "vide (clé/app modifiée ?)") return hits def fetch(self) -> list[Listing]: listings: list[Listing] = [] count = 0 seen: set[tuple[str, str]] = set() for p in self._properties(): if (p.get("status") or "") != "live": continue addr = p.get("address") or {} if ((addr.get("state") or "").strip().upper()) != "ON": continue # portefeuille national : Ontario seulement ici website = (p.get("website") or "").strip() if not website: continue host = (urlparse(website).hostname or "").lower() if any(host == d or host.endswith("." + d) for d in _SKIP_DOMAINS): continue # gestionnaire déjà couvert (metcap/cogir) # l'index contient des variantes du même immeuble (« 1637 - 1645 » # vs « 1637-1645 Bathurst ») : dédoublonnage titre+microsite key = (host, re.sub(r"\s*-\s*", "-", re.sub(r"\s+", " ", (p.get("title") or "").strip().lower()))) if key in seen: continue seen.add(key) if count >= self.max_properties: break count += 1 try: lst = self._listing(p, addr, website) if lst is not None: listings.append(lst) except Exception: continue return listings # -- une annonce par propriété (granularité immeuble, sans prix) ------------ def _listing(self, p: dict, addr: dict, website: str) -> Listing | None: title = re.sub(r"\s+", " ", (p.get("title") or "")).strip() city = (addr.get("city") or "").strip() if not title or not city: return None # adresse civique : champ street1 quand il existe (rare), sinon le # titre s'il ressemble à une adresse — sinon rien (pas d'invention) street = (addr.get("street1") or "").strip() if not street: m = _ADDR_RE.search(title) if m: street = m.group(0).strip() full_addr = "" if street: full_addr = f"{street}, {city}, ON" postal = (addr.get("zip") or "").strip() if postal: full_addr += f" {postal}" # coordonnées GPS de l'index (présentes sur ~5 % des fiches ; # finalize() valide la bbox Ontario) try: lat = float(addr.get("lat")) or None lng = float(addr.get("lng")) or None except (TypeError, ValueError): lat = lng = None # fiche microsite (description, téléphone, galerie) — via le cache BD : # revisitée seulement si la ligne de l'index change ext_id = str(p.get("objectID") or "").strip() feed_key = hashlib.sha1("|".join(str(x) for x in ( title, website, city, street, addr.get("zip"), lat, lng, )).encode("utf-8")).hexdigest() d: dict = {} try: d = self.detail(ext_id, feed_key, lambda: self._fetch_detail(website)) except Exception: pass # budget détail épuisé ou microsite injoignable details: dict = {"property_manager_site": website} phone = (d.get("phone") or "").strip() if phone: details["contact"] = {"phone": phone} return Listing( source=self.source_id, external_id=ext_id, url=website, title=title, address=full_addr, city=city, province="ON", description=d.get("description") or "", images=list(d.get("images") or [])[: self.max_images], details=details, lat=lat, lng=lng, ) # -- microsite : meta description + téléphone + galerie (HTML serveur) ------ def _fetch_detail(self, url: str) -> dict: """Extrait les champs sûrs du HTML serveur du microsite Duda. Les prix/plans d'étage sont volontairement ignorés : ils sont chargés côté client par un GraphQL PropertyVista inaccessible (voir en-tête). """ if self._detail_fetches >= self.max_details: raise RuntimeError("budget de microsites atteint") # non mis en cache self._detail_fetches += 1 out: dict = {"description": "", "phone": "", "images": []} try: html = self.get(url).text except Exception: return out # microsite mort/parqué : cache vide, on n'y # retournera que si la ligne de l'index change m = _META_DESC_RE.search(html) if m: import html as _h out["description"] = _h.unescape(m.group(1)).strip()[:900] m = _TEL_RE.search(html) if m: out["phone"] = re.sub(r"\s+", " ", m.group(1)).strip() # galerie : uniquement les visuels du site lui-même (chemin contenant # l'alias Duda du microsite), hors logos/drapeaux/pictos alias_m = _SITE_ALIAS_RE.search(html) alias = alias_m.group(1) if alias_m else None images: list[str] = [] for src in _DUDA_IMG_RE.findall(html): if alias and f"/{alias}/" not in src: continue # visuel partagé du gabarit (logo gestionnaire…) if _IMG_SKIP_RE.search(src): continue if src not in images: images.append(src) out["images"] = images[: self.max_images] return out