fix(tradein/ingest): 4 data-integrity дефекта в бэкфилле годов и импорте ДКП

A. ДОМ.РФ year backfill: авторитетный commission_year больше не экранируется
   невозможным существующим year_built. Валидное существующее значение
   выигрывает (COALESCE-семантика #2013 сохранена), но NULL/impossible
   (< 1850, > текущий+2, 0) заменяется валидным commission_year → zhkh_year.
   parse_int_field получил sanity-гейт [min,max] — мусорный commission_year
   не попадает в staging (7 таких строк в текущем staging).

B. propagate_listings_year: добавлен link-consistency guard. Раньше копировал
   houses.year_built на listings по house_id_fk без проверки — перепривязанный
   FK впрыскивал чужой когортный год. Теперь пропагация только если координаты
   объявления в пределах 500м от геометрии дома (ST_DWithin), либо (без коорд.)
   консервативный address-фоллбек по short_address. Live: блокирует 1 из 10
   текущих кандидатов (>500м mislink).

C. rosreestr_dkp_import: per-row INSERT-ошибки отделены от dedup-skip. Раньше
   except инкрементил тот же batch_skipped, что и легитимный ON CONFLICT — сбой
   маскировался под дедуп и run рапортовал success. Отдельный rows_errored +
   гейт по доле ошибок (> 5% → run FAILED, не silent-green).

D. rosreestr_dkp_import: ON CONFLICT DO NOTHING → DO UPDATE изменяемых сырых
   фактов Росреестра (price/area/rooms/floor/year/deal_date/...), чтобы
   исправленный/переопубликованный квартал обновлялся. Обогащение
   (lat/lon/geom/geocode_tried_at и пр.) не в SET-списке — не затирается.
   IS DISTINCT FROM guard сохраняет идемпотентность resume; RETURNING (xmax=0)
   отличает insert от update (rows_inserted vs rows_updated).

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

@@ -38,6 +38,7 @@ import tempfile
 import zipfile
 from collections.abc import Iterator
 from dataclasses import dataclass
+from datetime import date
 from pathlib import Path
 
 import httpx
@@ -61,6 +62,19 @@ DOWNLOAD_TIMEOUT_SEC = 180
 # сбойный чанк откатывается изолированно, остальные доезжают.
 UPSERT_CHUNK_SIZE = 500
 
+# Правдоподобные границы года постройки. Значение вне диапазона — почти наверняка
+# опечатка/мусор: для МКД-реестра капремонта дом < 1850 нереален, а > текущий+2 ещё
+# физически не введён в эксплуатацию. Такой год НЕ должен (а) попадать из ДОМ.РФ CSV
+# в staging (parse-гейт в parse_int_field), (б) экранировать авторитетный ДОМ.РФ
+# commission_year при бэкфилле houses.year_built (SQL-гейт в _BACKFILL_*_SQL).
+YEAR_BUILT_MIN = 1850
+YEAR_BUILT_FUTURE_SLACK = 2  # дом «в стройке» — текущий год + 2 ещё допустим
+
+
+def plausible_year_max() -> int:
+    """Верхняя граница правдоподобного года постройки (текущий год + slack)."""
+    return date.today().year + YEAR_BUILT_FUTURE_SLACK
+
 
 # ─────────────────────────────────────────────────────────────────────────────
 # Чистые хелперы парсинга (юнит-тестируются без сети/БД)
@@ -78,25 +92,40 @@ def parse_decimal_comma(raw: str | None) -> float | None:
         return None
 
 
-def parse_int_field(raw: str | None) -> int | None:
+def parse_int_field(
+    raw: str | None,
+    *,
+    min_value: int | None = None,
+    max_value: int | None = None,
+) -> int | None:
     """Устойчивый str → int|None (commission_year / number_floors_max).
 
     Пусто/None/нечисло → None. Терпит decimal-строки («5,0») через float-фоллбек —
     ДОМ.РФ CSV в принципе может так отдать целочисленные поля.
+
+    min_value/max_value — опциональный sanity-гейт: распарсенное значение вне
+    [min_value, max_value] трактуется как None (для commission_year передаём
+    YEAR_BUILT_MIN..plausible_year_max(), чтобы мусорный год не попал в staging;
+    для number_floors_max границы не задаём — там гейт не требуется).
     """
     if raw is None:
         return None
     s = raw.strip()
     if not s:
         return None
+    value: int | None = None
     try:
-        return int(s)
+        value = int(s)
     except ValueError:
-        pass
         try:
-        return int(float(s.replace(",", ".")))
+            value = int(float(s.replace(",", ".")))
         except ValueError:
             return None
+    if min_value is not None and value < min_value:
+        return None
+    if max_value is not None and value > max_value:
+        return None
+    return value
 
 
 @dataclass(slots=True)
@@ -146,7 +175,13 @@ def parse_kr11_csv(path: str | Path) -> dict[str, Kr11Row]:
             mkd_code=mkd_code,
             houseguid=(raw.get("houseguid") or "").strip() or None,
             address=(raw.get("address") or "").strip() or None,
-            commission_year=parse_int_field(raw.get("commission_year")),
+            # sanity-гейт: год вне [1850, текущий+2] — мусор, не пускаем в staging
+            # (иначе он позже мог бы записаться в houses.year_built как «авторитетный»).
+            commission_year=parse_int_field(
+                raw.get("commission_year"),
+                min_value=YEAR_BUILT_MIN,
+                max_value=plausible_year_max(),
+            ),
             number_floors_max=parse_int_field(raw.get("number_floors_max")),
             total_sq=parse_decimal_comma(raw.get("total_sq")),
         )
@@ -342,19 +377,46 @@ def load_domrf_kapremont(
 # Backfill houses.year_built/material_walls/total_floors
 # ─────────────────────────────────────────────────────────────────────────────
 # Шаг 1: дома с domrf-матчем — year_built/material_walls/total_floors из domrf,
-# С ЗАФОЛЖЕННЫМ zhkh_year/zhkh_floors фоллбеком внутри ТОЙ ЖЕ строки COALESCE
-# (COALESCE(year_built, domrf.commission_year, zhkh_year) — ровно как в issue).
-# Только заполняет NULL — никогда не перезаписывает существующий non-null year_built.
+# С ЗАФОЛЖЕННЫМ zhkh_year/zhkh_floors фоллбеком (COALESCE, как в issue #2013).
+#
+# ПРИОРИТЕТ year_built (Fix A): ВАЛИДНОЕ существующее значение выигрывает (COALESCE
+# семантика сохранена для valid). НО невозможное существующее значение (< :ymin,
+# > :ymax или NULL) НЕ должно экранировать авторитетный ДОМ.РФ commission_year —
+# для домов реестра капремонта commission_year это источник истины по году ввода.
+# Поэтому impossible/NULL заменяется валидным commission_year → затем валидным
+# zhkh_year; если валидной замены нет — оставляем существующее как есть (не обнуляем
+# даже мусор, чтобы не терять данные без выигрыша). Кандидаты замены тоже проходят
+# плаузибилити-гейт [:ymin, :ymax], чтобы в houses.year_built не попал мусорный год.
+# material_walls/total_floors — прежняя COALESCE-семантика (только заполнить NULL).
 _BACKFILL_HOUSES_FROM_DOMRF_SQL = text(
     """
     UPDATE houses h
-    SET year_built = COALESCE(h.year_built, d.commission_year, h.zhkh_year),
+    SET year_built = CASE
+            WHEN h.year_built BETWEEN CAST(:ymin AS int) AND CAST(:ymax AS int)
+                THEN h.year_built
+            ELSE COALESCE(
+                CASE WHEN d.commission_year BETWEEN CAST(:ymin AS int) AND CAST(:ymax AS int)
+                    THEN d.commission_year END,
+                CASE WHEN h.zhkh_year BETWEEN CAST(:ymin AS int) AND CAST(:ymax AS int)
+                    THEN h.zhkh_year END,
+                h.year_built
+            )
+        END,
         material_walls = COALESCE(h.material_walls, d.wall_material),
         total_floors = COALESCE(h.total_floors, d.number_floors_max, h.zhkh_floors)
     FROM domrf_kapremont d
     WHERE d.houseguid = COALESCE(h.gar_house_guid, h.house_fias_id, h.zhkh_house_guid)
       AND (
-            (h.year_built IS NULL AND COALESCE(d.commission_year, h.zhkh_year) IS NOT NULL)
+            (
+              (h.year_built IS NULL
+               OR h.year_built NOT BETWEEN CAST(:ymin AS int) AND CAST(:ymax AS int))
+              AND COALESCE(
+                    CASE WHEN d.commission_year BETWEEN CAST(:ymin AS int) AND CAST(:ymax AS int)
+                        THEN d.commission_year END,
+                    CASE WHEN h.zhkh_year BETWEEN CAST(:ymin AS int) AND CAST(:ymax AS int)
+                        THEN h.zhkh_year END
+                  ) IS NOT NULL
+            )
          OR (h.material_walls IS NULL AND d.wall_material IS NOT NULL)
          OR (h.total_floors IS NULL AND COALESCE(d.number_floors_max, h.zhkh_floors) IS NOT NULL)
       )
@@ -367,7 +429,16 @@ _BACKFILL_HOUSES_FROM_DOMRF_COUNT_SQL = text(
     FROM houses h
     JOIN domrf_kapremont d
       ON d.houseguid = COALESCE(h.gar_house_guid, h.house_fias_id, h.zhkh_house_guid)
-    WHERE (h.year_built IS NULL AND COALESCE(d.commission_year, h.zhkh_year) IS NOT NULL)
+    WHERE (
+            (h.year_built IS NULL
+             OR h.year_built NOT BETWEEN CAST(:ymin AS int) AND CAST(:ymax AS int))
+            AND COALESCE(
+                  CASE WHEN d.commission_year BETWEEN CAST(:ymin AS int) AND CAST(:ymax AS int)
+                      THEN d.commission_year END,
+                  CASE WHEN h.zhkh_year BETWEEN CAST(:ymin AS int) AND CAST(:ymax AS int)
+                      THEN h.zhkh_year END
+                ) IS NOT NULL
+          )
        OR (h.material_walls IS NULL AND d.wall_material IS NOT NULL)
        OR (h.total_floors IS NULL AND COALESCE(d.number_floors_max, h.zhkh_floors) IS NOT NULL)
     """
@@ -412,8 +483,9 @@ def backfill_houses_from_domrf(db: Session, *, dry_run: bool = False) -> dict[st
     Не коммитит (caller). dry_run — ноль записей, только SELECT count(*) по тем же
     предикатам.
     """
+    year_bounds = {"ymin": YEAR_BUILT_MIN, "ymax": plausible_year_max()}
     if dry_run:
-        domrf_matched = db.execute(_BACKFILL_HOUSES_FROM_DOMRF_COUNT_SQL).scalar_one()
+        domrf_matched = db.execute(_BACKFILL_HOUSES_FROM_DOMRF_COUNT_SQL, year_bounds).scalar_one()
         zhkh_fallback = db.execute(_BACKFILL_HOUSES_ZHKH_FALLBACK_COUNT_SQL).scalar_one()
         result = {
             "domrf_matched": domrf_matched,
@@ -423,7 +495,7 @@ def backfill_houses_from_domrf(db: Session, *, dry_run: bool = False) -> dict[st
         logger.info("backfill_houses_from_domrf DRY-RUN: %s", result)
         return result
 
-    domrf_updated = db.execute(_BACKFILL_HOUSES_FROM_DOMRF_SQL).rowcount
+    domrf_updated = db.execute(_BACKFILL_HOUSES_FROM_DOMRF_SQL, year_bounds).rowcount
     zhkh_updated = db.execute(_BACKFILL_HOUSES_ZHKH_FALLBACK_SQL).rowcount
 
     result = {
@@ -441,24 +513,51 @@ def backfill_houses_from_domrf(db: Session, *, dry_run: bool = False) -> dict[st
 # Только NULL listings.year_built, только когда houses.year_built уже известен —
 # никогда не перезаписывает существующий listings.year_built (source-provided данные
 # приоритетнее houses-агрегата).
+#
+# LINK-CONSISTENCY GUARD (Fix B): раньше пропагация шла по house_id_fk БЕЗ проверки,
+# что listing реально относится к этому дому. Перепривязанный FK (house dedup/merge
+# «съезжает» на дом за сотни метров) впрыскивал бы чужой когортный год в живой comp.
+# Пропагируем только если координаты самого объявления в пределах
+# PROPAGATE_MAX_DISTANCE_M от геометрии дома. Если у объявления нет координат —
+# консервативный address-фоллбек: адрес объявления должен содержать short_address
+# дома (тот же дом по адресной строке); нет short_address — не пропагируем.
+PROPAGATE_MAX_DISTANCE_M = 500
+
+# Общий link-consistency предикат для UPDATE и COUNT (единый источник — одна правка
+# порога PROPAGATE_MAX_DISTANCE_M меняет обе SQL). Порог — int-константа, не bind-param
+# (в ST_DWithin проще литерал; значение доверенное, не из пользовательского ввода).
+_PROPAGATE_LINK_GUARD = f"""
+      AND (
+            (l.geom IS NOT NULL AND h.geom IS NOT NULL
+                AND ST_DWithin(
+                    CAST(l.geom AS geography), CAST(h.geom AS geography),
+                    {PROPAGATE_MAX_DISTANCE_M}))
+         OR (l.geom IS NULL
+                AND h.short_address IS NOT NULL AND l.address IS NOT NULL
+                AND l.address ILIKE '%' || h.short_address || '%')
+      )
+"""
+
 _PROPAGATE_LISTINGS_YEAR_SQL = text(
-    """
+    f"""
     UPDATE listings l
     SET year_built = h.year_built
     FROM houses h
     WHERE l.house_id_fk = h.id
       AND l.year_built IS NULL
       AND h.year_built IS NOT NULL
+    {_PROPAGATE_LINK_GUARD}
     """
 )
 
 _PROPAGATE_LISTINGS_YEAR_COUNT_SQL = text(
-    """
+    f"""
     SELECT count(*)
     FROM listings l
     JOIN houses h ON l.house_id_fk = h.id
     WHERE l.year_built IS NULL
       AND h.year_built IS NOT NULL
+    {_PROPAGATE_LINK_GUARD}
     """
 )
 

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

@@ -37,6 +37,11 @@ from app.services import scrape_runs as runs_mod
 
 logger = logging.getLogger(__name__)
 
+# import_rosreestr_dkp: доля per-row INSERT-ошибок (rows_errored / rows_fetched), выше
+# которой прогон помечается FAILED, а не silent-green (Fix C). Единичные битые строки
+# (редкий bad row) не валят импорт; систематический сбой (≈100% ошибок) — валит.
+DKP_IMPORT_ERROR_RATE_THRESHOLD = 0.05
+
 
 def compute_next_run_at(
     window_start_hour: int,
@@ -217,7 +222,16 @@ def import_rosreestr_dkp(
     counters: dict[str, int] = {
         "rows_fetched": 0,
         "rows_inserted": 0,
+        # rows_updated: ON CONFLICT DO UPDATE обновил существующую строку
+        # (исправленный/переопубликованный квартал — Fix D).
+        "rows_updated": 0,
+        # rows_skipped: ТОЛЬКО легитимный dedup-пропуск (строка уже есть, факты
+        # идентичны — DO UPDATE ... WHERE distinct не сработал).
         "rows_skipped": 0,
+        # rows_errored: реальные per-row INSERT-ошибки, отделены от dedup-skip (Fix C),
+        # раньше обе категории клались в rows_skipped → систематический сбой выглядел
+        # как обычный дедуп и прогон рапортовал success.
+        "rows_errored": 0,
         "batches_done": 0,
     }
 
@@ -255,8 +269,9 @@ def import_rosreestr_dkp(
                 # #1182 Phase 2: кооперативный SIGTERM-drain (деплой recreate scraper).
                 # Это НЕ user-cancel → mark_done (partial), не mark_cancelled. Курсор
                 # уже зафиксирован update_heartbeat'ом каждый батч; следующий run
-                # пере-сканирует с id=0 (INSERT ... ON CONFLICT DO NOTHING идемпотентен)
-                # → уже вставленные строки пропускаются, остаток до-импортируется.
+                # пере-сканирует с id=0 — INSERT ... ON CONFLICT DO UPDATE идемпотентен
+                # для неизменных строк (WHERE ... IS DISTINCT FROM пропускает совпадающие
+                # факты), поэтому повторный проход не плодит дубликаты и не трогает данные.
                 # mark_done выводит run из 'running' → reap_zombies его не тронет.
                 logger.info(
                     "rosreestr_dkp_import run_id=%d: SIGTERM-drain — committing partial "
@@ -327,7 +342,9 @@ def import_rosreestr_dkp(
 
             total_batches += 1
             batch_inserted = 0
+            batch_updated = 0
             batch_skipped = 0
+            batch_errored = 0
             batch_max_id = last_id
 
             for row in batch_rows:
@@ -337,7 +354,17 @@ def import_rosreestr_dkp(
 
                 try:
                     with db.begin_nested():  # SAVEPOINT per row
-                        inserted = db.execute(
+                        # Fix D: ON CONFLICT DO UPDATE вместо прежнего no-op-дедупа — при
+                        # повторном импорте ИСПРАВЛЕННОГО/переопубликованного квартала обновляем
+                        # сырые факты Росреестра. WHERE ... IS DISTINCT FROM оставляет
+                        # неизменные строки нетронутыми (идемпотентность resume: строка
+                        # без изменений → 0 returned → legit dedup-skip). source/source_id
+                        # (identity) и dedup_hash (ключ конфликта) стабильны, не трогаем.
+                        # Обогащение (lat/lon/geom/geocode_tried_at, cadastral_number,
+                        # total_floors, house_type ...) НЕ в EXCLUDED-списке → сохраняется.
+                        # RETURNING (xmax = 0): freshly-inserted → xmax=0 (was_inserted),
+                        # обновлённая по ON CONFLICT → xmax<>0; отличаем insert от update.
+                        result = db.execute(
                             text("""
                                 INSERT INTO deals (
                                     source, dedup_hash, source_id, address, region_code, city,
@@ -359,8 +386,28 @@ def import_rosreestr_dkp(
                                     CAST(:price_per_m2 AS int),
                                     CAST(:deal_date AS date)
                                 )
-                                ON CONFLICT (dedup_hash) DO NOTHING
-                                RETURNING id
+                                ON CONFLICT (dedup_hash) DO UPDATE SET
+                                    address = EXCLUDED.address,
+                                    region_code = EXCLUDED.region_code,
+                                    city = EXCLUDED.city,
+                                    rooms = EXCLUDED.rooms,
+                                    area_m2 = EXCLUDED.area_m2,
+                                    floor = EXCLUDED.floor,
+                                    year_built = EXCLUDED.year_built,
+                                    price_rub = EXCLUDED.price_rub,
+                                    price_per_m2 = EXCLUDED.price_per_m2,
+                                    deal_date = EXCLUDED.deal_date
+                                WHERE deals.address IS DISTINCT FROM EXCLUDED.address
+                                   OR deals.region_code IS DISTINCT FROM EXCLUDED.region_code
+                                   OR deals.city IS DISTINCT FROM EXCLUDED.city
+                                   OR deals.rooms IS DISTINCT FROM EXCLUDED.rooms
+                                   OR deals.area_m2 IS DISTINCT FROM EXCLUDED.area_m2
+                                   OR deals.floor IS DISTINCT FROM EXCLUDED.floor
+                                   OR deals.year_built IS DISTINCT FROM EXCLUDED.year_built
+                                   OR deals.price_rub IS DISTINCT FROM EXCLUDED.price_rub
+                                   OR deals.price_per_m2 IS DISTINCT FROM EXCLUDED.price_per_m2
+                                   OR deals.deal_date IS DISTINCT FROM EXCLUDED.deal_date
+                                RETURNING (xmax = 0) AS was_inserted
                             """),
                             {
                                 "dedup_hash": row["dedup_hash"],
@@ -377,25 +424,33 @@ def import_rosreestr_dkp(
                                 "deal_date": row["deal_date"],
                             },
                         ).fetchone()
-                        if inserted is not None:
+                        if result is None:
+                            # DO UPDATE ... WHERE distinct не сработал → строка есть и
+                            # факты идентичны = легитимный dedup-skip (Fix C: только это
+                            # теперь считается skip, INSERT-ошибки — отдельно ниже).
+                            batch_skipped += 1
+                        elif result[0]:
                             batch_inserted += 1
                         else:
-                            batch_skipped += 1
+                            batch_updated += 1
                 except Exception as exc:
+                    # Fix C: реальная per-row INSERT-ошибка — ОТДЕЛЬНЫЙ счётчик, не skip.
                     logger.warning(
                         "rosreestr_dkp_import run_id=%d: row id=%d INSERT failed: %s",
                         run_id,
                         row_id,
                         exc,
                     )
-                    batch_skipped += 1
+                    batch_errored += 1
 
             last_id = batch_max_id
             db.commit()
 
             counters["rows_fetched"] += len(batch_rows)
             counters["rows_inserted"] += batch_inserted
+            counters["rows_updated"] += batch_updated
             counters["rows_skipped"] += batch_skipped
+            counters["rows_errored"] += batch_errored
             counters["batches_done"] = total_batches
             counters["last_id"] = last_id  # type: ignore[assignment]
 
@@ -403,26 +458,42 @@ def import_rosreestr_dkp(
             runs_mod.update_heartbeat(db, run_id, counters)
             logger.info(
                 "rosreestr_dkp_import run_id=%d: batch=%d fetched=%d "
-                "inserted=%d skipped=%d last_id=%d",
+                "inserted=%d updated=%d skipped=%d errored=%d last_id=%d",
                 run_id,
                 total_batches,
                 len(batch_rows),
                 batch_inserted,
+                batch_updated,
                 batch_skipped,
+                batch_errored,
                 last_id,
             )
 
             if len(batch_rows) < batch_size:
                 break  # Last partial batch — no more rows
 
+        # Fix C: систематический per-row INSERT-сбой больше не рапортует success.
+        # Отделив rows_errored от rows_skipped, проверяем долю ошибок: выше порога —
+        # прогон FAILED (raise → внешний except → mark_failed), а не silent-green.
+        fetched = counters["rows_fetched"]
+        errored = counters["rows_errored"]
+        if fetched > 0 and errored / fetched > DKP_IMPORT_ERROR_RATE_THRESHOLD:
+            raise RuntimeError(
+                f"rosreestr_dkp_import: per-row INSERT error rate "
+                f"{errored}/{fetched} ({errored / fetched:.1%}) exceeds "
+                f"{DKP_IMPORT_ERROR_RATE_THRESHOLD:.0%} threshold — marking run failed"
+            )
+
         runs_mod.mark_done(db, run_id, counters)
         logger.info(
             "rosreestr_dkp_import run_id=%d done: "
-            "total_fetched=%d inserted=%d skipped=%d batches=%d",
+            "total_fetched=%d inserted=%d updated=%d skipped=%d errored=%d batches=%d",
             run_id,
             counters["rows_fetched"],
             counters["rows_inserted"],
+            counters["rows_updated"],
             counters["rows_skipped"],
+            counters["rows_errored"],
             total_batches,
         )
 

tradein-mvp/backend/tests/test_domrf_kapremont_loader.py

@@ -141,6 +141,28 @@ def test_parse_int_field() -> None:
     assert dkl.parse_int_field("5,0") == 5  # decimal-фоллбек
 
 
+def test_parse_int_field_year_sanity_gate() -> None:
+    # Fix A: значение вне [min_value, max_value] → None (мусорный год не в staging).
+    ymax = dkl.plausible_year_max()
+    assert dkl.parse_int_field("1841", min_value=1850, max_value=ymax) is None  # слишком старый
+    assert dkl.parse_int_field("0", min_value=1850, max_value=ymax) is None  # ноль
+    assert dkl.parse_int_field(str(ymax + 5), min_value=1850, max_value=ymax) is None  # будущее
+    assert dkl.parse_int_field("1964", min_value=1850, max_value=ymax) == 1964  # валидный
+    assert dkl.parse_int_field("1850", min_value=1850, max_value=ymax) == 1850  # граница вкл.
+    # Без границ (number_floors_max) — гейт не применяется.
+    assert dkl.parse_int_field("1841") == 1841
+
+
+def test_parse_kr11_gates_impossible_commission_year(tmp_path: Path) -> None:
+    # Fix A: строка КР1.1 с невозможным годом (1200) → commission_year=None.
+    bad_row = _KR11_ROW_3.replace("1964;", "1200;", 1)
+    csv_text = "\n".join([_KR11_HEADER, bad_row]) + "\n"
+    p = tmp_path / "bad_year.csv"
+    p.write_text(csv_text, encoding="utf-8-sig")
+    rows = dkl.parse_kr11_csv(p)
+    assert rows["2170017185"].commission_year is None
+
+
 # ─────────────────────────────────────────────────────────────────────────────
 # parse_kr11_csv — house registry
 # ─────────────────────────────────────────────────────────────────────────────
@@ -286,9 +308,17 @@ _BACKFILL_DOMRF_SQL = re.sub(r"\s+", " ", str(dkl._BACKFILL_HOUSES_FROM_DOMRF_SQ
 _BACKFILL_ZHKH_SQL = re.sub(r"\s+", " ", str(dkl._BACKFILL_HOUSES_ZHKH_FALLBACK_SQL.text))
 
 
-def test_backfill_domrf_sql_coalesce_order_matches_spec() -> None:
-    # COALESCE(year_built, domrf.commission_year, zhkh_year) — ровно в этом порядке (#2013).
-    assert "COALESCE(h.year_built, d.commission_year, h.zhkh_year)" in _BACKFILL_DOMRF_SQL
+def test_backfill_domrf_sql_year_priority_and_plausibility_guard() -> None:
+    bound = "BETWEEN CAST(:ymin AS int) AND CAST(:ymax AS int)"
+    # Fix A: year_built — валидное существующее (в границах :ymin..:ymax) выигрывает,
+    # иначе (NULL/impossible) заменяется авторитетным commission_year → zhkh_year.
+    assert f"WHEN h.year_built {bound}" in _BACKFILL_DOMRF_SQL
+    assert "THEN h.year_built" in _BACKFILL_DOMRF_SQL
+    # commission_year и zhkh_year тоже проходят плаузибилити-гейт перед записью.
+    assert f"CASE WHEN d.commission_year {bound}" in _BACKFILL_DOMRF_SQL
+    # Невозможный существующий год НЕ экранирует замену (WHERE ловит NOT BETWEEN).
+    assert f"h.year_built NOT {bound}" in _BACKFILL_DOMRF_SQL
+    # material_walls / total_floors — прежняя COALESCE-семантика (только заполнить NULL).
     assert "COALESCE(h.material_walls, d.wall_material)" in _BACKFILL_DOMRF_SQL
     assert "COALESCE(h.total_floors, d.number_floors_max, h.zhkh_floors)" in _BACKFILL_DOMRF_SQL
 
@@ -339,6 +369,10 @@ def test_backfill_houses_real_run_sums_both_steps_rowcount() -> None:
     executed_sql = [str(call.args[0]) for call in db.execute.call_args_list]
     assert "UPDATE houses" in executed_sql[0]
     assert "UPDATE houses" in executed_sql[1]
+    # Fix A: domrf-шаг прокидывает плаузибилити-границы года как bind-параметры.
+    domrf_params = db.execute.call_args_list[0].args[1]
+    assert domrf_params["ymin"] == dkl.YEAR_BUILT_MIN
+    assert domrf_params["ymax"] == dkl.plausible_year_max()
 
 
 def test_backfill_houses_dry_run_only_counts_no_update() -> None:
@@ -367,6 +401,26 @@ def test_propagate_sql_only_fills_null_listings_year() -> None:
     assert "l.house_id_fk = h.id" in _PROPAGATE_SQL
 
 
+def test_propagate_sql_has_link_consistency_distance_guard() -> None:
+    # Fix B: пропагация только если координаты объявления в пределах порога от дома
+    # (ST_DWithin, geography → метры) — перепривязанный FK не впрыснет чужой год.
+    assert "ST_DWithin(" in _PROPAGATE_SQL
+    assert "CAST(l.geom AS geography)" in _PROPAGATE_SQL
+    assert "CAST(h.geom AS geography)" in _PROPAGATE_SQL
+    assert str(dkl.PROPAGATE_MAX_DISTANCE_M) in _PROPAGATE_SQL
+    # Оба SQL (UPDATE и COUNT) несут один и тот же guard.
+    count_sql = re.sub(r"\s+", " ", str(dkl._PROPAGATE_LISTINGS_YEAR_COUNT_SQL.text))
+    assert "ST_DWithin(" in count_sql
+
+
+def test_propagate_sql_no_coords_falls_back_to_address_check() -> None:
+    # Fix B: объявление без координат → консервативный address-фоллбек
+    # (адрес объявления содержит short_address дома), не пропагируем «вслепую».
+    assert "l.geom IS NULL" in _PROPAGATE_SQL
+    assert "h.short_address IS NOT NULL" in _PROPAGATE_SQL
+    assert "l.address ILIKE" in _PROPAGATE_SQL
+
+
 def test_propagate_listings_year_real_run_returns_rowcount() -> None:
     db = MagicMock()
     db.execute.return_value.rowcount = 42

tradein-mvp/backend/tests/test_rosreestr_dedup_key.py

@@ -46,9 +46,67 @@ def test_import_populates_source_id() -> None:
     assert '"source_id": str(row["source_id_src"])' in _IMPORT_SRC
 
 
-def test_import_keeps_on_conflict_dedup_hash() -> None:
-    """Дедуп по-прежнему через ON CONFLICT (dedup_hash) DO NOTHING."""
-    assert "ON CONFLICT (dedup_hash) DO NOTHING" in _IMPORT_SRC
+def test_import_uses_on_conflict_do_update_for_corrections() -> None:
+    """Fix D: дедуп по dedup_hash, но ON CONFLICT DO UPDATE (не DO NOTHING) —
+
+    исправленный/переопубликованный квартал обновляет сырые факты Росреестра.
+    """
+    assert "ON CONFLICT (dedup_hash) DO UPDATE SET" in _IMPORT_SRC
+    assert "DO NOTHING" not in _IMPORT_BODY
+    # Обновляются изменяемые сырые факты...
+    for col in (
+        "address = EXCLUDED.address",
+        "region_code = EXCLUDED.region_code",
+        "city = EXCLUDED.city",
+        "rooms = EXCLUDED.rooms",
+        "area_m2 = EXCLUDED.area_m2",
+        "floor = EXCLUDED.floor",
+        "year_built = EXCLUDED.year_built",
+        "price_rub = EXCLUDED.price_rub",
+        "price_per_m2 = EXCLUDED.price_per_m2",
+        "deal_date = EXCLUDED.deal_date",
+    ):
+        assert col in _IMPORT_SRC, f"missing DO UPDATE SET column: {col!r}"
+    # ...но идемпотентность через IS DISTINCT FROM (неизменные строки не трогаются).
+    assert "IS DISTINCT FROM EXCLUDED.price_rub" in _IMPORT_SRC
+    # xmax-дискриминатор insert vs update.
+    assert "RETURNING (xmax = 0)" in _IMPORT_SRC
+
+
+def test_import_do_update_never_overwrites_enrichment_columns() -> None:
+    """Fix D: обогащение (lat/lon/geom/geocode/cadastral_number...) не в SET-списке.
+
+    Эти колонки заполняются пост-импортом (geocode_deals_*) — повторный импорт не
+    должен их затирать. Отсутствие в SET DO UPDATE гарантирует сохранность.
+    """
+    conflict_clause = _IMPORT_SRC.split("ON CONFLICT (dedup_hash) DO UPDATE SET", 1)[1]
+    set_and_where = conflict_clause.split("RETURNING", 1)[0]
+    for enrich in (
+        "lat =",
+        "lon =",
+        "geom =",
+        "geocode_tried_at =",
+        "cadastral_number =",
+        "total_floors =",
+        "house_type =",
+    ):
+        assert enrich not in set_and_where, f"enrichment column leaked into UPDATE: {enrich!r}"
+
+
+def test_import_separates_errors_from_dedup_skips() -> None:
+    """Fix C: INSERT-ошибки считаются отдельным счётчиком, не как dedup-skip."""
+    assert '"rows_errored": 0' in _IMPORT_SRC
+    assert '"rows_updated": 0' in _IMPORT_SRC
+    assert "batch_errored += 1" in _IMPORT_SRC
+    # Per-row except-ветка (после лога "INSERT failed", до конца батча) инкрементит
+    # ТОЛЬКО errored — больше НЕ skip (это был баг silent-green: ошибки маскировались
+    # под дедуп).
+    per_row_except = _IMPORT_SRC.split("INSERT failed", 1)[1].split("last_id = batch_max_id", 1)[0]
+    assert "batch_errored += 1" in per_row_except
+    assert "batch_skipped += 1" not in per_row_except
+    # Порог ошибок → run failed, а не silent-green.
+    assert "DKP_IMPORT_ERROR_RATE_THRESHOLD" in _IMPORT_SRC
+    assert "marking run failed" in _IMPORT_SRC
 
 
 def test_import_uses_psycopg_v3_casts_not_double_colon() -> None: