tradein-mvp/backend/scripts/README.md

@@ -0,0 +1,116 @@
+# tradein-mvp/backend/scripts/
+
+Ops scripts that touch the production database directly. Run via `python -m
+scripts.<name>` from the `backend/` working directory after `uv sync`.
+
+All scripts are idempotent / resumable where they write — re-running the same
+`--batch` label skips already-processed rows (UNIQUE constraints in target
+tables). Failures inside a per-row loop never roll back the outer transaction;
+each row is wrapped in a SAVEPOINT (`db.begin_nested()`) per `.claude/rules/backend.md`.
+
+---
+
+## Address audit + backfill (issue #582)
+
+End-to-end address quality pipeline. Three scripts, two helpers, two SQL files.
+
+### `audit_address_mismatch.py` — Phase 1 baseline (PR #583)
+
+Stratified-sample audit (200 EKB houses) comparing `houses.address` vs
+Yandex Geocoder reverse lookup. Writes one row per house into
+`address_mismatch_audit` with the snapped point + canonical address + distance.
+
+```bash
+DATABASE_URL=postgresql+psycopg://... \
+YANDEX_GEOCODER_API_KEY=... \
+uv run python -m scripts.audit_address_mismatch \
+    --batch 2026-05-25_run1 \
+    --limit-per-district 25
+```
+
+Mode `auto` picks API if the key is set, otherwise Playwright (CAPTCHA-aware,
+4-7s sleep between calls). API tier free is 25k req/day → 200-row sample
+takes ~10s with no quota concern.
+
+Report:
+
+```bash
+psql "$DATABASE_URL" -v batch='2026-05-25_run1' \
+    -f scripts/address_audit_report.sql
+```
+
+### `backfill_house_coords.py` — Phase 2-3 (PR for #582)
+
+Two modes (`--audit-only` flag switches between them):
+
+**Backfill (default)** — forward-geocode `houses.address` for the ~4141 rows
+WHERE `lat IS NULL OR lon IS NULL`. Only writes back if Yandex returns
+`precision='exact'` or `'number'` (skips street-only / locality matches).
+Each processed row gets an `address_mismatch_audit` entry with status
+`backfill` / `imprecise` / `no_match` / `error`.
+
+```bash
+DATABASE_URL=postgresql+psycopg://... \
+YANDEX_GEOCODER_API_KEY=... \
+uv run python -m scripts.backfill_house_coords \
+    --batch 2026-05-27_backfill
+```
+
+Expected duration (~4141 rows, 50ms between calls, ~250ms RTT per request):
+20-25 min. Expected output split (rough baseline from Phase 1 numbers):
+
+| Status      | Approx rows | What it means                                       |
+|-------------|-------------|-----------------------------------------------------|
+| `backfill`  | ~3.3k–3.7k  | UPDATE landed, lat/lon now populated                |
+| `imprecise` | ~300–500    | Match returned but precision too low — needs review |
+| `no_match`  | ~100–300    | Yandex couldn't resolve; address probably mangled   |
+| `error`     | <50         | HTTP errors / timeouts — re-run picks them up       |
+
+**Audit-only** — reverse-geocode the ~4452 houses WITH coords, write
+audit rows with status `ok` (≤50m) / `mismatch` (>50m) / `no_match` / `error`.
+Does NOT modify the `houses` table.
+
+```bash
+uv run python -m scripts.backfill_house_coords \
+    --batch 2026-05-27_audit --audit-only
+```
+
+Combined budget for both phases (~8.6k requests) is well under the 25k/day
+Geocoder free tier.
+
+### Common ops
+
+Canary first — run with `--limit 100` and inspect the audit table before
+letting the full job loose:
+
+```bash
+uv run python -m scripts.backfill_house_coords \
+    --batch canary_$(date +%F) --limit 100
+psql "$DATABASE_URL" -c "
+  SELECT audit_status, COUNT(*)
+    FROM address_mismatch_audit
+   WHERE audit_batch = 'canary_$(date +%F)'
+   GROUP BY audit_status;
+"
+```
+
+Resume after crash / quota hit — same `--batch` label, the UNIQUE
+`(house_id, audit_batch)` index skips finished rows:
+
+```bash
+uv run python -m scripts.backfill_house_coords --batch 2026-05-27_backfill
+# ... interruption ...
+uv run python -m scripts.backfill_house_coords --batch 2026-05-27_backfill
+# logs: "resuming batch 2026-05-27_backfill: N rows already processed"
+```
+
+### Helpers (not entry points)
+
+- `_yandex_reverse.py` — `forward_via_api()`, `reverse_via_api()`,
+  `reverse_via_playwright()`, `YandexReverseResult` dataclass. Both API
+  paths share `_parse_api_payload` because Yandex's forward/reverse
+  envelopes have the same shape.
+- `audit_address_sample.sql` — random sample for the Phase 1 audit (used
+  by `audit_address_mismatch.py`).
+- `address_audit_report.sql` — psql-driven post-run summary (p50/p75/p95
+  distance, top-20 outliers, per-district breakdown).

tradein-mvp/backend/scripts/_yandex_reverse.py

@@ -1,26 +1,33 @@
-"""Yandex reverse-geocoding helpers for the address-mismatch audit (issue #582).
+"""Yandex Geocoder helpers for the address-mismatch audit + backfill (issue #582).
 
-Two reverse paths exposed for the driver to choose between depending on
+Three geocoding paths exposed:
-runtime config and quota:
 
-- `reverse_via_api()` — Yandex Geocoder HTTP API. Fast, structured response,
+- `reverse_via_api()` — Yandex Geocoder HTTP API, lon/lat → address. Fast,
-  needs a valid API key (env `YANDEX_GEOCODER_API_KEY`). Free tier is 25k
+  structured response, needs a valid API key (env `YANDEX_GEOCODER_API_KEY`).
-  req/day so 200-row stratified sample is well within budget.
+  Free tier is 25k req/day, fine for ~8.5k houses + audit (~17k total).
 
 - `reverse_via_playwright()` — fallback when no API key is available. Drives
   a real browser session at https://yandex.ru/maps/?…&mode=whatshere. Slower
   and CAPTCHA-prone, so the driver inserts 4-7s sleeps between calls and we
   raise a dedicated exception on CAPTCHA so the batch can pause-and-resume.
 
-Both return a `YandexReverseResult` dataclass so the driver code stays
+- `forward_via_api()` — address → lon/lat + canonical address (Phase 2 of
-implementation-agnostic. The `raw` field always carries the full source
+  issue #582). Used by `backfill_house_coords.py` to fill `houses.lat/lon`
-payload — JSON for the API path, a small dict snapshot for the playwright
+  for the 4141 houses scraped from sources that didn't include coords (esp.
-path (state object or selector text) — so we can do post-hoc diagnostics.
+  yandex_valuation, which only returns an address string).
 
-Why both:
+All three return a `YandexReverseResult` dataclass — same shape regardless
+of direction so the driver code stays implementation-agnostic. The `raw`
+field always carries the full source payload for post-hoc diagnostics, and
+`precision` / `kind` are filled in by the API paths so the caller can skip
+imprecise matches (e.g. only-street-level results during backfill).
+
+Why three paths:
 The user (issue #582 discussion) wants the audit to run on dev machines
 that may not have an API key, but on prod we already provision the key for
-estimator.py. Supporting both keeps the script useful in both contexts.
+estimator.py. Forward geocode is API-only — Playwright forward geocoding
+through Yandex Maps search is too fragile (relevance ranking, suggest
+dropdown). For dev without a key, backfill simply doesn't run.
 """
 
 from __future__ import annotations
@@ -58,22 +65,34 @@ _API_TIMEOUT = httpx.Timeout(connect=5.0, read=10.0, write=5.0, pool=5.0)
 
 @dataclass
 class YandexReverseResult:
-    """Normalized result of a reverse-geocode call (API or browser).
+    """Normalized result of a geocode call (forward, reverse-API, or browser).
 
     Attributes:
-        address: Human-readable address Yandex thinks corresponds to lat/lon.
+        address: Human-readable canonical address Yandex returned. For
-            None if Yandex returned no match.
+            reverse, this is the snapped address at the queried point. For
+            forward, this is the canonical form of the input address. None
+            if Yandex returned no match.
         snapped_lat: Latitude of the matched object's geometric centre.
         snapped_lon: Longitude of the matched object's geometric centre.
+        precision: For forward calls — Yandex match precision tag (`exact`,
+            `number`, `near`, `range`, `street`, `other`). For reverse —
+            same field is filled when present (usually `house` / `street`).
+            None for the playwright path. Used by the backfill driver to
+            skip imprecise matches.
+        kind: Object kind from Yandex (`house`, `street`, `locality`, ...).
+            Same source as `precision` — see metaDataProperty.GeocoderMetaData.
         raw: Raw response payload retained for forensics (JSON dict from API,
             or snapshot dict from playwright). Used to populate
-            `address_mismatch_audit.raw_payload`.
+            `address_mismatch_audit.raw_payload` and
+            `houses.raw_payload.yandex_geocode`.
     """
 
     address: str | None
     snapped_lat: float | None
     snapped_lon: float | None
     raw: dict[str, Any] = field(default_factory=dict)
+    precision: str | None = None
+    kind: str | None = None
 
 
 class YandexBlockedError(RuntimeError):
@@ -143,7 +162,8 @@ def _parse_api_payload(data: dict[str, Any]) -> YandexReverseResult:
     """Extract address + snapped point from a Yandex Geocoder API JSON response.
 
     Split out so unit tests can feed a fixture file directly without spinning
-    up an HTTP mock.
+    up an HTTP mock. Same payload shape for forward and reverse calls —
+    Yandex's response envelope is symmetric.
     """
     try:
         members = data.get("response", {}).get("GeoObjectCollection", {}).get("featureMember", [])
@@ -156,6 +176,8 @@ def _parse_api_payload(data: dict[str, Any]) -> YandexReverseResult:
         # (full canonical) and fall back to `name` (street + house number).
         meta = geo_obj.get("metaDataProperty", {}).get("GeocoderMetaData", {})
         address = meta.get("text") or geo_obj.get("name")
+        precision = meta.get("precision")
+        kind = meta.get("kind")
 
         # Point format: "<lon> <lat>" — space-separated string.
         point_str = geo_obj.get("Point", {}).get("pos", "")
@@ -178,12 +200,79 @@ def _parse_api_payload(data: dict[str, Any]) -> YandexReverseResult:
             snapped_lat=snapped_lat,
             snapped_lon=snapped_lon,
             raw=data,
+            precision=precision,
+            kind=kind,
         )
     except Exception as e:  # pragma: no cover — defensive; tests cover happy paths
         logger.warning("yandex API payload parse failed: %s", e)
         return YandexReverseResult(address=None, snapped_lat=None, snapped_lon=None, raw=data)
 
 
+# ---------------------------------------------------------------------------
+# Path A.2 — Forward geocode (address → lon/lat) via HTTP API
+# ---------------------------------------------------------------------------
+
+
+async def forward_via_api(
+    address: str,
+    api_key: str,
+    *,
+    client: httpx.AsyncClient | None = None,
+) -> YandexReverseResult:
+    """Forward-geocode an address string via the Yandex Geocoder HTTP API.
+
+    Phase 2 of issue #582 — used by `backfill_house_coords.py` to populate
+    `houses.lat/lon` for houses that were scraped without coords (esp.
+    yandex_valuation rows, which only carry an address).
+
+    Args:
+        address: free-form address ("ул Малышева 51", "Екатеринбург, Ленина 5",
+            etc.). Yandex's NLU is forgiving — no need to pre-normalize.
+        api_key: Yandex Geocoder API key.
+        client: optional pre-built async client. If None, a one-shot client
+            is created (matches `reverse_via_api` ergonomics).
+
+    Returns:
+        `YandexReverseResult` with the canonical address + snapped point of
+        the first matching feature. `precision` and `kind` are populated so
+        the backfill driver can skip imprecise hits (e.g. precision='street'
+        means we landed on the road, not the building — too vague for
+        comparable-listings spatial queries).
+
+    Same envelope as `reverse_via_api` — `_parse_api_payload` handles both.
+    """
+    params = {
+        "apikey": api_key,
+        "geocode": address,
+        "format": "json",
+        # `kind=house` filters out street-only / locality-only matches at
+        # the API level when possible. Yandex still returns lower-precision
+        # results when no building matches, so the caller must double-check
+        # `precision` before writing to houses.
+        "kind": "house",
+        "results": "1",
+        # Locality bias for EKB — improves recall when the input address
+        # omits the city. The audit population is 99% EKB houses, so this
+        # is safe; non-EKB inputs (rare) still resolve, just with the bias.
+        "ll": "60.6122,56.8389",
+        "spn": "0.6,0.4",
+    }
+
+    own_client = client is None
+    if client is None:
+        client = httpx.AsyncClient(timeout=_API_TIMEOUT)
+
+    try:
+        resp = await client.get(_YANDEX_GEOCODE_API, params=params)
+        resp.raise_for_status()
+        data = resp.json()
+    finally:
+        if own_client:
+            await client.aclose()
+
+    return _parse_api_payload(data)
+
+
 # ---------------------------------------------------------------------------
 # Path B — Playwright fallback
 # ---------------------------------------------------------------------------

tradein-mvp/backend/scripts/audit_address_sample.sql

@@ -1,53 +1,47 @@
 -- audit_address_sample.sql
--- Stratified sample of EKB houses for the address-mismatch audit (issue #582).
+-- Random sample of EKB houses for the address-mismatch audit (issue #582).
 --
 -- Strategy:
---   1. Spatial-join houses → gendesign_ekb_districts_geom via ST_Contains
+--   1. Filter to houses with non-null lat/lon and non-empty address.
---      (only houses with lat/lon falling inside one of the 8 admin polygons).
+--   2. Random shuffle via `ORDER BY random()` — repeatable enough for spot
---   2. Partition by district, ROW_NUMBER() with a deterministic ORDER BY id
+--      sampling without needing a stable PRNG seed (the audit table dedupes
---      so repeated runs return the same sample (deterministic = matches the
+--      via UNIQUE (house_id, audit_batch), so re-running gives idempotent
---      "resumable per (house_id, audit_batch)" guarantee in 066 migration).
+--      results regardless of which rows land in the sample first).
---   3. Keep top :limit_per_district per district → ~200 rows total for 8
+--   3. Cap the result at :limit_per_district * 8 rows — keeps the bind-param
---      districts × 25.
+--      contract compatible with the old stratified sampler (`:limit_per_district`
+--      is still honored, just multiplied by the assumed 8-district count).
+--
+-- Why no spatial stratification anymore:
+--   The previous version JOINed to `gendesign_ekb_districts_geom` (FDW
+--   polygon table) to bucket houses by admin district. That join is fine on
+--   prod where FDW is wired, but it adds a dependency we don't need for
+--   Phase 2-3 (backfill + canonical reverse). Aggregation by district at
+--   report time still works — we re-derive district during the audit via
+--   spatial containment in the report SQL when needed.
 --
 -- Bind param:
---   :limit_per_district — default 25, expressed as :limit_per_district
+--   :limit_per_district — kept for back-compat with the audit driver.
---     (psql variables resolve in :name form for the report; for Python use
+--     Effective sample size = :limit_per_district * 8 (e.g. 25 → 200).
---     SQLAlchemy text() with bindparams).
 --
 -- Columns returned:
 --   id, address, lat, lon, district
+--   `district` is always NULL here — the audit driver will reverse-derive it
+--   from Yandex Geocoder response (Yandex returns admin component) or leave
+--   it NULL if not present in the response.
 --
 -- NB: uses CAST(:x AS int) per project sql.md rule (psycopg v3 ignores ::type
 -- after bind params).
 
-WITH houses_in_districts AS (
+SELECT
-    SELECT
+    h.id,
-        h.id,
+    h.address,
-        h.address,
+    h.lat,
-        h.lat,
+    h.lon,
-        h.lon,
+    NULL::text AS district
-        d.district_name AS district,
+FROM houses h
-        h.geom
+WHERE h.lat IS NOT NULL
-    FROM houses h
+  AND h.lon IS NOT NULL
-    JOIN gendesign_ekb_districts_geom d
+  AND h.address IS NOT NULL
-      ON ST_Contains(d.geom, h.geom)
+  AND length(trim(h.address)) > 0
-    WHERE h.lat IS NOT NULL
+ORDER BY random()
-      AND h.lon IS NOT NULL
+LIMIT CAST(:limit_per_district AS int) * 8;
-      AND h.address IS NOT NULL
-      AND length(trim(h.address)) > 0
-),
-ranked AS (
-    SELECT
-        id,
-        address,
-        lat,
-        lon,
-        district,
-        ROW_NUMBER() OVER (PARTITION BY district ORDER BY id) AS rn
-    FROM houses_in_districts
-)
-SELECT id, address, lat, lon, district
-FROM ranked
-WHERE rn <= CAST(:limit_per_district AS int)
-ORDER BY district, id;

tradein-mvp/backend/scripts/backfill_house_coords.py

@@ -0,0 +1,619 @@
+"""Forward-geocode houses through Yandex Geocoder API to backfill lat/lon
+and canonical address, plus optional reverse audit of already-geocoded houses.
+
+Phase 2-3 of Forgejo issue #582. Two modes (mutually exclusive):
+
+  1. Backfill (default) — for the ~4141 rows WHERE lat IS NULL OR lon IS NULL:
+     forward-geocode `houses.address` → snap to a Yandex `house`-precision
+     point, UPDATE houses with the new lat/lon + canonical address payload,
+     and write an `address_mismatch_audit` row with `audit_status='backfill'`.
+
+  2. Audit-only (--audit-only) — for the ~4452 rows that already have coords:
+     reverse-geocode (lat, lon) → snapped point + canonical address, compute
+     ST_Distance vs stored coords, write an `address_mismatch_audit` row with
+     status 'ok' (≤50m) or 'mismatch' (>50m). Does NOT touch houses.
+
+Design choices:
+- **Per-row SAVEPOINT** (`db.begin_nested()`): a single Yandex/PostGIS error
+  must not nuke the entire batch. Per backend.md, never use bare rollback
+  inside a loop.
+- **Resumable** via UNIQUE (house_id, audit_batch). Re-running the same
+  --batch label skips already-processed houses, so a partial run can be
+  picked up after CAPTCHA / network blip / 25k/day quota hit.
+- **Precision filter**: backfill skips matches with precision in
+  ('street', 'other', 'range', 'near', None) — those are too imprecise for
+  comparable-listing spatial queries and would silently degrade matching
+  recall. The audit row still records what Yandex returned for forensics.
+- **Rate limit**: 50ms between calls (~20 req/sec, well under Yandex's
+  25 req/sec service limit). Backfill mode runs single-threaded.
+- **Daily quota**: 4141 backfill + 4452 audit ≈ 8.6k requests. Free Geocoder
+  tier is 25k/day → comfortable buffer for retries.
+
+Usage:
+    YANDEX_GEOCODER_API_KEY=xxx \\
+    DATABASE_URL=postgresql+psycopg://... \\
+    python -m scripts.backfill_house_coords --batch 2026-05-27_backfill
+
+    # Audit-only on the 4452 already-geocoded houses
+    python -m scripts.backfill_house_coords --batch 2026-05-27_audit \\
+        --audit-only --limit 500
+
+Outputs:
+- Backfill mode: UPDATE rows in `houses`, INSERT rows in
+  `address_mismatch_audit` with status 'backfill' / 'no_match' / 'imprecise'.
+- Audit mode: INSERT rows in `address_mismatch_audit` with status 'ok' /
+  'mismatch' / 'no_match' / 'error'.
+- Per-batch progress is logged every 25 rows.
+"""
+
+from __future__ import annotations
+
+import argparse
+import asyncio
+import json
+import logging
+import os
+from dataclasses import dataclass
+from datetime import date
+from pathlib import Path
+from typing import Any
+
+import httpx
+from sqlalchemy import text
+from sqlalchemy.orm import Session
+
+# Allow running both as `python -m scripts.backfill_house_coords` (preferred)
+# and as a stand-alone file. Mirrors the audit_address_mismatch import dance.
+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
+
+try:
+    from scripts._yandex_reverse import (  # type: ignore[import-not-found]
+        YandexReverseResult,
+        forward_via_api,
+        reverse_via_api,
+    )
+except ImportError:
+    from _yandex_reverse import (  # type: ignore[no-redef]
+        YandexReverseResult,
+        forward_via_api,
+        reverse_via_api,
+    )
+
+logging.basicConfig(
+    level=logging.INFO,
+    format="%(asctime)s %(levelname)s %(name)s %(message)s",
+)
+logger = logging.getLogger("backfill_house_coords")
+
+# Yandex Geocoder service limits per docs (as of 2026-05):
+#   - 25k requests/day free tier
+#   - 25 requests/sec sustained
+# 50ms between calls = ~20 req/sec, leaving headroom for connection ramp-up.
+_REQUEST_DELAY_S = 0.05
+
+# Precision values we ACCEPT for backfill — anything else means Yandex didn't
+# resolve to a specific building, and writing the result back into houses
+# would degrade matching recall.
+#   `exact`  → match found at the exact address (best case)
+#   `number` → house number matched, but unit/entrance unspecified (acceptable)
+# `near` / `range` / `street` / `other` / None → skipped (logged for analysis).
+_BACKFILL_OK_PRECISION = frozenset({"exact", "number"})
+
+# Audit threshold per issue #582 — distances above this flag a "mismatch"
+# (the row still goes in the audit table, just with status='mismatch' for
+# the report SQL to bucket separately).
+_MISMATCH_DISTANCE_M = 50.0
+
+
+# ---------------------------------------------------------------------------
+# Domain types
+# ---------------------------------------------------------------------------
+
+
+@dataclass
+class HouseRow:
+    """One house from the source query — minimal fields needed for geocode."""
+
+    id: int
+    address: str
+    lat: float | None
+    lon: float | None
+
+
+# ---------------------------------------------------------------------------
+# Source-row queries
+# ---------------------------------------------------------------------------
+
+
+def _select_houses_without_coords(db: Session, limit: int | None) -> list[HouseRow]:
+    """Pull houses needing forward geocode (lat IS NULL OR lon IS NULL).
+
+    Skips rows with empty address — there's nothing to geocode there, they
+    need a separate cleanup pass.
+    """
+    sql = (
+        "SELECT id, address, lat, lon "
+        "FROM houses "
+        "WHERE (lat IS NULL OR lon IS NULL) "
+        "  AND address IS NOT NULL "
+        "  AND length(trim(address)) > 0 "
+        "ORDER BY id"
+    )
+    if limit is not None:
+        sql += " LIMIT CAST(:limit AS int)"
+        rows = db.execute(text(sql), {"limit": limit}).mappings().all()
+    else:
+        rows = db.execute(text(sql)).mappings().all()
+    return [
+        HouseRow(id=r["id"], address=r["address"], lat=r["lat"], lon=r["lon"]) for r in rows
+    ]
+
+
+def _select_houses_with_coords(db: Session, limit: int | None) -> list[HouseRow]:
+    """Pull houses needing reverse audit (both lat AND lon present)."""
+    sql = (
+        "SELECT id, address, lat, lon "
+        "FROM houses "
+        "WHERE lat IS NOT NULL "
+        "  AND lon IS NOT NULL "
+        "  AND address IS NOT NULL "
+        "  AND length(trim(address)) > 0 "
+        "ORDER BY id"
+    )
+    if limit is not None:
+        sql += " LIMIT CAST(:limit AS int)"
+        rows = db.execute(text(sql), {"limit": limit}).mappings().all()
+    else:
+        rows = db.execute(text(sql)).mappings().all()
+    return [
+        HouseRow(id=r["id"], address=r["address"], lat=r["lat"], lon=r["lon"]) for r in rows
+    ]
+
+
+def _already_processed_ids(db: Session, batch: str) -> set[int]:
+    """house_ids already in address_mismatch_audit for this batch → skip set."""
+    rows = db.execute(
+        text("SELECT house_id FROM address_mismatch_audit WHERE audit_batch = CAST(:b AS text)"),
+        {"b": batch},
+    ).all()
+    return {r[0] for r in rows}
+
+
+# ---------------------------------------------------------------------------
+# Distance helper — PostGIS, lon/lat order
+# ---------------------------------------------------------------------------
+
+
+def _distance_meters(
+    db: Session, olat: float, olon: float, slat: float, slon: float
+) -> float | None:
+    """Great-circle distance (meters) via PostGIS geography type.
+
+    Lifted from `audit_address_mismatch.py` to keep the two scripts using
+    the same authority for distance computation. ST_MakePoint takes lon
+    first per PostGIS convention.
+    """
+    row = db.execute(
+        text(
+            "SELECT ST_Distance("
+            "  ST_SetSRID(ST_MakePoint(CAST(:olon AS double precision), "
+            "                          CAST(:olat AS double precision)), 4326)::geography, "
+            "  ST_SetSRID(ST_MakePoint(CAST(:slon AS double precision), "
+            "                          CAST(:slat AS double precision)), 4326)::geography"
+            ") AS m"
+        ),
+        {"olat": olat, "olon": olon, "slat": slat, "slon": slon},
+    ).first()
+    if row is None or row[0] is None:
+        return None
+    return float(row[0])
+
+
+# ---------------------------------------------------------------------------
+# DB writers
+# ---------------------------------------------------------------------------
+
+
+def _update_house_coords(
+    db: Session,
+    *,
+    house_id: int,
+    lat: float,
+    lon: float,
+    payload: dict[str, Any],
+) -> None:
+    """UPDATE houses SET lat/lon + merge yandex_geocode into raw_payload.
+
+    The `houses_set_geom_trg` BEFORE UPDATE trigger (009_houses.sql) maintains
+    `geom` automatically when lat/lon change, so we don't need to set geom
+    explicitly here. `raw_payload || jsonb_build_object(...)` is the idiomatic
+    psycopg-safe way to merge — single ALTER, no read-modify-write race.
+    """
+    db.execute(
+        text(
+            "UPDATE houses "
+            "   SET lat = CAST(:lat AS double precision), "
+            "       lon = CAST(:lon AS double precision), "
+            "       raw_payload = COALESCE(raw_payload, '{}'::jsonb) "
+            "                     || jsonb_build_object('yandex_geocode', "
+            "                                           CAST(:payload AS jsonb)) "
+            " WHERE id = CAST(:id AS bigint)"
+        ),
+        {"id": house_id, "lat": lat, "lon": lon, "payload": json.dumps(payload)},
+    )
+
+
+def _insert_audit_row(
+    db: Session,
+    *,
+    house_id: int,
+    batch: str,
+    original_address: str | None,
+    original_lat: float | None,
+    original_lon: float | None,
+    snapped_address: str | None,
+    snapped_lat: float | None,
+    snapped_lon: float | None,
+    distance_m: float | None,
+    audit_status: str,
+    error_message: str | None,
+    raw_payload: dict[str, Any] | None,
+) -> None:
+    """INSERT … ON CONFLICT DO NOTHING into address_mismatch_audit.
+
+    Same column shape as `audit_address_mismatch._insert_audit_row` but the
+    `district` and `street_differs` fields are left NULL — backfill/audit
+    here doesn't have a stratification basis and we let the report SQL
+    derive district at query time if needed (via Yandex address parse).
+
+    Caller wraps in `begin_nested()` per backend.md SAVEPOINT pattern.
+    """
+    db.execute(
+        text(
+            "INSERT INTO address_mismatch_audit ("
+            "  house_id, audit_batch, district,"
+            "  original_address, original_lat, original_lon,"
+            "  snapped_address, snapped_lat, snapped_lon,"
+            "  distance_m, street_differs,"
+            "  audit_status, error_message, raw_payload"
+            ") VALUES ("
+            "  CAST(:house_id AS bigint), CAST(:batch AS text), NULL,"
+            "  :original_address, :original_lat, :original_lon,"
+            "  :snapped_address, :snapped_lat, :snapped_lon,"
+            "  :distance_m, NULL,"
+            "  CAST(:audit_status AS text), :error_message,"
+            "  CAST(:raw_payload AS jsonb)"
+            ") ON CONFLICT (house_id, audit_batch) DO NOTHING"
+        ),
+        {
+            "house_id": house_id,
+            "batch": batch,
+            "original_address": original_address,
+            "original_lat": original_lat,
+            "original_lon": original_lon,
+            "snapped_address": snapped_address,
+            "snapped_lat": snapped_lat,
+            "snapped_lon": snapped_lon,
+            "distance_m": distance_m,
+            "audit_status": audit_status,
+            "error_message": error_message,
+            "raw_payload": json.dumps(raw_payload) if raw_payload is not None else None,
+        },
+    )
+
+
+# ---------------------------------------------------------------------------
+# Backfill loop (forward geocode, lat IS NULL houses)
+# ---------------------------------------------------------------------------
+
+
+def _classify_backfill_status(res: YandexReverseResult | None) -> str:
+    """Translate a forward-geocode result into an audit_status value.
+
+    'backfill' — Yandex returned a precise hit, lat/lon will be written.
+    'imprecise' — match returned but precision is too low (street/other/...).
+    'no_match' — Yandex returned an empty featureMember.
+    'error' — handled by the caller's exception branch.
+    """
+    if res is None or res.address is None:
+        return "no_match"
+    if res.precision not in _BACKFILL_OK_PRECISION:
+        return "imprecise"
+    if res.snapped_lat is None or res.snapped_lon is None:
+        return "no_match"
+    return "backfill"
+
+
+async def _run_backfill_mode(
+    db: Session, sample: list[HouseRow], batch: str, api_key: str
+) -> int:
+    """Forward-geocode each house, UPDATE coords on precise hits, audit-log all."""
+    processed = 0
+    updated = 0
+    n_imprecise = 0
+    n_no_match = 0
+    n_error = 0
+
+    async with httpx.AsyncClient(timeout=httpx.Timeout(10.0)) as client:
+        for i, row in enumerate(sample, start=1):
+            status = "backfill"
+            err: str | None = None
+            res: YandexReverseResult | None = None
+            try:
+                res = await forward_via_api(row.address, api_key, client=client)
+            except httpx.HTTPError as e:
+                status = "error"
+                err = f"http_error: {e!s}"
+                n_error += 1
+            except Exception as e:  # pragma: no cover — defensive
+                status = "error"
+                err = f"unhandled: {e!s}"
+                n_error += 1
+
+            if status != "error":
+                status = _classify_backfill_status(res)
+                if status == "imprecise":
+                    n_imprecise += 1
+                elif status == "no_match":
+                    n_no_match += 1
+
+            try:
+                with db.begin_nested():
+                    if status == "backfill" and res is not None and res.snapped_lat is not None:
+                        # safe: status='backfill' guarantees snapped_lat/lon non-None.
+                        assert res.snapped_lon is not None
+                        _update_house_coords(
+                            db,
+                            house_id=row.id,
+                            lat=res.snapped_lat,
+                            lon=res.snapped_lon,
+                            payload={
+                                "address": res.address,
+                                "precision": res.precision,
+                                "kind": res.kind,
+                                "batch": batch,
+                                "source": "yandex_geocoder_api",
+                            },
+                        )
+                        updated += 1
+                    _insert_audit_row(
+                        db,
+                        house_id=row.id,
+                        batch=batch,
+                        original_address=row.address,
+                        original_lat=row.lat,
+                        original_lon=row.lon,
+                        snapped_address=res.address if res else None,
+                        snapped_lat=res.snapped_lat if res else None,
+                        snapped_lon=res.snapped_lon if res else None,
+                        distance_m=None,
+                        audit_status=status,
+                        error_message=err,
+                        raw_payload=res.raw if res else None,
+                    )
+                # Per-row commit so resume picks up exactly where we crashed.
+                db.commit()
+                processed += 1
+            except Exception as e:
+                db.rollback()
+                logger.warning("backfill insert failed for house_id=%s: %s", row.id, e)
+
+            if i % 25 == 0:
+                logger.info(
+                    "backfill progress %d/%d updated=%d imprecise=%d no_match=%d error=%d",
+                    i,
+                    len(sample),
+                    updated,
+                    n_imprecise,
+                    n_no_match,
+                    n_error,
+                )
+
+            # Yandex 25 req/sec → 50ms between calls is plenty of headroom.
+            if i < len(sample):
+                await asyncio.sleep(_REQUEST_DELAY_S)
+
+    logger.info(
+        "backfill done: processed=%d updated=%d imprecise=%d no_match=%d error=%d",
+        processed,
+        updated,
+        n_imprecise,
+        n_no_match,
+        n_error,
+    )
+    return processed
+
+
+# ---------------------------------------------------------------------------
+# Audit-only loop (reverse geocode, lat IS NOT NULL houses)
+# ---------------------------------------------------------------------------
+
+
+async def _run_audit_mode(
+    db: Session, sample: list[HouseRow], batch: str, api_key: str
+) -> int:
+    """Reverse-geocode each house, compute distance, audit-log status/mismatch."""
+    processed = 0
+    n_ok = 0
+    n_mismatch = 0
+    n_no_match = 0
+    n_error = 0
+
+    async with httpx.AsyncClient(timeout=httpx.Timeout(10.0)) as client:
+        for i, row in enumerate(sample, start=1):
+            # Type-narrow: audit mode only feeds rows with non-null coords.
+            assert row.lat is not None and row.lon is not None
+            status = "ok"
+            err: str | None = None
+            res: YandexReverseResult | None = None
+            try:
+                res = await reverse_via_api(row.lat, row.lon, api_key, client=client)
+            except httpx.HTTPError as e:
+                status = "error"
+                err = f"http_error: {e!s}"
+                n_error += 1
+            except Exception as e:  # pragma: no cover — defensive
+                status = "error"
+                err = f"unhandled: {e!s}"
+                n_error += 1
+
+            distance = None
+            if res is not None and status == "ok":
+                if res.address is None:
+                    status = "no_match"
+                    n_no_match += 1
+                else:
+                    if res.snapped_lat is not None and res.snapped_lon is not None:
+                        distance = _distance_meters(
+                            db, row.lat, row.lon, res.snapped_lat, res.snapped_lon
+                        )
+                        if distance is not None and distance > _MISMATCH_DISTANCE_M:
+                            status = "mismatch"
+                            n_mismatch += 1
+                        else:
+                            n_ok += 1
+                    else:
+                        n_ok += 1
+
+            try:
+                with db.begin_nested():
+                    _insert_audit_row(
+                        db,
+                        house_id=row.id,
+                        batch=batch,
+                        original_address=row.address,
+                        original_lat=row.lat,
+                        original_lon=row.lon,
+                        snapped_address=res.address if res else None,
+                        snapped_lat=res.snapped_lat if res else None,
+                        snapped_lon=res.snapped_lon if res else None,
+                        distance_m=distance,
+                        audit_status=status,
+                        error_message=err,
+                        raw_payload=res.raw if res else None,
+                    )
+                db.commit()
+                processed += 1
+            except Exception as e:
+                db.rollback()
+                logger.warning("audit insert failed for house_id=%s: %s", row.id, e)
+
+            if i % 25 == 0:
+                logger.info(
+                    "audit progress %d/%d ok=%d mismatch=%d no_match=%d error=%d",
+                    i,
+                    len(sample),
+                    n_ok,
+                    n_mismatch,
+                    n_no_match,
+                    n_error,
+                )
+
+            if i < len(sample):
+                await asyncio.sleep(_REQUEST_DELAY_S)
+
+    logger.info(
+        "audit done: processed=%d ok=%d mismatch=%d no_match=%d error=%d",
+        processed,
+        n_ok,
+        n_mismatch,
+        n_no_match,
+        n_error,
+    )
+    return processed
+
+
+# ---------------------------------------------------------------------------
+# CLI
+# ---------------------------------------------------------------------------
+
+
+def _parse_args(argv: list[str] | None = None) -> argparse.Namespace:
+    """argparse setup, factored out for testability."""
+    p = argparse.ArgumentParser(
+        description=(
+            "Phase 2-3 of issue #582 — backfill houses.lat/lon via Yandex forward "
+            "geocode, or audit already-geocoded houses via reverse geocode."
+        ),
+    )
+    p.add_argument(
+        "--batch",
+        default=f"{date.today().isoformat()}_backfill",
+        help="Audit batch label. Same batch re-run skips already-processed houses.",
+    )
+    p.add_argument(
+        "--audit-only",
+        action="store_true",
+        help=(
+            "Run reverse-geocode audit on houses WITH coords instead of forward "
+            "backfill on houses WITHOUT coords. Does not modify the houses table."
+        ),
+    )
+    p.add_argument(
+        "--limit",
+        type=int,
+        default=None,
+        help=(
+            "Optional cap on source-row count. Useful for canary runs "
+            "(e.g. --limit 100 before letting the full 4k loose)."
+        ),
+    )
+    return p.parse_args(argv)
+
+
+async def main(argv: list[str] | None = None) -> int:
+    """CLI entry point. Returns the number of rows processed this run."""
+    args = _parse_args(argv)
+    api_key = os.environ.get("YANDEX_GEOCODER_API_KEY")
+    if not api_key:
+        raise SystemExit(
+            "YANDEX_GEOCODER_API_KEY is required — forward geocode is API-only."
+        )
+
+    mode = "audit" if args.audit_only else "backfill"
+    logger.info(
+        "starting batch=%s mode=%s limit=%s",
+        args.batch,
+        mode,
+        args.limit if args.limit is not None else "all",
+    )
+
+    db = SessionLocal()
+    try:
+        if args.audit_only:
+            sample = _select_houses_with_coords(db, args.limit)
+        else:
+            sample = _select_houses_without_coords(db, args.limit)
+        logger.info("loaded source rows: %d", len(sample))
+
+        done = _already_processed_ids(db, args.batch)
+        if done:
+            logger.info(
+                "resuming batch %s: %d rows already processed",
+                args.batch,
+                len(done),
+            )
+        remaining = [s for s in sample if s.id not in done]
+        if not remaining:
+            logger.info("nothing to do — batch %s is complete for the loaded sample", args.batch)
+            return 0
+
+        if args.audit_only:
+            n = await _run_audit_mode(db, remaining, args.batch, api_key)
+        else:
+            n = await _run_backfill_mode(db, remaining, args.batch, api_key)
+
+        logger.info("done: processed=%d batch=%s mode=%s", n, args.batch, mode)
+        return n
+    finally:
+        db.close()
+
+
+if __name__ == "__main__":  # pragma: no cover
+    asyncio.run(main())

tradein-mvp/backend/tests/test_backfill_house_coords.py

@@ -0,0 +1,510 @@
+"""Unit tests for the Phase 2-3 backfill/audit script (issue #582).
+
+Coverage:
+  - `forward_via_api` request shape — verifies geocode/format/locality bias.
+  - `_parse_api_payload` precision + kind extraction.
+  - `_classify_backfill_status` precision filter rules.
+  - `_update_house_coords` — UPDATE shape + raw_payload merge.
+  - `_run_backfill_mode` — happy path UPDATE + audit row, plus imprecise-skip.
+  - `_run_audit_mode` — ok / mismatch / no_match distinction.
+  - `main()` resumability — second pass on same batch inserts 0.
+
+No real Postgres in unit tests (same convention as test_audit_address_mismatch).
+DB is a MagicMock that records INSERT/UPDATE calls and routes SELECT side-effects.
+"""
+
+from __future__ import annotations
+
+import json
+import os
+from unittest.mock import AsyncMock, MagicMock, patch
+
+# Same dance as test_audit_address_mismatch — settings needs a DSN at import.
+os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost/test_db")
+
+import httpx
+import pytest
+
+from scripts._yandex_reverse import (
+    YandexReverseResult,
+    _parse_api_payload,
+    forward_via_api,
+)
+from scripts.backfill_house_coords import (
+    HouseRow,
+    _classify_backfill_status,
+    _run_audit_mode,
+    _run_backfill_mode,
+    _update_house_coords,
+    main,
+)
+
+# ---------------------------------------------------------------------------
+# forward_via_api — request shape
+# ---------------------------------------------------------------------------
+
+
+async def test_forward_api_request_shape():
+    """Verify the GET param dict — address as `geocode`, kind=house, EKB bias."""
+    fixture = {
+        "response": {
+            "GeoObjectCollection": {
+                "featureMember": [
+                    {
+                        "GeoObject": {
+                            "metaDataProperty": {
+                                "GeocoderMetaData": {
+                                    "text": "Россия, Свердловская область, Екатеринбург, "
+                                    "улица Малышева, 51",
+                                    "precision": "exact",
+                                    "kind": "house",
+                                }
+                            },
+                            "name": "улица Малышева, 51",
+                            "Point": {"pos": "60.586155 56.838004"},
+                        }
+                    }
+                ]
+            }
+        }
+    }
+    captured: dict[str, httpx.Request] = {}
+
+    def handler(request: httpx.Request) -> httpx.Response:
+        captured["req"] = request
+        return httpx.Response(200, json=fixture)
+
+    transport = httpx.MockTransport(handler)
+    async with httpx.AsyncClient(transport=transport) as client:
+        res = await forward_via_api("ул Малышева 51", "DUMMY_KEY", client=client)
+
+    assert res.address is not None and "Малышева" in res.address
+    assert res.precision == "exact"
+    assert res.kind == "house"
+    assert res.snapped_lon == pytest.approx(60.586155, abs=1e-6)
+    assert res.snapped_lat == pytest.approx(56.838004, abs=1e-6)
+
+    qs = dict(httpx.QueryParams(captured["req"].url.query))
+    assert qs["apikey"] == "DUMMY_KEY"
+    assert qs["geocode"] == "ул Малышева 51"
+    assert qs["format"] == "json"
+    assert qs["kind"] == "house"
+    # EKB locality bias for forward geocode — important so addresses without
+    # the city resolve to the correct Малышева (there's one in Moscow too).
+    assert "ll" in qs
+    assert "spn" in qs
+
+
+# ---------------------------------------------------------------------------
+# Precision / kind passthrough in _parse_api_payload
+# ---------------------------------------------------------------------------
+
+
+def test_parse_api_payload_propagates_precision_and_kind():
+    data = {
+        "response": {
+            "GeoObjectCollection": {
+                "featureMember": [
+                    {
+                        "GeoObject": {
+                            "metaDataProperty": {
+                                "GeocoderMetaData": {
+                                    "text": "ул Ленина 5",
+                                    "precision": "exact",
+                                    "kind": "house",
+                                }
+                            },
+                            "name": "ул Ленина 5",
+                            "Point": {"pos": "60.6 56.8"},
+                        }
+                    }
+                ]
+            }
+        }
+    }
+    res = _parse_api_payload(data)
+    assert res.precision == "exact"
+    assert res.kind == "house"
+
+
+# ---------------------------------------------------------------------------
+# _classify_backfill_status — precision filter rules
+# ---------------------------------------------------------------------------
+
+
+def test_classify_backfill_status_exact_match():
+    res = YandexReverseResult(
+        address="ул Малышева 51",
+        snapped_lat=56.838,
+        snapped_lon=60.586,
+        precision="exact",
+        kind="house",
+    )
+    assert _classify_backfill_status(res) == "backfill"
+
+
+def test_classify_backfill_status_number_match():
+    res = YandexReverseResult(
+        address="ул Ленина 5",
+        snapped_lat=56.840,
+        snapped_lon=60.600,
+        precision="number",
+        kind="house",
+    )
+    assert _classify_backfill_status(res) == "backfill"
+
+
+def test_classify_backfill_status_street_is_imprecise():
+    res = YandexReverseResult(
+        address="ул Ленина",
+        snapped_lat=56.840,
+        snapped_lon=60.600,
+        precision="street",
+        kind="street",
+    )
+    assert _classify_backfill_status(res) == "imprecise"
+
+
+def test_classify_backfill_status_other_is_imprecise():
+    res = YandexReverseResult(
+        address="Свердловская область",
+        snapped_lat=56.8,
+        snapped_lon=60.6,
+        precision="other",
+        kind="locality",
+    )
+    assert _classify_backfill_status(res) == "imprecise"
+
+
+def test_classify_backfill_status_no_match():
+    res = YandexReverseResult(address=None, snapped_lat=None, snapped_lon=None)
+    assert _classify_backfill_status(res) == "no_match"
+
+
+def test_classify_backfill_status_none():
+    assert _classify_backfill_status(None) == "no_match"
+
+
+def test_classify_backfill_status_precision_ok_but_no_coords():
+    """Defensive: precision=exact but snapped point missing → no_match, not backfill."""
+    res = YandexReverseResult(
+        address="ул Малышева 51",
+        snapped_lat=None,
+        snapped_lon=None,
+        precision="exact",
+        kind="house",
+    )
+    assert _classify_backfill_status(res) == "no_match"
+
+
+# ---------------------------------------------------------------------------
+# _update_house_coords — UPDATE shape verification
+# ---------------------------------------------------------------------------
+
+
+def test_update_house_coords_passes_bindings():
+    db = MagicMock()
+    _update_house_coords(
+        db,
+        house_id=42,
+        lat=56.838,
+        lon=60.586,
+        payload={"address": "ул Малышева 51", "precision": "exact"},
+    )
+    args, _kw = db.execute.call_args
+    sql_str = str(args[0])
+    binds = args[1]
+    assert "UPDATE houses" in sql_str
+    assert "raw_payload" in sql_str
+    assert "yandex_geocode" in sql_str
+    assert binds["id"] == 42
+    assert binds["lat"] == 56.838
+    assert binds["lon"] == 60.586
+    # payload bound as JSON string for CAST(:payload AS jsonb)
+    decoded = json.loads(binds["payload"])
+    assert decoded["address"] == "ул Малышева 51"
+
+
+# ---------------------------------------------------------------------------
+# DB mock helper — same approach as test_audit_address_mismatch
+# ---------------------------------------------------------------------------
+
+
+def _make_db_mock(
+    backfill_sample: list[dict] | None = None,
+    audit_sample: list[dict] | None = None,
+    processed_ids: set[int] | None = None,
+    distance_value: float = 12.5,
+):
+    """MagicMock DB that:
+    - returns `backfill_sample` for `lat IS NULL OR lon IS NULL` SELECT
+    - returns `audit_sample` for `lat IS NOT NULL` SELECT
+    - returns `processed_ids` for the resume SELECT
+    - records INSERTs and UPDATEs
+    - returns `distance_value` for ST_Distance calls
+    """
+    backfill_sample = backfill_sample or []
+    audit_sample = audit_sample or []
+    processed_ids = processed_ids if processed_ids is not None else set()
+
+    inserted: list[dict] = []
+    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 "FROM houses" in sql_str and "lat IS NULL OR lon IS NULL" in sql_str:
+            result.mappings.return_value.all.return_value = backfill_sample
+        elif "FROM houses" in sql_str and "lat IS NOT NULL" in sql_str:
+            result.mappings.return_value.all.return_value = audit_sample
+        elif "FROM address_mismatch_audit" in sql_str and "house_id" in sql_str:
+            result.all.return_value = [(hid,) for hid in processed_ids]
+        elif "INSERT INTO address_mismatch_audit" in sql_str:
+            inserted.append(dict(params))
+            processed_ids.add(params["house_id"])
+        elif "UPDATE houses" in sql_str:
+            updated.append(dict(params))
+        elif "ST_Distance" in sql_str:
+            result.first.return_value = (distance_value,)
+        return result
+
+    db.execute.side_effect = execute_side_effect
+    db.commit = MagicMock()
+    db.rollback = MagicMock()
+    db.close = MagicMock()
+    return db, inserted, updated
+
+
+# ---------------------------------------------------------------------------
+# _run_backfill_mode — happy path + imprecise-skip
+# ---------------------------------------------------------------------------
+
+
+async def test_run_backfill_mode_writes_update_and_audit():
+    sample = [
+        HouseRow(id=1, address="ул Малышева 51", lat=None, lon=None),
+    ]
+    db, inserted, updated = _make_db_mock()
+    res = YandexReverseResult(
+        address="Россия, Екатеринбург, улица Малышева, 51",
+        snapped_lat=56.838,
+        snapped_lon=60.586,
+        precision="exact",
+        kind="house",
+        raw={"ok": True},
+    )
+    with patch(
+        "scripts.backfill_house_coords.forward_via_api",
+        new=AsyncMock(return_value=res),
+    ):
+        n = await _run_backfill_mode(db, sample, "b1", "KEY")
+    assert n == 1
+    assert len(updated) == 1
+    assert updated[0]["id"] == 1
+    assert updated[0]["lat"] == 56.838
+    assert updated[0]["lon"] == 60.586
+    assert len(inserted) == 1
+    assert inserted[0]["audit_status"] == "backfill"
+    assert inserted[0]["snapped_address"] == "Россия, Екатеринбург, улица Малышева, 51"
+
+
+async def test_run_backfill_mode_imprecise_skips_update():
+    """precision='street' → audit row written with status=imprecise, no UPDATE."""
+    sample = [HouseRow(id=2, address="ул Ленина", lat=None, lon=None)]
+    db, inserted, updated = _make_db_mock()
+    res = YandexReverseResult(
+        address="ул Ленина",
+        snapped_lat=56.840,
+        snapped_lon=60.600,
+        precision="street",
+        kind="street",
+        raw={"oh_well": True},
+    )
+    with patch(
+        "scripts.backfill_house_coords.forward_via_api",
+        new=AsyncMock(return_value=res),
+    ):
+        n = await _run_backfill_mode(db, sample, "b2", "KEY")
+    assert n == 1
+    assert updated == []
+    assert len(inserted) == 1
+    assert inserted[0]["audit_status"] == "imprecise"
+
+
+async def test_run_backfill_mode_no_match():
+    """Yandex returns empty result → status=no_match, no UPDATE."""
+    sample = [HouseRow(id=3, address="несуществующая улица 99", lat=None, lon=None)]
+    db, inserted, updated = _make_db_mock()
+    res = YandexReverseResult(
+        address=None, snapped_lat=None, snapped_lon=None, raw={"empty": True}
+    )
+    with patch(
+        "scripts.backfill_house_coords.forward_via_api",
+        new=AsyncMock(return_value=res),
+    ):
+        n = await _run_backfill_mode(db, sample, "b3", "KEY")
+    assert n == 1
+    assert updated == []
+    assert inserted[0]["audit_status"] == "no_match"
+
+
+async def test_run_backfill_mode_http_error_marks_error():
+    sample = [HouseRow(id=4, address="ул X 1", lat=None, lon=None)]
+    db, inserted, updated = _make_db_mock()
+    with patch(
+        "scripts.backfill_house_coords.forward_via_api",
+        new=AsyncMock(side_effect=httpx.HTTPError("boom")),
+    ):
+        n = await _run_backfill_mode(db, sample, "b4", "KEY")
+    assert n == 1
+    assert updated == []
+    assert inserted[0]["audit_status"] == "error"
+    assert "boom" in (inserted[0]["error_message"] or "")
+
+
+# ---------------------------------------------------------------------------
+# _run_audit_mode — ok / mismatch / no_match
+# ---------------------------------------------------------------------------
+
+
+async def test_run_audit_mode_ok_within_50m():
+    sample = [HouseRow(id=10, address="ул Малышева 51", lat=56.838, lon=60.586)]
+    db, inserted, _updated = _make_db_mock(distance_value=12.5)
+    res = YandexReverseResult(
+        address="Россия, Екатеринбург, улица Малышева, 51",
+        snapped_lat=56.838004,
+        snapped_lon=60.586155,
+        precision="exact",
+        kind="house",
+        raw={"r": 1},
+    )
+    with patch(
+        "scripts.backfill_house_coords.reverse_via_api",
+        new=AsyncMock(return_value=res),
+    ):
+        n = await _run_audit_mode(db, sample, "ba1", "KEY")
+    assert n == 1
+    assert inserted[0]["audit_status"] == "ok"
+    assert inserted[0]["distance_m"] == 12.5
+
+
+async def test_run_audit_mode_mismatch_above_50m():
+    sample = [HouseRow(id=11, address="ул Ленина 5", lat=56.840, lon=60.600)]
+    db, inserted, _updated = _make_db_mock(distance_value=312.0)
+    res = YandexReverseResult(
+        address="Россия, Екатеринбург, улица Ленина, 7",
+        snapped_lat=56.841,
+        snapped_lon=60.601,
+        precision="exact",
+        kind="house",
+        raw={"r": 2},
+    )
+    with patch(
+        "scripts.backfill_house_coords.reverse_via_api",
+        new=AsyncMock(return_value=res),
+    ):
+        n = await _run_audit_mode(db, sample, "ba2", "KEY")
+    assert n == 1
+    assert inserted[0]["audit_status"] == "mismatch"
+    assert inserted[0]["distance_m"] == 312.0
+
+
+async def test_run_audit_mode_no_match():
+    sample = [HouseRow(id=12, address="ул X 99", lat=56.0, lon=60.0)]
+    db, inserted, _updated = _make_db_mock()
+    res = YandexReverseResult(
+        address=None, snapped_lat=None, snapped_lon=None, raw={"empty": True}
+    )
+    with patch(
+        "scripts.backfill_house_coords.reverse_via_api",
+        new=AsyncMock(return_value=res),
+    ):
+        n = await _run_audit_mode(db, sample, "ba3", "KEY")
+    assert n == 1
+    assert inserted[0]["audit_status"] == "no_match"
+
+
+# ---------------------------------------------------------------------------
+# Resumability — second pass on same batch inserts 0
+# ---------------------------------------------------------------------------
+
+
+async def test_main_resumable_skips_processed(monkeypatch):
+    """Run main() twice with same batch — second pass processes nothing."""
+    backfill_sample = [
+        {"id": 1, "address": "ул Малышева 51", "lat": None, "lon": None},
+        {"id": 2, "address": "ул Ленина 5", "lat": None, "lon": None},
+    ]
+    processed_ids: set[int] = set()
+    db, inserted, updated = _make_db_mock(
+        backfill_sample=backfill_sample, processed_ids=processed_ids
+    )
+
+    monkeypatch.setenv("YANDEX_GEOCODER_API_KEY", "TEST_KEY")
+    fake = YandexReverseResult(
+        address="ул Малышева 51",
+        snapped_lat=56.838,
+        snapped_lon=60.586,
+        precision="exact",
+        kind="house",
+        raw={"ok": True},
+    )
+
+    with (
+        patch("scripts.backfill_house_coords.SessionLocal", return_value=db),
+        patch(
+            "scripts.backfill_house_coords.forward_via_api",
+            new=AsyncMock(return_value=fake),
+        ),
+    ):
+        n1 = await main(["--batch", "resume_test"])
+        assert n1 == 2
+        assert len(inserted) == 2
+        assert len(updated) == 2
+
+        inserted.clear()
+        updated.clear()
+        n2 = await main(["--batch", "resume_test"])
+        assert n2 == 0
+        assert inserted == []
+        assert updated == []
+
+
+async def test_main_requires_api_key(monkeypatch):
+    """Without YANDEX_GEOCODER_API_KEY the script exits cleanly."""
+    monkeypatch.delenv("YANDEX_GEOCODER_API_KEY", raising=False)
+    with pytest.raises(SystemExit):
+        await main(["--batch", "no_key"])
+
+
+async def test_main_audit_only_flag_routes_to_audit_loop(monkeypatch):
+    """--audit-only switches sample query + loop, no UPDATE expected."""
+    audit_sample = [
+        {"id": 50, "address": "ул Малышева 51", "lat": 56.838, "lon": 60.586},
+    ]
+    db, inserted, updated = _make_db_mock(audit_sample=audit_sample, distance_value=8.0)
+    monkeypatch.setenv("YANDEX_GEOCODER_API_KEY", "TEST_KEY")
+    fake = YandexReverseResult(
+        address="Россия, Екатеринбург, улица Малышева, 51",
+        snapped_lat=56.838004,
+        snapped_lon=60.586155,
+        precision="exact",
+        kind="house",
+        raw={"r": 1},
+    )
+    with (
+        patch("scripts.backfill_house_coords.SessionLocal", return_value=db),
+        patch(
+            "scripts.backfill_house_coords.reverse_via_api",
+            new=AsyncMock(return_value=fake),
+        ),
+    ):
+        n = await main(["--batch", "audit_run", "--audit-only"])
+    assert n == 1
+    assert updated == []  # audit mode never updates houses
+    assert inserted[0]["audit_status"] == "ok"
+    assert inserted[0]["distance_m"] == 8.0