diff --git a/tradein-mvp/backend/app/services/house_dedup_merge.py b/tradein-mvp/backend/app/services/house_dedup_merge.py index 0f244b97..511386a1 100644 --- a/tradein-mvp/backend/app/services/house_dedup_merge.py +++ b/tradein-mvp/backend/app/services/house_dedup_merge.py @@ -13,8 +13,10 @@ WHAT this is: pipeline, run inside ONE transaction so a crash leaves the table untouched. Cluster key: CANONICAL address via tradein_canon_addr() over the CLEAN address - COALESCE(short_address, full_address, address) (cadastral_number is 100% NULL on prod — - confirmed in migration 040 — so address is the real building key). The clean source matters: + 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 @@ -116,6 +118,43 @@ MERGE JOURNAL — the merge is REVERSIBLE (#2690, migration 230): 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. @@ -165,6 +204,14 @@ _COMPLETENESS_EXPR = """ # правилу. Последствие не косметическое: объявления проигравшего переезжают на запись, на которую # корпус никогда не ссылался, а 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, @@ -199,37 +246,17 @@ _CANON_KEY_EXPR = """ """ -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. +def _ranked_cte(cluster_key_case: str) -> str: + """Render the `WITH … ranked AS (…)` prelude: cluster → rank → expose the keeper per row. - 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). + 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. """ - 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 WITH clustered AS ( SELECT id, @@ -281,7 +308,41 @@ def _mapping_sql(cluster_key_case: str, *, apply_geo_guard: bool = True) -> str: 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 @@ -314,6 +375,54 @@ _BUILD_MAPPING_SQL = text(_mapping_sql(_CANON_KEY_EXPR)) # 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) ─────────────────────────── @@ -726,6 +835,15 @@ class DedupMergeResult: 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) @@ -736,6 +854,12 @@ class DedupMergeResult: "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), } @@ -853,6 +977,49 @@ def _run_merge_pass( 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, *, @@ -908,6 +1075,10 @@ def merge_duplicate_houses( 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() diff --git a/tradein-mvp/backend/tests/test_house_dedup_merge.py b/tradein-mvp/backend/tests/test_house_dedup_merge.py index 63db7822..4355e323 100644 --- a/tradein-mvp/backend/tests/test_house_dedup_merge.py +++ b/tradein-mvp/backend/tests/test_house_dedup_merge.py @@ -18,6 +18,7 @@ import inspect import os import re from pathlib import Path +from types import SimpleNamespace from typing import Any import pytest @@ -486,9 +487,13 @@ class _FakeDB: mapping_rows: list[_Row], step_rowcount: int = 1, fk_children: dict[str, str] | None = None, + residual: dict[str, int] | None = None, + residual_raises: bool = False, ): self._mapping_rows = mapping_rows self._step_rowcount = step_rowcount + self._residual = residual + self._residual_raises = residual_raises self._mapping_served = False # The catalog the FK-child guard reads; defaults to the real live set. self._fk_children = _FK_CHILDREN if fk_children is None else fk_children @@ -503,6 +508,12 @@ class _FakeDB: return _FakeResult() if "FROM pg_constraint" in sql: return _FakeResult(rows=[_FKChild(t, c) for t, c in self._fk_children.items()]) + if "AS residual_rows" in sql: # residual census (#2690) — read-only, after both passes + if self._residual_raises: + raise RuntimeError("census exploded") + if self._residual is None: + return _FakeResult(rows=[]) + return _FakeResult(rows=[SimpleNamespace(**self._residual)]) if "SELECT loser_id, keeper_id, norm_address" in sql: # The service now runs TWO passes (fias, then canon). Model «fias pass found the # duplicates, canon pass is clean»: serve the scripted mapping once, empty afterwards. @@ -550,6 +561,85 @@ def test_dry_run_computes_counts_but_rolls_back() -> None: assert db.rollbacks == 1 +# ── Residual census (#2690 п.2/п.4): остаток фиксируется числом, а не оценкой ── + + +def test_residual_census_reuses_the_canon_mapping_prelude() -> None: + """Перепись остатка обязана считать РОВНО те строки, о которых рассуждает канон-проход. + + Общий `_ranked_cte` — не косметика: собственная копия кластеризации разъехалась бы с + проходом, который она описывает, и разъезд был бы невидим (это тот же класс ошибки, что и + ключ в #2690 — два выражения, похожие друг на друга и не равные). RED до фикса: переписи + нет вовсе. + """ + residual = _flat(str(hdm._RESIDUAL_SQL.text)) + prelude = _flat(hdm._ranked_cte(hdm._CANON_KEY_EXPR)) + assert prelude in residual + assert prelude in _flat(hdm._mapping_sql(hdm._CANON_KEY_EXPR)) + + +def test_residual_census_keeps_silent_guard_apart_from_rejecting_guard() -> None: + """«Координат нет» и «дальше 250 м» — противоположные факты, в одну сумму их нельзя. + + Первое означает, что страж не смог высказаться (остаток неизвестен), второе — что он + высказался по существу и дома РАЗНЫЕ (это вообще не дубли). Сумма из этих двух читается как + «долг», которого нет. Плюс: перепись НЕ применяет гео-фильтр — иначе она считала бы только + то, что и так слилось. + """ + residual = _flat(str(hdm._RESIDUAL_SQL.text)) + for bucket in ( + "AS residual_rows", + "AS residual_listings", + "AS residual_no_geom", + "AS residual_far", + "AS residual_cross_fias", + "AS residual_mergeable", + ): + assert bucket in residual, bucket + # Гео-страж 250 м здесь — РАЗДЕЛИТЕЛЬ корзин, а не фильтр строк. + assert "AND ST_DistanceSphere(loser_geom, keeper_geom) <= 250" not in residual + assert set(hdm.DedupMergeResult().to_counters()) >= { + "residual_rows", + "residual_listings", + "residual_no_geom", + "residual_far", + "residual_cross_fias", + "residual_mergeable", + } + + +def test_residual_counters_reported_even_when_nothing_merged() -> None: + """Прогон, который не слил ничего, — ровно тот, чьему остатку нужно число.""" + db = _FakeDB( + mapping_rows=[], + residual={ + "residual_rows": 963, + "residual_listings": 1765, + "residual_no_geom": 326, + "residual_far": 568, + "residual_cross_fias": 8, + "residual_mergeable": 61, + }, + ) + out = hdm.merge_duplicate_houses(db, dry_run=False) # type: ignore[arg-type] + assert out["losers_deleted"] == 0 + assert out["residual_rows"] == 963 + assert out["residual_listings"] == 1765 + assert out["residual_no_geom"] == 326 + assert out["residual_far"] == 568 + assert out["residual_cross_fias"] == 8 + assert out["residual_mergeable"] == 61 + + +def test_residual_census_failure_never_breaks_the_merge() -> None: + """Перепись — приборы, слияние — продукт: упавший счётчик не отменяет коммит.""" + db = _FakeDB(mapping_rows=[_Row(2, 1, "ул. мира, 10")], residual_raises=True) + out = hdm.merge_duplicate_houses(db, dry_run=False) # type: ignore[arg-type] + assert out["losers_deleted"] == 1 + assert out["residual_rows"] == 0 # неизвестно — и это видно, а не выдумано + assert db.commits == 1 + + def test_real_merge_commits() -> None: """dry_run=False with dups → exactly one commit, no rollback.""" rows = [_Row(2, 1, "ул. мира, 10")]