757 lines
33 KiB
Python
757 lines
33 KiB
Python
"""Recurring, idempotent, collision-safe house-deduplication merge (#1772).
|
||
|
||
WHY (recurrence, not a one-shot):
|
||
Migration 108_merge_duplicate_houses.sql did a ONE-TIME merge of duplicate `houses`
|
||
rows, but the duplicates RECUR: the matching pipeline keeps creating them
|
||
(Tier-3 geo-jitter splits one physical building across rows + per-source ext_id INSERTs).
|
||
A single migration cannot fix a steady-state inflow — we need a RECURRING merge that
|
||
re-runs on a schedule and is a no-op when there is nothing to merge.
|
||
|
||
WHAT this is:
|
||
A direct, set-based reimplementation of 108's proven
|
||
cluster → pick canonical → re-point children (UNIQUE-collision-safe) → delete losers
|
||
pipeline, run inside ONE transaction so a crash leaves the table untouched.
|
||
|
||
Cluster key: CANONICAL address via tradein_canon_addr() over the CLEAN address
|
||
COALESCE(short_address, full_address, address) (cadastral_number is 100% NULL on prod —
|
||
confirmed in migration 040 — so address is the real building key). The clean source matters:
|
||
`address` can carry район-noise the canon does not strip (e.g. «улица Вайнера, 66 · р-н Центр»
|
||
→ canon «вайнера66рнцентр»), while `short_address` holds the clean «улица Вайнера, 66»
|
||
(→ «вайнера66») — preferring the clean field lets such a row cluster with its twin. The canon
|
||
collapses spelling/район variants of the SAME building (ул→улица, strips город/район/«россия»/
|
||
«м-н»/«р-н» noise, removes spaces/punct, PRESERVES корпус) so e.g. «ул. Вайнера,66» and
|
||
«улица Вайнера, 66» share one cluster_key. Rows whose canon is NULL/blank are never clustered
|
||
(cluster_key NULL → ignored). Only canons shared by >1 house_id form a cluster.
|
||
|
||
GEO GUARD (anti over-merge): because the canon strips город/район, two different buildings
|
||
with the same street+number in different region-66 towns (e.g. «Ленина 5») would share a
|
||
canon. To avoid merging them, within a canon-cluster a house is a LOSER only if it is within
|
||
250 m of the keeper (ST_DistanceSphere on the WGS84 geom). Same-canon houses >250 m away — or
|
||
with NULL geom on either side — are left as separate rows (conservative; never over-merges).
|
||
|
||
TWO PASSES (2026-07-02 follow-up): the SAME cluster→keeper→re-point→carry-identity→delete
|
||
pipeline now runs TWICE inside one transaction, parametrised by the cluster-key expression
|
||
(`_mapping_sql`) — no 300-line copy-paste:
|
||
1. FIAS pass — clusters by lower(NULLIF(house_fias_id, '')) (the ФИАС/ГАР building UUID,
|
||
DaData-backfilled 2026-07-02 for 5 122 houses). Only non-null fias shared by >1 house_id
|
||
cluster. Catches duplicates the canon MISSES: slash-collapse («Сулимова, 32» vs
|
||
«Сулимова, 3/2» → same canon «сулимова32») and посёлок truncation (Кедровка/Шувакиш
|
||
colliding with a same-named ЕКБ street).
|
||
2. CANON pass — the canonical-address clustering above, now with a CROSS-FIAS GUARD: within a
|
||
canon cluster a loser is NOT merged when it AND the keeper both carry a non-null but
|
||
DIFFERENT house_fias_id — provably different buildings the canon collapsed (the
|
||
slash-collapse class). Critical anti-over-merge fix. The 250 m geo guard applies to both
|
||
passes (after the backfill, fias rows almost always have geom → conservative).
|
||
|
||
IDENTITY CARRY-OVER (each pass, BEFORE deleting losers): the keeper's NULL identity / geo-QC
|
||
fields are filled from its losers with COALESCE semantics (keeper value wins; donor = the
|
||
loser with geom, then min id): house_fias_id, cadastral_number, gar_house_guid, gar_flat_count,
|
||
gar_matched_at, gar_match_method, dadata_qc_geo, dadata_qc_house, dadata_enriched_at. Without
|
||
this the merge would silently drop DaData/ГАР enrichment that lived only on a loser row.
|
||
|
||
KEEPER RULE (one canonical row per cluster), in priority order:
|
||
1. geom NOT NULL first (a geocoded row is the better survivor)
|
||
2. most linked listings (the row the corpus already points at)
|
||
3. most-populated fields (the richest metadata row)
|
||
4. min(id) (deterministic tie-break)
|
||
|
||
CHILD-COLLISION HANDLING — the migration-133 lesson:
|
||
Migration 133 broke on prod because a child UNIQUE constraint was missed. EVERY FK child
|
||
of houses(id) is handled below. The FK set was re-audited against the LIVE schema
|
||
(grep "REFERENCES houses" data/sql/*.sql) — it is the SAME 11 children 108 handled; no
|
||
new FK child was added after 108. For each UNIQUE-constrained child we DELETE the loser
|
||
rows that would collide with the keeper (deduping by the TARGET key so dup-vs-dup
|
||
collisions inside clusters >2 are also caught) BEFORE re-pointing the survivors.
|
||
|
||
IMPORTANT divergence from 108: houses_price_dynamics's UNIQUE was REPLACED by
|
||
migration 029 — it is now the 6-column
|
||
(house_id, source, room_count, prices_type, period, month_date)
|
||
NOT 108's stale 3-column (house_id, month_date, source). We dedup on the LIVE 6-column
|
||
key; deduping on the old key would still violate the live constraint on re-point.
|
||
|
||
Per-child handling (loser=L, keeper=K):
|
||
listings.house_id_fk no UNIQUE → plain re-point
|
||
house_placement_history.house_id no UNIQUE → plain re-point
|
||
house_reviews.house_id no blocking UNIQUE→ plain re-point
|
||
house_reliability_checks.house_id no blocking UNIQUE→ plain re-point
|
||
external_valuations.house_id no blocking UNIQUE→ plain re-point
|
||
house_sources UNIQUE(ext_source,ext_id) → delete colliding L, re-point rest
|
||
house_address_aliases UNIQUE(normalized_address)→ delete colliding L, re-point rest
|
||
houses_price_dynamics UNIQUE(6-col, migr 029) → dedup by target 6-col key, re-point
|
||
house_imv_evaluations UNIQUE(house_id) → dedup by target house_id, re-point
|
||
house_suggestions UNIQUE(house_id,ext_item_id)→ dedup by target key, re-point
|
||
address_mismatch_audit UNIQUE(house_id,audit_batch)→ dedup by target key, re-point
|
||
|
||
BACKFILL (reduces recurrence):
|
||
After deleting losers we backfill house_sources + house_address_aliases onto the keeper
|
||
(same as 108) so the matching pipeline's Tier-1/Tier-2 finds the keeper next scrape and
|
||
does not immediately re-split it.
|
||
|
||
IDEMPOTENCY:
|
||
Every UPDATE/DELETE keys off a temp mapping of (loser→keeper). On a clean table the
|
||
mapping is empty → every statement touches 0 rows → no-op. Re-running is safe.
|
||
|
||
DRY-RUN:
|
||
dry_run=True computes the same mapping and counts, then ROLLS BACK — no writes. Used by
|
||
the orchestrator to preview a merge before enabling the (DESTRUCTIVE) schedule.
|
||
|
||
psycopg v3: all SQL uses CAST(:x AS type), never the colon-colon bound-param cast form.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import logging
|
||
import time
|
||
from dataclasses import dataclass, field
|
||
from typing import Any
|
||
|
||
from sqlalchemy import text
|
||
from sqlalchemy.orm import Session
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
# Significant columns counted to score per-row completeness when picking the keeper.
|
||
# Mirrors 108's completeness expression (all confirmed present in the live houses schema).
|
||
_COMPLETENESS_EXPR = """
|
||
(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
|
||
"""
|
||
|
||
# Keeper ORDER BY, shared by the ROW_NUMBER() rank and the first_value() keeper pick so they
|
||
# agree row-for-row. Priority: geom present → most linked listings → most-populated → min id.
|
||
_KEEPER_ORDER = f"""
|
||
(h.geom IS NOT NULL) DESC,
|
||
listing_cnt DESC,
|
||
({_COMPLETENESS_EXPR}) DESC,
|
||
h.id ASC
|
||
"""
|
||
|
||
# Cluster-key expressions, one per merge pass. The pipeline (build mapping → keeper → re-point
|
||
# children → carry identity → delete losers → backfill) is IDENTICAL for both passes; only the
|
||
# cluster key differs, parametrised into `_mapping_sql`, so there is NO copy-paste of the pipeline.
|
||
#
|
||
# FIAS key (#1772 follow-up, 2026-07-02): the ФИАС/ГАР building UUID (DaData backfill). Catches
|
||
# duplicates the canon misses — slash-collapse («Сулимова, 32» vs «Сулимова, 3/2») and посёлок
|
||
# truncation (Кедровка/Шувакиш). Only non-null fias clusters.
|
||
_FIAS_KEY_EXPR = """
|
||
CASE WHEN NULLIF(house_fias_id, '') IS NOT NULL
|
||
THEN 'fias:' || lower(house_fias_id)
|
||
END
|
||
"""
|
||
|
||
# CANON key: canonical address over the CLEAN address (short_address→full_address→address). The
|
||
# canon collapses spelling/район variants of the SAME building (ул→улица, strips город/район noise,
|
||
# preserves корпус); the digit guard drops street-less degenerate canons; blank canons never join.
|
||
_CANON_KEY_EXPR = """
|
||
CASE WHEN NULLIF(
|
||
tradein_canon_addr(COALESCE(short_address, full_address, address)), ''
|
||
) IS NOT NULL
|
||
AND tradein_canon_addr(COALESCE(short_address, full_address, address)) ~ '[0-9]'
|
||
THEN 'addr:'
|
||
|| tradein_canon_addr(COALESCE(short_address, full_address, address))
|
||
END
|
||
"""
|
||
|
||
|
||
def _mapping_sql(cluster_key_case: str) -> str:
|
||
"""Render the loser→keeper mapping SQL for one pass, given its cluster-key CASE expression.
|
||
|
||
Only cluster keys shared by >1 house_id form a cluster; the keeper is rn=1 per cluster, losers
|
||
are rn>1. The final mapping is filtered by the 250 m geo guard AND the CROSS-FIAS guard (both
|
||
no-ops for the fias pass, where every clustered row shares one fias by construction).
|
||
`cluster_key_case` is a STATIC module constant (never runtime data) — no value injection.
|
||
"""
|
||
return f"""
|
||
CREATE TEMP TABLE _1772_dup_mapping ON COMMIT DROP AS
|
||
WITH clustered AS (
|
||
SELECT
|
||
id,
|
||
{cluster_key_case} 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
|
||
),
|
||
listing_counts AS (
|
||
SELECT house_id_fk AS house_id, count(*) AS listing_cnt
|
||
FROM listings
|
||
WHERE house_id_fk IS NOT NULL
|
||
GROUP BY house_id_fk
|
||
),
|
||
ranked AS (
|
||
SELECT
|
||
dh.id,
|
||
dh.cluster_key,
|
||
lower(trim(COALESCE(h.short_address, h.full_address, h.address))) AS norm_address,
|
||
h.geom AS loser_geom,
|
||
h.house_fias_id AS loser_fias,
|
||
ROW_NUMBER() OVER (
|
||
PARTITION BY dh.cluster_key
|
||
ORDER BY {_KEEPER_ORDER}
|
||
) AS rn,
|
||
first_value(dh.id) OVER (
|
||
PARTITION BY dh.cluster_key
|
||
ORDER BY {_KEEPER_ORDER}
|
||
) AS keeper_id,
|
||
first_value(h.geom) OVER (
|
||
PARTITION BY dh.cluster_key
|
||
ORDER BY {_KEEPER_ORDER}
|
||
) AS keeper_geom,
|
||
first_value(h.house_fias_id) OVER (
|
||
PARTITION BY dh.cluster_key
|
||
ORDER BY {_KEEPER_ORDER}
|
||
) AS keeper_fias
|
||
FROM dup_houses dh
|
||
JOIN houses h ON h.id = dh.id
|
||
LEFT JOIN listing_counts lc ON lc.house_id = dh.id
|
||
)
|
||
-- Geo guard against cross-town over-merge: tradein_canon_addr strips город/район, so two
|
||
-- different buildings sharing a street+number canon (e.g. «Ленина 5» in different towns)
|
||
-- collapse to one cluster_key. A loser is only merged when it is geographically next to the
|
||
-- keeper (<=250 m — покрывает geocode-разброс одного здания, прод: дубли Мраморская 34к4 в
|
||
-- 222 m; города региона-66 в км+ друг от друга → 250 m безопасно от cross-town). Houses in
|
||
-- the same canon-cluster but >250 m away, or with NULL geom on either side, are left as
|
||
-- separate rows (conservative — never over-merges). Cluster_key также требует цифру в canon
|
||
-- (номер дома) — исключает вырожденные canon вида «екатеринбург»/«сооружение».
|
||
-- CROSS-FIAS guard (#1772 follow-up): never merge two rows that BOTH carry a non-null but
|
||
-- DIFFERENT house_fias_id — they are provably different buildings the cluster key happened to
|
||
-- collapse (canon slash-collapse «Сулимова, 32»/«Сулимова, 3/2»). No-op for the fias pass (one
|
||
-- fias per cluster) and for canon clusters where at most one side has a fias.
|
||
SELECT id AS loser_id, keeper_id, norm_address
|
||
FROM ranked
|
||
WHERE rn > 1
|
||
AND id <> keeper_id
|
||
AND keeper_geom IS NOT NULL
|
||
AND loser_geom IS NOT NULL
|
||
AND ST_DistanceSphere(loser_geom, keeper_geom) <= 250
|
||
AND NOT (
|
||
NULLIF(loser_fias, '') IS NOT NULL
|
||
AND NULLIF(keeper_fias, '') IS NOT NULL
|
||
AND lower(loser_fias) <> lower(keeper_fias)
|
||
)
|
||
"""
|
||
|
||
|
||
# Canon-pass mapping — kept as a module constant for backward-compatible introspection/tests.
|
||
_BUILD_MAPPING_SQL = text(_mapping_sql(_CANON_KEY_EXPR))
|
||
# Fias-pass mapping — same pipeline, clustered by the ФИАС building UUID (runs first).
|
||
_BUILD_MAPPING_SQL_FIAS = text(_mapping_sql(_FIAS_KEY_EXPR))
|
||
|
||
# Each step keys off _1772_dup_mapping → empty mapping ⇒ 0 rows touched ⇒ idempotent no-op.
|
||
_STEPS: list[tuple[str, str]] = [
|
||
# ── Plain re-point (no UNIQUE on the FK column) ───────────────────────────
|
||
(
|
||
"listings",
|
||
"""
|
||
UPDATE listings l
|
||
SET house_id_fk = m.keeper_id
|
||
FROM _1772_dup_mapping m
|
||
WHERE l.house_id_fk = m.loser_id
|
||
""",
|
||
),
|
||
(
|
||
"house_placement_history",
|
||
"""
|
||
UPDATE house_placement_history hph
|
||
SET house_id = m.keeper_id
|
||
FROM _1772_dup_mapping m
|
||
WHERE hph.house_id = m.loser_id
|
||
""",
|
||
),
|
||
(
|
||
"house_reviews",
|
||
"""
|
||
UPDATE house_reviews hr
|
||
SET house_id = m.keeper_id
|
||
FROM _1772_dup_mapping m
|
||
WHERE hr.house_id = m.loser_id
|
||
""",
|
||
),
|
||
(
|
||
"house_reliability_checks",
|
||
"""
|
||
UPDATE house_reliability_checks hrc
|
||
SET house_id = m.keeper_id
|
||
FROM _1772_dup_mapping m
|
||
WHERE hrc.house_id = m.loser_id
|
||
""",
|
||
),
|
||
(
|
||
"external_valuations",
|
||
"""
|
||
UPDATE external_valuations ev
|
||
SET house_id = m.keeper_id
|
||
FROM _1772_dup_mapping m
|
||
WHERE ev.house_id = m.loser_id
|
||
""",
|
||
),
|
||
# ── UNIQUE(ext_source, ext_id): delete colliding losers, re-point rest ─────
|
||
(
|
||
"house_sources(collision-delete)",
|
||
"""
|
||
DELETE FROM house_sources hs
|
||
USING _1772_dup_mapping m
|
||
WHERE hs.house_id = m.loser_id
|
||
AND EXISTS (
|
||
SELECT 1 FROM house_sources hs2
|
||
WHERE hs2.house_id = m.keeper_id
|
||
AND hs2.ext_source = hs.ext_source
|
||
AND hs2.ext_id = hs.ext_id
|
||
)
|
||
""",
|
||
),
|
||
(
|
||
"house_sources(re-point)",
|
||
"""
|
||
UPDATE house_sources hs
|
||
SET house_id = m.keeper_id
|
||
FROM _1772_dup_mapping m
|
||
WHERE hs.house_id = m.loser_id
|
||
""",
|
||
),
|
||
# ── UNIQUE(normalized_address): delete colliding losers, re-point rest ─────
|
||
(
|
||
"house_address_aliases(collision-delete)",
|
||
"""
|
||
DELETE FROM house_address_aliases haa
|
||
USING _1772_dup_mapping m
|
||
WHERE haa.house_id = m.loser_id
|
||
AND EXISTS (
|
||
SELECT 1 FROM house_address_aliases haa2
|
||
WHERE haa2.house_id = m.keeper_id
|
||
AND haa2.normalized_address = haa.normalized_address
|
||
)
|
||
""",
|
||
),
|
||
(
|
||
"house_address_aliases(re-point)",
|
||
"""
|
||
UPDATE house_address_aliases haa
|
||
SET house_id = m.keeper_id
|
||
FROM _1772_dup_mapping m
|
||
WHERE haa.house_id = m.loser_id
|
||
""",
|
||
),
|
||
# ── UNIQUE(house_id, source, room_count, prices_type, period, month_date) ──
|
||
# LIVE 6-col key (migration 029 replaced 108's stale 3-col). Dedup by the TARGET key
|
||
# (keeper survives, else min id) so dup-vs-dup collisions in clusters >2 are also caught.
|
||
(
|
||
"houses_price_dynamics(dedup)",
|
||
"""
|
||
DELETE FROM houses_price_dynamics t
|
||
USING (
|
||
SELECT t2.id,
|
||
ROW_NUMBER() OVER (
|
||
PARTITION BY COALESCE(m.keeper_id, t2.house_id),
|
||
t2.source, t2.room_count, t2.prices_type,
|
||
t2.period, t2.month_date
|
||
ORDER BY (m.keeper_id IS NULL) DESC, t2.id ASC
|
||
) AS rn
|
||
FROM houses_price_dynamics t2
|
||
LEFT JOIN _1772_dup_mapping m ON m.loser_id = t2.house_id
|
||
) d
|
||
WHERE t.id = d.id AND d.rn > 1
|
||
""",
|
||
),
|
||
(
|
||
"houses_price_dynamics(re-point)",
|
||
"""
|
||
UPDATE houses_price_dynamics hpd
|
||
SET house_id = m.keeper_id
|
||
FROM _1772_dup_mapping m
|
||
WHERE hpd.house_id = m.loser_id
|
||
""",
|
||
),
|
||
# ── UNIQUE(house_id): one evaluation per keeper ───────────────────────────
|
||
(
|
||
"house_imv_evaluations(dedup)",
|
||
"""
|
||
DELETE FROM house_imv_evaluations t
|
||
USING (
|
||
SELECT t2.id,
|
||
ROW_NUMBER() OVER (
|
||
PARTITION BY COALESCE(m.keeper_id, t2.house_id)
|
||
ORDER BY (m.keeper_id IS NULL) DESC, t2.id ASC
|
||
) AS rn
|
||
FROM house_imv_evaluations t2
|
||
LEFT JOIN _1772_dup_mapping m ON m.loser_id = t2.house_id
|
||
) d
|
||
WHERE t.id = d.id AND d.rn > 1
|
||
""",
|
||
),
|
||
(
|
||
"house_imv_evaluations(re-point)",
|
||
"""
|
||
UPDATE house_imv_evaluations hie
|
||
SET house_id = m.keeper_id
|
||
FROM _1772_dup_mapping m
|
||
WHERE hie.house_id = m.loser_id
|
||
""",
|
||
),
|
||
# ── UNIQUE(house_id, ext_item_id) ─────────────────────────────────────────
|
||
(
|
||
"house_suggestions(dedup)",
|
||
"""
|
||
DELETE FROM house_suggestions t
|
||
USING (
|
||
SELECT t2.id,
|
||
ROW_NUMBER() OVER (
|
||
PARTITION BY COALESCE(m.keeper_id, t2.house_id), t2.ext_item_id
|
||
ORDER BY (m.keeper_id IS NULL) DESC, t2.id ASC
|
||
) AS rn
|
||
FROM house_suggestions t2
|
||
LEFT JOIN _1772_dup_mapping m ON m.loser_id = t2.house_id
|
||
) d
|
||
WHERE t.id = d.id AND d.rn > 1
|
||
""",
|
||
),
|
||
(
|
||
"house_suggestions(re-point)",
|
||
"""
|
||
UPDATE house_suggestions hs
|
||
SET house_id = m.keeper_id
|
||
FROM _1772_dup_mapping m
|
||
WHERE hs.house_id = m.loser_id
|
||
""",
|
||
),
|
||
# ── UNIQUE(house_id, audit_batch) ─────────────────────────────────────────
|
||
(
|
||
"address_mismatch_audit(dedup)",
|
||
"""
|
||
DELETE FROM address_mismatch_audit t
|
||
USING (
|
||
SELECT t2.id,
|
||
ROW_NUMBER() OVER (
|
||
PARTITION BY COALESCE(m.keeper_id, t2.house_id), t2.audit_batch
|
||
ORDER BY (m.keeper_id IS NULL) DESC, t2.id ASC
|
||
) AS rn
|
||
FROM address_mismatch_audit t2
|
||
LEFT JOIN _1772_dup_mapping m ON m.loser_id = t2.house_id
|
||
) d
|
||
WHERE t.id = d.id AND d.rn > 1
|
||
""",
|
||
),
|
||
(
|
||
"address_mismatch_audit(re-point)",
|
||
"""
|
||
UPDATE address_mismatch_audit ama
|
||
SET house_id = m.keeper_id
|
||
FROM _1772_dup_mapping m
|
||
WHERE ama.house_id = m.loser_id
|
||
""",
|
||
),
|
||
]
|
||
|
||
# Delete the loser houses — all FK children are re-pointed or CASCADE by now.
|
||
_DELETE_LOSERS_SQL = text(
|
||
"""
|
||
DELETE FROM houses h
|
||
USING _1772_dup_mapping m
|
||
WHERE h.id = m.loser_id
|
||
"""
|
||
)
|
||
|
||
# Backfill house_sources for keeper avito-rows so Tier-1 (source_exact) finds the keeper next
|
||
# scrape (reduces recurrence). Same shape as 108 Step 5.
|
||
_BACKFILL_SOURCES_SQL = text(
|
||
"""
|
||
INSERT INTO house_sources (house_id, ext_source, ext_id, confidence, matched_method, matched_at)
|
||
SELECT h.id, h.source, h.ext_house_id, 1.0, 'backfill_dedup_merge', NOW()
|
||
FROM houses h
|
||
WHERE h.source = 'avito'
|
||
AND h.ext_house_id IS NOT NULL
|
||
AND h.id IN (SELECT DISTINCT keeper_id FROM _1772_dup_mapping)
|
||
ON CONFLICT (ext_source, ext_id) DO NOTHING
|
||
"""
|
||
)
|
||
|
||
# Backfill normalized-address aliases onto the keeper so Tier-2 (address) finds it next scrape.
|
||
_BACKFILL_ALIASES_SQL = text(
|
||
"""
|
||
INSERT INTO house_address_aliases (house_id, normalized_address, fingerprint, source)
|
||
SELECT h.id, lower(trim(h.address)), NULL, 'backfill_dedup_merge'
|
||
FROM houses h
|
||
WHERE h.address IS NOT NULL
|
||
AND length(trim(h.address)) >= 5
|
||
AND h.id IN (SELECT DISTINCT keeper_id FROM _1772_dup_mapping)
|
||
ON CONFLICT (normalized_address) DO NOTHING
|
||
"""
|
||
)
|
||
|
||
# Carry identity / geo-QC / ГАР-enrichment onto the keeper BEFORE deleting losers (#1772
|
||
# follow-up). COALESCE — the keeper's own populated value always wins; a NULL keeper field is
|
||
# filled from a donor loser. Donor = the loser with geom first, then min id (deterministic):
|
||
# array_agg(... ORDER BY geom-present DESC, id ASC) FILTER (WHERE <field> NOT NULL) [1] takes the
|
||
# first non-null value in that order. Without this the merge would drop DaData/ГАР enrichment that
|
||
# lived only on a loser row (e.g. house_fias_id backfilled onto the loser but not the keeper).
|
||
_CARRY_OVER_IDENTITY_SQL = text(
|
||
"""
|
||
UPDATE houses k
|
||
SET
|
||
house_fias_id = COALESCE(k.house_fias_id, d.house_fias_id),
|
||
cadastral_number = COALESCE(k.cadastral_number, d.cadastral_number),
|
||
gar_house_guid = COALESCE(k.gar_house_guid, d.gar_house_guid),
|
||
gar_flat_count = COALESCE(k.gar_flat_count, d.gar_flat_count),
|
||
gar_matched_at = COALESCE(k.gar_matched_at, d.gar_matched_at),
|
||
gar_match_method = COALESCE(k.gar_match_method, d.gar_match_method),
|
||
dadata_qc_geo = COALESCE(k.dadata_qc_geo, d.dadata_qc_geo),
|
||
dadata_qc_house = COALESCE(k.dadata_qc_house, d.dadata_qc_house),
|
||
dadata_enriched_at = COALESCE(k.dadata_enriched_at, d.dadata_enriched_at)
|
||
FROM (
|
||
SELECT
|
||
m.keeper_id,
|
||
(array_agg(l.house_fias_id
|
||
ORDER BY (l.geom IS NOT NULL) DESC, l.id ASC)
|
||
FILTER (WHERE l.house_fias_id IS NOT NULL))[1] AS house_fias_id,
|
||
(array_agg(l.cadastral_number
|
||
ORDER BY (l.geom IS NOT NULL) DESC, l.id ASC)
|
||
FILTER (WHERE l.cadastral_number IS NOT NULL))[1] AS cadastral_number,
|
||
(array_agg(l.gar_house_guid
|
||
ORDER BY (l.geom IS NOT NULL) DESC, l.id ASC)
|
||
FILTER (WHERE l.gar_house_guid IS NOT NULL))[1] AS gar_house_guid,
|
||
(array_agg(l.gar_flat_count
|
||
ORDER BY (l.geom IS NOT NULL) DESC, l.id ASC)
|
||
FILTER (WHERE l.gar_flat_count IS NOT NULL))[1] AS gar_flat_count,
|
||
(array_agg(l.gar_matched_at
|
||
ORDER BY (l.geom IS NOT NULL) DESC, l.id ASC)
|
||
FILTER (WHERE l.gar_matched_at IS NOT NULL))[1] AS gar_matched_at,
|
||
(array_agg(l.gar_match_method
|
||
ORDER BY (l.geom IS NOT NULL) DESC, l.id ASC)
|
||
FILTER (WHERE l.gar_match_method IS NOT NULL))[1] AS gar_match_method,
|
||
(array_agg(l.dadata_qc_geo
|
||
ORDER BY (l.geom IS NOT NULL) DESC, l.id ASC)
|
||
FILTER (WHERE l.dadata_qc_geo IS NOT NULL))[1] AS dadata_qc_geo,
|
||
(array_agg(l.dadata_qc_house
|
||
ORDER BY (l.geom IS NOT NULL) DESC, l.id ASC)
|
||
FILTER (WHERE l.dadata_qc_house IS NOT NULL))[1] AS dadata_qc_house,
|
||
(array_agg(l.dadata_enriched_at
|
||
ORDER BY (l.geom IS NOT NULL) DESC, l.id ASC)
|
||
FILTER (WHERE l.dadata_enriched_at IS NOT NULL))[1] AS dadata_enriched_at
|
||
FROM _1772_dup_mapping m
|
||
JOIN houses l ON l.id = m.loser_id
|
||
GROUP BY m.keeper_id
|
||
) d
|
||
WHERE k.id = d.keeper_id
|
||
"""
|
||
)
|
||
|
||
|
||
@dataclass
|
||
class DedupMergeResult:
|
||
"""Counters from one merge pass (also the dry-run preview)."""
|
||
|
||
clusters_merged: int = 0 # distinct keepers that absorbed at least one loser
|
||
losers_deleted: int = 0 # duplicate house rows removed (0 on a clean table)
|
||
listings_repointed: int = 0 # listings.house_id_fk moved loser→keeper
|
||
children_deleted: int = 0 # collision/dedup deletions across all UNIQUE children
|
||
children_repointed: int = 0 # survivor child rows moved loser→keeper
|
||
dry_run: bool = False
|
||
duration_sec: float = field(default=0.0)
|
||
|
||
def to_counters(self) -> dict[str, int]:
|
||
return {
|
||
"clusters_merged": self.clusters_merged,
|
||
"losers_deleted": self.losers_deleted,
|
||
"listings_repointed": self.listings_repointed,
|
||
"children_deleted": self.children_deleted,
|
||
"children_repointed": self.children_repointed,
|
||
"dry_run": int(self.dry_run),
|
||
"duration_sec": int(self.duration_sec),
|
||
}
|
||
|
||
|
||
def _run_merge_pass(
|
||
db: Session,
|
||
*,
|
||
build_sql: Any,
|
||
pass_label: str,
|
||
result: DedupMergeResult,
|
||
) -> None:
|
||
"""Run ONE merge pass (fias- or canon-key) inside the caller's open transaction.
|
||
|
||
Builds a fresh loser→keeper mapping for this pass's cluster key, re-points every FK child
|
||
(UNIQUE-collision-safe), carries identity/enrichment onto the keeper, deletes the losers and
|
||
backfills sources/aliases. Accumulates counters onto `result`. NEVER commits/rolls back — the
|
||
caller owns the single transaction wrapping both passes.
|
||
"""
|
||
# Fresh mapping for this pass. ON COMMIT DROP only fires at txn end, so drop the temp table
|
||
# explicitly — the second pass must rebuild the same-named table within the one transaction.
|
||
db.execute(text("DROP TABLE IF EXISTS _1772_dup_mapping"))
|
||
db.execute(build_sql)
|
||
|
||
mapping = db.execute(
|
||
text(
|
||
"SELECT loser_id, keeper_id, norm_address FROM _1772_dup_mapping "
|
||
"ORDER BY keeper_id, loser_id"
|
||
)
|
||
).all()
|
||
if not mapping:
|
||
logger.info("merge_duplicate_houses: pass=%s no clusters found (no-op)", pass_label)
|
||
return
|
||
|
||
result.losers_deleted += len(mapping)
|
||
result.clusters_merged += len({row.keeper_id for row in mapping})
|
||
|
||
# Audit log: every loser→keeper move with its address, for traceability.
|
||
for row in mapping:
|
||
logger.info(
|
||
"merge_duplicate_houses: pass=%s merge loser_id=%d → keeper_id=%d address=%r",
|
||
pass_label,
|
||
row.loser_id,
|
||
row.keeper_id,
|
||
row.norm_address,
|
||
)
|
||
|
||
for label, sql in _STEPS:
|
||
res = db.execute(text(sql))
|
||
rowcount = res.rowcount or 0
|
||
if label == "listings":
|
||
result.listings_repointed += rowcount
|
||
elif label.endswith("(collision-delete)") or label.endswith("(dedup)"):
|
||
result.children_deleted += rowcount
|
||
elif label.endswith("(re-point)") or label in (
|
||
"house_placement_history",
|
||
"house_reviews",
|
||
"house_reliability_checks",
|
||
"external_valuations",
|
||
):
|
||
result.children_repointed += rowcount
|
||
logger.debug("merge_duplicate_houses: pass=%s step=%s rows=%d", pass_label, label, rowcount)
|
||
|
||
# Carry identity/enrichment onto the keeper BEFORE the losers vanish, then delete + backfill.
|
||
db.execute(_CARRY_OVER_IDENTITY_SQL)
|
||
db.execute(_DELETE_LOSERS_SQL)
|
||
db.execute(_BACKFILL_SOURCES_SQL)
|
||
db.execute(_BACKFILL_ALIASES_SQL)
|
||
|
||
|
||
def merge_duplicate_houses(db: Session, *, dry_run: bool = False) -> dict[str, int]:
|
||
"""Cluster houses by fias UUID, then by canonical address, merging dups onto one keeper.
|
||
|
||
Re-implements migration 108's proven collision-safe pipeline as a RECURRING TWO-PASS job:
|
||
1. FIAS pass — cluster by lower(NULLIF(house_fias_id, '')) (catches slash-collapse /
|
||
посёлок canon bugs the address canon misses).
|
||
2. CANON pass — cluster by canonical address (250 m geo guard + cross-fias anti-over-merge
|
||
guard).
|
||
Each pass: pick keeper → re-point children (UNIQUE-collision-safe) → carry identity onto the
|
||
keeper → delete losers → backfill sources/aliases. BOTH passes run in ONE transaction.
|
||
dry_run=True computes counts then ROLLS BACK (no writes). Idempotent: a clean table yields an
|
||
empty mapping in each pass → every statement is a 0-row no-op.
|
||
|
||
Returns the counter dict (DedupMergeResult.to_counters()).
|
||
"""
|
||
start = time.monotonic()
|
||
result = DedupMergeResult(dry_run=dry_run)
|
||
|
||
try:
|
||
# Pass 1: cluster by the ФИАС building UUID (runs first — most precise building identity).
|
||
_run_merge_pass(db, build_sql=_BUILD_MAPPING_SQL_FIAS, pass_label="fias", result=result)
|
||
# Pass 2: cluster by canonical address, with the cross-fias anti-over-merge guard.
|
||
_run_merge_pass(db, build_sql=_BUILD_MAPPING_SQL, pass_label="canon", result=result)
|
||
|
||
if result.losers_deleted == 0:
|
||
# Clean table — both passes empty. Roll back (we only opened temp tables).
|
||
db.rollback()
|
||
result.duration_sec = time.monotonic() - start
|
||
logger.info(
|
||
"merge_duplicate_houses: no duplicate clusters found (no-op) dry_run=%s",
|
||
dry_run,
|
||
)
|
||
return result.to_counters()
|
||
|
||
if dry_run:
|
||
db.rollback()
|
||
logger.info(
|
||
"merge_duplicate_houses: DRY-RUN computed clusters=%d losers=%d "
|
||
"listings_repointed=%d children_deleted=%d children_repointed=%d — ROLLED BACK",
|
||
result.clusters_merged,
|
||
result.losers_deleted,
|
||
result.listings_repointed,
|
||
result.children_deleted,
|
||
result.children_repointed,
|
||
)
|
||
else:
|
||
db.commit()
|
||
logger.info(
|
||
"merge_duplicate_houses: COMMITTED clusters=%d losers=%d "
|
||
"listings_repointed=%d children_deleted=%d children_repointed=%d",
|
||
result.clusters_merged,
|
||
result.losers_deleted,
|
||
result.listings_repointed,
|
||
result.children_deleted,
|
||
result.children_repointed,
|
||
)
|
||
except Exception:
|
||
logger.exception("merge_duplicate_houses: FAILED — rolling back")
|
||
try:
|
||
db.rollback()
|
||
except Exception:
|
||
logger.exception("merge_duplicate_houses: rollback also failed")
|
||
raise
|
||
|
||
result.duration_sec = time.monotonic() - start
|
||
return result.to_counters()
|
||
|
||
|
||
# ── Run lifecycle wrapper (scheduler entrypoint) ─────────────────────────────
|
||
def run_house_dedup_merge(db: Session, *, run_id: int, params: dict) -> dict[str, int]:
|
||
"""Run-lifecycle wrapper for the recurring house-dedup merge (sync, DB-only).
|
||
|
||
Launched by the in-app scheduler (source='house_dedup_merge') via
|
||
trigger_house_dedup_merge_run, or manually. Mirrors run_cadastral_geo_match: pure
|
||
internal DB op, finalises scrape_runs (mark_done / mark_failed) with counters.
|
||
|
||
Params (default_params jsonb):
|
||
dry_run: compute + return counts then ROLLBACK without writing (default false).
|
||
The deploy seeds the schedule DISABLED; the orchestrator can flip a single
|
||
manual run to dry_run=true to preview before enabling.
|
||
"""
|
||
from app.services import scrape_runs as runs_mod
|
||
|
||
dry_run = bool(params.get("dry_run", False))
|
||
counters: dict[str, int] = {
|
||
"clusters_merged": 0,
|
||
"losers_deleted": 0,
|
||
"listings_repointed": 0,
|
||
"children_deleted": 0,
|
||
"children_repointed": 0,
|
||
"dry_run": int(dry_run),
|
||
}
|
||
try:
|
||
runs_mod.update_heartbeat(db, run_id, counters)
|
||
counters = merge_duplicate_houses(db, dry_run=dry_run)
|
||
runs_mod.mark_done(db, run_id, counters)
|
||
logger.info(
|
||
"run_house_dedup_merge: run_id=%d DONE clusters=%d losers=%d dry_run=%s",
|
||
run_id,
|
||
counters.get("clusters_merged", 0),
|
||
counters.get("losers_deleted", 0),
|
||
dry_run,
|
||
)
|
||
return counters
|
||
except Exception as exc:
|
||
logger.exception("run_house_dedup_merge: run_id=%d FAILED", run_id)
|
||
try:
|
||
db.rollback()
|
||
except Exception:
|
||
pass
|
||
runs_mod.mark_failed(db, run_id, str(exc)[:1000], counters)
|
||
raise
|