fix(tradein): plain natural-key dedup_hash for rosreestr deals (#576) (#638)

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

@@ -352,7 +352,9 @@ def import_rosreestr_dkp(
       - doc_type = 'ДКП'  (только вторичка — #549 / Fix_Rosreestr_Dkp_Filter_May24)
       - period_start_date >= since  (default '2024-01-01')
 
-    dedup_hash: md5('ros:dkp:' || id) — соответствует #549 / Fix_Rosreestr_Dkp_Filter_May24.
+    dedup_hash: 'ros:dkp:' || id — плоский натуральный ключ (инъективный, без коллизий,
+      human-readable). До #576 здесь был md5('ros:dkp:' || id); миграция 077 конвертировала
+      существующие строки. source_id хранит исходный rosreestr id (дедуп переустанавливаем).
     Rooms: выводятся из площади (Росреестр не отдаёт кол-во комнат).
 
     Batch-процессинг: читаем из FDW батчами по batch_size через cursor-based пагинацию
@@ -413,7 +415,8 @@ def import_rosreestr_dkp(
                 text("""
                     SELECT
                         id,
-                        md5('ros:dkp:' || CAST(id AS text))     AS dedup_hash,
+                        id                                       AS source_id_src,
+                        'ros:dkp:' || CAST(id AS text)           AS dedup_hash,
                         'Екатеринбург, ' || trim(street)         AS address,
                         CASE
                             WHEN area < 30 THEN 0
@@ -467,12 +470,13 @@ def import_rosreestr_dkp(
                         inserted = db.execute(
                             text("""
                                 INSERT INTO deals (
-                                    source, dedup_hash, address, rooms, area_m2,
+                                    source, dedup_hash, source_id, address, rooms, area_m2,
                                     floor, year_built, price_rub, price_per_m2, deal_date
                                 )
                                 VALUES (
                                     'rosreestr',
                                     CAST(:dedup_hash AS text),
+                                    CAST(:source_id AS text),
                                     CAST(:address AS text),
                                     CAST(:rooms AS int),
                                     CAST(:area_m2 AS numeric),
@@ -487,6 +491,7 @@ def import_rosreestr_dkp(
                             """),
                             {
                                 "dedup_hash": row["dedup_hash"],
+                                "source_id": str(row["source_id_src"]),
                                 "address": row["address"],
                                 "rooms": row["rooms"],
                                 "area_m2": row["area_m2"],

tradein-mvp/backend/data/sql/077_dedup_hash_plain_key_backfill.sql

@@ -0,0 +1,48 @@
+-- 077: #576 — migrate rosreestr deals dedup_hash md5('ros:dkp:'||id) -> plain 'ros:dkp:'||id
+--
+-- До этой миграции dedup_hash хранил md5-хэш натурального ключа: md5('ros:dkp:' || id)
+-- (collision-prone, не human-readable, дедуп не восстановить из значения). После — плоский
+-- натуральный ключ 'ros:dkp:N' (инъективный, zero-collision, обратимый в исходный id).
+--
+-- source_id никогда не записывался (NULL для всех 49 791 rosreestr-строк), поэтому исходный
+-- id восстанавливаем из FDW foreign table gendesign_rosreestr_deals, матча по тому же md5.
+-- Заодно проставляем source_id = id (дедуп переустанавливаем навсегда).
+--
+-- Idempotent: повторный прогон затрагивает только строки, оставшиеся в md5-форме
+-- (после первого прогона их 0 — UPDATE no-op). Плоские ключи ('ros:dkp:N') физически
+-- не могут совпасть с md5-hex (разный формат), поэтому транзиентного UNIQUE-violation
+-- по deals_dedup_hash_key во время UPDATE не возникает.
+--
+-- FDW dependency: читает foreign table gendesign_rosreestr_deals (SERVER gendesign_remote,
+-- создан в 060_postgres_fdw_extension.sql; foreign table — в 072_..._cian_rosreestr.sql;
+-- USER MAPPING — backend startup core/fdw.py). Деплой применяет миграции ПОСЛЕ `compose up -d`,
+-- т.е. FDW-сервер уже поднят и foreign table читаема.
+--
+-- Фильтр src CTE — байт-в-байт совпадает с live-импортом (scheduler.py:437-444 /
+-- import-rosreestr.sh), чтобы каждая импортированная сделка нашлась и сконвертировалась.
+-- Apply after: 076_account_estimate_quota.sql
+
+BEGIN;
+
+WITH src AS (
+    SELECT
+        id,
+        md5('ros:dkp:' || CAST(id AS text)) AS h
+    FROM gendesign_rosreestr_deals
+    WHERE region_code = 66
+      AND city ILIKE '%катеринбург%'
+      AND realestate_type_code = '002001003000'
+      AND area BETWEEN 18 AND 200
+      AND deal_price BETWEEN 1000000 AND 100000000
+      AND street IS NOT NULL AND trim(street) <> ''
+      AND doc_type = 'ДКП'
+      AND period_start_date >= DATE '2024-01-01'
+)
+UPDATE deals d
+SET dedup_hash = 'ros:dkp:' || CAST(src.id AS text),
+    source_id  = CAST(src.id AS text)
+FROM src
+WHERE d.source = 'rosreestr'
+  AND d.dedup_hash = src.h;
+
+COMMIT;

tradein-mvp/backend/tests/test_rosreestr_dedup_key.py

@@ -0,0 +1,105 @@
+"""Unit-тесты для формата dedup_hash rosreestr ДКП-сделок (#576).
+
+До #576 dedup_hash хранил md5('ros:dkp:' || id) — collision-prone, не human-readable,
+дедуп не восстановить из значения. После — плоский натуральный ключ 'ros:dkp:N'
+(инъективный, zero-collision, обратимый в исходный rosreestr id).
+
+Чисто-юнит: ассертим SQL-текст, эмитируемый import_rosreestr_dkp (через inspect.getsource),
+плюс содержимое backfill-миграции 077 — без живого FDW / DB-фикстуры.
+"""
+import inspect
+import os
+from pathlib import Path
+
+# Импорт app.services.scheduler тянет app.core.config.Settings → требует DATABASE_URL.
+# Ставим заглушку ДО импорта (как в test_scheduler.py) — тест чисто-статический, БД не трогает.
+os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test")
+
+from app.services import scheduler
+
+_SQL_DIR = Path(__file__).resolve().parents[1] / "data" / "sql"
+_MIGRATION_077 = _SQL_DIR / "077_dedup_hash_plain_key_backfill.sql"
+
+# Источник функции импорта — читаем один раз.
+_IMPORT_SRC = inspect.getsource(scheduler.import_rosreestr_dkp)
+# Тело без docstring: docstring упоминает старое md5-выражение в пояснительных целях,
+# поэтому negative-ассерты («md5 больше нет») гоняем по коду, а не по докстрингу.
+_IMPORT_BODY = _IMPORT_SRC.split('"""', 2)[-1]
+
+
+def test_import_uses_plain_natural_key_not_md5() -> None:
+    """SELECT-выражение dedup_hash — плоский ключ 'ros:dkp:' || id, без md5()."""
+    # Плоский натуральный ключ присутствует как SELECT-выражение dedup_hash.
+    assert "'ros:dkp:' || CAST(id AS text)           AS dedup_hash" in _IMPORT_SRC
+    # md5-обёртки натурального ключа в коде больше нет (старое выражение #549).
+    assert "md5('ros:dkp:'" not in _IMPORT_BODY
+    assert "md5(" not in _IMPORT_BODY
+
+
+def test_import_populates_source_id() -> None:
+    """Новые строки сохраняют исходный rosreestr id в source_id (дедуп переустанавливаем)."""
+    # source_id попадает в SELECT, в список колонок INSERT и в params.
+    assert "AS source_id_src" in _IMPORT_SRC
+    assert "source, dedup_hash, source_id," in _IMPORT_SRC
+    assert "CAST(:source_id AS text)" in _IMPORT_SRC
+    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_psycopg_v3_casts_not_double_colon() -> None:
+    """psycopg v3: параметры приводятся через CAST(:x AS type), никогда :x::type.
+
+    Литеральные ::-касты на колонках FDW (round(deal_price)::bigint и т.п.) допустимы —
+    запрет касается только :param::type, который psycopg v3 ломает.
+    """
+    import re
+
+    # Ни один bind-параметр (:name) не должен использовать ::type.
+    assert not re.search(r":\w+::", _IMPORT_SRC)
+    # Параметры приводятся через CAST(:param AS type).
+    assert "CAST(:since AS date)" in _IMPORT_SRC
+    assert "CAST(:last_id AS bigint)" in _IMPORT_SRC
+    assert "CAST(:source_id AS text)" in _IMPORT_SRC
+
+
+# ── backfill-миграция 077 ────────────────────────────────────────────────────
+
+
+def test_migration_077_exists() -> None:
+    assert _MIGRATION_077.is_file(), f"missing migration: {_MIGRATION_077}"
+
+
+def test_migration_077_converts_md5_to_plain_key() -> None:
+    """077 матчит существующие строки по md5 и пишет плоский ключ + source_id."""
+    sql = _MIGRATION_077.read_text("utf-8")
+    # Восстанавливаем id из FDW по тому же md5, что писался раньше.
+    assert "md5('ros:dkp:' || CAST(id AS text))" in sql
+    assert "FROM gendesign_rosreestr_deals" in sql
+    # Целевые значения — плоский ключ и source_id.
+    assert "dedup_hash = 'ros:dkp:' || CAST(src.id AS text)" in sql
+    assert "source_id  = CAST(src.id AS text)" in sql
+    # Только rosreestr-строки, матч по старому md5-значению.
+    assert "d.source = 'rosreestr'" in sql
+    assert "d.dedup_hash = src.h" in sql
+    # Транзакционно.
+    assert "BEGIN;" in sql and "COMMIT;" in sql
+
+
+def test_migration_077_filter_matches_live_import() -> None:
+    """Фильтр src CTE байт-в-байт совпадает с живым импортом → все строки сконвертируются."""
+    sql = _MIGRATION_077.read_text("utf-8")
+    for clause in (
+        "region_code = 66",
+        "city ILIKE '%катеринбург%'",
+        "realestate_type_code = '002001003000'",
+        "area BETWEEN 18 AND 200",
+        "deal_price BETWEEN 1000000 AND 100000000",
+        "street IS NOT NULL AND trim(street) <> ''",
+        "doc_type = 'ДКП'",
+    ):
+        assert clause in sql, f"missing filter clause in migration: {clause!r}"
+        assert clause in _IMPORT_SRC, f"missing filter clause in import: {clause!r}"

tradein-mvp/deploy/import-rosreestr.sh

@@ -23,18 +23,21 @@ echo "[$(date -u +%H:%M:%S)] import-rosreestr: ЕКБ квартиры с $SINCE
 
 # 1. Staging-таблица в tradein.
 docker exec "$DST_PG" psql -U tradein -d tradein -v ON_ERROR_STOP=on -c "
-  CREATE TABLE IF NOT EXISTS deals_ros_staging (
-    dedup_hash text PRIMARY KEY, address text, rooms int, area_m2 numeric,
+  -- DROP перед CREATE: страхует от устаревшей формы staging-таблицы, если прошлый
+  -- запуск упал между CREATE и финальным DROP под старым кодом (без source_id).
+  DROP TABLE IF EXISTS deals_ros_staging;
+  CREATE TABLE deals_ros_staging (
+    dedup_hash text PRIMARY KEY, source_id text, address text, rooms int, area_m2 numeric,
     floor int, year_built int, price_rub bigint, price_per_m2 int, deal_date date
   );
-  TRUNCATE deals_ros_staging;
 "
 
 # 2. Трансформируем на gendesign-стороне, пайпим CSV в staging.
 docker exec "$SRC_PG" psql -U "$SRC_USER" -d "$SRC_DB" -v ON_ERROR_STOP=on -c "
   COPY (
     SELECT
-      md5('ros:dkp:' || id::text)                          AS dedup_hash,
+      'ros:dkp:' || id::text                               AS dedup_hash,
+      id::text                                             AS source_id,
       'Екатеринбург, ' || trim(street)                     AS address,
       CASE
         WHEN area < 30 THEN 0  WHEN area < 44 THEN 1
@@ -68,10 +71,10 @@ docker exec "$DST_PG" psql -U tradein -d tradein -v ON_ERROR_STOP=on -c "
   DELETE FROM deals WHERE address = 'Екатеринбург, реальная сделка';
 
   INSERT INTO deals (
-    source, dedup_hash, address, rooms, area_m2, floor,
+    source, dedup_hash, source_id, address, rooms, area_m2, floor,
     year_built, price_rub, price_per_m2, deal_date
   )
-  SELECT 'rosreestr', dedup_hash, address, rooms, area_m2, floor,
+  SELECT 'rosreestr', dedup_hash, source_id, address, rooms, area_m2, floor,
          year_built, price_rub, price_per_m2, deal_date
   FROM deals_ros_staging
   ON CONFLICT (dedup_hash) DO NOTHING;