Merge pull request 'fix(tradein): avito houses dedup через match_or_create_house + merge-миграция дублей (108)' (#1546) from fix/avito-houses-dedup-merge into main

Reviewed-on: #1546

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

@@ -33,6 +33,7 @@ from typing import TYPE_CHECKING, Any
 from sqlalchemy import text
 from sqlalchemy.orm import Session
 
+from app.services.matching.houses import match_or_create_house
 from app.services.scrapers.avito import _is_firewall_page
 from app.services.scrapers.avito_exceptions import AvitoBlockedError, AvitoRateLimitedError
 
@@ -755,9 +756,10 @@ async def fetch_house_catalog(
 
 
 def upsert_house(db: Session, h: HouseInfo) -> int:
-    """INSERT INTO houses ON CONFLICT (source, ext_house_id) DO UPDATE.
+    """Match-or-create канонический дом через match_or_create_house, затем обновить enrichment.
 
     Source='avito'. Возвращает houses.id.
+    Делегирует в _persist_house (без url — url пуст как раньше для этого пути).
 
     Маппинг полей HouseInfo → DDL колонки (009 + 010):
       ext_id          → ext_house_id (text)
@@ -768,87 +770,87 @@ def upsert_house(db: Session, h: HouseInfo) -> int:
       rating_score    → rating_score (010, numeric(4,2))
       rating (DDL)    → округлённый numeric(2,1) из rating_score
     """
-    row = db.execute(
+    return _persist_house(db, h, "")
+
+
+def _upsert_house_with_url(db: Session, h: HouseInfo, house_url: str) -> int:
+    """Внутренний вариант upsert_house с правильным URL.
+
+    Делегирует в _persist_house — используй его вместо прямого вызова.
+    """
+    return _persist_house(db, h, house_url)
+
+
+def _persist_house(db: Session, h: HouseInfo, house_url: str) -> int:
+    """Канонический путь сохранения дома-avito через match_or_create_house.
+
+    Шаги:
+      1. match_or_create_house — находит или создаёт канонический houses-ряд
+         (3-tier: source_exact → fingerprint → geo → new), защищён advisory lock.
+      2. UPDATE houses SET ... — записывает все avito-enrichment поля в найденный/новый дом.
+         - Avito-авторитетные поля (rating*, developer*, технические хар-ки, url,
+           avito_id_hash) перезаписываются напрямую.
+         - address/lat/lon/year_built — через COALESCE чтобы не затирать данные
+           канонического дома от cian/кадастра.
+      3. Не вызывает db.commit() — коммитит вызывающий save_house_catalog_enrichment
+         (advisory lock xact-scoped, должен держаться до commit).
+
+    Returns:
+        houses.id (int) канонического дома.
+    """
+    house_id, _conf, _method = match_or_create_house(
+        db,
+        ext_source="avito",
+        ext_id=str(h.ext_id),
+        # short preferred: aliases have no city/region prefix; full is last resort
+        address=h.short_address or h.full_address,
+        lat=h.lat,
+        lon=h.lon,
+        year_built=h.year_built,
+        source_url=house_url or None,
+    )
+
+    db.execute(
         text("""
-            INSERT INTO houses (
+            UPDATE houses SET
-                source, ext_house_id, url,
+                url                          = CAST(:url AS text),
-                address, short_address, full_address,
+                short_address                = CAST(:short_address AS text),
-                lat, lon,
+                full_address                 = CAST(:full_address AS text),
-                avito_id_hash,
+                avito_id_hash                = CAST(:avito_id_hash AS text),
-                year_built, house_type, total_floors,
+                house_type                   = CAST(:house_type AS text),
-                passenger_elevators, cargo_elevators,
+                total_floors                 = CAST(:total_floors AS int),
-                has_concierge, closed_yard,
+                passenger_elevators          = CAST(:passenger_elevators AS int),
-                material_floors, hot_water,
+                cargo_elevators              = CAST(:cargo_elevators AS int),
-                parking_type, has_playground,
+                has_concierge                = CAST(:has_concierge AS boolean),
-                developer_name, developer_key,
+                closed_yard                  = CAST(:closed_yard AS boolean),
-                infrastructure_summary, infrastructure_walk_distance,
+                material_floors              = CAST(:material_floors AS text),
-                rating, rating_score, rating_string, reviews_count,
+                hot_water                    = CAST(:hot_water AS text),
-                rating_distribution,
+                parking_type                 = CAST(:parking_type AS text),
-                map_pins,
+                has_playground               = CAST(:has_playground AS boolean),
-                raw_characteristics,
+                developer_name               = CAST(:developer_name AS text),
-                last_scraped_at
+                developer_key                = CAST(:developer_key AS text),
-            ) VALUES (
+                infrastructure_summary       = CAST(:infrastructure_summary AS text),
-                'avito', CAST(:ext_house_id AS text), CAST(:url AS text),
+                infrastructure_walk_distance = CAST(:infrastructure_walk_distance AS text),
-                CAST(:address AS text), CAST(:short_address AS text), CAST(:full_address AS text),
+                rating                       = CAST(:rating AS numeric),
-                CAST(:lat AS double precision), CAST(:lon AS double precision),
+                rating_score                 = CAST(:rating_score AS numeric),
-                CAST(:avito_id_hash AS text),
+                rating_string                = CAST(:rating_string AS text),
-                CAST(:year_built AS int), CAST(:house_type AS text), CAST(:total_floors AS int),
+                reviews_count                = CAST(:reviews_count AS int),
-                CAST(:passenger_elevators AS int), CAST(:cargo_elevators AS int),
+                rating_distribution          = CAST(:rating_distribution AS jsonb),
-                CAST(:has_concierge AS boolean), CAST(:closed_yard AS boolean),
+                map_pins                     = CAST(:map_pins AS jsonb),
-                CAST(:material_floors AS text), CAST(:hot_water AS text),
+                raw_characteristics          = CAST(:raw_characteristics AS jsonb),
-                CAST(:parking_type AS text), CAST(:has_playground AS boolean),
+                address   = COALESCE(houses.address, CAST(:address AS text)),
-                CAST(:developer_name AS text), CAST(:developer_key AS text),
+                lat       = COALESCE(houses.lat,     CAST(:lat AS double precision)),
-                CAST(:infrastructure_summary AS text), CAST(:infrastructure_walk_distance AS text),
+                lon       = COALESCE(houses.lon,     CAST(:lon AS double precision)),
-                CAST(:rating AS numeric), CAST(:rating_score AS numeric),
+                year_built = COALESCE(houses.year_built, CAST(:year_built AS int)),
-                CAST(:rating_string AS text), CAST(:reviews_count AS int),
+                last_scraped_at              = NOW()
-                CAST(:rating_distribution AS jsonb),
+            WHERE id = CAST(:house_id AS bigint)
-                CAST(:map_pins AS jsonb),
-                CAST(:raw_characteristics AS jsonb),
-                NOW()
-            )
-            ON CONFLICT (source, ext_house_id) DO UPDATE SET
-                url                         = EXCLUDED.url,
-                address                     = EXCLUDED.address,
-                short_address               = EXCLUDED.short_address,
-                full_address                = EXCLUDED.full_address,
-                lat                         = EXCLUDED.lat,
-                lon                         = EXCLUDED.lon,
-                avito_id_hash               = EXCLUDED.avito_id_hash,
-                year_built                  = EXCLUDED.year_built,
-                house_type                  = EXCLUDED.house_type,
-                total_floors                = EXCLUDED.total_floors,
-                passenger_elevators         = EXCLUDED.passenger_elevators,
-                cargo_elevators             = EXCLUDED.cargo_elevators,
-                has_concierge               = EXCLUDED.has_concierge,
-                closed_yard                 = EXCLUDED.closed_yard,
-                material_floors             = EXCLUDED.material_floors,
-                hot_water                   = EXCLUDED.hot_water,
-                parking_type                = EXCLUDED.parking_type,
-                has_playground              = EXCLUDED.has_playground,
-                developer_name              = EXCLUDED.developer_name,
-                developer_key               = EXCLUDED.developer_key,
-                infrastructure_summary      = EXCLUDED.infrastructure_summary,
-                infrastructure_walk_distance = EXCLUDED.infrastructure_walk_distance,
-                rating                      = EXCLUDED.rating,
-                rating_score                = EXCLUDED.rating_score,
-                rating_string               = EXCLUDED.rating_string,
-                reviews_count               = EXCLUDED.reviews_count,
-                rating_distribution         = EXCLUDED.rating_distribution,
-                map_pins                    = EXCLUDED.map_pins,
-                raw_characteristics         = EXCLUDED.raw_characteristics,
-                last_scraped_at             = NOW()
-            RETURNING id
         """),
         {
-            "ext_house_id": str(h.ext_id),
+            "house_id": house_id,
-            "url": "",  # caller должен передавать house_url через enrichment
+            "url": house_url or "",
-            "address": h.short_address,
             "short_address": h.short_address,
             "full_address": h.full_address,
-            "lat": h.lat,
-            "lon": h.lon,
             "avito_id_hash": h.ext_id_hash,
-            "year_built": h.year_built,
             "house_type": h.house_type,
             "total_floors": h.total_floors,
             "passenger_elevators": h.passenger_elevators,
@@ -871,124 +873,21 @@ def upsert_house(db: Session, h: HouseInfo) -> int:
             "rating_distribution": json.dumps(h.rating_distribution, ensure_ascii=False),
             "map_pins": json.dumps(h.map_pins, ensure_ascii=False),
             "raw_characteristics": json.dumps(h.raw_characteristics, ensure_ascii=False),
-        },
-    )
-    house_id: int = row.scalar_one()
-    logger.info("upsert_house: ext_id=%s → houses.id=%s", h.ext_id, house_id)
-    return house_id
-
-
-def _upsert_house_with_url(db: Session, h: HouseInfo, house_url: str) -> int:
-    """Внутренний вариант upsert_house с правильным URL."""
-    row = db.execute(
-        text("""
-            INSERT INTO houses (
-                source, ext_house_id, url,
-                address, short_address, full_address,
-                lat, lon,
-                avito_id_hash,
-                year_built, house_type, total_floors,
-                passenger_elevators, cargo_elevators,
-                has_concierge, closed_yard,
-                material_floors, hot_water,
-                parking_type, has_playground,
-                developer_name, developer_key,
-                infrastructure_summary, infrastructure_walk_distance,
-                rating, rating_score, rating_string, reviews_count,
-                rating_distribution,
-                map_pins,
-                raw_characteristics,
-                last_scraped_at
-            ) VALUES (
-                'avito', CAST(:ext_house_id AS text), CAST(:url AS text),
-                CAST(:address AS text), CAST(:short_address AS text), CAST(:full_address AS text),
-                CAST(:lat AS double precision), CAST(:lon AS double precision),
-                CAST(:avito_id_hash AS text),
-                CAST(:year_built AS int), CAST(:house_type AS text), CAST(:total_floors AS int),
-                CAST(:passenger_elevators AS int), CAST(:cargo_elevators AS int),
-                CAST(:has_concierge AS boolean), CAST(:closed_yard AS boolean),
-                CAST(:material_floors AS text), CAST(:hot_water AS text),
-                CAST(:parking_type AS text), CAST(:has_playground AS boolean),
-                CAST(:developer_name AS text), CAST(:developer_key AS text),
-                CAST(:infrastructure_summary AS text), CAST(:infrastructure_walk_distance AS text),
-                CAST(:rating AS numeric), CAST(:rating_score AS numeric),
-                CAST(:rating_string AS text), CAST(:reviews_count AS int),
-                CAST(:rating_distribution AS jsonb),
-                CAST(:map_pins AS jsonb),
-                CAST(:raw_characteristics AS jsonb),
-                NOW()
-            )
-            ON CONFLICT (source, ext_house_id) DO UPDATE SET
-                url                         = EXCLUDED.url,
-                address                     = EXCLUDED.address,
-                short_address               = EXCLUDED.short_address,
-                full_address                = EXCLUDED.full_address,
-                lat                         = EXCLUDED.lat,
-                lon                         = EXCLUDED.lon,
-                avito_id_hash               = EXCLUDED.avito_id_hash,
-                year_built                  = EXCLUDED.year_built,
-                house_type                  = EXCLUDED.house_type,
-                total_floors                = EXCLUDED.total_floors,
-                passenger_elevators         = EXCLUDED.passenger_elevators,
-                cargo_elevators             = EXCLUDED.cargo_elevators,
-                has_concierge               = EXCLUDED.has_concierge,
-                closed_yard                 = EXCLUDED.closed_yard,
-                material_floors             = EXCLUDED.material_floors,
-                hot_water                   = EXCLUDED.hot_water,
-                parking_type                = EXCLUDED.parking_type,
-                has_playground              = EXCLUDED.has_playground,
-                developer_name              = EXCLUDED.developer_name,
-                developer_key               = EXCLUDED.developer_key,
-                infrastructure_summary      = EXCLUDED.infrastructure_summary,
-                infrastructure_walk_distance = EXCLUDED.infrastructure_walk_distance,
-                rating                      = EXCLUDED.rating,
-                rating_score                = EXCLUDED.rating_score,
-                rating_string               = EXCLUDED.rating_string,
-                reviews_count               = EXCLUDED.reviews_count,
-                rating_distribution         = EXCLUDED.rating_distribution,
-                map_pins                    = EXCLUDED.map_pins,
-                raw_characteristics         = EXCLUDED.raw_characteristics,
-                last_scraped_at             = NOW()
-            RETURNING id
-        """),
-        {
-            "ext_house_id": str(h.ext_id),
-            "url": house_url,
             "address": h.short_address,
-            "short_address": h.short_address,
-            "full_address": h.full_address,
             "lat": h.lat,
             "lon": h.lon,
-            "avito_id_hash": h.ext_id_hash,
             "year_built": h.year_built,
-            "house_type": h.house_type,
-            "total_floors": h.total_floors,
-            "passenger_elevators": h.passenger_elevators,
-            "cargo_elevators": h.cargo_elevators,
-            "has_concierge": h.has_concierge,
-            "closed_yard": h.closed_yard,
-            "material_floors": h.material_floors,
-            "hot_water": h.hot_water,
-            "parking_type": h.parking_type,
-            "has_playground": h.has_playground,
-            "developer_name": h.developer_name,
-            "developer_key": h.developer_key,
-            "infrastructure_summary": h.infrastructure_summary,
-            "infrastructure_walk_distance": h.infrastructure_walk_distance,
-            "rating": round(h.rating_score, 1) if h.rating_score is not None else None,
-            "rating_score": h.rating_score,
-            "rating_string": h.rating_string,
-            "reviews_count": h.reviews_count,
-            "rating_distribution": json.dumps(h.rating_distribution, ensure_ascii=False),
-            "map_pins": json.dumps(h.map_pins, ensure_ascii=False),
-            "raw_characteristics": json.dumps(h.raw_characteristics, ensure_ascii=False),
         },
     )
-    house_id: int = row.scalar_one()
+
     logger.info(
-        "_upsert_house_with_url: ext_id=%s url=%s → houses.id=%s", h.ext_id, house_url, house_id
+        "_persist_house: ext_id=%s url=%s → houses.id=%s (method=%s)",
+        h.ext_id,
+        house_url,
+        house_id,
+        _method,
     )
-    return house_id
+    return int(house_id)
 
 
 def save_house_reviews(db: Session, house_id: int, reviews: list[HouseReview]) -> int:

tradein-mvp/backend/data/sql/108_merge_duplicate_houses.sql

@@ -0,0 +1,347 @@
+-- 108_merge_duplicate_houses.sql
+-- Цель: объединить дублирующиеся строки в таблице houses, возникшие из-за того, что
+--   avito_houses.py до fix #108 использовал прямой INSERT ON CONFLICT(source, ext_house_id)
+--   вместо canonического match_or_create_house. В результате один физический дом
+--   мог иметь несколько строк — по одной на каждый source (avito-каталог vs listing-scraper).
+--
+-- Алгоритм:
+--   1. Построить cluster_key по (cadastral_number / address / geo) — консервативная кластеризация.
+--   2. Выбрать canonical дом в кластере: максимум не-NULL колонок, geom присутствует, min(id).
+--   3. Перепривязать все зависимые таблицы с dup → canonical (обработка UNIQUE-коллизий).
+--   4. Удалить дубли.
+--   5. Вставить house_sources + address_aliases для backfill Tier1/2 на следующем скрейпе.
+--   6. RAISE NOTICE с итоговыми счётчиками.
+--
+-- Зависимые таблицы (FK на houses.id):
+--   listings               (house_id_fk)  — NO ACTION  — обязателен репойнт
+--   house_placement_history (house_id)    — NO ACTION  — обязателен репойнт
+--   house_reviews          (house_id)     — CASCADE    — репойнт перед DELETE чтобы обновить house_id
+--   houses_price_dynamics  (house_id)     — CASCADE    — UNIQUE(house_id,month_date,source) → dedup
+--   house_reliability_checks(house_id)    — CASCADE    — нет ограничивающего UNIQUE
+--   house_sources          (house_id)     — CASCADE    — UNIQUE(ext_source,ext_id) → dedup
+--   house_address_aliases  (house_id)     — CASCADE    — UNIQUE(normalized_address) → dedup
+--   house_imv_evaluations  (house_id)     — CASCADE    — UNIQUE(house_id) → dedup
+--   house_suggestions      (house_id)     — CASCADE    — UNIQUE(house_id, ext_item_id) → dedup
+--   external_valuations    (house_id)     — CASCADE    — нет ограничивающего UNIQUE
+--   address_mismatch_audit (house_id)     — CASCADE    — нет ограничивающего UNIQUE
+--
+-- Идемпотентность: все UPDATE/DELETE ищут по mapping временной таблицы. При повторном
+--   запуске mapping будет пустым (дублей нет) → все UPDATE/DELETE затронут 0 строк → no-op.
+--
+-- Стиль: следует 063_backfill_houses_and_link_listings.sql.
+--
+-- Деплой: после 107_scrape_schedules_seed_cian_city_sweep.sql
+
+BEGIN;
+
+-- -------------------------------------------------------------------------
+-- Step 0: Счётчик до
+-- -------------------------------------------------------------------------
+DO $$
+DECLARE
+    v_before bigint;
+BEGIN
+    SELECT count(*) INTO v_before FROM houses;
+    RAISE NOTICE 'merge 108 start: houses_before=%', v_before;
+END $$;
+
+-- -------------------------------------------------------------------------
+-- Step 1: Построить mapping дублей → canonical
+--
+-- cluster_key (приоритет):
+--   1. cadastral_number (наиболее точный идентификатор)
+--   2. 'addr:' || lower(trim(address))
+--   3. 'geo:'  || round(lat,4) || ',' || round(lon,4)
+--   строки без всех трёх полей — cluster_key IS NULL → пропускаем (не трогаем)
+--
+-- Канонический дом: первый по (geom NOT NULL DESC, count NOT NULL fields DESC, id ASC).
+-- Считаем «значимые» NOT NULL поля чтобы выбрать самый полный ряд.
+-- -------------------------------------------------------------------------
+CREATE TEMP TABLE IF NOT EXISTS _108_dup_mapping AS
+WITH clustered AS (
+    SELECT
+        id,
+        COALESCE(
+            NULLIF(trim(cadastral_number), ''),
+            CASE WHEN NULLIF(lower(trim(address)), '') IS NOT NULL
+                 THEN 'addr:' || lower(trim(address)) END,
+            CASE WHEN lat IS NOT NULL AND lon IS NOT NULL
+                 THEN 'geo:' || round(lat::numeric, 4)::text || ',' || round(lon::numeric, 4)::text
+            END
+        ) AS cluster_key
+    FROM houses
+),
+clusters_with_count AS (
+    SELECT cluster_key, count(*) AS n
+    FROM clustered
+    WHERE cluster_key IS NOT NULL
+    GROUP BY cluster_key
+    HAVING count(*) > 1
+),
+-- Только дома, входящие в кластер-дубль
+dup_houses AS (
+    SELECT h.id, c.cluster_key
+    FROM houses h
+    JOIN clustered cl ON cl.id = h.id
+    JOIN clusters_with_count cw ON cw.cluster_key = cl.cluster_key
+),
+-- Выбрать canonical (rn=1) и дубли (rn>1) внутри каждого кластера
+ranked AS (
+    SELECT
+        dh.id,
+        dh.cluster_key,
+        ROW_NUMBER() OVER (
+            PARTITION BY dh.cluster_key
+            ORDER BY
+                (h.geom IS NOT NULL) DESC,
+                -- Считаем не-NULL значимые колонки для оценки полноты
+                (
+                    (h.address IS NOT NULL)::int +
+                    (h.lat IS NOT NULL)::int +
+                    (h.lon IS NOT NULL)::int +
+                    (h.year_built IS NOT NULL)::int +
+                    (h.house_type IS NOT NULL)::int +
+                    (h.total_floors IS NOT NULL)::int +
+                    (h.cadastral_number IS NOT NULL)::int +
+                    (h.developer_name IS NOT NULL)::int +
+                    (h.rating_score IS NOT NULL)::int +
+                    (h.avito_id_hash IS NOT NULL)::int
+                ) DESC,
+                dh.id ASC
+        ) AS rn,
+        first_value(dh.id) OVER (
+            PARTITION BY dh.cluster_key
+            ORDER BY
+                (h.geom IS NOT NULL) DESC,
+                (
+                    (h.address IS NOT NULL)::int +
+                    (h.lat IS NOT NULL)::int +
+                    (h.lon IS NOT NULL)::int +
+                    (h.year_built IS NOT NULL)::int +
+                    (h.house_type IS NOT NULL)::int +
+                    (h.total_floors IS NOT NULL)::int +
+                    (h.cadastral_number IS NOT NULL)::int +
+                    (h.developer_name IS NOT NULL)::int +
+                    (h.rating_score IS NOT NULL)::int +
+                    (h.avito_id_hash IS NOT NULL)::int
+                ) DESC,
+                dh.id ASC
+        ) AS canonical_id
+    FROM dup_houses dh
+    JOIN houses h ON h.id = dh.id
+)
+SELECT id AS dup_id, canonical_id
+FROM ranked
+WHERE rn > 1
+  AND id <> canonical_id;
+
+-- Лог размера mapping
+DO $$
+DECLARE
+    v_dups bigint;
+    v_clusters bigint;
+BEGIN
+    SELECT count(*), count(DISTINCT canonical_id) INTO v_dups, v_clusters
+    FROM _108_dup_mapping;
+    RAISE NOTICE 'merge 108: mapping built — dup_rows=%, clusters_to_merge=%', v_dups, v_clusters;
+END $$;
+
+-- -------------------------------------------------------------------------
+-- Step 2: Репойнт зависимых таблиц (NO ACTION FK — обязательно до DELETE)
+-- -------------------------------------------------------------------------
+
+-- 2a. listings.house_id_fk  (NO ACTION — ключевой)
+UPDATE listings l
+SET house_id_fk = m.canonical_id
+FROM _108_dup_mapping m
+WHERE l.house_id_fk = m.dup_id;
+
+-- 2b. house_placement_history.house_id  (NO ACTION)
+UPDATE house_placement_history hph
+SET house_id = m.canonical_id
+FROM _108_dup_mapping m
+WHERE hph.house_id = m.dup_id;
+
+-- -------------------------------------------------------------------------
+-- Step 3: Репойнт таблиц с CASCADE FK + UNIQUE-коллизиями
+--
+-- Паттерн: сначала удалить дублирующую строку дубля если canonical уже имеет
+-- такой же ключ, затем репойнтить оставшееся.
+-- -------------------------------------------------------------------------
+
+-- 3a. house_sources  UNIQUE(ext_source, ext_id)
+--   Удалить строки дубля, у которых (ext_source,ext_id) уже есть у canonical.
+DELETE FROM house_sources hs
+USING _108_dup_mapping m
+WHERE hs.house_id = m.dup_id
+  AND EXISTS (
+      SELECT 1 FROM house_sources hs2
+      WHERE hs2.house_id = m.canonical_id
+        AND hs2.ext_source = hs.ext_source
+        AND hs2.ext_id = hs.ext_id
+  );
+--   Репойнтить оставшиеся строки дубля
+UPDATE house_sources hs
+SET house_id = m.canonical_id
+FROM _108_dup_mapping m
+WHERE hs.house_id = m.dup_id;
+
+-- 3b. house_address_aliases  UNIQUE(normalized_address)
+--   Удалить строки дубля, у которых normalized_address уже есть у canonical.
+DELETE FROM house_address_aliases haa
+USING _108_dup_mapping m
+WHERE haa.house_id = m.dup_id
+  AND EXISTS (
+      SELECT 1 FROM house_address_aliases haa2
+      WHERE haa2.house_id = m.canonical_id
+        AND haa2.normalized_address = haa.normalized_address
+  );
+--   Репойнтить оставшиеся
+UPDATE house_address_aliases haa
+SET house_id = m.canonical_id
+FROM _108_dup_mapping m
+WHERE haa.house_id = m.dup_id;
+
+-- 3c. houses_price_dynamics  UNIQUE(house_id, month_date, source)
+--   Удалить строки дубля, у которых (month_date,source) уже есть у canonical.
+DELETE FROM houses_price_dynamics hpd
+USING _108_dup_mapping m
+WHERE hpd.house_id = m.dup_id
+  AND EXISTS (
+      SELECT 1 FROM houses_price_dynamics hpd2
+      WHERE hpd2.house_id = m.canonical_id
+        AND hpd2.month_date = hpd.month_date
+        AND hpd2.source = hpd.source
+  );
+--   Репойнтить оставшиеся
+UPDATE houses_price_dynamics hpd
+SET house_id = m.canonical_id
+FROM _108_dup_mapping m
+WHERE hpd.house_id = m.dup_id;
+
+-- 3d. house_imv_evaluations  UNIQUE(house_id) — у canonical уже может быть запись
+--   Удалить строки дубля если canonical уже имеет evaluation.
+DELETE FROM house_imv_evaluations hie
+USING _108_dup_mapping m
+WHERE hie.house_id = m.dup_id
+  AND EXISTS (
+      SELECT 1 FROM house_imv_evaluations hie2
+      WHERE hie2.house_id = m.canonical_id
+  );
+--   Репойнтить оставшиеся (canonical не имеет evaluation — забираем от дубля)
+UPDATE house_imv_evaluations hie
+SET house_id = m.canonical_id
+FROM _108_dup_mapping m
+WHERE hie.house_id = m.dup_id;
+
+-- 3e. house_suggestions  UNIQUE(house_id, ext_item_id)
+--   Удалить строки дубля, у которых (ext_item_id) уже есть у canonical.
+DELETE FROM house_suggestions hs
+USING _108_dup_mapping m
+WHERE hs.house_id = m.dup_id
+  AND EXISTS (
+      SELECT 1 FROM house_suggestions hs2
+      WHERE hs2.house_id = m.canonical_id
+        AND hs2.ext_item_id = hs.ext_item_id
+  );
+--   Репойнтить оставшиеся
+UPDATE house_suggestions hs
+SET house_id = m.canonical_id
+FROM _108_dup_mapping m
+WHERE hs.house_id = m.dup_id;
+
+-- 3f. house_reviews (CASCADE, нет UNIQUE конфликта — просто репойнт)
+UPDATE house_reviews hr
+SET house_id = m.canonical_id
+FROM _108_dup_mapping m
+WHERE hr.house_id = m.dup_id;
+
+-- 3g. house_reliability_checks (CASCADE, нет блокирующего UNIQUE)
+UPDATE house_reliability_checks hrc
+SET house_id = m.canonical_id
+FROM _108_dup_mapping m
+WHERE hrc.house_id = m.dup_id;
+
+-- 3h. external_valuations (CASCADE, нет блокирующего UNIQUE)
+UPDATE external_valuations ev
+SET house_id = m.canonical_id
+FROM _108_dup_mapping m
+WHERE ev.house_id = m.dup_id;
+
+-- 3i. address_mismatch_audit (CASCADE, нет блокирующего UNIQUE)
+UPDATE address_mismatch_audit ama
+SET house_id = m.canonical_id
+FROM _108_dup_mapping m
+WHERE ama.house_id = m.dup_id;
+
+-- -------------------------------------------------------------------------
+-- Step 4: Удалить дубли (все FK уже перепривязаны или CASCADE)
+-- -------------------------------------------------------------------------
+DELETE FROM houses h
+USING _108_dup_mapping m
+WHERE h.id = m.dup_id;
+
+-- -------------------------------------------------------------------------
+-- Step 5: Backfill house_sources для canonical-домов
+--   Гарантируем что у каждого canonical'а есть source-запись для avito,
+--   чтобы Tier1 (source_exact) в match_or_create_house сработал на следующем скрейпе.
+--
+--   Используем ext_house_id из houses.ext_house_id — только для домов с source='avito'.
+--   Для домов source='derived' ориентируемся на address — backfill address_aliases ниже.
+-- -------------------------------------------------------------------------
+INSERT INTO house_sources (house_id, ext_source, ext_id, confidence, matched_method, matched_at, last_seen_at)
+SELECT
+    h.id,
+    h.source,
+    h.ext_house_id,
+    1.0,
+    'backfill_merge',
+    NOW(),
+    NOW()
+FROM houses h
+WHERE h.source = 'avito'
+  AND h.ext_house_id IS NOT NULL
+  AND h.id IN (SELECT canonical_id FROM _108_dup_mapping)
+ON CONFLICT (ext_source, ext_id) DO NOTHING;
+
+-- Backfill address_aliases — нормализованный адрес как low(trim(address)).
+-- Питоновский normalize_address() недоступен в SQL; используем lower(trim(address))
+-- как приближение. Если это недостаточно точно — forward-fix само залечит через
+-- geo-tier при следующем скрейпе (Tier3 строит alias автоматически).
+INSERT INTO house_address_aliases (house_id, normalized_address, fingerprint, source)
+SELECT
+    h.id,
+    lower(trim(h.address)),
+    NULL,  -- fingerprint не вычисляем в SQL — обновится при следующем скрейпе
+    'backfill_merge'
+FROM houses h
+WHERE h.address IS NOT NULL
+  AND length(trim(h.address)) >= 5
+  AND h.id IN (SELECT canonical_id FROM _108_dup_mapping)
+ON CONFLICT (normalized_address) DO NOTHING;
+
+-- -------------------------------------------------------------------------
+-- Step 6: Итоговые счётчики
+-- -------------------------------------------------------------------------
+DO $$
+DECLARE
+    v_before          bigint;
+    v_clusters_merged bigint;
+    v_rows_deleted    bigint;
+    v_after           bigint;
+BEGIN
+    -- before был зафиксирован в Step 0 RAISE NOTICE; здесь пересчитываем через mapping
+    SELECT count(DISTINCT canonical_id) INTO v_clusters_merged FROM _108_dup_mapping;
+    SELECT count(*)                     INTO v_rows_deleted    FROM _108_dup_mapping;
+    SELECT count(*)                     INTO v_after           FROM houses;
+    -- before = after + deleted
+    v_before := v_after + v_rows_deleted;
+
+    RAISE NOTICE
+        'merge 108 done: houses before=%, clusters_merged=%, rows_deleted=%, after=%',
+        v_before, v_clusters_merged, v_rows_deleted, v_after;
+END $$;
+
+DROP TABLE IF EXISTS _108_dup_mapping;
+
+COMMIT;

tradein-mvp/backend/tests/test_avito_houses_dedup.py

@@ -0,0 +1,271 @@
+"""Tests for avito_houses dedup fix — _persist_house делегирует в match_or_create_house.
+
+Проверяет:
+  1. Два HouseInfo с разными ext_id но одинаковым адресом и координатами
+     → _persist_house возвращает ОДИН house_id (не плодит дубль).
+  2. Enrichment-поля (rating_score, developer_name) реально записываются в UPDATE.
+  3. _persist_house не вызывает db.commit().
+  4. upsert_house и _upsert_house_with_url делегируют в _persist_house.
+
+Использует тот же паттерн mock-DB что в test_matching.py и test_same_building_match.py.
+"""
+
+import json
+import os
+
+os.environ.setdefault("DATABASE_URL", "postgresql://test:test@localhost/test_db")
+
+from unittest.mock import MagicMock, patch
+
+from app.services.scrapers.avito_houses import (
+    HouseInfo,
+    _persist_house,
+    _upsert_house_with_url,
+    upsert_house,
+)
+
+# ---------------------------------------------------------------------------
+# Helpers
+# ---------------------------------------------------------------------------
+
+
+def _make_house(ext_id: int = 111, address: str = "ул. Ленина, 5") -> HouseInfo:
+    return HouseInfo(
+        ext_id=ext_id,
+        ext_id_hash="hash_" + str(ext_id),
+        short_address=address,
+        full_address="Свердловская обл., Екатеринбург, " + address,
+        lat=56.8378,
+        lon=60.5946,
+        year_built=2010,
+        house_type="monolith",
+        total_floors=16,
+        passenger_elevators=2,
+        cargo_elevators=1,
+        has_concierge=True,
+        closed_yard=False,
+        material_floors="concrete",
+        hot_water="central",
+        parking_type="underground",
+        has_playground=True,
+        developer_name="ПИК",
+        developer_key="pik",
+        infrastructure_summary="Школа, детсад",
+        infrastructure_walk_distance="5 мин",
+        rating_score=4.5,
+        rating_string="4.5 из 5",
+        reviews_count=120,
+        rating_distribution=[{"score": 5, "count": 80}],
+        map_pins=[{"type": "metro", "name": "Уральская"}],
+        raw_characteristics=[{"title": "Год постройки", "value": "2010"}],
+    )
+
+
+def _make_db_for_persist(canonical_id: int = 42) -> MagicMock:
+    """Mock DB для _persist_house: match_or_create_house ходит через db.execute,
+    поэтому патчим match_or_create_house целиком и проверяем вызов UPDATE.
+    """
+    db = MagicMock()
+    # db.execute для UPDATE houses SET ... — просто возвращаем None (rowcount не важен)
+    db.execute.return_value = MagicMock()
+    return db
+
+
+# ---------------------------------------------------------------------------
+# Test 1: два HouseInfo с разными ext_id, одинаковый адрес → один house_id
+# ---------------------------------------------------------------------------
+
+
+def test_persist_house_same_address_returns_same_id():
+    """Два HouseInfo с разными ext_id но одинаковым адресом → один canonical house_id.
+
+    Моделируем случай: первый вызов (ext_id=111) → match_or_create_house вернул 42 (new).
+    Второй вызов (ext_id=222, тот же адрес) → match_or_create_house вернул 42 (fingerprint).
+    Оба _persist_house должны вернуть 42.
+    """
+    db = MagicMock()
+    db.execute.return_value = MagicMock()
+
+    h1 = _make_house(ext_id=111)
+    h2 = _make_house(ext_id=222)  # другой ext_id, тот же адрес/координаты
+
+    with patch(
+        "app.services.scrapers.avito_houses.match_or_create_house",
+        side_effect=[(42, 1.0, "new"), (42, 0.9, "fingerprint")],
+    ) as mock_match:
+        id1 = _persist_house(db, h1, "https://avito.ru/house/111")
+        id2 = _persist_house(db, h2, "https://avito.ru/house/222")
+
+    assert id1 == 42, f"Первый вызов должен вернуть 42, получили {id1}"
+    assert id2 == 42, f"Второй вызов должен вернуть тот же 42, получили {id2}"
+    assert id1 == id2, "Оба вызова должны вернуть один house_id (нет дубля)"
+    assert mock_match.call_count == 2
+
+
+# ---------------------------------------------------------------------------
+# Test 2: enrichment-поля записываются в UPDATE
+# ---------------------------------------------------------------------------
+
+
+def test_persist_house_enrichment_fields_written():
+    """Enrichment-поля (rating_score, developer_name) передаются в UPDATE-запрос."""
+    db = MagicMock()
+    db.execute.return_value = MagicMock()
+
+    h = _make_house(ext_id=111)
+
+    with patch(
+        "app.services.scrapers.avito_houses.match_or_create_house",
+        return_value=(42, 1.0, "source_exact"),
+    ):
+        _persist_house(db, h, "https://avito.ru/house/111")
+
+    # db.execute должен быть вызван 1 раз (UPDATE houses SET ...)
+    assert (
+        db.execute.call_count == 1
+    ), f"Ожидался 1 вызов db.execute (UPDATE), получили {db.execute.call_count}"
+
+    # Проверяем параметры bind
+    call_args = db.execute.call_args
+    sql_obj = call_args[0][0]
+    bind = call_args[0][1]
+
+    assert "UPDATE houses" in str(sql_obj), "SQL должен содержать UPDATE houses"
+    assert bind["rating_score"] == 4.5, f"rating_score={bind.get('rating_score')!r}"
+    assert bind["developer_name"] == "ПИК", f"developer_name={bind.get('developer_name')!r}"
+    assert bind["house_id"] == 42, f"house_id={bind.get('house_id')!r}"
+
+    # avito_id_hash должен быть передан
+    assert bind["avito_id_hash"] == "hash_111", f"avito_id_hash={bind.get('avito_id_hash')!r}"
+
+    # rating_distribution должен быть JSON-строкой
+    rd = bind.get("rating_distribution")
+    assert isinstance(rd, str), f"rating_distribution должен быть str (JSON), получили {type(rd)}"
+    assert json.loads(rd) == [{"score": 5, "count": 80}]
+
+
+# ---------------------------------------------------------------------------
+# Test 3: _persist_house не вызывает db.commit()
+# ---------------------------------------------------------------------------
+
+
+def test_persist_house_no_commit():
+    """_persist_house не должен вызывать db.commit() — коммитит вызывающий."""
+    db = MagicMock()
+    db.execute.return_value = MagicMock()
+
+    h = _make_house()
+
+    with patch(
+        "app.services.scrapers.avito_houses.match_or_create_house",
+        return_value=(7, 1.0, "new"),
+    ):
+        _persist_house(db, h, "")
+
+    db.commit.assert_not_called()
+
+
+# ---------------------------------------------------------------------------
+# Test 4: upsert_house делегирует в _persist_house (с пустым url)
+# ---------------------------------------------------------------------------
+
+
+def test_upsert_house_delegates_to_persist_house():
+    """upsert_house(db, h) → _persist_house(db, h, "") → возвращает house_id."""
+    db = MagicMock()
+    db.execute.return_value = MagicMock()
+    h = _make_house()
+
+    with patch(
+        "app.services.scrapers.avito_houses.match_or_create_house",
+        return_value=(55, 1.0, "new"),
+    ):
+        result = upsert_house(db, h)
+
+    assert result == 55
+
+    # Проверяем что UPDATE был вызван с url="" (пустая строка — нет url в этом пути)
+    bind = db.execute.call_args[0][1]
+    assert bind["url"] == "", f"url должен быть пустой строкой, получили {bind.get('url')!r}"
+
+
+# ---------------------------------------------------------------------------
+# Test 5: _upsert_house_with_url делегирует в _persist_house (с реальным url)
+# ---------------------------------------------------------------------------
+
+
+def test_upsert_house_with_url_delegates_to_persist_house():
+    """_upsert_house_with_url(db, h, url) → _persist_house(db, h, url) → house_id."""
+    db = MagicMock()
+    db.execute.return_value = MagicMock()
+    h = _make_house()
+    url = "https://www.avito.ru/catalog/houses/ekb/dom-1234"
+
+    with patch(
+        "app.services.scrapers.avito_houses.match_or_create_house",
+        return_value=(99, 1.0, "new"),
+    ):
+        result = _upsert_house_with_url(db, h, url)
+
+    assert result == 99
+
+    # Проверяем что правильный url передан в UPDATE
+    bind = db.execute.call_args[0][1]
+    assert bind["url"] == url, f"url должен быть {url!r}, получили {bind.get('url')!r}"
+
+
+# ---------------------------------------------------------------------------
+# Test 6: match_or_create_house вызывается с правильными аргументами
+# ---------------------------------------------------------------------------
+
+
+def test_persist_house_calls_match_with_correct_args():
+    """_persist_house передаёт правильные аргументы в match_or_create_house."""
+    db = MagicMock()
+    db.execute.return_value = MagicMock()
+
+    h = _make_house(ext_id=333)
+    url = "https://avito.ru/333"
+
+    with patch(
+        "app.services.scrapers.avito_houses.match_or_create_house",
+        return_value=(10, 1.0, "new"),
+    ) as mock_match:
+        _persist_house(db, h, url)
+
+    mock_match.assert_called_once_with(
+        db,
+        ext_source="avito",
+        ext_id="333",
+        address=h.short_address or h.full_address,
+        lat=h.lat,
+        lon=h.lon,
+        year_built=h.year_built,
+        source_url=url,
+    )
+
+
+# ---------------------------------------------------------------------------
+# Test 7: COALESCE — адрес/координаты не затираются через UPDATE параметры
+#   (проверяем что bind содержит address/lat/lon для COALESCE в SQL)
+# ---------------------------------------------------------------------------
+
+
+def test_persist_house_coalesce_fields_in_bind():
+    """UPDATE передаёт address/lat/lon в bind для COALESCE(houses.<col>, CAST(:x AS ...))."""
+    db = MagicMock()
+    db.execute.return_value = MagicMock()
+
+    h = _make_house()
+
+    with patch(
+        "app.services.scrapers.avito_houses.match_or_create_house",
+        return_value=(1, 1.0, "new"),
+    ):
+        _persist_house(db, h, "")
+
+    bind = db.execute.call_args[0][1]
+    assert bind["address"] == h.short_address
+    assert bind["lat"] == h.lat
+    assert bind["lon"] == h.lon
+    assert bind["year_built"] == h.year_built