Python 49.6%
TypeScript 25.5%
CSS 24.1%
1# -----------------------------------------------------------------------------2# Home-Ka — US real-estate aggregator (Groupe KA)3# Author: Simon-Pierre Boucher — contact@spboucher.ai4# connectors/base.py : connector base class + fetch backends.5#6# Same philosophy as immo-ka: SOURCE → connector → raw ingestion →7# normalization → property matching → deduplication → canonical listing →8# search index → Home-Ka. One connector adapts ONE source; the pipeline9# (ingest.py) handles the diff with the database.10#11# Two kinds of connectors:12# - FAMILY connectors (reso/, rets/, xml/, json/, csv/, sftp/, public_data/):13# generic, config-driven — one class serves many sources. Instantiated14# with (source_id, config) from a row of the `sources` table. They carry15# a class attribute `family` (e.g. "reso").16# - CUSTOM connectors (custom_broker/*.py): one class per site, class17# attribute `source_id` set, zero-arg constructible — exactly like18# immo-ka's per-agency connectors.19#20# Data-sourcing policy (Home-Ka): favor the LOWEST-LEVEL source available —21# direct feeds (RESO Web API, RETS, XML/JSON/CSV drops) over site crawling.22# Crawling is a discovery/bootstrap tool, not the foundation; the proxy23# backends (Firecrawl/Scrapfly) exist as a LAST RESORT only.24# -----------------------------------------------------------------------------25from __future__ import annotations2627import json28import os29import time3031import requests3233from ..schema import Listing3435USER_AGENT = ("Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) "36 "AppleWebKit/537.36 (KHTML, like Gecko) Chrome/126 Safari/537.36 "37 "HomeKaBot/1.0 (+https://www.home-ka.com/bot; contact@spboucher.ai)")3839FIRECRAWL_API = "https://api.firecrawl.dev/v1/scrape"40SCRAPFLY_API = "https://api.scrapfly.io/scrape"414243def expand_env(config: dict) -> dict:44 """Resolve '$ENV:NAME' placeholders in a source config — credentials live45 in the environment (.env), never in the database."""46 out = {}47 for k, v in (config or {}).items():48 if isinstance(v, str) and v.startswith("$ENV:"):49 out[k] = os.environ.get(v[5:], "")50 elif isinstance(v, dict):51 out[k] = expand_env(v)52 else:53 out[k] = v54 return out555657class BaseConnector:58 """A connector = one adapter for one listing source.5960 Family subclasses: set `family`, accept (source_id, config) and implement61 `fetch()`. Custom subclasses: set `source_id` and implement `fetch()`.62 `fetch()` returns the complete list of listings currently visible at the63 source; the pipeline handles additions/changes/removals.64 """6566 family: str = "" # set on generic, config-driven families67 source_id: str = "" # set on custom one-site connectors68 request_delay: float = 0.6 # politeness between requests69 timeout: int = 3070 use_detail_cache: bool = True # DB cache of detail pages7172 def __init__(self, source_id: str | None = None,73 config: dict | None = None) -> None:74 if source_id:75 self.source_id = source_id76 self.config = expand_env(config or {})77 if self.config.get("request_delay"):78 self.request_delay = float(self.config["request_delay"])79 self.session = requests.Session()80 self.session.headers["User-Agent"] = USER_AGENT81 self._last_request = 0.082 self._detail_con = None8384 # -- backends -------------------------------------------------------------85 def _throttle(self) -> None:86 wait = self.request_delay - (time.time() - self._last_request)87 if wait > 0:88 time.sleep(wait)8990 def get(self, url: str, **kw) -> requests.Response:91 """Polite direct GET."""92 self._throttle()93 resp = self.session.get(url, timeout=self.timeout, **kw)94 self._last_request = time.time()95 resp.raise_for_status()96 return resp9798 def post(self, url: str, **kw) -> requests.Response:99 """Polite direct POST (internal search APIs)."""100 self._throttle()101 resp = self.session.post(url, timeout=self.timeout, **kw)102 self._last_request = time.time()103 resp.raise_for_status()104 return resp105106 def fc_scrape(self, url: str, proxy: str = "auto", render: bool = False,107 wait_for: int = 0, retries: int = 2) -> str:108 """Raw HTML via Firecrawl — used as a THROUGHPUT backend (parallel109 fetching without hammering the source directly) and to reach110 bot-challenged sites. `maxAge` leans on Firecrawl's cache so weekly111 re-syncs stay cheap. Retries on 429/5xx with backoff."""112 key = os.environ.get("FIRECRAWL_API_KEY")113 if not key:114 raise RuntimeError("FIRECRAWL_API_KEY missing (see .env)")115 payload: dict = {"url": url, "timeout": 60000,116 "formats": ["html"] if render else ["rawHtml"],117 "proxy": proxy,118 "maxAge": 6 * 86400 * 1000}119 if wait_for:120 payload["waitFor"] = wait_for121 last = ""122 for attempt in range(retries + 1):123 try:124 resp = requests.post(125 FIRECRAWL_API, json=payload,126 headers={"Authorization": f"Bearer {key}"}, timeout=120)127 if resp.status_code in (429, 500, 502, 503):128 time.sleep(3 * (attempt + 1))129 continue130 resp.raise_for_status()131 data = resp.json().get("data") or {}132 return data.get("rawHtml") or data.get("html") or ""133 except requests.RequestException as exc:134 last = str(exc)135 time.sleep(2 * (attempt + 1))136 raise RuntimeError(f"firecrawl failed for {url}: {last}")137138 def get_rendered(self, url: str, wait_for: int = 0,139 proxy: str | None = None) -> str:140 """LAST RESORT — JS-rendered HTML via Firecrawl (FIRECRAWL_API_KEY)."""141 key = os.environ.get("FIRECRAWL_API_KEY")142 if not key:143 raise RuntimeError("FIRECRAWL_API_KEY missing (see .env)")144 payload: dict = {"url": url, "formats": ["html"], "timeout": 90000}145 if wait_for:146 payload["waitFor"] = wait_for147 if proxy:148 payload["proxy"] = proxy149 resp = requests.post(FIRECRAWL_API, json=payload,150 headers={"Authorization": f"Bearer {key}"},151 timeout=150)152 resp.raise_for_status()153 return (resp.json().get("data") or {}).get("html", "")154155 def scrapfly(self, url: str, render_js: bool = True, asp: bool = True,156 rendering_wait: int = 0, country: str = "us",157 wait_for_selector: str | None = None,158 js_scenario: list | str | None = None,159 proxy_pool: str | None = None, headers: dict | None = None,160 method: str = "GET", body: str | None = None) -> dict:161 """LAST RESORT — full Scrapfly call, returns the `result` dict.162 Home-Ka policy: only for sources with no lower-level path; convert163 every interesting source to a direct feed instead."""164 import base64165 key = os.environ.get("SCRAPFLY_KEY")166 if not key:167 raise RuntimeError("SCRAPFLY_KEY missing (see .env)")168 params: dict = {"key": key, "url": url, "country": country}169 if asp:170 params["asp"] = "true"171 if render_js:172 params["render_js"] = "true"173 if rendering_wait:174 params["rendering_wait"] = rendering_wait175 if wait_for_selector:176 params["wait_for_selector"] = wait_for_selector177 if proxy_pool:178 params["proxy_pool"] = proxy_pool179 if js_scenario is not None:180 js = js_scenario if isinstance(js_scenario, str) else json.dumps(js_scenario)181 params["js_scenario"] = base64.urlsafe_b64encode(js.encode()).decode()182 if headers:183 for k, v in headers.items():184 params[f"headers[{k}]"] = v185 self._throttle()186 if method.upper() == "POST":187 resp = requests.post(SCRAPFLY_API, params=params,188 data=(body or ""), timeout=180)189 else:190 resp = requests.get(SCRAPFLY_API, params=params, timeout=180)191 self._last_request = time.time()192 try:193 return resp.json().get("result") or {}194 except ValueError:195 return {}196197 def detail(self, external_id: str, key: str, fetch_fn) -> dict:198 """Detail-page payload with cache: `fetch_fn` only runs when the199 listing is new or its list-content hash changed."""200 if not self.use_detail_cache:201 return fetch_fn() or {}202 from .. import db203 if self._detail_con is None:204 self._detail_con = db.connect()205 cached = db.get_cached_detail(self._detail_con, self.source_id,206 str(external_id), key)207 if cached is not None:208 return cached209 payload = fetch_fn() or {}210 db.put_cached_detail(self._detail_con, self.source_id,211 str(external_id), key, payload)212 return payload213214 # -- contract --------------------------------------------------------------215 def fetch(self) -> list[Listing]:216 raise NotImplementedError217218219# =============================================================================220# Anti-bot resilience (Groupe KA) — auto-escalation of get() without touching221# connector bodies: an anti-bot block (403/429/503/challenge) or a network cut222# triggers the fallback chain (Oxylabs residential -> Scrapfly ASP -> Bright223# Data). See connectors/_resilient.py. Idempotent (_KA_RESILIENT_WRAPPED).224# Direct fetch stays the default — proxies only fire when blocked.225# =============================================================================226if not getattr(BaseConnector, "_KA_RESILIENT_WRAPPED", False):227 import requests as _ka_requests # noqa: E402228 from . import _resilient as _kar # noqa: E402229230 _ka_orig_get = BaseConnector.get231232 def _ka_full_url(url, kw):233 try:234 return _ka_requests.Request("GET", url,235 params=kw.get("params")).prepare().url236 except Exception: # noqa: BLE001237 return url238239 def _ka_resilient_get(self, url, **kw):240 timeout = getattr(self, "timeout", 30)241 headers = kw.get("headers")242 try:243 return _ka_orig_get(self, url, **kw)244 except _ka_requests.HTTPError as exc:245 r = getattr(exc, "response", None)246 if r is not None and _kar.is_blocked(r):247 target = getattr(r, "url", None) or _ka_full_url(url, kw)248 better = _kar.escalate_if_blocked(249 r, target, timeout=timeout, headers=headers)250 if better is not None and getattr(better, "status_code", 0) == 200:251 return better252 raise253 except (_ka_requests.ConnectionError, _ka_requests.Timeout):254 better = _kar.escalate(_ka_full_url(url, kw),255 timeout=timeout, headers=headers)256 if better is not None and getattr(better, "status_code", 0) == 200:257 return better258 raise259260 def _ka_get_resilient(self, url, *, render_js=False, country="us", **kw):261 """Explicit anti-bot fetch: forces the fallback chain when needed."""262 timeout = getattr(self, "timeout", 30)263 headers = kw.get("headers")264 try:265 resp = _ka_orig_get(self, url, **kw)266 except _ka_requests.HTTPError as exc:267 resp = getattr(exc, "response", None)268 except (_ka_requests.ConnectionError, _ka_requests.Timeout):269 resp = None270 target = _ka_full_url(url, kw)271 if resp is not None and getattr(resp, "url", None):272 target = resp.url273 return _kar.escalate_if_blocked(resp, target, timeout=timeout,274 country=country, render_js=render_js,275 headers=headers)276277 BaseConnector.get = _ka_resilient_get278 BaseConnector.get_resilient = _ka_get_resilient279 BaseConnector._KA_RESILIENT_WRAPPED = True280