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/barnes_quebec.py : BARNES Québec (barnes-quebec.com)5#   Site WordPress indexé dans Algolia. La config expose l'App ID et une clé6#   API dans le HTML ; l'index « quebec_all » contient les propriétés (type7#   "property") avec adresse, prix, MLS, chambres, salles de bains, superficie8#   et géolocalisation. Les enregistrements sont dupliqués par langue → on9#   dédoublonne par référence MLS. Les permaliens pointent vers le domaine de10#   staging Kinsta ; on les réécrit vers le domaine de production.11# -----------------------------------------------------------------------------12from __future__ import annotations1314import os15import re1617from .base import BaseConnector18from . import _detailutil as du19from ..schema import PropertyListing2021DETAIL_LIMIT = int(os.environ.get("IMMOKA_BARNES_DETAIL_LIMIT", "400"))22# Galerie WordPress : .../wp-content/uploads/AAAA/MM/{ref}-{hash}-{L}x{H}.jpg23_IMG_RE = re.compile(24    r'https://barnes-quebec\.com/wp-content/uploads/\d{4}/\d{2}/[^"\'\\ ]+?\.(?:jpg|jpeg|png|webp)',25    re.I)26_SIZE_RE = re.compile(r'-(\d{2,4})x(\d{2,4})(?=\.[a-z]+$)', re.I)2728APP_ID = "HCW55VIQNM"29API_KEY = "f4a20779fb6ded84a9c96a9b5976328b"30INDEX = "quebec_all"31QUERY_URL = f"https://{APP_ID}-dsn.algolia.net/1/indexes/{INDEX}/query"32SITE = "https://barnes-quebec.com"33STAGING = "stg-quebec-staging.kinsta.cloud"34HITS_PER_PAGE = 10035MAX_PAGES = 40363738class BarnesQuebecConnector(BaseConnector):39    source_id = "barnes_quebec"40    request_delay = 0.2541    use_detail_cache = False4243    def fetch(self) -> list[PropertyListing]:44        best: dict[str, tuple[int, PropertyListing]] = {}45        page = 046        while page < MAX_PAGES:47            data = self._query(page)48            hits = data.get("hits", [])49            if not hits:50                break51            for h in hits:52                if h.get("type") != "property":53                    continue54                lst = self._to_listing(h)55                if lst is None:56                    continue57                # une propriété apparaît en plusieurs langues et parfois en58                # doublon « Contact us » (sans prix). Clé stable = adresse59                # normalisée ; on garde la meilleure fiche (prix connu + FR).60                key = " ".join(lst.title.split()).lower()61                score = (2 if lst.price else 0) + (1 if h.get("lang_fr") == 1 else 0)62                if key not in best or score > best[key][0]:63                    best[key] = (score, lst)64            if page + 1 >= data.get("nbPages", 0):65                break66            page += 167        listings = [lst for _, lst in best.values()]68        # Algolia ne contient pas les photos : on les récupère sur la fiche.69        du.enrich(self, listings, DETAIL_LIMIT, parse_barnes_detail, key="v2")70        return listings7172    def _query(self, page: int) -> dict:73        resp = self.post(74            QUERY_URL,75            headers={"X-Algolia-API-Key": API_KEY,76                     "X-Algolia-Application-Id": APP_ID,77                     "Content-Type": "application/json"},78            json={"params": f"hitsPerPage={HITS_PER_PAGE}&page={page}"},79        )80        return resp.json()8182    def _to_listing(self, h: dict) -> PropertyListing | None:83        object_id = str(h.get("objectID") or "")84        mls = str(h.get("property_mls_reference") or "").strip()85        if not object_id and not mls:86            return None8788        permalink = (h.get("permalink") or "").replace(STAGING, "barnes-quebec.com")89        if permalink.startswith("http://"):90            permalink = "https://" + permalink[len("http://"):]9192        price = h.get("property_price") or 093        try:94            price = float(price)95        except (TypeError, ValueError):96            price = 0.097        price_label = h.get("property_pretty_price") or ""9899        area = _pos(h.get("property_area"))100        land = _pos(h.get("property_land_area"))101        lat = h.get("property_address_latitude") or None102        lng = h.get("property_address_longitude") or None103        try:104            lat = float(lat) if lat else None105            lng = float(lng) if lng else None106        except (TypeError, ValueError):107            lat = lng = None108        if lat == 0 or lng == 0:109            lat = lng = None110111        return PropertyListing(112            source=self.source_id,113            external_id=mls or object_id,114            url=permalink or SITE + "/property/",115            title=(h.get("title") or "").strip(),116            address=(h.get("title") or "").strip(),117            city=(h.get("property_address_city") or "").strip(),118            region=(h.get("region") or "").strip(),119            property_type=(h.get("property_type") or "").strip(),120            price=price if price > 0 else None,121            price_label=price_label,122            bedrooms=_pos(h.get("property_bedrooms_integer")),123            bathrooms=_pos(h.get("property_bathrooms")),124            area_sqft=area,125            lot_sqft=land,126            mls=mls,127            description=(h.get("content") or "")[:4000],128            lat=lat,129            lng=lng,130            broker_name="BARNES Québec",131        )132133134def _pos(v):135    try:136        n = float(v)137        return int(n) if n and n > 0 else None138    except (TypeError, ValueError):139        return None140141142def parse_barnes_detail(html: str) -> dict:143    """Galerie photo pleine résolution (absente d'Algolia) + description/pièces."""144    out: dict = {}145    # regroupe par image de base (sans le suffixe -LxH), garde la plus grande146    best: dict[str, tuple[int, str]] = {}147    for u in _IMG_RE.findall(html):148        m = _SIZE_RE.search(u)149        area = int(m.group(1)) * int(m.group(2)) if m else 10 ** 8   # sans suffixe = original150        base = _SIZE_RE.sub("", u)151        if base not in best or area > best[base][0]:152            best[base] = (area, u)153    imgs = [u for _, u in best.values()]154    # ignore les vignettes de courtiers/logos (gardent souvent une réf MLS chiffrée)155    imgs = [u for u in imgs if re.search(r"/\d{6,}", u)] or imgs156    if imgs:157        out["images"] = imgs[:60]158    desc = du.ld_description(html)159    if desc:160        out["description"] = desc161    _det = du.centris_details(du.flatten(html))162    if _det:163        out.setdefault("details", {}).update(_det)164    return out165