fix(cian): recover new-build house number via cadastral anchor + listings (source,source_id) dedup (#1773)

PART 1 — cian-новостройки теряли номер дома: geo.address[] для ЖК часто без
части {type:"house"}, адрес выходил house-less («…улица Евгения Савкова» без
«29») и раздробленные корпуса схлопывались в catch-all → same-building anchor
рвался. Теперь для новостроек без house восстанавливаем номер по приоритету:
(a) building.houseNumber / building.address, (b) building_cadastral_number как
anchor-ключ, (c) хвостовой дом-токен из newbuilding.name — без catch-all
фабрикации. building_cadastral_number предпочтён как house anchor
(house_source=cian_building_cad) → Tier 0 cadastr_exact схлопывает корпуса;
nb_id остаётся fallback'ом.

PART 2 — дубль-листинги: миграция 133 дедупит listings по (source, source_id)
(keeper = house_id_fk → активность → свежесть → id), безопасно перепривязывает
детей (price_history/snapshots/listing_sources/external_valuations/merged_into)
collision-safe и вешает частичный UNIQUE (source, source_id) WHERE source_id IS
NOT NULL. Идемпотентно, недеструктивно для keeper. save_listings оставлен на
ON CONFLICT (dedup_hash) — арбитр согласован с новым constraint, source_id=NULL
(yandex/url-only) не ломается.

Refs #1773

tradein-mvp/backend/app/services/scrapers/base.py

@@ -355,6 +355,13 @@ def save_listings(
                     :geo_precision, :card_hash,
                     NOW(), NOW()
                 )
+                -- Конфликт-арбитр — dedup_hash (sha256(source|source_id)).
+                -- Для одного (source, source_id) формула даёт ОДИН dedup_hash,
+                -- поэтому этот апсерт уже не вставит второй раз → согласован с
+                -- частичным UNIQUE (source, source_id) WHERE source_id IS NOT NULL
+                -- (133_listings_uq_source_source_id.sql). Менять арбитр на
+                -- (source, source_id) НЕЛЬЗЯ: yandex/url-only дают source_id=NULL,
+                -- а NULL не годится как conflict-target → дубли вернулись бы (#1773).
                 ON CONFLICT (dedup_hash) DO UPDATE
                     SET last_seen_at = NOW(),
                         is_active = true,

tradein-mvp/backend/app/services/scrapers/cian.py

@@ -28,6 +28,7 @@ import hashlib
 import inspect
 import logging
 import math
+import re
 from collections.abc import Callable
 from datetime import UTC, datetime
 from typing import Any
@@ -839,8 +840,21 @@ class CianScraper(BaseScraper):
                 lat = lon = None
 
             # ── Адрес из geo.address[] ───────────────────────────────────────
+            # newbuilding нужен ДО _format_address: для ЖК geo.address[] часто без
+            # части {type:"house"} → дом-номер восстанавливаем из building/кадастра/
+            # названия ЖК (#1773), иначе все корпуса схлопываются в house-less
+            # catch-all и same-building anchor рвётся.
             address_parts = geo.get("address") or []
-            address = _format_address(address_parts)
+            newbuilding: dict[str, Any] = offer.get("newbuilding") or {}
+            nb_id = newbuilding.get("id")
+            is_newbuilding = bool(nb_id) and int(nb_id) > 0
+            address = _format_address(
+                address_parts,
+                building=building,
+                building_cadastral_number=building_cadastral_number,
+                newbuilding=newbuilding,
+                is_newbuilding=is_newbuilding,
+            )
 
             # ── Metro stations ───────────────────────────────────────────────
             undergrounds: list[dict[str, Any]] = geo.get("undergrounds") or []
@@ -858,16 +872,24 @@ class CianScraper(BaseScraper):
             ]
 
             # ── Newbuilding link ─────────────────────────────────────────────
-            newbuilding: dict[str, Any] = offer.get("newbuilding") or {}
-            nb_id = newbuilding.get("id")
+            # newbuilding / nb_id / is_newbuilding уже вычислены выше (до адреса).
             # id == 0 — sentinel для вторички (per Schema sec 20.5)
             house_source: str | None = None
             house_ext_id: str | None = None
             listing_segment: str | None = None
 
-            if nb_id and int(nb_id) > 0:
-                house_source = "cian_newbuilding"
-                house_ext_id = str(nb_id)
+            if is_newbuilding:
+                # Предпочитаем кадастр ДОМА как same-building anchor (#1773): nb_id
+                # группирует по ЖК целиком, а один ЖК — это несколько корпусов
+                # (6679/11110/7019). buildingCadastralNumber стабильно идентифицирует
+                # ФИЗИЧЕСКИЙ дом → Tier 0 cadastr_exact в match_or_create_house
+                # схлопывает раздробленные корпуса. nb_id оставляем fallback'ом.
+                if building_cadastral_number:
+                    house_source = "cian_building_cad"
+                    house_ext_id = building_cadastral_number
+                else:
+                    house_source = "cian_newbuilding"
+                    house_ext_id = str(nb_id)
                 listing_segment = "novostroyki"
             else:
                 house_source = "cian"
@@ -1020,27 +1042,113 @@ class CianScraper(BaseScraper):
 
 # ── Address formatter ────────────────────────────────────────────────────────
 
+# Дом-токен в хвосте строки: «29», «29А», «12 к1», «5/2», «10 стр.3», «1С7».
+# Якорим в конец строки (\Z) — берём именно завершающий номер, не цифры из
+# названия ЖК («Суходольский квартал»). Не матчим «р-н», «мкр» и пр. слова.
+_HOUSE_TOKEN_RE = re.compile(
+    r"(\d+[А-Яа-яA-Za-z]?"  # базовый номер: 29 / 29А / 1С
+    r"(?:\s*(?:к|корп\.?|стр\.?|с)\s*\d+[А-Яа-яA-Za-z]?)?"  # корпус/строение
+    r"(?:\s*/\s*\d+[А-Яа-яA-Za-z]?)?)"  # дробь 5/2
+    r"\s*\Z"
+)
 
-def _format_address(address_parts: list[dict[str, Any]]) -> str:
+
+def _recover_newbuilding_house(
+    *,
+    building: dict[str, Any] | None,
+    building_cadastral_number: str | None,
+    newbuilding: dict[str, Any] | None,
+) -> str | None:
+    """Восстановить дом-токен для новостройки, когда geo.address[] без {type:"house"}.
+
+    Приоритет (#1773):
+      (a) building.houseNumber / building.address — если SERP-state их отдал;
+      (b) building_cadastral_number — стабильный same-building anchor: если ни
+          номер, ни название не дали дом, добавляем «кад. 66:41:...:11396» как
+          ключ-якорь, чтобы раздробленные корпуса схлопывались по кадастру;
+      (c) хвостовой дом-токен из newbuilding.name («… 29» / «… 29А к1»).
+
+    НЕ выдумываем catch-all номер: если ничего из (a)/(b)/(c) не нашли — None.
+    Кадастр (b) предпочтительнее как house anchor (см. _offer_to_lot).
+    """
+    building = building or {}
+    newbuilding = newbuilding or {}
+
+    # (a) building.houseNumber / building.address — прямые поля SERP-state
+    house_number = building.get("houseNumber")
+    if house_number is not None:
+        token = str(house_number).strip()
+        if token:
+            return token
+    b_addr = (building.get("address") or "").strip()
+    if b_addr:
+        m = _HOUSE_TOKEN_RE.search(b_addr)
+        if m:
+            return m.group(1).strip()
+
+    # (c) хвостовой дом-токен из названия ЖК (до кадастра — он читабельнее)
+    nb_name = (newbuilding.get("name") or "").strip()
+    if nb_name:
+        m = _HOUSE_TOKEN_RE.search(nb_name)
+        if m:
+            return m.group(1).strip()
+
+    # (b) кадастр дома как anchor-ключ (последний приоритет в строке адреса,
+    # но самый надёжный для матчинга — Tier 0 cadastr_exact)
+    if building_cadastral_number:
+        cad = building_cadastral_number.strip()
+        if cad:
+            return f"кад. {cad}"
+
+    return None
+
+
+def _format_address(
+    address_parts: list[dict[str, Any]],
+    *,
+    building: dict[str, Any] | None = None,
+    building_cadastral_number: str | None = None,
+    newbuilding: dict[str, Any] | None = None,
+    is_newbuilding: bool = False,
+) -> str:
     """Сформировать читаемый адрес из geo.address[] Cian.
 
     Пропускаем location (страна/регион) и metro — берём раион/улицу/дом.
     Пример: [location=Москва, raion=Пресненский, street=..., house=1С7, metro=Москва-Сити]
     → 'Пресненский, улица ..., 1С7'
+
+    #1773: для новостроек (is_newbuilding) geo.address[] часто без {type:"house"}.
+    Если части house нет — восстанавливаем номер дома через _recover_newbuilding_house
+    (building → название ЖК → кадастр) и дописываем его в конец строки, чтобы
+    same-building anchor (fingerprint/кадастр) не рвался по house-less улице.
     """
     if not address_parts:
         return "Екатеринбург (Cian)"
 
     skip_types = {"location", "metro"}
     parts: list[str] = []
+    has_house = False
     for part in address_parts:
         if not isinstance(part, dict):
             continue
         ptype = part.get("type", "")
+        if ptype == "house":
+            has_house = True
         if ptype in skip_types:
             continue
         name = (part.get("fullName") or part.get("name") or "").strip()
         if name:
             parts.append(name)
 
+    # Восстановление дома только для новостроек без части house — вторичка с
+    # house-less адресом не трогается (не выдумываем номер).
+    if is_newbuilding and not has_house:
+        recovered = _recover_newbuilding_house(
+            building=building,
+            building_cadastral_number=building_cadastral_number,
+            newbuilding=newbuilding,
+        )
+        if recovered:
+            parts.append(recovered)
+
     return ", ".join(parts) if parts else "Екатеринбург (Cian)"

tradein-mvp/backend/data/sql/133_listings_uq_source_source_id.sql

@@ -0,0 +1,207 @@
+-- 133_listings_uq_source_source_id.sql
+-- Дедуп дублей listings по (source, source_id) + UNIQUE-констрейнт против повторов.
+--
+-- WHY (root cause — #1773):
+--   Один и тот же оффер задваивается в listings: cian 330047129 лежит в ДВУХ
+--   строках — одна с house_id_fk (привязана к дому), другая с house_id_fk=NULL.
+--   Корень — историческая смена формулы dedup_hash (#753): раньше hash включал
+--   цену, поэтому каждое изменение цены порождало новую строку с другим
+--   dedup_hash. Текущий UPSERT идёт ON CONFLICT (dedup_hash), а старые «ghost»-
+--   строки несут price-inclusive hash, который новым апсертом уже не матчится →
+--   живут вечно как дубль того же (source, source_id). 113_deactivate_ghost_
+--   duplicate_listings.sql лишь ГАСИЛ (is_active=false) лишние строки, но
+--   физически их не удалял и не запрещал появление новых → дубли остаются и
+--   рвут same-building anchor / раздувают инвентарь.
+--
+-- WHAT:
+--   1) Внутри каждой группы (source, source_id) с >1 строк выбираем «лучшую»
+--      строку-keeper и схлопываем группу на неё:
+--        keeper = приоритет (house_id_fk IS NOT NULL) DESC,
+--                 затем is_active DESC, затем last_seen_at DESC, затем id DESC.
+--      То есть оставляем строку, уже привязанную к дому и самую свежую.
+--   2) Перепривязываем дочерние записи (offer_price_history, listings_snapshots,
+--      listing_sources, external_valuations) и self-ref listings.merged_into
+--      с loser-строк на keeper — collision-safe (WHERE NOT EXISTS), чтобы не
+--      нарушить их собственные UNIQUE/PK. Остаток (коллизии — у keeper уже есть
+--      эквивалент) уедет CASCADE при удалении loser.
+--   3) Удаляем loser-строки (FK ON DELETE CASCADE подчищает остаточных детей).
+--   4) Вешаем UNIQUE-индекс на (source, source_id) WHERE source_id IS NOT NULL —
+--      частичный, т.к. источники без source_id (yandex/url-only) дают source_id
+--      NULL, а NULL в Postgres-UNIQUE считаются различными → их не трогаем.
+--
+-- IDEMPOTENCY / SAFETY (важно для прода):
+--   - Весь файл в одной транзакции BEGIN/COMMIT — атомарно.
+--   - Дедуп НЕДЕСТРУКТИВЕН для keeper: его дети сохраняются полностью, удаляются
+--     только дубли-loser и их (коллизионные) дети, у которых keeper уже имеет
+--     эквивалент.
+--   - Повторный прогон: после первого раза дублей нет → дедуп-шаги no-op;
+--     индекс создаётся через CREATE UNIQUE INDEX IF NOT EXISTS → no-op.
+--   - source_id IS NULL строки полностью игнорируются (частичный индекс).
+--
+-- RELATION к app/services/scrapers/base.py save_listings:
+--   save_listings апсертит ON CONFLICT (dedup_hash). Текущая формула dedup_hash =
+--   sha256(source || '|' || source_id) детерминирована из (source, source_id),
+--   поэтому одинаковый (source, source_id) всегда даёт ОДИН dedup_hash →
+--   ON CONFLICT (dedup_hash) и так не вставит второй раз. Новый UNIQUE
+--   (source, source_id) согласован с этим (страхует только от
+--   историко-формульных дублей), НЕ конфликтует с апсертом и НЕ требует менять
+--   save_listings на ON CONFLICT (source, source_id). source_id=NULL не попадает
+--   под частичный индекс → апсерты yandex/url-only не ломаются.
+--
+-- DEPENDENCIES:
+--   - listings (id, source, source_id, house_id_fk, is_active, last_seen_at,
+--     merged_into) -- 002_core_tables.sql + 011_listings_alter.sql + 028
+--   - offer_price_history.listing_id -> listings(id) ON DELETE CASCADE -- 023
+--   - listings_snapshots PK (listing_id, snapshot_date) -- 016
+--   - listing_sources UNIQUE (ext_source, ext_id) -- 028
+--   - external_valuations partial UNIQUE (source, house_id, listing_id) -- 044
+--
+-- Deploy order: after 131_fix_diff_percent_overflow.sql (latest revision).
+
+BEGIN;
+
+DO $$
+DECLARE
+    v_groups          bigint;
+    v_losers          bigint;
+    v_deleted         bigint;
+    v_remaining_dupes bigint;
+BEGIN
+    -- ---------------------------------------------------------------------
+    -- 0. Материализуем keeper-per-group и список loser-строк во временную
+    --    таблицу. keeper выбираем оконной функцией с детерминированным
+    --    ORDER BY: привязка к дому > активность > свежесть > id.
+    -- ---------------------------------------------------------------------
+    CREATE TEMP TABLE _ld_map ON COMMIT DROP AS
+    WITH ranked AS (
+        SELECT
+            id,
+            source,
+            source_id,
+            first_value(id) OVER (
+                PARTITION BY source, source_id
+                ORDER BY
+                    (house_id_fk IS NOT NULL) DESC,
+                    is_active DESC,
+                    last_seen_at DESC NULLS LAST,
+                    id DESC
+            ) AS keeper_id
+        FROM listings
+        WHERE source_id IS NOT NULL
+          AND (source, source_id) IN (
+              SELECT source, source_id
+              FROM listings
+              WHERE source_id IS NOT NULL
+              GROUP BY source, source_id
+              HAVING count(*) > 1
+          )
+    )
+    SELECT id AS loser_id, keeper_id
+    FROM ranked
+    WHERE id <> keeper_id;
+
+    SELECT count(*)                       INTO v_losers FROM _ld_map;
+    SELECT count(DISTINCT keeper_id)      INTO v_groups FROM _ld_map;
+
+    IF v_losers = 0 THEN
+        RAISE NOTICE 'migration 133: no duplicate (source, source_id) groups — dedup no-op';
+    ELSE
+        -- -----------------------------------------------------------------
+        -- 1. Перепривязка дочерних записей loser -> keeper (collision-safe).
+        --    Где у keeper уже есть запись с тем же UNIQUE/PK-ключом —
+        --    loser-строку НЕ двигаем (она уедет CASCADE на шаге 3).
+        -- -----------------------------------------------------------------
+
+        -- offer_price_history: нет UNIQUE на listing_id → двигаем все.
+        UPDATE offer_price_history o
+        SET listing_id = m.keeper_id
+        FROM _ld_map m
+        WHERE o.listing_id = m.loser_id;
+
+        -- listings_snapshots: PK (listing_id, snapshot_date) → не двигаем дни,
+        -- которые у keeper уже есть.
+        UPDATE listings_snapshots s
+        SET listing_id = m.keeper_id
+        FROM _ld_map m
+        WHERE s.listing_id = m.loser_id
+          AND NOT EXISTS (
+              SELECT 1 FROM listings_snapshots s2
+              WHERE s2.listing_id = m.keeper_id
+                AND s2.snapshot_date = s.snapshot_date
+          );
+
+        -- listing_sources: UNIQUE (ext_source, ext_id) → не двигаем те, что
+        -- у keeper уже есть.
+        UPDATE listing_sources ls
+        SET listing_id = m.keeper_id
+        FROM _ld_map m
+        WHERE ls.listing_id = m.loser_id
+          AND NOT EXISTS (
+              SELECT 1 FROM listing_sources ls2
+              WHERE ls2.listing_id = m.keeper_id
+                AND ls2.ext_source = ls.ext_source
+                AND ls2.ext_id     = ls.ext_id
+          );
+
+        -- external_valuations: partial UNIQUE (source, house_id, listing_id)
+        -- WHERE house_id IS NOT NULL AND listing_id IS NOT NULL.
+        UPDATE external_valuations ev
+        SET listing_id = m.keeper_id
+        FROM _ld_map m
+        WHERE ev.listing_id = m.loser_id
+          AND NOT EXISTS (
+              SELECT 1 FROM external_valuations ev2
+              WHERE ev2.listing_id = m.keeper_id
+                AND ev2.source = ev.source
+                AND ev2.house_id IS NOT DISTINCT FROM ev.house_id
+          );
+
+        -- self-ref: listings.merged_into указывает на loser → переводим на keeper.
+        UPDATE listings l
+        SET merged_into = m.keeper_id
+        FROM _ld_map m
+        WHERE l.merged_into = m.loser_id;
+
+        -- -----------------------------------------------------------------
+        -- 2. Удаляем loser-строки. FK ON DELETE CASCADE подчистит остаточных
+        --    детей (коллизии, не перенесённые на шаге 1) — у keeper для них
+        --    уже есть эквивалент.
+        -- -----------------------------------------------------------------
+        DELETE FROM listings l
+        USING _ld_map m
+        WHERE l.id = m.loser_id;
+
+        GET DIAGNOSTICS v_deleted = ROW_COUNT;
+
+        RAISE NOTICE
+            'migration 133: dedup done — groups=% losers=% deleted=% (kept best row per group)',
+            v_groups, v_losers, v_deleted;
+    END IF;
+
+    -- Контроль: после дедупа не должно остаться групп с дублями.
+    SELECT count(*) INTO v_remaining_dupes
+    FROM (
+        SELECT 1
+        FROM listings
+        WHERE source_id IS NOT NULL
+        GROUP BY source, source_id
+        HAVING count(*) > 1
+    ) d;
+
+    IF v_remaining_dupes > 0 THEN
+        RAISE EXCEPTION
+            'migration 133: still % duplicate (source, source_id) groups after dedup — aborting',
+            v_remaining_dupes;
+    END IF;
+END $$;
+
+-- -------------------------------------------------------------------------
+-- 3. UNIQUE-индекс против будущих дублей. Частичный (source_id IS NOT NULL):
+--    источники без source_id (yandex/url-only) дают NULL и не должны падать.
+--    IF NOT EXISTS → идемпотентно при повторном прогоне.
+-- -------------------------------------------------------------------------
+CREATE UNIQUE INDEX IF NOT EXISTS listings_source_source_id_uq
+    ON listings (source, source_id)
+    WHERE source_id IS NOT NULL;
+
+COMMIT;

tradein-mvp/backend/tests/test_cian_serp_scraper.py

@@ -279,13 +279,31 @@ def test_cian_serp_phones():
 
 
 def test_cian_serp_house_ext_id_newbuilding():
-    """house_ext_id заполнен если offer.newbuilding.id > 0."""
+    """Новостройка с кадастром дома → anchor = building cad (#1773), не nb_id.
+
+    _make_full_offer несёт building_cadastral_number, поэтому same-building anchor
+    предпочитает кадастр дома (надёжнее, чем nb_id, который группирует весь ЖК)."""
     offer = _make_full_offer(newbuilding_id=54016)
     html = _make_cian_serp_html([offer])
 
     scraper = CianScraper()
     lots = scraper._parse_serp_html(html)
 
+    lot = lots[0]
+    assert lot.house_source == "cian_building_cad"
+    assert lot.house_ext_id == "66:41:0204016:100"
+    assert lot.listing_segment == "novostroyki"
+
+
+def test_cian_serp_house_ext_id_newbuilding_nb_id_fallback():
+    """Без кадастра дома anchor падает на nb_id (ЖК) — back-compat fallback."""
+    offer = _make_full_offer(newbuilding_id=54016, building_cadastral_number="")
+    offer["buildingCadastralNumber"] = None
+    html = _make_cian_serp_html([offer])
+
+    scraper = CianScraper()
+    lots = scraper._parse_serp_html(html)
+
     lot = lots[0]
     assert lot.house_source == "cian_newbuilding"
     assert lot.house_ext_id == "54016"
@@ -576,3 +594,136 @@ def test_format_address_standard():
 def test_format_address_empty():
     """Пустой address[] → fallback string."""
     assert _format_address([]) == "Екатеринбург (Cian)"
+
+
+# ── #1773: новостройка без part{type:"house"} → восстановление номера дома ─────
+
+
+def _make_newbuilding_houseless_offer(
+    cian_id: int = 330047129,
+    newbuilding_id: int = 54016,
+    newbuilding_name: str = "Суходольский квартал",
+    building_cadastral_number: str | None = "66:41:0313010:11396",
+    building_house_number: str | None = None,
+) -> dict:
+    """Cian-оффер новостройки: geo.address[] БЕЗ {type:"house"} (ЖК Суходольский,
+    ул. Евгения Савкова 29, cian 330047129 — реальный кейс из #1773)."""
+    offer = _make_full_offer(
+        cian_id=cian_id,
+        newbuilding_id=newbuilding_id,
+        building_cadastral_number=building_cadastral_number or "",
+    )
+    # geo.address[] без части house — корень бага: «…улица Евгения Савкова» без «29»
+    offer["geo"]["address"] = [
+        {"type": "location", "name": "Свердловская область"},
+        {"type": "city", "name": "Екатеринбург"},
+        {"type": "district", "name": "Академический"},
+        {"type": "street", "name": "улица Евгения Савкова"},
+    ]
+    offer["newbuilding"] = {"id": newbuilding_id, "name": newbuilding_name}
+    building = offer.setdefault("building", {})
+    if building_house_number is not None:
+        building["houseNumber"] = building_house_number
+    if building_cadastral_number is None:
+        offer["buildingCadastralNumber"] = None
+    return offer
+
+
+def test_cian_newbuilding_cad_number_anchor():
+    """Новостройка с кадастром дома → house_source/house_ext_id = building cad
+    (same-building anchor #1773), адрес дополнен кадастром."""
+    offer = _make_newbuilding_houseless_offer(
+        building_cadastral_number="66:41:0313010:11396",
+    )
+    html = _make_cian_serp_html([offer])
+
+    scraper = CianScraper()
+    lots = scraper._parse_serp_html(html)
+
+    assert len(lots) == 1
+    lot = lots[0]
+    # Кадастр дома предпочтён как anchor-ключ вместо nb_id (ЖК целиком)
+    assert lot.house_source == "cian_building_cad"
+    assert lot.house_ext_id == "66:41:0313010:11396"
+    assert lot.building_cadastral_number == "66:41:0313010:11396"
+    assert lot.listing_segment == "novostroyki"
+    # geo.address[] был без house → кадастр дописан в адрес как якорь
+    assert "Евгения Савкова" in (lot.address or "")
+    assert "66:41:0313010:11396" in (lot.address or "")
+
+
+def test_cian_newbuilding_recovers_house_from_building_number():
+    """building.houseNumber имеет приоритет (a): «29» восстановлен в адрес,
+    кадастр остаётся anchor-ключом."""
+    offer = _make_newbuilding_houseless_offer(
+        building_cadastral_number="66:41:0313010:11396",
+        building_house_number="29",
+    )
+    html = _make_cian_serp_html([offer])
+
+    scraper = CianScraper()
+    lots = scraper._parse_serp_html(html)
+
+    lot = lots[0]
+    addr = lot.address or ""
+    assert "Евгения Савкова" in addr
+    assert "29" in addr  # номер дома восстановлен из building.houseNumber
+    # анкор по-прежнему кадастр дома (надёжнее nb_id)
+    assert lot.house_ext_id == "66:41:0313010:11396"
+
+
+def test_cian_newbuilding_recovers_house_from_name_when_no_cad():
+    """Без кадастра дома: дом-токен «29» вытягивается из названия ЖК (приоритет c),
+    anchor падает на nb_id."""
+    offer = _make_newbuilding_houseless_offer(
+        newbuilding_name="ЖК на Савкова 29",
+        building_cadastral_number=None,
+    )
+    html = _make_cian_serp_html([offer])
+
+    scraper = CianScraper()
+    lots = scraper._parse_serp_html(html)
+
+    lot = lots[0]
+    assert lot.building_cadastral_number is None
+    # нет кадастра → anchor = nb_id (ЖК)
+    assert lot.house_source == "cian_newbuilding"
+    assert lot.house_ext_id == "54016"
+    addr = lot.address or ""
+    assert "29" in addr  # дом восстановлен из названия ЖК
+
+
+def test_cian_newbuilding_no_house_no_fabrication():
+    """Ни building, ни кадастра, ни номера в названии → номер НЕ выдумываем."""
+    offer = _make_newbuilding_houseless_offer(
+        newbuilding_name="Суходольский квартал",  # без числа-хвоста
+        building_cadastral_number=None,
+    )
+    html = _make_cian_serp_html([offer])
+
+    scraper = CianScraper()
+    lots = scraper._parse_serp_html(html)
+
+    lot = lots[0]
+    addr = lot.address or ""
+    assert "Евгения Савкова" in addr
+    # никаких фейковых catch-all номеров (например 7019) в адресе
+    assert not addr.rstrip().endswith("7019")
+
+
+def test_format_address_vtorichka_houseless_untouched():
+    """Вторичка с house-less адресом НЕ трогается — номер не выдумываем."""
+    parts = [
+        {"type": "city", "name": "Екатеринбург"},
+        {"type": "street", "name": "улица Евгения Савкова"},
+    ]
+    result = _format_address(
+        parts,
+        building={"houseNumber": "29"},
+        building_cadastral_number="66:41:0313010:11396",
+        newbuilding={"id": 0, "name": ""},
+        is_newbuilding=False,
+    )
+    # вторичка: is_newbuilding=False → ни «29», ни кадастр не дописываются
+    assert "29" not in result
+    assert "66:41" not in result