feat(#3051): deals.doc_type + регион и тип документа как параметры импорта Росреестра

ПОЧЕМУ: расширение на Москву упирается в два литерала. В источнике за 2024 по региону 77
лежат 30 627 ДДУ с медианой 112 743 против 107 005 ДКП с медианой 256 250 — это цены
котлована, и без различимого признака в deals они развалят любую оценку. При этом тип
сделки терялся при загрузке вовсе (в deals колонки не было), а фильтры region_code = 66
и doc_type = 'ДКП' стояли литералами в scheduler.import_rosreestr_dkp и в двойнике
deploy/import-rosreestr.sh — сменить регион было нельзя, не правя код.

ЧТО:
- миграция 288: deals.doc_type text (idempotent) + бэкфилл 'ДКП' для source='rosreestr'
  (корректен, а не эвристика: всё загруженное прошло фильтр ДКП — и в импорте, и в 077)
  + явный region_code=66 в default_params расписания rosreestr_dkp_import вместо неявного
  дефолта в коде. Индекс НЕ добавлен: 2-3 значения, живые выборки идут по
  region_code/deal_date/geom — заведём частичный, когда появится режущий запрос;
- import_rosreestr_dkp: region_code (default 66) и doc_types (default ['ДКП']) из params,
  фильтры через bind-параметры CAST(:region_code AS int) / ANY(CAST(:doc_types AS text[])),
  doc_type едет из SELECT в INSERT и в ON CONFLICT DO UPDATE. Дефолты сохраняют текущее
  прод-поведение байт-в-байт;
- dedup_hash оставлен как 'ros:dkp:' || id: id уникален в источнике независимо от типа
  документа, а смена формы ключа осиротила бы уже загруженные строки (ровно то, что
  разгребала миграция 077);
- deploy/import-rosreestr.sh: REGION_CODE / DOC_TYPE как env со старыми дефолтами,
  doc_type протащен через staging в deals; шапка про «ЕКБ квартиры» переписана честно —
  city-фильтр снят давно, скоуп = весь регион;
- тесты: test_rosreestr_dedup_key переведён с ассертов на литералы на проверку
  «параметр + дефолт = скоуп 077»; новый test_3051_* проверяет bind-параметры реальным
  вызовом с моком Session, дефолты 66/['ДКП'], doc_type в колонках INSERT и текст 288.

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,63 @@
+-- 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;
+
+-- 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_3051_deals_doc_type_region_param.py

@@ -0,0 +1,148 @@
+"""#3051 п.3 + п.6: doc_type в deals и параметризация региона/типа документа на импорте.
+
+ЗАЧЕМ: для Москвы в источнике за 2024 лежат 30 627 ДДУ с медианой 112 743 ₽/м² против
+107 005 ДКП с медианой 256 250 — это цены котлована. Тип сделки терялся при загрузке
+(в deals не приезжал вовсе), а фильтры `region_code = 66` и `doc_type = 'ДКП'` стояли
+литералами, поэтому регион 77 было не подключить, не правя код. Тесты фиксируют, что
+параметры реально доезжают до bind-словаря, дефолты воспроизводят прежнее поведение,
+а doc_type пишется в deals.
+
+Чисто-юнит: ассертим SQL-текст import_rosreestr_dkp (inspect.getsource) + текст миграции
+288, плюс вызываем функцию с моком Session, чтобы увидеть фактические bind-параметры.
+Живой FDW / БД не нужны.
+"""
+
+import inspect
+import os
+import re
+from pathlib import Path
+from typing import Any
+from unittest.mock import MagicMock
+
+# Как в test_rosreestr_dedup_key.py: Settings требует DATABASE_URL до импорта.
+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_288 = _SQL_DIR / "288_deals_doc_type.sql"
+_IMPORT_SRC = inspect.getsource(scheduler.import_rosreestr_dkp)
+
+
+# ── SQL-форма импорта ────────────────────────────────────────────────────────
+
+
+def test_import_filters_region_and_doc_type_via_bind_params() -> None:
+    """Оба фильтра — bind-параметры, литералов 66 / 'ДКП' в SQL больше нет."""
+    assert "region_code = CAST(:region_code AS int)" in _IMPORT_SRC
+    assert "doc_type = ANY(CAST(:doc_types AS text[]))" in _IMPORT_SRC
+    assert "WHERE region_code = 66" not in _IMPORT_SRC
+    assert "AND doc_type = 'ДКП'" not in _IMPORT_SRC
+    # psycopg v3: ни один bind-параметр не приводится через :name::type.
+    assert not re.search(r":\w+::", _IMPORT_SRC)
+
+
+def test_import_writes_doc_type_into_deals() -> None:
+    """doc_type едет из SELECT в список колонок INSERT, в VALUES и в ON CONFLICT."""
+    assert "doc_type                                 AS doc_type" in _IMPORT_SRC
+    assert "deal_date, doc_type" in _IMPORT_SRC
+    assert "CAST(:doc_type AS text)" in _IMPORT_SRC
+    # Переопубликованный квартал может исправить тип документа — обновляем.
+    assert "doc_type = EXCLUDED.doc_type" in _IMPORT_SRC
+    assert "IS DISTINCT FROM EXCLUDED.doc_type" in _IMPORT_SRC
+    assert '"doc_type": row["doc_type"]' in _IMPORT_SRC
+
+
+def test_import_keeps_dedup_key_shape_unchanged() -> None:
+    """Ключ дедупа НЕ зависит от типа документа — иначе осиротели бы старые строки.
+
+    id уникален в источнике сам по себе; префикс 'ros:dkp:' остаётся историческим
+    именем формата (см. миграцию 077, которая уже конвертировала все строки).
+    """
+    assert "'ros:dkp:' || CAST(id AS text)           AS dedup_hash" in _IMPORT_SRC
+    assert "ON CONFLICT (dedup_hash) DO UPDATE SET" in _IMPORT_SRC
+
+
+# ── фактические bind-параметры (вызов с моком) ───────────────────────────────
+
+
+def _capture_select_params(params: dict[str, Any]) -> dict[str, Any]:
+    """Прогоняет import_rosreestr_dkp до первого SELECT из FDW и возвращает его binds.
+
+    Мок Session: cleanup-DELETE и _resume_dkp_cursor получают пустые результаты,
+    первый батч — пустой список, поэтому цикл завершается сразу после SELECT.
+    """
+    captured: list[dict[str, Any]] = []
+
+    def _execute(stmt: Any, binds: Any = None, *a: Any, **kw: Any) -> MagicMock:
+        sql = str(stmt)
+        if "FROM gendesign_rosreestr_deals" in sql and isinstance(binds, dict):
+            captured.append(binds)
+        result = MagicMock()
+        result.fetchall.return_value = []
+        result.fetchone.return_value = None
+        result.mappings.return_value.all.return_value = []
+        result.scalar.return_value = None
+        return result
+
+    db = MagicMock()
+    db.execute.side_effect = _execute
+
+    monkeys = {
+        "is_cancelled": False,
+    }
+    orig_is_cancelled = scheduler.runs_mod.is_cancelled
+    orig_mark_done = scheduler.runs_mod.mark_done
+    orig_heartbeat = scheduler.kit_runs.update_heartbeat
+    try:
+        scheduler.runs_mod.is_cancelled = lambda *a, **kw: monkeys["is_cancelled"]  # type: ignore[assignment]
+        scheduler.runs_mod.mark_done = lambda *a, **kw: None  # type: ignore[assignment]
+        scheduler.kit_runs.update_heartbeat = lambda *a, **kw: None  # type: ignore[assignment]
+        scheduler.import_rosreestr_dkp(db, 1, params)
+    finally:
+        scheduler.runs_mod.is_cancelled = orig_is_cancelled  # type: ignore[assignment]
+        scheduler.runs_mod.mark_done = orig_mark_done  # type: ignore[assignment]
+        scheduler.kit_runs.update_heartbeat = orig_heartbeat  # type: ignore[assignment]
+
+    assert captured, "SELECT из gendesign_rosreestr_deals не выполнился"
+    return captured[0]
+
+
+def test_defaults_preserve_current_production_behaviour() -> None:
+    """Без параметров — ровно прежний скоуп: Свердловская обл., только ДКП."""
+    binds = _capture_select_params({})
+    assert binds["region_code"] == 66
+    assert binds["doc_types"] == ["ДКП"]
+
+
+def test_params_override_region_and_doc_types() -> None:
+    """Москва + оба типа документа приходят из params, а не из кода."""
+    binds = _capture_select_params({"region_code": 77, "doc_types": ["ДКП", "ДДУ"]})
+    assert binds["region_code"] == 77
+    assert binds["doc_types"] == ["ДКП", "ДДУ"]
+
+
+def test_region_code_accepts_string_from_jsonb_params() -> None:
+    """default_params прилетают из jsonb — строковый регион не должен ломать CAST."""
+    binds = _capture_select_params({"region_code": "77"})
+    assert binds["region_code"] == 77
+    assert isinstance(binds["region_code"], int)
+
+
+# ── миграция 288 ─────────────────────────────────────────────────────────────
+
+
+def test_migration_288_adds_column_backfills_and_sets_schedule_region() -> None:
+    """288: idempotent-колонка, бэкфилл только rosreestr-строк, явный регион в расписании."""
+    assert _MIGRATION_288.is_file(), f"missing migration: {_MIGRATION_288}"
+    sql = _MIGRATION_288.read_text("utf-8")
+    assert "ALTER TABLE deals ADD COLUMN IF NOT EXISTS doc_type text" in sql
+    # Бэкфилл корректен ровно потому, что всё загруженное прошло фильтр ДКП.
+    assert "SET doc_type = 'ДКП'" in sql
+    assert "WHERE source = 'rosreestr'" in sql
+    assert "AND doc_type IS NULL" in sql
+    # Явный region_code в расписании вместо неявного дефолта в коде.
+    assert "UPDATE scrape_schedules" in sql
+    assert "'{\"region_code\": 66}'::jsonb" in sql
+    assert "source = 'rosreestr_dkp_import'" in sql
+    assert "BEGIN;" in sql and "COMMIT;" in sql

tradein-mvp/backend/tests/test_rosreestr_dedup_key.py

@@ -151,23 +151,45 @@ 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 проверяется через дефолты — см. тест ниже.
+    Остальные клозы обязаны совпадать байт-в-байт, иначе 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_parametrized_region_and_doc_type_keep_077_defaults() -> None:
+    """#3051: регион и типы документов — bind-параметры, но ДЕФОЛТЫ = скоуп 077.
+
+    Литералы `region_code = 66` / `doc_type = 'ДКП'` ушли из SQL живого импорта
+    (иначе регион 77 не подключить), но дефолты обязаны воспроизводить ровно тот
+    набор строк, который 077 когда-то backfill'ил, — иначе штатный прогон молча
+    сменил бы скоуп.
+    """
+    # В SQL — параметры, не литералы.
+    assert "region_code = CAST(:region_code AS int)" in _IMPORT_SRC
+    assert "doc_type = ANY(CAST(:doc_types AS text[]))" in _IMPORT_SRC
+    assert "region_code = 66" not in _IMPORT_BODY
+    assert "doc_type = 'ДКП'" not in _IMPORT_BODY.split("params.get", 1)[-1].split("SELECT", 1)[0]
+    # Дефолты — прежнее поведение.
+    assert 'params.get("region_code", 66)' in _IMPORT_SRC
+    assert 'params.get("doc_types") or ["ДКП"]' in _IMPORT_SRC
 
 
 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;