tradein-mvp/backend/app/services/matching/conflict_resolution.py

@@ -0,0 +1,196 @@
+"""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
+
+from typing import Any
+
+# ---------------------------------------------------------------------------
+# 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"],
+
+    # Yandex raw RU sale-type phrase (vs existing `sale_type` enum)
+    "sale_type_text":        ["yandex_detail"],
+}
+
+
+def _resolve(
+    priority: dict[str, list[str] | str],
+    field: str,
+    candidates: dict[str, Any],
+) -> Any:
+    """Pick value from candidates per priority dict semantics."""
+    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":
+            for v in candidates.values():
+                if v is not None:
+                    return v
+            return None
+        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":
+            # Caller decides — return median by default
+            nums = [v for v in candidates.values() if isinstance(v, (int, float))]
+            if not nums:
+                return None
+            nums.sort()
+            return nums[len(nums) // 2]
+        # Unknown rule
+        return next((v for v in candidates.values() if v is not None), None)
+
+    # Priority list: pick value from highest-ranked source present (non-null)
+    for src in rule:
+        if src in candidates and candidates[src] is not None:
+            return candidates[src]
+    # Fallback: any non-null
+    return next((v for v in candidates.values() if v is not None), None)
+
+
+def resolve_house_field(field: str, candidates: dict[str, Any]) -> Any:
+    """Pick canonical house field value per HOUSE_FIELD_PRIORITY."""
+    return _resolve(HOUSE_FIELD_PRIORITY, field, candidates)
+
+
+def resolve_listing_field(field: str, candidates: dict[str, Any]) -> Any:
+    """Pick canonical listing field value per LISTING_FIELD_PRIORITY."""
+    return _resolve(LISTING_FIELD_PRIORITY, field, candidates)
+
+
+# ---------------------------------------------------------------------------
+# 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

tradein-mvp/backend/tests/matching/test_conflict_resolution.py

@@ -0,0 +1,230 @@
+"""Per-field priority resolution tests."""
+from app.services.matching.conflict_resolution import (
+    HOUSE_FIELD_PRIORITY,
+    LISTING_FIELD_PRIORITY,
+    resolve_house_field,
+    resolve_listing_field,
+)
+
+# ---------------------------------------------------------------------------
+# Pre-existing tests (preserved)
+# ---------------------------------------------------------------------------
+
+def test_house_year_built_prefers_cian_bti() -> None:
+    out = resolve_house_field(
+        "year_built",
+        {"avito_houses_catalog": 2020, "cian_bti": 2021, "cian_serp": 2022},
+    )
+    assert out == 2021
+
+
+def test_house_unknown_field_first_non_null() -> None:
+    assert resolve_house_field("totally_unknown", {"x": None, "y": 5}) == 5
+
+
+def test_listing_owners_count_avito_domoteka_only() -> None:
+    out = resolve_listing_field("owners_count", {"avito_domoteka": 2, "cian_serp": 99})
+    assert out == 2
+
+
+def test_listing_photo_urls_union() -> None:
+    out = resolve_listing_field(
+        "photo_urls", {"avito": ["a.jpg", "b.jpg"], "cian": ["b.jpg", "c.jpg"]},
+    )
+    assert set(out) == {"a.jpg", "b.jpg", "c.jpg"}
+
+
+def test_listing_kadastr_first_non_null() -> None:
+    out = resolve_listing_field("kadastr_num", {"avito": None, "cian": "66:1:1:1"})
+    assert out == "66:1:1:1"
+
+
+def test_house_priority_dict_has_year_built() -> None:
+    assert "cian_bti" in HOUSE_FIELD_PRIORITY["year_built"]
+
+
+def test_listing_priority_dict_has_owners() -> None:
+    assert "avito_domoteka" in LISTING_FIELD_PRIORITY["owners_count"]
+
+
+# ---------------------------------------------------------------------------
+# Yandex house priority tests
+# ---------------------------------------------------------------------------
+
+class TestYandexHousePriority:
+    def test_house_lat_yandex_realty_nb_when_cian_missing(self) -> None:
+        out = resolve_house_field("lat", {"yandex_realty_nb": 56.85})
+        assert out == 56.85
+
+    def test_house_lat_cian_preferred_over_yandex(self) -> None:
+        out = resolve_house_field(
+            "lat", {"cian_serp": 56.83, "yandex_realty_nb": 56.85}
+        )
+        assert out == 56.83
+
+    def test_house_year_built_yandex_valuation_picked(self) -> None:
+        out = resolve_house_field("year_built", {"yandex_valuation": 1981})
+        assert out == 1981
+
+    def test_house_year_built_cian_bti_preferred(self) -> None:
+        out = resolve_house_field(
+            "year_built", {"cian_bti": 1980, "yandex_valuation": 1981}
+        )
+        assert out == 1980
+
+    def test_house_text_reviews_count_yandex_only(self) -> None:
+        out = resolve_house_field("text_reviews_count", {"yandex_realty_nb": 353})
+        assert out == 353
+
+    def test_house_corpus_count_yandex_only(self) -> None:
+        out = resolve_house_field("corpus_count", {"yandex_realty_nb": 3})
+        assert out == 3
+
+    def test_house_commission_year_cian_serp_preferred(self) -> None:
+        out = resolve_house_field(
+            "commission_year", {"cian_serp": 2022, "yandex_realty_nb": 2023}
+        )
+        assert out == 2022
+
+    def test_house_commission_month_yandex_only(self) -> None:
+        out = resolve_house_field("commission_month", {"yandex_realty_nb": "июнь"})
+        assert out == "июнь"
+
+    def test_house_developer_name_cian_preferred(self) -> None:
+        out = resolve_house_field(
+            "developer_name",
+            {"cian": "PRINZIP", "yandex_realty_nb": "PRINZIP недвижимость"},
+        )
+        assert out == "PRINZIP"
+
+    def test_house_has_lift_cian_bti_preferred_over_yandex_valuation(self) -> None:
+        out = resolve_house_field(
+            "has_lift", {"cian_bti": True, "yandex_valuation": True}
+        )
+        assert out is True
+
+    def test_house_ceiling_height_cian_detail_preferred(self) -> None:
+        out = resolve_house_field(
+            "ceiling_height", {"cian_detail": 2.7, "yandex_valuation": 2.5}
+        )
+        assert out == 2.7
+
+    def test_house_has_panorama_yandex_valuation_only(self) -> None:
+        out = resolve_house_field("has_panorama", {"yandex_valuation": True})
+        assert out is True
+
+    def test_house_yandex_total_listings_yandex_valuation_only(self) -> None:
+        out = resolve_house_field("yandex_total_listings", {"yandex_valuation": 42})
+        assert out == 42
+
+    def test_house_house_class_yandex_realty_nb_fallback(self) -> None:
+        out = resolve_house_field("house_class", {"yandex_realty_nb": "бизнес"})
+        assert out == "бизнес"
+
+    def test_house_house_class_avito_preferred_over_yandex(self) -> None:
+        out = resolve_house_field(
+            "house_class",
+            {"avito_houses_catalog": "комфорт", "yandex_realty_nb": "бизнес"},
+        )
+        assert out == "комфорт"
+
+    def test_house_total_area_ha_yandex_only(self) -> None:
+        out = resolve_house_field("total_area_ha", {"yandex_realty_nb": 12.5})
+        assert out == 12.5
+
+    def test_house_has_lift_yandex_valuation_fallback_when_cian_missing(self) -> None:
+        out = resolve_house_field("has_lift", {"yandex_valuation": False})
+        assert out is False
+
+
+# ---------------------------------------------------------------------------
+# Yandex listing priority tests
+# ---------------------------------------------------------------------------
+
+class TestYandexListingPriority:
+    def test_listing_description_cian_preferred_over_yandex_detail(self) -> None:
+        out = resolve_listing_field(
+            "description",
+            {"cian_serp": "cian text", "yandex_detail": "yandex text"},
+        )
+        assert out == "cian text"
+
+    def test_listing_house_type_yandex_detail_used_when_cian_avito_missing(self) -> None:
+        out = resolve_listing_field("house_type", {"yandex_detail": "brick"})
+        assert out == "brick"
+
+    def test_listing_agency_name_yandex_only(self) -> None:
+        out = resolve_listing_field(
+            "agency_name", {"yandex_detail": "Агентство «Диал»"}
+        )
+        assert out == "Агентство «Диал»"
+
+    def test_listing_agency_founded_year_yandex_only(self) -> None:
+        out = resolve_listing_field("agency_founded_year", {"yandex_detail": 2005})
+        assert out == 2005
+
+    def test_listing_agency_objects_count_yandex_only(self) -> None:
+        out = resolve_listing_field("agency_objects_count", {"yandex_detail": 120})
+        assert out == 120
+
+    def test_listing_views_total_yandex_only(self) -> None:
+        out = resolve_listing_field("views_total_yandex", {"yandex_detail": 874})
+        assert out == 874
+
+    def test_listing_publish_date_relative_yandex_only(self) -> None:
+        out = resolve_listing_field(
+            "publish_date_relative", {"yandex_detail": "3 дня назад"}
+        )
+        assert out == "3 дня назад"
+
+    def test_listing_sale_type_text_yandex_only(self) -> None:
+        out = resolve_listing_field(
+            "sale_type_text", {"yandex_detail": "Прямая продажа"}
+        )
+        assert out == "Прямая продажа"
+
+    def test_listing_agency_name_none_when_no_yandex(self) -> None:
+        out = resolve_listing_field("agency_name", {"cian_serp": None})
+        assert out is None
+
+    def test_listing_description_avito_detail_preferred_over_yandex(self) -> None:
+        out = resolve_listing_field(
+            "description",
+            {"avito_detail": "avito desc", "yandex_detail": "yandex desc"},
+        )
+        assert out == "avito desc"
+
+    def test_listing_house_type_cian_preferred_over_yandex_detail(self) -> None:
+        out = resolve_listing_field(
+            "house_type",
+            {"cian_serp": "панель", "yandex_detail": "кирпич"},
+        )
+        assert out == "панель"
+
+    def test_listing_yandex_detail_keys_registered(self) -> None:
+        """Ensure all Yandex-unique listing keys are in priority dict."""
+        yandex_keys = [
+            "agency_name",
+            "agency_founded_year",
+            "agency_objects_count",
+            "views_total_yandex",
+            "publish_date_relative",
+            "sale_type_text",
+        ]
+        for key in yandex_keys:
+            assert key in LISTING_FIELD_PRIORITY, f"{key!r} missing from LISTING_FIELD_PRIORITY"
+            rule = LISTING_FIELD_PRIORITY[key]
+            assert rule == ["yandex_detail"], f"{key!r} rule mismatch: {rule!r}"
+
+    def test_house_yandex_keys_registered(self) -> None:
+        """Ensure all Yandex-unique house keys are in priority dict."""
+        yandex_keys = [
+            "text_reviews_count",
+            "corpus_count",
+            "total_area_ha",
+            "commission_month",
+            "has_panorama",
+            "yandex_total_listings",
+        ]
+        for key in yandex_keys:
+            assert key in HOUSE_FIELD_PRIORITY, f"{key!r} missing from HOUSE_FIELD_PRIORITY"