From b72af1cc499566e42fc056bf639af45ac73c713c Mon Sep 17 00:00:00 2001 From: bot-backend Date: Mon, 15 Jun 2026 22:40:46 +0300 Subject: [PATCH] =?UTF-8?q?fix(tradein):=20avito=20houses=20dedup=20=D1=87?= =?UTF-8?q?=D0=B5=D1=80=D0=B5=D0=B7=20match=5For=5Fcreate=5Fhouse=20+=20me?= =?UTF-8?q?rge-=D0=BC=D0=B8=D0=B3=D1=80=D0=B0=D1=86=D0=B8=D1=8F=20=D0=B4?= =?UTF-8?q?=D1=83=D0=B1=D0=BB=D0=B5=D0=B9=20(108)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Корень проблемы: avito_houses.py обходил канонический matching-путь через прямой INSERT ON CONFLICT(source, ext_house_id), создавая дубли для физически одного дома (~33% таблицы houses = ~4479 лишних строк). Forward fix: upsert_house/upsert_house_with_url → _persist_house(db, h, url), который вызывает match_or_create_house (3-tier: source_exact→fingerprint→geo→new, с pg_advisory_xact_lock) и затем UPDATE с enrichment-полями. avito-авторитетные поля (rating, developer, технические хар-ки) перезаписываются напрямую; address/lat/lon/year_built — через COALESCE чтобы не затирать canonical-данные. Merge-миграция 108: cluster_key по cadastral/address/geo → canonical (max не-NULL полей, geom присутствует, min(id)) → репойнт всех 11 зависимых таблиц с обработкой UNIQUE-коллизий → DELETE дублей → backfill house_sources/aliases. --- .../app/services/scrapers/avito_houses.py | 270 +++++--------- .../data/sql/108_merge_duplicate_houses.sql | 347 ++++++++++++++++++ .../backend/tests/test_avito_houses_dedup.py | 271 ++++++++++++++ 3 files changed, 702 insertions(+), 186 deletions(-) create mode 100644 tradein-mvp/backend/data/sql/108_merge_duplicate_houses.sql create mode 100644 tradein-mvp/backend/tests/test_avito_houses_dedup.py diff --git a/tradein-mvp/backend/app/services/scrapers/avito_houses.py b/tradein-mvp/backend/app/services/scrapers/avito_houses.py index c443b810..24dabf85 100644 --- a/tradein-mvp/backend/app/services/scrapers/avito_houses.py +++ b/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,86 @@ 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), + address=h.full_address or h.short_address, + lat=h.lat, + lon=h.lon, + year_built=h.year_built, + source_url=house_url or None, + ) + + 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 + UPDATE houses SET + url = CAST(:url AS text), + short_address = CAST(:short_address AS text), + full_address = CAST(:full_address AS text), + avito_id_hash = CAST(:avito_id_hash AS text), + house_type = CAST(:house_type AS text), + total_floors = CAST(:total_floors AS int), + passenger_elevators = CAST(:passenger_elevators AS int), + cargo_elevators = CAST(:cargo_elevators AS int), + has_concierge = CAST(:has_concierge AS boolean), + closed_yard = CAST(:closed_yard AS boolean), + material_floors = CAST(:material_floors AS text), + hot_water = CAST(:hot_water AS text), + parking_type = CAST(:parking_type AS text), + has_playground = CAST(:has_playground AS boolean), + developer_name = CAST(:developer_name AS text), + developer_key = CAST(:developer_key AS text), + infrastructure_summary = CAST(:infrastructure_summary AS text), + infrastructure_walk_distance = CAST(:infrastructure_walk_distance AS text), + rating = CAST(:rating AS numeric), + rating_score = CAST(:rating_score AS numeric), + rating_string = CAST(:rating_string AS text), + reviews_count = CAST(:reviews_count AS int), + rating_distribution = CAST(:rating_distribution AS jsonb), + map_pins = CAST(:map_pins AS jsonb), + raw_characteristics = CAST(:raw_characteristics AS jsonb), + address = COALESCE(houses.address, CAST(:address AS text)), + lat = COALESCE(houses.lat, CAST(:lat AS double precision)), + lon = COALESCE(houses.lon, CAST(:lon AS double precision)), + year_built = COALESCE(houses.year_built, CAST(:year_built AS int)), + last_scraped_at = NOW() + WHERE id = CAST(:house_id AS bigint) """), { - "ext_house_id": str(h.ext_id), - "url": "", # caller должен передавать house_url через enrichment - "address": h.short_address, + "house_id": house_id, + "url": house_url or "", "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 +872,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: diff --git a/tradein-mvp/backend/data/sql/108_merge_duplicate_houses.sql b/tradein-mvp/backend/data/sql/108_merge_duplicate_houses.sql new file mode 100644 index 00000000..beb9992e --- /dev/null +++ b/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; diff --git a/tradein-mvp/backend/tests/test_avito_houses_dedup.py b/tradein-mvp/backend/tests/test_avito_houses_dedup.py new file mode 100644 index 00000000..48b35dee --- /dev/null +++ b/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.full_address or h.short_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., 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