# ----------------------------------------------------------------------------- # Home-Ka — US real-estate aggregator (Groupe KA) # Author: Simon-Pierre Boucher — contact@spboucher.ai # connectors/reso/webapi.py : RESO Web API (OData) connector — the PREFERRED # feed type. Works with MLS Grid, Trestle (CoreLogic), Bridge Interactive, # Spark/FBS, RMLS direct RESO endpoints, and any Data Dictionary compliant # server. One `sources` row per feed: # # {"id": "mlsgrid_actris", "connector_type": "reso", "config": { # "base_url": "https://api.mlsgrid.com/v2", # "token": "$ENV:MLSGRID_TOKEN", # or OAuth2 below # "token_url": "", "client_id": "", "client_secret": "", # "resource": "Property", # "filter": "StandardStatus eq 'Active' and PropertyType eq 'Residential'", # "expand": "Media", # "page_size": 1000, "max_records": 50000, # "originating_system": "ACTRIS"}} # # Field mapping is RESO Data Dictionary — schema.Listing understands the # standard names natively (details passthrough), with the explicit mapping # below for the core columns. # ----------------------------------------------------------------------------- from __future__ import annotations import time import requests from ..base import BaseConnector from ...schema import Listing class RESOWebAPIConnector(BaseConnector): family = "reso" request_delay = 0.3 # -- auth ------------------------------------------------------------- def _token(self) -> str: cfg = self.config if cfg.get("token"): return cfg["token"] if cfg.get("token_url"): resp = requests.post(cfg["token_url"], data={ "grant_type": "client_credentials", "client_id": cfg.get("client_id", ""), "client_secret": cfg.get("client_secret", ""), "scope": cfg.get("scope", "api"), }, timeout=30) resp.raise_for_status() return resp.json()["access_token"] raise RuntimeError(f"{self.source_id}: no RESO token/credentials configured") # -- fetch ------------------------------------------------------------ def fetch(self) -> list[Listing]: cfg = self.config base = (cfg.get("base_url") or "").rstrip("/") if not base: raise RuntimeError(f"{self.source_id}: base_url missing") resource = cfg.get("resource", "Property") page_size = int(cfg.get("page_size", 1000)) max_records = int(cfg.get("max_records", 100_000)) params = {"$top": page_size} if cfg.get("filter"): params["$filter"] = cfg["filter"] if cfg.get("select"): params["$select"] = cfg["select"] if cfg.get("expand"): params["$expand"] = cfg["expand"] self.session.headers["Authorization"] = f"Bearer {self._token()}" self.session.headers["Accept"] = "application/json" out: list[Listing] = [] url: str | None = f"{base}/{resource}" first = True while url and len(out) < max_records: resp = self.get(url, params=params if first else None) first = False data = resp.json() for rec in data.get("value", []): lst = self._to_listing(rec) if lst is not None: out.append(lst) url = data.get("@odata.nextLink") return out def _to_listing(self, rec: dict) -> Listing | None: key = rec.get("ListingKey") or rec.get("ListingId") if not key: return None media = rec.get("Media") or [] images = [m.get("MediaURL") for m in media if isinstance(m, dict) and m.get("MediaURL") and (m.get("MediaCategory") in (None, "Photo"))] images.sort(key=lambda u: u or "") if media and all(isinstance(m, dict) for m in media): ordered = sorted(media, key=lambda m: m.get("Order") or 0) images = [m.get("MediaURL") for m in ordered if m.get("MediaURL")] details = {k: v for k, v in rec.items() if isinstance(v, (str, int, float, bool))} street = rec.get("UnparsedAddress") or " ".join( str(rec.get(k) or "") for k in ("StreetNumber", "StreetDirPrefix", "StreetName", "StreetSuffix") ).strip() return Listing( source=self.source_id, external_id=str(key), url=self.config.get("listing_url_template", "").format(**rec) if self.config.get("listing_url_template") else "", street_address=street, unit=str(rec.get("UnitNumber") or ""), city=str(rec.get("City") or ""), state=str(rec.get("StateOrProvince") or ""), zip_code=str(rec.get("PostalCode") or ""), county=str(rec.get("CountyOrParish") or ""), property_type=str(rec.get("PropertySubType") or rec.get("PropertyType") or ""), property_subtype=str(rec.get("PropertySubType") or ""), list_price=rec.get("ListPrice"), bedrooms=rec.get("BedroomsTotal"), bathrooms_full=rec.get("BathroomsFull"), bathrooms_half=rec.get("BathroomsHalf"), bathrooms=rec.get("BathroomsTotalInteger"), living_area_sqft=rec.get("LivingArea"), lot_size_sqft=rec.get("LotSizeSquareFeet"), year_built=rec.get("YearBuilt"), apn=str(rec.get("ParcelNumber") or ""), mls_id=str(rec.get("ListingId") or ""), mls_name=str(rec.get("OriginatingSystemName") or self.config.get("originating_system") or ""), status=str(rec.get("StandardStatus") or "Active"), listed_at=str(rec.get("ListingContractDate") or ""), brokerage_name=str(rec.get("ListOfficeName") or ""), office_name=str(rec.get("ListOfficeName") or ""), agent_name=str(rec.get("ListAgentFullName") or ""), agent_phone=str(rec.get("ListAgentPreferredPhone") or rec.get("ListAgentDirectPhone") or ""), description=str(rec.get("PublicRemarks") or ""), details=details, images=images, lat=rec.get("Latitude"), lng=rec.get("Longitude"), ) def probe(base_url: str, token: str = "") -> dict: """Quick health probe of a RESO endpoint ($metadata reachability) — used by the admin and the brokerage discovery pipeline.""" headers = {"Accept": "application/json"} if token: headers["Authorization"] = f"Bearer {token}" t0 = time.time() try: resp = requests.get(base_url.rstrip("/") + "/$metadata", headers=headers, timeout=20) return {"ok": resp.status_code < 500, "status": resp.status_code, "seconds": round(time.time() - t0, 2)} except requests.RequestException as exc: return {"ok": False, "error": str(exc)}