# ----------------------------------------------------------------------------- # Lou-Ka — Agrégateur de logements à louer (province de Québec) # Auteur : Simon-Pierre Boucher — contact@spboucher.ai # connectors/immeubles_dcl.py : connecteur Les Immeubles DCL # (lesimmeublesdcl.com — Rouyn-Noranda, Val-d'Or, Malartic ; volet logements # abordables). Site Laravel + **FleetCart** (catalogue e-commerce détourné # en vitrine locative, Vue.js côté client) : la page /logements-a-louer est # vide côté serveur, mais son API interne répond en JSON structuré — # POST /logements-a-louer (X-CSRF-TOKEN lu dans `window.FleetCart`, session # Laravel du GET initial) -> products.data : adresse, ville, code postal, # **GPS (lat_lng)**, prix, catégorie (typologie « 1 ½ »… -> unit_type) et # attributs (Ameublement -> meublé, Animaux -> pets, Pièces, Chambres, # Salle de bain, Commodités, Inclusions, Secteurs, Disponibilité). Fiche # /logements-a-louer/ (via cache BD) : description et galerie complète # (images S3 `img.popup-img`). Les logements loués vivent dans un catalogue # séparé (/logements-loues) : jamais mélangés. robots.txt ouvert. # ----------------------------------------------------------------------------- from __future__ import annotations import hashlib import re import time from bs4 import BeautifulSoup from ..schema import Listing, normalize_unit_type from .base import BaseConnector BASE = "https://www.lesimmeublesdcl.com" LIST_URL = f"{BASE}/logements-a-louer" _CSRF_RE = re.compile(r"csrfToken:\s*'([^']+)'") def _attr_values(product: dict) -> dict[str, list[str]]: """Attributs FleetCart du produit -> {nom: [valeurs]}.""" out: dict[str, list[str]] = {} for attr in product.get("attributes") or []: name = (attr.get("name") or "").strip() vals = [v.get("value", "").strip() for v in attr.get("values") or [] if v.get("value")] if name and vals: out[name] = vals return out class ImmeublesDCLConnector(BaseConnector): source_id = "immeubles_dcl" request_delay = 0.6 max_details = 25 # garde-fou fiches détail (vraies requêtes par sync) max_pages = 10 # garde-fou pagination API def _post_json(self, url: str, payload: dict, token: str): """POST JSON avec le même throttling poli que get().""" wait = self.request_delay - (time.time() - self._last_request) if wait > 0: time.sleep(wait) resp = self.session.post( url, json=payload, timeout=self.timeout, headers={"X-CSRF-TOKEN": token, "X-Requested-With": "XMLHttpRequest", "Accept": "application/json"}) self._last_request = time.time() resp.raise_for_status() return resp def fetch(self) -> list[Listing]: # 1) GET la page : cookie de session Laravel + jeton CSRF (window.FleetCart) html = self.get(LIST_URL).text m = _CSRF_RE.search(html) if not m: raise RuntimeError("csrfToken introuvable sur /logements-a-louer") token = m.group(1) # 2) API products.index (JSON), paginée via last_page self._fetched = 0 listings: dict[str, Listing] = {} page = 1 while page <= self.max_pages: payload = {"query": "", "fromPrice": 0, "toPrice": 100000, "perPage": 50, "page": page, "attribute": {}, "sort": "latest"} data = self._post_json(LIST_URL, payload, token).json() products = (data.get("products") or {}) for prod in products.get("data") or []: try: self._parse_product(prod, listings) except Exception: continue if page >= int(products.get("last_page") or 1): break page += 1 return list(listings.values()) # -- produit FleetCart (JSON) -------------------------------------------------------- def _parse_product(self, prod: dict, listings: dict[str, Listing]) -> None: ext_id = str(prod.get("id") or "") slug = prod.get("slug") or "" if not ext_id or not slug or ext_id in listings: return url = f"{LIST_URL}/{slug}" name = (prod.get("name") or "").strip() address = (prod.get("address") or "").strip() city = (prod.get("city") or "").strip() zip_code = (prod.get("zip") or "").strip() full_address = ", ".join(p for p in (address, city, zip_code) if p) # prix structuré (selling_price) + libellé du site (« 900 $CA ») price = None selling = prod.get("selling_price") or {} cur = selling.get("inCurrentCurrency") or {} if isinstance(cur.get("amount"), (int, float)): price = float(cur["amount"]) # « 1 425 $CA » : espaces (fines) insécables -> espace simple price_label = re.sub(r"\s+", " ", prod.get("formatted_price") or "").strip() # typologie = catégorie du catalogue (« 1 ½ », « 4 ½ »…) unit_type = "" for cat in prod.get("categories") or []: cname = (cat.get("name") or "").strip() if re.match(r"^\d\s*½", cname): unit_type = normalize_unit_type(cname.replace("½", "1/2")) break attrs = _attr_values(prod) availability = ", ".join(attrs.get("Disponibilité", [])) sector = ", ".join(attrs.get("Secteurs", [])) # animaux : champ structuré du site (prudence : conditions si permis) pets = None pets_raw = " / ".join(attrs.get("Animaux", [])) if pets_raw: if re.search(r"aucun|pas accept|non", pets_raw, re.I): pets = "non" elif re.search(r"accept|autoris", pets_raw, re.I): pets = "conditions" furnished = None furn_raw = " ".join(attrs.get("Ameublement", [])) if re.search(r"^non\s*meubl", furn_raw.strip(), re.I): furnished = False elif re.search(r"meubl", furn_raw, re.I): furnished = True amenities: list[str] = [] if attrs.get("Pièces"): amenities.append(f"{attrs['Pièces'][0]} pièce(s)") if attrs.get("Chambres à coucher"): amenities.append(f"{attrs['Chambres à coucher'][0]} chambre(s)") if attrs.get("Salle de bain"): amenities.append(f"{attrs['Salle de bain'][0]} salle(s) de bain") for grp in ("Commodités", "Commodités de l'immeuble"): amenities.extend(attrs.get(grp, [])) if attrs.get("Inclusions"): amenities.append("Inclus : " + ", ".join(attrs["Inclusions"])) # GPS structuré (lat_lng) — validé ensuite par finalize() lat = lng = None latlng = prod.get("lat_lng") or [] if isinstance(latlng, list) and len(latlng) == 2: try: lat, lng = float(latlng[0]), float(latlng[1]) except (TypeError, ValueError): lat = lng = None images = [m_.get("path", "") for m_ in prod.get("media") or [] if isinstance(m_, dict) and m_.get("path", "").startswith("http")] lst = Listing( source=self.source_id, external_id=ext_id, url=url, title=name, address=full_address, sector=sector, city=city, unit_type=unit_type, price=price, price_label=price_label, availability=availability, pets=pets, furnished=furnished, amenities=list(dict.fromkeys(amenities))[:25], images=images, lat=lat, lng=lng, ) key = hashlib.sha1( f"{name}|{price_label}|{availability}|{sector}" .encode("utf-8")).hexdigest() try: payload = self.detail(ext_id, key, lambda u=url: self._fetch_detail(u)) self._apply_detail(lst, payload) except Exception: pass listings[ext_id] = lst # -- fiche détail (/logements-a-louer/) ---------------------------------------- def _fetch_detail(self, url: str) -> dict: """Description libre et galerie complète (images S3 du carrousel).""" if self._fetched >= self.max_details: raise RuntimeError("budget de fiches détail atteint") self._fetched += 1 html = self.get(url).text soup = BeautifulSoup(html, "html.parser") out: dict = {} # bloc « Description » : titre (h5) suivi du texte libre for h in soup.find_all(["h2", "h3", "h4", "h5"]): if h.get_text(strip=True).lower() == "description": parts = [] for sib in h.find_next_siblings(): if sib.name in ("h2", "h3", "h4"): break parts.append(sib.get_text("\n", strip=True)) txt = "\n".join(p for p in parts if p) if txt: out["description"] = re.sub(r"[ \t]+", " ", txt).strip()[:1500] break # galerie : liens pleine taille du carrousel (a.popup-img -> S3) images: list[str] = [] for a in soup.select("a.popup-img[href]"): u = a["href"].strip() if u.startswith("http") and u not in images: images.append(u) out["images"] = images[:25] return out def _apply_detail(self, lst: Listing, d: dict) -> None: """Reporte le payload (frais ou en cache) sur l'annonce.""" if not d: return if d.get("description"): lst.description = d["description"] if d.get("images") and len(d["images"]) > len(lst.images): lst.images = d["images"]