SPB Git

spb/lou-ka Public

Lou·Ka — tous les logements à louer du Québec, un seul endroit.

HTML 99.7%
12.0 KB · 274 lines python
Raw Blame History
1# -----------------------------------------------------------------------------2# Lou-Ka — Agrégateur de logements à louer (province de Québec)3# Auteur : Simon-Pierre Boucher — contact@spboucher.ai4# connectors/logisbourg.py : connecteur Logisbourg (logisbourg.com)5#   Site ASP classique : tableau des disponibilités par immeuble6#   (Le Bourgarde, Les Terrasses du Bourg, Le Bourg du Maizerets,7#   Le Bourg du Fleuve) + fiche appartement.asp pour adresse (avec no8#   d'unité), lat/lng (carte Google), superficie (encadré dédié), images,9#   caractéristiques de l'appartement et de l'immeuble, services optionnels10#   et téléphone du bureau de location. Fiches mises en cache via11#   self.detail() : re-visitées seulement quand la ligne du tableau change.12# -----------------------------------------------------------------------------13from __future__ import annotations1415import hashlib16import html as htmllib17import re18from urllib.parse import quote1920from ..schema import Listing, infer_city, normalize_unit_type, parse_price21from .base import BaseConnector2223BASE = "https://www.logisbourg.com"24LIST_URL = f"{BASE}/appartements-disponibles.asp"2526# En-têtes d'immeuble : "Charlesbourg / Lebourgneuf  : Le Bourgarde"27_COMPLEX_RE = re.compile(28    r'class="dispos_td_complexe".*?<div class="nomComplexe">\s*(.*?)<BR>\s*'29    r'<div class="nomComplexeAdresse">(.*?)</div>', re.S | re.I)30# Lignes de logement : <TR id="tr_23_" ...> ... </TR>31_ROW_RE = re.compile(r'<TR id="tr_(\d+)_".*?</TR>', re.S | re.I)32_LINK_RE = re.compile(33    r"appartement\.asp\?SID=(\d+)&CID=(\d+)&LID=(\d+)&Secteur=([^&']*)&Complexe=([^']*)'")34# Fiche : icônes/texte des caractéristiques (<div class='carac_icone'>…</div>)35_CARAC_RE = re.compile(r"<div class=['\"]carac_icone['\"][^>]*>(.*?)</div>",36                       re.S | re.I)37# Fiche : superficie dans son encadré dédié (<div id='superficie_encadre'>)38_SUPERFICIE_RE = re.compile(39    r"id=['\"]superficie_encadre['\"][^>]*>\s*([\d\s]{2,7})\s*pi", re.I)40# Fiche : coordonnées de la carte Google (new google.maps.LatLng(46.84,-71.26))41_LATLNG_RE = re.compile(r"google\.maps\.LatLng\((-?\d{1,3}\.\d+),\s*(-?\d{1,3}\.\d+)\)")42# Fiche : adresse complète avec no d'unité (« 4855, 6e Avenue Ouest # 403 »)43_ADDR_UNIT_RE = re.compile(r'<div class="nomComplexeAdresse">(.*?)</div>', re.S | re.I)444546def _strip_tags(s: str) -> str:47    return re.sub(r"\s+", " ", htmllib.unescape(re.sub(r"<[^>]+>", " ", s))).strip()484950def _fix_sup(street: str) -> str:51    """« 6 e Avenue » (issu de 6<SUP>e</SUP>) -> « 6e Avenue »."""52    return re.sub(r"(\d)\s+(e|er|re)(\s+|$)", r"\1\2 ", street).strip()535455class _DetailBudget(Exception):56    """Plafond de vraies requêtes de fiches atteint pour cette synchro."""575859class LogisbourgConnector(BaseConnector):60    source_id = "logisbourg"61    request_delay = 0.662    max_details = 80        # plafond de VRAIES requêtes de fiches par synchro6364    def __init__(self) -> None:65        super().__init__()66        self._detail_fetches = 06768    # le serveur n'annonce pas de charset : forcer l'UTF-8 déclaré dans le HTML69    def _get_text(self, url: str) -> str:70        resp = self.get(url)71        resp.encoding = "utf-8"72        return resp.text7374    def fetch(self) -> list[Listing]:75        html = self._get_text(LIST_URL)7677        # 1) Position des en-têtes d'immeuble (secteur : nom + adresse de rue)78        complexes: list[tuple[int, str, str, str]] = []   # (pos, secteur, nom, rue)79        for m in _COMPLEX_RE.finditer(html):80            header = _strip_tags(m.group(1))              # "Secteur : Nom"81            street = _strip_tags(m.group(2))              # "6e Avenue Ouest"82            sector, _, name = header.partition(":")83            complexes.append((m.start(), sector.strip(), name.strip(), street))8485        # 2) Lignes de disponibilités86        listings: dict[str, Listing] = {}87        row_keys: dict[str, str] = {}                     # lid -> hash de la ligne88        for row_m in _ROW_RE.finditer(html):89            row = row_m.group(0)90            link = _LINK_RE.search(row)91            if not link:92                continue93            sid, cid, lid, sector_q, complex_q = link.groups()94            if lid in listings:95                continue9697            # immeuble courant = dernier en-tête avant cette ligne98            sector = name = street = ""99            for pos, sec, nom, rue in complexes:100                if pos < row_m.start():101                    sector, name, street = sec, nom, rue102                else:103                    break104            sector = sector or sector_q105            name = name or complex_q106107            def cell(cls: str) -> str:108                c = re.search(r'class="%s"[^>]*>(.*?)</TD>' % cls, row, re.S | re.I)109                return _strip_tags(c.group(1)) if c else ""110111            def cell_title(cls: str) -> str:112                c = re.search(r'class="%s"\s+TITLE="([^"]*)"' % cls, row, re.I)113                return htmllib.unescape(c.group(1)).replace("\xa0", " ") if c else ""114115            unit_raw = cell("Grandeur")116            avail = cell_title("DateDispo") or cell("DateDispo")117            floor = cell_title("Etage") or cell("Etage")118            price_label = cell("Prix")119            # inclusions (icônes avec ALT dans la colonne "Inclus")120            included = [htmllib.unescape(a) for a in121                        re.findall(r'ALT="([^"]+)"', row, re.I)122                        if "favoris" not in a.lower()]123124            url = (f"{BASE}/appartement.asp?SID={sid}&CID={cid}&LID={lid}"125                   f"&Secteur={quote(sector_q)}&Complexe={quote(complex_q)}")126            amenities = list(dict.fromkeys(included))127            if floor:128                amenities.append(f"Étage : {floor}")129130            # clé de cache des fiches : contenu de la ligne du tableau131            # (préfixe v2 = version de l'extraction, pour invalider le cache132            # quand le connecteur change)133            row_keys[lid] = "v3:" + hashlib.sha1(134                " | ".join([unit_raw, avail, floor, price_label] + included)135                .encode("utf-8")).hexdigest()136137            listings[lid] = Listing(138                source=self.source_id,139                external_id=lid,140                url=url,141                title=f"{name}{unit_raw}".strip(" —"),142                sector=sector,143                city=infer_city(sector, default="Québec"),144                unit_type=normalize_unit_type(unit_raw),145                price=parse_price(price_label),146                price_label=price_label,147                availability=avail,148                amenities=amenities,149            )150151        # 3) Fiches (avec cache BD) : adresse + no d'unité, lat/lng, superficie,152        #    images, caractéristiques, services optionnels, téléphone153        for lid, lst in listings.items():154            url = lst.url155            try:156                payload = self.detail(157                    lid, row_keys.get(lid, ""),158                    lambda u=url: self._fetch_detail(u))159            except _DetailBudget:160                continue161            except Exception:162                continue163            self._apply_detail(lst, payload)164165        return list(listings.values())166167    # -- fiche appartement.asp -------------------------------------------------168    def _fetch_detail(self, url: str) -> dict:169        if self._detail_fetches >= self.max_details:170            raise _DetailBudget(url)171        self._detail_fetches += 1172        detail = self._get_text(url)173        out: dict = {}174175        # adresse complète avec no d'unité (« 4855, 6e Avenue Ouest # 403 »)176        m = _ADDR_UNIT_RE.search(detail)177        if m:178            addr = _fix_sup(_strip_tags(m.group(1)))179            out["address"] = re.sub(r"#\s*", "#", addr)180        if not out.get("address"):181            # repli : variables JS adr1/adr2 de la carte Google182            m = re.search(r'var adr1 = "([^"]*)";\s*var adr2 = "([^"]*)"', detail)183            if m and m.group(1):184                out["address"] = f"{m.group(1)}, {_fix_sup(_strip_tags(m.group(2)))}".strip(" ,")185186        # coordonnées de la carte Google (structurées)187        m = _LATLNG_RE.search(detail)188        if m:189            out["lat"], out["lng"] = float(m.group(1)), float(m.group(2))190191        # superficie : encadré dédié de la fiche (structuré)192        m = _SUPERFICIE_RE.search(detail)193        if m:194            try:195                val = float(m.group(1).replace(" ", ""))196                if 80 <= val <= 20000:197                    out["area_sqft"] = val198            except ValueError:199                pass200201        # téléphone du bureau de location (lien tel: du formulaire de visite)202        m = re.search(r"Pour une visite.*?tel:\+?1?(\d{10})", detail, re.S | re.I)203        if m:204            d = m.group(1)205            out["phone"] = f"{d[0:3]}-{d[3:6]}-{d[6:10]}"206207        # retirer commentaires et scripts avant les extractions texte208        detail = re.sub(r"<!--.*?-->", " ", detail, flags=re.S)209        detail = re.sub(r"<script.*?</script>", " ", detail, flags=re.S | re.I)210211        # images de l'immeuble : ./<secteur>/<immeuble>/images/xxx.jpg212        imgs = re.findall(213            r'(?:\./)?([a-z0-9_\-]+/[a-z0-9_\-]+/images/[^"\'\s]+'214            r'\.(?:jpg|jpeg|png|webp))', detail, re.I)215        out["images"] = [f"{BASE}/{u}" for u in dict.fromkeys(imgs)][:25]216217        # caractéristiques de l'appartement : ALT des icônes (« Entrées218        # laveuse/sécheuse incluses ») + texte visible (« Balcon 60 pi² »,219        # « Un stationnement extérieur »…)220        amen: list[str] = []221        for c in _CARAC_RE.findall(detail):222            amen += [htmllib.unescape(a or b) for a, b in223                     re.findall(r"(?:ALT|alt)=(?:\"([^\"]{3,60})\"|'([^']{3,60})')", c)]224            amen.append(_strip_tags(c))225        # caractéristiques de l'immeuble (liste <li> : Bell Fibe, Intercom…)226        m = re.search(r"Caract[ée]ristiques immeuble(.*?)(?:Services optionnels|"227                      r"Les informations)", detail, re.S | re.I)228        immeuble_txt = ""229        if m:230            immeuble_txt = _strip_tags(m.group(1))231            amen += [_strip_tags(li) for li in232                     re.findall(r"<li>(.*?)</li>", m.group(1), re.S | re.I)]233        # revêtements de sol (« Revêtements : Bois flottant, céramique… »)234        m = re.search(r"Rev[êe]tements?\s*:\s*([^<]{3,120})", detail, re.I)235        if m:236            amen.append("Revêtements : " + _strip_tags(m.group(1)))237        out["amenities"] = [a for a in dict.fromkeys(amen)238                            if len(a) >= 3 and a.lower() != "superficie"][:25]239240        # description : présentation du complexe + immeuble + services optionnels241        desc_bits = []242        m = re.search(r'<div id="DescriptionComplexe">(.*?)</div>', detail,243                      re.S | re.I)244        if m and _strip_tags(m.group(1)):245            desc_bits.append(_strip_tags(m.group(1)))246        if immeuble_txt:247            desc_bits.append("Caractéristiques de l'immeuble : " + immeuble_txt)248        m = re.search(r"Services optionnels(.*?)Les informations", detail,249                      re.S | re.I)250        if m and _strip_tags(m.group(1)):251            desc_bits.append("Services optionnels : " + _strip_tags(m.group(1)))252        out["description"] = " — ".join(desc_bits)[:900]253        return out254255    def _apply_detail(self, lst: Listing, payload: dict) -> None:256        if not payload:257            return258        if payload.get("address"):259            lst.address = payload["address"]260            lst.title = f"{payload['address']}{lst.unit_type}".strip(" —")261        if payload.get("lat") is not None and payload.get("lng") is not None:262            lst.lat, lst.lng = payload["lat"], payload["lng"]263        if payload.get("area_sqft"):264            lst.area_sqft = payload["area_sqft"]265        if payload.get("phone"):266            lst.details.setdefault("contact", {})["phone"] = payload["phone"]267        for a in payload.get("amenities", []):268            if a not in lst.amenities:269                lst.amenities.append(a)270        if payload.get("images"):271            lst.images = payload["images"]272        if payload.get("description"):273            lst.description = payload["description"]274