SPB Git

spb/lou-ka Public

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

HTML 99.7%
10.6 KB · 251 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/quartier_sila.py : connecteur Quartier Sila (silalevis.ca)5#   Complexe de condos locatifs du Groupe Damco / Développement Beaubourg à6#   Saint-Romuald (Lévis) — 4 phases (Guillaume-Couture, J.-B.-Demers,7#   d'Anticosti), 3½ à 5½. Site WordPress/Elementor derrière Cloudflare ; la8#   page « Condos locatifs Sila » embarque le module de plans interactifs9#   Livya (app.livya.com, client « damco », projet « sila »). La page Next.js10#   du module est rendue côté serveur : son flux RSC (self.__next_f.push)11#   contient l'inventaire JSON complet — numéro d'unité, immeuble/étage,12#   statut, loyer, pièces, superficie, adresse civique, GPS et plans. Les13#   unités de la phase 4 (prélocation) n'ont pas encore de prix publié.14#   2 requêtes par sync : la page WordPress (id d'entité Livya + services et15#   commodités + bannières de phases) + la page du module.16# -----------------------------------------------------------------------------17from __future__ import annotations1819import codecs20import json21import re2223from bs4 import BeautifulSoup2425from ..schema import Listing, infer_city26from .base import BaseConnector2728BASE = "https://silalevis.ca"29LIST_URL = f"{BASE}/condos-locatifs-sila/"30LIVYA = "https://app.livya.com"31SECTOR = "Saint-Romuald"3233# fragments RSC de Next.js : self.__next_f.push([1,"...payload échappé..."])34_NEXT_F_RE = re.compile(r'self\.__next_f\.push\(\[1,"((?:[^"\\]|\\.)*)"\]\)')35# bannières de la page : « Phase 3 en construction pour livraison à l'été36# 2026! », « Phase 4 en prélocation à l'automne 2026! »37_PHASE_BANNER_RE = re.compile(38    r"Phase\s*(\d)\s*(en\s+(?:construction|pr[ée]location)[^!.<]{0,80})", re.I)39# item de commodité réservé à certaines phases : « (Phases 2 et 3) »40_AMEN_PHASE_RE = re.compile(r"phases?\s*([\d\s,et]+)", re.I)414243def _flight_blob(html: str) -> str:44    """Concatène et désérialise les fragments RSC d'une page Livya.4546    `unicode_escape` interprète les octets en latin-1 : on ré-encode pour47    retrouver l'UTF-8 d'origine (sinon « Étage » devient « Ãtage »).48    """49    blob = "".join(codecs.decode(c, "unicode_escape")50                   for c in _NEXT_F_RE.findall(html))51    return blob.encode("latin-1", "ignore").decode("utf-8", "ignore")525354def _json_arrays(blob: str, key: str) -> list[list]:55    """Toutes les valeurs de `"key":[...]` du flux (équilibrage de crochets)."""56    out: list[list] = []57    for m in re.finditer(re.escape(f'"{key}":['), blob):58        j = m.end() - 159        depth, in_str, esc = 0, False, False60        for k in range(j, len(blob)):61            c = blob[k]62            if in_str:63                if esc:64                    esc = False65                elif c == "\\":66                    esc = True67                elif c == '"':68                    in_str = False69            elif c == '"':70                in_str = True71            elif c == "[":72                depth += 173            elif c == "]":74                depth -= 175                if depth == 0:76                    try:77                        out.append(json.loads(blob[j:k + 1]))78                    except ValueError:79                        pass80                    break81    return out828384def _unit_type(rooms: float | None) -> str:85    """3.5 -> « 3½ » ; 0.5 -> « Studio » (finalize gère 6½+)."""86    if not rooms:87        return ""88    if rooms < 1:89        return "Studio"90    return f"{int(rooms)}½"919293class QuartierSilaConnector(BaseConnector):94    source_id = "quartier_sila"95    request_delay = 0.89697    # -- page WordPress : commodités listées sous « Services et commodités » ----98    @staticmethod99    def _amenities(soup: BeautifulSoup) -> list[str]:100        """Items `<li>` entre le titre « Services et commodités » et le titre101        suivant (le pied de page et le bandeau de cookies ont aussi des puces)."""102        items: list[str] = []103        collecting = False104        for el in soup.find_all(["h1", "h2", "h3", "h4", "li"]):105            text = re.sub(r"\s+", " ", el.get_text(" ", strip=True))106            if el.name != "li":107                if collecting:108                    break109                collecting = "services et commodit" in text.lower()110                continue111            if collecting and 3 <= len(text) <= 120:112                items.append(text)113        return list(dict.fromkeys(items))[:20]114115    @staticmethod116    def _amenities_for_phase(amenities: list[str], phase: str) -> list[str]:117        """Écarte les items réservés à d'autres phases (« (Phases 2 et 3) »)."""118        out = []119        for item in amenities:120            m = _AMEN_PHASE_RE.search(item)121            if m and phase and phase not in re.findall(r"\d", m.group(1)):122                continue123            out.append(item)124        return out125126    def fetch(self) -> list[Listing]:127        # 1) Page « Condos locatifs Sila » : id d'entité Livya + commodités +128        #    bannières d'état des phases + téléphone du bureau de location129        wp = self.get(LIST_URL).text130        m = re.search(r"<[^>]*livya-module-container-plans[^>]*>", wp)131        if not m:132            raise RuntimeError("module Livya introuvable sur la page condos")133        tag = m.group(0)134        client_m = re.search(r'data-client="([^"]+)"', wp)135        project = re.search(r'data-project="([^"]+)"', tag)136        entity = re.search(r'data-entity="([^"]+)"', tag)137        if not (client_m and project and entity):138            raise RuntimeError("attributs data-client/project/entity manquants")139140        soup = BeautifulSoup(wp, "html.parser")141        amenities = self._amenities(soup)142        contact: dict = {}143        tel = re.search(r'href="tel:([\d\s\-.]{10,14})"', wp)144        if tel:145            d = re.sub(r"\D", "", tel.group(1))[-10:]146            if len(d) == 10:147                contact["phone"] = f"{d[:3]}-{d[3:6]}-{d[6:]}"148        # état publié par phase (construction/prélocation + horizon de livraison)149        banners = {n: f"Phase {n} {re.sub(r'[  ]+', ' ', rest).strip()}"150                   for n, rest in _PHASE_BANNER_RE.findall(151                       soup.get_text(" ", strip=True))}152153        # 2) Page du module Livya (rendue serveur) -> inventaire JSON complet154        livya_url = (f"{LIVYA}/fr/{client_m.group(1)}/projects/"155                     f"{project.group(1)}/plans/{entity.group(1)}?noLayout=1")156        blob = _flight_blob(self.get(livya_url).text)157158        phase_names: dict[str, str] = {}159        for arr in _json_arrays(blob, "phases"):160            for p in arr:161                if isinstance(p, dict) and p.get("phaseId") and p.get("name"):162                    phase_names[p["phaseId"]] = str(p["name"])163164        units, seen = [], set()165        for arr in _json_arrays(blob, "units"):166            for u in arr:167                if isinstance(u, dict) and u.get("unitId") and u["unitId"] not in seen:168                    seen.add(u["unitId"])169                    units.append(u)170171        listings: list[Listing] = []172        for u in units:173            if u.get("availability") != "AVAILABLE" or not u.get("rental", True):174                continue175            num = str(u.get("number") or "").strip()176            price = u.get("rentalPrice")177            price = float(price) if isinstance(price, (int, float)) and price > 0 else None178            area = u.get("unitSize")179            area = float(area) if isinstance(area, (int, float)) and area > 0 else None180            phase = phase_names.get(u.get("phaseId") or "", "")181182            # immeuble + étage depuis « Sila 3 - Étage 6 »183            floor_disp = str(u.get("floorDisplayName") or "").strip()184            building, _, floor_label = (x.strip() for x in185                                        floor_disp.partition(" - "))186187            desc: list[str] = []188            if floor_label:189                desc.append(floor_label.capitalize())190            elif u.get("floorNumber"):191                desc.append(f"Étage {u['floorNumber']}")192            if phase:193                desc.append(f"Phase {phase}")194            if u.get("typeName"):195                desc.append(f"Modèle {u['typeName']}")196            if u.get("roomsBed"):197                desc.append(f"{u['roomsBed']} chambre(s)")198            if u.get("roomsBath"):199                desc.append(f"{u['roomsBath']} salle(s) de bain")200            if u.get("balconySize"):201                desc.append(f"Balcon de {u['balconySize']} pi²")202            if u.get("floorPlanUrl"):203                desc.append(f"Plan : {u['floorPlanUrl']}")204205            address = ", ".join(x for x in (206                u.get("address") or "", u.get("city") or "",207                u.get("postalCode") or "") if x)208209            images = [img.get("fullUrl") for img in (u.get("typeImages") or [])210                      if isinstance(img, dict) and img.get("fullUrl")]211            if u.get("floorPlanImageUrl"):212                images.append(u["floorPlanImageUrl"])213214            # disponibilité : date future publiée > bannière de phase215            # (« Phase 4 en prélocation à l'automne 2026 ») > statut du plan216            future = u.get("futureAvailability")217            availability = (str(future) if future218                            else banners.get(phase, "Disponible"))219220            details: dict = {}221            if contact:222                details["contact"] = dict(contact)223            floor_no = str(u.get("floorNumber") or "")224            if floor_no.isdigit():225                details["floor"] = int(floor_no)226227            titre = (f"Quartier Sila — {building}, unité {num}" if building228                     else f"Quartier Sila — unité {num}")229            lat, lng = u.get("latitude"), u.get("longitude")230            listings.append(Listing(231                source=self.source_id,232                external_id=str(u["unitId"]),233                url=LIST_URL,234                title=f"{titre} ({_unit_type(u.get('rooms'))})",235                address=address,236                sector=SECTOR,237                city=infer_city(SECTOR),238                unit_type=_unit_type(u.get("rooms")),239                price=price,240                price_label=f"{price:.0f} $ /mois" if price else "",241                availability=availability,242                area_sqft=area,243                description=" | ".join(desc),244                amenities=self._amenities_for_phase(amenities, phase),245                details=details,246                images=images[:12],247                lat=float(lat) if lat else None,248                lng=float(lng) if lng else None,249            ))250        return listings251