spb/immo-ka Public

1# -----------------------------------------------------------------------------2# Immo-Ka — Agrégateur de maisons à vendre (province de Québec)3# Auteur : Simon-Pierre Boucher — contact@spboucher.ai4# connectors/engel_volkers.py : Engel & Völkers Montréal / Québec (luxe)5#   Le site local immobilier.evquebec.com publie un property-sitemap.xml6#   exhaustif (~1240 fiches, /propriete/{no-centris}/). Chaque fiche a un7#   JSON-LD RealEstateListing (prix, adresse, description, n° Centris). La8#   galerie est servie séquentiellement sur media.evquebec.com/listings/{id}/NN.webp.9# -----------------------------------------------------------------------------10from __future__ import annotations1112import html as _html13import os14import re1516from .base import BaseConnector17from . import _detailutil as du18from ..schema import PropertyListing1920SITEMAP = "https://immobilier.evquebec.com/property-sitemap.xml"21MEDIA = "https://media.evquebec.com/listings"22DETAIL_LIMIT = int(os.environ.get("IMMOKA_EV_DETAIL_LIMIT", "400"))23GALLERY_MAX = 602425_PROP_RE = re.compile(r'https://immobilier\.evquebec\.com/propriete/(\d{5,})/')26_NAME_RE = re.compile(r'^(.*?)\s+à vendre\b\s*-?\s*(.*)$', re.I)272829class EngelVolkersConnector(BaseConnector):30    source_id = "engel_volkers"31    request_delay = 0.432    use_detail_cache = True3334    def fetch(self) -> list[PropertyListing]:35        try:36            xml = self.get(SITEMAP).text37        except Exception:38            return []39        ids, seen = [], set()40        for m in _PROP_RE.finditer(xml):     # ne garder que les URLs FR41            if m.group(1) not in seen:42                seen.add(m.group(1))43                ids.append(m.group(1))44        listings = [PropertyListing(source=self.source_id, external_id=cid,45                                    url=f"https://immobilier.evquebec.com/propriete/{cid}/",46                                    mls=cid, broker_name="Engel & Völkers Québec")47                    for cid in ids]48        self._enrich(listings)49        return listings5051    def _enrich(self, listings: list[PropertyListing]) -> None:52        from .. import db53        con = db.connect()54        budget = DETAIL_LIMIT55        try:56            for lst in listings:57                d = db.get_cached_detail(con, self.source_id, lst.external_id, "v1")58                if d is None:59                    if budget <= 0:60                        continue61                    try:62                        d = self._parse(self.get(lst.url).text, lst.external_id)63                    except Exception:64                        d = {}65                    db.put_cached_detail(con, self.source_id, lst.external_id, "v1", d)66                    budget -= 167                _apply(lst, d)68        finally:69            con.close()7071    def _parse(self, html: str, centris: str) -> dict:72        out: dict = {}73        for n in du.ld_nodes(html):74            t = n.get("@type")75            types = t if isinstance(t, list) else [t]76            if "RealEstateListing" in types:77                if n.get("description"):78                    out["description"] = _html.unescape(str(n["description"])).strip()79                offer = n.get("offers") or {}80                try:81                    p = float(offer.get("price"))82                    if p >= 10_000:83                        out["price"] = p84                except (TypeError, ValueError):85                    pass86                name = _html.unescape(n.get("name", "")).strip()87                if name:88                    out.update(_parse_name(name))89                break90        # bedrooms / bathrooms depuis le texte91        text = re.sub(r"\s+", " ", _html.unescape(re.sub(r"<[^>]+>", " ", html)))92        mb = re.search(r"(\d+)\s*chambre", text, re.I)93        if mb:94            out["bedrooms"] = int(mb.group(1))95        ms = re.search(r"(\d+)\s*salle[s]?\s*de\s*bain", text, re.I)96        if ms:97            out["bathrooms"] = int(ms.group(1))98        out["images"] = self._gallery(centris)99        return out100101    def _gallery(self, centris: str) -> list[str]:102        """Galerie séquentielle NN.webp — on s'arrête au premier trou (HEAD rapide)."""103        urls = []104        for i in range(1, GALLERY_MAX + 1):105            u = f"{MEDIA}/{centris}/{i:02d}.webp"106            try:107                r = self.session.head(u, timeout=8)108            except Exception:109                break110            if r.status_code != 200:111                break112            urls.append(u)113        return urls114115116def _parse_name(name: str) -> dict:117    """« Condo à vendre - Centre-ville - 1288 Av. …, Montréal (Ville-Marie), Montréal »."""118    out: dict = {}119    m = _NAME_RE.search(name)120    if not m:121        return out122    out["property_type"] = m.group(1).strip()123    rest = m.group(2).strip()124    segs = [s.strip() for s in rest.split(" - ") if s.strip()]125    loc = segs[-1] if segs else rest126    # municipalité = segment contenant « ( » ; adresse = ce qui précède127    commas = [c.strip() for c in loc.split(",")]128    muni_idx = next((i for i, c in enumerate(commas) if "(" in c), None)129    if muni_idx is not None:130        muni = commas[muni_idx]131        ps = re.search(r"\(([^)]+)\)", muni)132        out["city"] = re.sub(r"\s*\([^)]*\)", "", muni).strip()133        if ps:134            out["sector"] = ps.group(1).strip()135        addr = ", ".join(commas[:muni_idx]).strip()136        if addr:137            out["address"] = addr138    elif len(commas) >= 2:139        # pas d'arrondissement : dernier segment = municipalité, le reste = adresse140        out["city"] = commas[-1]141        out["address"] = ", ".join(commas[:-1]).strip()142    else:143        out["address"] = loc144    if len(segs) > 1 and not out.get("sector"):145        out["sector"] = segs[0]146    return out147148149def _apply(lst: PropertyListing, d: dict) -> None:150    if not d:151        return152    for f in ("address", "city", "sector", "property_type", "description"):153        if d.get(f) and not getattr(lst, f, ""):154            setattr(lst, f, d[f])155    for f in ("price", "bedrooms", "bathrooms"):156        if d.get(f) is not None and getattr(lst, f, None) in (None, 0):157            setattr(lst, f, d[f])158    if d.get("images") and len(d["images"]) > len(lst.images):159        lst.images = d["images"]160