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/csv/feed.py : CSV feed connector — brokerage exports, scheduled5# drops (also consumed by the sftp family after download).6#7# {"id": "acme_csv", "connector_type": "csv", "config": {8# "url": "https://feeds.acme.com/export.csv", # or "path": local file9# "delimiter": ",", "encoding": "utf-8-sig",10# "mapping": {"external_id": "MLS #", "list_price": "Price",11# "street_address": "Address", "city": "City", ...},12# "images_split": "|", # column with delimiter-joined URLs13# "static": {"state": "FL"}}}14# -----------------------------------------------------------------------------15from __future__ import annotations1617import csv18import io1920from ..base import BaseConnector21from .._maputil import apply_mapping22from ...schema import Listing232425class CSVFeedConnector(BaseConnector):26 family = "csv"27 request_delay = 0.22829 def fetch(self) -> list[Listing]:30 cfg = self.config31 if cfg.get("path"):32 text = open(cfg["path"], encoding=cfg.get("encoding", "utf-8-sig"),33 newline="").read()34 else:35 resp = self.get(cfg["url"])36 resp.encoding = cfg.get("encoding", "utf-8-sig")37 text = resp.text38 return parse_csv(self.source_id, text, cfg)394041def parse_csv(source_id: str, text: str, cfg: dict) -> list[Listing]:42 """Shared with the sftp family."""43 mapping = cfg.get("mapping") or {}44 static = cfg.get("static") or {}45 split = cfg.get("images_split", "|")46 out: list[Listing] = []47 reader = csv.DictReader(io.StringIO(text),48 delimiter=cfg.get("delimiter", ","))49 for rec in reader:50 lst = apply_mapping(source_id, rec, mapping, static)51 if lst is None:52 continue53 if lst.images and len(lst.images) == 1 and split in (lst.images[0] or ""):54 lst.images = [u.strip() for u in lst.images[0].split(split) if u.strip()]55 out.append(lst)56 return out57