refactor(tradein): unify kadastr_num -> cadastral_number (#732)

p3 naming-consistency (cadastral cols 0% filled). Migration 094_cadastral_unify.sql
(idempotent DO-block IF EXISTS): drop+recreate listings_search_mv (MV) +
v_data_quality (view), RENAME deals/house_metadata kadastr_num->cadastral_number,
DROP listings.kadastr_num + listings_kadastr_idx. Code SQL-string renames in
search_query/estimator/base (Lot dataclass field NOT renamed). 240 tests pass.

WARN: destructive DDL auto-applies to prod on deploy -> MERGE-GATE post-demo.
Reviewer: verify recreated MV/view DDL vs LIVE pg_get_viewdef before apply
(reconstructed from migration history 050/046, not live capture).

Refs #732

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

@@ -3420,7 +3420,7 @@ def _fetch_deals(
                 rooms, area_m2, floor, total_floors,
                 price_rub, price_per_m2,
                 deal_date, days_on_market,
-                kadastr_num,
+                cadastral_number,
                 ST_Distance(geom::geography, ST_MakePoint(:lon, :lat)::geography) AS distance_m
             FROM deals
             WHERE ST_DWithin(geom::geography, ST_MakePoint(:lon, :lat)::geography, :radius)
@@ -3644,7 +3644,7 @@ def _deal_to_analog(row: dict[str, Any]) -> AnalogLot:
         T0_per_house — kadastr_num exact match (НЕ доступно в open dataset Росреестра)
         T1_per_street — street-level only (default для всех ДКП open dataset)
     """
-    kad = row.get("kadastr_num")
+    kad = row.get("cadastral_number")
     # Per-house tier требует kadastr_num типа "66:41:0204016:10" (с участком).
     # Street-only patterns: "66:41:0000000:0" или NULL → T1.
     tier = "T0_per_house" if kad and not kad.endswith(":0") else "T1_per_street"

tradein-mvp/backend/app/services/scrapers/base.py

@@ -249,7 +249,7 @@ def save_listings(
                     source, source_url, source_id, dedup_hash,
                     address, lat, lon, region_code,
                     rooms, area_m2, floor, total_floors, year_built,
-                    house_type, repair_state, has_balcony, kadastr_num,
+                    house_type, repair_state, has_balcony,
                     house_source, house_ext_id, house_url, listing_segment,
                     price_rub, price_per_m2,
                     listing_date, days_on_market, photo_urls, raw_payload,
@@ -263,7 +263,7 @@ def save_listings(
                     :source, :source_url, :source_id, :dedup,
                     :address, :lat, :lon, 66,
                     :rooms, :area_m2, :floor, :total_floors, :year_built,
-                    :house_type, :repair_state, :has_balcony, :kadastr,
+                    :house_type, :repair_state, :has_balcony,
                     :house_source, :house_ext_id, :house_url, :listing_segment,
                     :price_rub, :ppm2,
                     :listing_date, :days_on_market,
@@ -317,7 +317,6 @@ def save_listings(
                 "house_type": lot.house_type,
                 "repair_state": lot.repair_state,
                 "has_balcony": lot.has_balcony,
-                "kadastr": lot.kadastr_num,
                 "house_source": lot.house_source,
                 "house_ext_id": lot.house_ext_id,
                 "house_url": lot.house_url,

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

@@ -84,7 +84,7 @@ def build_search_query(params: SearchParams) -> tuple[str, dict[str, object]]:
         args["fl_total_max"] = params.floors_total_max
 
     if params.has_kadastr:
-        where.append("kadastr_num IS NOT NULL")
+        where.append("cadastral_number IS NOT NULL")
 
     if params.sources:
         where.append("sources && CAST(:sources AS text[])")
@@ -111,7 +111,7 @@ def build_search_query(params: SearchParams) -> tuple[str, dict[str, object]]:
     sql = (
         "SELECT listing_id, source, source_url, address, lat, lng, "
         "rooms, total_area, floor, total_floors, price_rub, price_per_m2, "
-        "kadastr_num, scraped_at, "
+        "cadastral_number, scraped_at, "
         "house_id, year_built, house_class, developer_name, "
         "house_rating, house_ratings_count, "
         "source_count, sources, has_avito, has_cian, has_yandex, "

tradein-mvp/backend/data/sql/094_cadastral_unify.sql

@@ -0,0 +1,230 @@
+-- 094_cadastral_unify.sql
+-- Purpose: Унификация имени кадастрового столбца kadastr_num → cadastral_number
+--          во всех таблицах tradein-БД.
+--
+-- Контекст (Issue #732):
+--   deals.kadastr_num          → RENAME TO cadastral_number
+--   house_metadata.kadastr_num → RENAME TO cadastral_number
+--   listings.kadastr_num       → DROP (данные 0%; listings.cadastral_number уже существует
+--                                       — добавлен в 019_listings_alter_cian.sql)
+--   listings_search_mv         → DROP + RECREATE (ссылался на l.kadastr_num → l.cadastral_number)
+--   v_data_quality             → DROP + RECREATE (ссылался на kadastr_num в active_listings CTE)
+--
+-- Idempotency:
+--   * RENAME: обёрнуты в DO $$ ... $$ блок с проверкой information_schema.columns.
+--     Повторный запуск: столбец уже переименован → IF NOT EXISTS → skip.
+--   * DROP/DROP IF EXISTS: безопасны при повторном запуске.
+--   * RECREATE: CREATE OR REPLACE VIEW / CREATE MATERIALIZED VIEW после DROP IF EXISTS.
+--
+-- ВНИМАНИЕ: DESTRUCTIVE DDL. Применять только после проверки данных на проде.
+--   AUTO-APPLIES ON DEPLOY — merge-gate требуется (post-demo).
+--
+-- Dependencies:
+--   002_core_tables.sql (deals, listings, house_metadata)
+--   019_listings_alter_cian.sql (listings.cadastral_number already exists)
+--   046_views.sql (v_data_quality, v_price_divergence)
+--   050_search_optimization.sql (listings_search_mv)
+
+BEGIN;
+
+-- =============================================================================
+-- 1. Drop dependent objects (view + materialized view)
+--    Нужно до RENAME/DROP колонок, иначе PG откажет (зависимые объекты).
+-- =============================================================================
+
+DROP MATERIALIZED VIEW IF EXISTS listings_search_mv;
+DROP VIEW IF EXISTS v_data_quality;
+
+-- =============================================================================
+-- 2. Rename deals.kadastr_num → cadastral_number (idempotent)
+-- =============================================================================
+
+DO $$
+BEGIN
+    IF EXISTS (
+        SELECT 1 FROM information_schema.columns
+         WHERE table_schema = 'public'
+           AND table_name   = 'deals'
+           AND column_name  = 'kadastr_num'
+    ) THEN
+        ALTER TABLE deals RENAME COLUMN kadastr_num TO cadastral_number;
+    END IF;
+END
+$$;
+
+-- =============================================================================
+-- 3. Rename house_metadata.kadastr_num → cadastral_number (idempotent)
+--    house_metadata.kadastr_num имеет UNIQUE constraint — переименование сохраняет его.
+-- =============================================================================
+
+DO $$
+BEGIN
+    IF EXISTS (
+        SELECT 1 FROM information_schema.columns
+         WHERE table_schema = 'public'
+           AND table_name   = 'house_metadata'
+           AND column_name  = 'kadastr_num'
+    ) THEN
+        ALTER TABLE house_metadata RENAME COLUMN kadastr_num TO cadastral_number;
+    END IF;
+END
+$$;
+
+-- =============================================================================
+-- 4. Удаляем listings.kadastr_num + его индекс
+--    listings.cadastral_number (кадастровый номер квартиры, из Cian) уже существует
+--    с 019_listings_alter_cian.sql — данные в kadastr_num 0%, копировать не нужно.
+-- =============================================================================
+
+DROP INDEX IF EXISTS listings_kadastr_idx;
+
+DO $$
+BEGIN
+    IF EXISTS (
+        SELECT 1 FROM information_schema.columns
+         WHERE table_schema = 'public'
+           AND table_name   = 'listings'
+           AND column_name  = 'kadastr_num'
+    ) THEN
+        ALTER TABLE listings DROP COLUMN kadastr_num;
+    END IF;
+END
+$$;
+
+-- =============================================================================
+-- 5. Recreate v_data_quality
+--    Изменение: kadastr_num → cadastral_number (listings.cadastral_number из 019).
+--    CREATE OR REPLACE VIEW — идемпотентно.
+-- =============================================================================
+
+CREATE OR REPLACE VIEW v_data_quality AS
+WITH active_listings AS (
+    SELECT * FROM listings WHERE is_active = true
+)
+SELECT
+    (SELECT count(*) FROM houses)                                                  AS houses_total,
+    (SELECT count(*) FROM houses h
+       WHERE EXISTS (SELECT 1 FROM house_sources hs WHERE hs.house_id = h.id))      AS houses_with_source,
+    (SELECT count(*) FROM houses h
+       WHERE EXISTS (SELECT 1 FROM house_sources hs
+                      WHERE hs.house_id = h.id AND hs.ext_source = 'avito'))        AS houses_with_avito,
+    (SELECT count(*) FROM houses h
+       WHERE EXISTS (SELECT 1 FROM house_sources hs
+                      WHERE hs.house_id = h.id AND hs.ext_source LIKE 'cian%'))     AS houses_with_cian,
+    (SELECT count(*) FROM houses h
+       WHERE EXISTS (SELECT 1 FROM house_sources hs
+                      WHERE hs.house_id = h.id AND hs.ext_source = 'yandex'))       AS houses_with_yandex,
+    (SELECT count(*) FROM (
+        SELECT house_id FROM house_sources GROUP BY house_id HAVING count(*) >= 2
+    ) sub)                                                                          AS houses_2plus_sources,
+    (SELECT count(*) FROM (
+        SELECT house_id FROM house_sources GROUP BY house_id HAVING count(*) >= 3
+    ) sub)                                                                          AS houses_3plus_sources,
+    (SELECT count(*) FROM active_listings)                                         AS listings_active,
+    (SELECT count(*) FROM (
+        SELECT listing_id FROM listing_sources
+         WHERE listing_id IN (SELECT id FROM active_listings)
+         GROUP BY listing_id HAVING count(*) >= 2
+    ) sub)                                                                          AS listings_dedup_2sources,
+    (SELECT count(*) FROM active_listings WHERE lat IS NOT NULL) * 100.0
+        / NULLIF((SELECT count(*) FROM active_listings), 0)                         AS pct_geocoded,
+    (SELECT count(*) FROM active_listings WHERE cadastral_number IS NOT NULL) * 100.0
+        / NULLIF((SELECT count(*) FROM active_listings), 0)                         AS pct_cadastr,
+    (SELECT count(*) FROM active_listings WHERE description IS NOT NULL) * 100.0
+        / NULLIF((SELECT count(*) FROM active_listings), 0)                         AS pct_description,
+    (SELECT count(*) FROM active_listings l
+       JOIN houses h ON h.id = l.house_id_fk
+       WHERE h.year_built IS NOT NULL) * 100.0
+        / NULLIF((SELECT count(*) FROM active_listings), 0)                         AS pct_year_built,
+    NOW() - (SELECT max(scraped_at) FROM listings WHERE source = 'avito')           AS avito_last_scrape_ago,
+    NOW() - (SELECT max(scraped_at) FROM listings WHERE source = 'cian')            AS cian_last_scrape_ago,
+    NOW() - (SELECT max(scraped_at) FROM listings WHERE source = 'yandex')          AS yandex_last_scrape_ago,
+    (SELECT count(*) FROM v_price_divergence)                                       AS price_disagreements_count,
+    (SELECT count(*) FROM listings WHERE is_outlier = true)                         AS outliers_flagged;
+
+COMMENT ON VIEW v_data_quality IS
+  'KPI snapshot. Refreshed on-demand by /api/v1/admin/data-quality endpoint (Master Plan sec 8.1).';
+
+-- =============================================================================
+-- 6. Recreate listings_search_mv
+--    Изменение: l.kadastr_num → l.cadastral_number (apartment-level из 019_listings_alter_cian).
+--    Все 6 индексов воссоздаются (см. 050_search_optimization.sql).
+-- =============================================================================
+
+CREATE MATERIALIZED VIEW listings_search_mv AS
+SELECT
+    l.id                AS listing_id,
+    l.source,
+    l.source_url,
+    l.address,
+    l.geom,
+    l.lat,
+    l.lon              AS lng,
+    l.rooms,
+    l.area_m2          AS total_area,
+    l.floor,
+    l.total_floors,
+    l.price_rub,
+    l.price_per_m2,
+    l.cadastral_number,
+    l.is_active,
+    l.scraped_at,
+    -- House denorm
+    h.id               AS house_id,
+    h.year_built,
+    h.house_class,
+    h.developer_name,
+    h.rating           AS house_rating,
+    h.reviews_count    AS house_ratings_count,
+    -- Cross-source aggregates
+    (SELECT count(*)               FROM listing_sources ls WHERE ls.listing_id = l.id) AS source_count,
+    (SELECT array_agg(DISTINCT ext_source) FROM listing_sources ls WHERE ls.listing_id = l.id) AS sources,
+    (SELECT bool_or(ext_source = 'avito')         FROM listing_sources ls WHERE ls.listing_id = l.id) AS has_avito,
+    (SELECT bool_or(ext_source = 'cian')          FROM listing_sources ls WHERE ls.listing_id = l.id) AS has_cian,
+    (SELECT bool_or(ext_source = 'yandex_realty') FROM listing_sources ls WHERE ls.listing_id = l.id) AS has_yandex,
+    -- Price percentile within house
+    (SELECT percentile_cont(0.5) WITHIN GROUP (ORDER BY ll.price_per_m2)
+       FROM listings ll
+      WHERE ll.house_id_fk = l.house_id_fk AND ll.is_active = true) AS house_median_ppm2,
+    -- DEFAULT: district / distance_to_metro_m / last_price_change / photos_count
+    -- not present in current schema — placeholders.
+    NULL::text                          AS district,
+    NULL::int                           AS distance_to_metro_m,
+    NULL::timestamptz                   AS last_price_change,
+    NULL::int                           AS photos_count,
+    -- Trigram-ready columns
+    l.address                           AS address_trgm,
+    -- Aggregated tsv (description + address + developer_name)
+    to_tsvector('russian',
+        coalesce(l.description, '') || ' ' ||
+        coalesce(l.address, '')     || ' ' ||
+        coalesce(h.developer_name, '')
+    ) AS tsv
+FROM listings l
+LEFT JOIN houses h ON h.id = l.house_id_fk
+WHERE l.is_active = true
+  AND COALESCE(l.canonical, true) = true;
+
+-- 6 matview indexes (воссоздаются без IF NOT EXISTS — MV была только что создана)
+CREATE UNIQUE INDEX listings_search_mv_id_idx
+  ON listings_search_mv (listing_id);
+
+CREATE INDEX listings_search_mv_geom_idx
+  ON listings_search_mv USING GIST (geom);
+
+CREATE INDEX listings_search_mv_filters_idx
+  ON listings_search_mv (rooms, price_rub, total_area, scraped_at DESC);
+
+CREATE INDEX listings_search_mv_address_trgm_idx
+  ON listings_search_mv USING GIN (address_trgm gin_trgm_ops);
+
+CREATE INDEX listings_search_mv_tsv_idx
+  ON listings_search_mv USING GIN (tsv);
+
+CREATE INDEX listings_search_mv_sources_idx
+  ON listings_search_mv (has_avito, has_cian, has_yandex);
+
+-- Populate materialized view (первый запуск — CONCURRENTLY недоступен на новой MV).
+REFRESH MATERIALIZED VIEW listings_search_mv;
+
+COMMIT;

tradein-mvp/backend/tests/test_estimator_enrich.py

@@ -71,7 +71,7 @@ def test_deal_to_analog_carries_lat_lon() -> None:
         "price_per_m2": 183_333,
         "deal_date": None,
         "days_on_market": None,
-        "kadastr_num": "66:41:0204016:10",
+        "cadastral_number": "66:41:0204016:10",
         "source": "rosreestr",
         "distance_m": 80.0,
         "lat": 56.84,

tradein-mvp/backend/tests/test_search_api.py

@@ -34,16 +34,22 @@ def test_build_query_geo_radius():
 
 def test_build_query_all_filters():
     params = SearchParams(
-        rooms=2, area_m2_min=40, area_m2_max=80, price_rub_max=10_000_000,
+        rooms=2,
-        year_built_min=2000, has_kadastr=True, multi_source_only=True,
+        area_m2_min=40,
-        require_avito=True, address_query="Малышева",
+        area_m2_max=80,
+        price_rub_max=10_000_000,
+        year_built_min=2000,
+        has_kadastr=True,
+        multi_source_only=True,
+        require_avito=True,
+        address_query="Малышева",
     )
     sql, args = build_search_query(params)
     assert "rooms = CAST(:rooms AS integer)" in sql
     assert "total_area >= CAST(:area_min AS double precision)" in sql
     assert "price_rub <= CAST(:price_max AS bigint)" in sql
     assert "year_built >= CAST(:yb_min AS integer)" in sql
-    assert "kadastr_num IS NOT NULL" in sql
+    assert "cadastral_number IS NOT NULL" in sql
     assert "source_count >= 2" in sql
     assert "has_avito = true" in sql
     assert "address ILIKE" in sql
@@ -99,8 +105,12 @@ def _reset_cache_singleton(monkeypatch):
 
 def test_search_cache_hit(_reset_cache_singleton):
     cached_payload = {
-        "items": [], "total": 0, "page": 1, "page_size": 50,
+        "items": [],
-        "elapsed_ms": 0.0, "cache_hit": True,
+        "total": 0,
+        "page": 1,
+        "page_size": 50,
+        "elapsed_ms": 0.0,
+        "cache_hit": True,
     }
     _reset_cache_singleton.get = AsyncMock(return_value=cached_payload)
     client = TestClient(app)