spb/lou-ka Public

1# -----------------------------------------------------------------------------2# Lou-Ka — Agrégateur de logements à louer (province de Québec)3# Auteur : Simon-Pierre Boucher — contact@spboucher.ai4# connectors/bribourg.py : connecteur Bri Bourg (bribourg.com)5#   Site PHP simple (Charlesbourg, Vanier, Beauport). Les annonces de6#   logements à louer sont publiées via un blogue DropInBlog embarqué dans7#   Logements-a-louer.php ; on lit le flux RSS DropInBlog, puis on complète8#   avec la page d'immeuble correspondante (Immeubles.php) : photos, code9#   postal, descriptif de l'immeuble et services à proximité (via le cache10#   self.detail()). Contact (tel:/mailto:) pris sur la page principale.11# -----------------------------------------------------------------------------12from __future__ import annotations1314import hashlib15import html as htmllib16import re17import unicodedata18from urllib.parse import unquote, urlparse, parse_qs1920from ..schema import Listing, infer_city, normalize_unit_type, parse_price21from .base import BaseConnector2223BASE = "https://bribourg.com"24RENT_PAGE = f"{BASE}/Logements-a-louer.php"25# identifiant du blogue DropInBlog (repli si non détecté dans la page)26DEFAULT_BLOG_ID = "dcc6791b-355d-4d9b-bb91-a30d855f6cdf"27FEED_URL = "https://io.dropinblog.com/feed/{blog_id}/?limit=100"282930def _strip_tags(s: str) -> str:31    return re.sub(r"\s+", " ", htmllib.unescape(re.sub(r"<[^>]+>", " ", s))).strip()323334def _norm(s: str) -> str:35    s = unicodedata.normalize("NFD", s.lower())36    return "".join(c for c in s if unicodedata.category(c) != "Mn")373839class BribourgConnector(BaseConnector):40    source_id = "bribourg"41    request_delay = 0.64243    def fetch(self) -> list[Listing]:44        # 1) Détecter l'identifiant du blogue DropInBlog dans la page45        #    (+ contact du gestionnaire : liens tel:/mailto: structurés)46        blog_id = DEFAULT_BLOG_ID47        self._contact: dict = {}48        try:49            page = self.get(RENT_PAGE).text50            m = re.search(r"embedjs/([0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-"51                          r"[0-9a-f]{4}-[0-9a-f]{12})", page)52            if m:53                blog_id = m.group(1)54            m = re.search(r'href="tel:(\d{10})"', page)55            if m:56                d = m.group(1)57                self._contact["phone"] = f"{d[:3]}-{d[3:6]}-{d[6:]}"58            m = re.search(r'href="mailto:([^"?]+)"', page)59            if m:60                self._contact["email"] = m.group(1)61        except Exception:62            pass6364        # 2) Flux RSS des annonces65        try:66            feed = self.get(FEED_URL.format(blog_id=blog_id)).text67        except Exception:68            return []6970        # 3) Pages d'immeubles (pour compléter les photos)71        building_pages = self._building_pages()7273        listings: list[Listing] = []74        for item in re.findall(r"<item>(.*?)</item>", feed, re.S):75            try:76                lst = self._parse_item(item, building_pages)77                if lst:78                    listings.append(lst)79            except Exception:80                continue81        return listings8283    # -- inventaire des pages d'immeubles ---------------------------------------84    def _building_pages(self) -> list[str]:85        try:86            html = self.get(f"{BASE}/Immeubles.php").text87        except Exception:88            return []89        pages = re.findall(r'href="((?:\./)?\d[\w\-]+\.php)"', html)90        return sorted({p.lstrip("./") for p in pages})9192    # -- une annonce du flux -----------------------------------------------------93    def _parse_item(self, item: str, building_pages: list[str]) -> Listing | None:94        def tag(name: str) -> str:95            m = re.search(r"<%s>(.*?)</%s>" % (name, name), item, re.S)96            return htmllib.unescape(m.group(1).strip()) if m else ""9798        title = _strip_tags(tag("title"))99        link = tag("link") or RENT_PAGE100        m = re.search(r"<content:encoded>\s*<!\[CDATA\[(.*?)\]\]>", item, re.S)101        content = m.group(1) if m else tag("description")102103        # exclusions : stationnement / commercial / rangement104        if re.search(r"stationnement|commercial|rangement|garage|entrep[oô]t",105                     title, re.I):106            return None107108        text = _strip_tags(content)109        # "Adresse : 510, avenue Claude-Martin, Québec (Vanier)"110        addr_m = re.search(r"Adresse\s*:\s*([^A-Z]*?[\w\s,.'’\-()]+?)"111                           r"(?=\s*(?:Loyer|Disponibilit|Caract|$))", text)112        address = addr_m.group(1).strip(" ,") if addr_m else ""113        price_m = re.search(r"Loyer\s*:\s*([\d\s,]+\$[^A-Z]*?)(?=\s*(?:Disponibilit|Caract|$))",114                            text)115        price_label = price_m.group(1).strip() if price_m else ""116        avail_m = re.search(r"Disponibilit[ée]\s*:\s*(.+?)(?=\s*Caract|$)", text)117        availability = avail_m.group(1).strip() if avail_m else ""118119        # secteur : "(Vanier)" dans l'adresse ou le titre120        sector = ""121        m = re.search(r"\(([^)]+)\)", address) or re.search(r"\(([^)]+)\)", title)122        if m:123            sector = m.group(1).strip()124125        # caractéristiques (liste <li>)126        amenities = []127        for li in re.findall(r"<li[^>]*>(.*?)</li>", content, re.S):128            t = _strip_tags(li)129            if t and t not in amenities:130                amenities.append(t)131        unit_type = normalize_unit_type(title) or normalize_unit_type(" ".join(amenities))132133        # images de l'annonce (hébergées chez DropInBlog)134        images = [u for u in dict.fromkeys(135            re.findall(r'src="(https?://[^"]+\.(?:jpg|jpeg|png|webp))"', content, re.I))]136137        # identifiant stable : slug du paramètre ?p= du lien138        qs = parse_qs(urlparse(link).query)139        ext_id = unquote(qs.get("p", [""])[0]) or _norm(re.sub(r"\W+", "-", address or title))140141        # compléter avec la page de l'immeuble correspondant (numéro civique142        # + rue) : photos, code postal, descriptif, services à proximité —143        # via le cache self.detail() (clé = hash du contenu RSS de l'annonce)144        description = text[:600]145        civic_m = re.match(r"(\d+)", address)146        if civic_m and building_pages:147            civic = civic_m.group(1)148            street_words = [w for w in re.findall(r"[a-z]{4,}", _norm(address))149                            if w not in ("avenue", "boulevard", "quebec")]150            for pg in building_pages:151                pg_n = _norm(pg)152                if re.match(r"%s\D" % re.escape(civic), pg) and \153                        any(w in pg_n for w in street_words):154                    key = hashlib.sha1(item.encode("utf-8")).hexdigest()155                    payload = self.detail(156                        ext_id, key, lambda p=pg: self._building_payload(p))157                    for full in payload.get("images") or []:158                        if full not in images:159                            images.append(full)160                    postal = payload.get("postal") or ""161                    if postal and postal not in address:162                        address = f"{address}, {postal}"163                    bits = [text]164                    if payload.get("building_desc"):165                        bits.append("Immeuble : " + payload["building_desc"])166                    if payload.get("services"):167                        bits.append("Services à proximité : " +168                                    ", ".join(payload["services"]))169                    description = " — ".join(bits)[:600]170                    break171172        city = infer_city(sector, default="Québec")173        return Listing(174            source=self.source_id,175            external_id=ext_id,176            url=link,177            title=title,178            address=address,179            sector=sector,180            city=city,181            unit_type=unit_type,182            price=parse_price(price_label),183            price_label=price_label,184            availability=availability,185            description=description,186            amenities=amenities,187            details={"contact": dict(self._contact)} if self._contact else {},188            images=images[:25],189        )190191    # -- page d'immeuble (photos, code postal, descriptif, services) -----------192    def _building_payload(self, page: str) -> dict:193        try:194            bhtml = self.get(f"{BASE}/{page}").text195        except Exception:196            return {}197        out: dict = {}198        images: list[str] = []199        extra = re.findall(200            r'(?:src|href)="((?:\./)?images/[\w\-]+\.(?:jpg|jpeg|png|webp))"',201            bhtml, re.I)202        for u in dict.fromkeys(extra):203            if re.search(r"logo|favicon|apropos|ico-", u, re.I):204                continue205            images.append(f"{BASE}/{u.lstrip('./')}")206        if images:207            out["images"] = images[:25]208209        btext = _strip_tags(bhtml)210        m = re.search(r"Code postal\s*:\s*([A-Z]\d[A-Z]\s?\d[A-Z]\d)", btext)211        if m:212            out["postal"] = m.group(1)213        # descriptif : premier paragraphe substantiel (« Immeuble situé … »)214        for p in re.findall(r"<p[^>]*>(.*?)</p>", bhtml, re.S):215            t = _strip_tags(p)216            if len(t) > 60 and not re.search(r"RSS|FeedBurner", t):217                out["building_desc"] = t[:300]218                break219        # « SERVICES À PROXIMITÉ » (liste à puces)220        m = re.search(r"PROXIMIT.*?<ul[^>]*>(.*?)</ul>", bhtml, re.S)221        if m:222            services = [_strip_tags(li)223                        for li in re.findall(r"<li[^>]*>(.*?)</li>",224                                             m.group(1), re.S)]225            services = [s for s in services if s]226            if services:227                out["services"] = services[:12]228        return out229