# ----------------------------------------------------------------------------- # Lou-Ka — Agrégateur de logements à louer (province de Québec) # Auteur : Simon-Pierre Boucher — contact@spboucher.ai # connectors/nexliving.py : connecteur NexLiving Communities # (residential.nexliving.ca) — REIT pancanadien ; côté Québec : Gatineau # (Pointe-Gatineau, Place du Golf, Saint-James, Prado, Nelson-Morin, # Progrès, Futaie, Lanctot, JGR), Thurso (Fraser) et Val-d'Or # (Roland-Audet). Site Yardi RentCafe derrière Cloudflare (403 direct) : # tout passe par Scrapfly (ASP), SANS rendu JS — RentCafe embarque un bloc # SEO rendu serveur (#list-view-cards-hidden, li.property-box-hidden) avec # nom, adresse complète, lits/sdb/pi², prix « à partir de », téléphone et # vignette. Une annonce par propriété (uid = slug RentCafe stable). # Pages parcourues : /qc/gatineau/apartments, /qc/thurso/apartments, # /qc/val-d'or/apartments — SEULES les fiches /apartments/qc/ passent # (les parcs ON et NB du REIT sont exclus), avec la ville réelle par # annonce. La fiche propriété (galerie, description) passe par # self.detail() (cache BD) avec budget Scrapfly par synchronisation. # ----------------------------------------------------------------------------- from __future__ import annotations import hashlib import re from bs4 import BeautifulSoup from ..schema import Listing, parse_price from .base import BaseConnector BASE = "https://residential.nexliving.ca" # pages liste par ville québécoise (l'apostrophe de val-d'or est bien dans # l'URL du site) SEARCH_PATHS = ["/qc/gatineau/apartments", "/qc/thurso/apartments", "/qc/val-d'or/apartments"] # segment ville de /apartments/qc// -> ville affichée _QC_CITIES = { "gatineau": "Gatineau", "thurso": "Thurso", "val-d-or": "Val-d'Or", } _PROP_RE = re.compile(r"/apartments/qc/([a-z0-9\-.']+)/([a-z0-9\-]+)/") _BED_TYPES = {"0": "Studio", "1": "3½", "2": "4½", "3": "5½", "4": "6½+"} _PRICE_RE = re.compile(r"\$[\d,]+(?:\.\d{2})?(?:(?:\s*(?:-|to))+\s*" r"\$[\d,]+(?:\.\d{2})?)?") _IMG_RE = re.compile( r"https://resource\.rentcafe\.com/image/upload/[^\"'\s\\]+?" r"\.(?:jpg|jpeg|png|webp)", re.I) _SKIP_IMG = re.compile(r"logo|icon|favicon|placeholder|lockup", re.I) class _BudgetReached(Exception): """Plafond d'appels Scrapfly atteint pour cette synchronisation.""" class NexLivingConnector(BaseConnector): source_id = "nexliving" request_delay = 1.0 max_images = 20 max_details = 12 # budget de fiches propriété Scrapfly (hors cache) # -- page derrière Cloudflare (bloc SEO rendu serveur, pas de JS) ---------- def _page(self, url: str) -> str: html = self.get_scrapfly(url, render_js=False) if html and "property-box-hidden" in html: return html # seconde chance avec rendu JS complet (bloc SEO absent/incomplet) return self.get_scrapfly(url, rendering_wait=5000) def fetch(self) -> list[Listing]: self._fetched = 0 listings: list[Listing] = [] seen: set[str] = set() pages_ok = 0 for path in SEARCH_PATHS: try: html = self._page(BASE + path) soup = BeautifulSoup(html, "html.parser") cards = soup.select("#list-view-cards-hidden " "li.property-box-hidden") if not cards: continue pages_ok += 1 except Exception: continue for card in cards: try: lst = self._card_listing(card) except Exception: continue if lst and lst.external_id not in seen: seen.add(lst.external_id) listings.append(lst) if pages_ok == 0: # échec transitoire -> journalisé comme ÉCHEC par ingest (ok=0), # pas comme un passage « ok » à 0 annonce raise RuntimeError( "aucune page liste NexLiving accessible via Scrapfly") return listings # -- une annonce par propriété ----------------------------------------------- def _card_listing(self, card) -> Listing | None: a = card.select_one(".property-name a[href]") if not a: return None url = (a.get("href") or "").split("?")[0].replace("http://", "https://") m = _PROP_RE.search(url) if not m: return None # fiche hors Québec (ON/NB) : exclue city = _QC_CITIES.get(m.group(1).replace("'", "-")) slug = m.group(2) addr_el = card.select_one(".card-prop-address") address = addr_el.get_text(" ", strip=True) if addr_el else "" if not city: # ville QC inattendue : la lire dans l'adresse (« …, Ville, QC … ») am = re.search(r",\s*([^,]+),\s*QC\b", address) if not am: return None # jamais de ville inventée city = am.group(1).strip() name = re.sub(r"\s*opens in a new tab\s*", "", a.get_text(" ", strip=True)).strip() name = name or slug.replace("-", " ").title() # lits/sdb/pi² (le type d'unité seulement si la gamme est sans # ambiguïté : « 3.0Beds - 3.0Beds ») beds = baths = sqft = "" for li in card.select(".card-bed-bath-rent li"): it = li.get_text(" ", strip=True) if "Bed" in it: beds = it elif "Bath" in it: baths = it elif "Sq" in it: sqft = re.sub(r"\s*to\s*-\s*", " - ", it) unit_type = "" bm = re.match(r"^(?:(\d)\.0Beds\s*-\s*(\d)\.0Beds|Studio\s*-\s*Studio)$", beds or "") if bm: if bm.group(1) is None: unit_type = "Studio" elif bm.group(1) == bm.group(2): unit_type = _BED_TYPES.get(bm.group(1), "") # prix « $999.00 to - $1,899.00 » ou « Call for Details » price = None price_label = "" pm = _PRICE_RE.search(card.get_text(" ", strip=True)) if pm: price_label = re.sub(r"\s*to\s*-\s*", " - ", pm.group(0)) price = parse_price(price_label) if "-" in price_label: price_label = "À partir de " + price_label # téléphone du bureau de location (lien tel: structuré) phone = "" tel = card.select_one("a[href^='tel:']") if tel: tm = re.search(r"\(?([2-9]\d{2})\)?[\s.\-]?(\d{3})[\s.\-]?(\d{4})", tel.get("href", "")) if tm: phone = f"{tm.group(1)}-{tm.group(2)}-{tm.group(3)}" images: list[str] = [] img = card.select_one("img[src*='rentcafe']") if img and img.get("src"): images.append(img["src"]) # fiche propriété (galerie + description) via cache BD : Scrapfly # n'est rappelé que si la carte liste a changé key = hashlib.sha1(f"{name}|{address}|{beds}|{baths}|{sqft}|" f"{price_label}".encode("utf-8")).hexdigest() try: payload = self.detail(slug, key, lambda: self._fetch_detail(url)) except Exception: payload = {} for im in (payload.get("images") or []): if im not in images: images.append(im) prop_type = "" pt = card.select_one(".unit-types") if pt: prop_type = pt.get_text(" ", strip=True) desc_parts = ([payload["description"]] if payload.get("description") else []) desc_parts += [b for b in (prop_type, beds, baths, sqft) if b] details: dict = {} if phone: details["contact"] = {"phone": phone} return Listing( source=self.source_id, external_id=slug, url=url, title=name, address=address, sector="", city=city, unit_type=unit_type, price=price, price_label=price_label, description=" — ".join(desc_parts)[:800], details=details, images=images[: self.max_images], ) # -- fiche propriété (Scrapfly sans JS, budget par sync) --------------------- def _fetch_detail(self, url: str) -> dict: if self._fetched >= self.max_details: raise _BudgetReached() self._fetched += 1 payload: dict = {"description": "", "images": []} html = self.get_scrapfly(url, render_js=False) if not html: return payload soup = BeautifulSoup(html, "html.parser") for u in dict.fromkeys(_IMG_RE.findall(html)): if not _SKIP_IMG.search(u) and u not in payload["images"]: payload["images"].append(u) payload["images"] = payload["images"][: self.max_images] paras = [p.get_text(" ", strip=True) for p in soup.find_all("p")] paras = [p for p in paras if len(p) > 80] if paras: payload["description"] = " ".join(paras[:2])[:600] return payload