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/immeubles_dcl.py : connecteur Les Immeubles DCL5#   (lesimmeublesdcl.com — Rouyn-Noranda, Val-d'Or, Malartic ; volet logements6#   abordables). Site Laravel + **FleetCart** (catalogue e-commerce détourné7#   en vitrine locative, Vue.js côté client) : la page /logements-a-louer est8#   vide côté serveur, mais son API interne répond en JSON structuré —9#   POST /logements-a-louer (X-CSRF-TOKEN lu dans `window.FleetCart`, session10#   Laravel du GET initial) -> products.data : adresse, ville, code postal,11#   **GPS (lat_lng)**, prix, catégorie (typologie « 1 ½ »… -> unit_type) et12#   attributs (Ameublement -> meublé, Animaux -> pets, Pièces, Chambres,13#   Salle de bain, Commodités, Inclusions, Secteurs, Disponibilité). Fiche14#   /logements-a-louer/<slug> (via cache BD) : description et galerie complète15#   (images S3 `img.popup-img`). Les logements loués vivent dans un catalogue16#   séparé (/logements-loues) : jamais mélangés. robots.txt ouvert.17# -----------------------------------------------------------------------------18from __future__ import annotations1920import hashlib21import re22import time2324from bs4 import BeautifulSoup2526from ..schema import Listing, normalize_unit_type27from .base import BaseConnector2829BASE = "https://www.lesimmeublesdcl.com"30LIST_URL = f"{BASE}/logements-a-louer"3132_CSRF_RE = re.compile(r"csrfToken:\s*'([^']+)'")333435def _attr_values(product: dict) -> dict[str, list[str]]:36    """Attributs FleetCart du produit -> {nom: [valeurs]}."""37    out: dict[str, list[str]] = {}38    for attr in product.get("attributes") or []:39        name = (attr.get("name") or "").strip()40        vals = [v.get("value", "").strip()41                for v in attr.get("values") or [] if v.get("value")]42        if name and vals:43            out[name] = vals44    return out454647class ImmeublesDCLConnector(BaseConnector):48    source_id = "immeubles_dcl"49    request_delay = 0.650    max_details = 25     # garde-fou fiches détail (vraies requêtes par sync)51    max_pages = 10       # garde-fou pagination API5253    def _post_json(self, url: str, payload: dict, token: str):54        """POST JSON avec le même throttling poli que get()."""55        wait = self.request_delay - (time.time() - self._last_request)56        if wait > 0:57            time.sleep(wait)58        resp = self.session.post(59            url, json=payload, timeout=self.timeout,60            headers={"X-CSRF-TOKEN": token,61                     "X-Requested-With": "XMLHttpRequest",62                     "Accept": "application/json"})63        self._last_request = time.time()64        resp.raise_for_status()65        return resp6667    def fetch(self) -> list[Listing]:68        # 1) GET la page : cookie de session Laravel + jeton CSRF (window.FleetCart)69        html = self.get(LIST_URL).text70        m = _CSRF_RE.search(html)71        if not m:72            raise RuntimeError("csrfToken introuvable sur /logements-a-louer")73        token = m.group(1)7475        # 2) API products.index (JSON), paginée via last_page76        self._fetched = 077        listings: dict[str, Listing] = {}78        page = 179        while page <= self.max_pages:80            payload = {"query": "", "fromPrice": 0, "toPrice": 100000,81                       "perPage": 50, "page": page, "attribute": {},82                       "sort": "latest"}83            data = self._post_json(LIST_URL, payload, token).json()84            products = (data.get("products") or {})85            for prod in products.get("data") or []:86                try:87                    self._parse_product(prod, listings)88                except Exception:89                    continue90            if page >= int(products.get("last_page") or 1):91                break92            page += 193        return list(listings.values())9495    # -- produit FleetCart (JSON) --------------------------------------------------------96    def _parse_product(self, prod: dict, listings: dict[str, Listing]) -> None:97        ext_id = str(prod.get("id") or "")98        slug = prod.get("slug") or ""99        if not ext_id or not slug or ext_id in listings:100            return101        url = f"{LIST_URL}/{slug}"102103        name = (prod.get("name") or "").strip()104        address = (prod.get("address") or "").strip()105        city = (prod.get("city") or "").strip()106        zip_code = (prod.get("zip") or "").strip()107        full_address = ", ".join(p for p in (address, city, zip_code) if p)108109        # prix structuré (selling_price) + libellé du site (« 900 $CA »)110        price = None111        selling = prod.get("selling_price") or {}112        cur = selling.get("inCurrentCurrency") or {}113        if isinstance(cur.get("amount"), (int, float)):114            price = float(cur["amount"])115        # « 1 425 $CA » : espaces (fines) insécables -> espace simple116        price_label = re.sub(r"\s+", " ",117                             prod.get("formatted_price") or "").strip()118119        # typologie = catégorie du catalogue (« 1 ½ », « 4 ½ »…)120        unit_type = ""121        for cat in prod.get("categories") or []:122            cname = (cat.get("name") or "").strip()123            if re.match(r"^\d\s*½", cname):124                unit_type = normalize_unit_type(cname.replace("½", "1/2"))125                break126127        attrs = _attr_values(prod)128        availability = ", ".join(attrs.get("Disponibilité", []))129        sector = ", ".join(attrs.get("Secteurs", []))130131        # animaux : champ structuré du site (prudence : conditions si permis)132        pets = None133        pets_raw = " / ".join(attrs.get("Animaux", []))134        if pets_raw:135            if re.search(r"aucun|pas accept|non", pets_raw, re.I):136                pets = "non"137            elif re.search(r"accept|autoris", pets_raw, re.I):138                pets = "conditions"139140        furnished = None141        furn_raw = " ".join(attrs.get("Ameublement", []))142        if re.search(r"^non\s*meubl", furn_raw.strip(), re.I):143            furnished = False144        elif re.search(r"meubl", furn_raw, re.I):145            furnished = True146147        amenities: list[str] = []148        if attrs.get("Pièces"):149            amenities.append(f"{attrs['Pièces'][0]} pièce(s)")150        if attrs.get("Chambres à coucher"):151            amenities.append(f"{attrs['Chambres à coucher'][0]} chambre(s)")152        if attrs.get("Salle de bain"):153            amenities.append(f"{attrs['Salle de bain'][0]} salle(s) de bain")154        for grp in ("Commodités", "Commodités de l'immeuble"):155            amenities.extend(attrs.get(grp, []))156        if attrs.get("Inclusions"):157            amenities.append("Inclus : " + ", ".join(attrs["Inclusions"]))158159        # GPS structuré (lat_lng) — validé ensuite par finalize()160        lat = lng = None161        latlng = prod.get("lat_lng") or []162        if isinstance(latlng, list) and len(latlng) == 2:163            try:164                lat, lng = float(latlng[0]), float(latlng[1])165            except (TypeError, ValueError):166                lat = lng = None167168        images = [m_.get("path", "") for m_ in prod.get("media") or []169                  if isinstance(m_, dict) and m_.get("path", "").startswith("http")]170171        lst = Listing(172            source=self.source_id,173            external_id=ext_id,174            url=url,175            title=name,176            address=full_address,177            sector=sector,178            city=city,179            unit_type=unit_type,180            price=price,181            price_label=price_label,182            availability=availability,183            pets=pets,184            furnished=furnished,185            amenities=list(dict.fromkeys(amenities))[:25],186            images=images,187            lat=lat,188            lng=lng,189        )190191        key = hashlib.sha1(192            f"{name}|{price_label}|{availability}|{sector}"193            .encode("utf-8")).hexdigest()194        try:195            payload = self.detail(ext_id, key,196                                  lambda u=url: self._fetch_detail(u))197            self._apply_detail(lst, payload)198        except Exception:199            pass200        listings[ext_id] = lst201202    # -- fiche détail (/logements-a-louer/<slug>) ----------------------------------------203    def _fetch_detail(self, url: str) -> dict:204        """Description libre et galerie complète (images S3 du carrousel)."""205        if self._fetched >= self.max_details:206            raise RuntimeError("budget de fiches détail atteint")207        self._fetched += 1208        html = self.get(url).text209        soup = BeautifulSoup(html, "html.parser")210        out: dict = {}211212        # bloc « Description » : titre (h5) suivi du texte libre213        for h in soup.find_all(["h2", "h3", "h4", "h5"]):214            if h.get_text(strip=True).lower() == "description":215                parts = []216                for sib in h.find_next_siblings():217                    if sib.name in ("h2", "h3", "h4"):218                        break219                    parts.append(sib.get_text("\n", strip=True))220                txt = "\n".join(p for p in parts if p)221                if txt:222                    out["description"] = re.sub(r"[ \t]+", " ", txt).strip()[:1500]223                break224225        # galerie : liens pleine taille du carrousel (a.popup-img -> S3)226        images: list[str] = []227        for a in soup.select("a.popup-img[href]"):228            u = a["href"].strip()229            if u.startswith("http") and u not in images:230                images.append(u)231        out["images"] = images[:25]232        return out233234    def _apply_detail(self, lst: Listing, d: dict) -> None:235        """Reporte le payload (frais ou en cache) sur l'annonce."""236        if not d:237            return238        if d.get("description"):239            lst.description = d["description"]240        if d.get("images") and len(d["images"]) > len(lst.images):241            lst.images = d["images"]242