"""Recurring, idempotent, collision-safe house-deduplication merge (#1772).

WHY (recurrence, not a one-shot):
    Migration 108_merge_duplicate_houses.sql did a ONE-TIME merge of duplicate `houses`
    rows, but the duplicates RECUR: the matching pipeline keeps creating them
    (Tier-3 geo-jitter splits one physical building across rows + per-source ext_id INSERTs).
    A single migration cannot fix a steady-state inflow — we need a RECURRING merge that
    re-runs on a schedule and is a no-op when there is nothing to merge.

WHAT this is:
    A direct, set-based reimplementation of 108's proven
        cluster → pick canonical → re-point children (UNIQUE-collision-safe) → delete losers
    pipeline, run inside ONE transaction so a crash leaves the table untouched.

    Cluster key: CANONICAL address via tradein_canon_addr() over the CLEAN address
    COALESCE(short_address, full_address, address) (cadastral_number is 100% NULL on prod —
    confirmed in migration 040 — so address is the real building key). The clean source matters:
    `address` can carry район-noise the canon does not strip (e.g. «улица Вайнера, 66 · р-н Центр»
    → canon «вайнера66рнцентр»), while `short_address` holds the clean «улица Вайнера, 66»
    (→ «вайнера66») — preferring the clean field lets such a row cluster with its twin. The canon
    collapses spelling/район variants of the SAME building (ул→улица, strips город/район/«россия»/
    «м-н»/«р-н» noise, removes spaces/punct, PRESERVES корпус) so e.g. «ул. Вайнера,66» and
    «улица Вайнера, 66» share one cluster_key. Rows whose canon is NULL/blank are never clustered
    (cluster_key NULL → ignored). Only canons shared by >1 house_id form a cluster.

    GEO GUARD (anti over-merge, CANON PASS ONLY — #2187): because the canon strips город/район,
    two different buildings with the same street+number in different region-66 towns (e.g. «Ленина
    5») would share a canon. To avoid merging them, within a canon-cluster a house is a LOSER only
    if it is within 250 m of the keeper (ST_DistanceSphere on the WGS84 geom). Same-canon houses
    >250 m away — or with NULL geom on either side — are left as separate rows (conservative). The
    FIAS pass deliberately SKIPS this guard: a shared ФИАС/ГАР UUID IS the building identity and
    strictly outranks geo-proximity, so same-fias rows merge even with NULL geom on a side or
    >250 m apart (the geom-first keeper rule simultaneously repairs the broken coordinate).

TWO PASSES (2026-07-02 follow-up): the SAME cluster→keeper→re-point→carry-identity→delete
    pipeline now runs TWICE inside one transaction, parametrised by the cluster-key expression
    (`_mapping_sql`) — no 300-line copy-paste:
      1. FIAS pass — clusters by lower(NULLIF(house_fias_id, '')) (the ФИАС/ГАР building UUID,
         DaData-backfilled 2026-07-02 for 5 122 houses). Only non-null fias shared by >1 house_id
         cluster. Catches duplicates the canon MISSES: slash-collapse («Сулимова, 32» vs
         «Сулимова, 3/2» → same canon «сулимова32») and посёлок truncation (Кедровка/Шувакиш
         colliding with a same-named ЕКБ street).
      2. CANON pass — the canonical-address clustering above, now with a CROSS-FIAS GUARD: within a
         canon cluster a loser is NOT merged when it AND the keeper both carry a non-null but
         DIFFERENT house_fias_id — provably different buildings the canon collapsed (the
         slash-collapse class). Critical anti-over-merge fix. The 250 m geo guard applies to the
         CANON pass ONLY (#2187): the FIAS pass merges on UUID identity regardless of geom — a
         боевой прогон left 130 same-fias groups (335 houses, 1 529 listings) split because the
         guard blocked them (NULL geom on a side, or >250 m from a broken coordinate).

IDENTITY CARRY-OVER (each pass, BEFORE deleting losers): the keeper's NULL identity / geo-QC
    fields are filled from its losers with COALESCE semantics (keeper value wins; donor = the
    loser with geom, then min id): house_fias_id, cadastral_number, gar_house_guid, gar_flat_count,
    gar_matched_at, gar_match_method, dadata_qc_geo, dadata_qc_house, dadata_enriched_at. Without
    this the merge would silently drop DaData/ГАР enrichment that lived only on a loser row.

KEEPER RULE (one canonical row per cluster), in priority order:
    1. geom NOT NULL first      (a geocoded row is the better survivor)
    2. most linked listings     (the row the corpus already points at)
    3. most-populated fields     (the richest metadata row)
    4. min(id)                  (deterministic tie-break)

CHILD-COLLISION HANDLING — the migration-133 lesson:
    Migration 133 broke on prod because a child UNIQUE constraint was missed. EVERY FK child
    of houses(id) is handled below. The FK set was re-audited against the LIVE schema
    (grep "REFERENCES houses" data/sql/*.sql) — it is the SAME 11 children 108 handled; no
    new FK child was added after 108. For each UNIQUE-constrained child we DELETE the loser
    rows that would collide with the keeper (deduping by the TARGET key so dup-vs-dup
    collisions inside clusters >2 are also caught) BEFORE re-pointing the survivors.

    IMPORTANT divergence from 108: houses_price_dynamics's UNIQUE was REPLACED by
    migration 029 — it is now the 6-column
        (house_id, source, room_count, prices_type, period, month_date)
    NOT 108's stale 3-column (house_id, month_date, source). We dedup on the LIVE 6-column
    key; deduping on the old key would still violate the live constraint on re-point.

    Per-child handling (loser=L, keeper=K):
      listings.house_id_fk              no UNIQUE         → plain re-point
      house_placement_history.house_id  no UNIQUE         → plain re-point
      house_reviews.house_id            no blocking UNIQUE→ plain re-point
      house_reliability_checks.house_id no blocking UNIQUE→ plain re-point
      external_valuations.house_id      no blocking UNIQUE→ plain re-point
      house_sources             UNIQUE(ext_source,ext_id) → delete colliding L, re-point rest
      house_address_aliases     UNIQUE(normalized_address)→ delete colliding L, re-point rest
      houses_price_dynamics     UNIQUE(6-col, migr 029)   → dedup by target 6-col key, re-point
      house_imv_evaluations     UNIQUE(house_id)          → dedup by target house_id, re-point
      house_suggestions         UNIQUE(house_id,ext_item_id)→ dedup by target key, re-point
      address_mismatch_audit    UNIQUE(house_id,audit_batch)→ dedup by target key, re-point

BACKFILL (reduces recurrence):
    After deleting losers we backfill house_sources + house_address_aliases onto the keeper
    (same as 108) so the matching pipeline's Tier-1/Tier-2 finds the keeper next scrape and
    does not immediately re-split it.

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 typing import Any

from sqlalchemy import text
from sqlalchemy.orm import Session

logger = logging.getLogger(__name__)

# Significant columns counted to score per-row completeness when picking the keeper.
# Mirrors 108's completeness expression (all confirmed present in the live houses schema).
_COMPLETENESS_EXPR = """
    (h.address IS NOT NULL)::int +
    (h.lat IS NOT NULL)::int +
    (h.lon IS NOT NULL)::int +
    (h.year_built IS NOT NULL)::int +
    (h.house_type IS NOT NULL)::int +
    (h.total_floors IS NOT NULL)::int +
    (h.cadastral_number IS NOT NULL)::int +
    (h.developer_name IS NOT NULL)::int +
    (h.rating_score IS NOT NULL)::int +
    (h.avito_id_hash IS NOT NULL)::int
"""

# Keeper ORDER BY, shared by the ROW_NUMBER() rank and the first_value() keeper pick so they
# agree row-for-row. Priority: geom present → most linked listings → most-populated → min id.
_KEEPER_ORDER = f"""
    (h.geom IS NOT NULL) DESC,
    listing_cnt DESC,
    ({_COMPLETENESS_EXPR}) DESC,
    h.id ASC
"""

# Cluster-key expressions, one per merge pass. The pipeline (build mapping → keeper → re-point
# children → carry identity → delete losers → backfill) is IDENTICAL for both passes; only the
# cluster key differs, parametrised into `_mapping_sql`, so there is NO copy-paste of the pipeline.
#
# FIAS key (#1772 follow-up, 2026-07-02): the ФИАС/ГАР building UUID (DaData backfill). Catches
# duplicates the canon misses — slash-collapse («Сулимова, 32» vs «Сулимова, 3/2») and посёлок
# truncation (Кедровка/Шувакиш). Only non-null fias clusters.
_FIAS_KEY_EXPR = """
            CASE WHEN NULLIF(house_fias_id, '') IS NOT NULL
                 THEN 'fias:' || lower(house_fias_id)
                 END
"""

# CANON key: canonical address over the CLEAN address (short_address→full_address→address). The
# canon collapses spelling/район variants of the SAME building (ул→улица, strips город/район noise,
# preserves корпус); the digit guard drops street-less degenerate canons; blank canons never join.
_CANON_KEY_EXPR = """
            CASE WHEN NULLIF(
                       tradein_canon_addr(COALESCE(short_address, full_address, address)), ''
                     ) IS NOT NULL
                  AND tradein_canon_addr(COALESCE(short_address, full_address, address)) ~ '[0-9]'
                 THEN 'addr:'
                      || tradein_canon_addr(COALESCE(short_address, full_address, address))
                 END
"""


def _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
    WITH clustered AS (
        SELECT
            id,
            {cluster_key_case} AS cluster_key
        FROM houses
    ),
    clusters_with_count AS (
        SELECT cluster_key, count(*) AS n
        FROM clustered
        WHERE cluster_key IS NOT NULL
        GROUP BY cluster_key
        HAVING count(*) > 1
    ),
    dup_houses AS (
        SELECT h.id, cl.cluster_key
        FROM houses h
        JOIN clustered cl ON cl.id = h.id
        JOIN clusters_with_count cw ON cw.cluster_key = cl.cluster_key
    ),
    listing_counts AS (
        SELECT house_id_fk AS house_id, count(*) AS listing_cnt
        FROM listings
        WHERE house_id_fk IS NOT NULL
        GROUP BY house_id_fk
    ),
    ranked AS (
        SELECT
            dh.id,
            dh.cluster_key,
            lower(trim(COALESCE(h.short_address, h.full_address, h.address))) AS norm_address,
            h.geom AS loser_geom,
            h.house_fias_id AS loser_fias,
            ROW_NUMBER() OVER (
                PARTITION BY dh.cluster_key
                ORDER BY {_KEEPER_ORDER}
            ) AS rn,
            first_value(dh.id) OVER (
                PARTITION BY dh.cluster_key
                ORDER BY {_KEEPER_ORDER}
            ) AS keeper_id,
            first_value(h.geom) OVER (
                PARTITION BY dh.cluster_key
                ORDER BY {_KEEPER_ORDER}
            ) AS keeper_geom,
            first_value(h.house_fias_id) OVER (
                PARTITION BY dh.cluster_key
                ORDER BY {_KEEPER_ORDER}
            ) AS keeper_fias
        FROM dup_houses dh
        JOIN houses h ON h.id = dh.id
        LEFT JOIN listing_counts lc ON lc.house_id = dh.id
    )
    -- CROSS-FIAS guard (#1772 follow-up): never merge two rows that BOTH carry a non-null but
    -- DIFFERENT house_fias_id — provably different buildings the cluster key collapsed (canon
    -- slash-collapse «Сулимова, 32»/«Сулимова, 3/2»). No-op for the fias pass (one fias per
    -- cluster) and for canon clusters where at most one side carries a fias.
    SELECT id AS loser_id, keeper_id, norm_address
    FROM ranked
    WHERE rn > 1
      AND id <> keeper_id{geo_guard}
      AND NOT (
          NULLIF(loser_fias, '') IS NOT NULL
          AND NULLIF(keeper_fias, '') IS NOT NULL
          AND lower(loser_fias) <> lower(keeper_fias)
      )
    """


# Canon-pass mapping — geo guard ON (cross-town over-merge protection for street+number canons).
_BUILD_MAPPING_SQL = text(_mapping_sql(_CANON_KEY_EXPR))
# Fias-pass mapping — same pipeline, clustered by the ФИАС building UUID (runs first). Geo guard
# OFF (#2187): a shared ГАР UUID IS the building identity and outranks proximity — same-fias rows
# merge even with NULL geom or >250 m apart (the geom-first keeper rule fixes broken coords).
_BUILD_MAPPING_SQL_FIAS = text(_mapping_sql(_FIAS_KEY_EXPR, apply_geo_guard=False))

# 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
    """
)

# Carry identity / geo-QC / ГАР-enrichment onto the keeper BEFORE deleting losers (#1772
# follow-up). COALESCE — the keeper's own populated value always wins; a NULL keeper field is
# filled from a donor loser. Donor = the loser with geom first, then min id (deterministic):
# array_agg(... ORDER BY geom-present DESC, id ASC) FILTER (WHERE <field> NOT NULL) [1] takes the
# first non-null value in that order. Without this the merge would drop DaData/ГАР enrichment that
# lived only on a loser row (e.g. house_fias_id backfilled onto the loser but not the keeper).
_CARRY_OVER_IDENTITY_SQL = text(
    """
    UPDATE houses k
    SET
        house_fias_id      = COALESCE(k.house_fias_id, d.house_fias_id),
        cadastral_number   = COALESCE(k.cadastral_number, d.cadastral_number),
        gar_house_guid     = COALESCE(k.gar_house_guid, d.gar_house_guid),
        gar_flat_count     = COALESCE(k.gar_flat_count, d.gar_flat_count),
        gar_matched_at     = COALESCE(k.gar_matched_at, d.gar_matched_at),
        gar_match_method   = COALESCE(k.gar_match_method, d.gar_match_method),
        dadata_qc_geo      = COALESCE(k.dadata_qc_geo, d.dadata_qc_geo),
        dadata_qc_house    = COALESCE(k.dadata_qc_house, d.dadata_qc_house),
        dadata_enriched_at = COALESCE(k.dadata_enriched_at, d.dadata_enriched_at)
    FROM (
        SELECT
            m.keeper_id,
            (array_agg(l.house_fias_id
                ORDER BY (l.geom IS NOT NULL) DESC, l.id ASC)
                FILTER (WHERE l.house_fias_id IS NOT NULL))[1] AS house_fias_id,
            (array_agg(l.cadastral_number
                ORDER BY (l.geom IS NOT NULL) DESC, l.id ASC)
                FILTER (WHERE l.cadastral_number IS NOT NULL))[1] AS cadastral_number,
            (array_agg(l.gar_house_guid
                ORDER BY (l.geom IS NOT NULL) DESC, l.id ASC)
                FILTER (WHERE l.gar_house_guid IS NOT NULL))[1] AS gar_house_guid,
            (array_agg(l.gar_flat_count
                ORDER BY (l.geom IS NOT NULL) DESC, l.id ASC)
                FILTER (WHERE l.gar_flat_count IS NOT NULL))[1] AS gar_flat_count,
            (array_agg(l.gar_matched_at
                ORDER BY (l.geom IS NOT NULL) DESC, l.id ASC)
                FILTER (WHERE l.gar_matched_at IS NOT NULL))[1] AS gar_matched_at,
            (array_agg(l.gar_match_method
                ORDER BY (l.geom IS NOT NULL) DESC, l.id ASC)
                FILTER (WHERE l.gar_match_method IS NOT NULL))[1] AS gar_match_method,
            (array_agg(l.dadata_qc_geo
                ORDER BY (l.geom IS NOT NULL) DESC, l.id ASC)
                FILTER (WHERE l.dadata_qc_geo IS NOT NULL))[1] AS dadata_qc_geo,
            (array_agg(l.dadata_qc_house
                ORDER BY (l.geom IS NOT NULL) DESC, l.id ASC)
                FILTER (WHERE l.dadata_qc_house IS NOT NULL))[1] AS dadata_qc_house,
            (array_agg(l.dadata_enriched_at
                ORDER BY (l.geom IS NOT NULL) DESC, l.id ASC)
                FILTER (WHERE l.dadata_enriched_at IS NOT NULL))[1] AS dadata_enriched_at
        FROM _1772_dup_mapping m
        JOIN houses l ON l.id = m.loser_id
        GROUP BY m.keeper_id
    ) d
    WHERE k.id = d.keeper_id
    """
)


@dataclass
class DedupMergeResult:
    """Counters from one merge pass (also the dry-run preview)."""

    clusters_merged: int = 0  # distinct keepers that absorbed at least one loser
    losers_deleted: int = 0  # duplicate house rows removed (0 on a clean table)
    listings_repointed: int = 0  # listings.house_id_fk moved loser→keeper
    children_deleted: int = 0  # collision/dedup deletions across all UNIQUE children
    children_repointed: int = 0  # survivor child rows moved loser→keeper
    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 _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.

    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()).
    """
    start = time.monotonic()
    result = DedupMergeResult(dry_run=dry_run)

    try:
        # 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)

        if result.losers_deleted == 0:
            # Clean table — both passes empty. Roll back (we only opened temp tables).
            db.rollback()
            result.duration_sec = time.monotonic() - start
            logger.info(
                "merge_duplicate_houses: no duplicate clusters found (no-op) dry_run=%s",
                dry_run,
            )
            return result.to_counters()

        if dry_run:
            db.rollback()
            logger.info(
                "merge_duplicate_houses: DRY-RUN computed clusters=%d losers=%d "
                "listings_repointed=%d children_deleted=%d children_repointed=%d — ROLLED BACK",
                result.clusters_merged,
                result.losers_deleted,
                result.listings_repointed,
                result.children_deleted,
                result.children_repointed,
            )
        else:
            db.commit()
            logger.info(
                "merge_duplicate_houses: COMMITTED clusters=%d losers=%d "
                "listings_repointed=%d children_deleted=%d children_repointed=%d",
                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 kit scheduler (source='house_dedup_merge') via
    product_handlers._job_house_dedup_merge, or manually. Mirrors run_cadastral_geo_match:
    pure internal DB op, finalises scrape_runs (mark_done / mark_failed) with counters.

    Params (default_params jsonb):
        dry_run: compute + return counts then ROLLBACK without writing (default false).
                 The deploy seeds the schedule DISABLED; the orchestrator can flip a single
                 manual run to dry_run=true to preview before enabling.
    """
    from app.services import scrape_runs as runs_mod

    dry_run = bool(params.get("dry_run", False))
    counters: dict[str, int] = {
        "clusters_merged": 0,
        "losers_deleted": 0,
        "listings_repointed": 0,
        "children_deleted": 0,
        "children_repointed": 0,
        "dry_run": int(dry_run),
    }
    try:
        runs_mod.update_heartbeat(db, run_id, counters)
        counters = merge_duplicate_houses(db, dry_run=dry_run)
        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