bot-backend
dca3de10e6
All checks were successful
CI / changes (push) Successful in 9s
CI / backend-tests (push) Has been skipped
CI / frontend-tests (push) Has been skipped
CI / openapi-codegen-check (push) Has been skipped
CI / changes (pull_request) Successful in 8s
CI / backend-tests (pull_request) Has been skipped
CI / frontend-tests (pull_request) Has been skipped
CI / openapi-codegen-check (pull_request) Has been skipped
Деплой упал на 108: (1) c.cluster_key → cl.cluster_key (несуществующий алиас); (2) дедуп-перед-репойнтом сравнивал дубль только с каноном, но два разных дубля одного кластера с одинаковым unique-ключом коллизили при репойнте на канон. Переписаны 3c/houses_price_dynamics, 3d/house_imv_evaluations, 3e/house_suggestions, 3i/address_mismatch_audit на collision-safe паттерн (дедуп по целевому ключу, канон-строка выживает). address_mismatch_audit имеет UNIQUE(house_id,audit_batch) — заголовок ошибочно помечал «нет UNIQUE». Валидировано dry-run против прод-данных (4478 реальных дублей): houses 13456 → 8978, clusters_merged=3878, чисто end-to-end.
371 lines
16 KiB
PL/PgSQL
371 lines
16 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(house_id,audit_batch) → dedup
|
||
--
|
||
-- Идемпотентность: все 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, cl.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)
|
||
-- Collision-safe: дедуп по ЦЕЛЕВОМУ ключу (canonical, month_date, source) —
|
||
-- ловит и dup-vs-canonical, и dup-vs-dup в кластерах >2. Выживает строка
|
||
-- canonical'а (m.canonical_id IS NULL), иначе минимальный id.
|
||
DELETE FROM houses_price_dynamics t
|
||
USING (
|
||
SELECT t2.id,
|
||
ROW_NUMBER() OVER (
|
||
PARTITION BY COALESCE(m.canonical_id, t2.house_id), t2.month_date, t2.source
|
||
ORDER BY (m.canonical_id IS NULL) DESC, t2.id ASC
|
||
) AS rn
|
||
FROM houses_price_dynamics t2
|
||
LEFT JOIN _108_dup_mapping m ON m.dup_id = t2.house_id
|
||
) d
|
||
WHERE t.id = d.id AND d.rn > 1;
|
||
-- Репойнтить оставшиеся (целевой ключ теперь свободен)
|
||
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)
|
||
-- Collision-safe: одна evaluation на canonical. Выживает canonical'а, иначе min id.
|
||
DELETE FROM house_imv_evaluations t
|
||
USING (
|
||
SELECT t2.id,
|
||
ROW_NUMBER() OVER (
|
||
PARTITION BY COALESCE(m.canonical_id, t2.house_id)
|
||
ORDER BY (m.canonical_id IS NULL) DESC, t2.id ASC
|
||
) AS rn
|
||
FROM house_imv_evaluations t2
|
||
LEFT JOIN _108_dup_mapping m ON m.dup_id = t2.house_id
|
||
) d
|
||
WHERE t.id = d.id AND d.rn > 1;
|
||
-- Репойнтить оставшиеся
|
||
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)
|
||
-- Collision-safe: дедуп по (canonical, ext_item_id) — ловит dup-vs-dup.
|
||
DELETE FROM house_suggestions t
|
||
USING (
|
||
SELECT t2.id,
|
||
ROW_NUMBER() OVER (
|
||
PARTITION BY COALESCE(m.canonical_id, t2.house_id), t2.ext_item_id
|
||
ORDER BY (m.canonical_id IS NULL) DESC, t2.id ASC
|
||
) AS rn
|
||
FROM house_suggestions t2
|
||
LEFT JOIN _108_dup_mapping m ON m.dup_id = t2.house_id
|
||
) d
|
||
WHERE t.id = d.id AND d.rn > 1;
|
||
-- Репойнтить оставшиеся
|
||
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 UNIQUE(house_id, audit_batch) — collision-safe
|
||
-- (заголовок ранее ошибочно помечал «нет блокирующего UNIQUE»).
|
||
DELETE FROM address_mismatch_audit t
|
||
USING (
|
||
SELECT t2.id,
|
||
ROW_NUMBER() OVER (
|
||
PARTITION BY COALESCE(m.canonical_id, t2.house_id), t2.audit_batch
|
||
ORDER BY (m.canonical_id IS NULL) DESC, t2.id ASC
|
||
) AS rn
|
||
FROM address_mismatch_audit t2
|
||
LEFT JOIN _108_dup_mapping m ON m.dup_id = t2.house_id
|
||
) d
|
||
WHERE t.id = d.id AND d.rn > 1;
|
||
-- Репойнтить оставшиеся
|
||
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;
|