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/logisquebec.py : LogisQuébec (logisquebec.com) — portail locatif5#   québécois (Réglisse Média). ~9 800 fiches location, province entière.6#   Découverte par les sitemaps (99,7 % du parc vérifié vs compteurs des pages7#   liste) : URLs de fiche « {type}-a-louer-{ville}-l###### ». Chaque fiche8#   (SSR, Cloudflare tolérant) expose un JSON-LD complet : SingleFamilyResidence9#   (adresse + code postal + GPS + téléphone) et Product (nom, description,10#   offers.price, sku = id). Galerie lightGallery, disponibilité et11#   caractéristiques actives (span .caracteristique-texte) dans le HTML.12#   Fiche expirée -> HTTP 410 Gone (retrait immédiat). Cache détail avec TTL13#   (les prix changent) via _detailutil.TtlDetailCache.14#   Superficie : la fiche n'a AUCUN champ structuré (ni JSON-LD floorSize, ni15#   caractéristique dédiée — vérifié 2026-08-18) ; seule l'extraction texte de16#   finalize() (description/commodités) s'applique, ~26 % du parc en donne une.17# -----------------------------------------------------------------------------18from __future__ import annotations1920import os21import re2223from bs4 import BeautifulSoup2425from ..schema import Listing, normalize_unit_type26from .base import BaseConnector27from . import _detailutil as du2829BASE = "https://www.logisquebec.com"3031# type d'URL -> (type d'unité par défaut, court terme ?) ; les slugs32# commerciaux (local-commercial, bureau, entrepot-industriel…) sont exclus33TYPES = {34    "appartement": ("", False),35    "condo": ("", False),36    "maison": ("Maison", False),37    "studio": ("Studio", False),38    "loft": ("Loft", False),39    "chambre-et-colocation": ("Chambre", False),40    "chalet": ("Chalet", False),41    "appartement-condo-court-terme": ("", True),42    "maison-court-terme": ("Maison", True),43    "chambre-court-terme": ("Chambre", True),44}4546_FICHE_RE = re.compile(47    r"^https://www\.logisquebec\.com/(" + "|".join(map(re.escape, TYPES)) +48    r")-a-louer-([a-z0-9_-]+)-l(\d+)$")49_LOC_RE = re.compile(r"<loc>(.*?)</loc>")50_PHOTO_RE = re.compile(r"https://i\.logisquebec\.com/i-a-louer/\d+/\d+/[^'\"]+")51_DISPO_RE = re.compile(r"Disponible à partir de\s*:?\s*\|\s*([^|]{1,60})")5253# Dimensionnement du budget détail : parc ~6 000 fiches, TTL 7 j, watch ~1054# passes/jour -> ~90 rafraîchissements/passe suffisent à tenir le TTL. 40055# laisse 4x de marge (poussées de nouvelles fiches) en ramenant le cycle de56# ~19 min à ~5 min. Les fiches JAMAIS VUES sont servies en premier (voir57# fetch) : aucune perte de volume, le parc se complète au même rythme.58DETAIL_LIMIT = int(os.environ.get("LOUKA_LOGISQUEBEC_DETAIL_LIMIT", "400"))59TTL_DAYS = float(os.environ.get("LOUKA_LOGISQUEBEC_TTL_DAYS", "7"))60MAX_FICHES = int(os.environ.get("LOUKA_LOGISQUEBEC_MAX", "0"))   # 0 = tout616263def _parse_fiche(html: str) -> dict:64    """Extrait le payload complet d'une page fiche LogisQuébec."""65    out: dict = {}66    for node in du.ld_nodes(html):67        t = node.get("@type") or ""68        if t == "SingleFamilyResidence":69            addr = node.get("address") or {}70            out["address"] = (addr.get("streetAddress") or "").strip()71            out["locality"] = (addr.get("addressLocality") or "").strip()72            out["postal_code"] = (addr.get("postalCode") or "").strip()73            geo = node.get("geo") or {}74            try:75                out["lat"] = float(geo.get("latitude"))76                out["lng"] = float(geo.get("longitude"))77            except (TypeError, ValueError):78                pass79            if node.get("telephone"):80                out["telephone"] = str(node["telephone"]).strip()81            rooms = node.get("numberOfRooms")82            if isinstance(rooms, (int, float)) and rooms > 0:83                out["rooms"] = int(rooms)84        elif t == "Product":85            out["title"] = (node.get("name") or "").strip()86            out["description"] = (node.get("description") or "").strip()[:6000]87            offers = node.get("offers") or {}88            try:89                out["price"] = float(str(offers.get("price")).replace(",", "."))90            except (TypeError, ValueError):91                pass92            if node.get("sku"):93                out["sku"] = str(node["sku"])9495    photos, seen = [], set()96    for u in _PHOTO_RE.findall(html):97        if "/thumbnail/" in u or u in seen:98            continue99        seen.add(u)100        photos.append(u)101    if photos:102        out["images"] = photos103104    # « Disponible à partir de : Immédiatement / 1er juillet… » (texte aplati)105    flat = du.flatten(html)106    m = _DISPO_RE.search(flat)107    if m:108        out["availability"] = m.group(1).strip(" .")109110    # caractéristiques ACTIVES = items possédant un span .caracteristique-texte111    # (les items sans texte sont la légende inactive des icônes)112    soup = BeautifulSoup(html, "html.parser")113    amenities = []114    for it in soup.select(".caracteristique-item"):115        if it.select_one(".caracteristique-texte") is None:116            continue117        label = it.get_text(" ", strip=True)118        label = re.sub(r"\s+", " ", label)119        # le titre est dupliqué dans le texte : « 2 chambres 2 chambres »120        half = len(label) // 2121        if len(label) % 2 == 1 and label[:half] == label[half + 1:]:122            label = label[:half]123        if label and label not in amenities:124            amenities.append(label)125    if amenities:126        out["amenities"] = amenities127128    vt = du.virtual_tour(html)      # lien Matterport/iGUIDE… si la fiche en a un129    if vt:130        out["virtual_tour"] = vt131    return out132133134class LogisQuebecConnector(BaseConnector):135    source_id = "logisquebec"136    request_delay = 0.5137138    # -- découverte ------------------------------------------------------------139    def _fiche_urls(self) -> list[tuple[str, str, str, str]]:140        """(url, type_slug, ville_slug, id) des fiches location résidentielles."""141        index = self.get(f"{BASE}/sitemap.xml").text142        subs = [u for u in _LOC_RE.findall(index) if "/sitemap/" in u]143        out, seen = [], set()144        for sub in subs:145            try:146                xml = self.get(sub).text147            except Exception:148                continue149            for u in _LOC_RE.findall(xml):150                m = _FICHE_RE.match(u)151                if m and m.group(3) not in seen:152                    seen.add(m.group(3))153                    out.append((u, m.group(1), m.group(2), m.group(3)))154        return out155156    # -- construction ----------------------------------------------------------157    def _to_listing(self, url: str, type_slug: str, payload: dict) -> Listing | None:158        if not payload or payload.get("gone"):159            return None160        eid = payload.get("sku") or url.rsplit("-l", 1)[-1]161        default_type, court_terme = TYPES[type_slug]162163        title = payload.get("title") or ""164        unit_type = normalize_unit_type(title)165        if not re.match(r"^\d½|^6½\+$", unit_type or ""):166            unit_type = ""167        if not unit_type and payload.get("rooms"):168            n = payload["rooms"]169            unit_type = "6½+" if n >= 6 else f"{n}½"170        if not unit_type:                # « 2 chambres » des caractéristiques171            for a in payload.get("amenities") or []:172                if re.match(r"^\d+ chambres?$", a):173                    unit_type = normalize_unit_type(a)174                    break175        if not unit_type:176            unit_type = default_type177178        locality = payload.get("locality") or ""179        m = re.match(r"^(.*?)\s*\((.+)\)\s*$", locality)180        sector, city = (m.group(1), m.group(2)) if m else ("", locality)181182        price = payload.get("price")183        # clés canoniques (mêmes noms que normalize.extract_details : le184        # merge_details de finalize() fusionne contact{} clé par clé)185        details: dict = {}186        if payload.get("postal_code"):187            details["postal_code"] = payload["postal_code"]188        if payload.get("telephone"):189            details["contact"] = {"phone": payload["telephone"]}190        if payload.get("rooms"):191            details["rooms"] = payload["rooms"]   # numberOfRooms (JSON-LD)192        if court_terme:193            details["Location court terme"] = "oui"194        if payload.get("virtual_tour"):195            details["virtual_tour"] = payload["virtual_tour"]196197        amenities = payload.get("amenities") or []198        lst = Listing(199            source=self.source_id,200            external_id=str(eid),201            url=url,202            title=title,203            address=payload.get("address") or "",204            sector=sector,205            city=city,206            unit_type=unit_type,207            price=price,208            price_label=(f"{price:,.0f} $/mois".replace(",", " ") if price else ""),209            availability=payload.get("availability") or "",210            description=payload.get("description") or "",211            amenities=amenities,212            details=details,213            images=payload.get("images") or [],214            lat=payload.get("lat"),215            lng=payload.get("lng"),216        )217        low = [a.lower() for a in amenities]218        if any("animaux accept" in a for a in low):219            lst.pets = "oui"220        if "meublé" in low:221            lst.furnished = True222        return lst223224    def fetch(self) -> list[Listing]:225        fiches = self._fiche_urls()226        if MAX_FICHES:227            fiches = fiches[:MAX_FICHES]228        cache = du.TtlDetailCache(self, budget=DETAIL_LIMIT, ttl_days=TTL_DAYS)229        # priorité du budget aux fiches jamais visitées (tri stable) : les230        # nouveautés sont toujours captées au premier passage, le reste du231        # budget sert au rafraîchissement TTL232        seen = {r["external_id"] for r in cache.con.execute(233            "SELECT external_id FROM detail_cache WHERE source=?",234            (self.source_id,))}235        fiches.sort(key=lambda f: f[3] in seen)236        out: list[Listing] = []237        try:238            for url, type_slug, _ville, eid in fiches:239                payload = cache.get(eid, url, _parse_fiche)240                if payload is None:      # jamais visitée + budget épuisé :241                    continue             # captée à une prochaine synchronisation242                lst = self._to_listing(url, type_slug, payload)243                if lst is not None:244                    out.append(lst)245        finally:246            cache.close()247        return out248