# ----------------------------------------------------------------------------- # Rent-Ka — Rental listings aggregator (Canada, outside Québec) # Author: Simon-Pierre Boucher — contact@spboucher.ai # schema.py : standardized data model (Listing) + central enrichment # ----------------------------------------------------------------------------- """Standard listing schema and field normalization. Every connector, whatever its source site, must produce `Listing` objects that conform to this schema. `finalize()` then applies the common normalization layer (rentka/normalize.py): ISO dates, sqft areas, canonical unit types, prices, ENGLISH display labels — connectors can stay simple and fill raw fields (French labels from fr-ca templates included). """ from __future__ import annotations import hashlib import json from dataclasses import dataclass, field, asdict from .normalize import ( # ré-exportés pour les connecteurs existants bedrooms_from_unit_type, clean_address, clean_city, clean_sector, coerce_count, extract_details, merge_details, normalize_unit_type, parse_area_sqft, parse_availability_date, parse_bathrooms, parse_bedrooms, parse_price, price_is_from, strip_accents, ) __all__ = [ "Listing", "infer_city", "normalize_unit_type", "parse_price", "parse_availability_date", "parse_area_sqft", "clean_address", "strip_accents", ] @dataclass class Listing: """Standardized Rent-Ka listing.""" source: str # source id (see data/sources.json) external_id: str # identifier at the source url: str # listing page at the source title: str = "" # e.g. "200 Bay Street — 2 bedrooms" address: str = "" # civic address sector: str = "" # neighbourhood/borough (e.g. The Annex) city: str = "" # Toronto, Calgary, Halifax… province: str = "ON" # ON | BC | AB | SK | MB | NB | NS | PE | NL | YT | NT | NU unit_type: str = "" # Studio, n bedroom(s), Loft, Condo, House, Room… bedrooms: float | None = None # closed bedrooms bathrooms: float | None = None # bathrooms (1.5 = extra powder room) price: float | None = None # monthly rent ($ CAD), lowest when "from" price_label: str = "" # display text (e.g. "From $799/month") availability: str = "" # display text (e.g. "Available now") availability_date: str | None = None # ISO "2026-07-01", "now", or None area_sqft: float | None = None # area in sq ft pets: str | None = None # "yes" | "no" | "conditions" | None furnished: bool | None = None # furnished (None = unknown) description: str = "" digest: dict | None = None # structured description (rentka/textmine.py) amenities: list[str] = field(default_factory=list) # source text, for display 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: """Hash du contenu pour la détection de changements (pseudo-webhook).""" payload = asdict(self) blob = json.dumps(payload, sort_keys=True, ensure_ascii=False) return hashlib.sha256(blob.encode("utf-8")).hexdigest() def finalize(self) -> "Listing": """Applique la normalisation commune. Appelé par le pipeline d'ingestion. Idempotent ; ne remplace jamais une valeur explicite du connecteur. """ import html as _html from .normalize import translate_label_en self.title = _html.unescape(self.title).strip() self.address = clean_address(_html.unescape(self.address)) self.sector = clean_sector(self.sector) self.city = clean_city(self.city) self.unit_type = normalize_unit_type(self.unit_type) # English display labels (central FR -> EN funnel — many connectors # read fr-ca page templates; unknown strings pass through untouched) self.amenities = [translate_label_en(a) for a in self.amenities] self.availability = translate_label_en(self.availability) self.price_label = translate_label_en(self.price_label) if self.pets in ("oui", "non"): self.pets = "yes" if self.pets == "oui" else "no" if self.price is None: self.price = parse_price(self.price_label) if self.availability_date is None: self.availability_date = parse_availability_date(self.availability) if self.area_sqft is None: for texte in (self.description, " ".join(self.amenities), self.title): self.area_sqft = parse_area_sqft(texte) if self.area_sqft is not None: break # source-provided coordinates: reject any point outside the covered # territory (swapped lat/lng, 0/0, typos) — the geocoder takes over # from the address rather than showing a pin in Kazakhstan. if self.lat is not None and self.lng is not None: prov = (self.province or "ON").upper() lo_lat, hi_lat, lo_lng, hi_lng = PROVINCE_BBOX.get( prov, CANADA_BBOX) ok = lo_lat <= self.lat <= hi_lat and lo_lng <= self.lng <= hi_lng if not ok: self.lat = self.lng = None derived = extract_details(self.amenities, self.description, self.title) if self.price_label and price_is_from(self.price_label): derived["price_from"] = True self.details = merge_details(self.details, derived) # chambres / salles de bain : valeur explicite du connecteur d'abord, # puis champs structurés de details, puis type d'unité (n½ -> n-2), # puis extraction texte — jamais de valeur inventée (None = inconnu) if self.bedrooms is None: for k in ("bedrooms", "Chambres", "Chambre(s)"): self.bedrooms = coerce_count(self.details.get(k)) if self.bedrooms is not None: break if self.bedrooms is None: self.bedrooms = bedrooms_from_unit_type(self.unit_type) if self.bedrooms is None: self.bedrooms = parse_bedrooms(self.title, " | ".join(self.amenities), self.description) if self.bathrooms is None: for k in ("bathrooms", "Salles de bain", "Salle de bain", "Salle(s) de bain"): self.bathrooms = coerce_count(self.details.get(k)) if self.bathrooms is not None: break if self.bathrooms is None: self.bathrooms = parse_bathrooms(" | ".join(self.amenities), self.description) # description structurée (nettoyage + extraction + sections) if self.digest is None and self.description: try: from .textmine import analyser self.digest = analyser(self.description, price=self.price, sector=self.sector, city=self.city) except ImportError: pass # module absent : la fiche affichera le texte brut if self.pets is None: self.pets = self.details.get("pets") else: self.details["pets"] = self.pets if self.furnished is None: furn = self.details.get("furnished") self.furnished = furn if isinstance(furn, bool) else None else: self.details["furnished"] = self.furnished return self # --------------------------------------------------------------------------- # Geography — coordinate sanity boxes per province (Canada, outside Québec) # --------------------------------------------------------------------------- # (min_lat, max_lat, min_lng, max_lng) CANADA_BBOX = (41.6, 83.2, -141.0, -52.5) PROVINCE_BBOX = { "BC": (48.2, 60.0, -139.1, -114.0), "AB": (48.9, 60.0, -120.0, -109.9), "SK": (48.9, 60.0, -110.1, -101.3), "MB": (48.9, 60.0, -102.1, -88.9), "ON": (41.6, 56.9, -95.3, -74.3), "NB": (44.5, 48.1, -69.1, -63.7), "NS": (43.3, 47.1, -66.5, -59.6), "PE": (45.9, 47.1, -64.5, -61.9), "NL": (46.5, 60.5, -67.9, -52.5), "YT": (60.0, 69.7, -141.1, -123.7), "NT": (60.0, 78.9, -136.5, -101.9), "NU": (60.0, 83.2, -120.7, -61.0), # legacy rows / stray input — Québec box kept only for bbox sanity "QC": (44.5, 63.0, -80.0, -56.0), } def infer_city(sector: str, default: str = "") -> str: """Legacy helper (Québec-era): now a pass-through to the default.""" return default