spb/home-ka

1# -----------------------------------------------------------------------------2# Home-Ka — US real-estate aggregator (Groupe KA)3# Author: Simon-Pierre Boucher — contact@spboucher.ai4# connectors/xml/feed.py : generic XML feed connector (brokerage exports,5# franchise syndication feeds, legacy XML drops). Config-driven:6#7#   {"id": "acme_xml", "connector_type": "xml", "config": {8#       "url": "https://feeds.acmerealty.com/listings.xml",9#       "item_tag": "listing",                 # repeated element10#       "mapping": {"external_id": "mlsNumber", "list_price": "price",11#                   "street_address": "address/street", "city": "address/city",12#                   "state": "address/state", "zip_code": "address/zip",13#                   "images": "photos/photo", ...},   # ElementTree paths14#       "static": {"state": "TX"},15#       "headers": {}, "auth_user": "", "auth_pass": ""}}16# -----------------------------------------------------------------------------17from __future__ import annotations1819import xml.etree.ElementTree as ET2021from ..base import BaseConnector22from ...schema import Listing2324_LIST_FIELDS = {"images", "features"}252627def _strip_ns(tag: str) -> str:28    return tag.rsplit("}", 1)[-1]293031class XMLFeedConnector(BaseConnector):32    family = "xml"33    request_delay = 0.23435    def fetch(self) -> list[Listing]:36        cfg = self.config37        kw = {}38        if cfg.get("headers"):39            kw["headers"] = cfg["headers"]40        if cfg.get("auth_user"):41            kw["auth"] = (cfg["auth_user"], cfg.get("auth_pass", ""))42        if cfg.get("path"):43            body = open(cfg["path"], "rb").read()44        else:45            body = self.get(cfg["url"], **kw).content46        root = ET.fromstring(body)47        item_tag = cfg.get("item_tag", "listing")48        mapping = cfg.get("mapping") or {}49        static = cfg.get("static") or {}50        out: list[Listing] = []51        for el in root.iter():52            if _strip_ns(el.tag) != item_tag:53                continue54            lst = self._to_listing(el, mapping, static)55            if lst is not None:56                out.append(lst)57        return out5859    def _find(self, el: ET.Element, path: str):60        """Namespace-tolerant lookup ('address/city' or '@attr' or 'tag@attr')."""61        if path.startswith("@"):62            return [el.get(path[1:])]63        attr = None64        if "@" in path:65            path, _, attr = path.partition("@")66        parts = path.split("/")67        nodes = [el]68        for p in parts:69            nxt = []70            for n in nodes:71                nxt.extend(c for c in n if _strip_ns(c.tag) == p)72            nodes = nxt73        if attr:74            return [n.get(attr) for n in nodes]75        return [(n.text or "").strip() for n in nodes]7677    def _to_listing(self, el: ET.Element, mapping: dict,78                    static: dict) -> Listing | None:79        kw: dict = {"source": self.source_id, "external_id": "", "url": "",80                    "details": {}}81        for field, path in mapping.items():82            vals = [v for v in self._find(el, path) if v]83            if not vals:84                continue85            kw[field] = vals if field in _LIST_FIELDS else vals[0]86        for field, v in static.items():87            kw.setdefault(field, v)88        if not kw.get("external_id"):89            return None90        kw["external_id"] = str(kw["external_id"])91        # flat leaf values preserved for debugging/passthrough92        for c in el:93            if len(c) == 0 and c.text and c.text.strip():94                kw["details"].setdefault(_strip_ns(c.tag), c.text.strip())95        allowed = set(Listing.__dataclass_fields__)96        return Listing(**{k: v for k, v in kw.items() if k in allowed})97