bot-backend
0ed0140c9e
All checks were successful
Deploy Trade-In / changes (push) Successful in 13s
Deploy Trade-In / build-frontend (push) Has been skipped
Deploy Trade-In / build-browser (push) Has been skipped
Deploy Trade-In / test (push) Successful in 3m11s
Deploy Trade-In / build-backend (push) Successful in 1m4s
Deploy Trade-In / deploy (push) Successful in 1m57s
1172 lines
56 KiB
Python
1172 lines
56 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) — the address is the only building key we
|
||
have (why: the KEY section below; the older claim here, «cadastral_number is 100% NULL on
|
||
prod», is no longer true — 2 648 of 9 179 rows carry one — and the conclusion no longer
|
||
rests on it). 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, CANON PASS ONLY — #2187): 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). The
|
||
FIAS pass deliberately SKIPS this guard: a shared ФИАС/ГАР UUID IS the building identity and
|
||
strictly outranks geo-proximity, so same-fias rows merge even with NULL geom on a side or
|
||
>250 m apart (the geom-first keeper rule simultaneously repairs the broken coordinate).
|
||
|
||
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 the
|
||
CANON pass ONLY (#2187): the FIAS pass merges on UUID identity regardless of geom — a
|
||
боевой прогон left 130 same-fias groups (335 houses, 1 529 listings) split because the
|
||
guard blocked them (NULL geom on a side, or >250 m from a broken coordinate).
|
||
|
||
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.
|
||
|
||
MERGE JOURNAL — the merge is REVERSIBLE (#2690, migration 230):
|
||
Every loser gets a row in `house_merge_log` written in the SAME transaction as the merge:
|
||
the full jsonb snapshot of the deleted row, the keeper's snapshot BEFORE the identity
|
||
carry-over, the ids of every child row whose FK moved, the full snapshots of every child row
|
||
a UNIQUE collision destroyed, plus the grounds — which pass, which cluster-key VALUE fired,
|
||
whether the geo guard was on, and the keeper↔loser distance in metres.
|
||
|
||
This exists because the merge used to leave no restorable trace: losers were hard-deleted
|
||
with their children and the only record of «what went into what» was a log line, in a
|
||
container whose logs rotate faster than a day. A day after a run nobody could even NAME the
|
||
pairs, and the only rollback was restoring the whole database.
|
||
|
||
Undo: `SELECT * FROM house_merge_undo(batch_id)` inside a transaction — restores the loser
|
||
rows, points the children back, re-inserts the destroyed children, and un-does the identity
|
||
carry-over on the keeper, reporting per record what it could and could not restore.
|
||
|
||
NOTE the journal is deliberately NEUTRAL to the merge rule: it changes no cluster key, no
|
||
keeper rule and no guard. It only makes whatever the pass decides reversible — which is the
|
||
precondition for revisiting those decisions at all (#2690, #1772).
|
||
|
||
distance_m is recorded on BOTH passes, including the fias pass whose geo guard is off. That
|
||
asymmetry — merge allowed without a proximity check — was invisible in data before; now
|
||
«how many merges happened beyond N metres, on which key» is one query.
|
||
|
||
KEY — there is no second, address-independent observation. Measured on prod 2026-08-10 (#2690):
|
||
#2690 asked for a cluster key that does not come from the normalized address, so that two
|
||
rows merge on two independent statements of identity rather than one restated twice. Every
|
||
field `houses` carries was checked against the live table. None qualifies:
|
||
|
||
cadastral_number 2 648 filled, ALL 2 648 values DISTINCT → collapses nothing. Provenance:
|
||
all 2 648 also carry dadata_enriched_at and house_fias_id, i.e. they are
|
||
DaData's answer to our address string, not a second observation of the
|
||
building. (The other cadastre we hold, listings.building_cadastral_number,
|
||
is the KNN geo-nearest hint — 20.1% of its values cover >1 ГАР building;
|
||
#2674 refused it as an identity key and that stands.)
|
||
house_fias_id 3 678 filled, ALL DISTINCT → the FIAS pass merges 0 rows today. Same
|
||
DaData provenance.
|
||
gar_house_guid the key #2690 rejected, re-measured: of 458 same-guid pairs, 441 share
|
||
the canon (the guid restates it), 17 do not — and 5 of those 17 are
|
||
>250 m apart, worst 5 064 km. Still circular, still noisy.
|
||
zhkh_house_guid looks independent (ГИС ЖКХ is an external registry) and is not: the
|
||
loader sets it WHERE gar_house_guid = <guid>, i.e. it IS the ГАР guid for
|
||
4 268 of 4 663 rows. The 395 that differ come from the cadastre fallback
|
||
— keyed by that same KNN hint. Of its 194 pairs with a DIFFERENT canon,
|
||
193 come through the fallback, and 30 of the 31 pairs >250 m apart do too.
|
||
source+ext_house_id, cian_internal_house_id, yandex_jk_id
|
||
distinct by construction / 39 / 0 rows → nothing to cluster.
|
||
coordinates a real independent observation, but not an IDENTITY: neighbours share a
|
||
yard. It is already used the only way it can be — as the guard.
|
||
year_built+total_floors
|
||
a FALSE witness, not a corroborator: of the 391 same-canon pairs the
|
||
guard cannot judge, only 18 agree on both fields (357 have a NULL), while
|
||
306 pairs the guard rejected at >250 m DO agree — it would confirm merges
|
||
that are provably wrong.
|
||
|
||
Conclusion: do NOT strengthen the key, and do not read the leftover as a backlog. What the
|
||
canon key + 250 m guard reach IS the ceiling; what is left is counted, not queued — see the
|
||
residual census (`_RESIDUAL_SQL`), whose buckets keep «the guard was silent» apart from «the
|
||
guard rejected on the merits». Prod 2026-08-10, 963 excess rows: 568 of them are >250 m apart
|
||
(median 1 084 m) — those are not duplicates at all, the canon key is wrong about them.
|
||
|
||
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 json
|
||
import logging
|
||
import time
|
||
import uuid
|
||
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.
|
||
#
|
||
# NULLS LAST на listing_cnt (#2674): счётчик приходит из LEFT JOIN listing_counts, поэтому у дома
|
||
# БЕЗ объявлений он NULL, а `DESC` в Postgres по умолчанию NULLS FIRST — то есть строка с нулём
|
||
# объявлений обгоняла строку со 192 и забирала роль keeper'а, ровно наоборот задокументированному
|
||
# правилу. Последствие не косметическое: объявления проигравшего переезжают на запись, на которую
|
||
# корпус никогда не ссылался, а COALESCE-перенос полей неполон (год постройки / тип дома /
|
||
# этажность / застройщик не переносятся) — данные богатого проигравшего удаляются безвозвратно.
|
||
#
|
||
# ПРОВЕРЕНО ЗАДНИМ ЧИСЛОМ (#2690 п.3, 2026-08-10): первый прогон на исправленном правиле —
|
||
# 08.08, 821 слияние — разобран по house_merge_log (у проигравшего число объявлений = длина
|
||
# children_repointed['listings.house_id_fk'], у победителя — что висело на нём до слияния).
|
||
# Слияний, где победитель беднее проигравшего по объявлениям: 0 из 821. Контрфактика старого
|
||
# правила на тех же кластерах: 6 из 762 забрали бы пустого победителя (8 объявлений). Мерить
|
||
# «победителя до слияния» по listings.scraped_at НЕЛЬЗЯ — #2206 двигает его при каждом
|
||
# ре-подтверждении, отчего появляются 207 несуществующих «худших победителей».
|
||
_KEEPER_ORDER = f"""
|
||
(h.geom IS NOT NULL) DESC,
|
||
listing_cnt DESC NULLS LAST,
|
||
({_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 _ranked_cte(cluster_key_case: str) -> str:
|
||
"""Render the `WITH … ranked AS (…)` prelude: cluster → rank → expose the keeper per row.
|
||
|
||
Shared verbatim by the merge mapping (`_mapping_sql`) and the residual census
|
||
(`_RESIDUAL_SQL`) so the census counts EXACTLY the rows the merge reasons about — a census
|
||
built from its own copy of the clustering would drift from the pass it describes and the
|
||
drift would be invisible (it is the same class of error as #2690's cluster key: two
|
||
expressions that look alike and are not).
|
||
`cluster_key_case` is a STATIC module constant (never runtime data) — no value injection.
|
||
"""
|
||
return f"""
|
||
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
|
||
)"""
|
||
|
||
|
||
def _mapping_sql(cluster_key_case: str, *, apply_geo_guard: bool = True) -> 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 CROSS-FIAS guard always applies (a no-op for the fias pass, where every clustered
|
||
row shares one fias by construction).
|
||
|
||
apply_geo_guard (#2187): the 250 m ST_DistanceSphere guard is emitted ONLY when True.
|
||
- CANON pass → True: the canon strips город/район, so same-street-number buildings in
|
||
different region-66 towns share a canon; the guard stops the cross-town over-merge.
|
||
- FIAS pass → False: a shared ФИАС/ГАР UUID IS the building identity and strictly outranks
|
||
proximity, so same-fias rows merge even with NULL geom on a side or >250 m apart (the
|
||
geom-first keeper rule simultaneously repairs the broken coordinate).
|
||
`cluster_key_case` is a STATIC module constant (never runtime data) — no value injection.
|
||
"""
|
||
geo_guard = (
|
||
"""
|
||
-- GEO GUARD (canon pass only — #2187). tradein_canon_addr strips город/район, so two
|
||
-- different buildings sharing a street+number canon («Ленина 5» in different region-66
|
||
-- towns) collapse to one cluster_key. A loser merges only when geographically next to the
|
||
-- keeper (<=250 m — covers one building's geocode spread, prod: Мраморская 34к4 dupes at
|
||
-- 222 m; region-66 towns are km+ apart → 250 m is safe from cross-town). >250 m, or NULL
|
||
-- geom on either side, → left as separate rows (conservative — never over-merges).
|
||
AND keeper_geom IS NOT NULL
|
||
AND loser_geom IS NOT NULL
|
||
AND ST_DistanceSphere(loser_geom, keeper_geom) <= 250"""
|
||
if apply_geo_guard
|
||
else ""
|
||
)
|
||
return f"""
|
||
CREATE TEMP TABLE _1772_dup_mapping ON COMMIT DROP AS
|
||
{_ranked_cte(cluster_key_case)}
|
||
-- CROSS-FIAS guard (#1772 follow-up): never merge two rows that BOTH carry a non-null but
|
||
-- DIFFERENT house_fias_id — provably different buildings the cluster key collapsed (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 carries a fias.
|
||
--
|
||
-- cluster_key / distance_m are carried out of the mapping for the MERGE JOURNAL (#2690):
|
||
-- cluster_key records WHICH key value fired, distance_m how far apart the two rows were.
|
||
-- distance_m is computed even when the geo guard is OFF for this pass — that is precisely
|
||
-- the case where nothing else records the distance, and #2690 had no way to ask
|
||
-- «how many merges happened at distances the guard would have blocked» from data.
|
||
SELECT id AS loser_id, keeper_id, norm_address, cluster_key,
|
||
CASE WHEN keeper_geom IS NOT NULL AND loser_geom IS NOT NULL
|
||
THEN ST_DistanceSphere(loser_geom, keeper_geom)
|
||
END AS distance_m
|
||
FROM ranked
|
||
WHERE rn > 1
|
||
AND id <> keeper_id{geo_guard}
|
||
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 — geo guard ON (cross-town over-merge protection for street+number canons).
|
||
_BUILD_MAPPING_SQL = text(_mapping_sql(_CANON_KEY_EXPR))
|
||
# Fias-pass mapping — same pipeline, clustered by the ФИАС building UUID (runs first). Geo guard
|
||
# OFF (#2187): a shared ГАР UUID IS the building identity and outranks proximity — same-fias rows
|
||
# merge even with NULL geom or >250 m apart (the geom-first keeper rule fixes broken coords).
|
||
_BUILD_MAPPING_SQL_FIAS = text(_mapping_sql(_FIAS_KEY_EXPR, apply_geo_guard=False))
|
||
|
||
# ── RESIDUAL CENSUS (#2690 п.2/п.4) ───────────────────────────────────────────
|
||
#
|
||
# Read-only, run AFTER both passes: how many same-canon rows the merge LEFT BEHIND, and WHY.
|
||
# Same `ranked` prelude as the canon mapping, minus the guard — so every row the guard filtered
|
||
# out is counted here, bucketed by the reason it survived.
|
||
#
|
||
# WHY this exists. #2690 asked for a second, address-independent key; measured 2026-08-10, there
|
||
# is none (see the KEY section in the module docstring), so the remainder is a CEILING, not a
|
||
# backlog — and a ceiling has to be a live number, not a one-off. The one-off rots fast: the
|
||
# issue's own census (781 excess rows, 06.08) was 963 four days later, after a run deleted 821.
|
||
#
|
||
# The buckets are deliberately NOT summed into one «остаток». «Guard was silent» and «guard
|
||
# rejected» are opposite facts:
|
||
# residual_no_geom — one side has no coordinates: the guard could not speak. UNKNOWN.
|
||
# residual_far — both geocoded, >250 m apart: the guard spoke on the merits. These are
|
||
# NOT duplicates — the canon key is wrong about them (prod 2026-08-10:
|
||
# 568 rows, median 1084 m). Counting them as «дубли» inflates the debt.
|
||
# residual_cross_fias — provably different buildings (two different ФИАС UUIDs).
|
||
# residual_mergeable — passes every guard and STILL was not merged. Must be 0 after a real
|
||
# run; non-zero is a tripwire on the pass itself, not a census entry.
|
||
# residual_listings is the user-visible size of the remainder (listings hanging on those rows).
|
||
_RESIDUAL_SQL = text(
|
||
f"""
|
||
{_ranked_cte(_CANON_KEY_EXPR)}
|
||
SELECT
|
||
count(*) FILTER (WHERE rn > 1) AS residual_rows,
|
||
COALESCE(sum(lcnt) FILTER (WHERE rn > 1), 0) AS residual_listings,
|
||
count(*) FILTER (WHERE rn > 1 AND cross_fias) AS residual_cross_fias,
|
||
count(*) FILTER (WHERE rn > 1 AND NOT cross_fias AND dist IS NULL)
|
||
AS residual_no_geom,
|
||
count(*) FILTER (WHERE rn > 1 AND NOT cross_fias AND dist > 250) AS residual_far,
|
||
count(*) FILTER (WHERE rn > 1 AND NOT cross_fias AND dist <= 250)
|
||
AS residual_mergeable
|
||
FROM (
|
||
SELECT rn,
|
||
COALESCE(lc.listing_cnt, 0) AS lcnt,
|
||
CASE WHEN keeper_geom IS NOT NULL AND loser_geom IS NOT NULL
|
||
THEN ST_DistanceSphere(loser_geom, keeper_geom)
|
||
END AS dist,
|
||
(NULLIF(loser_fias, '') IS NOT NULL
|
||
AND NULLIF(keeper_fias, '') IS NOT NULL
|
||
AND lower(loser_fias) <> lower(keeper_fias)) AS cross_fias
|
||
FROM ranked
|
||
LEFT JOIN listing_counts lc ON lc.house_id = ranked.id
|
||
) r
|
||
"""
|
||
)
|
||
|
||
# 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
|
||
RETURNING m.loser_id, l.id AS child_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
|
||
RETURNING m.loser_id, hph.id AS child_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
|
||
RETURNING m.loser_id, hr.id AS child_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
|
||
RETURNING m.loser_id, hrc.id AS child_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
|
||
RETURNING m.loser_id, ev.id AS child_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
|
||
)
|
||
RETURNING hs.house_id AS loser_id, to_jsonb(hs.*) AS row_snapshot
|
||
""",
|
||
),
|
||
(
|
||
"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
|
||
RETURNING m.loser_id, hs.id AS child_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
|
||
)
|
||
RETURNING haa.house_id AS loser_id, to_jsonb(haa.*) AS row_snapshot
|
||
""",
|
||
),
|
||
(
|
||
"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
|
||
RETURNING m.loser_id, haa.id AS child_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
|
||
RETURNING t.house_id AS loser_id, to_jsonb(t.*) AS row_snapshot
|
||
""",
|
||
),
|
||
(
|
||
"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
|
||
RETURNING m.loser_id, hpd.id AS child_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
|
||
RETURNING t.house_id AS loser_id, to_jsonb(t.*) AS row_snapshot
|
||
""",
|
||
),
|
||
(
|
||
"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
|
||
RETURNING m.loser_id, hie.id AS child_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
|
||
RETURNING t.house_id AS loser_id, to_jsonb(t.*) AS row_snapshot
|
||
""",
|
||
),
|
||
(
|
||
"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
|
||
RETURNING m.loser_id, hs.id AS child_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
|
||
RETURNING t.house_id AS loser_id, to_jsonb(t.*) AS row_snapshot
|
||
""",
|
||
),
|
||
(
|
||
"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
|
||
RETURNING m.loser_id, ama.id AS child_id
|
||
""",
|
||
),
|
||
]
|
||
|
||
# ── MERGE JOURNAL (#2690) ─────────────────────────────────────────────────────
|
||
#
|
||
# Every child of houses(id) except `listings` references it through a column named house_id;
|
||
# listings uses house_id_fk. The undo function reads the column name back out of the journal
|
||
# key ("таблица.колонка"), so this mapping is what makes the reverse UPDATE possible.
|
||
_FK_COLUMN = {"listings": "house_id_fk"}
|
||
|
||
# The (table, column) pairs the _STEPS pipeline actually handles, derived FROM the steps so the
|
||
# set cannot drift away from them. Compared against pg_catalog before every merge — see
|
||
# _assert_all_fk_children_handled.
|
||
_HANDLED_CHILDREN: frozenset[tuple[str, str]] = frozenset(
|
||
(tbl, _FK_COLUMN.get(tbl, "house_id")) for tbl in {label.split("(")[0] for label, _ in _STEPS}
|
||
)
|
||
|
||
# Live FK children of houses(id), read from the catalog rather than trusted from a comment.
|
||
_FK_CHILDREN_SQL = text(
|
||
"""
|
||
SELECT CAST(CAST(c.conrelid AS regclass) AS text) AS child_table,
|
||
a.attname AS fk_column
|
||
FROM pg_constraint c
|
||
JOIN unnest(c.conkey) AS k(attnum) ON true
|
||
JOIN pg_attribute a ON a.attrelid = c.conrelid AND a.attnum = k.attnum
|
||
WHERE c.confrelid = CAST('houses' AS regclass)
|
||
AND c.contype = 'f'
|
||
"""
|
||
)
|
||
|
||
# One journal row per loser, written from the mapping BEFORE anything is mutated — so loser_row
|
||
# is the row as it stood, and keeper_before precedes the identity carry-over.
|
||
_JOURNAL_INSERT_SQL = text(
|
||
"""
|
||
INSERT INTO house_merge_log (
|
||
batch_id, run_id, initiator, merge_pass, cluster_key, geo_guard, distance_m,
|
||
norm_address, loser_id, keeper_id, loser_row, keeper_before
|
||
)
|
||
SELECT
|
||
CAST(:batch_id AS uuid),
|
||
CAST(:run_id AS bigint),
|
||
CAST(:initiator AS text),
|
||
CAST(:merge_pass AS text),
|
||
m.cluster_key,
|
||
CAST(:geo_guard AS boolean),
|
||
m.distance_m,
|
||
m.norm_address,
|
||
m.loser_id,
|
||
m.keeper_id,
|
||
to_jsonb(l.*),
|
||
to_jsonb(k.*)
|
||
FROM _1772_dup_mapping m
|
||
JOIN houses l ON l.id = m.loser_id
|
||
JOIN houses k ON k.id = m.keeper_id
|
||
"""
|
||
)
|
||
|
||
# Child bookkeeping lands after the steps ran — only then is it known which rows moved and which
|
||
# were destroyed by a UNIQUE collision.
|
||
_JOURNAL_CHILDREN_SQL = text(
|
||
"""
|
||
UPDATE house_merge_log
|
||
SET children_repointed = CAST(:children_repointed AS jsonb),
|
||
children_deleted = CAST(:children_deleted AS jsonb)
|
||
WHERE batch_id = CAST(:batch_id AS uuid)
|
||
AND loser_id = CAST(:loser_id AS bigint)
|
||
"""
|
||
)
|
||
|
||
|
||
def _assert_all_fk_children_handled(db: Session) -> None:
|
||
"""Fail the merge if houses(id) gained an FK child the _STEPS pipeline does not handle.
|
||
|
||
This is what makes the journal's promise true rather than merely documented. An unhandled
|
||
child is not a cosmetic gap: 9 of the 11 FKs are ON DELETE CASCADE, so `DELETE FROM houses`
|
||
would destroy its rows silently — no re-point step touches them, no RETURNING records them,
|
||
and the journal would claim a complete snapshot it does not have. Migration 133 already
|
||
broke on prod for exactly this (a missed child); there the failure was loud. Here it would
|
||
be silent, which is worse. Aborting the transaction costs one skipped merge cycle.
|
||
"""
|
||
live = {(r.child_table, r.fk_column) for r in db.execute(_FK_CHILDREN_SQL).all()}
|
||
unhandled = live - _HANDLED_CHILDREN
|
||
if unhandled:
|
||
raise RuntimeError(
|
||
"merge_duplicate_houses: houses(id) has FK children the merge does not handle: "
|
||
f"{sorted(unhandled)}. Their rows would be CASCADE-deleted without a journal entry. "
|
||
"Add a re-point step to _STEPS (and its RETURNING) before merging again."
|
||
)
|
||
|
||
|
||
# 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
|
||
# Residual census (#2690): same-canon rows STILL in the table after this run, by reason.
|
||
# Not a backlog — measured 2026-08-10 there is no address-independent key to shrink it with,
|
||
# so this is the ceiling of what this pass can reach. See _RESIDUAL_SQL.
|
||
residual_rows: int = 0 # excess same-canon rows left behind (sum of the three buckets)
|
||
residual_listings: int = 0 # listings hanging on them (the user-visible size)
|
||
residual_no_geom: int = 0 # guard was SILENT — one side has no coordinates
|
||
residual_far: int = 0 # guard SPOKE — >250 m apart, i.e. not the same building
|
||
residual_cross_fias: int = 0 # two different ФИАС UUIDs — provably different buildings
|
||
residual_mergeable: int = 0 # passed every guard and still unmerged — TRIPWIRE, expect 0
|
||
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,
|
||
"residual_rows": self.residual_rows,
|
||
"residual_listings": self.residual_listings,
|
||
"residual_no_geom": self.residual_no_geom,
|
||
"residual_far": self.residual_far,
|
||
"residual_cross_fias": self.residual_cross_fias,
|
||
"residual_mergeable": self.residual_mergeable,
|
||
"dry_run": int(self.dry_run),
|
||
"duration_sec": int(self.duration_sec),
|
||
}
|
||
|
||
|
||
def _run_merge_pass(
|
||
db: Session,
|
||
*,
|
||
build_sql: Any,
|
||
pass_label: str,
|
||
geo_guard: bool,
|
||
batch_id: str,
|
||
run_id: int | None,
|
||
initiator: 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, writes the MERGE JOURNAL
|
||
(#2690), 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, which is also what makes the journal atomic with the merge: there is no ordering in
|
||
which the rows vanish but the journal entry does not land (and dry_run rolls back both).
|
||
"""
|
||
# 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, cluster_key, distance_m "
|
||
"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})
|
||
|
||
# JOURNAL, phase 1 — snapshot loser + keeper BEFORE any statement mutates them.
|
||
db.execute(
|
||
_JOURNAL_INSERT_SQL,
|
||
{
|
||
"batch_id": batch_id,
|
||
"run_id": run_id,
|
||
"initiator": initiator,
|
||
"merge_pass": pass_label,
|
||
"geo_guard": geo_guard,
|
||
},
|
||
)
|
||
|
||
# Container logs rotate faster than a day (#2690), so this line is a convenience, not the
|
||
# record — house_merge_log is. Distance is logged too: it is the one number that says
|
||
# whether a merge would have survived the geo guard.
|
||
for row in mapping:
|
||
logger.info(
|
||
"merge_duplicate_houses: pass=%s merge loser_id=%d → keeper_id=%d address=%r "
|
||
"distance_m=%s batch=%s",
|
||
pass_label,
|
||
row.loser_id,
|
||
row.keeper_id,
|
||
row.norm_address,
|
||
"n/a" if row.distance_m is None else f"{row.distance_m:.0f}",
|
||
batch_id,
|
||
)
|
||
|
||
# Per-loser child bookkeeping, collected from each step's RETURNING: survivors by id (the
|
||
# rows are intact, only their FK moved), destroyed rows by full snapshot (nothing else is
|
||
# left of them).
|
||
repointed: dict[int, dict[str, list[int]]] = {}
|
||
deleted: dict[int, dict[str, list[Any]]] = {}
|
||
|
||
for label, sql in _STEPS:
|
||
rows = db.execute(text(sql)).all()
|
||
rowcount = len(rows)
|
||
table = label.split("(")[0]
|
||
if label.endswith("(collision-delete)") or label.endswith("(dedup)"):
|
||
result.children_deleted += rowcount
|
||
for r in rows:
|
||
deleted.setdefault(r.loser_id, {}).setdefault(table, []).append(r.row_snapshot)
|
||
else:
|
||
key = f"{table}.{_FK_COLUMN.get(table, 'house_id')}"
|
||
for r in rows:
|
||
repointed.setdefault(r.loser_id, {}).setdefault(key, []).append(r.child_id)
|
||
if label == "listings":
|
||
result.listings_repointed += rowcount
|
||
else:
|
||
result.children_repointed += rowcount
|
||
logger.debug("merge_duplicate_houses: pass=%s step=%s rows=%d", pass_label, label, rowcount)
|
||
|
||
# JOURNAL, phase 2 — attach the child bookkeeping to the rows written in phase 1.
|
||
touched = sorted(set(repointed) | set(deleted))
|
||
if touched:
|
||
db.execute(
|
||
_JOURNAL_CHILDREN_SQL,
|
||
[
|
||
{
|
||
"batch_id": batch_id,
|
||
"loser_id": loser_id,
|
||
"children_repointed": json.dumps(repointed.get(loser_id, {})),
|
||
"children_deleted": json.dumps(deleted.get(loser_id, {}), default=str),
|
||
}
|
||
for loser_id in touched
|
||
],
|
||
)
|
||
|
||
# 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 _measure_residual(db: Session, result: DedupMergeResult) -> None:
|
||
"""Count the same-canon rows this run did NOT merge, bucketed by the reason (#2690).
|
||
|
||
Read-only; runs after both passes, so it describes the table as the run leaves it (under
|
||
dry_run it sees the not-yet-rolled-back state, which is the correct preview). Kept out of
|
||
`_run_merge_pass` because the census is about the CANON key only and must be taken once per
|
||
call, not once per pass.
|
||
|
||
Never fails the merge: the merge itself is the product, the census is instrumentation, and a
|
||
census that can abort a committed-by-now transaction would be worse than a missing number.
|
||
"""
|
||
try:
|
||
rows = db.execute(_RESIDUAL_SQL).all()
|
||
except Exception:
|
||
logger.exception("merge_duplicate_houses: residual census failed — counters left at 0")
|
||
return
|
||
if not rows:
|
||
return
|
||
r = rows[0]
|
||
result.residual_rows = int(r.residual_rows or 0)
|
||
result.residual_listings = int(r.residual_listings or 0)
|
||
result.residual_no_geom = int(r.residual_no_geom or 0)
|
||
result.residual_far = int(r.residual_far or 0)
|
||
result.residual_cross_fias = int(r.residual_cross_fias or 0)
|
||
result.residual_mergeable = int(r.residual_mergeable or 0)
|
||
logger.info(
|
||
"merge_duplicate_houses: residual rows=%d listings=%d "
|
||
"(страж молчит=%d · страж отверг >250м=%d · cross-fias=%d · сливаемых=%d)",
|
||
result.residual_rows,
|
||
result.residual_listings,
|
||
result.residual_no_geom,
|
||
result.residual_far,
|
||
result.residual_cross_fias,
|
||
result.residual_mergeable,
|
||
)
|
||
if result.residual_mergeable:
|
||
logger.warning(
|
||
"merge_duplicate_houses: %d rows pass every guard yet were NOT merged — the pass "
|
||
"left work on the table (expected 0)",
|
||
result.residual_mergeable,
|
||
)
|
||
|
||
|
||
def merge_duplicate_houses(
|
||
db: Session,
|
||
*,
|
||
dry_run: bool = False,
|
||
run_id: int | None = None,
|
||
initiator: str = "manual",
|
||
) -> 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.
|
||
|
||
Every deleted row is journaled to house_merge_log in the SAME transaction (#2690), so a
|
||
merge is reversible via house_merge_undo(batch_id); the batch_id is returned in the log line
|
||
and stored on every journal row of this call.
|
||
|
||
Returns the counter dict (DedupMergeResult.to_counters()).
|
||
"""
|
||
start = time.monotonic()
|
||
result = DedupMergeResult(dry_run=dry_run)
|
||
batch_id = str(uuid.uuid4())
|
||
|
||
try:
|
||
# Refuse to merge at all if some FK child would be CASCADE-destroyed unjournaled.
|
||
_assert_all_fk_children_handled(db)
|
||
# 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",
|
||
geo_guard=False,
|
||
batch_id=batch_id,
|
||
run_id=run_id,
|
||
initiator=initiator,
|
||
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",
|
||
geo_guard=True,
|
||
batch_id=batch_id,
|
||
run_id=run_id,
|
||
initiator=initiator,
|
||
result=result,
|
||
)
|
||
|
||
# Census of what is LEFT (read-only). Runs before the no-op early return on purpose:
|
||
# a run that merged nothing is exactly the run whose remainder needs a number.
|
||
_measure_residual(db, 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 "
|
||
"batch_id=%s (undo: SELECT * FROM house_merge_undo('%s'))",
|
||
result.clusters_merged,
|
||
result.losers_deleted,
|
||
result.listings_repointed,
|
||
result.children_deleted,
|
||
result.children_repointed,
|
||
batch_id,
|
||
batch_id,
|
||
)
|
||
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 kit scheduler (source='house_dedup_merge') via
|
||
product_handlers._job_house_dedup_merge, 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, run_id=run_id, initiator="schedule")
|
||
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
|