From 3c506222e617aa35cc83e2d984b67c4ec99f63f5 Mon Sep 17 00:00:00 2001 From: Light1YT Date: Wed, 27 May 2026 06:30:54 +0000 Subject: [PATCH] =?UTF-8?q?feat(tradein):=20Phase=202-3=20of=20#582=20?= =?UTF-8?q?=E2=80=94=20Yandex=20backfill=20+=20canonical=20audit=20(#591)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- tradein-mvp/backend/scripts/README.md | 116 ++++ .../backend/scripts/_yandex_reverse.py | 123 +++- .../backend/scripts/audit_address_sample.sql | 76 +-- .../backend/scripts/backfill_house_coords.py | 619 ++++++++++++++++++ .../tests/test_backfill_house_coords.py | 510 +++++++++++++++ 5 files changed, 1386 insertions(+), 58 deletions(-) create mode 100644 tradein-mvp/backend/scripts/README.md create mode 100644 tradein-mvp/backend/scripts/backfill_house_coords.py create mode 100644 tradein-mvp/backend/tests/test_backfill_house_coords.py diff --git a/tradein-mvp/backend/scripts/README.md b/tradein-mvp/backend/scripts/README.md new file mode 100644 index 00000000..4a713ff2 --- /dev/null +++ b/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.` 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). diff --git a/tradein-mvp/backend/scripts/_yandex_reverse.py b/tradein-mvp/backend/scripts/_yandex_reverse.py index 05084eb1..2e01146c 100644 --- a/tradein-mvp/backend/scripts/_yandex_reverse.py +++ b/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 -runtime config and quota: +Three geocoding paths exposed: -- `reverse_via_api()` — Yandex Geocoder HTTP API. Fast, structured response, - needs a valid API key (env `YANDEX_GEOCODER_API_KEY`). Free tier is 25k - req/day so 200-row stratified sample is well within budget. +- `reverse_via_api()` — Yandex Geocoder HTTP API, lon/lat → address. Fast, + structured response, needs a valid API key (env `YANDEX_GEOCODER_API_KEY`). + 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 -implementation-agnostic. The `raw` field always carries the full source -payload — JSON for the API path, a small dict snapshot for the playwright -path (state object or selector text) — so we can do post-hoc diagnostics. +- `forward_via_api()` — address → lon/lat + canonical address (Phase 2 of + issue #582). Used by `backfill_house_coords.py` to fill `houses.lat/lon` + for the 4141 houses scraped from sources that didn't include coords (esp. + 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. - None if Yandex returned no match. + address: Human-readable canonical address Yandex returned. For + 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: " " — 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 # --------------------------------------------------------------------------- diff --git a/tradein-mvp/backend/scripts/audit_address_sample.sql b/tradein-mvp/backend/scripts/audit_address_sample.sql index b8465c1d..11e83b62 100644 --- a/tradein-mvp/backend/scripts/audit_address_sample.sql +++ b/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 --- (only houses with lat/lon falling inside one of the 8 admin polygons). --- 2. Partition by district, ROW_NUMBER() with a deterministic ORDER BY id --- so repeated runs return the same sample (deterministic = matches the --- "resumable per (house_id, audit_batch)" guarantee in 066 migration). --- 3. Keep top :limit_per_district per district → ~200 rows total for 8 --- districts × 25. +-- 1. Filter to houses with non-null lat/lon and non-empty address. +-- 2. Random shuffle via `ORDER BY random()` — repeatable enough for spot +-- sampling without needing a stable PRNG seed (the audit table dedupes +-- via UNIQUE (house_id, audit_batch), so re-running gives idempotent +-- results regardless of which rows land in the sample first). +-- 3. Cap the result at :limit_per_district * 8 rows — keeps the bind-param +-- 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 --- (psql variables resolve in :name form for the report; for Python use --- SQLAlchemy text() with bindparams). +-- :limit_per_district — kept for back-compat with the audit driver. +-- Effective sample size = :limit_per_district * 8 (e.g. 25 → 200). -- -- 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 - h.id, - h.address, - h.lat, - h.lon, - d.district_name AS district, - h.geom - FROM houses h - JOIN gendesign_ekb_districts_geom d - ON ST_Contains(d.geom, h.geom) - WHERE h.lat IS NOT NULL - AND h.lon IS NOT NULL - 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; +SELECT + h.id, + h.address, + h.lat, + h.lon, + NULL::text AS district +FROM houses h +WHERE h.lat IS NOT NULL + AND h.lon IS NOT NULL + AND h.address IS NOT NULL + AND length(trim(h.address)) > 0 +ORDER BY random() +LIMIT CAST(:limit_per_district AS int) * 8; diff --git a/tradein-mvp/backend/scripts/backfill_house_coords.py b/tradein-mvp/backend/scripts/backfill_house_coords.py new file mode 100644 index 00000000..7a607730 --- /dev/null +++ b/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()) diff --git a/tradein-mvp/backend/tests/test_backfill_house_coords.py b/tradein-mvp/backend/tests/test_backfill_house_coords.py new file mode 100644 index 00000000..2b8da977 --- /dev/null +++ b/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