Merge pull request 'feat(#3051): deals.doc_type + регион и тип документа как параметры импорта Росреестра' (#3421) from feat/3051-deals-doc-type-region-param into main

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

@@ -328,26 +328,36 @@ def import_rosreestr_dkp(
     SERVER gendesign_remote настроен в 060_postgres_fdw_extension.sql.
     USER MAPPING создаётся при startup в core/fdw.py (tradein_fdw_reader).
 
-    Область покрытия: вся Свердловская область (region_code=66), не только Екатеринбург —
-    прежний ILIKE-фильтр по подстроке города (ограничивавший импорт одним Екатеринбургом)
-    снят (Mera trade-in расширяется на весь регион, unlocks +47183 сделок вне ЕКБ уже
+    Область покрытия задаётся параметром, а не литералом (#3051 п.6): region_code
+    приходит из params, дефолт 66 = вся Свердловская область (не только Екатеринбург —
+    прежний ILIKE-фильтр по подстроке города снят, unlocks +47183 сделок вне ЕКБ уже
     сидящих в source foreign table). address и deals.city строятся из реального city
     источника (не хардкод "Екатеринбург"), deals.region_code заполняется из строки
-    источника (= 66 при текущем фильтре).
+    источника (= запрошенному региону при текущем фильтре).
+
+    Типы документов тоже параметр (#3051 п.3): doc_types, дефолт ['ДКП'] = прежнее
+    поведение (только вторичка — #549 / Fix_Rosreestr_Dkp_Filter_May24). Для Москвы
+    ДДУ идут по ценам котлована и медиану развалят, поэтому смешивать их с ДКП можно
+    только осознанно и с колонкой deals.doc_type (миграция 288), которая теперь
+    заполняется на импорте.
 
     Фильтры (совпадают с import-rosreestr.sh + Fix_Rosreestr_Dkp_Filter_May24):
-      - region_code = 66 (вся Свердловская область, все города)
+      - region_code = :region_code  (param, default 66 — вся Свердловская область)
       - city IS NOT NULL AND trim(city) != '' (непустой город → корректный address)
       - realestate_type_code = '002001003000' (квартира)
       - area BETWEEN 18 AND 200
       - deal_price BETWEEN 1000000 AND 100000000
       - street IS NOT NULL AND trim(street) != ''
-      - doc_type = 'ДКП'  (только вторичка — #549 / Fix_Rosreestr_Dkp_Filter_May24)
+      - doc_type = ANY(:doc_types)  (param, default ['ДКП'])
       - period_start_date >= since  (default '2024-01-01')
 
     dedup_hash: 'ros:dkp:' || id — плоский натуральный ключ (инъективный, без коллизий,
       human-readable). До #576 здесь был md5('ros:dkp:' || id); миграция 077 конвертировала
       существующие строки. source_id хранит исходный rosreestr id (дедуп переустанавливаем).
+      Префикс ':dkp:' НАМЕРЕННО оставлен неизменным после параметризации doc_types: id
+      уникален в источнике сам по себе, независимо от типа документа, поэтому ключ и без
+      того не коллизирует; а вот смена формы ключа осиротила бы все уже загруженные строки
+      (их пришлось бы конвертировать ещё одной миграцией — ровно то, что делала 077).
     Rooms: выводятся из площади (Росреестр не отдаёт кол-во комнат).
 
     Batch-процессинг: читаем из FDW батчами по batch_size через cursor-based пагинацию
@@ -364,6 +374,11 @@ def import_rosreestr_dkp(
     """
     since: str = str(params.get("since", "2024-01-01"))
     batch_size: int = int(params.get("batch_size", 2000))
+    # #3051 п.6: регион — параметр, дефолт 66 сохраняет текущее прод-поведение
+    # (расписание получает явный region_code в миграции 288).
+    region_code: int = int(params.get("region_code", 66))
+    # #3051 п.3: типы документов — параметр, дефолт ['ДКП'] = прежний литерал.
+    doc_types: list[str] = [str(t) for t in params.get("doc_types") or ["ДКП"]]
 
     counters: dict[str, int] = {
         "rows_fetched": 0,
@@ -472,21 +487,28 @@ def import_rosreestr_dkp(
                         year_build                               AS year_built,
                         round(deal_price)::bigint                AS price_rub,
                         round(price_per_sqm)::int                AS price_per_m2,
-                        period_start_date                        AS deal_date
+                        period_start_date                        AS deal_date,
+                        doc_type                                 AS doc_type
                     FROM gendesign_rosreestr_deals
-                    WHERE region_code = 66
+                    WHERE region_code = CAST(:region_code AS int)
                       AND city IS NOT NULL AND trim(city) <> ''
                       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 doc_type = ANY(CAST(:doc_types AS text[]))
                       AND period_start_date >= CAST(:since AS date)
                       AND id > CAST(:last_id AS bigint)
                     ORDER BY id
                     LIMIT CAST(:batch_size AS int)
                 """),
-                    {"since": since, "last_id": last_id, "batch_size": batch_size},
+                    {
+                        "since": since,
+                        "last_id": last_id,
+                        "batch_size": batch_size,
+                        "region_code": region_code,
+                        "doc_types": doc_types,
+                    },
                 )
                 .mappings()
                 .all()
@@ -524,7 +546,7 @@ def import_rosreestr_dkp(
                                 INSERT INTO deals (
                                     source, dedup_hash, source_id, address, region_code, city,
                                     rooms, area_m2, floor, year_built, price_rub, price_per_m2,
-                                    deal_date
+                                    deal_date, doc_type
                                 )
                                 VALUES (
                                     'rosreestr',
@@ -539,7 +561,8 @@ def import_rosreestr_dkp(
                                     CAST(:year_built AS int),
                                     CAST(:price_rub AS bigint),
                                     CAST(:price_per_m2 AS int),
-                                    CAST(:deal_date AS date)
+                                    CAST(:deal_date AS date),
+                                    CAST(:doc_type AS text)
                                 )
                                 ON CONFLICT (dedup_hash) DO UPDATE SET
                                     address = EXCLUDED.address,
@@ -551,7 +574,8 @@ def import_rosreestr_dkp(
                                     year_built = EXCLUDED.year_built,
                                     price_rub = EXCLUDED.price_rub,
                                     price_per_m2 = EXCLUDED.price_per_m2,
-                                    deal_date = EXCLUDED.deal_date
+                                    deal_date = EXCLUDED.deal_date,
+                                    doc_type = EXCLUDED.doc_type
                                 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
@@ -562,6 +586,7 @@ def import_rosreestr_dkp(
                                    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
+                                   OR deals.doc_type IS DISTINCT FROM EXCLUDED.doc_type
                                 RETURNING (xmax = 0) AS was_inserted
                             """),
                             {
@@ -577,6 +602,7 @@ def import_rosreestr_dkp(
                                 "price_rub": row["price_rub"],
                                 "price_per_m2": row["price_per_m2"],
                                 "deal_date": row["deal_date"],
+                                "doc_type": row["doc_type"],
                             },
                         ).fetchone()
                         if result is None:

tradein-mvp/backend/data/sql/288_deals_doc_type.sql

@@ -0,0 +1,67 @@
+-- 288_deals_doc_type.sql
+-- deals.doc_type — тип документа сделки Росреестра (#3051 п.3, п.6).
+--
+-- Dependencies: 002_core_tables.sql (deals), 015_scrape_runs.sql + 072_scrape_schedules_seed_cian_rosreestr.sql
+--               (scrape_schedules, строка source='rosreestr_dkp_import')
+-- Apply after: 287_proxy_run_attribution.sql
+--
+-- ЧАСТЬ A — WHY:
+--   Импорт Росреестра до сих пор ронял тип документа на пол: фильтр doc_type = 'ДКП'
+--   стоял литералом в WHERE, а в deals не приезжало ничего. Пока скоуп был один
+--   (Свердловская обл., только вторичка) это было безобидно — все строки источника
+--   'rosreestr' по построению ДКП. С расширением на Москву безобидность кончается:
+--   в источнике за 2024 по region_code=77 лежит 30 627 ДДУ с медианой 112 743 ₽/м²
+--   против 107 005 ДКП с медианой 256 250 ₽/м² — это цены котлована, и смешать их
+--   в одной таблице без различимого признака значит развалить любую оценку.
+--   Колонка нужна ДО того, как импорт начнёт тянуть больше одного типа.
+--
+--   ЧАСТЬ A — WHAT:
+--   doc_type text NULLable — источник (foreign table gendesign_rosreestr_deals.doc_type)
+--   тоже text и NULL допускает; строгий NOT NULL сломал бы не-росреестровые источники
+--   (etazhi / domklik_history), у которых понятия «тип документа» нет вовсе.
+--   Индекс НЕ добавляем: селективность низкая (2-3 значения), а все живые выборки
+--   по deals идут по region_code/deal_date/geom — doc_type там в лучшем случае
+--   довесок к уже отобранному диапазону. Появится запрос, который реально режет
+--   по doc_type на большом наборе — заведём частичный индекс тогда, по EXPLAIN.
+--
+--   ЧАСТЬ B — бэкфилл:
+--   Всё, что лежит в deals с source='rosreestr', прошло через WHERE doc_type = 'ДКП'
+--   (и в scheduler.import_rosreestr_dkp, и в deploy/import-rosreestr.sh, и в
+--   backfill-миграции 077) — других типов там физически быть не может. Поэтому
+--   проставить 'ДКП' задним числом корректно, а не эвристика.
+--   Остальные источники остаются NULL осознанно.
+--
+--   ЧАСТЬ C — расписание:
+--   scrape_schedules.default_params для 'rosreestr_dkp_import' получает явный
+--   region_code=66. Раньше регион был неявным дефолтом в коде; после параметризации
+--   (params.get('region_code', 66)) неявность становится ловушкой — прод-строка должна
+--   сама говорить, какой регион она тянет. doc_types в default_params НЕ пишем:
+--   дефолт ['ДКП'] в коде и есть текущее поведение, а запись его в расписание
+--   создала бы второе место, где надо не забыть поменять.
+--
+-- Идемпотентна, безопасна к повторному запуску.
+
+BEGIN;
+
+-- #2752: блокирующий DDL не должен вставать в очередь за чужой сессией и уводить
+-- за собой запросы приложения — лучше упасть по таймауту и повторить деплой.
+SET LOCAL lock_timeout = '5s';
+
+-- A: колонка
+ALTER TABLE deals ADD COLUMN IF NOT EXISTS doc_type text;
+
+COMMENT ON COLUMN deals.doc_type IS
+    'Тип документа сделки Росреестра (ДКП / ДДУ). NULL для источников без этого понятия.';
+
+-- B: бэкфилл — все rosreestr-строки прошли фильтр ДКП на импорте
+UPDATE deals
+   SET doc_type = 'ДКП'
+ WHERE source = 'rosreestr'
+   AND doc_type IS NULL;
+
+-- C: явный регион в расписании импорта вместо неявного дефолта в коде
+UPDATE scrape_schedules
+   SET default_params = default_params || '{"region_code": 66}'::jsonb
+ WHERE source = 'rosreestr_dkp_import';
+
+COMMIT;

tradein-mvp/backend/tests/test_rosreestr_dedup_key.py

@@ -151,23 +151,27 @@ def test_migration_077_converts_md5_to_plain_key() -> None:
 def test_migration_077_shared_filter_matches_live_import() -> None:
     """Дедуп-релевантные фильтры src CTE миграции 077 совпадают с живым импортом.
 
-    Исключение — city ILIKE (см. test_live_import_dropped_ekb_city_filter ниже):
-    077 backfill'ил ЕКБ-строки под EKB-only scope того времени; живой импорт расширен
-    на всю Свердловскую область (region_code=66, все города), поэтому city-фильтр из
-    живого импорта СНЯТ намеренно. Остальные клозы обязаны совпадать байт-в-байт,
-    иначе 077 конвертировал бы не тот набор строк.
+    Два исключения:
+      - city ILIKE (см. test_live_import_dropped_ekb_city_filter ниже): 077 backfill'ил
+        ЕКБ-строки под EKB-only scope того времени; живой импорт расширен на всю
+        Свердловскую область, поэтому city-фильтр из живого импорта СНЯТ намеренно;
+      - region_code / doc_type: в живом импорте это ПАРАМЕТРЫ (#3051), их совпадение
+        с 077 проверяется дефолтами params.get(..., 66) / ["ДКП"] в самом импорте.
+    Остальные клозы обязаны совпадать байт-в-байт, иначе 077 конвертировал бы не тот
+    набор строк.
     """
     sql = _MIGRATION_077.read_text("utf-8")
     for clause in (
-        "region_code = 66",
         "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}"
+    # Исторические литералы 077 остаются на месте (миграция применена, её не правят).
+    assert "region_code = 66" in sql
+    assert "doc_type = 'ДКП'" in sql
 
 
 def test_live_import_dropped_ekb_city_filter() -> None:

tradein-mvp/deploy/import-rosreestr.sh

@@ -2,13 +2,19 @@
 # Импорт реальных сделок Росреестра из gendesign-БД в tradein deals.
 #
 # Источник: gendesign-postgres-1 / rosreestr_deals (6.8М строк, партиц.).
-# Берём ЕКБ квартиры (realestate_type_code=002001003000) за последние ~18 мес.
+# Берём квартиры (realestate_type_code=002001003000) региона REGION_CODE — по умолчанию
+# 66 = вся Свердловская область, все города (не только ЕКБ: city-фильтр снят давно,
+# шапка про «ЕКБ квартиры» была неправдой).
 # rooms выводим из площади (Росреестр не отдаёт кол-во комнат).
 # Координаты — NULL, проставляются отдельно (geocode по street).
 #
-# Только ДКП (вторичка) — ДДУ застройщиков скёюят median, отделены сознательно. См. PR-A.
+# DOC_TYPE по умолчанию ДКП (вторичка): ДДУ идут по ценам котлована и скёюят median,
+# отделены сознательно (см. PR-A). С #3051 это параметр, а не литерал — для Москвы
+# (REGION_CODE=77) типы разделяются колонкой deals.doc_type (миграция 288), которую
+# скрипт теперь заполняет.
 #
 # Запуск на прод-хосте: ./import-rosreestr.sh
+#   REGION_CODE=77 DOC_TYPE='ДДУ' ./import-rosreestr.sh   # Москва, первичка
 # Повторяемо: дедуп по dedup_hash, новый запуск подтянет свежие кварталы.
 
 set -euo pipefail
@@ -18,8 +24,11 @@ DST_PG="${DST_PG:-tradein-postgres}"
 SRC_DB="${SRC_DB:-gendesign}"
 SRC_USER="${SRC_USER:-gendesign}"
 SINCE="${SINCE:-2024-01-01}"
+# #3051: регион и тип документа — параметры со старыми дефолтами (поведение не меняется).
+REGION_CODE="${REGION_CODE:-66}"
+DOC_TYPE="${DOC_TYPE:-ДКП}"
 
-echo "[$(date -u +%H:%M:%S)] import-rosreestr: ЕКБ квартиры с $SINCE"
+echo "[$(date -u +%H:%M:%S)] import-rosreestr: region=$REGION_CODE doc_type=$DOC_TYPE с $SINCE"
 
 # 1. Staging-таблица в tradein.
 docker exec "$DST_PG" psql -U tradein -d tradein -v ON_ERROR_STOP=on -c "
@@ -29,7 +38,8 @@ docker exec "$DST_PG" psql -U tradein -d tradein -v ON_ERROR_STOP=on -c "
   CREATE TABLE deals_ros_staging (
     dedup_hash text PRIMARY KEY, source_id text, address text, region_code int, city text,
     rooms int, area_m2 numeric,
-    floor int, year_built int, price_rub bigint, price_per_m2 int, deal_date date
+    floor int, year_built int, price_rub bigint, price_per_m2 int, deal_date date,
+    doc_type text
   );
 "
 
@@ -51,15 +61,16 @@ docker exec "$SRC_PG" psql -U "$SRC_USER" -d "$SRC_DB" -v ON_ERROR_STOP=on -c "
       year_build                                           AS year_built,
       round(deal_price)::bigint                            AS price_rub,
       round(price_per_sqm)::int                            AS price_per_m2,
-      period_start_date                                    AS deal_date
+      period_start_date                                    AS deal_date,
+      doc_type                                             AS doc_type
     FROM rosreestr_deals
-    WHERE region_code = 66
+    WHERE region_code = $REGION_CODE
       AND city IS NOT NULL AND trim(city) <> ''
       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 doc_type = '$DOC_TYPE'
       AND period_start_date >= '$SINCE'
   ) TO STDOUT WITH CSV
 " | docker exec -i "$DST_PG" psql -U tradein -d tradein -v ON_ERROR_STOP=on -c "
@@ -75,10 +86,10 @@ docker exec "$DST_PG" psql -U tradein -d tradein -v ON_ERROR_STOP=on -c "
 
   INSERT INTO deals (
     source, dedup_hash, source_id, address, region_code, city, rooms, area_m2, floor,
-    year_built, price_rub, price_per_m2, deal_date
+    year_built, price_rub, price_per_m2, deal_date, doc_type
   )
   SELECT 'rosreestr', dedup_hash, source_id, address, region_code, city, rooms, area_m2, floor,
-         year_built, price_rub, price_per_m2, deal_date
+         year_built, price_rub, price_per_m2, deal_date, doc_type
   FROM deals_ros_staging
   ON CONFLICT (dedup_hash) DO NOTHING;