# ----------------------------------------------------------------------------- # Lou-Ka — Agrégateur de logements à louer (province de Québec) # Auteur : Simon-Pierre Boucher — contact@spboucher.ai # connectors/citiluxx.py : connecteur Citiluxx (citiluxx.com — Groupe Cleary) # Condos locatifs tout inclus au Faubourg Cousineau, 8121 boulevard # Cousineau, Saint-Hubert (Longueuil). WordPress (Astra/Elementor) rendu # serveur : la page /plans-appartements-a-louer/ publie le prix plancher # par typologie (« 3 1/2 à louer : à partir de 1 932$/mois »…) et embarque # le widget Smart Condo Plans (API JSON publique smartcondoplans.silo.immo, # projet « citiluxx ») qui fournit le plan des ~199 unités (chambres, # salles de bain, superficies) — mais SANS prix ni disponibilité par unité. # Granularité : une annonce par typologie (prix « à partir de »), enrichie # des superficies min-max et du nombre d'unités via l'API Silo. # 2026-09-13 : SiteGround sert son anti-bot sgcaptcha (202 + challenge JS, # en-tête sg-captcha) aux IP datacenter — invisible du wrapper résilient # (2xx) ; on bascule alors la session sur un proxy résidentiel Oxylabs CA # (même pattern que boreal_abitibi, hébergeur SiteGround identique). # ----------------------------------------------------------------------------- from __future__ import annotations import os import re import time from urllib.parse import quote from ..schema import Listing, normalize_unit_type, parse_price from .base import BaseConnector BASE = "https://citiluxx.com" PLANS_URL = f"{BASE}/plans-appartements-a-louer/" SILO_API = ("https://smartcondoplans.silo.immo/v2/building/get-project-data" "?project=citiluxx&language=fr") ADDRESS = "8121, boulevard Cousineau, Saint-Hubert" CITY = "Longueuil" SECTOR = "Saint-Hubert" # « 3 1/2 à louer : à partir de 1 932$/mois » PRICE_RE = re.compile( r"(\d)\s*(?:½|1/2)\s*(?:\([^)]*\)\s*)?(?::|à louer\s*:)\s*" r"[àa] partir de\s*([\d\s,]{3,9})\s*\$\s*/\s*mois", re.I) IMG_RE = re.compile( r"https://citiluxx\.com/wp-content/uploads/" r'[^"\s\\)]+\.(?:jpe?g|png|webp)', re.I) # nb de chambres (API Silo, champ `room`) -> typologie québécoise UT_OF_ROOMS = {1: "3½", 2: "4½", 3: "5½"} # page interstitielle sgcaptcha SiteGround (meta refresh vers /.well-known/…) _SG_MARKER = "/.well-known/sgcaptcha/" def _sg_challenge(resp) -> bool: """True si la réponse est le challenge anti-bot SiteGround (202 + JS).""" if "challenge" in str(resp.headers.get("sg-captcha", "")).lower(): return True return _SG_MARKER in (resp.text or "")[:600] AMENITIES = [ "Formule tout inclus (chauffage, climatisation, électricité, eau chaude)", "5 électroménagers haut de gamme", "Concierge", "Hauts plafonds de 9 pieds", "Comptoirs en quartz", "Grand balcon-terrasse", "Casiers à colis intelligents", "Espace lave-auto", "Stationnement intérieur ($) avec espace vélo", "Animaux acceptés (max 15 lb, conditions)", ] class CitiluxxConnector(BaseConnector): source_id = "citiluxx" request_delay = 0.8 def _enable_residential_proxy(self) -> bool: """Route toute la session via Oxylabs résidentiel CA (session collante).""" endpoint = os.environ.get("OXYLABS_PROXY") user = os.environ.get("OXYLABS_PROXY_USER") pwd = os.environ.get("OXYLABS_PROXY_PASS") if not (endpoint and user and pwd): return False puser = f"{user}-cc-CA-sessid-citiluxx{int(time.time())}-sesstime-10" proxy = f"http://{quote(puser, safe='')}:{quote(pwd, safe='')}@{endpoint}" self.session.proxies = {"http": proxy, "https": proxy} self.session.verify = False # CA MITM du proxy Oxylabs (cf. _resilient) return True def _get_html(self, url: str) -> str: """GET avec contournement du challenge sgcaptcha (proxy résidentiel).""" resp = self.get(url) if not _sg_challenge(resp): return resp.text if not self.session.proxies and self._enable_residential_proxy(): resp = self.get(url) if not _sg_challenge(resp): return resp.text raise RuntimeError(f"challenge sgcaptcha non contourné — {url}") def fetch(self) -> list[Listing]: listings: list[Listing] = [] html = self._get_html(PLANS_URL) text = re.sub(r"\s+", " ", re.sub(r"<[^>]+>", " ", html)) images = [u for u in dict.fromkeys(IMG_RE.findall(html)) if not re.search(r"logo|icon|favicon|cropped-|Plans?\.webp|" r"-\d{2,3}x\d{2,3}\.", u, re.I)][:15] # enrichissement : plan des unités via l'API Silo (superficies, # nombre d'unités par typologie) — best effort, jamais bloquant silo: dict[str, dict] = {} try: data = self.get(SILO_API, headers={"Accept": "application/json"}).json() for u in data.get("units") or []: if u.get("not_unit") or u.get("hide"): continue ut = UT_OF_ROOMS.get(u.get("room")) if not ut: continue s = silo.setdefault(ut, {"n": 0, "areas": [], "baths": set()}) s["n"] += 1 if u.get("area"): s["areas"].append(float(u["area"])) if u.get("bathroom"): s["baths"].add(u["bathroom"]) except Exception: silo = {} seen: set[str] = set() for n, price_txt in PRICE_RE.findall(text): ut = normalize_unit_type(f"{n}½") if ut in seen: continue seen.add(ut) price_label = f"à partir de {price_txt.strip()}$/mois" desc = ("Condos locatifs neufs tout inclus Citiluxx au Faubourg " "Cousineau (Saint-Hubert, Longueuil) : espaces de vie " "ouverts, abondante luminosité, finitions haut de gamme. " "Phases 1 et 2 en occupation immédiate.") details: dict = {} area_min = None s = silo.get(ut) if s and s["areas"]: area_min = min(s["areas"]) desc += (f" Superficies de {int(min(s['areas']))} à " f"{int(max(s['areas']))} pi² selon l'unité " f"({s['n']} unités de ce type au plan).") details["building_units"] = s["n"] listings.append(Listing( source=self.source_id, external_id=f"citiluxx-{n}.5", url=PLANS_URL, title=f"Citiluxx — {ut} au Faubourg Cousineau", address=ADDRESS, sector=SECTOR, city=CITY, unit_type=ut, price=parse_price(price_label), price_label=price_label, availability="Occupation immédiate", area_sqft=area_min, description=desc, amenities=list(AMENITIES), details=details, images=images, )) return listings