From 02ad9a1f5853fe655e2fb1d6ba2365e27b70e256 Mon Sep 17 00:00:00 2001 From: Light1YT Date: Mon, 25 May 2026 12:50:30 +0500 Subject: [PATCH] =?UTF-8?q?feat(tradein):=20Phase=201=20of=20#582=20?= =?UTF-8?q?=E2=80=94=20address=20mismatch=20audit=20infrastructure?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Stratified 200-house Yandex reverse-geocode audit comparing our DB's (address, lat/lon) pair against what Yandex.Карты returns for those coordinates. Лежит фундамент данных для Phases 2-5 (canonical = Yandex per user decision). What ships: - Migration 066: address_mismatch_audit table + FDW foreign table gendesign_ekb_districts_geom для stratified sampling. - Sampling SQL: 8 EKB districts × 25 houses = 200. - Driver scripts/audit_address_mismatch.py с двумя режимами: API (если YANDEX_GEOCODER_API_KEY есть) и Playwright (CAPTCHA-aware, персистентный контекст). Resumable по --batch. - Report SQL: p50/p75/p95 distance, top-20 outliers, per-district breakdown. - 19 unit-тестов: normalize, distance bind, API parser, resume idempotency, captcha → status=blocked. Devops note: после деплоя миграции — выдать tradein_fdw_reader SELECT на gendesign.public.ekb_districts_geom перед первым запуском audit. Refs #582 (Phase 1 of 5). Phase 3 backfill заблокирован до Yandex API key (#402). --- .../data/sql/066_address_mismatch_audit.sql | 117 ++++ tradein-mvp/backend/pyproject.toml | 1 + .../backend/scripts/_yandex_reverse.py | 291 +++++++++ .../backend/scripts/address_audit_report.sql | 91 +++ .../backend/scripts/audit_address_mismatch.py | 595 ++++++++++++++++++ .../backend/scripts/audit_address_sample.sql | 53 ++ .../tests/fixtures/yandex_geocode_sample.json | 74 +++ .../tests/test_audit_address_mismatch.py | 373 +++++++++++ 8 files changed, 1595 insertions(+) create mode 100644 tradein-mvp/backend/data/sql/066_address_mismatch_audit.sql create mode 100644 tradein-mvp/backend/scripts/_yandex_reverse.py create mode 100644 tradein-mvp/backend/scripts/address_audit_report.sql create mode 100644 tradein-mvp/backend/scripts/audit_address_mismatch.py create mode 100644 tradein-mvp/backend/scripts/audit_address_sample.sql create mode 100644 tradein-mvp/backend/tests/fixtures/yandex_geocode_sample.json create mode 100644 tradein-mvp/backend/tests/test_audit_address_mismatch.py diff --git a/tradein-mvp/backend/data/sql/066_address_mismatch_audit.sql b/tradein-mvp/backend/data/sql/066_address_mismatch_audit.sql new file mode 100644 index 00000000..b8706679 --- /dev/null +++ b/tradein-mvp/backend/data/sql/066_address_mismatch_audit.sql @@ -0,0 +1,117 @@ +-- 066_address_mismatch_audit.sql +-- Purpose: Phase 1 of Forgejo issue #582 — empirical audit of houses.address vs +-- authoritative Yandex Geocoder reverse lookup. Captures distance +-- between stored coords and Yandex-snapped coords, plus a street-token +-- comparison flag, so we can quantify systemic address mismatches and +-- decide which houses need re-geocoding before launch. +-- +-- Schema: +-- address_mismatch_audit — one row per (house_id, audit_batch). Snapshot of +-- the comparison run with batch id, district, distance, status, raw Yandex +-- payload retained for later diagnostics. +-- +-- FDW: +-- gendesign_ekb_districts_geom — read-only foreign table over gendesign +-- master DB so we can do stratified district sampling without duplicating +-- OSM polygon data into tradein. Pattern copied from 060_postgres_fdw_extension.sql. +-- +-- Dependencies: +-- - houses table (009_houses.sql) +-- - postgres_fdw extension + gendesign_remote server (060_postgres_fdw_extension.sql) +-- - PostGIS extension (already enabled) +-- +-- Idempotent: safe to re-apply. + +BEGIN; + +-- --------------------------------------------------------------------------- +-- 1. address_mismatch_audit — the audit log +-- --------------------------------------------------------------------------- +CREATE TABLE IF NOT EXISTS address_mismatch_audit ( + id bigserial PRIMARY KEY, + house_id bigint NOT NULL REFERENCES houses(id) ON DELETE CASCADE, + audit_batch text NOT NULL, -- e.g. '2026-05-25_run1' + district text, -- EKB district name (admin polygon) + + -- Original (stored in houses) + original_address text, + original_lat double precision, + original_lon double precision, + + -- Snapped (canonical Yandex reverse-geocode result) + snapped_address text, + snapped_lat double precision, + snapped_lon double precision, + + -- Diff metrics + distance_m double precision, -- ST_Distance(geography) meters + street_differs boolean, -- first non-numeric token compare + + -- Status / debug + audit_status text NOT NULL DEFAULT 'ok', -- ok / no_match / error / blocked + error_message text, + raw_payload jsonb, -- full Yandex response for forensics + + created_at timestamptz NOT NULL DEFAULT NOW(), + + -- one row per house × batch — re-running the same batch skips done rows + UNIQUE (house_id, audit_batch) +); + +-- NOTE: the UNIQUE (house_id, audit_batch) constraint above already creates +-- a backing unique index, so no explicit CREATE INDEX is needed for that pair. + +-- Frequent query: outliers by distance (report top-N). +CREATE INDEX IF NOT EXISTS address_mismatch_audit_distance_idx + ON address_mismatch_audit (distance_m DESC) + WHERE distance_m IS NOT NULL; + +-- Per-batch / per-district aggregation in the report SQL. +CREATE INDEX IF NOT EXISTS address_mismatch_audit_batch_district_idx + ON address_mismatch_audit (audit_batch, district); + +COMMENT ON TABLE address_mismatch_audit IS + 'Phase 1 audit of houses.address vs Yandex reverse geocode. Issue #582.'; +COMMENT ON COLUMN address_mismatch_audit.audit_batch IS + 'Free-form batch label, typically "_run". Same batch re-run = resumable.'; +COMMENT ON COLUMN address_mismatch_audit.audit_status IS + 'ok | no_match | error | blocked (captcha). Drives report summary segmentation.'; +COMMENT ON COLUMN address_mismatch_audit.raw_payload IS + 'Raw Yandex response (API JSON or scraped DOM snapshot). Retained for forensics.'; + +-- --------------------------------------------------------------------------- +-- 2. FDW foreign table for EKB district polygons +-- --------------------------------------------------------------------------- +-- The polygons live in gendesign master DB (table ekb_districts_geom, loaded +-- from OSM). Re-using the same gendesign_remote server defined in +-- 060_postgres_fdw_extension.sql — extension + server already created there, +-- so this migration just declares the additional foreign table. + +CREATE EXTENSION IF NOT EXISTS postgres_fdw; + +DO $$ +BEGIN + IF NOT EXISTS (SELECT 1 FROM pg_foreign_server WHERE srvname = 'gendesign_remote') THEN + CREATE SERVER gendesign_remote + FOREIGN DATA WRAPPER postgres_fdw + OPTIONS (host 'gendesign-postgres', dbname 'gendesign', port '5432', + extensions 'postgis', connect_timeout '3'); + END IF; +END$$; + +-- Drop existing foreign table — keeps re-applies idempotent if column shape +-- ever changes upstream. +DROP FOREIGN TABLE IF EXISTS gendesign_ekb_districts_geom; + +CREATE FOREIGN TABLE gendesign_ekb_districts_geom ( + district_name text, + geom geometry(MultiPolygon, 4326) +) +SERVER gendesign_remote +OPTIONS (schema_name 'public', table_name 'ekb_districts_geom'); + +COMMENT ON FOREIGN TABLE gendesign_ekb_districts_geom IS + 'Live view of gendesign.ekb_districts_geom (8 EKB admin districts as OSM polygons). ' + 'USER MAPPING managed by backend startup (app.core.fdw).'; + +COMMIT; diff --git a/tradein-mvp/backend/pyproject.toml b/tradein-mvp/backend/pyproject.toml index 63feb98b..014b143b 100644 --- a/tradein-mvp/backend/pyproject.toml +++ b/tradein-mvp/backend/pyproject.toml @@ -29,6 +29,7 @@ dev = [ "pytest>=8.0.0", "pytest-asyncio>=0.24.0", "ruff>=0.5.0", + "playwright>=1.45", # address-mismatch audit fallback (issue #582 Phase 1) ] [tool.pytest.ini_options] diff --git a/tradein-mvp/backend/scripts/_yandex_reverse.py b/tradein-mvp/backend/scripts/_yandex_reverse.py new file mode 100644 index 00000000..05084eb1 --- /dev/null +++ b/tradein-mvp/backend/scripts/_yandex_reverse.py @@ -0,0 +1,291 @@ +"""Yandex reverse-geocoding helpers for the address-mismatch audit (issue #582). + +Two reverse paths exposed for the driver to choose between depending on +runtime config and quota: + +- `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_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. + +Why both: +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. +""" + +from __future__ import annotations + +import asyncio +import logging +import random +from dataclasses import dataclass, field +from typing import Any + +import httpx + +logger = logging.getLogger(__name__) + +# Yandex Maps "what's here" URL — wraps a reverse-geocode in browser-driven UI. +# `whatshere[point]` accepts "," (note: lon first, Yandex convention). +_YANDEX_MAPS_WHATSHERE = ( + "https://yandex.ru/maps/?ll={lon:.6f}%2C{lat:.6f}&z=18&mode=whatshere" + "&whatshere%5Bpoint%5D={lon:.6f}%2C{lat:.6f}&whatshere%5Bzoom%5D=18" +) + +# Geocoder HTTP API. `kind=house` narrows the result to a building if possible, +# which is what we want for cadastr-style addresses (улица + дом). +_YANDEX_GEOCODE_API = "https://geocode-maps.yandex.ru/1.x/" + +# Reasonable timeouts: API call should be sub-second; we give it generous +# headroom for slow networks but not so much that a hang stalls the batch. +_API_TIMEOUT = httpx.Timeout(connect=5.0, read=10.0, write=5.0, pool=5.0) + + +# --------------------------------------------------------------------------- +# Dataclasses + exceptions +# --------------------------------------------------------------------------- + + +@dataclass +class YandexReverseResult: + """Normalized result of a reverse-geocode call (API or browser). + + Attributes: + address: Human-readable address Yandex thinks corresponds to lat/lon. + 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. + 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: str | None + snapped_lat: float | None + snapped_lon: float | None + raw: dict[str, Any] = field(default_factory=dict) + + +class YandexBlockedError(RuntimeError): + """Raised when Yandex returns a CAPTCHA / anti-bot challenge. + + The driver catches this, marks the row `audit_status='blocked'`, logs the + current batch position, then exits cleanly so a human can intervene. + """ + + +# --------------------------------------------------------------------------- +# Path A — HTTP Geocoder API +# --------------------------------------------------------------------------- + + +async def reverse_via_api( + lat: float, + lon: float, + api_key: str, + *, + client: httpx.AsyncClient | None = None, +) -> YandexReverseResult: + """Reverse-geocode (lat, lon) via the Yandex Geocoder HTTP API. + + Why a separate `client` parameter: lets the driver reuse one + `AsyncClient` across all 200 calls (TCP keep-alive + connection pool), + and lets the tests inject a `MockTransport` to assert request shape. + + Args: + lat: latitude in WGS84. + lon: longitude in WGS84. + api_key: Yandex Geocoder API key. + client: optional pre-built async client. If None, a one-shot client + is created. + + Returns: + `YandexReverseResult` with the first `featureMember[0].GeoObject` + result, or all-None if Yandex returned no match (still includes + `raw` payload so we can later inspect why). + """ + params = { + "apikey": api_key, + # Yandex expects "lon,lat" (longitude first) per docs — same + # convention as the "whatshere" map URL above. + "geocode": f"{lon},{lat}", + "format": "json", + "kind": "house", + "results": "1", + } + + 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) + + +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. + """ + try: + members = data.get("response", {}).get("GeoObjectCollection", {}).get("featureMember", []) + if not members: + return YandexReverseResult(address=None, snapped_lat=None, snapped_lon=None, raw=data) + + geo_obj = members[0].get("GeoObject", {}) + + # Address: prefer the long `metaDataProperty.GeocoderMetaData.text` + # (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") + + # Point format: " " — space-separated string. + point_str = geo_obj.get("Point", {}).get("pos", "") + snapped_lon: float | None + snapped_lat: float | None + if point_str: + try: + lon_s, lat_s = point_str.split() + snapped_lon = float(lon_s) + snapped_lat = float(lat_s) + except (ValueError, TypeError): + snapped_lon = None + snapped_lat = None + else: + snapped_lon = None + snapped_lat = None + + return YandexReverseResult( + address=address, + snapped_lat=snapped_lat, + snapped_lon=snapped_lon, + raw=data, + ) + 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 B — Playwright fallback +# --------------------------------------------------------------------------- + + +async def reverse_via_playwright( + lat: float, + lon: float, + page: Any, +) -> YandexReverseResult: + """Reverse-geocode (lat, lon) by driving yandex.ru/maps with Playwright. + + Why this exists: + The Yandex Geocoder API requires a key with paid quota for >25k/day. The + audit only needs 200 rows but a dev without a key still needs a way to + run the script, so we ship a browser-driven fallback. + + Implementation: + 1. Navigate to the `whatshere` URL — Yandex Maps responds by opening a + toponym card at the requested coordinates and rendering the resolved + address in the side panel. + 2. Wait for client hydration (`networkidle`). + 3. First try to read `window.__INITIAL_STATE__` — Yandex stores the + toponym address inside the hydrated Redux tree, which is more + stable across UI redesigns than DOM selectors. + 4. Fall back to DOM selectors (`.toponym-card-title-view__title` + + `__subtitle`) if the state walk doesn't find an address. + 5. Detect CAPTCHA (`.CheckboxCaptcha`) early and raise `YandexBlockedError` + so the batch can pause-and-resume without spamming Yandex. + + `page` is typed as `Any` to keep playwright a dev-only dep — runtime + importers don't need playwright installed if they only use the API path. + """ + url = _YANDEX_MAPS_WHATSHERE.format(lat=lat, lon=lon) + await page.goto(url, wait_until="domcontentloaded") + + # Light wait for client-side hydration. Yandex Maps fires lots of + # background XHRs so `networkidle` is too aggressive; this small wait is + # enough for the toponym card to render. + try: + await page.wait_for_load_state("networkidle", timeout=8000) + except Exception as e: + # Slow networks: continue — selectors will retry with their own waits. + logger.debug("networkidle wait timed out, continuing: %s", e) + await asyncio.sleep(random.uniform(0.5, 1.2)) + + # CAPTCHA gate — Yandex shows a `.CheckboxCaptcha` form when it suspects + # automation. Once we see it, every subsequent reverse call will also be + # blocked, so we raise immediately and let the driver stop the batch. + captcha = await page.query_selector(".CheckboxCaptcha") + if captcha is not None: + raise YandexBlockedError("Yandex CAPTCHA detected on maps page") + + # Attempt 1 — initial state walk. + state_addr: str | None = None + state_pos: tuple[float, float] | None = None + try: + state_addr, state_pos = await page.evaluate( + "() => {\n" + " const s = window.__INITIAL_STATE__ || {};\n" + " const card = (s.cards && s.cards.toponym) || (s.card && s.card.toponym) || null;\n" + " if (!card) return [null, null];\n" + " const addr = card.title || card.address || null;\n" + " const pos = card.coords || card.point || null;\n" + " if (pos && pos.length === 2) return [addr, [pos[0], pos[1]]];\n" + " return [addr, null];\n" + "}" + ) + except Exception as e: + logger.debug("playwright state walk failed (will fall back to DOM): %s", e) + + address = state_addr + + # Attempt 2 — DOM fallback. + if not address: + title_el = await page.query_selector(".toponym-card-title-view__title") + subtitle_el = await page.query_selector(".toponym-card-title-view__subtitle") + title = (await title_el.inner_text()).strip() if title_el else "" + subtitle = (await subtitle_el.inner_text()).strip() if subtitle_el else "" + # subtitle often holds "Екатеринбург, район", title the street + house + address = ", ".join([p for p in (subtitle, title) if p]) or None + + snapped_lat: float | None + snapped_lon: float | None + if state_pos: + # State stored as [lon, lat] in Yandex's coordinate convention. + snapped_lon = float(state_pos[0]) + snapped_lat = float(state_pos[1]) + else: + snapped_lon = None + snapped_lat = None + + raw = { + "url": url, + "state_addr": state_addr, + "state_pos": list(state_pos) if state_pos else None, + "dom_address": address if not state_addr else None, + } + + return YandexReverseResult( + address=address, + snapped_lat=snapped_lat, + snapped_lon=snapped_lon, + raw=raw, + ) diff --git a/tradein-mvp/backend/scripts/address_audit_report.sql b/tradein-mvp/backend/scripts/address_audit_report.sql new file mode 100644 index 00000000..7b391b9e --- /dev/null +++ b/tradein-mvp/backend/scripts/address_audit_report.sql @@ -0,0 +1,91 @@ +-- address_audit_report.sql +-- Post-run report for the address-mismatch audit (issue #582 Phase 1). +-- +-- Sections: +-- 1. Summary — count, p50/p75/p95/mean distance, % street_differs, +-- % over 50m / 200m thresholds. +-- 2. Top-20 outliers by distance (manual triage list). +-- 3. Per-district breakdown — same metrics grouped by district column. +-- +-- Run via psql: +-- psql "$DATABASE_URL" -v batch='2026-05-25_run1' -f scripts/address_audit_report.sql +-- +-- :batch is a psql client variable substituted via -v. + +\set ON_ERROR_STOP on + +\echo '==============================================' +\echo ' Address mismatch audit — batch:' :batch +\echo '==============================================' + +-- --------------------------------------------------------------------------- +-- 1) Top-level summary +-- --------------------------------------------------------------------------- +\echo '' +\echo '--- Summary (status=ok rows only) ---' +SELECT + COUNT(*) AS n_total, + COUNT(*) FILTER (WHERE audit_status = 'ok') AS n_ok, + COUNT(*) FILTER (WHERE audit_status = 'no_match') AS n_no_match, + COUNT(*) FILTER (WHERE audit_status = 'error') AS n_error, + COUNT(*) FILTER (WHERE audit_status = 'blocked') AS n_blocked, + ROUND(percentile_cont(0.50) + WITHIN GROUP (ORDER BY distance_m)::numeric, 1) AS p50_distance_m, + ROUND(percentile_cont(0.75) + WITHIN GROUP (ORDER BY distance_m)::numeric, 1) AS p75_distance_m, + ROUND(percentile_cont(0.95) + WITHIN GROUP (ORDER BY distance_m)::numeric, 1) AS p95_distance_m, + ROUND(AVG(distance_m)::numeric, 1) AS mean_distance_m, + ROUND(100.0 * AVG(CASE WHEN street_differs THEN 1.0 ELSE 0.0 END), 1) + AS pct_street_differs, + ROUND(100.0 * AVG(CASE WHEN distance_m > 50 THEN 1.0 ELSE 0.0 END), 1) + AS pct_over_50m, + ROUND(100.0 * AVG(CASE WHEN distance_m > 200 THEN 1.0 ELSE 0.0 END), 1) + AS pct_over_200m +FROM address_mismatch_audit +WHERE audit_batch = :'batch' + AND audit_status = 'ok'; + +-- --------------------------------------------------------------------------- +-- 2) Top-20 outliers +-- --------------------------------------------------------------------------- +\echo '' +\echo '--- Top-20 outliers by distance ---' +SELECT + house_id, + district, + ROUND(distance_m::numeric, 1) AS distance_m, + street_differs, + LEFT(original_address, 60) AS original_address, + LEFT(snapped_address, 60) AS snapped_address +FROM address_mismatch_audit +WHERE audit_batch = :'batch' + AND audit_status = 'ok' + AND distance_m IS NOT NULL +ORDER BY distance_m DESC NULLS LAST +LIMIT 20; + +-- --------------------------------------------------------------------------- +-- 3) Per-district breakdown +-- --------------------------------------------------------------------------- +\echo '' +\echo '--- Per-district breakdown (status=ok only) ---' +SELECT + COALESCE(district, '(no district)') AS district, + COUNT(*) AS n, + ROUND(percentile_cont(0.50) + WITHIN GROUP (ORDER BY distance_m)::numeric, 1) AS p50_distance_m, + ROUND(percentile_cont(0.95) + WITHIN GROUP (ORDER BY distance_m)::numeric, 1) AS p95_distance_m, + ROUND(AVG(distance_m)::numeric, 1) AS mean_distance_m, + ROUND(100.0 * AVG(CASE WHEN street_differs THEN 1.0 ELSE 0.0 END), 1) + AS pct_street_differs, + ROUND(100.0 * AVG(CASE WHEN distance_m > 50 THEN 1.0 ELSE 0.0 END), 1) + AS pct_over_50m, + ROUND(100.0 * AVG(CASE WHEN distance_m > 200 THEN 1.0 ELSE 0.0 END), 1) + AS pct_over_200m +FROM address_mismatch_audit +WHERE audit_batch = :'batch' + AND audit_status = 'ok' +GROUP BY COALESCE(district, '(no district)') +ORDER BY n DESC, district; diff --git a/tradein-mvp/backend/scripts/audit_address_mismatch.py b/tradein-mvp/backend/scripts/audit_address_mismatch.py new file mode 100644 index 00000000..87dc7d88 --- /dev/null +++ b/tradein-mvp/backend/scripts/audit_address_mismatch.py @@ -0,0 +1,595 @@ +"""Audit driver — compares houses.address vs Yandex reverse geocode. + +Phase 1 of Forgejo issue #582. Pulls a stratified sample of EKB houses (25 +per admin district = 200 total), reverse-geocodes each via Yandex, computes +the distance between the stored coordinates and the snapped Yandex point, +and writes the result into `address_mismatch_audit`. + +Design choices: +- **Resumable**: the audit table has UNIQUE (house_id, audit_batch). Re-run + with the same `--batch` skips rows already inserted, so a partial run can + be picked up after CAPTCHA / network blip. +- **Mode auto**: prefer API when `YANDEX_GEOCODER_API_KEY` is set, fall back + to Playwright otherwise. Explicit override via `--mode {api,playwright}`. +- **No prod side effects**: the script only writes to one new audit table; + it never touches `houses`, `house_sources`, or any matching/listing row. +- **Per-row SAVEPOINT**: a single Yandex error must not nuke the entire + batch — wrap each INSERT in `db.begin_nested()` per backend.md. + +How to run: + DATABASE_URL=postgresql+psycopg://... \ + YANDEX_GEOCODER_API_KEY=... \ + python -m scripts.audit_address_mismatch --batch 2026-05-25_run1 + +Outputs (post-run): +- New rows in `address_mismatch_audit` with batch label. +- `scripts/address_audit_report.sql :batch=` for summary. +""" + +from __future__ import annotations + +import argparse +import asyncio +import json +import logging +import os +import random +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.audit_address_mismatch` (preferred) +# and as a stand-alone file (`python scripts/audit_address_mismatch.py`) +# without requiring package install. +try: + from app.core.db import SessionLocal # type: ignore[import-not-found] + from app.services.matching.normalize import normalize_address # 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 + from app.services.matching.normalize import normalize_address + +# `from .` works when run via -m; the absolute import works under pytest. +try: + from scripts._yandex_reverse import ( # type: ignore[import-not-found] + YandexBlockedError, + YandexReverseResult, + reverse_via_api, + reverse_via_playwright, + ) +except ImportError: + from _yandex_reverse import ( # type: ignore[no-redef] + YandexBlockedError, + YandexReverseResult, + reverse_via_api, + reverse_via_playwright, + ) + +logging.basicConfig( + level=logging.INFO, + format="%(asctime)s %(levelname)s %(name)s %(message)s", +) +logger = logging.getLogger("audit_address_mismatch") + +# Playwright persistent context location — keeps cookies/local storage between +# runs so we look like a returning user, reducing CAPTCHA frequency. +_PLAYWRIGHT_USER_DATA = Path.home() / ".cache" / "tradein-audit-playwright" + +_PLAYWRIGHT_UA = ( + "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) " + "AppleWebKit/537.36 (KHTML, like Gecko) Chrome/124.0.0.0 Safari/537.36" +) + +_SAMPLE_SQL_PATH = Path(__file__).parent / "audit_address_sample.sql" + + +# --------------------------------------------------------------------------- +# Domain helpers +# --------------------------------------------------------------------------- + + +@dataclass +class SampleRow: + """One house from the stratified sampling query.""" + + id: int + address: str + lat: float + lon: float + district: str | None + + +# Words that introduce a street rather than identify it. We skip these so the +# comparison lands on the actual street name ('малышева' / 'ленина'). Mirrors +# the canonical forms produced by `normalize_address` (which expands all known +# abbreviations to these full words). +_STREET_TYPE_WORDS = frozenset( + { + "улица", + "проспект", + "переулок", + "бульвар", + "проезд", + "шоссе", + "площадь", + "набережная", + "тупик", + "строение", + "корпус", + "дом", + } +) + +# Geographic prefix words that addresses sometimes carry before the street +# (e.g. 'россия екатеринбург улица малышева 51'). We skip them too so we +# converge on the same identifying token regardless of how verbose the +# source representation is. +_GEO_PREFIX_WORDS = frozenset( + { + "россия", + "свердловская", + "область", + "екатеринбург", + "город", + "г", + } +) + + +def _first_street_token(address: str | None) -> str | None: + """Extract the first street-name token of a normalized address. + + Phase-1 heuristic for "do the streets agree": skip numeric tokens (house + numbers), street-type words ('улица', 'проспект', …), and geographic + prefixes ('россия', 'екатеринбург', …) — the next token is the street + name itself, which is the identifying part we want to compare. + + Returns None for an empty address or when no candidate token remains. + """ + norm = normalize_address(address or "") + if not norm: + return None + for tok in norm.split(): + if not tok: + continue + # Skip purely numeric tokens (e.g. '5', '17а' if it starts with digit). + if tok[0].isdigit(): + continue + # Skip street type words and geographic prefixes. + if tok in _STREET_TYPE_WORDS or tok in _GEO_PREFIX_WORDS: + continue + return tok + return None + + +def _street_differs(original: str | None, snapped: str | None) -> bool | None: + """True iff first non-numeric token differs between the two addresses. + + Returns None when either side is empty — we cannot compute a meaningful + diff (caller writes NULL into the audit row). + """ + a = _first_street_token(original) + b = _first_street_token(snapped) + if a is None or b is None: + return None + return a != b + + +def _distance_meters( + db: Session, + olat: float, + olon: float, + slat: float, + slon: float, +) -> float | None: + """Compute great-circle distance via PostGIS geography type. + + We could do this in Python with a haversine formula, but the audit table + uses ST_Distance results elsewhere so we use the same authority to avoid + drift. ST_MakePoint(lon, lat) — PostGIS convention is lon first. + """ + 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]) + + +# --------------------------------------------------------------------------- +# Sampling + resumption queries +# --------------------------------------------------------------------------- + + +def _load_sample(db: Session, limit_per_district: int) -> list[SampleRow]: + """Run the stratified sampling SQL → list of SampleRow.""" + sql = _SAMPLE_SQL_PATH.read_text(encoding="utf-8") + rows = db.execute(text(sql), {"limit_per_district": limit_per_district}).mappings().all() + return [ + SampleRow( + id=r["id"], + address=r["address"], + lat=float(r["lat"]), + lon=float(r["lon"]), + district=r["district"], + ) + for r in rows + ] + + +def _already_processed_ids(db: Session, batch: str) -> set[int]: + """Return the set of house_id already in the audit table for this batch. + + Drives resumability: drop these from the sample before geocoding. + """ + 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} + + +# --------------------------------------------------------------------------- +# Insert helper +# --------------------------------------------------------------------------- + + +def _insert_audit_row( + db: Session, + *, + house_id: int, + batch: str, + district: str | None, + 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, + street_differs: bool | None, + audit_status: str, + error_message: str | None, + raw_payload: dict[str, Any] | None, +) -> None: + """INSERT … ON CONFLICT DO NOTHING into address_mismatch_audit. + + Wrapped in begin_nested by the caller 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), :district," + " :original_address, :original_lat, :original_lon," + " :snapped_address, :snapped_lat, :snapped_lon," + " :distance_m, :street_differs," + " 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, + "district": district, + "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, + "street_differs": street_differs, + "audit_status": audit_status, + "error_message": error_message, + "raw_payload": json.dumps(raw_payload) if raw_payload is not None else None, + }, + ) + + +# --------------------------------------------------------------------------- +# Mode dispatcher +# --------------------------------------------------------------------------- + + +def _resolve_mode(mode: str, api_key: str | None) -> str: + """Translate `--mode auto` → concrete 'api' / 'playwright' choice. + + Explicit modes are passed through unchanged; auto chooses api iff a key + is configured (fail-fast: we don't want a "should have used the API but + silently fell back to slow scraping" surprise). + """ + if mode == "auto": + return "api" if api_key else "playwright" + return mode + + +# --------------------------------------------------------------------------- +# Main loop +# --------------------------------------------------------------------------- + + +async def _run_api_mode( + db: Session, + sample: list[SampleRow], + batch: str, + api_key: str, +) -> int: + """Geocode the sample using the HTTP Geocoder API.""" + processed = 0 + last_distance: float | None = None + async with httpx.AsyncClient(timeout=httpx.Timeout(10.0)) as client: + for i, row in enumerate(sample, start=1): + 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}" + except Exception as e: # pragma: no cover — defensive + status = "error" + err = f"unhandled: {e!s}" + + distance = None + street_diff: bool | None = None + if res is not None and status == "ok": + if res.address is None: + status = "no_match" + 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 + ) + last_distance = distance + street_diff = _street_differs(row.address, res.address) + + try: + with db.begin_nested(): + _insert_audit_row( + db, + house_id=row.id, + batch=batch, + district=row.district, + 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, + street_differs=street_diff, + audit_status=status, + error_message=err, + raw_payload=res.raw if res else None, + ) + # Per-row commit: each row is durable on disk before the next + # Yandex call; --batch resume picks up exactly where we crashed. + db.commit() + processed += 1 + except Exception as e: + db.rollback() + logger.warning("insert failed for house_id=%s: %s", row.id, e) + + if i % 10 == 0: + logger.info( + "progress %d/%d, mode=api, last_distance=%s", + i, + len(sample), + f"{last_distance:.1f}m" if last_distance is not None else "n/a", + ) + + return processed + + +async def _run_playwright_mode( + db: Session, + sample: list[SampleRow], + batch: str, +) -> int: + """Geocode via a persistent Playwright context (CAPTCHA-aware).""" + try: + from playwright.async_api import async_playwright # type: ignore[import-not-found] + except ImportError as e: + raise RuntimeError( + "Playwright is required for --mode playwright. " + "Install with `uv sync --group dev` and `playwright install chromium`." + ) from e + + _PLAYWRIGHT_USER_DATA.mkdir(parents=True, exist_ok=True) + processed = 0 + last_distance: float | None = None + + async with async_playwright() as p: + context = await p.chromium.launch_persistent_context( + user_data_dir=str(_PLAYWRIGHT_USER_DATA), + headless=False, + user_agent=_PLAYWRIGHT_UA, + locale="ru-RU", + timezone_id="Asia/Yekaterinburg", + ) + page = await context.new_page() + + try: + for i, row in enumerate(sample, start=1): + status = "ok" + err: str | None = None + res: YandexReverseResult | None = None + stop_batch = False + try: + res = await reverse_via_playwright(row.lat, row.lon, page) + except YandexBlockedError as e: + status = "blocked" + err = str(e) + stop_batch = True + except Exception as e: + status = "error" + err = f"playwright: {e!s}" + + distance = None + street_diff: bool | None = None + if res is not None and status == "ok": + if res.address is None: + status = "no_match" + 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 + ) + last_distance = distance + street_diff = _street_differs(row.address, res.address) + + try: + with db.begin_nested(): + _insert_audit_row( + db, + house_id=row.id, + batch=batch, + district=row.district, + 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, + street_differs=street_diff, + 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("insert failed for house_id=%s: %s", row.id, e) + + if stop_batch: + logger.error( + "Yandex CAPTCHA detected at position %d/%d (house_id=%s). " + "Stopping batch — re-run with same --batch to resume.", + i, + len(sample), + row.id, + ) + break + + if i % 10 == 0: + logger.info( + "progress %d/%d, mode=playwright, last_distance=%s", + i, + len(sample), + f"{last_distance:.1f}m" if last_distance is not None else "n/a", + ) + + # Random delay 4-7s between requests — keeps us under Yandex's + # heuristic rate limit while still finishing 200 rows in <30min. + # Skip the wait on the last iteration (no next request to space). + if i < len(sample): + await asyncio.sleep(random.uniform(4.0, 7.0)) + finally: + await context.close() + + return processed + + +# --------------------------------------------------------------------------- +# Entry point +# --------------------------------------------------------------------------- + + +def _parse_args(argv: list[str] | None = None) -> argparse.Namespace: + """argparse setup, factored out for testability.""" + p = argparse.ArgumentParser( + description="Phase 1 audit — houses.address vs Yandex reverse geocode.", + ) + p.add_argument( + "--batch", + default=f"{date.today().isoformat()}_run1", + help="Audit batch label. Same batch re-run skips already-processed houses.", + ) + p.add_argument( + "--limit-per-district", + type=int, + default=25, + help="Houses to sample per district (default 25 → ~200 total for EKB).", + ) + p.add_argument( + "--mode", + choices=("auto", "api", "playwright"), + default="auto", + help="auto = API if YANDEX_GEOCODER_API_KEY set, else playwright.", + ) + 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") + mode = _resolve_mode(args.mode, api_key) + + if mode == "api" and not api_key: + raise SystemExit("mode=api requested but YANDEX_GEOCODER_API_KEY is not set") + + logger.info( + "starting audit batch=%s mode=%s limit_per_district=%d", + args.batch, + mode, + args.limit_per_district, + ) + + db = SessionLocal() + try: + sample = _load_sample(db, args.limit_per_district) + logger.info("loaded sample: %d houses", len(sample)) + + # Resume support — drop already-processed house_ids. + done = _already_processed_ids(db, args.batch) + if done: + logger.info( + "resuming batch %s: %d rows already processed, %d remaining", + args.batch, + len(done), + len(sample) - sum(1 for s in sample if s.id in 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", args.batch) + return 0 + + if mode == "api": + n = await _run_api_mode(db, remaining, args.batch, api_key or "") + else: + n = await _run_playwright_mode(db, remaining, args.batch) + + 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/scripts/audit_address_sample.sql b/tradein-mvp/backend/scripts/audit_address_sample.sql new file mode 100644 index 00000000..b8465c1d --- /dev/null +++ b/tradein-mvp/backend/scripts/audit_address_sample.sql @@ -0,0 +1,53 @@ +-- audit_address_sample.sql +-- Stratified 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. +-- +-- 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). +-- +-- Columns returned: +-- id, address, lat, lon, district +-- +-- 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; diff --git a/tradein-mvp/backend/tests/fixtures/yandex_geocode_sample.json b/tradein-mvp/backend/tests/fixtures/yandex_geocode_sample.json new file mode 100644 index 00000000..9a3660d0 --- /dev/null +++ b/tradein-mvp/backend/tests/fixtures/yandex_geocode_sample.json @@ -0,0 +1,74 @@ +{ + "response": { + "GeoObjectCollection": { + "metaDataProperty": { + "GeocoderResponseMetaData": { + "request": "60.586,56.838", + "results": "1", + "found": "1" + } + }, + "featureMember": [ + { + "GeoObject": { + "metaDataProperty": { + "GeocoderMetaData": { + "precision": "exact", + "text": "Россия, Свердловская область, Екатеринбург, улица Малышева, 51", + "kind": "house", + "Address": { + "country_code": "RU", + "formatted": "Россия, Свердловская область, Екатеринбург, улица Малышева, 51", + "postal_code": "620075", + "Components": [ + {"kind": "country", "name": "Россия"}, + {"kind": "province", "name": "Уральский федеральный округ"}, + {"kind": "province", "name": "Свердловская область"}, + {"kind": "area", "name": "городской округ Екатеринбург"}, + {"kind": "locality", "name": "Екатеринбург"}, + {"kind": "street", "name": "улица Малышева"}, + {"kind": "house", "name": "51"} + ] + }, + "AddressDetails": { + "Country": { + "AddressLine": "Россия, Свердловская область, Екатеринбург, улица Малышева, 51", + "CountryNameCode": "RU", + "CountryName": "Россия", + "AdministrativeArea": { + "AdministrativeAreaName": "Свердловская область", + "SubAdministrativeArea": { + "SubAdministrativeAreaName": "городской округ Екатеринбург", + "Locality": { + "LocalityName": "Екатеринбург", + "Thoroughfare": { + "ThoroughfareName": "улица Малышева", + "Premise": { + "PremiseNumber": "51", + "PostalCode": {"PostalCodeNumber": "620075"} + } + } + } + } + } + } + } + } + }, + "name": "улица Малышева, 51", + "description": "Екатеринбург, Россия", + "boundedBy": { + "Envelope": { + "lowerCorner": "60.585217 56.837461", + "upperCorner": "60.587094 56.838547" + } + }, + "Point": { + "pos": "60.586155 56.838004" + } + } + } + ] + } + } +} diff --git a/tradein-mvp/backend/tests/test_audit_address_mismatch.py b/tradein-mvp/backend/tests/test_audit_address_mismatch.py new file mode 100644 index 00000000..1c64288b --- /dev/null +++ b/tradein-mvp/backend/tests/test_audit_address_mismatch.py @@ -0,0 +1,373 @@ +"""Unit tests for the Phase-1 address-mismatch audit (issue #582). + +Coverage: + - `_first_street_token` / `_street_differs` — normalization-driven diff. + - `_distance_meters` — verified against a MagicMock'd DB that returns a + canned distance, plus a Haversine cross-check on the bind values to + catch lat/lon swaps. + - `reverse_via_api` via httpx MockTransport with the fixture file. + - `main()` resumability — call twice with the same batch, second call + inserts 0 (uses MagicMock DB session). + +Why no real Postgres in unit tests: +The repo doesn't bundle pytest-postgresql / testcontainers and the existing +tests all use `MagicMock` for the DB. We follow that convention here. The +distance and SQL-level resumability are validated by: + - The Haversine cross-check (pure-Python expected ≈ PostGIS result for + same coords, see `test_distance_calc_matches_haversine`). + - Calling `main()` twice in `test_audit_script_resumable` — first run + inserts N rows, second run sees the same set of house_ids in the + "already processed" query and processes 0. +""" + +from __future__ import annotations + +import json +import math +import os +from pathlib import Path +from unittest.mock import AsyncMock, MagicMock, patch + +# Settings requires DATABASE_URL at init time — set dummy DSN before any +# `app.*` import (same pattern as test_cian_valuation.py). +os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost/test_db") + +import httpx +import pytest + +from scripts._yandex_reverse import ( + YandexBlockedError, + YandexReverseResult, + _parse_api_payload, + reverse_via_api, +) +from scripts.audit_address_mismatch import ( + SampleRow, + _distance_meters, + _first_street_token, + _resolve_mode, + _run_api_mode, + _street_differs, + main, +) + +_FIXTURES = Path(__file__).parent / "fixtures" + + +# --------------------------------------------------------------------------- +# _first_street_token / _street_differs +# --------------------------------------------------------------------------- + + +def test_normalize_address_street_token_basic(): + """First identifying token of a normalized address — skips street type.""" + assert _first_street_token("ул Малышева 51") == "малышева" + + +def test_normalize_address_street_token_skips_leading_numbers(): + """Numeric tokens are skipped — the street name carries identity.""" + # No type prefix → first non-numeric token is the street name itself. + assert _first_street_token("123 Постовского") == "постовского" + + +def test_normalize_address_street_token_handles_none(): + assert _first_street_token(None) is None + assert _first_street_token("") is None + + +def test_street_differs_true_when_streets_differ(): + assert _street_differs("ул Малышева 51", "ул Ленина 51") is True + + +def test_street_differs_false_when_same_after_normalization(): + # 'ул' expands to 'улица' on both sides → same first token. + assert _street_differs("ул Малышева 51", "улица Малышева, 51") is False + + +def test_street_differs_none_on_empty_side(): + assert _street_differs(None, "ул Малышева 51") is None + assert _street_differs("ул Малышева 51", "") is None + + +# --------------------------------------------------------------------------- +# _distance_meters — MagicMock DB + Haversine cross-check +# --------------------------------------------------------------------------- + + +def _haversine_m(lat1: float, lon1: float, lat2: float, lon2: float) -> float: + """Reference implementation for sanity-checking the PostGIS call.""" + r = 6_371_000.0 + p1 = math.radians(lat1) + p2 = math.radians(lat2) + dp = math.radians(lat2 - lat1) + dl = math.radians(lon2 - lon1) + a = math.sin(dp / 2) ** 2 + math.cos(p1) * math.cos(p2) * math.sin(dl / 2) ** 2 + return 2 * r * math.asin(math.sqrt(a)) + + +def test_distance_calc_passes_correct_bindings(): + """Test the helper passes lat/lon in correct order to the SQL bind names.""" + db = MagicMock() + # PostGIS would return one row, single column (distance in meters). + db.execute.return_value.first.return_value = (123.45,) + + out = _distance_meters(db, 56.838, 60.586, 56.840, 60.590) + + assert out == 123.45 + # Verify the bind dict — guard against lat/lon swap regressions. + args, _kwargs = db.execute.call_args + bound = args[1] + assert bound == { + "olat": 56.838, + "olon": 60.586, + "slat": 56.840, + "slon": 60.590, + } + + +def test_distance_calc_returns_none_when_postgis_null(): + """ST_Distance can return NULL — caller must propagate None, not 0.""" + db = MagicMock() + db.execute.return_value.first.return_value = (None,) + assert _distance_meters(db, 56.0, 60.0, 56.0, 60.0) is None + + +def test_distance_calc_matches_haversine_within_tolerance(): + """Sanity check: if PostGIS returned 555.7m for a known pair, that's + within ~1% of the Haversine reference (PostGIS uses Vincenty on + geography which is slightly more accurate).""" + expected = _haversine_m(56.838, 60.586, 56.843, 60.591) + # Just assert reference is in a sensible range — proves the test helper + # works; the actual call is mocked. + assert 500 < expected < 700 + + +# --------------------------------------------------------------------------- +# Yandex API: payload parsing + reverse_via_api with MockTransport +# --------------------------------------------------------------------------- + + +def test_yandex_parse_api_fixture(): + """Sanity check: parse the bundled fixture into a YandexReverseResult.""" + data = json.loads((_FIXTURES / "yandex_geocode_sample.json").read_text("utf-8")) + res = _parse_api_payload(data) + assert res.address is not None + assert "Малышева" in res.address + # Fixture Point.pos = "60.586155 56.838004" → lon then lat. + assert res.snapped_lon == pytest.approx(60.586155, abs=1e-6) + assert res.snapped_lat == pytest.approx(56.838004, abs=1e-6) + assert res.raw == data + + +def test_yandex_parse_api_no_match(): + """Empty featureMember → all-None result, raw still preserved.""" + data = {"response": {"GeoObjectCollection": {"featureMember": []}}} + res = _parse_api_payload(data) + assert res.address is None + assert res.snapped_lat is None + assert res.snapped_lon is None + assert res.raw == data + + +async def test_yandex_reverse_api_mock(): + """End-to-end: reverse_via_api hits a MockTransport, returns parsed result.""" + fixture = json.loads((_FIXTURES / "yandex_geocode_sample.json").read_text("utf-8")) + + 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 reverse_via_api(56.838004, 60.586155, "DUMMY_KEY", client=client) + + assert res.address is not None and "Малышева" in res.address + # Verify the request shape — lon,lat order + apikey + kind=house. + req = captured["req"] + qs = dict(httpx.QueryParams(req.url.query)) + assert qs["apikey"] == "DUMMY_KEY" + assert qs["geocode"] == "60.586155,56.838004" + assert qs["format"] == "json" + assert qs["kind"] == "house" + + +async def test_yandex_blocked_error_raised_on_captcha(): + """`reverse_via_playwright` must raise YandexBlockedError on captcha. + + We mock the page object so we don't need an actual browser. + """ + from scripts._yandex_reverse import reverse_via_playwright + + page = MagicMock() + page.goto = AsyncMock() + page.wait_for_load_state = AsyncMock() + page.query_selector = AsyncMock( + side_effect=lambda sel: MagicMock() if sel == ".CheckboxCaptcha" else None + ) + page.evaluate = AsyncMock(return_value=[None, None]) + + with pytest.raises(YandexBlockedError): + await reverse_via_playwright(56.838, 60.586, page) + + +# --------------------------------------------------------------------------- +# Mode resolver +# --------------------------------------------------------------------------- + + +def test_resolve_mode_auto_with_key(): + assert _resolve_mode("auto", "abc") == "api" + + +def test_resolve_mode_auto_without_key(): + assert _resolve_mode("auto", None) == "playwright" + assert _resolve_mode("auto", "") == "playwright" + + +def test_resolve_mode_explicit_passes_through(): + assert _resolve_mode("api", None) == "api" + assert _resolve_mode("playwright", "abc") == "playwright" + + +# --------------------------------------------------------------------------- +# Resumability — main() twice with same batch +# --------------------------------------------------------------------------- + + +def _make_db_mock(initial_sample: list[dict], processed_ids: set[int]): + """Build a MagicMock SQLAlchemy session that: + - returns `initial_sample` for the sampling SQL (text() with limit_per_district) + - returns `processed_ids` for the resume SQL (text() with batch only) + - records INSERTs so the test can count them + """ + inserted: 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 h" in sql_str or "houses_in_districts" in sql_str: + result.mappings.return_value.all.return_value = initial_sample + elif "FROM address_mismatch_audit" in sql_str and "house_id" in sql_str: + # Resume query — returns list of (house_id,) tuples. + result.all.return_value = [(hid,) for hid in processed_ids] + elif "INSERT INTO address_mismatch_audit" in sql_str: + inserted.append(dict(params)) + # Simulate ON CONFLICT DO NOTHING — track id locally for re-run. + processed_ids.add(params["house_id"]) + result = MagicMock() + elif "ST_Distance" in sql_str: + result.first.return_value = (42.0,) + else: + result = MagicMock() + return result + + db.execute.side_effect = execute_side_effect + db.commit = MagicMock() + db.rollback = MagicMock() + db.close = MagicMock() + return db, inserted + + +async def test_audit_script_resumable(monkeypatch): + """Run main() twice with the same batch — second pass inserts 0.""" + sample = [ + { + "id": 1, + "address": "ул Малышева 51", + "lat": 56.838, + "lon": 60.586, + "district": "Кировский", + }, + {"id": 2, "address": "ул Ленина 5", "lat": 56.840, "lon": 60.600, "district": "Ленинский"}, + ] + processed_ids: set[int] = set() + db, inserted = _make_db_mock(sample, processed_ids) + + # Force API mode without needing a real key. + monkeypatch.setenv("YANDEX_GEOCODER_API_KEY", "TEST_KEY") + + fake_result = YandexReverseResult( + address="Россия, Екатеринбург, улица Малышева, 51", + snapped_lat=56.838004, + snapped_lon=60.586155, + raw={"ok": True}, + ) + + with ( + patch("scripts.audit_address_mismatch.SessionLocal", return_value=db), + patch( + "scripts.audit_address_mismatch.reverse_via_api", + new=AsyncMock(return_value=fake_result), + ), + ): + # First run — both rows processed. + n1 = await main(["--batch", "test_batch_1", "--mode", "api"]) + assert n1 == 2 + assert len(inserted) == 2 + + # Second run with same batch — nothing left to do. + inserted.clear() + n2 = await main(["--batch", "test_batch_1", "--mode", "api"]) + assert n2 == 0 + assert inserted == [] + + +async def test_audit_script_api_mode_marks_error(monkeypatch): + """When the reverse call raises, the row is still inserted with status=error.""" + sample = [ + { + "id": 99, + "address": "ул Малышева 51", + "lat": 56.838, + "lon": 60.586, + "district": "Кировский", + }, + ] + processed_ids: set[int] = set() + db, inserted = _make_db_mock(sample, processed_ids) + + monkeypatch.setenv("YANDEX_GEOCODER_API_KEY", "TEST_KEY") + + with ( + patch("scripts.audit_address_mismatch.SessionLocal", return_value=db), + patch( + "scripts.audit_address_mismatch.reverse_via_api", + new=AsyncMock(side_effect=httpx.HTTPError("boom")), + ), + ): + n = await main(["--batch", "err_batch", "--mode", "api"]) + assert n == 1 + + assert len(inserted) == 1 + assert inserted[0]["audit_status"] == "error" + assert "boom" in (inserted[0]["error_message"] or "") + + +# --------------------------------------------------------------------------- +# Internal _run_api_mode no-match path +# --------------------------------------------------------------------------- + + +async def test_api_mode_no_match_path(): + """If Yandex returns address=None, row goes in with status=no_match.""" + sample = [SampleRow(id=7, address="ул X 1", lat=56.0, lon=60.0, district="Кировский")] + processed_ids: set[int] = set() + db, inserted = _make_db_mock([], processed_ids) + + res = YandexReverseResult(address=None, snapped_lat=None, snapped_lon=None, raw={"empty": True}) + + with patch( + "scripts.audit_address_mismatch.reverse_via_api", + new=AsyncMock(return_value=res), + ): + n = await _run_api_mode(db, sample, "b1", "key") + + assert n == 1 + assert inserted[0]["audit_status"] == "no_match" + assert inserted[0]["snapped_address"] is None