From 6487813ec54a8623c2cc64c3bfc90533186c0e12 Mon Sep 17 00:00:00 2001 From: bot-backend Date: Sun, 12 Jul 2026 22:12:31 +0300 Subject: [PATCH] =?UTF-8?q?fix(tradein/ingest):=204=20data-integrity=20?= =?UTF-8?q?=D0=B4=D0=B5=D1=84=D0=B5=D0=BA=D1=82=D0=B0=20=D0=B2=20=D0=B1?= =?UTF-8?q?=D1=8D=D0=BA=D1=84=D0=B8=D0=BB=D0=BB=D0=B5=20=D0=B3=D0=BE=D0=B4?= =?UTF-8?q?=D0=BE=D0=B2=20=D0=B8=20=D0=B8=D0=BC=D0=BF=D0=BE=D1=80=D1=82?= =?UTF-8?q?=D0=B5=20=D0=94=D0=9A=D0=9F?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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). --- .../app/services/domrf_kapremont_loader.py | 133 +++++++++++++++--- tradein-mvp/backend/app/services/scheduler.py | 91 ++++++++++-- .../tests/test_domrf_kapremont_loader.py | 60 +++++++- .../backend/tests/test_rosreestr_dedup_key.py | 64 ++++++++- 4 files changed, 315 insertions(+), 33 deletions(-) diff --git a/tradein-mvp/backend/app/services/domrf_kapremont_loader.py b/tradein-mvp/backend/app/services/domrf_kapremont_loader.py index 4678ed4d..750c5059 100644 --- a/tradein-mvp/backend/app/services/domrf_kapremont_loader.py +++ b/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) - except ValueError: - pass - try: - return int(float(s.replace(",", "."))) + value = int(s) except ValueError: + try: + 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} """ ) diff --git a/tradein-mvp/backend/app/services/scheduler.py b/tradein-mvp/backend/app/services/scheduler.py index c44c1aee..114ac577 100644 --- a/tradein-mvp/backend/app/services/scheduler.py +++ b/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, ) diff --git a/tradein-mvp/backend/tests/test_domrf_kapremont_loader.py b/tradein-mvp/backend/tests/test_domrf_kapremont_loader.py index 8522936c..74690b0e 100644 --- a/tradein-mvp/backend/tests/test_domrf_kapremont_loader.py +++ b/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 diff --git a/tradein-mvp/backend/tests/test_rosreestr_dedup_key.py b/tradein-mvp/backend/tests/test_rosreestr_dedup_key.py index 28e822f5..90176d3c 100644 --- a/tradein-mvp/backend/tests/test_rosreestr_dedup_key.py +++ b/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: