spb/lou-ka

1# -----------------------------------------------------------------------------2# Lou-Ka — Agrégateur de logements à louer (province de Québec)3# Auteur : Simon-Pierre Boucher — contact@spboucher.ai4# connectors/deschenes_pepin.py : connecteur Groupe Deschênes Pépin5#   (groupedeschenespepin.com — Montérégie). Promoteur-gestionnaire : projets6#   locatifs « Le Blüm » à Longueuil et Sainte-Julie (50 ans et plus).7#   WordPress/Elementor rendu serveur : la page /louer/ liste les projets,8#   chaque fiche /projet/<slug>/ donne les types d'unités (« 3 ½, 4 ½ et9#   5 ½ »), l'adresse du bureau d'information, la description et la galerie.10#   Granularité : une annonce par projet × typologie (pas d'unités ni de prix11#   publiés sur le site).12#   2026-09-13 : SiteGround sert son anti-bot sgcaptcha (202 + challenge JS,13#   en-tête sg-captcha) aux IP datacenter — invisible du wrapper résilient14#   (2xx) ; on bascule alors la session sur un proxy résidentiel Oxylabs CA15#   (même pattern que boreal_abitibi/citiluxx/cromwell, hébergeur identique).16# -----------------------------------------------------------------------------17from __future__ import annotations1819import os20import re21import time22from urllib.parse import quote2324from bs4 import BeautifulSoup2526from ..schema import Listing, normalize_unit_type27from .base import BaseConnector2829BASE = "https://groupedeschenespepin.com"30LIST_URL = f"{BASE}/louer/"3132PROJECT_LINK_RE = re.compile(r"https://groupedeschenespepin\.com/projet/([\w\-]+)/?")33UNIT_TOKEN_RE = re.compile(r"\b(\d)\s*(?:½|1/2)")34IMG_RE = re.compile(35    r"https://groupedeschenespepin\.com/wp-content/uploads/"36    r'[^"\s\\)]+\.(?:jpe?g|png|webp|avif)', re.I)37PHONE_RE = re.compile(r"\b(\d{3})[\s.\-](\d{3})[\s.\-](\d{4})\b")38EMAIL_RE = re.compile(r"[\w.\-]+@[\w.\-]+\.\w{2,}")39# adresse civique « 1825 rue du Caribou, Longueuil, Québec J4N 0C9 »40ADDR_RE = re.compile(41    r"\d{1,5}[,]?\s+(?:rue|avenue|av\.|boulevard|boul\.?|chemin|ch\.|route|"42    r"mont[ée]e|place)\s+[^,<>]{2,60},?\s*(?:Longueuil|Sainte-Julie|Ste-Julie)"43    r"[^<>]{0,40}?(?:[A-Z]\d[A-Z]\s?\d[A-Z]\d)?", re.I)4445# page interstitielle sgcaptcha SiteGround (meta refresh vers /.well-known/…)46_SG_MARKER = "/.well-known/sgcaptcha/"474849def _sg_challenge(resp) -> bool:50    """True si la réponse est le challenge anti-bot SiteGround (202 + JS)."""51    if "challenge" in str(resp.headers.get("sg-captcha", "")).lower():52        return True53    return _SG_MARKER in (resp.text or "")[:600]545556def _city_of(slug: str, text: str) -> str:57    key = f"{slug} {text}".lower()58    if "sainte-julie" in key or "ste-julie" in key:59        return "Sainte-Julie"60    return "Longueuil"616263class DeschenesPepinConnector(BaseConnector):64    source_id = "deschenes_pepin"65    request_delay = 0.866    max_projects = 12          # garde-fou de crawl6768    def _enable_residential_proxy(self) -> bool:69        """Route toute la session via Oxylabs résidentiel CA (session collante)."""70        endpoint = os.environ.get("OXYLABS_PROXY")71        user = os.environ.get("OXYLABS_PROXY_USER")72        pwd = os.environ.get("OXYLABS_PROXY_PASS")73        if not (endpoint and user and pwd):74            return False75        puser = f"{user}-cc-CA-sessid-deschenes{int(time.time())}-sesstime-10"76        proxy = f"http://{quote(puser, safe='')}:{quote(pwd, safe='')}@{endpoint}"77        self.session.proxies = {"http": proxy, "https": proxy}78        self.session.verify = False  # CA MITM du proxy Oxylabs (cf. _resilient)79        return True8081    def _get_html(self, url: str) -> str:82        """GET avec contournement du challenge sgcaptcha (proxy résidentiel)."""83        resp = self.get(url)84        if not _sg_challenge(resp):85            return resp.text86        if not self.session.proxies and self._enable_residential_proxy():87            resp = self.get(url)88            if not _sg_challenge(resp):89                return resp.text90        raise RuntimeError(f"challenge sgcaptcha non contourné — {url}")9192    def fetch(self) -> list[Listing]:93        listings: list[Listing] = []94        # la page de liste échoue franchement (fini le « 0 trouvé ok »95        # silencieux du challenge sgcaptcha — ingest journalise ok=0)96        html = self._get_html(LIST_URL)9798        # Slugs des fiches projet référencées depuis la section « Louer »99        slugs = list(dict.fromkeys(PROJECT_LINK_RE.findall(html)))100        for slug in slugs[: self.max_projects]:101            try:102                listings.extend(self._fetch_project(slug))103            except Exception:104                continue105        return listings106107    def _fetch_project(self, slug: str) -> list[Listing]:108        url = f"{BASE}/projet/{slug}/"109        html = self._get_html(url)110        soup = BeautifulSoup(html, "html.parser")111112        h1 = soup.find("h1")113        name = h1.get_text(" ", strip=True) if h1 else slug.replace("-", " ").title()114        text = re.sub(r"\s+", " ", soup.get_text(" ", strip=True))115116        # à vendre/maisons de ville : on ne garde que les projets locatifs117        # (la page /louer/ ne liste que ceux-là, ceinture et bretelles)118        if re.search(r"\bà vendre\b", name, re.I):119            return []120121        city = _city_of(slug, name)122123        # « Types d'unités 3 ½, 4 ½ et 5 ½ » (bloc d'entête de la fiche)124        m = re.search(r"Types?\s+d'unit[ée]s?\s*([\d\s½,/et]+)", text, re.I)125        unit_zone = m.group(1) if m else text[:2000]126        unit_types = list(dict.fromkeys(127            normalize_unit_type(f"{n}½") for n in UNIT_TOKEN_RE.findall(unit_zone)))128        if not unit_types:129            unit_types = [""]130131        # adresse du bureau d'information (l'immeuble lui-même)132        address = ""133        m = ADDR_RE.search(text)134        if m:135            address = m.group(0).strip().rstrip(",")136137        # description : paragraphe de présentation du projet138        description = ""139        for p in soup.find_all("p"):140            pt = p.get_text(" ", strip=True)141            if len(pt) > 120 and not pt.lower().startswith(("pour offrir", "je reconnais")):142                description = pt[:900]143                break144145        contact: dict = {}146        m = PHONE_RE.search(text)147        if m:148            contact["phone"] = f"{m.group(1)}-{m.group(2)}-{m.group(3)}"149        m = EMAIL_RE.search(text)150        if m:151            contact["email"] = m.group(0).lower()152153        images = [u for u in dict.fromkeys(IMG_RE.findall(html))154                  if not re.search(r"logo|icon|favicon|cropped-|plan|"155                                   r"-\d{2,3}x\d{2,3}\.", u, re.I)][:20]156157        amenities = []158        if re.search(r"50 ans et plus", text, re.I):159            amenities.append("Pour les 50 ans et plus")160        for kw, label in [161            (r"salle d'entra[îi]nement", "Salle d'entraînement"),162            (r"terrasse sur le toit", "Terrasse sur le toit"),163            (r"piscine", "Piscine"),164            (r"stationnement int[ée]rieur", "Stationnement intérieur"),165            (r"ascenseur", "Ascenseur"),166            (r"tout inclus", "Formule tout inclus"),167        ]:168            if re.search(kw, text, re.I):169                amenities.append(label)170171        out: list[Listing] = []172        for ut in unit_types:173            ext = f"{slug}-{ut.replace('½', '.5')}" if ut else slug174            out.append(Listing(175                source=self.source_id,176                external_id=ext,177                url=url,178                title=f"{name} — {ut}" if ut else name,179                address=address,180                city=city,181                unit_type=ut,182                description=description,183                amenities=list(dict.fromkeys(amenities)),184                details={"contact": dict(contact)} if contact else {},185                images=images,186            ))187        return out188