"""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 = , 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 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