diff --git a/tradein-mvp/backend/app/services/scheduler.py b/tradein-mvp/backend/app/services/scheduler.py index 2094d01b..68acd4ec 100644 --- a/tradein-mvp/backend/app/services/scheduler.py +++ b/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"], diff --git a/tradein-mvp/backend/data/sql/077_dedup_hash_plain_key_backfill.sql b/tradein-mvp/backend/data/sql/077_dedup_hash_plain_key_backfill.sql new file mode 100644 index 00000000..16f839cf --- /dev/null +++ b/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; diff --git a/tradein-mvp/backend/tests/test_rosreestr_dedup_key.py b/tradein-mvp/backend/tests/test_rosreestr_dedup_key.py new file mode 100644 index 00000000..96390d58 --- /dev/null +++ b/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}" diff --git a/tradein-mvp/deploy/import-rosreestr.sh b/tradein-mvp/deploy/import-rosreestr.sh index 894c57fb..692cacb6 100755 --- a/tradein-mvp/deploy/import-rosreestr.sh +++ b/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;