diff --git a/tradein-mvp/backend/app/services/house_dedup_merge.py b/tradein-mvp/backend/app/services/house_dedup_merge.py index dd90c9d1..e8959900 100644 --- a/tradein-mvp/backend/app/services/house_dedup_merge.py +++ b/tradein-mvp/backend/app/services/house_dedup_merge.py @@ -29,6 +29,26 @@ WHAT this is: 250 m of the keeper (ST_DistanceSphere on the WGS84 geom). Same-canon houses >250 m away — or with NULL geom on either side — are left as separate rows (conservative; never over-merges). +TWO PASSES (2026-07-02 follow-up): the SAME cluster→keeper→re-point→carry-identity→delete + pipeline now runs TWICE inside one transaction, parametrised by the cluster-key expression + (`_mapping_sql`) — no 300-line copy-paste: + 1. FIAS pass — clusters by lower(NULLIF(house_fias_id, '')) (the ФИАС/ГАР building UUID, + DaData-backfilled 2026-07-02 for 5 122 houses). Only non-null fias shared by >1 house_id + cluster. Catches duplicates the canon MISSES: slash-collapse («Сулимова, 32» vs + «Сулимова, 3/2» → same canon «сулимова32») and посёлок truncation (Кедровка/Шувакиш + colliding with a same-named ЕКБ street). + 2. CANON pass — the canonical-address clustering above, now with a CROSS-FIAS GUARD: within a + canon cluster a loser is NOT merged when it AND the keeper both carry a non-null but + DIFFERENT house_fias_id — provably different buildings the canon collapsed (the + slash-collapse class). Critical anti-over-merge fix. The 250 m geo guard applies to both + passes (after the backfill, fias rows almost always have geom → conservative). + +IDENTITY CARRY-OVER (each pass, BEFORE deleting losers): the keeper's NULL identity / geo-QC + fields are filled from its losers with COALESCE semantics (keeper value wins; donor = the + loser with geom, then min id): house_fias_id, cadastral_number, gar_house_guid, gar_flat_count, + gar_matched_at, gar_match_method, dadata_qc_geo, dadata_qc_house, dadata_enriched_at. Without + this the merge would silently drop DaData/ГАР enrichment that lived only on a loser row. + KEEPER RULE (one canonical row per cluster), in priority order: 1. geom NOT NULL first (a geocoded row is the better survivor) 2. most linked listings (the row the corpus already points at) @@ -83,6 +103,7 @@ from __future__ import annotations import logging import time from dataclasses import dataclass, field +from typing import Any from sqlalchemy import text from sqlalchemy.orm import Session @@ -113,21 +134,47 @@ _KEEPER_ORDER = f""" h.id ASC """ -# Build the loser→keeper mapping as a TEMP table for this transaction. Only addresses shared -# by >1 house_id are clustered; the keeper is rn=1 per cluster, losers are rn>1. -_BUILD_MAPPING_SQL = text( - f""" - CREATE TEMP TABLE _1772_dup_mapping ON COMMIT DROP AS - WITH clustered AS ( - SELECT - id, +# 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 AS cluster_key + END +""" + + +def _mapping_sql(cluster_key_case: str) -> str: + """Render the loser→keeper mapping SQL for one pass, given its cluster-key CASE expression. + + Only cluster keys shared by >1 house_id form a cluster; the keeper is rn=1 per cluster, losers + are rn>1. The final mapping is filtered by the 250 m geo guard AND the CROSS-FIAS guard (both + no-ops for the fias pass, where every clustered row shares one fias by construction). + `cluster_key_case` is a STATIC module constant (never runtime data) — no value injection. + """ + return f""" + CREATE TEMP TABLE _1772_dup_mapping ON COMMIT DROP AS + WITH clustered AS ( + SELECT + id, + {cluster_key_case} AS cluster_key FROM houses ), clusters_with_count AS ( @@ -155,6 +202,7 @@ _BUILD_MAPPING_SQL = text( 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} @@ -166,7 +214,11 @@ _BUILD_MAPPING_SQL = text( first_value(h.geom) OVER ( PARTITION BY dh.cluster_key ORDER BY {_KEEPER_ORDER} - ) AS keeper_geom + ) 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 @@ -179,6 +231,10 @@ _BUILD_MAPPING_SQL = text( -- the same canon-cluster but >250 m away, or with NULL geom on either side, are left as -- separate rows (conservative — never over-merges). Cluster_key также требует цифру в canon -- (номер дома) — исключает вырожденные canon вида «екатеринбург»/«сооружение». + -- CROSS-FIAS guard (#1772 follow-up): never merge two rows that BOTH carry a non-null but + -- DIFFERENT house_fias_id — they are provably different buildings the cluster key happened to + -- collapse (canon slash-collapse «Сулимова, 32»/«Сулимова, 3/2»). No-op for the fias pass (one + -- fias per cluster) and for canon clusters where at most one side has a fias. SELECT id AS loser_id, keeper_id, norm_address FROM ranked WHERE rn > 1 @@ -186,8 +242,18 @@ _BUILD_MAPPING_SQL = text( AND keeper_geom IS NOT NULL AND loser_geom IS NOT NULL AND ST_DistanceSphere(loser_geom, keeper_geom) <= 250 + AND NOT ( + NULLIF(loser_fias, '') IS NOT NULL + AND NULLIF(keeper_fias, '') IS NOT NULL + AND lower(loser_fias) <> lower(keeper_fias) + ) """ -) + + +# Canon-pass mapping — kept as a module constant for backward-compatible introspection/tests. +_BUILD_MAPPING_SQL = text(_mapping_sql(_CANON_KEY_EXPR)) +# Fias-pass mapping — same pipeline, clustered by the ФИАС building UUID (runs first). +_BUILD_MAPPING_SQL_FIAS = text(_mapping_sql(_FIAS_KEY_EXPR)) # Each step keys off _1772_dup_mapping → empty mapping ⇒ 0 rows touched ⇒ idempotent no-op. _STEPS: list[tuple[str, str]] = [ @@ -430,6 +496,63 @@ _BACKFILL_ALIASES_SQL = text( """ ) +# 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: @@ -455,14 +578,83 @@ class DedupMergeResult: } -def merge_duplicate_houses(db: Session, *, dry_run: bool = False) -> dict[str, int]: - """Cluster houses by canonical address and merge duplicates onto one canonical keeper. +def _run_merge_pass( + db: Session, + *, + build_sql: Any, + pass_label: str, + result: DedupMergeResult, +) -> None: + """Run ONE merge pass (fias- or canon-key) inside the caller's open transaction. - Re-implements migration 108's proven collision-safe pipeline as a RECURRING job: - cluster (canon address + 250 m geo guard) → pick keeper → - re-point children (UNIQUE-collision-safe) → delete losers → backfill sources/aliases. - All in ONE transaction. dry_run=True computes counts then ROLLS BACK (no writes). - Idempotent: a clean table yields an empty mapping → every statement is a 0-row no-op. + Builds a fresh loser→keeper mapping for this pass's cluster key, re-points every FK child + (UNIQUE-collision-safe), carries identity/enrichment onto the keeper, deletes the losers and + backfills sources/aliases. Accumulates counters onto `result`. NEVER commits/rolls back — the + caller owns the single transaction wrapping both passes. + """ + # Fresh mapping for this pass. ON COMMIT DROP only fires at txn end, so drop the temp table + # explicitly — the second pass must rebuild the same-named table within the one transaction. + db.execute(text("DROP TABLE IF EXISTS _1772_dup_mapping")) + db.execute(build_sql) + + mapping = db.execute( + text( + "SELECT loser_id, keeper_id, norm_address FROM _1772_dup_mapping " + "ORDER BY keeper_id, loser_id" + ) + ).all() + if not mapping: + logger.info("merge_duplicate_houses: pass=%s no clusters found (no-op)", pass_label) + return + + result.losers_deleted += len(mapping) + result.clusters_merged += len({row.keeper_id for row in mapping}) + + # Audit log: every loser→keeper move with its address, for traceability. + for row in mapping: + logger.info( + "merge_duplicate_houses: pass=%s merge loser_id=%d → keeper_id=%d address=%r", + pass_label, + row.loser_id, + row.keeper_id, + row.norm_address, + ) + + for label, sql in _STEPS: + res = db.execute(text(sql)) + rowcount = res.rowcount or 0 + if label == "listings": + result.listings_repointed += rowcount + elif label.endswith("(collision-delete)") or label.endswith("(dedup)"): + result.children_deleted += rowcount + elif label.endswith("(re-point)") or label in ( + "house_placement_history", + "house_reviews", + "house_reliability_checks", + "external_valuations", + ): + result.children_repointed += rowcount + logger.debug("merge_duplicate_houses: pass=%s step=%s rows=%d", pass_label, label, rowcount) + + # Carry identity/enrichment onto the keeper BEFORE the losers vanish, then delete + backfill. + db.execute(_CARRY_OVER_IDENTITY_SQL) + db.execute(_DELETE_LOSERS_SQL) + db.execute(_BACKFILL_SOURCES_SQL) + db.execute(_BACKFILL_ALIASES_SQL) + + +def merge_duplicate_houses(db: Session, *, dry_run: bool = False) -> dict[str, int]: + """Cluster houses by fias UUID, then by canonical address, merging dups onto one keeper. + + Re-implements migration 108's proven collision-safe pipeline as a RECURRING TWO-PASS job: + 1. FIAS pass — cluster by lower(NULLIF(house_fias_id, '')) (catches slash-collapse / + посёлок canon bugs the address canon misses). + 2. CANON pass — cluster by canonical address (250 m geo guard + cross-fias anti-over-merge + guard). + Each pass: pick keeper → re-point children (UNIQUE-collision-safe) → carry identity onto the + keeper → delete losers → backfill sources/aliases. BOTH passes run in ONE transaction. + dry_run=True computes counts then ROLLS BACK (no writes). Idempotent: a clean table yields an + empty mapping in each pass → every statement is a 0-row no-op. Returns the counter dict (DedupMergeResult.to_counters()). """ @@ -470,56 +662,21 @@ def merge_duplicate_houses(db: Session, *, dry_run: bool = False) -> dict[str, i result = DedupMergeResult(dry_run=dry_run) try: - db.execute(_BUILD_MAPPING_SQL) + # Pass 1: cluster by the ФИАС building UUID (runs first — most precise building identity). + _run_merge_pass(db, build_sql=_BUILD_MAPPING_SQL_FIAS, pass_label="fias", result=result) + # Pass 2: cluster by canonical address, with the cross-fias anti-over-merge guard. + _run_merge_pass(db, build_sql=_BUILD_MAPPING_SQL, pass_label="canon", result=result) - mapping = db.execute( - text( - "SELECT loser_id, keeper_id, norm_address FROM _1772_dup_mapping " - "ORDER BY keeper_id, loser_id" - ) - ).all() - result.losers_deleted = len(mapping) - result.clusters_merged = len({row.keeper_id for row in mapping}) - - if not mapping: - # Clean table — nothing to merge. Still roll back (we only opened a temp table). + 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-address clusters found (no-op) dry_run=%s", + "merge_duplicate_houses: no duplicate clusters found (no-op) dry_run=%s", dry_run, ) return result.to_counters() - # Audit log: every loser→keeper move with its address, for traceability. - for row in mapping: - logger.info( - "merge_duplicate_houses: merge loser_id=%d → keeper_id=%d address=%r", - row.loser_id, - row.keeper_id, - row.norm_address, - ) - - for label, sql in _STEPS: - res = db.execute(text(sql)) - rowcount = res.rowcount or 0 - if label == "listings": - result.listings_repointed = rowcount - elif label.endswith("(collision-delete)") or label.endswith("(dedup)"): - result.children_deleted += rowcount - elif label.endswith("(re-point)") or label in ( - "house_placement_history", - "house_reviews", - "house_reliability_checks", - "external_valuations", - ): - result.children_repointed += rowcount - logger.debug("merge_duplicate_houses: step=%s rows=%d", label, rowcount) - - db.execute(_DELETE_LOSERS_SQL) - db.execute(_BACKFILL_SOURCES_SQL) - db.execute(_BACKFILL_ALIASES_SQL) - if dry_run: db.rollback() logger.info( diff --git a/tradein-mvp/backend/tests/test_house_dedup_merge.py b/tradein-mvp/backend/tests/test_house_dedup_merge.py index 862ce43a..b51b8c89 100644 --- a/tradein-mvp/backend/tests/test_house_dedup_merge.py +++ b/tradein-mvp/backend/tests/test_house_dedup_merge.py @@ -32,6 +32,8 @@ _SQL_DIR = Path(__file__).resolve().parents[1] / "data" / "sql" _MIGRATION_135 = _SQL_DIR / "135_scrape_schedules_seed_house_dedup_merge.sql" _MAPPING_SQL = str(hdm._BUILD_MAPPING_SQL.text) +_FIAS_MAPPING_SQL = str(hdm._BUILD_MAPPING_SQL_FIAS.text) +_CARRY_SQL = str(hdm._CARRY_OVER_IDENTITY_SQL.text) _STEP_SQL = {label: sql for label, sql in hdm._STEPS} _ALL_STEP_SQL = "\n".join(sql for _, sql in hdm._STEPS) _SERVICE_SRC = inspect.getsource(hdm) @@ -235,6 +237,104 @@ def test_no_fstring_or_format_in_parametrised_sql() -> None: assert ".format(" not in _SERVICE_SRC +# ── Fias-key pass + cross-fias guard + identity carry-over (#1772 follow-up) ── + + +def test_fias_pass_clusters_by_house_fias_id() -> None: + """A second cluster key: lower(NULLIF(house_fias_id,'')) — the ФИАС/ГАР building UUID. + + The fias pass catches duplicates the address canon misses (slash-collapse «Сулимова, 32» vs + «Сулимова, 3/2»; посёлок truncation). Only non-null fias shared by >1 house_id clusters. + """ + flat = _flat(_FIAS_MAPPING_SQL) + assert "'fias:' || lower(house_fias_id)" in flat + assert "NULLIF(house_fias_id, '') IS NOT NULL" in flat + # same collision-safe pipeline (same temp table + >1 grouping), NOT the canon cluster key. + assert "CREATE TEMP TABLE _1772_dup_mapping" in flat + assert "HAVING count(*) > 1" in flat + # the fias pass keys on the fias UUID, never on the 'addr:' canon key. + assert "'addr:'" not in flat + + +def test_both_passes_share_one_pipeline_no_copy_paste() -> None: + """The two mappings differ ONLY in the cluster-key expression (parametrised, no copy-paste).""" + canon = _flat(_MAPPING_SQL) + fias = _flat(_FIAS_MAPPING_SQL) + assert "'addr:'" in canon and "'addr:'" not in fias + assert "'fias:'" in fias and "'fias:'" not in canon + # both carry the same downstream skeleton (keeper rule, geo guard, cross-fias guard). + for token in ( + "clusters_with_count", + "listing_counts", + "ranked AS", + "AS keeper_geom", + "ST_DistanceSphere(loser_geom, keeper_geom) <= 250", + ): + assert token in canon and token in fias + + +def test_cross_fias_guard_blocks_slash_collapse_over_merge() -> None: + """Canon pass must NOT merge two rows that BOTH carry a non-null but DIFFERENT house_fias_id. + + This is the critical anti-over-merge fix for the canon slash-collapse class: «Сулимова, 32» + and «Сулимова, 3/2» share the canon «сулимова32» but are different buildings — their distinct + fias UUIDs veto the merge. + """ + flat = _flat(_MAPPING_SQL) + assert "h.house_fias_id AS loser_fias" in flat + assert "AS keeper_fias" in flat + assert "first_value(h.house_fias_id) OVER" in flat + # the guard: NOT (both non-null AND lower-different). + assert "NULLIF(loser_fias, '') IS NOT NULL" in flat + assert "NULLIF(keeper_fias, '') IS NOT NULL" in flat + assert "lower(loser_fias) <> lower(keeper_fias)" in flat + + +def test_identity_carry_over_fills_all_enrichment_fields() -> None: + """Before deleting losers, the keeper's NULL identity / geo-QC fields are filled from losers.""" + flat = _flat(_CARRY_SQL) + assert "UPDATE houses k" in flat + for col in ( + "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", + ): + # COALESCE keeper-first ⇒ a populated keeper field is NEVER overwritten by a loser. + assert f"COALESCE(k.{col}, d.{col})" in flat, f"identity field {col} not carried over" + # deterministic donor selection: loser with geom first, then min id. + assert "ORDER BY (l.geom IS NOT NULL) DESC, l.id ASC" in flat + assert "FROM _1772_dup_mapping m" in flat + assert "JOIN houses l ON l.id = m.loser_id" in flat + + +def test_carry_over_runs_before_delete_losers_in_each_pass() -> None: + """Identity carry-over must execute BEFORE the loser rows are deleted, in every pass.""" + src = inspect.getsource(hdm._run_merge_pass) + assert "_CARRY_OVER_IDENTITY_SQL" in src + assert src.index("_CARRY_OVER_IDENTITY_SQL") < src.index("_DELETE_LOSERS_SQL") + + +def test_merge_runs_fias_then_canon_pass_in_one_transaction() -> None: + """merge_duplicate_houses runs the fias pass first, then the canon pass, in ONE transaction.""" + src = inspect.getsource(hdm.merge_duplicate_houses) + assert 'pass_label="fias"' in src and 'pass_label="canon"' in src + assert src.index('pass_label="fias"') < src.index('pass_label="canon"') + assert "_BUILD_MAPPING_SQL_FIAS" in src + # single transaction wrapping BOTH passes: exactly one commit. + assert src.count("db.commit()") == 1 + + +def test_carry_over_no_psycopg_v3_colon_colon_cast() -> None: + """psycopg v3: the carry-over SQL must not use the :param::type bound-param cast form.""" + assert not re.search(r":\w+::", _CARRY_SQL) + + # ── Service signature + dry_run / idempotent contract ───────────────────────── @@ -288,6 +388,7 @@ class _FakeDB: def __init__(self, mapping_rows: list[_Row], step_rowcount: int = 1): self._mapping_rows = mapping_rows self._step_rowcount = step_rowcount + self._mapping_served = False self.commits = 0 self.rollbacks = 0 self.executed: list[str] = [] @@ -298,8 +399,13 @@ class _FakeDB: if "CREATE TEMP TABLE" in sql: return _FakeResult() 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. + if self._mapping_served: + return _FakeResult(rows=[]) + self._mapping_served = True return _FakeResult(rows=list(self._mapping_rows)) - # any UPDATE/DELETE/INSERT step + # any UPDATE/DELETE/INSERT step (incl. DROP TABLE, carry-over, delete, backfill) return _FakeResult(rowcount=self._step_rowcount) def commit(self) -> None: @@ -714,3 +820,107 @@ def test_real_canon_clusterkey_and_geo_guard_merge_semantics() -> None: db.execute(_t("DELETE FROM houses WHERE id BETWEEN 900010 AND 900015")) db.commit() db.close() + + +@pytest.mark.skipif(_live_session() is None, reason="no reachable Postgres test DB") +def test_real_fias_pass_cross_guard_and_identity_carryover() -> None: + """End-to-end on a real DB for the #1772 follow-up (fias pass + cross-fias guard + carry-over): + + A. same house_fias_id, DIFFERENT canon (different streets) → MERGED by the FIAS pass; + the loser's listing re-points onto the single survivor. + B. same canon (slash-collapse «32»/«3/2»), BOTH fias filled and DIFFERENT → NOT merged + (cross-fias guard vetoes — provably different buildings), both survive. + C. same canon, fias only on the LOSER → MERGED by the canon pass; the loser's + house_fias_id + gar_house_guid CARRY OVER onto the fias-less keeper. + D. a second full run is idempotent (nothing left to merge). + + Synthetic streets «*1772» avoid colliding with any real prod address/alias. + """ + from sqlalchemy import text as _t + + db = _live_session() + assert db is not None + try: + db.execute( + _t( + "INSERT INTO houses " + "(id, source, ext_house_id, address, lat, lon, house_fias_id, gar_house_guid, " + " dadata_enriched_at) VALUES " + # A — same fias, different canon (different streets), ~10 m apart → FIAS-pass merge + "(900020,'avito','EXT-F-K','ФиасОдин1772, 10', 56.84000,60.60000," + " 'F-SAME-1772',NULL,NULL)," + "(900021,'cian', 'EXT-F-L','СовсемДругая1772, 77',56.84009,60.60000," + " 'F-SAME-1772',NULL,NULL)," + # B — same canon (slash-collapse), DIFFERENT fias → cross-fias guard BLOCKS + "(900022,'avito','EXT-B-1','Клара1772, 32',56.84000,60.60000," + " 'F-B1-1772',NULL,NULL)," + "(900023,'cian', 'EXT-B-2','Клара1772, 3/2',56.84009,60.60000," + " 'F-B2-1772',NULL,NULL)," + # C — same canon, fias only on the loser → canon-pass merge + carry-over + "(900024,'avito','EXT-C-K','Донбасс1772, 8',56.84000,60.60000,NULL,NULL,NULL)," + "(900025,'cian', 'EXT-C-L','Донбасс1772, 8',56.84009,60.60000," + " 'F-CARRY-1772','G-CARRY-1772',NOW())" + ) + ) + # Listings: A keeper + A loser (loser's must re-point); C keeper (gives it the listing-count + # tie-break so the fias-LESS row wins the keeper slot and receives the carried-over fias). + db.execute( + _t( + "INSERT INTO listings " + "(id, source, source_url, source_id, dedup_hash, price_rub, house_id_fk) VALUES " + "(910020,'avito','http://t/f/k','LF-K','dh-f-keep',5000000,900020)," + "(910021,'cian', 'http://t/f/l','LF-L','dh-f-lose',6000000,900021)," + "(910024,'avito','http://t/c/k','LC-K','dh-c-keep',5500000,900024)" + ) + ) + db.commit() + + out = hdm.merge_duplicate_houses(db, dry_run=False) + assert out["losers_deleted"] >= 2 # at least A + C merged + + ids = [ + r.id + for r in db.execute( + _t("SELECT id FROM houses WHERE id BETWEEN 900020 AND 900025 ORDER BY id") + ).all() + ] + # A: fias merge — different canon, same fias → the loser is gone. + assert 900020 in ids and 900021 not in ids, "same fias must merge across differing canons" + # B: cross-fias guard — same canon but different fias → BOTH survive. + assert 900022 in ids and 900023 in ids, "different fias must veto a same-canon merge" + # C: canon merge — same canon, one fias → the fias-less keeper survives. + assert 900024 in ids and 900025 not in ids, "same-canon single-fias pair must merge" + + # A: the merged loser's listing re-pointed onto the surviving keeper. + repointed = db.execute(_t("SELECT house_id_fk FROM listings WHERE id = 910021")).scalar() + assert repointed == 900020 + + # C: identity carry-over — the loser's fias + ГАР guid moved onto the fias-less keeper. + keeper_c = db.execute( + _t("SELECT house_fias_id, gar_house_guid FROM houses WHERE id = 900024") + ).one() + assert keeper_c.house_fias_id == "F-CARRY-1772", "loser's house_fias_id must carry over" + assert keeper_c.gar_house_guid == "G-CARRY-1772", "loser's gar_house_guid must carry over" + + # D: idempotent second run — nothing left to merge. + again = hdm.merge_duplicate_houses(db, dry_run=False) + assert again["losers_deleted"] == 0 + finally: + db.rollback() + db.execute(_t("DELETE FROM listings WHERE id IN (910020,910021,910024)")) + db.execute( + _t( + "DELETE FROM house_sources WHERE ext_id IN " + "('EXT-F-K','EXT-F-L','EXT-B-1','EXT-B-2','EXT-C-K','EXT-C-L')" + ) + ) + db.execute( + _t( + "DELETE FROM house_address_aliases WHERE normalized_address IN " + "('фиасодин1772, 10','совсемдругая1772, 77','клара1772, 32','клара1772, 3/2'," + "'донбасс1772, 8')" + ) + ) + db.execute(_t("DELETE FROM houses WHERE id BETWEEN 900020 AND 900025")) + db.commit() + db.close()