SPB Git

spb/lou-ka Public

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

HTML 99.7%
8.7 KB · 226 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/sentinelle.py : connecteur La Sentinelle (lasentinellelevis.com)5#   Immeuble de 144 condos locatifs au 7002, boul. Guillaume-Couture6#   (Vieux-Lévis). La page /projet/ contient un TABLEAU structuré des unités7#   (numéro, étage, superficie pi², type, prix, statut, lien fiche) : on ne8#   visite que les fiches des unités « Disponible », via self.detail(...)9#   (cache BD, clé = ligne du tableau) — la fiche apporte la date de10#   disponibilité, la liste des commodités (ul.uk-list-disc), les photos11#   et le plan.12# -----------------------------------------------------------------------------13from __future__ import annotations1415import hashlib16import re1718from bs4 import BeautifulSoup1920from ..schema import Listing, infer_city, normalize_unit_type, parse_price21from .base import BaseConnector2223BASE = "https://lasentinellelevis.com"24LIST_URL = f"{BASE}/projet/"25SECTOR = "Vieux-Lévis"26ADDRESS = "7002, boul. Guillaume-Couture, Lévis, G6V 0C1"27PHONE = "418-741-3737"          # lien tel: du pied de page (bureau de location)2829_SKIP_IMG = re.compile(r"logo|favicon|icon|brochu|promenade", re.I)30_STATUS_RE = re.compile(r"^(Disponible|Loué|Réservé)$", re.I)313233class _BudgetReached(Exception):34    """Plafond de requêtes « fiche » atteint pour cette synchronisation."""353637class SentinelleConnector(BaseConnector):38    source_id = "sentinelle"39    request_delay = 0.640    max_details = 60          # plafond de vraies requêtes fiche par sync4142    def fetch(self) -> list[Listing]:43        html = self.get(LIST_URL).text44        soup = BeautifulSoup(html, "html.parser")45        self._fetches = 04647        listings: list[Listing] = []48        seen: set[str] = set()4950        # Tableau des unités : <tr data-url=".../unite-NNN/"><td>104</td>51        # <td>Étage 1</td><td>1241</td><td>4½+</td><td>1695$</td>52        # <td>Disponible</td>...53        rows = soup.select("tr[data-url*='/unite-']")54        for tr in rows:55            try:56                lst = self._row_listing(tr)57            except Exception:58                continue59            if lst and lst.external_id not in seen:60                seen.add(lst.external_id)61                listings.append(lst)62        if rows:63            return listings6465        # Repli (si le tableau disparaît) : balayer les liens de fiches66        unit_urls = sorted(set(re.findall(67            rf'href="({re.escape(BASE)}/projet/etage-\d+/unite-\d+/)"', html)))68        for url in unit_urls[: self.max_details]:69            unit_no = url.rstrip("/").split("-")[-1]70            if f"unite-{unit_no}" in seen:71                continue72            try:73                payload = self._fetch_detail(url)74            except Exception:75                continue76            if payload.get("status") not in (None, "Disponible"):77                continue78            lst = self._build(unit_no, url, payload)79            seen.add(lst.external_id)80            listings.append(lst)81        return listings8283    # -- ligne du tableau -------------------------------------------------------84    def _row_listing(self, tr) -> Listing | None:85        url = (tr.get("data-url") or "").split("?")[0]86        m = re.search(r"/unite-(\d+)/?$", url)87        if not m:88            return None89        unit_no = m.group(1)90        cells = [td.get_text(" ", strip=True) for td in tr.find_all("td")]9192        row: dict = {}93        for c in cells:94            if _STATUS_RE.match(c):95                row["status"] = c.capitalize()96            elif re.match(r"^\d[\d\s]*\$$", c):97                row["price_label"] = c98            elif re.match(r"^Étage\s+\d+$", c):99                row["floor"] = int(re.search(r"\d+", c).group(0))100            elif re.match(r"^\d\s*(?:½|1/2)\s*\+?$", c):101                row["type"] = c102            elif re.match(r"^\d{3,5}$", c) and c != unit_no:103                v = float(c)104                if 80 <= v <= 20000:105                    row["sqft"] = v106107        if row.get("status") and row["status"] != "Disponible":108            return None    # loué / réservé : pas une annonce active109110        # Fiche de l'unité (cache BD : re-téléchargée si la ligne change)111        key = hashlib.sha1(("|".join(cells)).encode("utf-8")).hexdigest()112        try:113            payload = self.detail(f"unite-{unit_no}", key,114                                  lambda u=url: self._fetch_detail(u))115        except Exception:      # _BudgetReached inclus : rien de caché116            payload = {}117        return self._build(unit_no, url, payload, row)118119    # -- fiche --------------------------------------------------------------------120    def _fetch_detail(self, url: str) -> dict:121        """Télécharge une fiche d'unité (appelé seulement hors cache)."""122        if self._fetches >= self.max_details:123            raise _BudgetReached()124        self._fetches += 1125        dhtml = self.get(url).text126        soup = BeautifulSoup(dhtml, "html.parser")127        text = soup.get_text("\n", strip=True)128        payload: dict = {}129130        m = re.search(r"Unité\s+(\d+)\s*-\s*(\d\s*(?:½|1/2)\s*\+?)", text)131        if m:132            payload["type"] = m.group(2).strip()133        fm = re.search(r"Étage\s+(\d+)", text)134        if fm:135            payload["floor"] = int(fm.group(1))136        sm = re.search(r"\n(Disponible|Loué|Réservé)\n", text)137        if sm:138            payload["status"] = sm.group(1)139140        am = re.search(r"Disponible à partir de\s*:\s*([^\n]+)", text)141        if am:142            payload["availability"] = \143                f"Disponible à partir de : {am.group(1).strip()}"144145        pm = re.search(r"^([\d\s  ]{3,})\$\s*$", text, re.M)146        if pm:147            payload["price_label"] = pm.group(0).strip()148149        sqm = re.search(r"Superficie\s+([\d\s]+)pi", text)150        if sqm:151            try:152                v = float(sqm.group(1).replace(" ", ""))153                if 80 <= v <= 20000:154                    payload["sqft"] = v155            except ValueError:156                pass157158        # commodités : liste à puces de la fiche (ul.uk-list-disc)159        amenities: list[str] = []160        for ul in soup.select("ul.uk-list-disc"):161            for li in ul.find_all("li"):162                t = li.get_text(" ", strip=True)163                if t and not t.lower().startswith("superficie") \164                        and t not in amenities:165                    amenities.append(t)166        payload["amenities"] = amenities[:15]167168        if "espace bureau" in text.lower():169            payload["office"] = True170171        imgs = re.findall(172            rf'(?:src|href)="({re.escape(BASE)}/wp-content/uploads/'173            rf'[^"]+\.(?:jpg|jpeg|png|webp))"', dhtml, re.I)174        payload["images"] = [u for u in dict.fromkeys(imgs)175                             if not _SKIP_IMG.search(u)][:10]176        return payload177178    # -- assemblage -----------------------------------------------------------179    def _build(self, unit_no: str, url: str, payload: dict,180               row: dict | None = None) -> Listing:181        row = row or {}182        raw_type = payload.get("type") or row.get("type") or ""183        unit_type = normalize_unit_type(raw_type)184185        price_label = row.get("price_label") \186            or payload.get("price_label") or ""187        price = parse_price(price_label)188189        availability = payload.get("availability") \190            or (row.get("status") or "Disponible")191192        area = row.get("sqft") or payload.get("sqft")193        floor = row.get("floor") or payload.get("floor")194195        desc_parts = []196        if floor:197            desc_parts.append(f"Étage {floor}")198        if area:199            desc_parts.append(f"Superficie {area:.0f} pi²")200        if "+" in raw_type or payload.get("office"):201            desc_parts.append("Avec espace bureau")202203        details: dict = {"contact": {"phone": PHONE}}204        if floor and 0 < int(floor) <= 60:205            details["floor"] = int(floor)206207        return Listing(208            source=self.source_id,209            external_id=f"unite-{unit_no}",210            url=url,211            title=f"La Sentinelle — Unité {unit_no}"212                  f"{' (' + unit_type + ')' if unit_type else ''}",213            address=ADDRESS,214            sector=SECTOR,215            city=infer_city(SECTOR, default="Lévis"),216            unit_type=unit_type,217            price=price,218            price_label=price_label,219            availability=availability,220            area_sqft=float(area) if area else None,221            description=" | ".join(desc_parts),222            amenities=list(payload.get("amenities") or []),223            details=details,224            images=list(payload.get("images") or []),225        )226