# ----------------------------------------------------------------------------- # Lou-Ka — Agrégateur de logements à louer (province de Québec) # Auteur : Simon-Pierre Boucher — contact@spboucher.ai # connectors/kangalou.py : Kangalou (kangalou.com) — portail locatif québécois # géré (Solutions Kangalou, ~8 700 annonces province entière). Découverte en # UNE requête : l'endpoint carte `POST /fr/recherche/ajax/search-map/` # (pinsOnly=true) renvoie tous les pins JSON — id, URL de fiche, GPS, ville, # type (ti/tn ; 10=Commercial et 11=Garage exclus), « n ½ » (an), prix (mc), # photo et l'état loué (ir, exclu). La fiche /fr/annonce/// (SSR, # aucun anti-bot) complète : adresse civique, chambres/salles de bain, loyer, # étage, superficie, disponibilité, description longue, sections « toggler » # (Inclusions, Caractéristiques, Animaux, Sécurité, À proximité, Transport…) # et la galerie COMPLÈTE pleine résolution (gr-modal-gallery-src, sans # paramètre de redimensionnement) — via TtlDetailCache (budget/TTL). Le # budget par défaut couvre tout le parc (~8 000 fiches) : le premier sync est # long (~1 h 30), les suivants ne re-visitent que les fiches périmées (TTL). # L'API de liste HTML (POST /fr/api/search) existe aussi mais pagine par 29 # et masque l'état loué (CSS) : les pins sont la source fiable. # ----------------------------------------------------------------------------- from __future__ import annotations import os import re from ..schema import Listing, normalize_unit_type from .base import BaseConnector from . import _detailutil as du BASE = "https://www.kangalou.com" PINS_API = f"{BASE}/fr/recherche/ajax/search-map/" DETAIL_LIMIT = int(os.environ.get("LOUKA_KANGALOU_DETAIL_LIMIT", "9000")) TTL_DAYS = float(os.environ.get("LOUKA_KANGALOU_TTL_DAYS", "7")) MAX_PINS = int(os.environ.get("LOUKA_KANGALOU_MAX", "0")) # 0 = tout # type id -> type d'unité par défaut (le « n ½ » du pin prime quand présent) _TYPES = {1: "", 2: "Condo", 3: "Maison", 4: "Chambre", 5: "Chalet", 6: "Studio", 7: "Loft"} _EXCLUDED_TI = {10, 11} # Commercial, Garage : pas du résidentiel # fiche détail (SSR) _GPS_RE = re.compile(r'maps/search/\?query=(-?\d+\.\d+)%2C(-?\d+\.\d+)') _ADDR_RE = re.compile(r'buds--separator[^>]*u-text-size-sm">\s*([^<]+)', re.S) # galerie : pleine taille dans gr-modal-gallery-src (les sont lazyload) ; # hébergement variable (images/glide kangalou OU s3 shareimmo/lws…) _GALLERY_RE = re.compile(r'gr-modal-gallery-src="(https?://[^"]+)"') _FLAT_FIELD = { "loyer": re.compile(r"Loyer:\s*\|\s*([\d\s ,]+\$\s*/\s*mois)"), "etage": re.compile(r"Étage:\s*\|\s*(\d{1,2})\b"), "sdb": re.compile(r"Nombre de salles? de bain:\s*\|\s*(\d+(?:[.,]5)?)"), "chambres": re.compile(r"Nombre de chambres:\s*\|\s*(\d+)"), "superficie": re.compile(r"Superficie:\s*\|\s*([\d\s ,.]+)\s*m\s*\|\s*2"), "dispo": re.compile(r"Disponibilité:\s*\|\s*([^|]{1,60})"), } # description : le titre varie selon le type (« Description du logement / # du condo / de la maison / du studio… à louer ») — matcher générique. # Bloc « typography » complet ; borne préférée = le déclencheur « Voir plus » # (gr-read-more-trigger) pour survivre aux
imbriqués, sinon repli sur # le premier
(bloc simple

…

) _DESC_FULL_RE = re.compile( r'Description d[^<]{0,60}.*?
]*>' r'(.*?)
\s*
.*?
]*>(.*?)
', re.S) # sections « toggler » de la fiche :
Titre
# suivi d'items

libellé[valeurs]

_SECTION_ITEM_RE = re.compile( r"

\s*(.*?)(?:\s*(.*?))?", re.S) # sections « à proximité » -> details ; le reste (Inclusions, Caractéristiques # intérieures/extérieures, Détail, Animaux, Sécurité, Buanderie, Autres # services, Critères importants…) -> amenities _PROXIMITY_TITLES = {"À proximité", "Transport"} _PETS_NO = {"Aucun animal autorisé", "Aucun animal accepté", "Animaux refusés"} _PETS_YES = {"Animaux acceptés", "Animaux autorisés", "Tous animaux autorisés"} def _clean_text(fragment: str) -> str: """HTML -> texte propre (br/p/li = sauts de ligne, entités décodées).""" import html as _h txt = re.sub(r"", "\n", fragment) txt = re.sub(r"", "\n", txt) txt = re.sub(r"]*>", "- ", txt) txt = re.sub(r"<[^>]+>", " ", txt) txt = _h.unescape(txt) txt = re.sub(r"[ \t]+", " ", txt) txt = re.sub(r" ?\n ?", "\n", txt) return re.sub(r"\n{3,}", "\n\n", txt).strip() def _toggler_sections(html: str) -> list[tuple[str, list[tuple[str, str]]]]: """[(titre de section, [(libellé, valeur)])] depuis les blocs toggler.""" import html as _h out = [] for seg in re.split(r"gr-toggler-trigger>", html)[1:]: title = _h.unescape(seg.split("<", 1)[0]).strip() items = [] for lab, val in _SECTION_ITEM_RE.findall(seg): lab = _h.unescape(re.sub(r"<[^>]+>", " ", lab)).strip() val = _h.unescape(re.sub(r"<[^>]+>", " ", val or "")).strip() if lab and len(lab) <= 60: items.append((lab, val)) if title and items: out.append((title, items)) return out def _parse_fiche(html: str) -> dict: """Champs riches d'une fiche Kangalou (SSR).""" out: dict = {} m = _GPS_RE.search(html) if m: out["lat"], out["lng"] = float(m.group(1)), float(m.group(2)) m = _ADDR_RE.search(html) if m: import html as _h # certaines fiches double-encodent (« Carri&egrave;res ») : deux passes out["address"] = _h.unescape(_h.unescape(m.group(1))).strip() flat = du.flatten(html) vals = {k: rx.search(flat) for k, rx in _FLAT_FIELD.items()} if vals["loyer"]: label = re.sub(r"\s+", " ", vals["loyer"].group(1)).strip() out["price_label"] = label try: out["price"] = float(re.sub(r"[^\d,.]", "", label.split("$")[0]) .replace(",", "")) except ValueError: pass if not (100 <= (out.get("price") or 0) <= 20000): out.pop("price", None) if vals["chambres"]: out["bedrooms"] = float(vals["chambres"].group(1)) if vals["sdb"]: out["bathrooms"] = float(vals["sdb"].group(1).replace(",", ".")) if vals["dispo"]: out["availability"] = vals["dispo"].group(1).strip(" .") if vals["etage"]: out.setdefault("details", {})["floor"] = int(vals["etage"].group(1)) if vals["superficie"]: try: m2 = float(vals["superficie"].group(1) .replace(" ", "").replace(" ", "").replace(",", ".")) if 8 <= m2 <= 2000: out["area_sqft"] = round(m2 * 10.7639) except ValueError: pass m = _DESC_FULL_RE.search(html) or _DESC_RE.search(html) if m: desc = _clean_text(m.group(1)) if desc: out["description"] = desc[:6000] # sections « toggler » : commodités structurées + proximité + animaux. # Libellés jamais interprétés (texte source) ; « libellé : valeurs » # quand la section fournit des valeurs (ex. Type de plancher : Céramique). amen: list[str] = [] proximity: dict[str, str] = {} pets_labels: list[str] = [] for title, items in _toggler_sections(html): if title in _PROXIMITY_TITLES: for lab, val in items: proximity[lab] = val continue for lab, val in items: entry = f"{lab} : {val}" if val else lab if entry not in amen: amen.append(entry) if title == "Animaux": pets_labels += [lab for lab, _ in items] if amen: out["amenities"] = amen if proximity: out.setdefault("details", {})["proximity"] = proximity # animaux / meublé : depuis les libellés structurés seulement (le texte # libre est laissé à la normalisation centrale) if pets_labels: if any(l in _PETS_NO for l in pets_labels): out["pets"] = "non" elif any(l in _PETS_YES for l in pets_labels): out["pets"] = "oui" else: # chat seulement, petit chien, en cage… out["pets"] = "conditions" if "Meublé" in amen: out["furnished"] = True # galerie (slider--gallery ; les vignettes asnav dupliquées sont exclues) ig = html.find("slider--gallery") if ig >= 0: seg = html[ig:] cut = seg.find("slider__asnav") if cut > 0: seg = seg[:cut] photos = [] for u in _GALLERY_RE.findall(seg): if u not in photos: photos.append(u) if photos: out["images"] = photos vt = du.virtual_tour(html) if vt: out.setdefault("details", {})["virtual_tour"] = vt return out class KangalouConnector(BaseConnector): source_id = "kangalou" request_delay = 0.5 timeout = 90 # la réponse pins fait ~6 Mo def _pins(self) -> list[dict]: import time import requests # le POST n'a pas l'escalade résiliente de get() : retenter les # lenteurs transitoires du site (ReadTimeout constaté 2026-08-24) for attempt in range(3): try: resp = self.post(PINS_API, data={"pinsOnly": "true"}, headers={"Accept": "application/json", "X-Requested-With": "XMLHttpRequest"}) break except (requests.ConnectionError, requests.Timeout): if attempt == 2: raise time.sleep(15 * (attempt + 1)) return (resp.json() or {}).get("pins") or [] @staticmethod def _unit_type(pin: dict) -> str: an = (pin.get("an") or "").strip() # « 4 ½ », « 9 ½ + » if an: ut = normalize_unit_type(an) if re.match(r"^\d½$|^6½\+$", ut or ""): return ut return _TYPES.get(pin.get("ti"), "") def fetch(self) -> list[Listing]: pins = self._pins() if MAX_PINS: pins = pins[:MAX_PINS] # key=v2 : parseur enrichi 2026-08-18 (galerie complète déjà en v1 ; # ajout loyer/pets/meublé/proximité + description multi-blocs) — force # une re-visite du parc complet une seule fois cache = du.TtlDetailCache(self, budget=DETAIL_LIMIT, ttl_days=TTL_DAYS, key="v2") out: list[Listing] = [] try: for pin in pins: if pin.get("ir"): continue # déjà loué : pas une offre active if pin.get("ti") in _EXCLUDED_TI or not pin.get("id"): continue # commercial / garage / malformé url = pin.get("l") or "" if not url: continue eid = str(pin["id"]) price = None try: price = float(str(pin.get("mc")).replace(",", "")) except (TypeError, ValueError): pass if price is not None and not (100 <= price <= 20000): price = None # affichage « sur demande » etc. lat = lng = None try: lat, lng = float(pin["lt"]), float(pin["lg"]) except (KeyError, TypeError, ValueError): pass photo = pin.get("p") or "" lst = Listing( source=self.source_id, external_id=eid, url=url, title=(pin.get("n") or "").strip(), city=(pin.get("c") or "").strip(), unit_type=self._unit_type(pin), price=price, price_label=(f"{price:,.0f} $/mois".replace(",", " ") if price else ""), images=[photo] if photo else [], lat=lat, lng=lng, ) payload = cache.get(eid, url, _parse_fiche) if payload: du.apply_detail(lst, payload) if payload.get("address"): lst.title = payload["address"] out.append(lst) finally: cache.close() return out