lekss361/gendesign
1
0
Fork 0
Code Issues 57 Pull requests Projects Releases Packages Wiki Activity Actions

feat(tradein): Phase 2-3 of #582 — backfill houses + canonical reverse audit

Browse source 
Builds on Phase 1 audit infra (PR #583). Forward-geocode 4141 houses
без coords + reverse audit 4452 already-geocoded houses через Yandex
Geocoder API. Verified mismatch 870м для «Малышева 125» — наш point
в другом квартале от реального здания.

Scripts (new):
- `backfill_house_coords.py` (619 LOC) — driver:
  * Forward mode: UPDATE houses SET lat/lon когда precision ∈ {exact, number}
  * `--audit-only`: reverse геокод + distance check, flag >50м как mismatch
  * Per-row SAVEPOINT, idempotent через UNIQUE (house_id, audit_batch)
  * EKB locality bias (ll=60.6122,56.8389, spn=0.6,0.4)
  * Rate-limit 50ms между запросами, 25k/день квота Yandex покрывает 8.5k houses

- `_yandex_reverse.py` (+123 LOC) — added `forward_via_api()`,
  precision/kind в YandexReverseResult dataclass

- `audit_address_sample.sql` (-7 +47) — FIX: убрана зависимость от FDW
  `gendesign_ekb_districts_geom` (нет на проде). Простой
  `ORDER BY random()` без district stratification. Bind-param контракт
  сохранён (`:limit_per_district` * 8 = 200 default).

Tests:
- `test_backfill_house_coords.py` (510 LOC) — 20 tests: request shape,
  precision filter, UPDATE shape, backfill happy/imprecise/no_match/error,
  audit ok/mismatch/no_match, resumability, missing-key exit, mode routing
- Existing `test_audit_address_mismatch.py` still green
- 39/39 pass, ruff clean

Decisions (см. PR body):
- Precision filter = {exact, number} only (street/range/near → audit-imprecise)
- Mismatch threshold 50м (consistent с Phase 1 report)
- API-only (no Playwright fallback) — exit если key missing

Closes #582 (Phase 2-3 portion; Phase 1 was #583)
This commit is contained in:
Light1YT 2026-05-27 11:30:25 +05:00
parent 56a7a36c9a
commit 20b4a195b3
5 changed files with 1386 additions and 58 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

116
tradein-mvp/backend/scripts/README.md Normal file
View file

@ -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).

123
tradein-mvp/backend/scripts/_yandex_reverse.py
View file

@ -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: "<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
# ---------------------------------------------------------------------------

76
tradein-mvp/backend/scripts/audit_address_sample.sql
View file

@ -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;

619
tradein-mvp/backend/scripts/backfill_house_coords.py Normal file
View file

@ -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())

510
tradein-mvp/backend/tests/test_backfill_house_coords.py Normal file
View file

@ -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