spb/lou-ka Public

1# -----------------------------------------------------------------------------2# Lou-Ka — Agrégateur de logements à louer (province de Québec)3# Auteur : Simon-Pierre Boucher — contact@spboucher.ai4# connectors/sdg.py : connecteur SDG Immobilier (sdgimmobilier.ca)5#   Site WordPress (thème Houzez) : /appartements-a-louer/ liste des6#   immeubles (fiches « immeuble »). Une annonce par immeuble (types7#   d'unités affichés, pas de prix ni de disponibilités publiés).8#   Galerie via data-images ; fiche : description, caractéristiques,9#   bloc adresse structuré (adresse, arrondissement, code postal, ville),10#   téléphone (lien tel:). Fiches derrière self.detail(...) (cache BD).11# -----------------------------------------------------------------------------12from __future__ import annotations1314import hashlib15import json16import re1718from bs4 import BeautifulSoup1920from ..schema import Listing, infer_city, normalize_unit_type21from .base import BaseConnector2223BASE = "https://www.sdgimmobilier.ca"24LIST_URL = f"{BASE}/appartements-a-louer/"2526IMG_RE = re.compile(27    r"https://www\.sdgimmobilier\.ca/wp-content/uploads/[^\"'\\\s\)]+"28    r"\.(?:jpg|jpeg|png|webp)", re.I)29IMG_NOISE_RE = re.compile(r"logo|favicon|icon|-\d+x\d+\.", re.I)30EXCLUDE_RE = re.compile(r"stationnement|commercial|rangement|entrepos|bureau",31                        re.I)323334class _BudgetReached(Exception):35    """Plafond de requêtes « fiche » atteint pour cette synchronisation."""363738class SDGConnector(BaseConnector):39    source_id = "sdg"40    request_delay = 0.641    max_details = 40             # plafond de vraies requêtes fiche par sync4243    def fetch(self) -> list[Listing]:44        html = self.get(LIST_URL).text45        soup = BeautifulSoup(html, "html.parser")4647        listings: dict[str, Listing] = {}48        cards: dict[str, str] = {}   # ext_id -> texte de carte (clé de cache)49        for card in soup.select("div.item-listing-wrap[data-hz-id]"):50            try:51                lst = self._parse_card(card)52            except Exception:53                continue54            if lst and lst.external_id not in listings:55                listings[lst.external_id] = lst56                cards[lst.external_id] = card.get_text(" ", strip=True)5758        # Fiches immeuble (cache BD) : description, types, commodités, adresse59        self._fetches = 060        for lst in listings.values():61            key = hashlib.sha1(62                f"{lst.title}|{cards.get(lst.external_id, '')}"63                .encode("utf-8")).hexdigest()64            try:65                payload = self.detail(66                    lst.external_id, key,67                    lambda u=lst.url: self._fetch_detail(u))68            except Exception:    # _BudgetReached inclus : rien de caché69                continue70            self._apply_detail(lst, payload)7172        return list(listings.values())7374    def _parse_card(self, card) -> Listing | None:75        ext_id = card.get("data-hz-id", "").strip()76        title_a = card.select_one(".item-title a")77        if not ext_id or not title_a:78            return None79        url = title_a.get("href", "")80        if "/immeuble/" not in url:81            return None82        title = title_a.get_text(" ", strip=True)83        if EXCLUDE_RE.search(f"{title} {url}"):84            return None8586        addr_el = card.select_one(".item-address span") or \87            card.select_one(".item-address")88        sector = addr_el.get_text(" ", strip=True) if addr_el else ""8990        # Galerie complète fournie dans l'attribut data-images (JSON)91        images: list[str] = []92        raw = card.get("data-images", "")93        if raw:94            try:95                images = [d.get("image", "") for d in json.loads(raw)96                          if d.get("image")]97            except (ValueError, TypeError):98                images = []99        if not images:100            img_el = card.select_one(".listing-thumb img")101            if img_el and (img_el.get("src") or "").startswith("http"):102                images = [img_el["src"]]103104        return Listing(105            source=self.source_id,106            external_id=ext_id,107            url=url,108            title=title,109            address=title if re.match(r"^\d", title) else "",110            sector=sector,111            city=infer_city(sector),112            images=list(dict.fromkeys(images))[:25],113        )114115    def _fetch_detail(self, url: str) -> dict:116        """Télécharge une fiche /immeuble/<slug>/ (appelé seulement hors cache)."""117        if self._fetches >= self.max_details:118            raise _BudgetReached()119        self._fetches += 1120        html = self.get(url).text121        soup = BeautifulSoup(html, "html.parser")122        payload: dict = {}123124        # Description125        desc_el = soup.select_one(".property-description-content") or \126            soup.select_one("#property-description-wrap")127        if desc_el:128            payload["description"] = desc_el.get_text(" ", strip=True)[:600]129130        # Caractéristiques : « Type d'unités: 3 ½ » + commodités131        payload["features"] = [a.get_text(" ", strip=True)132                               for a in soup.select("#property-features-wrap li a")]133134        # Bloc « Adresse » Houzez (structuré : adresse, arrondissement,135        # code postal, ville)136        addr: dict = {}137        for row in soup.select("#property-address-wrap .list-lined-item"):138            label = row.find("strong")139            value = row.find("span")140            if not label or not value:141                continue142            key = label.get_text(strip=True).lower().rstrip(" :")143            val = value.get_text(" ", strip=True)144            if val:145                addr[key] = val146        payload["address_block"] = addr147148        # Téléphone du gestionnaire (lien tel: structuré)149        tel = soup.select_one('a[href^="tel:"]')150        if tel:151            m = re.search(r"\(?([2-9]\d{2})\)?[\s.\-]?(\d{3})[\s.\-]?(\d{4})",152                          tel.get("href", ""))153            if m:154                payload["phone"] = f"{m.group(1)}-{m.group(2)}-{m.group(3)}"155156        # Photos de la fiche157        payload["images"] = [u for u in dict.fromkeys(IMG_RE.findall(html))158                             if not IMG_NOISE_RE.search(u)]159        return payload160161    @staticmethod162    def _apply_detail(lst: Listing, payload: dict) -> None:163        if payload.get("description"):164            lst.description = payload["description"]165166        amenities = []167        for f in (payload.get("features") or []):168            m = re.search(r"types? d.unités?\s*:?\s*(.+)", f, re.I)169            if m and not lst.unit_type:170                # « 2 ½, 3 ½, 4 ½ et 5 ½ » -> plus petit type (standard Lou-Ka)171                lst.unit_type = normalize_unit_type(m.group(1))172            if f and not m:173                amenities.append(f)174            elif m and "," in m.group(1):175                amenities.append(f)      # garder le détail multi-types176        if amenities:177            lst.amenities = amenities[:20]178179        # Adresse structurée : civique + code postal ; arrondissement ; ville180        addr = payload.get("address_block") or {}181        for key, val in addr.items():182            if key.startswith("adresse"):183                lst.address = val184            elif key.startswith("arrondissement") and not lst.sector:185                lst.sector = val186                lst.city = infer_city(val)187        if addr.get("ville"):188            lst.city = addr["ville"]                  # ville structurée189        if addr.get("code postal") and lst.address \190                and addr["code postal"] not in lst.address:191            lst.address = f"{lst.address}, {addr['code postal']}"192193        if payload.get("phone"):194            lst.details = {**lst.details,195                           "contact": {"phone": payload["phone"]}}196197        # Compléter la galerie avec les photos de la fiche198        lst.images = list(dict.fromkeys(199            lst.images + (payload.get("images") or [])))[:25]200