# ----------------------------------------------------------------------------- # Home-Ka — US real-estate aggregator (Groupe KA) # Author: Simon-Pierre Boucher — contact@spboucher.ai # schema.py : standardized data model (Listing), RESO Data Dictionary aligned # # Same philosophy as immo-ka/schema.py: every connector, whatever the source # (RESO Web API, RETS, XML/JSON feed, CSV drop, county records, brokerage # site), must produce `Listing` objects conforming to this schema. `finalize()` # then applies the shared normalization layer (homeka/normalize.py) so the # connectors stay simple and fill raw fields only. # # Home-Ka separates PROPERTY from LISTING: # - a Listing is one publication of a sale by one source (this dataclass); # - a Property is the physical asset, persisted in the `properties` table # and matched by address/APN (homeka/propertymatch.py). A property keeps # existing after its listing goes off-market. # ----------------------------------------------------------------------------- from __future__ import annotations import hashlib import json from dataclasses import dataclass, field, asdict from .normalize import ( clean_address, clean_description, clean_title, extract_beds_baths, normalize_property_type, normalize_state, normalize_status, normalize_zip, parse_area_sqft, parse_int, parse_lot_sqft, parse_price, parse_year, price_is_from, ) __all__ = ["Listing"] # RESO Data Dictionary fields commonly promoted from `details` when a # connector passed them raw (Web API / RETS payloads keep original keys). _RESO_ALIASES = { "list_price": ("ListPrice",), "street_address": ("UnparsedAddress", "StreetAddressFull"), "city": ("City",), "state": ("StateOrProvince",), "zip_code": ("PostalCode",), "county": ("CountyOrParish",), "property_type": ("PropertySubType", "PropertyType"), "bedrooms": ("BedroomsTotal",), "bathrooms_full": ("BathroomsFull",), "bathrooms_half": ("BathroomsHalf",), "living_area_sqft": ("LivingArea", "BuildingAreaTotal"), "lot_size_sqft": ("LotSizeSquareFeet",), "year_built": ("YearBuilt",), "apn": ("ParcelNumber",), "mls_id": ("ListingId",), "status": ("StandardStatus", "MlsStatus"), "lat": ("Latitude",), "lng": ("Longitude",), } @dataclass class Listing: """Standardized Home-Ka for-sale listing (one publication at one source).""" source: str # source id (row in the `sources` table) external_id: str # id at the source (often the MLS number) url: str # listing page at the source title: str = "" # e.g. "Craftsman bungalow — Austin, TX" street_address: str = "" # street address (no city/state/zip) unit: str = "" # apt/unit/suite city: str = "" state: str = "" # 2-letter USPS code zip_code: str = "" # 5-digit ZIP county: str = "" property_type: str = "" # canonical (Single Family, Condo, …) property_subtype: str = "" # raw RESO PropertySubType if available list_price: float | None = None # asking price ($ USD) price_label: str = "" # original text (e.g. "$459,000") bedrooms: int | None = None bathrooms_full: int | None = None bathrooms_half: int | None = None bathrooms: float | None = None # total (full + 0.5*half) when known living_area_sqft: float | None = None lot_size_sqft: float | None = None year_built: int | None = None apn: str = "" # assessor parcel number (property match) mls_id: str = "" # MLS listing number if displayed mls_name: str = "" # originating MLS (e.g. "ACTRIS") status: str = "active" # active | pending | sold | withdrawn | coming-soon listed_at: str = "" # listing/contract date if known (ISO) brokerage_name: str = "" # listing brokerage office_name: str = "" # office/branch agent_name: str = "" agent_phone: str = "" agent_email: str = "" description: str = "" features: list[str] = field(default_factory=list) # source feature texts details: dict = field(default_factory=dict) # structured fields (JSON) images: list[str] = field(default_factory=list) # absolute URLs lat: float | None = None lng: float | None = None @property def uid(self) -> str: return f"{self.source}:{self.external_id}" def content_hash(self) -> str: """Content hash for change detection (pseudo-webhook, as in immo-ka).""" payload = asdict(self) blob = json.dumps(payload, sort_keys=True, ensure_ascii=False, default=str) return hashlib.sha256(blob.encode("utf-8")).hexdigest() # -- normalization --------------------------------------------------------- def finalize(self) -> "Listing": """Apply the shared normalization. Called by the ingestion pipeline. Idempotent; never overrides an explicit connector value. """ # RESO passthrough: promote Data Dictionary keys left in `details` for attr, keys in _RESO_ALIASES.items(): if getattr(self, attr, None) in (None, "", 0): for k in keys: v = self.details.get(k) if v not in (None, "", [], {}): setattr(self, attr, v) break self.title = clean_title(self.title) self.street_address = clean_address(str(self.street_address)) self.city = clean_title(str(self.city or "").strip()) self.state = normalize_state(str(self.state)) self.zip_code = normalize_zip(self.zip_code) self.county = clean_title(str(self.county or "").replace(" County", "").strip()) self.status = normalize_status(self.status) if self.property_subtype and not self.property_type: self.property_type = self.property_subtype self.property_type = normalize_property_type(str(self.property_type)) self.description = clean_description(self.description) self.mls_id = str(self.mls_id or "").strip() self.apn = str(self.apn or "").strip().replace(" ", "") # gallery: valid absolute URLs only, dedup (see homeka/imgaudit.py) try: from .imgaudit import clean_gallery self.images = clean_gallery(self.images) except Exception: self.images = [u for u in (self.images or []) if isinstance(u, str) and u.startswith("http")] if self.list_price is None: self.list_price = parse_price(self.price_label) else: self.list_price = parse_price(self.list_price) if self.price_label and price_is_from(self.price_label): self.details.setdefault("price_from", True) # numeric coercion (RESO passthrough may leave strings/decimals) self.bedrooms = parse_int(self.bedrooms) self.bathrooms_full = parse_int(self.bathrooms_full) self.bathrooms_half = parse_int(self.bathrooms_half) self.living_area_sqft = parse_area_sqft(self.living_area_sqft) self.lot_size_sqft = parse_lot_sqft(self.lot_size_sqft) self.year_built = parse_year(self.year_built) try: self.lat = float(self.lat) if self.lat is not None else None self.lng = float(self.lng) if self.lng is not None else None except (TypeError, ValueError): self.lat = self.lng = None # total bathrooms if self.bathrooms is None: if self.bathrooms_full is not None: self.bathrooms = self.bathrooms_full + 0.5 * (self.bathrooms_half or 0) else: try: self.bathrooms = float(self.bathrooms) except (TypeError, ValueError): self.bathrooms = None # free-text extraction fallback text = " ".join(filter(None, (self.title, self.description, " ".join(self.features)))) if self.bedrooms is None or self.bathrooms is None: beds, baths = extract_beds_baths(text) if self.bedrooms is None: self.bedrooms = beds if self.bathrooms is None: self.bathrooms = baths if self.living_area_sqft is None: self.living_area_sqft = parse_area_sqft(text) # office fallback: office/agent name when brokerage is missing if not self.brokerage_name: self.brokerage_name = self.office_name or self.agent_name # source coordinates: reject anything outside the covered territory — # contiguous US + Alaska + Hawaii (swapped lat/lng, 0/0, typos) if self.lat is not None and self.lng is not None: ok = (18.5 <= self.lat <= 71.5 and -180.0 <= self.lng <= -66.0) if not ok: self.lat = self.lng = None return self