lekss361/gendesign
1
0
Fork 0
Code Issues 57 Pull requests Projects Releases Packages Wiki Activity Actions

fix(tradein): avito houses dedup через match_or_create_house + merge-миграция дублей (108) #1546

Merged
lekss361 merged 2 commits from fix/avito-houses-dedup-merge into main 2026-06-16 05:38:24 +00:00
Conversation 0 Commits 2 Files changed 3 +703 -186
3 changed files with 703 additions and 186 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

271
tradein-mvp/backend/app/services/scrapers/avito_houses.py
View file

@ -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 (
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 +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:

347
tradein-mvp/backend/data/sql/108_merge_duplicate_houses.sql Normal file
View file

@ -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;

271
tradein-mvp/backend/tests/test_avito_houses_dedup.py Normal file
View file

@ -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