From fa5e5d83f190f814c5d7b84c4a470459e2fc19e9 Mon Sep 17 00:00:00 2001 From: Light1YT Date: Sat, 27 Jun 2026 02:23:22 +0500 Subject: [PATCH 1/2] feat(tradein): recurring house-dedup merge (schedule dormant) (#1772) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Reuses migration 108's proven collision-safe pattern (cluster → canonical keeper → re-point children with UNIQUE-collision handling → delete losers → backfill sources/aliases) as a RECURRING, idempotent scheduled job — dups recur because the matching pipeline keeps creating them, so a one-time migration cannot fix the steady-state inflow. - Service app/services/house_dedup_merge.py: merge_duplicate_houses(db, *, dry_run=False). Clusters houses by EXACT address (cadastral 100% NULL → address is the real key). Keeper: geom NOT NULL → most listings → most-populated → min(id). Handles ALL 11 FK children incl. live 6-col houses_price_dynamics UNIQUE (migration 029 replaced 108's stale 3-col). One transaction; dry_run computes counts then ROLLBACK; idempotent (clean table → empty mapping → 0-row no-op). Audit-logs every merge. - Scheduler: trigger_house_dedup_merge_run (run_in_executor, sync DB-only) + dispatch branch + run_house_dedup_merge lifecycle wrapper. - Migration 135: seeds scrape_schedules house_dedup_merge enabled=FALSE (DORMANT — deploy neutral; enable deliberately after dry-run + manual run), weekly window, ON CONFLICT DO NOTHING. - Tests: static SQL-shape (every FK child + collision handling, migration-133 guard), fake-db dry_run/idempotent, scheduler wiring, optional real-DB end-to-end merge test (self-skips without a DB). --- .../backend/app/services/house_dedup_merge.py | 567 ++++++++++++++++++ tradein-mvp/backend/app/services/scheduler.py | 64 ++ ...crape_schedules_seed_house_dedup_merge.sql | 79 +++ .../backend/tests/test_house_dedup_merge.py | 554 +++++++++++++++++ 4 files changed, 1264 insertions(+) create mode 100644 tradein-mvp/backend/app/services/house_dedup_merge.py create mode 100644 tradein-mvp/backend/data/sql/135_scrape_schedules_seed_house_dedup_merge.sql create mode 100644 tradein-mvp/backend/tests/test_house_dedup_merge.py diff --git a/tradein-mvp/backend/app/services/house_dedup_merge.py b/tradein-mvp/backend/app/services/house_dedup_merge.py new file mode 100644 index 00000000..00151345 --- /dev/null +++ b/tradein-mvp/backend/app/services/house_dedup_merge.py @@ -0,0 +1,567 @@ +"""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: EXACT `houses.address` (cadastral_number is 100% NULL on prod — confirmed + in migration 040 — so address is the real building key). Rows whose address is NULL/blank + are never clustered (cluster_key NULL → ignored). Only addresses shared by >1 house_id + form a cluster. + +KEEPER RULE (one canonical row per cluster), in priority order: + 1. geom NOT NULL first (a geocoded row is the better survivor) + 2. most linked listings (the row the corpus already points at) + 3. most-populated fields (the richest metadata row) + 4. min(id) (deterministic tie-break) + +CHILD-COLLISION HANDLING — the migration-133 lesson: + Migration 133 broke on prod because a child UNIQUE constraint was missed. EVERY FK child + of houses(id) is handled below. The FK set was re-audited against the LIVE schema + (grep "REFERENCES houses" data/sql/*.sql) — it is the SAME 11 children 108 handled; no + new FK child was added after 108. For each UNIQUE-constrained child we DELETE the loser + rows that would collide with the keeper (deduping by the TARGET key so dup-vs-dup + collisions inside clusters >2 are also caught) BEFORE re-pointing the survivors. + + IMPORTANT divergence from 108: houses_price_dynamics's UNIQUE was REPLACED by + migration 029 — it is now the 6-column + (house_id, source, room_count, prices_type, period, month_date) + NOT 108's stale 3-column (house_id, month_date, source). We dedup on the LIVE 6-column + key; deduping on the old key would still violate the live constraint on re-point. + + Per-child handling (loser=L, keeper=K): + listings.house_id_fk no UNIQUE → plain re-point + house_placement_history.house_id no UNIQUE → plain re-point + house_reviews.house_id no blocking UNIQUE→ plain re-point + house_reliability_checks.house_id no blocking UNIQUE→ plain re-point + external_valuations.house_id no blocking UNIQUE→ plain re-point + house_sources UNIQUE(ext_source,ext_id) → delete colliding L, re-point rest + house_address_aliases UNIQUE(normalized_address)→ delete colliding L, re-point rest + houses_price_dynamics UNIQUE(6-col, migr 029) → dedup by target 6-col key, re-point + house_imv_evaluations UNIQUE(house_id) → dedup by target house_id, re-point + house_suggestions UNIQUE(house_id,ext_item_id)→ dedup by target key, re-point + address_mismatch_audit UNIQUE(house_id,audit_batch)→ dedup by target key, re-point + +BACKFILL (reduces recurrence): + After deleting losers we backfill house_sources + house_address_aliases onto the keeper + (same as 108) so the matching pipeline's Tier-1/Tier-2 finds the keeper next scrape and + does not immediately re-split it. + +IDEMPOTENCY: + Every UPDATE/DELETE keys off a temp mapping of (loser→keeper). On a clean table the + mapping is empty → every statement touches 0 rows → no-op. Re-running is safe. + +DRY-RUN: + dry_run=True computes the same mapping and counts, then ROLLS BACK — no writes. Used by + the orchestrator to preview a merge before enabling the (DESTRUCTIVE) schedule. + +psycopg v3: all SQL uses CAST(:x AS type), never the colon-colon bound-param cast form. +""" + +from __future__ import annotations + +import logging +import time +from dataclasses import dataclass, field + +from sqlalchemy import text +from sqlalchemy.orm import Session + +logger = logging.getLogger(__name__) + +# Significant columns counted to score per-row completeness when picking the keeper. +# Mirrors 108's completeness expression (all confirmed present in the live houses schema). +_COMPLETENESS_EXPR = """ + (h.address IS NOT NULL)::int + + (h.lat IS NOT NULL)::int + + (h.lon IS NOT NULL)::int + + (h.year_built IS NOT NULL)::int + + (h.house_type IS NOT NULL)::int + + (h.total_floors IS NOT NULL)::int + + (h.cadastral_number IS NOT NULL)::int + + (h.developer_name IS NOT NULL)::int + + (h.rating_score IS NOT NULL)::int + + (h.avito_id_hash IS NOT NULL)::int +""" + +# Keeper ORDER BY, shared by the ROW_NUMBER() rank and the first_value() keeper pick so they +# agree row-for-row. Priority: geom present → most linked listings → most-populated → min id. +_KEEPER_ORDER = f""" + (h.geom IS NOT NULL) DESC, + listing_cnt DESC, + ({_COMPLETENESS_EXPR}) DESC, + h.id ASC +""" + +# 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, + CASE WHEN NULLIF(lower(trim(address)), '') IS NOT NULL + THEN 'addr:' || lower(trim(address)) END 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(h.address)) AS norm_address, + 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 + FROM dup_houses dh + JOIN houses h ON h.id = dh.id + LEFT JOIN listing_counts lc ON lc.house_id = dh.id + ) + SELECT id AS loser_id, keeper_id, norm_address + FROM ranked + WHERE rn > 1 + AND id <> keeper_id + """ +) + +# Each step keys off _1772_dup_mapping → empty mapping ⇒ 0 rows touched ⇒ idempotent no-op. +_STEPS: list[tuple[str, str]] = [ + # ── Plain re-point (no UNIQUE on the FK column) ─────────────────────────── + ( + "listings", + """ + UPDATE listings l + SET house_id_fk = m.keeper_id + FROM _1772_dup_mapping m + WHERE l.house_id_fk = m.loser_id + """, + ), + ( + "house_placement_history", + """ + UPDATE house_placement_history hph + SET house_id = m.keeper_id + FROM _1772_dup_mapping m + WHERE hph.house_id = m.loser_id + """, + ), + ( + "house_reviews", + """ + UPDATE house_reviews hr + SET house_id = m.keeper_id + FROM _1772_dup_mapping m + WHERE hr.house_id = m.loser_id + """, + ), + ( + "house_reliability_checks", + """ + UPDATE house_reliability_checks hrc + SET house_id = m.keeper_id + FROM _1772_dup_mapping m + WHERE hrc.house_id = m.loser_id + """, + ), + ( + "external_valuations", + """ + UPDATE external_valuations ev + SET house_id = m.keeper_id + FROM _1772_dup_mapping m + WHERE ev.house_id = m.loser_id + """, + ), + # ── UNIQUE(ext_source, ext_id): delete colliding losers, re-point rest ───── + ( + "house_sources(collision-delete)", + """ + DELETE FROM house_sources hs + USING _1772_dup_mapping m + WHERE hs.house_id = m.loser_id + AND EXISTS ( + SELECT 1 FROM house_sources hs2 + WHERE hs2.house_id = m.keeper_id + AND hs2.ext_source = hs.ext_source + AND hs2.ext_id = hs.ext_id + ) + """, + ), + ( + "house_sources(re-point)", + """ + UPDATE house_sources hs + SET house_id = m.keeper_id + FROM _1772_dup_mapping m + WHERE hs.house_id = m.loser_id + """, + ), + # ── UNIQUE(normalized_address): delete colliding losers, re-point rest ───── + ( + "house_address_aliases(collision-delete)", + """ + DELETE FROM house_address_aliases haa + USING _1772_dup_mapping m + WHERE haa.house_id = m.loser_id + AND EXISTS ( + SELECT 1 FROM house_address_aliases haa2 + WHERE haa2.house_id = m.keeper_id + AND haa2.normalized_address = haa.normalized_address + ) + """, + ), + ( + "house_address_aliases(re-point)", + """ + UPDATE house_address_aliases haa + SET house_id = m.keeper_id + FROM _1772_dup_mapping m + WHERE haa.house_id = m.loser_id + """, + ), + # ── UNIQUE(house_id, source, room_count, prices_type, period, month_date) ── + # LIVE 6-col key (migration 029 replaced 108's stale 3-col). Dedup by the TARGET key + # (keeper survives, else min id) so dup-vs-dup collisions in clusters >2 are also caught. + ( + "houses_price_dynamics(dedup)", + """ + DELETE FROM houses_price_dynamics t + USING ( + SELECT t2.id, + ROW_NUMBER() OVER ( + PARTITION BY COALESCE(m.keeper_id, t2.house_id), + t2.source, t2.room_count, t2.prices_type, + t2.period, t2.month_date + ORDER BY (m.keeper_id IS NULL) DESC, t2.id ASC + ) AS rn + FROM houses_price_dynamics t2 + LEFT JOIN _1772_dup_mapping m ON m.loser_id = t2.house_id + ) d + WHERE t.id = d.id AND d.rn > 1 + """, + ), + ( + "houses_price_dynamics(re-point)", + """ + UPDATE houses_price_dynamics hpd + SET house_id = m.keeper_id + FROM _1772_dup_mapping m + WHERE hpd.house_id = m.loser_id + """, + ), + # ── UNIQUE(house_id): one evaluation per keeper ─────────────────────────── + ( + "house_imv_evaluations(dedup)", + """ + DELETE FROM house_imv_evaluations t + USING ( + SELECT t2.id, + ROW_NUMBER() OVER ( + PARTITION BY COALESCE(m.keeper_id, t2.house_id) + ORDER BY (m.keeper_id IS NULL) DESC, t2.id ASC + ) AS rn + FROM house_imv_evaluations t2 + LEFT JOIN _1772_dup_mapping m ON m.loser_id = t2.house_id + ) d + WHERE t.id = d.id AND d.rn > 1 + """, + ), + ( + "house_imv_evaluations(re-point)", + """ + UPDATE house_imv_evaluations hie + SET house_id = m.keeper_id + FROM _1772_dup_mapping m + WHERE hie.house_id = m.loser_id + """, + ), + # ── UNIQUE(house_id, ext_item_id) ───────────────────────────────────────── + ( + "house_suggestions(dedup)", + """ + DELETE FROM house_suggestions t + USING ( + SELECT t2.id, + ROW_NUMBER() OVER ( + PARTITION BY COALESCE(m.keeper_id, t2.house_id), t2.ext_item_id + ORDER BY (m.keeper_id IS NULL) DESC, t2.id ASC + ) AS rn + FROM house_suggestions t2 + LEFT JOIN _1772_dup_mapping m ON m.loser_id = t2.house_id + ) d + WHERE t.id = d.id AND d.rn > 1 + """, + ), + ( + "house_suggestions(re-point)", + """ + UPDATE house_suggestions hs + SET house_id = m.keeper_id + FROM _1772_dup_mapping m + WHERE hs.house_id = m.loser_id + """, + ), + # ── UNIQUE(house_id, audit_batch) ───────────────────────────────────────── + ( + "address_mismatch_audit(dedup)", + """ + DELETE FROM address_mismatch_audit t + USING ( + SELECT t2.id, + ROW_NUMBER() OVER ( + PARTITION BY COALESCE(m.keeper_id, t2.house_id), t2.audit_batch + ORDER BY (m.keeper_id IS NULL) DESC, t2.id ASC + ) AS rn + FROM address_mismatch_audit t2 + LEFT JOIN _1772_dup_mapping m ON m.loser_id = t2.house_id + ) d + WHERE t.id = d.id AND d.rn > 1 + """, + ), + ( + "address_mismatch_audit(re-point)", + """ + UPDATE address_mismatch_audit ama + SET house_id = m.keeper_id + FROM _1772_dup_mapping m + WHERE ama.house_id = m.loser_id + """, + ), +] + +# Delete the loser houses — all FK children are re-pointed or CASCADE by now. +_DELETE_LOSERS_SQL = text( + """ + DELETE FROM houses h + USING _1772_dup_mapping m + WHERE h.id = m.loser_id + """ +) + +# Backfill house_sources for keeper avito-rows so Tier-1 (source_exact) finds the keeper next +# scrape (reduces recurrence). Same shape as 108 Step 5. +_BACKFILL_SOURCES_SQL = text( + """ + INSERT INTO house_sources (house_id, ext_source, ext_id, confidence, matched_method, matched_at) + SELECT h.id, h.source, h.ext_house_id, 1.0, 'backfill_dedup_merge', NOW() + FROM houses h + WHERE h.source = 'avito' + AND h.ext_house_id IS NOT NULL + AND h.id IN (SELECT DISTINCT keeper_id FROM _1772_dup_mapping) + ON CONFLICT (ext_source, ext_id) DO NOTHING + """ +) + +# Backfill normalized-address aliases onto the keeper so Tier-2 (address) finds it next scrape. +_BACKFILL_ALIASES_SQL = text( + """ + INSERT INTO house_address_aliases (house_id, normalized_address, fingerprint, source) + SELECT h.id, lower(trim(h.address)), NULL, 'backfill_dedup_merge' + FROM houses h + WHERE h.address IS NOT NULL + AND length(trim(h.address)) >= 5 + AND h.id IN (SELECT DISTINCT keeper_id FROM _1772_dup_mapping) + ON CONFLICT (normalized_address) DO NOTHING + """ +) + + +@dataclass +class DedupMergeResult: + """Counters from one merge pass (also the dry-run preview).""" + + clusters_merged: int = 0 # distinct keepers that absorbed at least one loser + losers_deleted: int = 0 # duplicate house rows removed (0 on a clean table) + listings_repointed: int = 0 # listings.house_id_fk moved loser→keeper + children_deleted: int = 0 # collision/dedup deletions across all UNIQUE children + children_repointed: int = 0 # survivor child rows moved loser→keeper + dry_run: bool = False + duration_sec: float = field(default=0.0) + + def to_counters(self) -> dict[str, int]: + return { + "clusters_merged": self.clusters_merged, + "losers_deleted": self.losers_deleted, + "listings_repointed": self.listings_repointed, + "children_deleted": self.children_deleted, + "children_repointed": self.children_repointed, + "dry_run": int(self.dry_run), + "duration_sec": int(self.duration_sec), + } + + +def merge_duplicate_houses(db: Session, *, dry_run: bool = False) -> dict[str, int]: + """Cluster houses by exact address and merge duplicates onto one canonical keeper. + + Re-implements migration 108's proven collision-safe pipeline as a RECURRING job: + cluster (exact address) → pick keeper → re-point children (UNIQUE-collision-safe) → + delete losers → backfill sources/aliases onto keeper. + 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. + + Returns the counter dict (DedupMergeResult.to_counters()). + """ + start = time.monotonic() + result = DedupMergeResult(dry_run=dry_run) + + try: + db.execute(_BUILD_MAPPING_SQL) + + 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). + db.rollback() + result.duration_sec = time.monotonic() - start + logger.info( + "merge_duplicate_houses: no duplicate-address 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( + "merge_duplicate_houses: DRY-RUN computed clusters=%d losers=%d " + "listings_repointed=%d children_deleted=%d children_repointed=%d — ROLLED BACK", + result.clusters_merged, + result.losers_deleted, + result.listings_repointed, + result.children_deleted, + result.children_repointed, + ) + else: + db.commit() + logger.info( + "merge_duplicate_houses: COMMITTED clusters=%d losers=%d " + "listings_repointed=%d children_deleted=%d children_repointed=%d", + result.clusters_merged, + result.losers_deleted, + result.listings_repointed, + result.children_deleted, + result.children_repointed, + ) + except Exception: + logger.exception("merge_duplicate_houses: FAILED — rolling back") + try: + db.rollback() + except Exception: + logger.exception("merge_duplicate_houses: rollback also failed") + raise + + result.duration_sec = time.monotonic() - start + return result.to_counters() + + +# ── Run lifecycle wrapper (scheduler entrypoint) ───────────────────────────── +def run_house_dedup_merge(db: Session, *, run_id: int, params: dict) -> dict[str, int]: + """Run-lifecycle wrapper for the recurring house-dedup merge (sync, DB-only). + + Launched by the in-app scheduler (source='house_dedup_merge') via + trigger_house_dedup_merge_run, or manually. Mirrors run_cadastral_geo_match: pure + internal DB op, finalises scrape_runs (mark_done / mark_failed) with counters. + + Params (default_params jsonb): + dry_run: compute + return counts then ROLLBACK without writing (default false). + The deploy seeds the schedule DISABLED; the orchestrator can flip a single + manual run to dry_run=true to preview before enabling. + """ + from app.services import scrape_runs as runs_mod + + dry_run = bool(params.get("dry_run", False)) + counters: dict[str, int] = { + "clusters_merged": 0, + "losers_deleted": 0, + "listings_repointed": 0, + "children_deleted": 0, + "children_repointed": 0, + "dry_run": int(dry_run), + } + try: + runs_mod.update_heartbeat(db, run_id, counters) + counters = merge_duplicate_houses(db, dry_run=dry_run) + runs_mod.mark_done(db, run_id, counters) + logger.info( + "run_house_dedup_merge: run_id=%d DONE clusters=%d losers=%d dry_run=%s", + run_id, + counters.get("clusters_merged", 0), + counters.get("losers_deleted", 0), + dry_run, + ) + return counters + except Exception as exc: + logger.exception("run_house_dedup_merge: run_id=%d FAILED", run_id) + try: + db.rollback() + except Exception: + pass + runs_mod.mark_failed(db, run_id, str(exc)[:1000], counters) + raise diff --git a/tradein-mvp/backend/app/services/scheduler.py b/tradein-mvp/backend/app/services/scheduler.py index 17f3c7ae..281369f5 100644 --- a/tradein-mvp/backend/app/services/scheduler.py +++ b/tradein-mvp/backend/app/services/scheduler.py @@ -110,6 +110,21 @@ Sources: использует собственную curl_cffi-сессию; window 15:00-17:00 UTC — после avito_full_load 13-15 UTC и утреннего Avito-блока, дома уже geocoded/enriched дневными sweep'ами) + - house_dedup_merge → run_house_dedup_merge + (services/house_dedup_merge.py, #1772; recurring, idempotent, + collision-safe house-deduplication merge. Clusters houses by + EXACT address (cadastral_number is 100% NULL), picks one canonical + keeper per cluster (geom NOT NULL → most listings → most-populated + → min id), re-points all FK children with UNIQUE-collision handling + — reuses migration 108's proven pattern — deletes loser rows, + backfills house_sources/house_address_aliases onto the keeper so the + matching pipeline finds the keeper next scrape. All in ONE + transaction; dry_run param computes counts then ROLLBACKs. + DESTRUCTIVE (DELETEs duplicate house rows) but idempotent: clean + table → empty mapping → 0-row no-op. Seed 135 ships the schedule + DISABLED (enabled=false) — deploy is neutral; enable deliberately + after a validated dry-run + manual run; weekly window 04:00-05:00 + UTC — quiet hour, no scraper/proxy contention) """ from __future__ import annotations @@ -1654,6 +1669,53 @@ async def trigger_house_imv_backfill_run(db: Session, schedule_row: dict[str, An return run_id +async def trigger_house_dedup_merge_run(db: Session, schedule_row: dict[str, Any]) -> int | None: + """Создать scrape_runs + launch run_house_dedup_merge в executor (sync DB-only task, #1772). + + Recurring, idempotent, collision-safe house-deduplication merge: clusters `houses` by + EXACT address, picks one canonical keeper per cluster (geom NOT NULL → most listings → + most-populated → min id), re-points all FK children with UNIQUE-collision handling + (reuses migration 108's proven pattern), deletes the loser rows, backfills + house_sources/house_address_aliases onto the keeper — all in ONE transaction. + + DESTRUCTIVE (DELETEs duplicate house rows) but idempotent: a clean table yields an empty + mapping → every statement is a 0-row no-op. Seed 135 ships the schedule DISABLED + (enabled=false) — the deploy is neutral; the orchestrator dry-runs + does one validated + manual run, then enables deliberately. default_params.dry_run=true previews without writes. + + Sync task (set-based UPDATE/DELETE, no async HTTP) — run in run_in_executor by the same + pattern as trigger_cadastral_geo_match_run. run_house_dedup_merge owns the scrape_runs + lifecycle (mark_done/mark_failed). + + Returns run_id или None (skip — already running). + """ + run_id = _claim_run(db, schedule_row) + if run_id is None: + return None + + params = schedule_row.get("default_params") or {} + + async def _run() -> None: + run_db = SessionLocal() + try: + from app.services.house_dedup_merge import run_house_dedup_merge + + loop = asyncio.get_event_loop() + await loop.run_in_executor( + None, + lambda: run_house_dedup_merge(run_db, run_id=run_id, params=params), + ) + except Exception: + logger.exception("scheduler: run_house_dedup_merge crashed run_id=%d", run_id) + finally: + run_db.close() + + task = asyncio.create_task(_run()) + task.add_done_callback(lambda t: t.exception() if not t.cancelled() else None) + logger.info("scheduler: triggered house_dedup_merge run_id=%d", run_id) + return run_id + + def get_due_schedules(db: Session) -> list[dict[str, Any]]: """SELECT scrape_schedules WHERE enabled AND (next_run_at IS NULL OR next_run_at <= NOW()).""" rows = ( @@ -1737,6 +1799,8 @@ async def scheduler_loop() -> None: await trigger_cadastral_geo_match_run(db, sch) elif source == "house_imv_backfill": await trigger_house_imv_backfill_run(db, sch) + elif source == "house_dedup_merge": + await trigger_house_dedup_merge_run(db, sch) else: logger.warning("scheduler: unknown source=%s, skip", source) finally: diff --git a/tradein-mvp/backend/data/sql/135_scrape_schedules_seed_house_dedup_merge.sql b/tradein-mvp/backend/data/sql/135_scrape_schedules_seed_house_dedup_merge.sql new file mode 100644 index 00000000..e7c29483 --- /dev/null +++ b/tradein-mvp/backend/data/sql/135_scrape_schedules_seed_house_dedup_merge.sql @@ -0,0 +1,79 @@ +-- 135_scrape_schedules_seed_house_dedup_merge.sql +-- Scheduler seed for the recurring house-deduplication merge (#1772). +-- +-- WHAT (source='house_dedup_merge'): +-- trigger_house_dedup_merge_run (scheduler.py) → run_house_dedup_merge +-- (services/house_dedup_merge.py) → merge_duplicate_houses. Recurring, idempotent, +-- collision-safe merge of duplicate `houses` rows: +-- 1. Cluster houses by EXACT address (cadastral_number is 100% NULL on prod — confirmed +-- in migration 040 — so address is the real building key). Only addresses shared by +-- >1 house_id form a cluster. +-- 2. Pick one canonical keeper per cluster: geom NOT NULL → most linked listings → +-- most-populated fields → min(id). +-- 3. Re-point every FK child of houses(id) onto the keeper with UNIQUE-collision handling +-- (reuses migration 108's proven cluster→canonical→re-point→delete pattern). The 11 +-- children + their constraints (listings, house_placement_history, house_reviews, +-- house_reliability_checks, external_valuations, house_sources, house_address_aliases, +-- houses_price_dynamics [LIVE 6-col UNIQUE per migration 029], house_imv_evaluations, +-- house_suggestions, address_mismatch_audit). +-- 4. DELETE the loser house rows; backfill house_sources + house_address_aliases onto the +-- keeper so the matching pipeline (Tier-1/2) finds the keeper next scrape (reduces +-- recurrence). ALL in ONE transaction. +-- Idempotent: a clean table yields an empty mapping → every statement is a 0-row no-op. +-- dry_run param computes the same counts then ROLLBACKs (no writes). +-- +-- WHY RECURRING (not a one-shot migration): +-- Migration 108 did a ONE-TIME merge, but duplicates RECUR — the matching pipeline keeps +-- creating them (Tier-3 geo-jitter + per-source ext_id INSERTs). A migration cannot fix a +-- steady-state inflow; this schedule re-runs and is a no-op when there is nothing to merge. +-- +-- *** enabled=false (DORMANT) — DEPLOY MUST BE NEUTRAL *** +-- This is an auto-recurring DESTRUCTIVE merge (it DELETEs duplicate house rows). The deploy +-- ships it DISABLED so nothing fires automatically. The orchestrator dry-runs it +-- (default_params.dry_run=true on a manual run) + does ONE validated manual run, then flips +-- enabled=true deliberately: +-- UPDATE scrape_schedules SET enabled = true WHERE source = 'house_dedup_merge'; +-- +-- Window 04:00-05:00 UTC (07:00-08:00 MSK): +-- - Quiet hour: after the overnight scraper/sweep block and before the morning Avito block +-- (city_sweep 06-07, detail_backfill 09-12). Pure internal DB op — no scraper/proxy +-- contention. interval_days=7 → weekly cadence (default_params). +-- - next_run_at = tomorrow 04:00 UTC (NOT NULL — follows the seed convention; harmless while +-- enabled=false, since get_due_schedules filters enabled=true first). +-- +-- default_params: +-- dry_run -- false: actually merge. The orchestrator overrides to true on a manual run +-- to preview counts without writing. +-- interval_days -- 7: weekly cadence (compute_next_run_at reads this). +-- +-- DEPENDENCIES: 052_scrape_schedules.sql (table + UNIQUE(source)), +-- 108_merge_duplicate_houses.sql (the proven collision-safe pattern this reuses), +-- 029_extend_matching_valuation_dynamics.sql (LIVE 6-col houses_price_dynamics UNIQUE). +-- Idempotent: ON CONFLICT (source) DO NOTHING. +-- Runner applies migrations WITHOUT --single-transaction, so wrap in explicit BEGIN/COMMIT. + +BEGIN; + +INSERT INTO scrape_schedules ( + source, + enabled, + window_start_hour, + window_end_hour, + next_run_at, + default_params +) +VALUES +( + 'house_dedup_merge', + false, + 4, + 5, + ((CURRENT_DATE + INTERVAL '1 day') + make_interval(hours => 4)) AT TIME ZONE 'UTC', + '{"dry_run": false, "interval_days": 7}'::jsonb +) +ON CONFLICT (source) DO NOTHING; + +COMMENT ON TABLE scrape_schedules IS + 'In-app scheduler config (replaces cron-script setup). Sources: avito_city_sweep, yandex_city_sweep (dormant, #561), cian_history_backfill, rosreestr_dkp_import, listing_source_snapshot (#570), asking_to_sold_ratio_refresh (#648), refresh_search_matview (#769), yandex_address_backfill (#855, EKB pilot), sber_index_pull (#887, monthly), rosreestr_quarter_poll (#888, monthly), cian_city_sweep (dormant, #973), yandex_newbuilding_sweep (dormant, #974), geocode_missing_listings (#1: listings geom backfill, all sources), avito_detail_backfill (#1551: nightly detail-enrichment backfill for legacy avito listings), house_imv_backfill (#854: nightly bulk Avito IMV at house level → house_imv_evaluations), house_dedup_merge (#1772: recurring collision-safe merge of duplicate houses by exact address — DORMANT enabled=false, DESTRUCTIVE, enable deliberately after a validated dry-run + manual run).'; + +COMMIT; diff --git a/tradein-mvp/backend/tests/test_house_dedup_merge.py b/tradein-mvp/backend/tests/test_house_dedup_merge.py new file mode 100644 index 00000000..a6e750fc --- /dev/null +++ b/tradein-mvp/backend/tests/test_house_dedup_merge.py @@ -0,0 +1,554 @@ +"""Tests for the recurring house-deduplication merge (#1772). + +Convention mirrors tests/tasks/test_cadastral_geo_match.py and +tests/test_house_imv_backfill_scheduler.py: the merge is SQL-heavy and the CI gate has no +live Postgres, so the bulk of coverage is STATIC analysis of the emitted SQL (text() clauses ++ inspect.getsource) plus a fake-db behavioural pass driving the dry_run / idempotent / count +logic. An OPTIONAL real-Postgres behavioural test asserts the actual merge semantics (re-point +listings, dedup colliding children WITHOUT a UNIQUE violation, delete loser, keeper survives, +idempotent second run, dry_run no-write) and self-SKIPS when no DB is reachable. + +The static block is the migration-133 guard: it asserts EVERY FK child of houses(id) is +handled, and that the UNIQUE-constrained children dedup by the TARGET key before re-pointing. +""" + +from __future__ import annotations + +import inspect +import os +import re +from pathlib import Path +from typing import Any + +import pytest + +# settings needs a DSN at import (same dance as the sibling tests); these are static/fake-db. +os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test") + +from app.services import house_dedup_merge as hdm +from app.services import scheduler + +_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) +_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) + + +def _flat(sql: str) -> str: + return re.sub(r"\s+", " ", sql) + + +# ── Cluster key + keeper rule ───────────────────────────────────────────────── + + +def test_cluster_key_is_exact_address_not_cadastral() -> None: + """Spec: cadastral is unpopulated → cluster by EXACT address only.""" + flat = _flat(_MAPPING_SQL) + assert "'addr:' || lower(trim(address))" in flat + # cadastral_number must NOT be part of the cluster key (it is 100% NULL). + assert "cadastral_number" not in _flat( + _MAPPING_SQL[: _MAPPING_SQL.index("ranked")] + ) # not in cluster construction + # Only addresses shared by >1 house form a cluster. + assert "HAVING count(*) > 1" in flat + + +def test_keeper_rule_priority_order() -> None: + """Keeper: geom NOT NULL → most linked listings → most-populated → min(id).""" + order = _flat(hdm._KEEPER_ORDER) + assert "(h.geom IS NOT NULL) DESC" in order + assert "listing_cnt DESC" in order + assert "h.id ASC" in order + # geom precedes listing count precedes id (ordering matters). + assert order.index("geom IS NOT NULL") < order.index("listing_cnt") + assert order.index("listing_cnt") < order.index("h.id ASC") + # completeness is counted from the live houses columns. + for col in ("year_built", "house_type", "total_floors", "developer_name", "rating_score"): + assert col in hdm._COMPLETENESS_EXPR + + +def test_mapping_selects_losers_only() -> None: + """The mapping is loser_id→keeper_id; the keeper itself is excluded (rn>1, id<>keeper).""" + flat = _flat(_MAPPING_SQL) + assert "SELECT id AS loser_id, keeper_id" in flat + assert "WHERE rn > 1" in flat + assert "id <> keeper_id" in flat + + +def test_mapping_temp_table_drops_on_commit() -> None: + """TEMP ... ON COMMIT DROP — the mapping never leaks across runs (idempotent setup).""" + assert "CREATE TEMP TABLE _1772_dup_mapping ON COMMIT DROP" in _flat(_MAPPING_SQL) + + +# ── EVERY FK child of houses(id) is handled (the migration-133 guard) ────────── + +# The authoritative FK-child set (grep "REFERENCES houses" data/sql/*.sql). If a new child +# is ever added this list must grow with it — the test then forces a matching merge step. +_FK_CHILDREN = { + "listings": "house_id_fk", + "house_placement_history": "house_id", + "house_reviews": "house_id", + "house_reliability_checks": "house_id", + "external_valuations": "house_id", + "house_sources": "house_id", + "house_address_aliases": "house_id", + "houses_price_dynamics": "house_id", + "house_imv_evaluations": "house_id", + "house_suggestions": "house_id", + "address_mismatch_audit": "house_id", +} + + +def test_every_fk_child_of_houses_is_repointed() -> None: + """Each FK child must be re-pointed loser→keeper somewhere in the merge steps.""" + for table in _FK_CHILDREN: + assert table in _ALL_STEP_SQL, f"FK child {table} is never touched by the merge" + + +def test_unique_constrained_children_dedup_before_repoint() -> None: + """UNIQUE-constrained children must DELETE colliding losers before re-pointing survivors. + + Missing one of these is exactly what broke migration 133. We assert each has both a + collision/dedup DELETE step and a re-point UPDATE step, and that the DELETE precedes + the UPDATE in step order. + """ + unique_children = [ + ("house_sources", "house_sources(collision-delete)", "house_sources(re-point)"), + ( + "house_address_aliases", + "house_address_aliases(collision-delete)", + "house_address_aliases(re-point)", + ), + ( + "houses_price_dynamics", + "houses_price_dynamics(dedup)", + "houses_price_dynamics(re-point)", + ), + ( + "house_imv_evaluations", + "house_imv_evaluations(dedup)", + "house_imv_evaluations(re-point)", + ), + ("house_suggestions", "house_suggestions(dedup)", "house_suggestions(re-point)"), + ( + "address_mismatch_audit", + "address_mismatch_audit(dedup)", + "address_mismatch_audit(re-point)", + ), + ] + labels = [label for label, _ in hdm._STEPS] + for _table, del_label, rep_label in unique_children: + assert del_label in _STEP_SQL, f"{del_label} missing" + assert rep_label in _STEP_SQL, f"{rep_label} missing" + assert "DELETE FROM" in _STEP_SQL[del_label] + assert "UPDATE" in _STEP_SQL[rep_label] + assert labels.index(del_label) < labels.index(rep_label) + + +def test_price_dynamics_dedup_uses_live_6col_key_not_stale_3col() -> None: + """Migration 029 replaced the price_dynamics UNIQUE with a 6-col key — dedup on it. + + Deduping on 108's stale (house_id, month_date, source) would still violate the live + 6-col constraint on re-point. The dedup PARTITION must include all 6 dimension columns. + """ + sql = _flat(_STEP_SQL["houses_price_dynamics(dedup)"]) + for col in ("source", "room_count", "prices_type", "period", "month_date"): + assert col in sql, f"price_dynamics dedup missing dimension column {col}" + # partition over the TARGET keeper (COALESCE(keeper_id, house_id)) so dup-vs-dup is caught. + assert "PARTITION BY COALESCE(m.keeper_id, t2.house_id)" in sql + + +def test_collision_dedup_partitions_on_target_keeper() -> None: + """All dedup steps partition on the keeper target so cluster-of-3 dup-vs-dup collisions + are caught, with the keeper's own row surviving (keeper_id IS NULL DESC).""" + for label in ( + "houses_price_dynamics(dedup)", + "house_imv_evaluations(dedup)", + "house_suggestions(dedup)", + "address_mismatch_audit(dedup)", + ): + sql = _flat(_STEP_SQL[label]) + assert "PARTITION BY COALESCE(m.keeper_id, t2.house_id)" in sql + assert "(m.keeper_id IS NULL) DESC" in sql + + +def test_losers_deleted_and_keeper_backfilled() -> None: + """Loser houses deleted; keeper backfilled with sources + aliases to reduce recurrence.""" + assert "DELETE FROM houses" in _flat(str(hdm._DELETE_LOSERS_SQL.text)) + bf_sources = _flat(str(hdm._BACKFILL_SOURCES_SQL.text)) + assert "INSERT INTO house_sources" in bf_sources + assert "ON CONFLICT (ext_source, ext_id) DO NOTHING" in bf_sources + bf_aliases = _flat(str(hdm._BACKFILL_ALIASES_SQL.text)) + assert "INSERT INTO house_address_aliases" in bf_aliases + assert "ON CONFLICT (normalized_address) DO NOTHING" in bf_aliases + + +def test_no_psycopg_v3_colon_colon_cast() -> None: + """psycopg v3: never :param::type — must use CAST(:param AS type).""" + assert not re.search(r":\w+::", _MAPPING_SQL) + assert not re.search(r":\w+::", _ALL_STEP_SQL) + assert not re.search(r":\w+::", _SERVICE_SRC) + + +def test_no_fstring_or_format_in_parametrised_sql() -> None: + """SQL identifiers are static; no .format()/f-string interpolation of *values* into SQL.""" + # The only f-strings build static column-list fragments (_KEEPER_ORDER / completeness), + # never user/runtime values — assert no '.format(' value-injection into SQL text. + assert ".format(" not in _SERVICE_SRC + + +# ── Service signature + dry_run / idempotent contract ───────────────────────── + + +def test_merge_signature_and_dry_run_default() -> None: + sig = inspect.signature(hdm.merge_duplicate_houses) + assert "dry_run" in sig.parameters + assert sig.parameters["dry_run"].default is False + # dry_run is keyword-only (the `*` in the spec). + assert sig.parameters["dry_run"].kind is inspect.Parameter.KEYWORD_ONLY + + +def test_service_logs_every_merge_for_audit() -> None: + """Every loser→keeper move is logged with its address (audit trail).""" + assert "loser_id=%d → keeper_id=%d address=%r" in _SERVICE_SRC + + +def test_single_transaction_commit_or_rollback() -> None: + """One transaction: dry_run → rollback; real → commit; failure → rollback + re-raise.""" + src = inspect.getsource(hdm.merge_duplicate_houses) + assert "db.rollback()" in src + assert "db.commit()" in src + assert "raise" in src # failures re-raise, never swallowed + + +# ── Fake-db behavioural: dry_run / idempotent / counts (no Postgres) ────────── + + +class _FakeResult: + def __init__(self, rowcount: int = 0, rows: list[Any] | None = None, scalar: Any = None): + self.rowcount = rowcount + self._rows = rows or [] + self._scalar = scalar + + def all(self) -> list[Any]: + return self._rows + + def scalar(self) -> Any: + return self._scalar + + +class _Row: + def __init__(self, loser_id: int, keeper_id: int, norm_address: str): + self.loser_id = loser_id + self.keeper_id = keeper_id + self.norm_address = norm_address + + +class _FakeDB: + """Session stand-in: build-mapping + a scripted SELECT result, then per-step rowcounts.""" + + def __init__(self, mapping_rows: list[_Row], step_rowcount: int = 1): + self._mapping_rows = mapping_rows + self._step_rowcount = step_rowcount + self.commits = 0 + self.rollbacks = 0 + self.executed: list[str] = [] + + def execute(self, clause: Any, params: dict | None = None) -> _FakeResult: + sql = str(getattr(clause, "text", clause)) + self.executed.append(sql) + if "CREATE TEMP TABLE" in sql: + return _FakeResult() + if "SELECT loser_id, keeper_id, norm_address" in sql: + return _FakeResult(rows=list(self._mapping_rows)) + # any UPDATE/DELETE/INSERT step + return _FakeResult(rowcount=self._step_rowcount) + + def commit(self) -> None: + self.commits += 1 + + def rollback(self) -> None: + self.rollbacks += 1 + + +def test_idempotent_no_op_when_no_dups() -> None: + """Empty mapping → no commit (only the temp table was opened) → all-zero counters.""" + db = _FakeDB(mapping_rows=[]) + out = hdm.merge_duplicate_houses(db, dry_run=False) # type: ignore[arg-type] + assert out["clusters_merged"] == 0 + assert out["losers_deleted"] == 0 + assert out["listings_repointed"] == 0 + assert db.commits == 0 # nothing to commit + assert db.rollbacks == 1 # the no-op rollback + + +def test_dry_run_computes_counts_but_rolls_back() -> None: + """dry_run=True → counts populated, ZERO commits, exactly one rollback (no writes).""" + rows = [_Row(2, 1, "ул. ленина, 5"), _Row(3, 1, "ул. ленина, 5")] + db = _FakeDB(mapping_rows=rows, step_rowcount=2) + out = hdm.merge_duplicate_houses(db, dry_run=True) # type: ignore[arg-type] + assert out["dry_run"] == 1 + assert out["clusters_merged"] == 1 # both losers → one keeper + assert out["losers_deleted"] == 2 + assert out["listings_repointed"] == 2 # the 'listings' step rowcount + assert db.commits == 0 # dry-run NEVER commits + assert db.rollbacks == 1 + + +def test_real_merge_commits() -> None: + """dry_run=False with dups → exactly one commit, no rollback.""" + rows = [_Row(2, 1, "ул. мира, 10")] + db = _FakeDB(mapping_rows=rows, step_rowcount=1) + out = hdm.merge_duplicate_houses(db, dry_run=False) # type: ignore[arg-type] + assert out["dry_run"] == 0 + assert out["losers_deleted"] == 1 + assert db.commits == 1 + assert db.rollbacks == 0 + + +# ── Run-lifecycle wrapper ───────────────────────────────────────────────────── + + +def test_run_wrapper_marks_done_with_counters(monkeypatch: pytest.MonkeyPatch) -> None: + marked: dict[str, Any] = {} + from app.services import scrape_runs as runs_mod + + monkeypatch.setattr(runs_mod, "update_heartbeat", lambda *a, **k: None) + monkeypatch.setattr( + runs_mod, + "mark_done", + lambda _db, run_id, counters: marked.update(run_id=run_id, counters=dict(counters)), + ) + monkeypatch.setattr(runs_mod, "mark_failed", lambda *a, **k: None) + monkeypatch.setattr( + hdm, + "merge_duplicate_houses", + lambda _db, dry_run=False: {"clusters_merged": 3, "losers_deleted": 5, "dry_run": 0}, + ) + + out = hdm.run_house_dedup_merge(object(), run_id=42, params={"dry_run": False}) # type: ignore[arg-type] + assert out["clusters_merged"] == 3 + assert marked["run_id"] == 42 + assert marked["counters"]["losers_deleted"] == 5 + + +def test_run_wrapper_passes_dry_run_param(monkeypatch: pytest.MonkeyPatch) -> None: + captured: dict[str, Any] = {} + from app.services import scrape_runs as runs_mod + + monkeypatch.setattr(runs_mod, "update_heartbeat", lambda *a, **k: None) + monkeypatch.setattr(runs_mod, "mark_done", lambda *a, **k: None) + monkeypatch.setattr(runs_mod, "mark_failed", lambda *a, **k: None) + + def _fake_merge(_db: Any, dry_run: bool = False) -> dict[str, int]: + captured["dry_run"] = dry_run + return {"dry_run": int(dry_run)} + + monkeypatch.setattr(hdm, "merge_duplicate_houses", _fake_merge) + hdm.run_house_dedup_merge(object(), run_id=1, params={"dry_run": True}) # type: ignore[arg-type] + assert captured["dry_run"] is True + + +def test_run_wrapper_marks_failed_on_error(monkeypatch: pytest.MonkeyPatch) -> None: + failed: dict[str, Any] = {} + from app.services import scrape_runs as runs_mod + + monkeypatch.setattr(runs_mod, "update_heartbeat", lambda *a, **k: None) + monkeypatch.setattr(runs_mod, "mark_done", lambda *a, **k: None) + monkeypatch.setattr( + runs_mod, + "mark_failed", + lambda _db, run_id, err, counters: failed.update(run_id=run_id, err=err), + ) + + def _boom(_db: Any, dry_run: bool = False) -> dict[str, int]: + raise RuntimeError("merge exploded") + + monkeypatch.setattr(hdm, "merge_duplicate_houses", _boom) + + class _DB: + def rollback(self) -> None: + pass + + with pytest.raises(RuntimeError): + hdm.run_house_dedup_merge(_DB(), run_id=9, params={}) # type: ignore[arg-type] + assert failed["run_id"] == 9 + assert "merge exploded" in failed["err"] + + +# ── Scheduler wiring ────────────────────────────────────────────────────────── + + +def test_scheduler_has_trigger_and_dispatch() -> None: + assert hasattr(scheduler, "trigger_house_dedup_merge_run") + loop_src = inspect.getsource(scheduler.scheduler_loop) + assert 'source == "house_dedup_merge"' in loop_src + assert "trigger_house_dedup_merge_run(db, sch)" in loop_src + + +def test_trigger_claims_run_and_runs_in_executor() -> None: + src = inspect.getsource(scheduler.trigger_house_dedup_merge_run) + assert "_claim_run(db, schedule_row)" in src + assert "if run_id is None:" in src + # sync DB-only task → run_in_executor (mirror cadastral_geo_match, not a bare await). + assert "run_in_executor" in src + assert "run_house_dedup_merge" in src + # fresh session inside the spawned task + RUF006 keep-alive callback. + assert "run_db = SessionLocal()" in src + assert "task.add_done_callback(" in src + assert "run_db.close()" in src + + +# ── Migration content ───────────────────────────────────────────────────────── + + +def test_migration_135_seeds_schedule_disabled() -> None: + """The deploy must be NEUTRAL — schedule seeded enabled=false (DORMANT).""" + sql = _MIGRATION_135.read_text(encoding="utf-8") + assert "INSERT INTO scrape_schedules" in sql + assert "'house_dedup_merge'" in sql + assert "ON CONFLICT (source) DO NOTHING" in sql + assert "BEGIN;" in sql and "COMMIT;" in sql + # enabled=false in the VALUES (not enabled=true). Find the source line's VALUES tuple. + flat = re.sub(r"\s+", " ", sql) + m = re.search(r"'house_dedup_merge',\s*(\w+)", flat) + assert m is not None and m.group(1) == "false", "schedule MUST be seeded enabled=false" + + +def test_migration_135_weekly_window_and_dry_run_param() -> None: + sql = _MIGRATION_135.read_text(encoding="utf-8") + # weekly cadence + dry_run default param present. + assert '"interval_days": 7' in sql + assert '"dry_run": false' in sql + # next_run_at NOT NULL = tomorrow (seed convention; harmless while disabled). + assert "CURRENT_DATE + INTERVAL '1 day'" in sql + assert "make_interval(hours => 4)" in sql + + +def test_migration_135_no_psycopg_v3_colon_colon_cast() -> None: + sql = _MIGRATION_135.read_text(encoding="utf-8") + assert not re.search(r":\w+::", sql) + + +# ── Optional real-Postgres behavioural merge test (self-skips without a DB) ──── + + +def _live_session() -> Any | None: + """Return a SQLAlchemy Session if a non-placeholder Postgres is reachable, else None.""" + try: + from sqlalchemy import create_engine + from sqlalchemy.orm import sessionmaker + + dsn = os.environ.get("TEST_DATABASE_URL") or os.environ.get("DATABASE_URL", "") + if not dsn or "localhost:5432/test" in dsn: + return None + engine = create_engine(dsn, future=True) + conn = engine.connect() + from sqlalchemy import text as _t + + conn.execute(_t("SELECT 1")) + conn.close() + return sessionmaker(bind=engine, future=True)() + except Exception: + return None + + +@pytest.mark.skipif(_live_session() is None, reason="no reachable Postgres test DB") +def test_real_merge_repoints_dedups_deletes_and_is_idempotent() -> None: + """End-to-end on a real DB: 2 houses sharing an address + colliding children → + merge re-points listings, dedups the colliding children WITHOUT a UNIQUE violation, + deletes the loser, keeper survives with merged data; second run is a no-op; dry_run + makes no writes.""" + from sqlalchemy import text as _t + + db = _live_session() + assert db is not None + try: + # Two houses at the SAME address. Keeper (geom present) should win. + db.execute( + _t( + "INSERT INTO houses (id, source, ext_house_id, address, lat, lon) VALUES " + "(900001, 'avito', 'EXT-KEEP', 'тестдом 1772, 1', 56.84, 60.60)," + "(900002, 'cian', 'EXT-LOSE', 'тестдом 1772, 1', NULL, NULL)" + ) + ) + # listings pointing at BOTH (the loser's must be re-pointed). + db.execute( + _t( + "INSERT INTO listings (id, source, source_id, house_id_fk) VALUES " + "(910001, 'avito', 'L-KEEP', 900001)," + "(910002, 'cian', 'L-LOSE', 900002)" + ) + ) + # COLLIDING children: both houses have a price_dynamics row with the SAME 6-col key. + db.execute( + _t( + "INSERT INTO houses_price_dynamics " + "(house_id, month_date, source, room_count, prices_type, period, avg_price) VALUES " + "(900001, DATE '2026-01-01', 'cian', 'all', 'priceSqm', 'allTime', 100000)," + "(900002, DATE '2026-01-01', 'cian', 'all', 'priceSqm', 'allTime', 999999)" + ) + ) + # COLLIDING house_sources: same (ext_source, ext_id) on both would violate on re-point. + db.execute( + _t( + "INSERT INTO house_sources " + "(house_id, ext_source, ext_id, confidence, matched_method) VALUES " + "(900001, 'avito', 'SRC-DUP', 1.0, 'test')," + "(900002, 'avito', 'SRC-DUP', 1.0, 'test')" + ) + ) + db.commit() + + # ── dry_run makes NO writes ── + before = db.execute(_t("SELECT count(*) FROM houses WHERE id IN (900001,900002)")).scalar() + dry = hdm.merge_duplicate_houses(db, dry_run=True) + assert dry["losers_deleted"] >= 1 + after_dry = db.execute( + _t("SELECT count(*) FROM houses WHERE id IN (900001,900002)") + ).scalar() + assert after_dry == before == 2, "dry_run must not delete anything" + + # ── real merge ── + out = hdm.merge_duplicate_houses(db, dry_run=False) + assert out["losers_deleted"] >= 1 + # loser gone, keeper survives. + survivors = db.execute( + _t("SELECT id FROM houses WHERE id IN (900001,900002) ORDER BY id") + ).all() + assert [r.id for r in survivors] == [900001] + # loser's listing re-pointed to keeper. + repointed = db.execute( + _t("SELECT house_id_fk FROM listings WHERE id = 910002") + ).scalar() + assert repointed == 900001 + # price_dynamics deduped — exactly one row for the keeper at that 6-col key (no dup). + pd_cnt = db.execute( + _t( + "SELECT count(*) FROM houses_price_dynamics " + "WHERE house_id = 900001 AND month_date = DATE '2026-01-01' " + "AND source='cian' AND room_count='all' AND prices_type='priceSqm' " + "AND period='allTime'" + ) + ).scalar() + assert pd_cnt == 1, "colliding price_dynamics must be deduped, not duplicated" + # house_sources collision resolved — one SRC-DUP row on the keeper. + hs_cnt = db.execute( + _t("SELECT count(*) FROM house_sources WHERE ext_source='avito' AND ext_id='SRC-DUP'") + ).scalar() + assert hs_cnt == 1 + + # ── idempotent second run = no-op ── + again = hdm.merge_duplicate_houses(db, dry_run=False) + assert again["losers_deleted"] == 0 + finally: + # cleanup (children CASCADE on houses delete). + db.rollback() + db.execute(_t("DELETE FROM listings WHERE id IN (910001,910002)")) + db.execute(_t("DELETE FROM houses WHERE id IN (900001,900002)")) + db.execute(_t("DELETE FROM house_sources WHERE ext_id='SRC-DUP'")) + db.commit() + db.close() -- 2.45.3 From 70ca65b62b4c4c3ff54fe1ece5be0fe78440e01c Mon Sep 17 00:00:00 2001 From: Light1YT Date: Sat, 27 Jun 2026 02:38:15 +0500 Subject: [PATCH 2/2] test(tradein): fix real-DB merge test schema (#1772, #1933 review) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The optional real-DB end-to-end test self-skipped without a DB but would have errored WITH one (the safety fixture never actually ran). Fixed two schema bugs + one impossible-collision setup, validated against a live PostGIS 16 container (test now passes end-to-end, not just collected): - houses_price_dynamics insert used a non-existent `avg_price` column → `price_per_sqm` (the real NOT NULL value column, migration 024). The two rows share all 5 dim fields with differing house_id → a genuine LIVE 6-col UNIQUE collision the dedup step must resolve. Also assert the surviving row is the keeper's own (price_per_sqm=100000, not 999999). - listings insert omitted NOT NULL no-default columns → added source_url, dedup_hash (also UNIQUE → distinct per row), price_rub (migration 002). - house_sources UNIQUE(ext_source,ext_id) is GLOBAL, so the original same-key-on-both-houses setup violated at INSERT time (before the merge). Replaced with distinct loser/keeper source rows + assert the loser's row re-points onto the keeper and none dangle on the deleted loser. Still self-skips in CI (no Postgres) — correct; now valid against a tunnel/local DB. Full suite 2444 passed / 0 failed; ruff clean. --- .../backend/tests/test_house_dedup_merge.py | 74 ++++++++++++++----- 1 file changed, 55 insertions(+), 19 deletions(-) diff --git a/tradein-mvp/backend/tests/test_house_dedup_merge.py b/tradein-mvp/backend/tests/test_house_dedup_merge.py index a6e750fc..24455db6 100644 --- a/tradein-mvp/backend/tests/test_house_dedup_merge.py +++ b/tradein-mvp/backend/tests/test_house_dedup_merge.py @@ -475,30 +475,44 @@ def test_real_merge_repoints_dedups_deletes_and_is_idempotent() -> None: "(900002, 'cian', 'EXT-LOSE', 'тестдом 1772, 1', NULL, NULL)" ) ) - # listings pointing at BOTH (the loser's must be re-pointed). + # listings pointing at BOTH (the loser's must be re-pointed). source_url, dedup_hash, + # price_rub are NOT NULL with no default (002_core_tables.sql); dedup_hash is also + # UNIQUE, so each row needs a distinct value. db.execute( _t( - "INSERT INTO listings (id, source, source_id, house_id_fk) VALUES " - "(910001, 'avito', 'L-KEEP', 900001)," - "(910002, 'cian', 'L-LOSE', 900002)" + "INSERT INTO listings " + "(id, source, source_url, source_id, dedup_hash, price_rub, house_id_fk) " + "VALUES " + "(910001, 'avito', 'http://t/1772/k', 'L-KEEP', 'dh-1772-keep', 5000000, 900001)," + "(910002, 'cian', 'http://t/1772/l', 'L-LOSE', 'dh-1772-lose', 6000000, 900002)" ) ) - # COLLIDING children: both houses have a price_dynamics row with the SAME 6-col key. + # COLLIDING children: both houses have a price_dynamics row whose 5 dimension fields + # (source, room_count, prices_type, period, month_date) are IDENTICAL and only house_id + # differs (loser vs keeper). After the loser re-points to the keeper, both rows would + # share the LIVE 6-col UNIQUE key → the dedup step must drop the loser's row first. + # price_per_sqm is the real NOT NULL value column (024_houses_price_dynamics.sql); + # there is no avg_price column. db.execute( _t( "INSERT INTO houses_price_dynamics " - "(house_id, month_date, source, room_count, prices_type, period, avg_price) VALUES " + "(house_id, month_date, source, room_count, prices_type, period, price_per_sqm) " + "VALUES " "(900001, DATE '2026-01-01', 'cian', 'all', 'priceSqm', 'allTime', 100000)," "(900002, DATE '2026-01-01', 'cian', 'all', 'priceSqm', 'allTime', 999999)" ) ) - # COLLIDING house_sources: same (ext_source, ext_id) on both would violate on re-point. + # house_sources UNIQUE(ext_source, ext_id) is GLOBAL (not scoped to house_id), so two + # rows can never share the key at insert time — a true same-key collision is impossible + # to seed. We give the loser and keeper DISTINCT source rows; the merge must re-point the + # loser's onto the keeper (the collision-delete guard is a defensive no-op here, and both + # rows survive on the keeper). db.execute( _t( "INSERT INTO house_sources " "(house_id, ext_source, ext_id, confidence, matched_method) VALUES " - "(900001, 'avito', 'SRC-DUP', 1.0, 'test')," - "(900002, 'avito', 'SRC-DUP', 1.0, 'test')" + "(900001, 'avito', 'SRC-KEEP', 1.0, 'test')," + "(900002, 'cian', 'SRC-LOSE', 1.0, 'test')" ) ) db.commit() @@ -525,30 +539,52 @@ def test_real_merge_repoints_dedups_deletes_and_is_idempotent() -> None: _t("SELECT house_id_fk FROM listings WHERE id = 910002") ).scalar() assert repointed == 900001 - # price_dynamics deduped — exactly one row for the keeper at that 6-col key (no dup). - pd_cnt = db.execute( + # price_dynamics deduped — exactly one row for the keeper at that 6-col key (no dup), + # and it is the KEEPER's own row (price_per_sqm=100000), not the loser's (999999): + # the dedup keeps keeper rows (keeper_id IS NULL DESC) over loser rows. + pd_rows = db.execute( _t( - "SELECT count(*) FROM houses_price_dynamics " + "SELECT price_per_sqm FROM houses_price_dynamics " "WHERE house_id = 900001 AND month_date = DATE '2026-01-01' " "AND source='cian' AND room_count='all' AND prices_type='priceSqm' " "AND period='allTime'" ) + ).all() + assert len(pd_rows) == 1, "colliding price_dynamics must be deduped, not duplicated" + assert pd_rows[0].price_per_sqm == 100000, "dedup must keep the keeper's own row" + # house_sources: the loser's source row re-pointed onto the keeper — both now on 900001 + # (plus the backfill row for the keeper's own avito ext_house_id). The loser's row must + # be present and re-pointed; no row may dangle on the deleted loser. + hs_on_keeper = db.execute( + _t( + "SELECT count(*) FROM house_sources " + "WHERE house_id = 900001 AND ext_id IN ('SRC-KEEP','SRC-LOSE')" + ) ).scalar() - assert pd_cnt == 1, "colliding price_dynamics must be deduped, not duplicated" - # house_sources collision resolved — one SRC-DUP row on the keeper. - hs_cnt = db.execute( - _t("SELECT count(*) FROM house_sources WHERE ext_source='avito' AND ext_id='SRC-DUP'") + assert hs_on_keeper == 2, "both source rows must be re-pointed onto the keeper" + hs_dangling = db.execute( + _t("SELECT count(*) FROM house_sources WHERE house_id = 900002") ).scalar() - assert hs_cnt == 1 + assert hs_dangling == 0, "no source row may remain on the deleted loser" # ── idempotent second run = no-op ── again = hdm.merge_duplicate_houses(db, dry_run=False) assert again["losers_deleted"] == 0 finally: - # cleanup (children CASCADE on houses delete). + # cleanup: drop listings first (house_id_fk has no CASCADE), then the houses — which + # CASCADE-deletes house_sources / price_dynamics / aliases / etc. Also sweep any + # backfill house_sources rows the merge may have inserted for the keeper. db.rollback() db.execute(_t("DELETE FROM listings WHERE id IN (910001,910002)")) + db.execute( + _t("DELETE FROM house_sources WHERE ext_id IN ('SRC-KEEP','SRC-LOSE','EXT-KEEP')") + ) + db.execute( + _t( + "DELETE FROM house_address_aliases " + "WHERE normalized_address = 'тестдом 1772, 1'" + ) + ) db.execute(_t("DELETE FROM houses WHERE id IN (900001,900002)")) - db.execute(_t("DELETE FROM house_sources WHERE ext_id='SRC-DUP'")) db.commit() db.close() -- 2.45.3