"""Per-field source priority for cross-source canonical merge. Direct port of Cross_Source_Matching_Strategy.md sec 3.5 (houses) + 4.5 (listings). Source-of-truth dicts read by merge logic in match_or_create_house/listing. Sources covered: avito (serp/detail/houses_catalog/domoteka/imv), cian (serp/bti/detail/stats/valuation), yandex (serp/detail/realty_nb/valuation). """ from __future__ import annotations import logging from typing import Any logger = logging.getLogger(__name__) # cross_validate: warn if spread between sources exceeds this fraction of the median. _CROSS_VALIDATE_DIVERGENCE_THRESHOLD = 0.10 # --------------------------------------------------------------------------- # HOUSE_FIELD_PRIORITY — vault sec 3.5 # --------------------------------------------------------------------------- HOUSE_FIELD_PRIORITY: dict[str, list[str] | str] = { "address": ["cian", "avito", "yandex"], "lat": ["cian_serp", "avito_houses_catalog", "yandex_realty_nb"], "lon": ["cian_serp", "avito_houses_catalog", "yandex_realty_nb"], "year_built": [ "cian_bti", "cian_serp", "avito_houses_catalog", "yandex_valuation", "yandex_realty_nb", ], "house_type": ["cian_bti", "cian_serp", "avito", "yandex_valuation", "yandex_realty_nb"], "series_name": ["cian_bti"], "passenger_lifts_count": ["cian", "avito"], "cargo_lifts_count": ["cian", "avito"], "has_concierge": ["cian", "avito"], "closed_yard": ["cian", "avito"], "parking_type": ["cian"], "flat_count": ["cian_bti"], "entrances": ["cian_bti"], "is_emergency": ["cian_bti"], "management_company_id": ["cian_valuation"], "house_class": ["avito_houses_catalog", "cian", "yandex_realty_nb"], "rating_score": ["avito_houses_catalog", "cian", "yandex_realty_nb"], "reviews_count": ["avito_houses_catalog", "cian", "yandex_realty_nb"], # Yandex unique fields (newbuilding landing only) "text_reviews_count": ["yandex_realty_nb"], # 353 text reviews — Yandex strongpoint "corpus_count": ["yandex_realty_nb"], # "три башни" → 3 "total_area_ha": ["yandex_realty_nb"], # ЖК footprint "commission_year": ["cian_serp", "yandex_realty_nb"], "commission_month": ["yandex_realty_nb"], # raw RU month name "developer_name": ["cian", "yandex_realty_nb"], "has_panorama": ["yandex_valuation"], # Yandex 3D panorama flag "yandex_total_listings": ["yandex_valuation"], # "N объектов" в истории # Yandex Valuation enrichment (existing house attrs) "has_lift": ["cian_bti", "cian_detail", "yandex_valuation"], "ceiling_height": ["cian_detail", "yandex_valuation"], } # --------------------------------------------------------------------------- # LISTING_FIELD_PRIORITY — vault sec 4.5 # --------------------------------------------------------------------------- LISTING_FIELD_PRIORITY: dict[str, list[str] | str] = { "address": ["cian_serp", "avito_detail", "avito_serp"], "lat": ["cian_serp", "avito_detail"], "lon": ["cian_serp", "avito_detail"], "area_m2": ["cian_serp", "avito_detail"], "living_area_m2": ["cian_serp"], "kitchen_area_m2": ["cian_serp", "avito_detail"], "ceiling_height": ["cian_detail"], "floor": ["cian_serp", "avito_detail"], "total_floors": ["cian_serp", "avito_detail"], "year_built": ["cian_serp"], "house_type": ["cian_serp", "avito_detail", "yandex_detail"], "repair_state": ["cian_detail", "avito_detail"], "has_balcony": ["cian_serp", "avito_detail"], "balconies_count": ["cian_serp"], "loggias_count": ["cian_serp"], "windows_view_type": ["cian_detail", "avito_detail"], "separate_wcs_count": ["cian_detail"], "combined_wcs_count": ["cian_detail"], "room_type": ["cian_detail", "avito_detail"], "has_furniture": ["cian_serp", "avito_detail"], "phones": ["cian_serp"], "description": ["cian_serp", "avito_detail", "yandex_detail"], "photo_urls": "union", # Avito Domoteka unique "owners_count": ["avito_domoteka"], "owners_at_least": ["avito_domoteka"], "last_owner_change_date": ["avito_domoteka"], "encumbrances_clean": ["avito_domoteka"], "registry_match": ["avito_domoteka"], # Cian-only "is_rosreestr_checked": ["cian_serp"], "is_layout_approved": ["cian_serp"], "is_commercial_ownership_verified": ["cian_serp"], # Cross-validation "price_rub": "cross_validate", "kadastr_num": "first_non_null", # NOTE: task description claims price_rub=['cian','avito','yandex']; vault sec 4.5 # says 'cross_validate' (flag if diff > 10%). Following vault — change to list if # cross_validate semantics aren't desired at caller. # Yandex unique (agency block — OfferCardAuthorInfo) "agency_name": ["yandex_detail"], "agency_founded_year": ["yandex_detail"], "agency_objects_count": ["yandex_detail"], # Yandex parallel views column (NOT existing `views_total` which is Cian's) "views_total_yandex": ["yandex_detail"], # Yandex raw publish-date text (relative form) "publish_date_relative": ["yandex_detail"], } def _freshest_non_null( candidates: dict[str, Any], timestamps: dict[str, Any] | None, ) -> Any: """Pick the non-null candidate from the freshest source (max last_seen_at). Deterministic replacement for dict-insertion-order ``first_non_null`` (#1539). Among sources with a non-null value: * if ``timestamps`` is provided, choose the value whose source has the most recent ``last_seen_at``; ties (equal / missing timestamps) break by source name (lexicographic) so the result is reproducible regardless of dict order; * if no usable timestamps are available, fall back to first non-null in sorted-key order (still deterministic, unlike raw insertion order). """ non_null = {src: v for src, v in candidates.items() if v is not None} if not non_null: return None timestamps = timestamps or {} timed = [src for src in non_null if timestamps.get(src) is not None] if timed: # Freshest last_seen_at wins; equal timestamps tie-break by source name # (lexicographically smallest) so the result never depends on dict order. latest = max(timestamps[src] for src in timed) winner = min(src for src in timed if timestamps[src] == latest) return non_null[winner] # No timestamps anywhere: deterministic by sorted source name. return non_null[min(non_null)] def _resolve( priority: dict[str, list[str] | str], field: str, candidates: dict[str, Any], timestamps: dict[str, Any] | None = None, ) -> Any: """Pick value from candidates per priority dict semantics. ``candidates`` maps source name -> value for ``field``. ``timestamps`` maps source name -> ``last_seen_at`` (any comparable, usually ``datetime``); when supplied it makes conflict resolution prefer the freshest source instead of relying on non-deterministic dict insertion order (#1539). """ if not candidates: return None rule = priority.get(field, "first_non_null") if isinstance(rule, str): if rule == "union": out: list[Any] = [] for v in candidates.values(): if v is None: continue if isinstance(v, list | tuple | set): out.extend(v) else: out.append(v) # Preserve order, dedup seen: set[Any] = set() uniq: list[Any] = [] for x in out: key = repr(x) if key in seen: continue seen.add(key) uniq.append(x) return uniq if rule == "first_non_null": return _freshest_non_null(candidates, timestamps) if rule == "max": non_null = [v for v in candidates.values() if v is not None] return max(non_null) if non_null else None if rule == "min": non_null = [v for v in candidates.values() if v is not None] return min(non_null) if non_null else None if rule == "cross_validate": # Return the median; flag divergence (vault sec 4.5: warn if diff > 10%). nums = [v for v in candidates.values() if isinstance(v, int | float)] if not nums: return None nums.sort() n = len(nums) if n % 2: median = nums[n // 2] else: # True median for even n: mean of the two central values. median = (nums[n // 2 - 1] + nums[n // 2]) / 2 if median and (max(nums) - min(nums)) > _CROSS_VALIDATE_DIVERGENCE_THRESHOLD * abs( median ): logger.warning( "cross_validate divergence for field %r: min=%s max=%s median=%s " "(spread exceeds %.0f%% of median); candidates=%r", field, min(nums), max(nums), median, _CROSS_VALIDATE_DIVERGENCE_THRESHOLD * 100, candidates, ) return median # Unknown rule — same deterministic freshness fallback as first_non_null. return _freshest_non_null(candidates, timestamps) # Priority list: pick value from highest-ranked source present (non-null). # Explicit per-field ranking is authoritative and beats freshness. for src in rule: if src in candidates and candidates[src] is not None: return candidates[src] # Fallback: no ranked source present → freshest non-null (deterministic). return _freshest_non_null(candidates, timestamps) def resolve_house_field( field: str, candidates: dict[str, Any], timestamps: dict[str, Any] | None = None, ) -> Any: """Pick canonical house field value per HOUSE_FIELD_PRIORITY. ``timestamps`` (source -> last_seen_at) makes first_non_null / fallback resolution prefer the freshest source deterministically (#1539). """ return _resolve(HOUSE_FIELD_PRIORITY, field, candidates, timestamps) def resolve_listing_field( field: str, candidates: dict[str, Any], timestamps: dict[str, Any] | None = None, ) -> Any: """Pick canonical listing field value per LISTING_FIELD_PRIORITY. ``timestamps`` (source -> last_seen_at) makes first_non_null / fallback resolution prefer the freshest source deterministically (#1539). """ return _resolve(LISTING_FIELD_PRIORITY, field, candidates, timestamps) # --------------------------------------------------------------------------- # Legacy stub — kept for backward compat with existing __init__.py and tests # --------------------------------------------------------------------------- def update_canonical_fields( db: Any, listing_id: int, ext_source: str, lot_data: object, ) -> None: """Legacy Stage 8 v1 stub — full arbitration deferred to Stage 8.x.""" pass