spb/home-ka

1# -----------------------------------------------------------------------------2# Home-Ka — US real-estate aggregator (Groupe KA)3# Author: Simon-Pierre Boucher — contact@spboucher.ai4# connectors/sftp/drop.py : SFTP drop connector — brokerages/MLS vendors that5# push periodic CSV/XML exports to an SFTP folder. Downloads the newest6# matching file then dispatches to the csv/xml parser.7#8#   {"id": "acme_sftp", "connector_type": "sftp", "config": {9#       "host": "sftp.acmefeeds.com", "port": 22,10#       "username": "$ENV:ACME_SFTP_USER", "password": "$ENV:ACME_SFTP_PASS",11#       "key_path": "",                       # or private key file12#       "remote_dir": "/exports", "pattern": "listings_*.csv",13#       "format": "csv",                      # csv | xml14#       ... then the csv/xml family options (mapping, static, ...)}}15#16# Requires `paramiko` (in requirements.txt).17# -----------------------------------------------------------------------------18from __future__ import annotations1920import fnmatch21import io2223from ..base import BaseConnector24from ...schema import Listing252627class SFTPDropConnector(BaseConnector):28    family = "sftp"2930    def _download(self) -> bytes:31        try:32            import paramiko33        except ImportError:34            raise RuntimeError("paramiko missing — pip install paramiko")35        cfg = self.config36        client = paramiko.SSHClient()37        client.set_missing_host_key_policy(paramiko.AutoAddPolicy())38        kw: dict = {"username": cfg.get("username", "")}39        if cfg.get("key_path"):40            kw["key_filename"] = cfg["key_path"]41        else:42            kw["password"] = cfg.get("password", "")43        client.connect(cfg["host"], port=int(cfg.get("port", 22)),44                       timeout=30, **kw)45        try:46            sftp = client.open_sftp()47            remote_dir = cfg.get("remote_dir", ".")48            pattern = cfg.get("pattern", "*")49            names = [n for n in sftp.listdir(remote_dir)50                     if fnmatch.fnmatch(n, pattern)]51            if not names:52                raise RuntimeError(f"{self.source_id}: no file matching "53                                   f"{pattern} in {remote_dir}")54            # newest file by mtime55            newest = max(names, key=lambda n: sftp.stat(f"{remote_dir}/{n}").st_mtime)56            buf = io.BytesIO()57            sftp.getfo(f"{remote_dir}/{newest}", buf)58            return buf.getvalue()59        finally:60            client.close()6162    def fetch(self) -> list[Listing]:63        data = self._download()64        fmt = self.config.get("format", "csv")65        if fmt == "csv":66            from ..csv.feed import parse_csv67            return parse_csv(self.source_id,68                             data.decode(self.config.get("encoding", "utf-8-sig"),69                                         errors="replace"),70                             self.config)71        if fmt == "xml":72            from ..xml.feed import XMLFeedConnector73            conn = XMLFeedConnector(self.source_id,74                                    {**self.config, "path": None})75            import xml.etree.ElementTree as ET76            root = ET.fromstring(data)77            item_tag = self.config.get("item_tag", "listing")78            mapping = self.config.get("mapping") or {}79            static = self.config.get("static") or {}80            out = []81            for el in root.iter():82                if el.tag.rsplit("}", 1)[-1] != item_tag:83                    continue84                lst = conn._to_listing(el, mapping, static)85                if lst is not None:86                    out.append(lst)87            return out88        raise RuntimeError(f"{self.source_id}: unknown sftp format {fmt!r}")89