fix(tradein/ingest): раз-хардкод rosreestr DKP импорта на всю обл. 66

- scheduler.py import_rosreestr_dkp: снят фильтр city ILIKE '%катеринбург%', address из реального city источника, deals.region_code + новая deals.city заполняются (было: хардкод 'Екатеринбург,' + region_code NULL)
- migration 177: deals.city + индекс + бэкфилл существующих EKB-строк region_code=66/city
- guard city IS NOT NULL → address не NULL для ~15 null-city строк источника
- sync deploy/import-rosreestr.sh (ops-fallback) под тот же oblast-scope
- split dedup-теста: 077 (историческая) хранит EKB-фильтр, живой импорт — нет

Открывает +47183 не-ЕКБ сделок region 66, уже сидящих в источнике, ранее резавшихся на импорте.

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

@@ -180,9 +180,16 @@ 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 сделок вне ЕКБ уже
+    сидящих в source foreign table). address и deals.city строятся из реального city
+    источника (не хардкод "Екатеринбург"), deals.region_code заполняется из строки
+    источника (= 66 при текущем фильтре).
+
     Фильтры (совпадают с import-rosreestr.sh + Fix_Rosreestr_Dkp_Filter_May24):
-      - region_code = 66 (Свердловская область)
-      - city ILIKE '%катеринбург%'
+      - region_code = 66 (вся Свердловская область, все города)
+      - city IS NOT NULL AND trim(city) != '' (непустой город → корректный address)
       - realestate_type_code = '002001003000' (квартира)
       - area BETWEEN 18 AND 200
       - deal_price BETWEEN 1000000 AND 100000000
@@ -273,7 +280,9 @@ def import_rosreestr_dkp(
                         id,
                         id                                       AS source_id_src,
                         'ros:dkp:' || CAST(id AS text)           AS dedup_hash,
-                        'Екатеринбург, ' || trim(street)         AS address,
+                        trim(city) || ', ' || trim(street)       AS address,
+                        region_code,
+                        trim(city)                               AS city,
                         CASE
                             WHEN area < 30 THEN 0
                             WHEN area < 44 THEN 1
@@ -296,7 +305,7 @@ def import_rosreestr_dkp(
                         period_start_date                        AS deal_date
                     FROM gendesign_rosreestr_deals
                     WHERE region_code = 66
-                      AND city ILIKE '%катеринбург%'
+                      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
@@ -331,14 +340,17 @@ def import_rosreestr_dkp(
                         inserted = db.execute(
                             text("""
                                 INSERT INTO deals (
-                                    source, dedup_hash, source_id, address, rooms, area_m2,
-                                    floor, year_built, price_rub, price_per_m2, deal_date
+                                    source, dedup_hash, source_id, address, region_code, city,
+                                    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(:region_code AS int),
+                                    CAST(:city AS text),
                                     CAST(:rooms AS int),
                                     CAST(:area_m2 AS numeric),
                                     CAST(:floor_num AS int),
@@ -354,6 +366,8 @@ def import_rosreestr_dkp(
                                 "dedup_hash": row["dedup_hash"],
                                 "source_id": str(row["source_id_src"]),
                                 "address": row["address"],
+                                "region_code": row["region_code"],
+                                "city": row["city"],
                                 "rooms": row["rooms"],
                                 "area_m2": row["area_m2"],
                                 "floor_num": row["floor_num"],

tradein-mvp/backend/data/sql/177_deals_city_region.sql

@@ -0,0 +1,51 @@
+-- 177_deals_city_region.sql
+-- deals.city + region_code backfill — расширение Mera trade-in estimator с
+-- Екатеринбурга на ВСЮ Свердловскую область (region_code=66, все города).
+--
+-- WHY:
+--   Ночной import_rosreestr_dkp (scheduler.py, source='rosreestr_dkp_import')
+--   до сих пор фильтровал `city ILIKE '%катеринбург%'` и хардкодил
+--   'Екатеринбург, ' в address, а deals.region_code никогда не заполнялся
+--   (NULL на всех 49791 текущих строках). Foreign table
+--   gendesign_rosreestr_deals уже отдаёт чистые city + region_code из
+--   source-таблицы (migration 072) — снятие фильтра открывает +47183
+--   не-ЕКБ сделок, уже сидящих в источнике.
+--
+-- WHAT:
+--   - deals.region_code уже существует (002_core_tables.sql), но deals.city
+--     отсутствовал (было только текстовое поле address с хардкод-префиксом
+--     'Екатеринбург, '). Добавляем city text — импортёр (scheduler.py,
+--     тот же PR) теперь заполняет его из реального source.city.
+--   - Индекс (city, deal_date) — под будущие per-city аналитические запросы
+--     (аналог deals_source_idx для (source, deal_date)).
+--   - Backfill существующих 49791 строк: все текущие deals с source='rosreestr'
+--     импортированы ДО этой миграции старым EKB-only кодом, т.е. фактически
+--     все они — Екатеринбург (то же допущение, что зашито в address
+--     'Екатеринбург, ' + street). Простановка region_code=66, city='Екатеринбург'
+--     задним числом делает старые строки self-describing наравне с новыми.
+--
+-- SCOPE: только schema (city column + index) + backfill старых rosreestr-строк.
+--   Новые импорты (после деплоя scheduler.py) заполняют оба поля сами.
+--
+-- IDEMPOTENCY / SAFETY:
+--   - ADD COLUMN IF NOT EXISTS / CREATE INDEX IF NOT EXISTS — безопасный re-run.
+--   - Backfill UPDATE ограничен `region_code IS NULL OR city IS NULL` —
+--     повторный прогон no-op (после первого раза условие уже false).
+--
+-- Dependencies: 002_core_tables.sql (deals, deals.region_code)
+
+BEGIN;
+
+ALTER TABLE deals ADD COLUMN IF NOT EXISTS city text;
+
+CREATE INDEX IF NOT EXISTS deals_city_deal_date_idx ON deals (city, deal_date);
+
+-- Backfill: все существующие rosreestr-сделки импортированы старым EKB-only
+-- кодом (до этой миграции) => гарантированно Екатеринбург, region_code=66.
+UPDATE deals
+   SET region_code = 66,
+       city = 'Екатеринбург'
+ WHERE source = 'rosreestr'
+   AND (region_code IS NULL OR city IS NULL);
+
+COMMIT;

tradein-mvp/backend/tests/test_rosreestr_dedup_key.py

@@ -7,6 +7,7 @@
 Чисто-юнит: ассертим SQL-текст, эмитируемый import_rosreestr_dkp (через inspect.getsource),
 плюс содержимое backfill-миграции 077 — без живого FDW / DB-фикстуры.
 """
+
 import inspect
 import os
 from pathlib import Path
@@ -89,12 +90,18 @@ def test_migration_077_converts_md5_to_plain_key() -> None:
     assert "BEGIN;" in sql and "COMMIT;" in sql
 
 
-def test_migration_077_filter_matches_live_import() -> None:
-    """Фильтр src CTE байт-в-байт совпадает с живым импортом → все строки сконвертируются."""
+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 конвертировал бы не тот набор строк.
+    """
     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",
@@ -103,3 +110,17 @@ def test_migration_077_filter_matches_live_import() -> None:
     ):
         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}"
+
+
+def test_live_import_dropped_ekb_city_filter() -> None:
+    """Живой импорт БОЛЬШЕ не фильтрует по городу — Mera расширена на всю обл. 66.
+
+    Историческая миграция 077 (уже применена на прод, трогать нельзя) сохраняет свой
+    EKB-only city-фильтр; живой импорт — нет. Guard против случайного возврата
+    ЕКБ-скоупа, который снова отрезал бы +47k не-ЕКБ сделок.
+    """
+    sql = _MIGRATION_077.read_text("utf-8")
+    # 077 — исторический EKB-scope, city-фильтр там остаётся неизменным.
+    assert "city ILIKE '%катеринбург%'" in sql
+    # Живой импорт city-фильтр снял намеренно (region_code=66 = вся область).
+    assert "ILIKE '%катеринбург%'" not in _IMPORT_SRC

tradein-mvp/deploy/import-rosreestr.sh

@@ -27,7 +27,8 @@ docker exec "$DST_PG" psql -U tradein -d tradein -v ON_ERROR_STOP=on -c "
   -- запуск упал между 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,
+    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
   );
 "
@@ -38,7 +39,9 @@ docker exec "$SRC_PG" psql -U "$SRC_USER" -d "$SRC_DB" -v ON_ERROR_STOP=on -c "
     SELECT
       'ros:dkp:' || id::text                               AS dedup_hash,
       id::text                                             AS source_id,
-      'Екатеринбург, ' || trim(street)                     AS address,
+      trim(city) || ', ' || trim(street)                   AS address,
+      region_code                                          AS region_code,
+      trim(city)                                           AS city,
       CASE
         WHEN area < 30 THEN 0  WHEN area < 44 THEN 1
         WHEN area < 62 THEN 2  WHEN area < 85 THEN 3
@@ -51,7 +54,7 @@ docker exec "$SRC_PG" psql -U "$SRC_USER" -d "$SRC_DB" -v ON_ERROR_STOP=on -c "
       period_start_date                                    AS deal_date
     FROM rosreestr_deals
     WHERE region_code = 66
-      AND city ILIKE '%катеринбург%'
+      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
@@ -71,10 +74,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, source_id, address, rooms, area_m2, floor,
+    source, dedup_hash, source_id, address, region_code, city, rooms, area_m2, floor,
     year_built, price_rub, price_per_m2, deal_date
   )
-  SELECT 'rosreestr', dedup_hash, source_id, address, rooms, area_m2, floor,
+  SELECT 'rosreestr', dedup_hash, source_id, address, region_code, city, rooms, area_m2, floor,
          year_built, price_rub, price_per_m2, deal_date
   FROM deals_ros_staging
   ON CONFLICT (dedup_hash) DO NOTHING;