spb/home-ka

1# -----------------------------------------------------------------------------2# Home-Ka — US real-estate aggregator (Groupe KA)3# Author: Simon-Pierre Boucher — contact@spboucher.ai4# connectors/_maputil.py : shared helpers for config-driven field mapping.5#6# Family connectors (json, xml, csv, sftp, reso) turn a raw record into a7# Listing through a declarative mapping stored in the source config:8#   "mapping": {"external_id": "ListingId", "list_price": "price.amount",9#               "images": "photos[].url", "city": "address.city", ...}10# Paths are dot-separated; "[]" fans out over a list. Unmapped raw fields are11# preserved in Listing.details for the RESO passthrough in schema.finalize().12# -----------------------------------------------------------------------------13from __future__ import annotations1415from ..schema import Listing1617__all__ = ["dig", "apply_mapping"]1819_LIST_FIELDS = {"images", "features"}202122def dig(record, path: str):23    """Resolve a dot path in a nested dict/list structure.2425    "photos[].url" → [p["url"] for p in record["photos"]];26    "address.city" → record["address"]["city"]. Returns None when absent.27    """28    cur = record29    for part in str(path).split("."):30        if cur is None:31            return None32        fan = part.endswith("[]")33        key = part[:-2] if fan else part34        if key:35            if isinstance(cur, dict):36                cur = cur.get(key)37            elif isinstance(cur, list):38                cur = [c.get(key) if isinstance(c, dict) else None for c in cur]39            else:40                return None41        if fan and not isinstance(cur, list):42            cur = [cur] if cur is not None else []43    return cur444546def apply_mapping(source_id: str, record: dict, mapping: dict,47                  static: dict | None = None) -> Listing | None:48    """Build a Listing from a raw record and a declarative mapping."""49    kw: dict = {"source": source_id, "external_id": "", "url": ""}50    for field, path in (mapping or {}).items():51        v = dig(record, path)52        if v is None:53            continue54        if field in _LIST_FIELDS:55            v = [x for x in (v if isinstance(v, list) else [v]) if x]56        kw[field] = v57    for field, v in (static or {}).items():58        kw.setdefault(field, v)59    if not kw.get("external_id"):60        return None61    kw["external_id"] = str(kw["external_id"])62    kw.setdefault("details", {})63    # keep flat raw scalars for the RESO passthrough / debugging64    if isinstance(record, dict):65        for k, v in record.items():66            if isinstance(v, (str, int, float, bool)) and k not in kw["details"]:67                kw["details"][k] = v68    allowed = set(Listing.__dataclass_fields__)69    return Listing(**{k: v for k, v in kw.items() if k in allowed})70