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/sourceimmo.py : base réutilisable pour les agences bâties sur la5#   plateforme « source.immo » (plugin WordPress source-immo). L'app AngularJS6#   appelle une API ASP.NET publique :7#     GET https://api-v1.source.immo/api/listing/view/{view}/{locale}/items8#   avec les en-têtes x-si-account / x-si-api / x-si-appId (exposés dans la9#   config publique {_configs.json} du site). Pagination par `next_token`.10#   Une sous-classe fournit account_id / api_key / app_id / view_id.11# -----------------------------------------------------------------------------12from __future__ import annotations1314from .base import BaseConnector15from ..schema import PropertyListing1617API_ROOT = "https://api-v1.source.immo/api"1819# Codes de catégorie source.immo -> vocabulaire brut (finalize() normalise)20_CATEGORY = {21    "RESIDENTIAL": "Maison", "SINGLE-FAMILY": "Maison",22    "MULTI-FAMILY": "Multiplex", "CONDO": "Condo", "APARTMENT": "Condo",23    "LOT": "Terrain", "LAND": "Terrain", "FARM": "Fermette/Agricole",24    "COMMERCIAL": "Commercial", "INDUSTRIAL": "Commercial",25}26_SUBCATEGORY = {27    "2X": "Duplex", "3X": "Triplex", "4X": "Quadruplex", "5X": "Quintuplex",28}293031class SourceImmoConnector(BaseConnector):32    """Base : définir source_id, account_id, api_key, app_id, view_id, site_url."""3334    account_id = ""35    api_key = ""36    app_id = ""37    view_id = ""38    site_url = ""39    locale = "fr"40    request_delay = 0.2541    use_detail_cache = False42    max_pages = 400          # garde-fou (100 items/page)4344    def _headers(self) -> dict:45        return {46            "x-si-account": self.account_id,47            "x-si-api": self.api_key,48            "x-si-appId": self.app_id,49            "Accept": "application/json",50        }5152    def fetch(self) -> list[PropertyListing]:53        url = f"{API_ROOT}/listing/view/{self.view_id}/{self.locale}/items"54        out: list[PropertyListing] = []55        params: dict = {}56        seen: set[str] = set()57        prev_token = None58        for _ in range(self.max_pages):59            try:60                data = self.get(url, headers=self._headers(), params=params).json()61            except Exception:62                break63            items = data.get("items", []) if isinstance(data, dict) else []64            if not items:65                break66            before = len(out)67            for it in items:68                lst = self._to_listing(it)69                if lst and lst.uid not in seen:70                    seen.add(lst.uid)71                    out.append(lst)72            # Pagination source.immo : l'API attend `st` (search_token) + `nt`73            # (next_token) — c'est ce que fait l'app AngularJS. Avec le mauvais74            # nom de paramètre, l'API renvoie toujours la 1re page (faux plafond75            # de 100 items). Avec st+nt, on parcourt tout l'inventaire.76            meta = data.get("metadata") or {}77            token = meta.get("next_token")78            if not token or token == prev_token or len(out) == before:79                break80            prev_token = token81            params = {"st": meta.get("search_token") or "", "nt": token}82        return out8384    # -- mapping ---------------------------------------------------------------85    def _to_listing(self, it: dict) -> PropertyListing | None:86        ref = it.get("ref_number") or it.get("id")87        if not ref or not it.get("for_sale_flag", True):88            return None89        if (it.get("status_code") or "").upper() not in ("", "AVAILABLE"):90            return None9192        loc = it.get("location") or {}93        addr = loc.get("address") or {}94        street = " ".join(filter(None, [addr.get("street_number"),95                                        addr.get("street_name")])).strip()96        price = ((it.get("price") or {}).get("sell") or {}).get("amount")9798        cat = (it.get("category_code") or "").upper()99        sub = (it.get("subcategory_code") or "").upper()100        prop_type = _SUBCATEGORY.get(sub) or _CATEGORY.get(cat, "")101102        unit = it.get("main_unit") or {}103        building = it.get("building") or {}104        dim = building.get("dimension") or {}105        area = dim.get("area_sf")106        land = it.get("land") or {}107        lot = (land.get("dimension") or {}).get("area_sf")108109        photo = it.get("photo_url") or ""110        images = [photo.replace("-sm.", "-lg.")] if photo else []111112        # caractéristiques disponibles dans l'item (aucune requête détail)113        details = {"postal_code": addr.get("postal_code", "")}114        b_ass = (building.get("assessment") or {}).get("amount")115        l_ass = (land.get("assessment") or {}).get("amount")116        if b_ass or l_ass:117            details["Évaluation municipale"] = (118                f"{(b_ass or 0) + (l_ass or 0):,.0f} $".replace(",", " "))119        if dim.get("area_sf"):120            details["Superficie du bâtiment (au sol)"] = f"{dim['area_sf']:,.0f} pi²".replace(",", " ")121        if unit.get("room_count"):122            details["Nombre de pièces"] = unit["room_count"]123        features = []124        attrs = it.get("attributes") or {}125        if attrs.get("PARKING"):126            features.append("Stationnement")127        if it.get("virtual_tour_flag"):128            features.append("Visite virtuelle")129        if it.get("video_flag"):130            features.append("Vidéo")131132        return PropertyListing(133            source=self.source_id,134            external_id=str(ref),135            url=self._listing_url(it),136            title=street,137            address=street,138            city="",                       # code ville source.immo non résolu ici139            sector="",140            region=str(loc.get("region_code") or ""),141            property_type=prop_type,142            price=float(price) if price is not None else None,143            price_label=(f"{float(price):,.0f} $".replace(",", " ") if price else ""),144            bedrooms=unit.get("bedroom_count"),145            bathrooms=unit.get("bathroom_count"),146            powder_rooms=unit.get("waterroom_count"),147            area_sqft=float(area) if area else None,148            lot_sqft=float(lot) if lot else None,149            year_built=building.get("construction_year"),150            mls=str(ref),151            lat=loc.get("latitude"),152            lng=loc.get("longitude"),153            images=images,154            features=features,155            details=details,156        )157158    def _listing_url(self, it: dict) -> str:159        # URL lisible non essentielle ; on pointe la recherche du site par défaut.160        return self.site_url161