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/gestion_mj.py : connecteur Gestion Immobilière M.J (gestionmj.com)5#   Immeuble du 9155 rue Meilleur, Montréal (Ahuntsic / district Chabanel).6#   WordPress (thème Astra) rendu serveur avec CPT « a-louer » exposé en7#   REST (/wp-json/wp/v2/a-louer) : on énumère les annonces via l'API puis8#   on parse chaque page détail (h2.price, p.address, p.availability,9#   section Détails = commodités, Description, galerie).10#   ⚠️ Les blocs icônes (chambres / pi² / étage) et la date de disponibilité11#   sont des valeurs de gabarit IDENTIQUES sur toutes les pages (« 212#   chambre(s) », « 900 pi² », « 02/03/2025 ») → chambres/typologie dérivées13#   du TITRE (« Studio moderne », « Appartement 2/3 chambres »), icônes14#   reléguées en details. Granularité : une annonce par unité (post CPT).15#   2026-09-13 : SiteGround sert son anti-bot sgcaptcha (202 + challenge JS,16#   en-tête sg-captcha) aux IP datacenter — invisible du wrapper résilient17#   (2xx) ; on bascule alors la session sur un proxy résidentiel Oxylabs CA18#   (même pattern que boreal_abitibi/citiluxx/cromwell/deschenes_pepin/19#   gestion_habitation/gimcote/appartements_rimouski/lbm, hébergeur20#   identique).21# -----------------------------------------------------------------------------22from __future__ import annotations2324import json25import os26import re27import time28from urllib.parse import quote2930from bs4 import BeautifulSoup3132from ..schema import Listing, normalize_unit_type, parse_price33from .base import BaseConnector3435BASE = "https://gestionmj.com"36API_URL = (f"{BASE}/wp-json/wp/v2/a-louer?per_page=100"37           "&_fields=id,slug,link,title,modified")3839# page interstitielle sgcaptcha SiteGround (meta refresh vers /.well-known/…)40_SG_MARKER = "/.well-known/sgcaptcha/"414243def _sg_challenge(resp) -> bool:44    """True si la réponse est le challenge anti-bot SiteGround (202 + JS)."""45    if "challenge" in str(resp.headers.get("sg-captcha", "")).lower():46        return True47    return _SG_MARKER in (resp.text or "")[:600]4849BED_TITLE_RE = re.compile(r"(\d)\s*chambres?", re.I)50IMG_RE = re.compile(r"https://gestionmj\.com/wp-content/uploads/"51                    r'[^"\s\\)]+\.(?:jpe?g|png|webp)', re.I)525354class GestionMJConnector(BaseConnector):55    source_id = "gestion_mj"56    request_delay = 0.75758    def _enable_residential_proxy(self) -> bool:59        """Route toute la session via Oxylabs résidentiel CA (session collante)."""60        endpoint = os.environ.get("OXYLABS_PROXY")61        user = os.environ.get("OXYLABS_PROXY_USER")62        pwd = os.environ.get("OXYLABS_PROXY_PASS")63        if not (endpoint and user and pwd):64            return False65        puser = f"{user}-cc-CA-sessid-gestionmj{int(time.time())}-sesstime-10"66        proxy = f"http://{quote(puser, safe='')}:{quote(pwd, safe='')}@{endpoint}"67        self.session.proxies = {"http": proxy, "https": proxy}68        self.session.verify = False  # CA MITM du proxy Oxylabs69        return True7071    def _get_html(self, url: str) -> str:72        """GET avec contournement du challenge sgcaptcha (proxy résidentiel)."""73        resp = self.get(url)74        if not _sg_challenge(resp):75            return resp.text76        if not self.session.proxies and self._enable_residential_proxy():77            resp = self.get(url)78            if not _sg_challenge(resp):79                return resp.text80        raise RuntimeError(f"challenge sgcaptcha non contourné — {url}")8182    def fetch(self) -> list[Listing]:83        listings: list[Listing] = []84        # échec franc sur challenge/panne (fini le « 0 trouvé ok » silencieux)85        posts = json.loads(self._get_html(API_URL))86        if not isinstance(posts, list):87            raise RuntimeError("réponse wp-json inattendue (pas une liste)")8889        for post in posts:90            try:91                slug = post.get("slug", "")92                url = (post.get("link") or "").split("?")[0]93                if not slug or "/a-louer/" not in url:94                    continue95                title = BeautifulSoup(96                    (post.get("title") or {}).get("rendered", ""),97                    "html.parser").get_text(" ", strip=True) or slug9899                d = self.detail(slug, str(post.get("modified", "")),100                                lambda url=url: self._fetch_detail(url))101102                # typologie/chambres depuis le TITRE (icônes non fiables)103                bedrooms = None104                if re.search(r"\bstudio\b", title, re.I):105                    bedrooms = 0.0106                else:107                    m = BED_TITLE_RE.search(title)108                    if m:109                        bedrooms = float(m.group(1))110                unit_type = normalize_unit_type(111                    "Studio" if bedrooms == 0.0112                    else (f"{int(bedrooms)} chambres" if bedrooms else ""))113114                listings.append(Listing(115                    source=self.source_id,116                    external_id=slug,117                    url=url,118                    title=title,119                    address=d.get("address", ""),120                    city="Montréal",121                    sector="Ahuntsic-Cartierville",122                    unit_type=unit_type,123                    bedrooms=bedrooms,124                    price=d.get("price"),125                    availability=d.get("availability", ""),126                    furnished="Meublé" in (d.get("amenities") or []),127                    description=d.get("description", ""),128                    amenities=d.get("amenities") or [],129                    details=d.get("details") or {},130                    images=d.get("images") or [],131                ))132            except Exception:133                continue134        return listings135136    def _fetch_detail(self, url: str) -> dict:137        out: dict = {}138        # _get_html lève sur challenge : pas de {} vide dans detail_cache139        # (clé figée jamais invalidée)140        html = self._get_html(url)141        soup = BeautifulSoup(html, "html.parser")142143        el = soup.select_one("h2.price")144        if el:145            out["price"] = parse_price(el.get_text(" ", strip=True))146        el = soup.select_one("p.address")147        if el:148            out["address"] = re.sub(r"\s+", " ",149                                    el.get_text(" ", strip=True)).strip()150        el = soup.select_one("p.availability")151        if el:152            out["availability"] = re.sub(153                r"\s+", " ", el.get_text(" ", strip=True)).strip()154155        # section « Détails » : commodités (li)156        h = soup.find(string=re.compile(r"^\s*Détails\s*$"))157        amenities: list[str] = []158        if h:159            sec = h.find_parent(["div", "section"])160            for _ in range(3):161                if sec is None:162                    break163                lis = sec.select("li")164                if lis:165                    amenities = [li.get_text(" ", strip=True)166                                 for li in lis167                                 if 0 < len(li.get_text(strip=True)) <= 60]168                    break169                sec = sec.parent170        out["amenities"] = amenities[:15]171172        # description : paragraphes après le titre « Description »173        h = soup.find(string=re.compile(r"^\s*Description\s*$"))174        if h:175            par = h.find_parent(["div", "section"])176            ps = [p.get_text(" ", strip=True)177                  for p in (par.find_all_next("p", limit=3) if par else [])178                  if len(p.get_text(strip=True)) > 60]179            if ps:180                out["description"] = re.sub(r"\s+", " ",181                                            " ".join(ps[:2]))[:900]182183        # icônes gabarit (non fiables) → details, à titre indicatif184        details: dict = {}185        for sel, key in (("p.icon.square-footage", "gabarit_superficie"),186                         ("p.icon.floor", "gabarit_etage")):187            el = soup.select_one(sel)188            if el:189                details[key] = el.get_text(" ", strip=True)190        out["details"] = details191192        out["images"] = [193            u for u in dict.fromkeys(IMG_RE.findall(html))194            if not re.search(r"logo|icon|favicon|-\d{2,3}x\d{2,3}\.", u, re.I)195        ][:20]196        return out197