spb/home-ka

1# -----------------------------------------------------------------------------2# Home-Ka — US real-estate aggregator (Groupe KA)3# Author: Simon-Pierre Boucher — contact@spboucher.ai4# connectors/rets/client.py : legacy RETS connector (login + DMQL2 search,5# COMPACT-DECODED). Many small MLSs still hand out RETS credentials faster6# than RESO Web API access — this keeps those feeds first-class.7#8#   {"id": "rets_example", "connector_type": "rets", "config": {9#       "login_url": "https://rets.example-mls.com/rets/login.ashx",10#       "username": "$ENV:RETS_EXAMPLE_USER", "password": "$ENV:RETS_EXAMPLE_PASS",11#       "search_url": "",                       # discovered from login when empty12#       "resource": "Property", "class": "Residential",13#       "query": "(Status=|A)", "select": "",14#       "limit": 5000, "mapping": { ... Listing field -> RETS system name ... },15#       "user_agent": "HomeKa/1.0"}}16# -----------------------------------------------------------------------------17from __future__ import annotations1819import re20import xml.etree.ElementTree as ET2122import requests23from requests.auth import HTTPBasicAuth, HTTPDigestAuth2425from ..base import BaseConnector26from .._maputil import apply_mapping27from ...schema import Listing2829_CAP_RE = re.compile(r"<RETS-RESPONSE>(.*?)</RETS-RESPONSE>", re.S | re.I)303132class RETSConnector(BaseConnector):33    family = "rets"34    request_delay = 1.03536    def _login(self) -> str:37        """RETS Login transaction — returns the Search capability URL."""38        cfg = self.config39        if cfg.get("user_agent"):40            self.session.headers["User-Agent"] = cfg["user_agent"]41        self.session.headers["RETS-Version"] = cfg.get("rets_version", "RETS/1.7.2")42        auth_cls = HTTPDigestAuth if cfg.get("auth", "digest") == "digest" else HTTPBasicAuth43        self.session.auth = auth_cls(cfg.get("username", ""), cfg.get("password", ""))44        resp = self.get(cfg["login_url"])45        if cfg.get("search_url"):46            return cfg["search_url"]47        m = _CAP_RE.search(resp.text)48        caps = {}49        if m:50            for line in m.group(1).splitlines():51                if "=" in line:52                    k, _, v = line.partition("=")53                    caps[k.strip().lower()] = v.strip()54        search = caps.get("search")55        if not search:56            raise RuntimeError(f"{self.source_id}: no Search capability in RETS login")57        if search.startswith("/"):58            from urllib.parse import urlsplit59            u = urlsplit(cfg["login_url"])60            search = f"{u.scheme}://{u.netloc}{search}"61        return search6263    def fetch(self) -> list[Listing]:64        cfg = self.config65        search_url = self._login()66        params = {67            "SearchType": cfg.get("resource", "Property"),68            "Class": cfg.get("class", "Residential"),69            "Query": cfg.get("query", "(Status=|A)"),70            "QueryType": "DMQL2",71            "Format": "COMPACT-DECODED",72            "Limit": str(cfg.get("limit", 5000)),73            "StandardNames": cfg.get("standard_names", "0"),74        }75        if cfg.get("select"):76            params["Select"] = cfg["select"]77        resp = self.get(search_url, params=params)78        return self._parse_compact(resp.text)7980    def _parse_compact(self, body: str) -> list[Listing]:81        """Parse a COMPACT-DECODED RETS payload into Listings via the mapping."""82        try:83            root = ET.fromstring(body)84        except ET.ParseError as exc:85            raise RuntimeError(f"{self.source_id}: invalid RETS XML: {exc}")86        delim = "\t"87        d = root.find("DELIMITER")88        if d is not None and d.get("value"):89            delim = chr(int(d.get("value")))90        cols_el = root.find("COLUMNS")91        if cols_el is None or not cols_el.text:92            return []93        cols = cols_el.text.strip(delim).split(delim)94        mapping = self.config.get("mapping") or _DEFAULT_RETS_MAPPING95        static = self.config.get("static") or {}96        out: list[Listing] = []97        for row in root.findall("DATA"):98            vals = (row.text or "").strip("\n").strip(delim).split(delim)99            rec = {c: v for c, v in zip(cols, vals)}100            lst = apply_mapping(self.source_id, rec, mapping, static)101            if lst is not None:102                out.append(lst)103        return out104105106# RETS StandardNames (StandardNames=1) default mapping; system-name feeds107# override it entirely in the source config.108_DEFAULT_RETS_MAPPING = {109    "external_id": "ListingID",110    "mls_id": "ListingID",111    "list_price": "ListPrice",112    "street_address": "StreetAddress",113    "city": "City",114    "state": "StateOrProvince",115    "zip_code": "PostalCode",116    "county": "County",117    "property_type": "PropertyType",118    "bedrooms": "Beds",119    "bathrooms": "Baths",120    "living_area_sqft": "LivingArea",121    "year_built": "YearBuilt",122    "status": "Status",123    "agent_name": "ListAgentName",124    "brokerage_name": "ListOfficeName",125    "description": "PublicRemarks",126    "lat": "Latitude",127    "lng": "Longitude",128}129