"""Transform DOM.RIA raw data into a data_collector payload.""" from typing import Any, Dict, List, Optional def _first(val): """Helper: return first element if list, else the value itself.""" if isinstance(val, list) and val: return val[0] return val def _safe_int(val) -> Optional[int]: try: return int(float(str(val).replace(" ", "").replace(",", "."))) except (ValueError, TypeError): return None def _safe_float(val) -> Optional[float]: try: return float(str(val).replace(" ", "").replace(",", ".")) except (ValueError, TypeError): return None def _extract_photos(raw_photos) -> List[Dict]: """Normalise DOM.RIA photo field to a list of dicts. Sometimes photos come as a list, sometimes as a dict keyed by ordering. """ if isinstance(raw_photos, list): return raw_photos if isinstance(raw_photos, dict): return list(raw_photos.values()) return [] def _build_photo_urls(photos: List[Dict]) -> List[str]: """Turn photo metadata into full HTTPS URLs. DOM.RIA photos are served from cdn.riastatic.com. We can inject 'xl' before the extension for a larger variant. """ urls: List[str] = [] for p in photos: if not isinstance(p, dict): continue base = ( p.get("file") or p.get("beautifulUrl") or p.get("photo_base_url") or p.get("url") or p.get("src") ) if not base: continue if base.startswith("//"): base = "https:" + base elif base.startswith("/"): base = "https://cdn.riastatic.com" + base elif not base.startswith("http"): base = "https://cdn.riastatic.com/" + base # xl variant if plain .jpg if base.endswith(".jpg") and "xl" not in base.split("/")[-1]: base = base.replace(".jpg", "xl.jpg") urls.append(base) return urls def normalize_listing( catalog_item: Dict[str, Any], detail_realty: Optional[Dict[str, Any]] = None, city_name_meta: str = "", ) -> Dict[str, Any]: """Create the payload dict that will be sent to data_collector. Strategy: * Simple / stable fields → normalized scalar values. * Complex / nested / volatile fields → kept raw under `raw_domria`. """ c = catalog_item # shorthand d = detail_realty or {} # --- normalized simple fields ------------------------------------------------ price_usd = _safe_int(c.get("price_USD")) or _safe_int(c.get("price_usd")) # Try detail priceObj first (most explicit) if price_usd is None: price_usd = _safe_int(d.get("priceObj", {}).get("priceUSD")) # Try catalog priceArr (dict with keys 1/2/3 or list of objects) if price_usd is None: pa = c.get("priceArr") or d.get("priceArr") if isinstance(pa, dict): # DOM.RIA sends {'1': '130 000', '2': '112 317', '3': '5 863 000'} # Key '1' is USD in all observed cases price_usd = _safe_int(pa.get("1")) elif isinstance(pa, list): for entry in pa: if isinstance(entry, dict) and entry.get("currency") == "USD": price_usd = _safe_int(entry.get("price")) break # Fallback to plain price field (usually already in USD on sale pages) if price_usd is None: price_usd = _safe_int(c.get("price")) # --- photos ------------------------------------------------------------------ raw_catalog_photos = _extract_photos(c.get("photos")) raw_detail_photos = _extract_photos(d.get("photos")) # Prefer detail photos (usually more / higher quality), fallback to catalog photo_source = raw_detail_photos if raw_detail_photos else raw_catalog_photos photo_urls = _build_photo_urls(photo_source) if not photo_urls: # Last resort: detail sometimes has a single `main_photo` dict main = d.get("main_photo") if isinstance(main, dict): photo_urls = _build_photo_urls([main]) normalized = { "external_id": str(c.get("realty_id")), "url": f"https://dom.ria.com/uk/{(c.get('beautiful_url') or '').lstrip('/')}", "title": (d.get("title") or c.get("title") or "").strip(), "description": (d.get("description") or c.get("description") or "").strip(), "price_usd": price_usd, "price_raw": c.get("priceArr") or c.get("price"), "area_total_m2": _safe_float(c.get("total_square_meters")), "area_living_m2": _safe_float( d.get("living_square_meters") or c.get("living_square_meters") ), "area_kitchen_m2": _safe_float( d.get("kitchen_square_meters") or c.get("kitchen_square_meters") ), "rooms_count": _safe_int(c.get("rooms_count")), "floor": _safe_int(c.get("floor")), "floors_total": _safe_int( d.get("floors_count") or c.get("floors_count") ), "city_name": (d.get("city_name") or c.get("city_name") or city_name_meta or "").strip(), "district_name": (d.get("district_name") or c.get("district_name") or "").strip(), "street_name": (d.get("street_name") or c.get("street_name") or "").strip(), "building_number": (d.get("building_number") or c.get("building_number") or "").strip(), "latitude": _safe_float(c.get("lat")), "longitude": _safe_float(c.get("lng")), "photos": photo_urls, "photos_count": len(photo_urls), "tags": [t.get("name_uk") or t.get("name") for t in (d.get("tags") or c.get("tags") or []) if isinstance(t, dict)], "contact_phones": _extract_phones(d), "raw_domria": { "catalog_item": c, "detail_realty": d, }, } # Remove None values to keep payload compact cleaned = {k: v for k, v in normalized.items() if v is not None} return cleaned def _extract_phones(detail: Dict[str, Any]) -> List[str]: """Pull phone numbers from the detail object if available.""" phones: List[str] = [] user = detail.get("user") user_contacts = user.get("contacts") if isinstance(user, dict) else None contacts = detail.get("contacts") or user_contacts if not isinstance(contacts, dict): return phones for key in ("phones", "phone", "mobile", "tel"): vals = contacts.get(key) if isinstance(vals, list): for v in vals: if isinstance(v, dict): phones.append(v.get("phone") or v.get("number") or str(v)) elif isinstance(v, str): phones.append(v) elif isinstance(vals, str): phones.append(vals) # dedupe preserving order seen = set() uniq = [] for p in phones: p = str(p).strip() if p and p not in seen: seen.add(p) uniq.append(p) return uniq