tradein-mvp/backend/app/api/v1/admin.py

@@ -1865,7 +1865,7 @@ async def scrape_cian_detail(
 
     saved = False
     if listing_id is not None:
-        save_detail_enrichment(db, listing_id, enrichment)
+        save_detail_enrichment(db, listing_id, enrichment, matcher=RealMatcherAdapter())
         saved = True
 
     return CianDetailTriggerResp(

tradein-mvp/backend/app/services/cian_price_history.py

@@ -26,7 +26,7 @@ from scraper_kit.providers.cian.detail import fetch_detail, save_detail_enrichme
 from sqlalchemy import text
 from sqlalchemy.orm import Session
 
-from app.services.scraper_adapters import RealScraperConfig
+from app.services.scraper_adapters import RealMatcherAdapter, RealScraperConfig
 from app.services.scraper_settings import get_scraper_delay
 
 logger = logging.getLogger(__name__)
@@ -147,7 +147,9 @@ async def backfill_cian_price_history(
                     ),
                     {"lid": lid},
                 ).scalar_one()
-                save_detail_enrichment(db, lid, enrichment)
+                # matcher (#2435): резолвит канонический дом из bti_data (address/geo
+                # листинга) — иначе BTI-снапшот распарсен, но выбрасывается.
+                save_detail_enrichment(db, lid, enrichment, matcher=RealMatcherAdapter())
                 after = db.execute(
                     text(
                         "SELECT COUNT(*) FROM offer_price_history "

tradein-mvp/backend/app/tasks/cian_history_backfill.py

@@ -34,7 +34,7 @@ from sqlalchemy import text
 from sqlalchemy.orm import Session
 
 from app.core.config import settings
-from app.services.scraper_adapters import RealScraperConfig
+from app.services.scraper_adapters import RealMatcherAdapter, RealScraperConfig
 from app.services.scraper_settings import get_scraper_delay
 
 logger = logging.getLogger(__name__)
@@ -146,7 +146,11 @@ async def backfill_cian_history(
                         continue
 
                     try:
-                        save_detail_enrichment(db, listing_id, enrichment)
+                        # matcher (#2435): резолвит канонический дом из bti_data (address/geo
+                        # листинга) — иначе BTI-снапшот распарсен, но выбрасывается.
+                        save_detail_enrichment(
+                            db, listing_id, enrichment, matcher=RealMatcherAdapter()
+                        )
                         result.listings_succeeded += 1
                         result.price_changes_attempted += len(enrichment.price_changes or [])
                     except Exception as exc:

tradein-mvp/backend/tests/test_cian_bti_house_persist.py

@@ -0,0 +1,412 @@
+"""Тесты для Cian BTI / Valuation house-persist (#2435).
+
+`bti_data` (cian/detail.py) и valuation `house_info`/`managementCompany`
+(cian/valuation.py) парсились, но раньше выбрасывались («это задача Stage 6
+(houses), где есть house_id») — теперь пишутся в канонический houses-ряд:
+
+- detail.py: резолвит дом через инжектируемый `matcher: HouseMatcher`
+  (`match_or_create_house`, address/geo листинга) — mirror avito's `_persist_house`.
+- valuation.py: `house_id` уже резолвлен ВЫЗЫВАЮЩИМ (estimator.py, read-only
+  `match_house_readonly`) — здесь только COALESCE-UPDATE, без создания дома.
+
+MagicMock db — без реальной БД, тот же паттерн, что test_snapshot_writer.py /
+test_extval_house_id_write_path.py.
+"""
+
+from __future__ import annotations
+
+import os
+
+# Settings требует DATABASE_URL на import (см. test_extval_house_id_write_path.py).
+os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost/test_db")
+
+from contextlib import contextmanager
+from unittest.mock import MagicMock
+
+from scraper_kit.providers.cian.detail import DetailEnrichment, save_detail_enrichment
+from scraper_kit.providers.cian.valuation import CianValuationResult, _save_to_cache
+
+# ── Part 1: bti_data → houses (detail.py) ────────────────────────────────────
+
+_BTI_SAMPLE = {
+    "yearRelease": 1961,
+    "houseMaterialType": "brick",
+    "floorMax": 4,
+    "entrances": 4,
+    "flatCount": 58,
+    "isEmergency": False,
+    "houseHeatSupplyType": "central",
+    "houseGasSupplyType": "central",
+    "houseOverlapType": "concrete",
+    "lifts": 0,
+    "seriesName": "1-447",
+}
+
+
+def _mock_matcher(house_id: int | None = 501, method: str = "new") -> MagicMock:
+    matcher = MagicMock()
+    matcher.match_or_create_house.return_value = (house_id, 1.0, method)
+    return matcher
+
+
+def _mock_db_bti(
+    address: str | None = "Екатеринбург, улица Малышева, 51",
+    lat: float = 56.83,
+    lon: float = 60.6,
+) -> MagicMock:
+    """Mock db: SELECT address/lat/lon FROM listings → mappings().first() dict."""
+    db = MagicMock()
+
+    @contextmanager
+    def _nested():
+        yield MagicMock()
+
+    db.begin_nested.side_effect = _nested
+
+    def _exec(sql, params=None):
+        sql_str = str(sql)
+        mock_result = MagicMock()
+        mock_result.fetchone.return_value = None
+        mock_result.scalar_one_or_none.return_value = None
+        if "SELECT address, lat, lon FROM listings" in sql_str:
+            mock_result.mappings.return_value.first.return_value = {
+                "address": address,
+                "lat": lat,
+                "lon": lon,
+            }
+        else:
+            mock_result.mappings.return_value.first.return_value = None
+        return mock_result
+
+    db.execute.side_effect = _exec
+    return db
+
+
+def _bti_update_call(db: MagicMock) -> tuple[str, dict] | None:
+    for call_args in db.execute.call_args_list:
+        if not call_args.args:
+            continue
+        sql = str(call_args.args[0])
+        if "UPDATE houses" in sql and "series_name" in sql:
+            params = call_args.args[1] if len(call_args.args) >= 2 else {}
+            return sql, params
+    return None
+
+
+def test_bti_data_present_updates_house_columns_via_coalesce():
+    """bti_data + matcher → UPDATE houses SET series_name=COALESCE(...) с BTI-полями."""
+    db = _mock_db_bti()
+    matcher = _mock_matcher(house_id=501)
+    enrichment = DetailEnrichment(bti_data=dict(_BTI_SAMPLE))
+
+    save_detail_enrichment(db, 123, enrichment, matcher=matcher)
+
+    matcher.match_or_create_house.assert_called_once()
+    kwargs = matcher.match_or_create_house.call_args.kwargs
+    assert kwargs["ext_source"] == "cian_bti"
+    assert kwargs["ext_id"] == "123"
+    assert kwargs["address"] == "Екатеринбург, улица Малышева, 51"
+
+    call = _bti_update_call(db)
+    assert call is not None, "UPDATE houses с BTI-полями не найден"
+    sql, params = call
+    assert "COALESCE" in sql
+    assert params["house_id"] == 501
+    assert params["series_name"] == "1-447"
+    assert params["entrances"] == 4
+    assert params["flat_count"] == 58
+    assert params["is_emergency"] is False
+    assert params["heat_supply_type"] == "central"
+    assert params["gas_supply_type"] == "central"
+    assert params["overlap_type"] == "concrete"
+    db.commit.assert_called_once()
+
+
+def test_bti_data_absent_is_noop_no_crash():
+    """bti_data=None (default) → matcher не вызывается, UPDATE houses не выполняется."""
+    db = _mock_db_bti()
+    matcher = _mock_matcher()
+    enrichment = DetailEnrichment()  # bti_data=None
+
+    save_detail_enrichment(db, 124, enrichment, matcher=matcher)
+
+    matcher.match_or_create_house.assert_not_called()
+    assert _bti_update_call(db) is None
+    db.commit.assert_called_once()
+
+
+def test_bti_data_present_but_no_matcher_is_noop():
+    """matcher=None (default, backward-compat для существующих вызовов) → bti_data игнорируется."""
+    db = _mock_db_bti()
+    enrichment = DetailEnrichment(bti_data=dict(_BTI_SAMPLE))
+
+    save_detail_enrichment(db, 125, enrichment)  # no matcher kwarg
+
+    assert _bti_update_call(db) is None
+    db.commit.assert_called_once()
+
+
+def test_bti_house_resolution_refused_no_crash():
+    """matcher отказал (безномерный адрес, method='no_house_number') → graceful skip."""
+    db = _mock_db_bti()
+    matcher = _mock_matcher(house_id=None, method="no_house_number")
+    enrichment = DetailEnrichment(bti_data=dict(_BTI_SAMPLE))
+
+    save_detail_enrichment(db, 126, enrichment, matcher=matcher)
+
+    assert _bti_update_call(db) is None
+    db.commit.assert_called_once()
+
+
+def test_bti_persist_exception_does_not_abort_main_save():
+    """Исключение в BTI house-persist (напр. UPDATE houses упал) не прерывает основной save."""
+    db = _mock_db_bti()
+    matcher = _mock_matcher(house_id=501)
+    enrichment = DetailEnrichment(bti_data=dict(_BTI_SAMPLE))
+
+    original_side_effect = db.execute.side_effect
+
+    def _boom(sql, params=None):
+        if "UPDATE houses" in str(sql):
+            raise RuntimeError("boom")
+        return original_side_effect(sql, params)
+
+    db.execute.side_effect = _boom
+
+    # Не должно бросить исключение наружу.
+    save_detail_enrichment(db, 127, enrichment, matcher=matcher)
+    db.commit.assert_called_once()
+
+
+def test_bti_missing_listing_address_skips_resolve():
+    """Листинг без адреса → matcher не вызывается, no crash (best-effort)."""
+    db = _mock_db_bti(address="")
+    matcher = _mock_matcher()
+    enrichment = DetailEnrichment(bti_data=dict(_BTI_SAMPLE))
+
+    save_detail_enrichment(db, 128, enrichment, matcher=matcher)
+
+    matcher.match_or_create_house.assert_not_called()
+    db.commit.assert_called_once()
+
+
+# ── Part 2: valuation house_info/managementCompany → houses (valuation.py) ──
+
+_HOUSE_INFO_SAMPLE = [
+    {"title": "Год постройки", "value": 2005},
+    {"title": "Тип дома", "value": "Кирпичный"},
+    {"title": "Этажность", "value": 16},
+    {"title": "Газоснабжение", "value": "Отсутствует"},
+    {"title": "Отопление", "value": "Центральное"},
+    {"title": "Тип перекрытий", "value": "Железобетонные"},
+    {"title": "Подъездов", "value": 1},
+    {"title": "Количество лифтов", "value": "1 пассажирский, 1 грузовой"},
+    {"title": "Мусоропровод", "value": "Нет"},
+    {"title": "Квартир", "value": 105},
+    {"title": "Реновация", "value": "Нет"},
+    {"title": "Аварийность", "value": "Нет"},
+    {"title": "Детская площадка", "value": "Нет"},
+    {"title": "Спортивная площадка", "value": "Нет"},
+]
+
+
+def _cian_result(
+    house_info: list[dict] | None = None,
+    management_company: dict | None = None,
+    external_house_id: int | None = 54016,
+) -> CianValuationResult:
+    return CianValuationResult(
+        sale_price_rub=10_000_000,
+        sale_accuracy=85.0,
+        chart=[],
+        external_house_id=external_house_id,
+        house_info=house_info or [],
+        management_company=management_company,
+    )
+
+
+def _mock_db_valuation(
+    mc_insert_id: int | None = 900, mc_existing_id: int | None = None
+) -> MagicMock:
+    """Mock db для _save_to_cache: SELECT (dedup) → UPDATE-or-INSERT management_companies,
+    затем UPDATE houses.
+
+    `mc_existing_id` simulates the dedup SELECT finding a prior row (existing.fetchone()
+    returns that id) — the UPDATE branch is taken instead of INSERT. Default None means
+    "no existing row", exercising the INSERT branch (the original test-suite behavior).
+    """
+    db = MagicMock()
+
+    @contextmanager
+    def _nested():
+        yield MagicMock()
+
+    db.begin_nested.side_effect = _nested
+
+    def _exec(sql, params=None):
+        sql_str = str(sql)
+        mock_result = MagicMock()
+        if "SELECT id FROM management_companies" in sql_str:
+            mock_result.fetchone.return_value = (
+                (mc_existing_id,) if mc_existing_id is not None else None
+            )
+        elif "INSERT INTO management_companies" in sql_str:
+            mock_result.fetchone.return_value = (
+                (mc_insert_id,) if mc_insert_id is not None else None
+            )
+        else:
+            mock_result.fetchone.return_value = None
+        return mock_result
+
+    db.execute.side_effect = _exec
+    return db
+
+
+def _mc_calls(db: MagicMock) -> list[str]:
+    """SQL statement kinds touching management_companies, in call order."""
+    kinds = []
+    for call_args in db.execute.call_args_list:
+        if not call_args.args:
+            continue
+        sql = str(call_args.args[0])
+        if "SELECT id FROM management_companies" in sql:
+            kinds.append("select")
+        elif "INSERT INTO management_companies" in sql:
+            kinds.append("insert")
+        elif "UPDATE management_companies" in sql:
+            kinds.append("update")
+    return kinds
+
+
+def _houses_update_call(db: MagicMock) -> tuple[str, dict] | None:
+    for call_args in db.execute.call_args_list:
+        if not call_args.args:
+            continue
+        sql = str(call_args.args[0])
+        if "UPDATE houses" in sql:
+            params = call_args.args[1] if len(call_args.args) >= 2 else {}
+            return sql, params
+    return None
+
+
+def _call_save_to_cache(db: MagicMock, result: CianValuationResult, house_id: int | None) -> None:
+    _save_to_cache(
+        db,
+        cache_key="ck",
+        address="Екатеринбург, улица Учителей, 18",
+        total_area=50.0,
+        rooms_count=2,
+        floor=3,
+        total_floors=16,
+        repair_type="cosmetic",
+        deal_type="sale",
+        result=result,
+        house_id=house_id,
+    )
+
+
+def test_valuation_house_info_and_management_company_land_in_houses_update():
+    """house_id задан, УК ранее не встречалась → SELECT (не найдено) → INSERT management_companies,
+    UPDATE houses с management_company_id + house_info-полями."""
+    db = _mock_db_valuation(mc_insert_id=900)
+    result = _cian_result(
+        house_info=_HOUSE_INFO_SAMPLE,
+        management_company={
+            "name": 'ТСЖ "Учителей, 18"',
+            "phones": ["+7 (343) 216-59-12"],
+            "email": "tsg.u.18@mail.ru",
+            "openingHours": [{"Пн-Пт": ["10:00-20:00"]}],
+            "chiefName": "Сабиров Наиль Вагизович",
+        },
+    )
+
+    _call_save_to_cache(db, result, house_id=42)
+
+    assert _mc_calls(db) == ["select", "insert"]
+    call = _houses_update_call(db)
+    assert call is not None
+    sql, params = call
+    assert "COALESCE" in sql
+    assert params["hid"] == 42
+    assert params["mcid"] == 900
+    assert params["cihi"] == 54016
+    assert params["yb"] == 2005
+    assert params["ht"] == "Кирпичный"
+    assert params["tf"] == 16
+    assert params["gst"] == "Отсутствует"
+    assert params["hst"] == "Центральное"
+    assert params["ot"] == "Железобетонные"
+    assert params["ent"] == 1
+    assert params["fc"] == 105
+    assert params["ie"] is False
+    assert params["hp"] is False
+    assert params["pe"] == 1
+    assert params["ce"] == 1
+    db.commit.assert_called_once()
+
+
+def test_valuation_management_company_dedups_on_repeat_call_same_name():
+    """Regression for the NULLS-DISTINCT dedup bug (#2435 review): a repeat valuation
+    call for a management company already persisted (same ext_source/name, ext_id=NULL)
+    must find it via the explicit SELECT and UPDATE it — not INSERT a duplicate row.
+    `ON CONFLICT (ext_source, name, ext_id)` alone can't dedup ext_id=NULL rows under
+    Postgres's default NULLS DISTINCT (two NULLs never compare equal), which is why
+    dedup is done via SELECT-before-INSERT rather than relying on the constraint."""
+    db = _mock_db_valuation(mc_existing_id=900)
+    result = _cian_result(
+        house_info=[],
+        management_company={"name": 'ТСЖ "Учителей, 18"', "phones": ["+7 (343) 216-59-12"]},
+    )
+
+    _call_save_to_cache(db, result, house_id=42)
+
+    assert _mc_calls(db) == ["select", "update"]
+    call = _houses_update_call(db)
+    assert call is not None
+    _, params = call
+    assert params["mcid"] == 900
+    db.commit.assert_called_once()
+
+
+def test_valuation_house_id_none_skips_house_update():
+    """house_id=None → нет UPDATE houses (best-effort no-op, тот же no-crash контракт,
+    что test_extval_house_id_write_path.py::test_cian_save_null_house_id_no_exception)."""
+    db = _mock_db_valuation()
+    result = _cian_result(house_info=_HOUSE_INFO_SAMPLE, management_company={"name": "УК Тест"})
+
+    _call_save_to_cache(db, result, house_id=None)
+
+    assert _houses_update_call(db) is None
+    db.commit.assert_called_once()
+
+
+def test_valuation_house_persist_exception_does_not_abort_main_save():
+    """UPDATE houses упал → перехвачено, external_valuations INSERT + commit всё равно проходят."""
+    db = _mock_db_valuation()
+    result = _cian_result(house_info=[], management_company=None)
+
+    def _boom(sql, params=None):
+        if "UPDATE houses" in str(sql):
+            raise RuntimeError("boom")
+        mock_result = MagicMock()
+        mock_result.fetchone.return_value = None
+        return mock_result
+
+    db.execute.side_effect = _boom
+
+    _call_save_to_cache(db, result, house_id=42)
+
+    db.commit.assert_called_once()
+
+
+def test_valuation_no_management_company_leaves_mcid_none():
+    """management_company=None → mc_id=None → management_company_id COALESCE(None, existing)."""
+    db = _mock_db_valuation()
+    result = _cian_result(house_info=[], management_company=None)
+
+    _call_save_to_cache(db, result, house_id=42)
+
+    call = _houses_update_call(db)
+    assert call is not None
+    _sql, params = call
+    assert params["mcid"] is None

tradein-mvp/packages/scraper-kit/src/scraper_kit/orchestration/pipeline.py

@@ -2344,7 +2344,9 @@ async def run_cian_city_sweep(
                                 source_url, config=config, proxy_provider=proxy_provider
                             )
                             if cian_enrichment is not None:
-                                cian_save_detail_enrichment(db, listing_id, cian_enrichment)
+                                cian_save_detail_enrichment(
+                                    db, listing_id, cian_enrichment, matcher=matcher
+                                )
                                 counters.detail_enriched += 1
                             else:
                                 counters.detail_failed += 1
@@ -2818,7 +2820,9 @@ async def run_cian_full_load(
                         source_url, config=config, proxy_provider=proxy_provider
                     )
                     if cian_enrichment is not None:
-                        cian_save_detail_enrichment(db, listing_id, cian_enrichment)
+                        cian_save_detail_enrichment(
+                            db, listing_id, cian_enrichment, matcher=matcher
+                        )
                         counters.detail_enriched += 1
                     else:
                         counters.detail_failed += 1

tradein-mvp/packages/scraper-kit/src/scraper_kit/providers/cian/detail.py

@@ -34,7 +34,7 @@ if TYPE_CHECKING:
     from curl_cffi.requests import AsyncSession
 
     from scraper_kit.browser_fetcher import BrowserFetcher
-    from scraper_kit.contracts import ProxyProvider, ScraperConfig
+    from scraper_kit.contracts import HouseMatcher, ProxyProvider, ScraperConfig
 
 logger = logging.getLogger(__name__)
 
@@ -307,13 +307,22 @@ def _extract_agent_profile(offer: dict[str, Any]) -> dict[str, Any] | None:
 # ── Save helpers ──────────────────────────────────────────────────────────────
 
 
-def save_detail_enrichment(db: Session, listing_id: int, enrichment: DetailEnrichment) -> None:
+def save_detail_enrichment(
+    db: Session,
+    listing_id: int,
+    enrichment: DetailEnrichment,
+    *,
+    matcher: HouseMatcher | None = None,
+) -> None:
     """Persist DetailEnrichment to DB:
 
     - UPDATE listings SET windows_view_type, ceiling_height, ... WHERE id = listing_id
     - INSERT INTO offer_price_history rows (if price_changes non-empty)
     - INSERT INTO agents (if agent_profile) and link via agent_id_fk
-    - bti_data не сохраняется здесь — это задача Stage 6 (houses), где есть house_id
+    - bti_data → канонический houses-ряд (#2435), best-effort, только если `matcher`
+      передан вызывающим (инжектируемый `HouseMatcher`, тот же паттерн, что
+      avito/houses.py::_persist_house). `matcher=None` (default) — backward-compat
+      no-op для существующих вызовов, которые ещё не прокинули matcher.
 
     Idempotency: re-running безопасно (UPSERT / ON CONFLICT DO NOTHING semantics).
     """
@@ -447,6 +456,23 @@ def save_detail_enrichment(db: Session, listing_id: int, enrichment: DetailEnric
                 {"aid": agent_db_id, "lid": listing_id},
             )
 
+    # BTI house-persist (#2435): bti_data распарсен (fetch_detail), но раньше выбрасывался —
+    # ранее здесь стоял комментарий "это задача Stage 6 (houses)". matcher резолвит
+    # канонический дом по address/geo листинга (см. _persist_cian_bti_house) — best-effort,
+    # SAVEPOINT изолирует сбой от уже выполненных выше UPDATE/INSERT (тот же паттерн, что
+    # estimator.py::_backfill_house_fias).
+    if enrichment.bti_data and matcher is not None:
+        try:
+            with db.begin_nested():
+                _persist_cian_bti_house(db, listing_id, enrichment.bti_data, matcher=matcher)
+        except Exception as exc:
+            logger.warning(
+                "Cian BTI house-persist failed for listing_id=%s (graceful): %s",
+                listing_id,
+                exc,
+                exc_info=True,
+            )
+
     db.commit()
     logger.info(
         "Cian detail enrichment saved for listing_id=%s (price_changes=%d, agent=%s, snapshot=%s)",
@@ -455,3 +481,110 @@ def save_detail_enrichment(db: Session, listing_id: int, enrichment: DetailEnric
         bool(enrichment.agent_profile),
         _snap_written,
     )
+
+
+# ── BTI house-persist (#2435) ─────────────────────────────────────────────────
+
+
+def _persist_cian_bti_house(
+    db: Session,
+    listing_id: int,
+    bti_data: dict[str, Any],
+    *,
+    matcher: HouseMatcher,
+) -> int | None:
+    """Резолвит канонический дом и пишет BTI-поля дома (#2435).
+
+    `bti_data` — raw ``offerData.bti.houseData`` снапшот (см. `DetailEnrichment.bti_data`),
+    shape (Schema_Cian_SERP_Inventory sec 20.2):
+        {yearRelease, houseMaterialType, floorMax, entrances, flatCount, isEmergency,
+         demolishedInMoscowProgramm, heatIndex, houseHeatSupplyType, houseOverhaulFundType,
+         houseGasSupplyType, houseOverlapType, lifts, seriesName}
+
+    Персистятся только BTI-эксклюзивные колонки из `020_houses_alter_cian.sql`
+    (cian_bti — единственный источник для них, ни один другой скраппер их не пишет):
+        seriesName          → series_name
+        entrances            → entrances
+        flatCount            → flat_count
+        isEmergency          → is_emergency
+        houseHeatSupplyType  → heat_supply_type
+        houseGasSupplyType   → gas_supply_type
+        houseOverlapType     → overlap_type
+    COALESCE(new, existing) — свежий BTI-снапшот побеждает, но не затирает существующее
+    значение NULL'ом (напр. если повторный fetch вернул неполный houseData).
+
+    NOT persisted (gap — нет чистого маппинга на существующую колонку, см. summary):
+        yearRelease/houseMaterialType/floorMax (генерик-колонки year_built/house_type/
+        total_floors существуют, но вне explicit scope БTI-колонок #2435 — future work),
+        demolishedInMoscowProgramm, heatIndex, houseOverhaulFundType, lifts (нет колонки
+        под недифференцированное «лифтов всего» — passenger/cargo split недоступен в BTI).
+
+    address/lat/lon листинга — bti_data адреса не несёт, читаем из `listings` по listing_id.
+
+    Returns:
+        houses.id, либо None если у листинга нет адреса, либо matcher отказал
+        (безномерный адрес без кадастра — метод 'no_house_number', #1772/P1).
+    """
+    row = (
+        db.execute(
+            text("SELECT address, lat, lon FROM listings WHERE id = CAST(:lid AS bigint)"),
+            {"lid": listing_id},
+        )
+        .mappings()
+        .first()
+    )
+    if row is None or not row["address"]:
+        logger.info(
+            "_persist_cian_bti_house: listing_id=%s missing address — skip house resolve",
+            listing_id,
+        )
+        return None
+
+    house_id, _conf, method = matcher.match_or_create_house(
+        db,
+        ext_source="cian_bti",
+        ext_id=str(listing_id),
+        address=row["address"],
+        lat=row["lat"],
+        lon=row["lon"],
+    )
+    if house_id is None:
+        logger.info(
+            "_persist_cian_bti_house: skip listing_id=%s addr=%r (method=%s)",
+            listing_id,
+            row["address"],
+            method,
+        )
+        return None
+
+    db.execute(
+        text("""
+            UPDATE houses SET
+                series_name      = COALESCE(CAST(:series_name AS text), series_name),
+                entrances        = COALESCE(CAST(:entrances AS int), entrances),
+                flat_count       = COALESCE(CAST(:flat_count AS int), flat_count),
+                is_emergency     = COALESCE(CAST(:is_emergency AS boolean), is_emergency),
+                heat_supply_type = COALESCE(CAST(:heat_supply_type AS text), heat_supply_type),
+                gas_supply_type  = COALESCE(CAST(:gas_supply_type AS text), gas_supply_type),
+                overlap_type     = COALESCE(CAST(:overlap_type AS text), overlap_type)
+            WHERE id = CAST(:house_id AS bigint)
+        """),
+        {
+            "house_id": house_id,
+            "series_name": bti_data.get("seriesName"),
+            "entrances": bti_data.get("entrances"),
+            "flat_count": bti_data.get("flatCount"),
+            "is_emergency": bti_data.get("isEmergency"),
+            "heat_supply_type": bti_data.get("houseHeatSupplyType"),
+            "gas_supply_type": bti_data.get("houseGasSupplyType"),
+            "overlap_type": bti_data.get("houseOverlapType"),
+        },
+    )
+
+    logger.info(
+        "_persist_cian_bti_house: listing_id=%s → house_id=%s (method=%s)",
+        listing_id,
+        house_id,
+        method,
+    )
+    return int(house_id)

tradein-mvp/packages/scraper-kit/src/scraper_kit/providers/cian/valuation.py

@@ -23,6 +23,7 @@ from __future__ import annotations
 import hashlib
 import json
 import logging
+import re
 from dataclasses import dataclass, field
 from typing import TYPE_CHECKING, Any
 from urllib.parse import urlencode
@@ -557,7 +558,251 @@ def _save_to_cache(
                 "raw": json.dumps(result.raw_state) if result.raw_state else None,
             },
         )
+
+        # Best-effort house enrichment (#2435): management_company/house_info/
+        # cian_internal_house_id → канонический houses-ряд. house_id уже резолвлен
+        # ВЫЗЫВАЮЩИМ (estimator.py::match_house_readonly — read-only, не создаёт дом из
+        # ephemeral valuation-запроса) — здесь только COALESCE-UPDATE, никакого
+        # match_or_create_house. SAVEPOINT изолирует сбой от уже выполненного INSERT
+        # выше (тот же паттерн, что estimator.py::_backfill_house_fias).
+        try:
+            with db.begin_nested():
+                _persist_cian_valuation_house(db, house_id=house_id, result=result)
+        except Exception as exc:
+            logger.warning(
+                "Cian valuation house-persist failed for house_id=%s (graceful): %s",
+                house_id,
+                exc,
+                exc_info=True,
+            )
+
         db.commit()
     except Exception as exc:
         logger.error("Cian valuation cache save failed: %s", exc, exc_info=True)
         raise
+
+
+# ── House enrichment (#2435) ──────────────────────────────────────────────────
+
+# houseInfo.data.items title (RU, lowercased) → houses column. Только поля с чистым
+# 1:1 маппингом на существующую колонку (Schema_Cian_SERP_Inventory sec 24.6).
+# Пропущенные (мусоропровод/реновация/спортплощадка/фонд капремонта) — нет колонки
+# в houses, см. gap-заметку в docstring _persist_cian_valuation_house.
+_HOUSE_INFO_TITLE_MAP: dict[str, str] = {
+    "год постройки": "year_built",
+    "тип дома": "house_type",
+    "этажность": "total_floors",
+    "газоснабжение": "gas_supply_type",
+    "отопление": "heat_supply_type",
+    "тип перекрытий": "overlap_type",
+    "подъездов": "entrances",
+    "квартир": "flat_count",
+    "аварийность": "is_emergency",
+    "детская площадка": "has_playground",
+}
+_HOUSE_INFO_BOOL_FIELDS = {"is_emergency", "has_playground"}
+_BOOL_YES_RU = {"да"}
+_BOOL_NO_RU = {"нет"}
+
+# "Количество лифтов": free-text "1 пассажирский, 1 грузовой" (не структурировано Cian'ом).
+_ELEVATORS_PASSENGER_RE = re.compile(r"(\d+)\s*пассажир", re.IGNORECASE)
+_ELEVATORS_CARGO_RE = re.compile(r"(\d+)\s*груз", re.IGNORECASE)
+
+
+def _map_house_info_items(items: list[dict[str, Any]]) -> dict[str, Any]:
+    """houseInfo.data.items (RU title/value пары) → known houses columns (#2435).
+
+    Best-effort парсинг: неизвестные/непонятные title просто пропускаются (нет
+    исключений). "Количество лифтов" — единственное поле с free-text значением
+    ("N пассажирский, M грузовой") — парсится regex'ом, при несовпадении паттерна
+    соответствующий счётчик остаётся не заполненным (не 0 — 0 означало бы "нет лифтов").
+    """
+    out: dict[str, Any] = {}
+    for item in items:
+        if not isinstance(item, dict):
+            continue
+        title = str(item.get("title") or "").strip().lower()
+        value = item.get("value")
+
+        if title == "количество лифтов" and isinstance(value, str):
+            pm = _ELEVATORS_PASSENGER_RE.search(value)
+            if pm:
+                out["passenger_elevators"] = int(pm.group(1))
+            cm = _ELEVATORS_CARGO_RE.search(value)
+            if cm:
+                out["cargo_elevators"] = int(cm.group(1))
+            continue
+
+        field_name = _HOUSE_INFO_TITLE_MAP.get(title)
+        if field_name is None:
+            continue
+
+        if field_name in _HOUSE_INFO_BOOL_FIELDS:
+            if isinstance(value, bool):
+                out[field_name] = value
+            elif isinstance(value, str):
+                v_norm = value.strip().lower()
+                if v_norm in _BOOL_YES_RU:
+                    out[field_name] = True
+                elif v_norm in _BOOL_NO_RU:
+                    out[field_name] = False
+        else:
+            out[field_name] = value
+    return out
+
+
+def _upsert_cian_valuation_management_company(
+    db: Session, mc: dict[str, Any] | None
+) -> int | None:
+    """UPSERT management_companies из Cian Valuation Calculator's managementCompany.data.
+
+    Тот же намерение, что providers/cian/newbuilding.py::save_newbuilding_enrichment
+    (независимая копия — newbuilding.py вне scope #2435, не рефакторим/не переиспользуем),
+    НО другой механизм dedup. ext_source='cian_valuation' (не 'cian') — валюация не даёт
+    числового id УК (в отличие от newbuilding.managementCompany.id), поэтому пишем
+    ext_id=NULL. `UNIQUE (ext_source, name, ext_id)` под Postgres-дефолтом NULLS DISTINCT
+    не считает два NULL равными — `ON CONFLICT` с ext_id=NULL никогда бы не сработал (каждый
+    вызов создавал бы новую дублирующую строку). Поэтому dedup сделан явно: SELECT по
+    (ext_source, name) WHERE ext_id IS NULL перед INSERT, а не через constraint.
+
+    opening_hours — Cian отдаёт список словарей ([{"Пн–Пт": ["10:00-20:00"]}, ...]), не
+    строку; json.dumps в text-колонку (в отличие от newbuilding.py, которая пишет объект
+    напрямую — тот путь не в scope этой задачи).
+    """
+    if not mc or not isinstance(mc, dict):
+        return None
+    name = mc.get("name")
+    if not name:
+        return None
+
+    opening_hours = mc.get("openingHours")
+    phones = mc.get("phones") or []
+    email = mc.get("email")
+    oh_json = json.dumps(opening_hours, ensure_ascii=False) if opening_hours else None
+    chief = mc.get("chiefName")
+
+    existing = db.execute(
+        text("""
+            SELECT id FROM management_companies
+            WHERE ext_source = 'cian_valuation' AND name = :name AND ext_id IS NULL
+            LIMIT 1
+        """),
+        {"name": name},
+    ).fetchone()
+    if existing:
+        mc_id = int(existing[0])
+        db.execute(
+            text("""
+                UPDATE management_companies SET
+                    phones = COALESCE(CAST(:phones AS text[]), phones),
+                    email  = COALESCE(:email, email)
+                WHERE id = CAST(:mc_id AS bigint)
+            """),
+            {"phones": phones, "email": email, "mc_id": mc_id},
+        )
+        return mc_id
+
+    row = db.execute(
+        text("""
+            INSERT INTO management_companies (
+                name, phones, email, opening_hours, chief_name, ext_source, ext_id
+            ) VALUES (
+                :name, CAST(:phones AS text[]), :email, :oh, :chief, 'cian_valuation', NULL
+            )
+            RETURNING id
+        """),
+        {
+            "name": name,
+            "phones": phones,
+            "email": email,
+            "oh": oh_json,
+            "chief": chief,
+        },
+    ).fetchone()
+    return int(row[0]) if row else None
+
+
+def _persist_cian_valuation_house(
+    db: Session,
+    *,
+    house_id: int | None,
+    result: CianValuationResult,
+) -> None:
+    """Пишет managementCompany/houseInfo/houseId в канонический houses-ряд (#2435).
+
+    `house_id` уже резолвлен ВЫЗЫВАЮЩИМ (estimator.py, read-only match_house_readonly) —
+    valuation-запрос по эфемерному целевому адресу не должен САМ создавать дом (в отличие
+    от cian_bti #2435 Part 1, где match_or_create_house уместен — там источник привязан
+    к конкретному листингу/дому, а не к произвольному введённому адресу). house_id=None →
+    no-op, best-effort.
+
+    COALESCE-направление — по app.services.matching.conflict_resolution.HOUSE_FIELD_PRIORITY:
+      - management_company_id: HOUSE_FIELD_PRIORITY["management_company_id"] == ["cian_valuation"]
+        — единственный источник → COALESCE(new, existing) (свежее значение побеждает).
+      - cian_internal_house_id (filters.houseId): не в реестре, но тот же принцип, что
+        newbuilding.py применяет к этой же колонке → COALESCE(new, existing).
+      - houseInfo-производные (year_built/house_type/total_floors/entrances/flat_count/
+        is_emergency/heat_supply_type/gas_supply_type/overlap_type/has_playground/
+        passenger_elevators/cargo_elevators): cian_valuation НЕ входит в приоритет для этих
+        полей вовсе (или cian_bti/serp/avito ранжированы выше) → COALESCE(existing, new) —
+        валюация только заполняет пробелы, не перетирает более авторитетные источники.
+    """
+    if house_id is None:
+        return
+
+    mc_id = _upsert_cian_valuation_management_company(db, result.management_company)
+    info = _map_house_info_items(result.house_info)
+
+    db.execute(
+        text("""
+            UPDATE houses SET
+                management_company_id  = COALESCE(
+                    CAST(:mcid AS bigint), houses.management_company_id
+                ),
+                cian_internal_house_id = COALESCE(
+                    CAST(:cihi AS bigint), houses.cian_internal_house_id
+                ),
+                year_built              = COALESCE(houses.year_built, CAST(:yb AS int)),
+                house_type              = COALESCE(houses.house_type, CAST(:ht AS text)),
+                total_floors            = COALESCE(houses.total_floors, CAST(:tf AS int)),
+                entrances               = COALESCE(houses.entrances, CAST(:ent AS int)),
+                flat_count              = COALESCE(houses.flat_count, CAST(:fc AS int)),
+                is_emergency            = COALESCE(houses.is_emergency, CAST(:ie AS boolean)),
+                heat_supply_type        = COALESCE(
+                    houses.heat_supply_type, CAST(:hst AS text)
+                ),
+                gas_supply_type         = COALESCE(houses.gas_supply_type, CAST(:gst AS text)),
+                overlap_type            = COALESCE(houses.overlap_type, CAST(:ot AS text)),
+                has_playground          = COALESCE(
+                    houses.has_playground, CAST(:hp AS boolean)
+                ),
+                passenger_elevators     = COALESCE(
+                    houses.passenger_elevators, CAST(:pe AS int)
+                ),
+                cargo_elevators         = COALESCE(houses.cargo_elevators, CAST(:ce AS int))
+            WHERE id = CAST(:hid AS bigint)
+        """),
+        {
+            "hid": house_id,
+            "mcid": mc_id,
+            "cihi": result.external_house_id,
+            "yb": info.get("year_built"),
+            "ht": info.get("house_type"),
+            "tf": info.get("total_floors"),
+            "ent": info.get("entrances"),
+            "fc": info.get("flat_count"),
+            "ie": info.get("is_emergency"),
+            "hst": info.get("heat_supply_type"),
+            "gst": info.get("gas_supply_type"),
+            "ot": info.get("overlap_type"),
+            "hp": info.get("has_playground"),
+            "pe": info.get("passenger_elevators"),
+            "ce": info.get("cargo_elevators"),
+        },
+    )
+    logger.info(
+        "_persist_cian_valuation_house: house_id=%s mc_id=%s cian_internal_house_id=%s",
+        house_id,
+        mc_id,
+        result.external_house_id,
+    )