spb/home-ka

1# -----------------------------------------------------------------------------2# Home-Ka — US real-estate aggregator (Groupe KA)3# Author: Simon-Pierre Boucher — contact@spboucher.ai4# connectors/json/feed.py : generic JSON API connector — partner APIs,5# brokerage internal search APIs, WordPress REST (wp-json) listing CPTs.6# Config-driven; pagination by page parameter or by next-link path.7#8#   {"id": "acme_api", "connector_type": "json", "config": {9#       "url": "https://www.acmerealty.com/api/listings?page={page}",10#       "method": "GET", "headers": {}, "body": null,11#       "items_path": "data.results",          # dot path to the record list12#       "next_path": "data.next",              # OR page-based: start_page/max_pages13#       "start_page": 1, "max_pages": 50, "stop_when_empty": true,14#       "mapping": {"external_id": "id", "url": "permalink",15#                   "list_price": "price", "images": "photos[].url", ...},16#       "static": {"state": "CO"}}}17# -----------------------------------------------------------------------------18from __future__ import annotations1920from ..base import BaseConnector21from .._maputil import apply_mapping, dig22from ...schema import Listing232425class JSONFeedConnector(BaseConnector):26    family = "json"27    request_delay = 0.52829    def fetch(self) -> list[Listing]:30        cfg = self.config31        mapping = cfg.get("mapping") or {}32        static = cfg.get("static") or {}33        items_path = cfg.get("items_path", "")34        headers = cfg.get("headers") or {}35        out: list[Listing] = []36        seen: set[str] = set()3738        url_tpl = cfg.get("url", "")39        page = int(cfg.get("start_page", 1))40        max_pages = int(cfg.get("max_pages", 100))41        next_url: str | None = None4243        for _ in range(max_pages):44            url = next_url or url_tpl.format(page=page)45            if cfg.get("method", "GET").upper() == "POST":46                body = cfg.get("body")47                if isinstance(body, dict):48                    body = {k: (str(v).format(page=page) if isinstance(v, str) else v)49                            for k, v in body.items()}50                resp = self.post(url, json=body, headers=headers)51            else:52                resp = self.get(url, headers=headers)53            data = resp.json()54            items = dig(data, items_path) if items_path else data55            if not isinstance(items, list):56                items = [items] if isinstance(items, dict) else []57            new = 058            for rec in items:59                if not isinstance(rec, dict):60                    continue61                lst = apply_mapping(self.source_id, rec, mapping, static)62                if lst is not None and lst.uid not in seen:63                    seen.add(lst.uid)64                    out.append(lst)65                    new += 166            if cfg.get("next_path"):67                next_url = dig(data, cfg["next_path"])68                if not next_url:69                    break70            else:71                page += 172                if cfg.get("stop_when_empty", True) and new == 0:73                    break74                if "{page}" not in url_tpl:75                    break76        return out77