tradein-mvp/backend/scripts/geocode_deals_from_houses.py

@@ -0,0 +1,591 @@
+"""Backfill deals.lat/lon via a per-street centroid join against houses.
+
+Issue #569, Step 2. The `deals` table (49,791 Rosreestr ДКП sale records) has
+100% NULL lat/lon/geom, so `estimator._fetch_deals()` — which matches via
+`ST_DWithin(geom, point, radius)` — never returns a single deal. The "real
+deals" comparable feature is silently dead.
+
+Approach — **street-centroid join, zero external geocoder calls**:
+  1. Build a per-street centroid map from `houses WHERE geom IS NOT NULL`
+     (~8,600 rows already geocoded): derive a normalized street key from
+     `houses.address`, group by it, compute the centroid as AVG(lat)/AVG(lon).
+  2. For each `deals` row with lat IS NULL: derive the SAME street key from
+     `deals.address` ('Екатеринбург, <street>', street-only — no house number),
+     look up the centroid and `UPDATE deals SET lat, lon, geocode_tried_at=NOW()`.
+
+Street-level precision is acceptable — the estimator search radius is
+1000-2000 m, so all deals on the same street land inside the same comparable
+window regardless of which house we used for the centroid.
+
+Why a street key (bare street name) and not `normalize_address`:
+  `deals.address` carries NO house number and sometimes NO street-type word
+  ('Екатеринбург, Малышева'), while `houses.address` is full and varied
+  ('Свердловская обл., Екатеринбург, ул. Большакова, 17'). `normalize_address`
+  keeps the house number and the type word, so the two sides would never
+  collide. `_street_key` strips city/region prefix, the street-type token AND
+  the house number, leaving just the lowercased street name ('малышева',
+  '8 марта') — the only token both sides reliably share.
+
+The `deals_set_geom_trg` BEFORE INSERT OR UPDATE OF lat, lon trigger
+(002_core_tables.sql) auto-fills `geom` from lat/lon via listings_set_geom(),
+so this script sets lat/lon ONLY — never geom directly.
+
+Resume-safe: only `deals WHERE lat IS NULL` are processed (matches the
+`deals_geocode_pending_idx` partial index from 005_geocode_tracking.sql); a
+successful UPDATE drops the row out of the candidate set on the next run.
+
+Usage:
+    DATABASE_URL=postgresql+psycopg://... \\
+    python -m scripts.geocode_deals_from_houses --dry-run
+
+    # real backfill, default cap 5000 rows/run
+    python -m scripts.geocode_deals_from_houses --batch 2026-05-28
+
+Flags:
+    --dry-run      report coverage projection + top-10 unmatched streets,
+                   no DB writes.
+    --limit N      max deals to process this run (default 5000).
+    --batch LABEL  log label (default `deals_geo_YYYY-MM-DD`).
+"""
+
+from __future__ import annotations
+
+import argparse
+import logging
+import re
+import unicodedata
+from collections import Counter
+from dataclasses import dataclass, field
+from datetime import date
+from pathlib import Path
+
+from sqlalchemy import text
+from sqlalchemy.orm import Session
+
+# Allow running both as `python -m scripts.geocode_deals_from_houses` (preferred,
+# matches the pattern from backfill_houses_dadata.py) and as a stand-alone file.
+try:
+    from app.core.db import SessionLocal  # type: ignore[import-not-found]
+except ImportError:  # pragma: no cover — fallback for adhoc invocation
+    import sys
+
+    sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
+    from app.core.db import SessionLocal
+
+logging.basicConfig(
+    level=logging.INFO,
+    format="%(asctime)s %(levelname)s %(name)s %(message)s",
+)
+logger = logging.getLogger("geocode_deals_from_houses")
+
+# Default per-run cap. The full deals backlog is ~50k; a SAVEPOINT-per-row
+# UPDATE loop over 50k is fine in one pass, but a cap keeps canary runs cheap
+# and lets the caller chunk if they want.
+_DEFAULT_LIMIT = 5000
+
+# Progress log cadence.
+_LOG_EVERY = 1000
+
+# A street key shorter than this is almost certainly a parse failure (an empty
+# string, a stray house number, a one-letter token) — skip it on both the
+# houses (centroid) and deals (lookup) side so garbage never anchors a match.
+_MIN_KEY_LEN = 3
+
+
+# ---------------------------------------------------------------------------
+# Street-key normalization — the crux of match rate
+# ---------------------------------------------------------------------------
+
+# Comma-separated admin chunks to drop wholesale: «Россия», «РФ», a region
+# («Свердловская область» / «...обл.»), a district («... р-н» / «... район»),
+# a city («г. Екатеринбург» / «Екатеринбург»). Each pattern matches a WHOLE
+# comma-segment (anchored ^…$ against the segment) so it can never bite a
+# partial word — segments that don't match are kept verbatim. Order doesn't
+# matter; we test every leading segment until one fails to match.
+_ADMIN_SEGMENT_RES = [
+    re.compile(r"^(?:россия|рф|российская\s+федерация)$", flags=re.UNICODE),
+    re.compile(r"^[а-яё][а-яё\s.-]*\bобл(?:асть|\.)?$", flags=re.UNICODE),
+    re.compile(
+        r"^[а-яё][а-яё\s.-]*\b(?:р-н|район|округ|край|республика)$",
+        flags=re.UNICODE,
+    ),
+    re.compile(r"^(?:г|гор|город)\.?\s+[а-яё][а-яё-]+$", flags=re.UNICODE),
+    re.compile(r"^екатеринбург$", flags=re.UNICODE),
+]
+
+# Street-type token at the START of the street segment. Stripped because
+# deals.address sometimes omits it entirely ('Екатеринбург, Малышева'), so the
+# type word must NOT be part of the key or '<type> малышева' vs 'малышева'
+# would diverge. Hyphenated forms first (longest-match), then short variants.
+# `\b` after the token forbids matching the head of a real name (e.g. the 'ал'
+# of 'алмазная' or the 'пр' of 'пришвина').
+_STREET_TYPE_RE = re.compile(
+    r"^(?:"
+    r"пр-кт|пр-т|пр-д|б-р|кв-л"
+    r"|улица|проспект|переулок|бульвар|шоссе|набережная|проезд|тракт"
+    r"|площадь|микрорайон|тупик|аллея|квартал"
+    r"|ул|пр|пер|наб|пл|мкр|туп"
+    r")(?:\.|\b)\s+",
+    flags=re.UNICODE,
+)
+
+# District / apartment / building tail INSIDE the street segment. Anchored on a
+# left boundary ((?<=^)|(?<=[\s,])) so the keyword is a standalone token, never
+# the middle of a word ('к' must not bite 'катеринбург'). Drops everything from
+# the marker to end-of-segment.
+_DISTRICT_SUFFIX_RE = re.compile(r"\s*[·|].*$", flags=re.UNICODE)
+# Right side requires a DIGIT after the token (`кв 12`, `корп. 2`, `стр 10`) so the
+# alternation can't eat real street names that merely START with these letters
+# («Строителей», «Офицеров», «Корпусная», «Помолова» → would collapse to 'ул.'
+# garbage key → wrong centroid → wrong coords). Caught in pre-push review.
+_APT_SUFFIX_RE = re.compile(
+    r"(?:^|(?<=[\s,]))(?:кв|корп|оф|пом|стр|строение|подъезд)\.?\s*\d.*$",
+    flags=re.UNICODE,
+)
+
+# Whitespace collapse.
+_WS_RE = re.compile(r"\s+")
+
+# A street name that legitimately STARTS with a number followed by a word —
+# '8 марта', '1905 года', '40 лет октября'. We must keep these intact while
+# still stripping a pure house number like '125' or '44-а'.
+_NUMERIC_STREET_RE = re.compile(r"^\d+\s+[а-яё]", flags=re.UNICODE)
+
+
+def _strip_house_tail(segment: str) -> str:
+    """Remove a trailing house-number / building token from a street segment.
+
+    'малышева 125'      → 'малышева'
+    'большакова, 17'    → 'большакова'
+    'крауля 48/2'       → 'крауля'
+    'репина 75/2 стр.'  → 'репина'   (apt suffix already stripped upstream)
+    '8 марта 100'       → '8 марта'  (leading numeric street preserved)
+    '8 марта'           → '8 марта'
+    'малышева'          → 'малышева' (no number → unchanged)
+
+    Strategy: walk tokens left→right, keeping tokens until we hit one that
+    starts with a digit AND is not the leading numeric-street token. A token
+    is "house-like" if it starts with a digit; the only digit-leading token we
+    keep is position 0 of a recognized numeric street name ('8 марта').
+    """
+    seg = segment.replace(",", " ")
+    seg = _WS_RE.sub(" ", seg).strip()
+    if not seg:
+        return ""
+    tokens = seg.split(" ")
+    keep: list[str] = []
+    numeric_street = bool(_NUMERIC_STREET_RE.match(seg))
+    for i, tok in enumerate(tokens):
+        if tok and tok[0].isdigit():
+            # First token of a numeric street ('8' in '8 марта') is kept; any
+            # later digit-leading token is a house number → stop here.
+            if i == 0 and numeric_street:
+                keep.append(tok)
+                continue
+            break
+        keep.append(tok)
+    return " ".join(keep).strip()
+
+
+def _street_key(address: str | None) -> str:
+    """Reduce any address to a bare-street-name key for the centroid join.
+
+    Both sides must produce the SAME key or the join under-matches:
+      'Екатеринбург, ул. Малышева, 125'   → 'малышева'
+      'г Екатеринбург, улица Малышева'     → 'малышева'
+      'Екатеринбург, Малышева'             → 'малышева'
+      'Свердловская обл., Екатеринбург, ул. Большакова, 17' → 'большакова'
+      'улица Яскина, 12 · р-н Октябрьский' → 'яскина'
+      'г. Екатеринбург, проспект Ленина, 50' → 'ленина'
+      'Екатеринбург, ул. 8 Марта, 100'     → '8 марта'
+
+    Steps:
+      1. NFC normalize, lowercase, ё→е (deals/houses differ on ё usage).
+      2. Drop a trailing district marker (' · р-н ...', ' | ...').
+      3. Split on commas; drop leading segments that are admin chunks
+         (Россия / region / district / город / Екатеринбург). The first
+         non-admin segment is the street segment.
+      4. Drop apartment/corpus/строение noise inside that segment.
+      5. Strip the street-type token at the start (ул/улица/проспект/...).
+      6. Strip the trailing house number, preserving numeric street names.
+      7. Collapse whitespace.
+
+    Returns '' when nothing usable remains (caller filters by _MIN_KEY_LEN).
+    """
+    if not address:
+        return ""
+    s = unicodedata.normalize("NFC", address).lower().replace("ё", "е")
+    s = _WS_RE.sub(" ", s).strip()
+    if not s:
+        return ""
+
+    # 2. Drop trailing district marker (' · Октябрьский', ' | ...').
+    s = _DISTRICT_SUFFIX_RE.sub("", s)
+
+    # 3. Split on commas, drop leading admin segments. Each segment is matched
+    # whole, so a non-admin street segment is never partially eaten.
+    segments = [seg.strip() for seg in s.split(",") if seg.strip()]
+    street_seg = ""
+    for seg in segments:
+        if any(rx.match(seg) for rx in _ADMIN_SEGMENT_RES):
+            continue
+        street_seg = seg
+        break
+    if not street_seg:
+        return ""
+
+    # 4. Drop apartment/corpus/строение noise inside the street segment.
+    street_seg = _APT_SUFFIX_RE.sub("", street_seg).strip()
+
+    # 5. Strip the street-type token if present.
+    street_seg = _STREET_TYPE_RE.sub("", street_seg).strip()
+
+    # 6. Strip trailing house number (keep '8 марта' style numeric streets).
+    street_seg = _strip_house_tail(street_seg)
+
+    return _WS_RE.sub(" ", street_seg).strip()
+
+
+# ---------------------------------------------------------------------------
+# Domain types
+# ---------------------------------------------------------------------------
+
+
+@dataclass
+class Centroid:
+    """One street's centroid, averaged over all geocoded houses on it."""
+
+    lat: float
+    lon: float
+    house_count: int
+
+
+@dataclass
+class DealRow:
+    """Minimal deals fields needed for the centroid lookup."""
+
+    id: int
+    address: str | None
+
+
+@dataclass
+class Stats:
+    """Final-summary counters."""
+
+    processed: int = 0
+    geocoded: int = 0
+    no_street_match: int = 0
+    failed: int = 0
+    # street_key → count of deals that had no house centroid (dry-run report).
+    unmatched_streets: Counter[str] = field(default_factory=Counter)
+
+
+# ---------------------------------------------------------------------------
+# Source queries
+# ---------------------------------------------------------------------------
+
+
+def _build_centroid_map(db: Session) -> dict[str, Centroid]:
+    """Per-street centroid from houses WHERE geom IS NOT NULL.
+
+    We read raw (address, lat, lon) and aggregate in Python so the street-key
+    derivation is the SAME code path as the deals side — pushing it into SQL
+    would require duplicating the regex logic in plpgsql and risk drift.
+    8,600 rows is trivial to hold in memory.
+    """
+    rows = db.execute(
+        text(
+            "SELECT address, lat, lon "
+            "FROM houses "
+            "WHERE geom IS NOT NULL "
+            "  AND lat IS NOT NULL "
+            "  AND lon IS NOT NULL "
+            "  AND address IS NOT NULL "
+            "  AND length(trim(address)) > 0"
+        )
+    ).mappings().all()
+
+    # street_key → running [lat_sum, lon_sum, n]
+    acc: dict[str, list[float]] = {}
+    for r in rows:
+        key = _street_key(r["address"])
+        if len(key) < _MIN_KEY_LEN:
+            continue
+        bucket = acc.setdefault(key, [0.0, 0.0, 0.0])
+        bucket[0] += float(r["lat"])
+        bucket[1] += float(r["lon"])
+        bucket[2] += 1.0
+
+    return {
+        key: Centroid(lat=lat_sum / n, lon=lon_sum / n, house_count=int(n))
+        for key, (lat_sum, lon_sum, n) in acc.items()
+    }
+
+
+def _select_deals_without_coords(db: Session, limit: int) -> list[DealRow]:
+    """deals needing coords (lat IS NULL) — resume-safe candidate set.
+
+    Matches `deals_geocode_pending_idx` (WHERE lat IS NULL). A successful
+    UPDATE sets lat NOT NULL, dropping the row out on the next run.
+    """
+    rows = db.execute(
+        text(
+            "SELECT id, address "
+            "FROM deals "
+            "WHERE lat IS NULL "
+            "  AND address IS NOT NULL "
+            "  AND length(trim(address)) > 0 "
+            "ORDER BY id "
+            "LIMIT CAST(:lim AS int)"
+        ),
+        {"lim": limit},
+    ).mappings().all()
+    return [DealRow(id=r["id"], address=r["address"]) for r in rows]
+
+
+# ---------------------------------------------------------------------------
+# DB writer
+# ---------------------------------------------------------------------------
+
+
+def _update_deal_coords(db: Session, *, deal_id: int, lat: float, lon: float) -> None:
+    """UPDATE deals SET lat/lon + geocode_tried_at=NOW(); geom auto-fills.
+
+    The `deals_set_geom_trg` BEFORE UPDATE OF lat, lon trigger
+    (002_core_tables.sql, reuses listings_set_geom()) populates geom from the
+    new lat/lon, so we never touch geom here. Setting geocode_tried_at marks
+    the row processed for the partial index / future cron passes.
+    """
+    db.execute(
+        text(
+            "UPDATE deals "
+            "   SET lat = CAST(:lat AS double precision), "
+            "       lon = CAST(:lon AS double precision), "
+            "       geocode_tried_at = NOW() "
+            " WHERE id = CAST(:id AS bigint)"
+        ),
+        {"id": deal_id, "lat": lat, "lon": lon},
+    )
+
+
+# ---------------------------------------------------------------------------
+# Main loop
+# ---------------------------------------------------------------------------
+
+
+def _run_backfill(
+    db: Session,
+    deals: list[DealRow],
+    centroids: dict[str, Centroid],
+    *,
+    batch: str,
+    dry_run: bool,
+) -> Stats:
+    """For each deal, look up its street centroid and UPDATE lat/lon.
+
+    Per-row SAVEPOINT (`db.begin_nested()`) so one bad UPDATE can't abort the
+    batch (backend.md SAVEPOINT rule). Per-row commit on success → a crash
+    mid-run leaves already-geocoded rows persisted, and resume picks up the
+    rest (lat IS NULL filter).
+    """
+    stats = Stats()
+
+    for i, deal in enumerate(deals, start=1):
+        key = _street_key(deal.address)
+        centroid = centroids.get(key) if len(key) >= _MIN_KEY_LEN else None
+
+        if centroid is None:
+            stats.no_street_match += 1
+            # Track the raw key (or a sentinel) so the dry-run report can show
+            # which streets we're missing. Empty key → '<no-street-parsed>'.
+            stats.unmatched_streets[key or "<no-street-parsed>"] += 1
+            stats.processed += 1
+            if dry_run and i % _LOG_EVERY == 0:
+                logger.info(
+                    "DRY-RUN deal_id=%s addr=%r key=%r → no centroid",
+                    deal.id,
+                    (deal.address or "")[:60],
+                    key,
+                )
+            continue
+
+        if dry_run:
+            stats.geocoded += 1
+            stats.processed += 1
+        else:
+            try:
+                with db.begin_nested():
+                    _update_deal_coords(
+                        db, deal_id=deal.id, lat=centroid.lat, lon=centroid.lon
+                    )
+                # Per-row commit so resume picks up exactly where we crashed.
+                db.commit()
+                stats.geocoded += 1
+                stats.processed += 1
+            except Exception as exc:  # defensive — isolate one bad UPDATE
+                db.rollback()
+                stats.failed += 1
+                logger.warning("db_write failed for deal_id=%s: %s", deal.id, exc)
+
+        if i % _LOG_EVERY == 0:
+            logger.info(
+                "batch=%s progress %d/%d geocoded=%d no_match=%d failed=%d",
+                batch,
+                i,
+                len(deals),
+                stats.geocoded,
+                stats.no_street_match,
+                stats.failed,
+            )
+
+    return stats
+
+
+# ---------------------------------------------------------------------------
+# Dry-run reporting
+# ---------------------------------------------------------------------------
+
+
+def _report_dry_run(
+    stats: Stats,
+    centroids: dict[str, Centroid],
+    *,
+    total_deals_null: int,
+    candidates: int,
+) -> None:
+    """Log coverage projection + the top-10 unmatched deal streets.
+
+    `candidates` is how many lat-IS-NULL deals we actually scanned this run
+    (capped by --limit); `total_deals_null` is the full backlog so the
+    projected % extrapolates honestly when --limit < backlog.
+    """
+    distinct_streets = len(centroids)
+    matched = stats.geocoded
+    scanned = candidates
+
+    # Coverage on the scanned slice, then projected onto the full backlog.
+    match_rate = (matched / scanned) if scanned else 0.0
+    projected = round(match_rate * total_deals_null)
+
+    logger.info("─" * 60)
+    logger.info("DRY-RUN SUMMARY (no DB writes)")
+    logger.info("distinct streets with a house centroid: %d", distinct_streets)
+    logger.info(
+        "deals scanned this run (lat IS NULL, capped by --limit): %d", scanned
+    )
+    logger.info("deals matched to a centroid: %d", matched)
+    logger.info("deals with no street match: %d", stats.no_street_match)
+    logger.info("match rate on scanned slice: %.1f%%", match_rate * 100.0)
+    logger.info(
+        "full lat-IS-NULL backlog: %d → projected matched ≈ %d (%.1f%%)",
+        total_deals_null,
+        projected,
+        match_rate * 100.0,
+    )
+    logger.info("top-10 unmatched deal streets (by row count):")
+    for street, cnt in stats.unmatched_streets.most_common(10):
+        logger.info("    %6d  %s", cnt, street)
+    logger.info("─" * 60)
+
+
+# ---------------------------------------------------------------------------
+# CLI
+# ---------------------------------------------------------------------------
+
+
+def _count_deals_null(db: Session) -> int:
+    """Full count of deals WHERE lat IS NULL — denominator for projection."""
+    row = db.execute(
+        text(
+            "SELECT count(*) AS n FROM deals "
+            "WHERE lat IS NULL AND address IS NOT NULL AND length(trim(address)) > 0"
+        )
+    ).first()
+    return int(row[0]) if row else 0
+
+
+def _parse_args(argv: list[str] | None = None) -> argparse.Namespace:
+    """argparse setup, factored out for testability."""
+    p = argparse.ArgumentParser(
+        description=(
+            "Issue #569 Step 2 — backfill deals.lat/lon from per-street house "
+            "centroids (no external geocoder)."
+        ),
+    )
+    p.add_argument(
+        "--limit",
+        type=int,
+        default=_DEFAULT_LIMIT,
+        help=f"Max deals to process this run (default {_DEFAULT_LIMIT}).",
+    )
+    p.add_argument(
+        "--batch",
+        default=f"deals_geo_{date.today().isoformat()}",
+        help="Log batch label (does not affect DB filters — logs only).",
+    )
+    p.add_argument(
+        "--dry-run",
+        action="store_true",
+        help=(
+            "Report distinct house streets, projected coverage %% of the "
+            "lat-IS-NULL backlog, and the top-10 unmatched deal streets. "
+            "No DB writes."
+        ),
+    )
+    return p.parse_args(argv)
+
+
+def main(argv: list[str] | None = None) -> int:
+    """CLI entry point. Returns the number of deals geocoded this run."""
+    args = _parse_args(argv)
+    logger.info(
+        "starting batch=%s limit=%s dry_run=%s",
+        args.batch,
+        args.limit,
+        args.dry_run,
+    )
+
+    db = SessionLocal()
+    try:
+        centroids = _build_centroid_map(db)
+        logger.info("built centroid map: %d distinct streets", len(centroids))
+        if not centroids:
+            logger.warning(
+                "no house centroids — houses table has no geocoded rows; nothing to do"
+            )
+            return 0
+
+        deals = _select_deals_without_coords(db, args.limit)
+        logger.info("loaded deals without coords: %d", len(deals))
+        if not deals:
+            logger.info("nothing to do — no deals with lat IS NULL and an address")
+            return 0
+
+        stats = _run_backfill(
+            db, deals, centroids, batch=args.batch, dry_run=args.dry_run
+        )
+
+        if args.dry_run:
+            total_null = _count_deals_null(db)
+            _report_dry_run(
+                stats,
+                centroids,
+                total_deals_null=total_null,
+                candidates=len(deals),
+            )
+
+        logger.info(
+            "done: batch=%s processed=%d geocoded=%d no_street_match=%d failed=%d",
+            args.batch,
+            stats.processed,
+            stats.geocoded,
+            stats.no_street_match,
+            stats.failed,
+        )
+        return stats.geocoded
+    finally:
+        db.close()
+
+
+if __name__ == "__main__":  # pragma: no cover
+    raise SystemExit(0 if main() >= 0 else 1)

tradein-mvp/backend/tests/scripts/test_geocode_deals_from_houses.py

@@ -0,0 +1,440 @@
+"""Unit tests for issue #569 Step 2 — geocode_deals_from_houses.py.
+
+Coverage:
+    - _street_key normalization: the three spec examples collapse to one key,
+      plus boundary cases (numeric streets '8 марта', names that start with a
+      type-token substring like 'алмазная', district/apt/region stripping).
+    - _build_centroid_map: AVG(lat)/AVG(lon) over multiple houses on one street.
+    - _run_backfill happy path: a deal whose street matches → UPDATE with the
+      centroid lat/lon, geocoded counter bumps.
+    - _run_backfill miss: a deal whose street has no centroid → no UPDATE,
+      counted as no_street_match + tracked for the dry-run report.
+    - --dry-run: no UPDATE / no commit, counters still move.
+    - per-row SAVEPOINT isolates a failing UPDATE.
+    - main() wires SessionLocal, respects --limit, and returns geocoded count.
+
+No real Postgres. Session is a MagicMock that routes SELECT side-effects by
+SQL substring and records UPDATE binds (same pattern as
+test_backfill_houses_dadata.py).
+"""
+
+from __future__ import annotations
+
+import os
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+# Settings needs a DSN at import time — set a dummy before any app.* import.
+os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost/test_db")
+
+from scripts.geocode_deals_from_houses import (
+    Centroid,
+    DealRow,
+    Stats,
+    _build_centroid_map,
+    _run_backfill,
+    _street_key,
+    _update_deal_coords,
+    main,
+)
+
+# ---------------------------------------------------------------------------
+# Mock-DB helper
+# ---------------------------------------------------------------------------
+
+
+def _make_db_mock(
+    *,
+    house_rows: list[dict] | None = None,
+    deal_rows: list[dict] | None = None,
+    null_count: int = 0,
+) -> tuple[MagicMock, list[dict]]:
+    """MagicMock Session that:
+
+    - returns `house_rows` for the centroid SELECT (`FROM houses`)
+    - returns `deal_rows` for the candidate SELECT (`FROM deals ... lat IS NULL`)
+    - returns `null_count` for the `count(*)` SELECT
+    - records UPDATE binds into the returned `updated` list
+    - supports `db.begin_nested()` as a context manager
+    """
+    house_rows = house_rows or []
+    deal_rows = deal_rows or []
+    updated: list[dict] = []
+
+    db = MagicMock()
+    db.begin_nested.return_value.__enter__ = lambda self: self
+    db.begin_nested.return_value.__exit__ = lambda self, *a: False
+
+    def execute_side_effect(sql, params=None):
+        sql_str = str(sql)
+        result = MagicMock()
+        if "UPDATE deals" in sql_str:
+            updated.append(dict(params) if params else {})
+            return result
+        if "count(*)" in sql_str:
+            result.first.return_value = (null_count,)
+            return result
+        if "FROM houses" in sql_str:
+            result.mappings.return_value.all.return_value = house_rows
+            return result
+        if "FROM deals" in sql_str:
+            lim = params.get("lim", len(deal_rows)) if params else len(deal_rows)
+            result.mappings.return_value.all.return_value = deal_rows[:lim]
+            return result
+        return result
+
+    db.execute.side_effect = execute_side_effect
+    db.commit = MagicMock()
+    db.rollback = MagicMock()
+    db.close = MagicMock()
+    return db, updated
+
+
+# ---------------------------------------------------------------------------
+# _street_key — normalization (the crux)
+# ---------------------------------------------------------------------------
+
+
+def test_street_key_three_spec_variants_collapse_to_same_key():
+    """The spec's three Малышева variants must produce one identical key."""
+    a = _street_key("Екатеринбург, ул. Малышева, 125")
+    b = _street_key("г Екатеринбург, улица Малышева")
+    c = _street_key("Екатеринбург, Малышева")
+    assert a == b == c == "малышева"
+
+
+def test_street_key_strips_region_and_house_number():
+    assert (
+        _street_key("Свердловская обл., Екатеринбург, ул. Большакова, 17")
+        == "большакова"
+    )
+
+
+def test_street_key_strips_district_marker():
+    assert _street_key("улица Яскина, 12 · р-н Октябрьский") == "яскина"
+
+
+def test_street_key_handles_prospekt():
+    assert _street_key("г. Екатеринбург, проспект Ленина, 50") == "ленина"
+
+
+def test_street_key_preserves_numeric_street_name():
+    """'8 марта' is a real street — the leading number must survive while the
+    trailing house number is stripped."""
+    assert _street_key("Екатеринбург, ул. 8 Марта, 100") == "8 марта"
+    assert _street_key("Екатеринбург, 8 Марта") == "8 марта"
+
+
+def test_street_key_preserves_multiword_numeric_street():
+    assert _street_key("Екатеринбург, ул. 40 лет Октября") == "40 лет октября"
+
+
+def test_street_key_does_not_eat_name_starting_like_type_token():
+    """'алмазная' must NOT lose 'ал', 'пришвина' must NOT lose 'пр' — the
+    street-type strip is \\b-bounded."""
+    assert _street_key("ул. Алмазная, 7") == "алмазная"
+    assert _street_key("ул. Пришвина, 3") == "пришвина"
+
+
+def test_street_key_strips_korpus_and_kv_suffix():
+    assert _street_key("Екатеринбург, ул. Крауля, 48, корп. 2") == "крауля"
+    assert (
+        _street_key("РФ, Свердловская обл., Екатеринбург, ул. Ленина, 5, кв. 12")
+        == "ленина"
+    )
+
+
+def test_street_key_does_not_eat_names_starting_like_apt_suffix():
+    """Regression (pre-push review): un-anchored _APT_SUFFIX_RE collapsed real
+    streets starting with кв/корп/оф/пом/стр («Строителей», «Офицеров»,
+    «Корпусная», «Помолова») to the garbage key 'ул.' → wrong centroid → wrong
+    coords. Suffix strip now requires a DIGIT after the token."""
+    assert _street_key("Екатеринбург, ул. Строителей, 5") == "строителей"
+    assert _street_key("Екатеринбург, ул. Офицеров") == "офицеров"
+    assert _street_key("Екатеринбург, ул. Корпусная, 14") == "корпусная"
+    assert _street_key("Екатеринбург, ул. Помолова, 3") == "помолова"
+    # deals-form (no house number) must still resolve to the street, not 'ул.'
+    assert _street_key("Екатеринбург, Стрелочников") == "стрелочников"
+
+
+def test_street_key_normalizes_yo_to_e():
+    """Рабочей Молодёжи (ё) and Молодежи (е) must collapse — sources differ."""
+    assert _street_key("Екатеринбург, наб. Рабочей Молодёжи, 2") == "рабочей молодежи"
+    assert _street_key("Екатеринбург, Рабочей Молодежи") == "рабочей молодежи"
+
+
+def test_street_key_house_prefix_dom():
+    assert _street_key("Екатеринбург, пр. Ленина, д. 5") == "ленина"
+
+
+def test_street_key_none_and_empty_and_city_only():
+    assert _street_key(None) == ""
+    assert _street_key("") == ""
+    # City alone has no street segment → empty key (caller treats as no-match).
+    assert _street_key("Екатеринбург") == ""
+
+
+# ---------------------------------------------------------------------------
+# _build_centroid_map — AVG over multiple houses on a street
+# ---------------------------------------------------------------------------
+
+
+def test_build_centroid_map_averages_houses_on_one_street():
+    """Two houses on Малышева → centroid is the mean of their coords."""
+    house_rows = [
+        {"address": "Екатеринбург, ул. Малышева, 10", "lat": 56.80, "lon": 60.50},
+        {"address": "Екатеринбург, ул. Малышева, 20", "lat": 56.90, "lon": 60.70},
+    ]
+    db, _ = _make_db_mock(house_rows=house_rows)
+
+    centroids = _build_centroid_map(db)
+
+    assert set(centroids) == {"малышева"}
+    c = centroids["малышева"]
+    assert c.lat == pytest.approx(56.85)
+    assert c.lon == pytest.approx(60.60)
+    assert c.house_count == 2
+
+
+def test_build_centroid_map_groups_distinct_streets():
+    house_rows = [
+        {"address": "Екатеринбург, ул. Малышева, 10", "lat": 56.80, "lon": 60.50},
+        {"address": "г Екатеринбург, Малышева", "lat": 56.82, "lon": 60.54},
+        {"address": "Екатеринбург, проспект Ленина, 5", "lat": 56.84, "lon": 60.60},
+    ]
+    db, _ = _make_db_mock(house_rows=house_rows)
+
+    centroids = _build_centroid_map(db)
+
+    assert set(centroids) == {"малышева", "ленина"}
+    assert centroids["малышева"].house_count == 2
+    assert centroids["ленина"].house_count == 1
+    # Малышева centroid = mean of the two Малышева rows.
+    assert centroids["малышева"].lat == pytest.approx(56.81)
+    assert centroids["ленина"].lat == pytest.approx(56.84)
+
+
+def test_build_centroid_map_skips_unparseable_address():
+    """A house whose address yields too-short a key is dropped, not crashed."""
+    house_rows = [
+        {"address": "Екатеринбург", "lat": 56.8, "lon": 60.6},  # city only → ''
+        {"address": "Екатеринбург, ул. Малышева, 1", "lat": 56.84, "lon": 60.6},
+    ]
+    db, _ = _make_db_mock(house_rows=house_rows)
+
+    centroids = _build_centroid_map(db)
+    assert set(centroids) == {"малышева"}
+
+
+# ---------------------------------------------------------------------------
+# _run_backfill — happy path: street match → UPDATE with centroid coords
+# ---------------------------------------------------------------------------
+
+
+def test_run_backfill_matched_street_issues_update_with_centroid():
+    deal = DealRow(id=42, address="Екатеринбург, ул. Малышева, 125")
+    centroids = {"малышева": Centroid(lat=56.838, lon=60.586, house_count=3)}
+    db, updated = _make_db_mock()
+
+    stats = _run_backfill(db, [deal], centroids, batch="b1", dry_run=False)
+
+    assert stats.geocoded == 1
+    assert stats.no_street_match == 0
+    assert stats.failed == 0
+    assert stats.processed == 1
+    assert len(updated) == 1
+    assert updated[0]["id"] == 42
+    assert updated[0]["lat"] == 56.838
+    assert updated[0]["lon"] == 60.586
+    assert db.commit.call_count == 1
+
+
+def test_run_backfill_deal_with_only_street_name_matches():
+    """Deal address with no house number / no type word still matches."""
+    deal = DealRow(id=7, address="Екатеринбург, Малышева")
+    centroids = {"малышева": Centroid(lat=56.8, lon=60.5, house_count=1)}
+    db, updated = _make_db_mock()
+
+    stats = _run_backfill(db, [deal], centroids, batch="b", dry_run=False)
+
+    assert stats.geocoded == 1
+    assert len(updated) == 1
+    assert updated[0]["lat"] == 56.8
+
+
+# ---------------------------------------------------------------------------
+# _run_backfill — miss: no centroid → no UPDATE, counted as no_street_match
+# ---------------------------------------------------------------------------
+
+
+def test_run_backfill_unmatched_street_no_update():
+    deal = DealRow(id=99, address="Екатеринбург, ул. Несуществующая, 1")
+    centroids = {"малышева": Centroid(lat=56.8, lon=60.5, house_count=1)}
+    db, updated = _make_db_mock()
+
+    stats = _run_backfill(db, [deal], centroids, batch="b", dry_run=False)
+
+    assert stats.geocoded == 0
+    assert stats.no_street_match == 1
+    assert stats.processed == 1
+    assert updated == []
+    assert db.commit.call_count == 0
+    # Unmatched street tracked for the dry-run report.
+    assert stats.unmatched_streets["несуществующая"] == 1
+
+
+def test_run_backfill_unparseable_deal_tracked_as_sentinel():
+    """A deal whose address yields an empty key is a no-match under a sentinel."""
+    deal = DealRow(id=5, address="Екатеринбург")
+    centroids = {"малышева": Centroid(lat=56.8, lon=60.5, house_count=1)}
+    db, updated = _make_db_mock()
+
+    stats = _run_backfill(db, [deal], centroids, batch="b", dry_run=False)
+
+    assert stats.no_street_match == 1
+    assert updated == []
+    assert stats.unmatched_streets["<no-street-parsed>"] == 1
+
+
+# ---------------------------------------------------------------------------
+# --dry-run — no DB writes
+# ---------------------------------------------------------------------------
+
+
+def test_run_backfill_dry_run_issues_no_update():
+    deal = DealRow(id=1, address="Екатеринбург, ул. Малышева, 1")
+    centroids = {"малышева": Centroid(lat=56.8, lon=60.5, house_count=1)}
+    db, updated = _make_db_mock()
+
+    stats = _run_backfill(db, [deal], centroids, batch="dry", dry_run=True)
+
+    assert stats.geocoded == 1
+    assert stats.processed == 1
+    assert updated == []  # no UPDATE
+    assert db.commit.call_count == 0  # no commit
+
+
+# ---------------------------------------------------------------------------
+# per-row SAVEPOINT — one bad UPDATE doesn't abort the batch
+# ---------------------------------------------------------------------------
+
+
+def test_run_backfill_db_write_failure_isolated_to_row():
+    deals = [
+        DealRow(id=10, address="Екатеринбург, ул. Малышева, 1"),
+        DealRow(id=11, address="Екатеринбург, ул. Малышева, 2"),
+    ]
+    centroids = {"малышева": Centroid(lat=56.8, lon=60.5, house_count=1)}
+    db, updated = _make_db_mock()
+
+    # Make the FIRST UPDATE raise, the rest succeed.
+    call = {"n": 0}
+    real_side_effect = db.execute.side_effect
+
+    def failing_execute(sql, params=None):
+        if "UPDATE deals" in str(sql):
+            call["n"] += 1
+            if call["n"] == 1:
+                raise RuntimeError("constraint blew up")
+        return real_side_effect(sql, params)
+
+    db.execute.side_effect = failing_execute
+
+    stats = _run_backfill(db, deals, centroids, batch="b", dry_run=False)
+
+    assert stats.failed == 1
+    assert stats.geocoded == 1
+    assert db.rollback.call_count == 1
+    # Only the second deal's UPDATE was recorded.
+    assert [u["id"] for u in updated] == [11]
+
+
+# ---------------------------------------------------------------------------
+# _update_deal_coords — bind shape + geom NOT set manually
+# ---------------------------------------------------------------------------
+
+
+def test_update_deal_coords_sets_lat_lon_tried_at_not_geom():
+    db = MagicMock()
+    _update_deal_coords(db, deal_id=3, lat=56.1, lon=60.2)
+    args, _kw = db.execute.call_args
+    sql_str = str(args[0])
+    binds = args[1]
+    assert "UPDATE deals" in sql_str
+    assert "geocode_tried_at = NOW()" in sql_str
+    # geom must NOT be set manually — the deals_set_geom_trg trigger fills it.
+    assert "geom" not in sql_str
+    assert binds == {"id": 3, "lat": 56.1, "lon": 60.2}
+
+
+# ---------------------------------------------------------------------------
+# main() — wiring, --limit, return value
+# ---------------------------------------------------------------------------
+
+
+def test_main_respects_limit_and_returns_geocoded():
+    house_rows = [
+        {"address": "Екатеринбург, ул. Малышева, 1", "lat": 56.80, "lon": 60.50},
+        {"address": "Екатеринбург, ул. Малышева, 2", "lat": 56.90, "lon": 60.70},
+    ]
+    deal_rows = [
+        {"id": 1, "address": "Екатеринбург, Малышева"},
+        {"id": 2, "address": "Екатеринбург, ул. Малышева, 9"},
+        {"id": 3, "address": "Екатеринбург, ул. Малышева, 99"},
+    ]
+    db, updated = _make_db_mock(house_rows=house_rows, deal_rows=deal_rows)
+
+    with patch("scripts.geocode_deals_from_houses.SessionLocal", return_value=db):
+        n = main(["--limit", "1", "--batch", "test_limit"])
+
+    assert n == 1
+    assert len(updated) == 1
+    assert updated[0]["id"] == 1
+    # Centroid used = mean of the two house rows.
+    assert updated[0]["lat"] == pytest.approx(56.85)
+    assert updated[0]["lon"] == pytest.approx(60.60)
+
+
+def test_main_dry_run_writes_nothing():
+    house_rows = [
+        {"address": "Екатеринбург, ул. Малышева, 1", "lat": 56.80, "lon": 60.50},
+    ]
+    deal_rows = [
+        {"id": 1, "address": "Екатеринбург, Малышева"},
+        {"id": 2, "address": "Екатеринбург, ул. Неизвестная, 5"},
+    ]
+    db, updated = _make_db_mock(house_rows=house_rows, deal_rows=deal_rows, null_count=2)
+
+    with patch("scripts.geocode_deals_from_houses.SessionLocal", return_value=db):
+        n = main(["--dry-run"])
+
+    # dry-run returns geocoded count (would-match), writes nothing.
+    assert n == 1
+    assert updated == []
+    assert db.commit.call_count == 0
+
+
+def test_main_no_centroids_returns_zero():
+    """Empty houses table → nothing to join against → graceful 0."""
+    db, updated = _make_db_mock(house_rows=[], deal_rows=[{"id": 1, "address": "x"}])
+
+    with patch("scripts.geocode_deals_from_houses.SessionLocal", return_value=db):
+        n = main([])
+
+    assert n == 0
+    assert updated == []
+
+
+# ---------------------------------------------------------------------------
+# Stats dataclass
+# ---------------------------------------------------------------------------
+
+
+def test_stats_unmatched_streets_counter_defaults_empty():
+    s = Stats()
+    assert s.processed == 0
+    assert s.geocoded == 0
+    assert s.no_street_match == 0
+    assert s.failed == 0
+    assert s.unmatched_streets.most_common(3) == []