bot-backend
b72af1cc49
Корень проблемы: 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.
347 lines
15 KiB
PL/PgSQL
347 lines
15 KiB
PL/PgSQL
-- 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;
|