diff --git a/tradein-mvp/backend/scripts/geocode_deals_nominatim.py b/tradein-mvp/backend/scripts/geocode_deals_nominatim.py new file mode 100644 index 00000000..80947023 --- /dev/null +++ b/tradein-mvp/backend/scripts/geocode_deals_nominatim.py @@ -0,0 +1,564 @@ +"""Geocode deals still `lat IS NULL` after the house-centroid pass, via geocoder. + +Issue #569, Step 3 — the Nominatim (cache → Cadastral FDW → Yandex → Nominatim) +fallback for the ~14% of deals the centroid join can never cover. + +Background +---------- +Step 2 (`geocode_deals_from_houses.py`) backfilled `deals.lat/lon` by matching +each deal's normalized street to a centroid computed from already-geocoded +`houses`. On prod that covered 86% (42,840 / 49,791 rows). The remaining ~14% +(~6,951 rows) sit on streets with NO geocoded house at all, so the centroid join +has nothing to match against — those need a real geocoder call. + +This script picks up exactly those leftovers (`deals WHERE lat IS NULL`) and runs +each through `app.services.geocoder.geocode(address, db)`, which already does +cache → Cadastral FDW → Yandex → Nominatim with rate-limiting and writes the +result into `geocode_cache`. We do NOT call Nominatim directly — that keeps a +single source of truth for provider order, ЕКБ bbox filtering, and the 1 req/sec +policy. + +Why dedup by address (not one call per row) +------------------------------------------- +`deals.address` is street-only ('Екатеринбург, '), so thousands of rows +share the same address. The geocoder caches by normalized address, but we also +GROUP BY address up front so the real call count is driven by DISTINCT streets, +not the ~6,951 row backlog. One geocode call → UPDATE every deal on that street. + +Street-level precision is accepted: the estimator's comparable search uses a +1000-2000 m radius, so a street-level point lands every deal on that street in +the same comparable window. The `deals_set_geom_trg` BEFORE UPDATE OF lat, lon +trigger (002_core_tables.sql) fills `geom` from lat/lon, so we set lat/lon ONLY, +never geom. + +Resumable + bounded retries (critical — Nominatim is ~1 req/s) +-------------------------------------------------------------- +We stamp `geocode_tried_at = NOW()` on EVERY attempt — success AND failure — so +an un-geocodable address (Nominatim returns nothing) drops out of the candidate +set and is NOT re-scanned every run. Without this, each run would re-try the +full ~7k failures at 1 req/s. The candidate query combines: + - `lat IS NULL` (the `deals_geocode_pending_idx` partial index, 005), and + - `geocode_tried_at IS NULL OR geocode_tried_at < NOW() - interval '30 days'` +so failed addresses are retried at most once every 30 days, not every run. + +Per-address `db.begin_nested()` SAVEPOINT (backend.md rule) isolates one bad +UPDATE from the rest of the batch; a per-address commit means a crash mid-run +leaves already-geocoded streets persisted and resume picks up the rest. + +Usage: + DATABASE_URL=postgresql+psycopg://... \\ + YANDEX_GEOCODER_API_KEY=... # optional — geocoder falls back to Nominatim \\ + python -m scripts.geocode_deals_nominatim --dry-run + + # real run, default cap 2000 deal rows / run + python -m scripts.geocode_deals_nominatim --batch 2026-05-28 + +Flags: + --limit N max deal ROWS to fan out over this run (default 2000). The + number of geocode CALLS is the distinct-address count within + that slice, which is far smaller. + --dry-run report how many rows / distinct streets would be attempted, + plus the geocode hit/miss split. No DB writes. + --batch LABEL log label (default `deals_nominatim_YYYY-MM-DD`). + --stale-days N retry addresses last tried more than N days ago (default 30). +""" + +from __future__ import annotations + +import argparse +import asyncio +import logging +from dataclasses import dataclass +from datetime import date +from pathlib import Path + +from sqlalchemy import text +from sqlalchemy.orm import Session + +# Allow running both as `python -m scripts.geocode_deals_nominatim` (preferred, +# matches backfill_houses_dadata.py) and as a stand-alone file. +try: + from app.core.db import SessionLocal # type: ignore[import-not-found] +except ImportError: # pragma: no cover — fallback for adhoc invocation + import sys + + sys.path.insert(0, str(Path(__file__).resolve().parents[1])) + from app.core.db import SessionLocal + +try: + from app.services.geocoder import ( # type: ignore[import-not-found] + GeocodeResult, + geocode, + ) +except ImportError: # pragma: no cover + import sys + + sys.path.insert(0, str(Path(__file__).resolve().parents[1])) + from app.services.geocoder import GeocodeResult, geocode + +logging.basicConfig( + level=logging.INFO, + format="%(asctime)s %(levelname)s %(name)s %(message)s", +) +logger = logging.getLogger("geocode_deals_nominatim") + +# Default per-run row cap. Nominatim is ~1 req/sec; the number of CALLS is the +# distinct-address count inside this slice (far below the row count thanks to +# dedup + cache), but we still cap rows so a single run can't block for hours. +_DEFAULT_LIMIT = 2000 + +# Retry an address last tried more than this many days ago. Bounds retries on +# un-geocodable addresses (failed today → not re-scanned until 30 days pass). +_DEFAULT_STALE_DAYS = 30 + +# Progress log cadence (per distinct address processed). +_LOG_EVERY = 25 + + +# --------------------------------------------------------------------------- +# Domain types +# --------------------------------------------------------------------------- + + +@dataclass +class AddressGroup: + """One distinct deals.address and how many lat-IS-NULL rows share it.""" + + address: str + deals_count: int + + +@dataclass +class Stats: + """Final-summary counters. + + - processed distinct addresses fed to the geocoder this run. + - geocoded addresses the geocoder resolved to coords. + - geocode_failed addresses the geocoder returned None for (still stamped). + - skipped distinct addresses skipped before any geocode call + (too short to be a usable query). + - deals_updated total deal ROWS that received lat/lon (sum over geocoded + addresses). + - cache_hits / cache_misses provider == 'cache' vs a real provider call. + """ + + processed: int = 0 + geocoded: int = 0 + geocode_failed: int = 0 + skipped: int = 0 + deals_updated: int = 0 + cache_hits: int = 0 + cache_misses: int = 0 + + +# --------------------------------------------------------------------------- +# Source query — distinct addresses among the leftover lat-IS-NULL deals +# --------------------------------------------------------------------------- + + +def _select_pending_addresses( + db: Session, *, limit: int, stale_days: int +) -> list[AddressGroup]: + """Distinct deals.address still needing coords — resume-safe candidate set. + + Combines the `deals_geocode_pending_idx` partial index predicate + (`lat IS NULL`) with a staleness filter so failed/un-geocodable addresses + are retried at most once every `stale_days`, never every run. + + The GROUP BY collapses the ~6,951-row backlog into its distinct streets; + `:limit` caps the deal ROWS fanned out, computed from the running SUM of + per-address counts so a single huge street can't blow past the cap. We + order by occurrence DESC (biggest ROI per geocode call first) then address + for a deterministic resume order. + """ + rows = db.execute( + text( + "SELECT address, deals_count FROM (" + " SELECT address, " + " COUNT(*) AS deals_count, " + " SUM(COUNT(*)) OVER (" + " ORDER BY COUNT(*) DESC, address ASC" + " ) AS running_rows " + " FROM deals " + " WHERE lat IS NULL " + " AND address IS NOT NULL " + " AND length(trim(address)) >= 3 " + " AND (geocode_tried_at IS NULL " + " OR geocode_tried_at < NOW() " + " - make_interval(days => CAST(:stale_days AS int))) " + " GROUP BY address " + ") g " + "WHERE running_rows - deals_count < CAST(:limit AS int) " + "ORDER BY deals_count DESC, address ASC" + ), + {"limit": limit, "stale_days": stale_days}, + ).mappings().all() + return [AddressGroup(address=r["address"], deals_count=r["deals_count"]) for r in rows] + + +def _count_pending_total(db: Session, *, stale_days: int) -> tuple[int, int]: + """Full backlog: (distinct addresses, total rows) eligible this pass. + + Denominators for the dry-run projection — counts every lat-IS-NULL deal + that passes the staleness filter, ignoring --limit. + """ + row = db.execute( + text( + "SELECT COUNT(DISTINCT address) AS streets, COUNT(*) AS rows " + "FROM deals " + "WHERE lat IS NULL " + " AND address IS NOT NULL " + " AND length(trim(address)) >= 3 " + " AND (geocode_tried_at IS NULL " + " OR geocode_tried_at < NOW() " + " - make_interval(days => CAST(:stale_days AS int)))" + ), + {"stale_days": stale_days}, + ).first() + if row is None: + return (0, 0) + return (int(row[0]), int(row[1])) + + +# --------------------------------------------------------------------------- +# DB writers +# --------------------------------------------------------------------------- + + +def _update_deals_geocoded(db: Session, *, address: str, lat: float, lon: float) -> int: + """UPDATE every lat-IS-NULL deal on `address`; geom auto-fills via trigger. + + The `deals_set_geom_trg` BEFORE UPDATE OF lat, lon trigger + (002_core_tables.sql, reuses listings_set_geom()) populates geom from the + new lat/lon, so we never touch geom here. Stamps `geocode_tried_at = NOW()` + so the row drops out of the candidate set. The `AND lat IS NULL` guard + keeps this idempotent and avoids clobbering rows another pass already set. + + Returns the number of deal rows updated. + """ + result = db.execute( + text( + "UPDATE deals " + " SET lat = CAST(:lat AS double precision), " + " lon = CAST(:lon AS double precision), " + " geocode_tried_at = NOW() " + " WHERE address = CAST(:addr AS text) " + " AND lat IS NULL" + ), + {"addr": address, "lat": lat, "lon": lon}, + ) + return result.rowcount + + +def _mark_deals_tried(db: Session, *, address: str) -> int: + """Stamp `geocode_tried_at = NOW()` WITHOUT touching lat/lon (geocode miss). + + Critical for resume: an address the geocoder can't resolve must still drop + out of the candidate set, or the next run re-scans it at ~1 req/s. We leave + lat/lon NULL (still "pending coords") but the staleness filter now excludes + it for `stale_days`. The `AND lat IS NULL` guard means a concurrent success + can't be downgraded. + + Returns the number of deal rows stamped. + """ + result = db.execute( + text( + "UPDATE deals " + " SET geocode_tried_at = NOW() " + " WHERE address = CAST(:addr AS text) " + " AND lat IS NULL" + ), + {"addr": address}, + ) + return result.rowcount + + +# --------------------------------------------------------------------------- +# Main loop +# --------------------------------------------------------------------------- + + +async def _run_backfill( + db: Session, + groups: list[AddressGroup], + *, + batch: str, + dry_run: bool, +) -> Stats: + """For each distinct address: geocode once, then UPDATE all its deals. + + Per-address SAVEPOINT (`db.begin_nested()`) so one bad UPDATE can't abort + the batch (backend.md SAVEPOINT rule). Per-address commit on success → a + crash mid-run leaves already-geocoded streets persisted, and resume picks + up the rest (lat IS NULL + staleness filter). + + Rate-limiting lives inside `geocode()` (Nominatim 1 req/s sleep, Yandex + retry/backoff), so we don't add our own sleep here — that would double the + walltime. A geocode that raises is treated as a failure for THIS run but is + NOT stamped (left for the next pass to retry sooner than a clean miss). + """ + stats = Stats() + + for i, group in enumerate(groups, start=1): + address = group.address + + # The geocoder itself rejects <3 chars, but skip here too so the dry-run + # report and counters stay honest (no phantom "processed" address). + if len(address.strip()) < 3: + stats.skipped += 1 + continue + + result: GeocodeResult | None = None + try: + result = await geocode(address, db) + except Exception as exc: # defensive — one geocode error must not kill batch + stats.geocode_failed += 1 + stats.processed += 1 + logger.warning( + "geocode raised for addr=%r (%d deals): %s", + address[:60], + group.deals_count, + exc, + ) + # Do NOT stamp on a raised exception — a transient network/HTTP error + # should be retried on the next run, not deferred for stale_days. + continue + + if result is None: + # Clean miss — geocoder exhausted all providers. Stamp so we don't + # re-scan this un-geocodable address every run. + stats.geocode_failed += 1 + stats.processed += 1 + if dry_run: + logger.info( + "DRY-RUN addr=%r (%d deals) → NOT FOUND (would stamp tried)", + address[:60], + group.deals_count, + ) + else: + try: + with db.begin_nested(): + _mark_deals_tried(db, address=address) + db.commit() + except Exception as exc: # defensive — isolate one bad UPDATE + db.rollback() + logger.warning("mark_tried failed for addr=%r: %s", address[:60], exc) + _maybe_log_progress(i, groups, batch, stats) + continue + + # Hit — record provider split for the summary. + if result.provider == "cache": + stats.cache_hits += 1 + else: + stats.cache_misses += 1 + stats.geocoded += 1 + stats.processed += 1 + + if dry_run: + logger.info( + "DRY-RUN addr=%r → (%.5f, %.5f) provider=%s would update %d deals", + address[:60], + result.lat, + result.lon, + result.provider, + group.deals_count, + ) + _maybe_log_progress(i, groups, batch, stats) + continue + + try: + with db.begin_nested(): + n = _update_deals_geocoded( + db, address=address, lat=result.lat, lon=result.lon + ) + # Per-address commit so resume picks up exactly where we crashed. + db.commit() + stats.deals_updated += n + except Exception as exc: # defensive — isolate one bad UPDATE + db.rollback() + # The geocode itself succeeded (and is cached); only the write + # failed. Count the address as geocoded but log the write failure. + logger.warning("db_write failed for addr=%r: %s", address[:60], exc) + + _maybe_log_progress(i, groups, batch, stats) + + return stats + + +def _maybe_log_progress(i: int, groups: list[AddressGroup], batch: str, stats: Stats) -> None: + """Emit a progress line every `_LOG_EVERY` distinct addresses.""" + if i % _LOG_EVERY == 0: + logger.info( + "batch=%s progress %d/%d geocoded=%d failed=%d deals_updated=%d " + "cache=(hit=%d miss=%d)", + batch, + i, + len(groups), + stats.geocoded, + stats.geocode_failed, + stats.deals_updated, + stats.cache_hits, + stats.cache_misses, + ) + + +# --------------------------------------------------------------------------- +# Dry-run reporting +# --------------------------------------------------------------------------- + + +def _report_dry_run( + stats: Stats, + *, + total_streets: int, + total_rows: int, + scanned_streets: int, +) -> None: + """Log the would-attempt counts + the geocode hit/miss split. + + `scanned_streets` is how many distinct addresses we actually fed to the + geocoder this run (capped by --limit fan-out); `total_streets` / `total_rows` + are the full eligible backlog so the projection extrapolates honestly. + """ + hit_rate = (stats.geocoded / stats.processed) if stats.processed else 0.0 + projected_streets = round(hit_rate * total_streets) + + logger.info("─" * 60) + logger.info("DRY-RUN SUMMARY (no DB writes)") + logger.info( + "full eligible backlog: %d distinct streets across %d deal rows " + "(lat IS NULL, not tried within stale window)", + total_streets, + total_rows, + ) + logger.info("distinct streets attempted this run (capped by --limit): %d", scanned_streets) + logger.info(" → geocoded: %d", stats.geocoded) + logger.info(" → not found: %d", stats.geocode_failed) + logger.info(" → skipped (short): %d", stats.skipped) + logger.info(" → cache hit/miss: %d / %d", stats.cache_hits, stats.cache_misses) + logger.info("hit rate on attempted streets: %.1f%%", hit_rate * 100.0) + logger.info( + "projected over full backlog: ≈ %d / %d streets resolvable (%.1f%%)", + projected_streets, + total_streets, + hit_rate * 100.0, + ) + logger.info("─" * 60) + + +# --------------------------------------------------------------------------- +# CLI +# --------------------------------------------------------------------------- + + +def _parse_args(argv: list[str] | None = None) -> argparse.Namespace: + """argparse setup, factored out for testability.""" + p = argparse.ArgumentParser( + description=( + "Issue #569 Step 3 — geocode deals still lat IS NULL after the " + "house-centroid pass, via app.services.geocoder.geocode " + "(cache → Cadastral FDW → Yandex → Nominatim)." + ), + ) + p.add_argument( + "--limit", + type=int, + default=_DEFAULT_LIMIT, + help=( + f"Max deal ROWS to fan out over this run (default {_DEFAULT_LIMIT}). " + "Geocode CALLS = distinct-address count within that slice (smaller)." + ), + ) + p.add_argument( + "--batch", + default=f"deals_nominatim_{date.today().isoformat()}", + help="Log batch label (does not affect DB filters — logs only).", + ) + p.add_argument( + "--dry-run", + action="store_true", + help=( + "Report how many rows / distinct streets would be attempted and the " + "geocode hit/miss split. No DB writes." + ), + ) + p.add_argument( + "--stale-days", + type=int, + default=_DEFAULT_STALE_DAYS, + help=( + f"Retry addresses last tried more than N days ago (default " + f"{_DEFAULT_STALE_DAYS}). Bounds retries on un-geocodable addresses." + ), + ) + return p.parse_args(argv) + + +async def main(argv: list[str] | None = None) -> int: + """CLI entry point. Returns the number of distinct addresses geocoded. + + Async because `geocode()` is a coroutine; run via `asyncio.run(main())` + below, mirroring backfill_houses_dadata.py / backfill_house_coords.py. + No external creds are required — the geocoder degrades to Nominatim when + YANDEX_GEOCODER_API_KEY is unset, so we don't fail-fast on a missing key. + """ + args = _parse_args(argv) + logger.info( + "starting batch=%s limit=%s stale_days=%s dry_run=%s", + args.batch, + args.limit, + args.stale_days, + args.dry_run, + ) + + db = SessionLocal() + try: + groups = _select_pending_addresses( + db, limit=args.limit, stale_days=args.stale_days + ) + total_rows = sum(g.deals_count for g in groups) + logger.info( + "loaded %d distinct addresses (%d deal rows) needing coords", + len(groups), + total_rows, + ) + if not groups: + logger.info( + "nothing to do — no deals with lat IS NULL eligible (all tried " + "within the last %d days, or no addressable rows)", + args.stale_days, + ) + return 0 + + stats = await _run_backfill(db, groups, batch=args.batch, dry_run=args.dry_run) + + if args.dry_run: + total_streets, backlog_rows = _count_pending_total( + db, stale_days=args.stale_days + ) + _report_dry_run( + stats, + total_streets=total_streets, + total_rows=backlog_rows, + scanned_streets=len(groups), + ) + + logger.info( + "done: batch=%s processed=%d geocoded=%d geocode_failed=%d " + "skipped=%d deals_updated=%d cache=(hit=%d miss=%d)", + args.batch, + stats.processed, + stats.geocoded, + stats.geocode_failed, + stats.skipped, + stats.deals_updated, + stats.cache_hits, + stats.cache_misses, + ) + return stats.geocoded + finally: + db.close() + + +if __name__ == "__main__": # pragma: no cover + raise SystemExit(asyncio.run(main())) diff --git a/tradein-mvp/backend/tests/scripts/test_geocode_deals_nominatim.py b/tradein-mvp/backend/tests/scripts/test_geocode_deals_nominatim.py new file mode 100644 index 00000000..794555cc --- /dev/null +++ b/tradein-mvp/backend/tests/scripts/test_geocode_deals_nominatim.py @@ -0,0 +1,456 @@ +"""Unit tests for issue #569 Step 3 — geocode_deals_nominatim.py. + +Coverage (per the issue's test plan): + - a NULL-coords deal gets geocoded → UPDATE deals issued with lat/lon + + geocode_tried_at, counted as geocoded, deals_updated reflects rowcount. + - geocode() returns None → geocode_tried_at still stamped (a "mark tried" + UPDATE with no lat/lon), no coords UPDATE, counted as geocode_failed. + - --dry-run issues no UPDATE / no commit, counters still move. + - the resume filter excludes already-tried-recently rows (SQL carries the + lat IS NULL + staleness predicate, and binds --stale-days through). + - per-address SAVEPOINT isolation: one failing UPDATE doesn't abort the + batch. + - dedup: distinct addresses drive geocode call count (one call per address). + - main() wiring: SessionLocal, --limit bind, returns geocoded count. + +No real Postgres. `geocode()` is async → patched with AsyncMock (mirrors +tests/tasks/test_geocode_missing.py). The Session is a MagicMock that routes +SELECT side-effects by SQL substring and records UPDATE binds (mirrors +tests/scripts/test_geocode_deals_from_houses.py). +""" + +from __future__ import annotations + +import os +from unittest.mock import AsyncMock, MagicMock, patch + +# Settings needs a DSN at import time — set a dummy before any app.* import. +os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost/test_db") + +from app.services.geocoder import GeocodeResult +from scripts.geocode_deals_nominatim import ( + AddressGroup, + Stats, + _mark_deals_tried, + _run_backfill, + _select_pending_addresses, + _update_deals_geocoded, + main, +) + +_GEOCODE_PATH = "scripts.geocode_deals_nominatim.geocode" + + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + + +def _result(provider: str = "nominatim", lat: float = 56.838, lon: float = 60.605) -> GeocodeResult: + return GeocodeResult( + lat=lat, + lon=lon, + full_address="Екатеринбург, ул. Тестовая", + provider=provider, # type: ignore[arg-type] + confidence="approximate", + ) + + +def _make_db_mock( + *, + address_rows: list[dict] | None = None, + total_streets: int = 0, + total_rows: int = 0, + update_rowcount: int = 1, +) -> tuple[MagicMock, list[dict], list[dict]]: + """MagicMock Session that: + + - returns `address_rows` for the candidate SELECT (`FROM deals` GROUP BY) + - returns (total_streets, total_rows) for the dry-run COUNT + - records coords-UPDATE binds into `coord_updates` + - records mark-tried-UPDATE binds into `tried_updates` + - every UPDATE returns a result with `.rowcount == update_rowcount` + - supports `db.begin_nested()` as a context manager + + Returns (db, coord_updates, tried_updates). + """ + address_rows = address_rows or [] + coord_updates: list[dict] = [] + tried_updates: list[dict] = [] + + db = MagicMock() + db.begin_nested.return_value.__enter__ = lambda self: self + db.begin_nested.return_value.__exit__ = lambda self, *a: False + + def execute_side_effect(sql, params=None): + sql_str = str(sql) + result = MagicMock() + if "UPDATE deals" in sql_str: + # Distinguish the coords UPDATE (sets lat) from the mark-tried UPDATE. + if "SET lat" in sql_str: + coord_updates.append(dict(params) if params else {}) + else: + tried_updates.append(dict(params) if params else {}) + result.rowcount = update_rowcount + return result + if "COUNT(DISTINCT address)" in sql_str: + result.first.return_value = (total_streets, total_rows) + return result + if "FROM deals" in sql_str: + result.mappings.return_value.all.return_value = address_rows + return result + return result + + db.execute.side_effect = execute_side_effect + db.commit = MagicMock() + db.rollback = MagicMock() + db.close = MagicMock() + return db, coord_updates, tried_updates + + +# --------------------------------------------------------------------------- +# _run_backfill — happy path: geocoded → coords UPDATE + tried_at stamped +# --------------------------------------------------------------------------- + + +async def test_run_backfill_geocoded_issues_coords_update(): + groups = [AddressGroup(address="Екатеринбург, Тестовая", deals_count=5)] + db, coord_updates, tried_updates = _make_db_mock(update_rowcount=5) + + with patch(_GEOCODE_PATH, new_callable=AsyncMock, return_value=_result("nominatim")): + stats = await _run_backfill(db, groups, batch="b1", dry_run=False) + + assert stats.geocoded == 1 + assert stats.geocode_failed == 0 + assert stats.processed == 1 + assert stats.deals_updated == 5 + assert stats.cache_misses == 1 + assert stats.cache_hits == 0 + # Exactly one coords UPDATE, bound with lat/lon/addr; no mark-tried UPDATE. + assert len(coord_updates) == 1 + assert coord_updates[0]["addr"] == "Екатеринбург, Тестовая" + assert coord_updates[0]["lat"] == 56.838 + assert coord_updates[0]["lon"] == 60.605 + assert tried_updates == [] + assert db.commit.call_count == 1 + + +async def test_run_backfill_cache_hit_counted_separately(): + groups = [AddressGroup(address="Екатеринбург, Кэшевая", deals_count=2)] + db, coord_updates, _ = _make_db_mock(update_rowcount=2) + + with patch(_GEOCODE_PATH, new_callable=AsyncMock, return_value=_result("cache")): + stats = await _run_backfill(db, groups, batch="b", dry_run=False) + + assert stats.geocoded == 1 + assert stats.cache_hits == 1 + assert stats.cache_misses == 0 + assert len(coord_updates) == 1 + + +# --------------------------------------------------------------------------- +# geocode None → geocode_tried_at stamped, NO coords UPDATE, counted failed +# --------------------------------------------------------------------------- + + +async def test_run_backfill_geocode_none_stamps_tried_no_coords(): + groups = [AddressGroup(address="Екатеринбург, Несуществующая", deals_count=3)] + db, coord_updates, tried_updates = _make_db_mock(update_rowcount=3) + + with patch(_GEOCODE_PATH, new_callable=AsyncMock, return_value=None): + stats = await _run_backfill(db, groups, batch="b", dry_run=False) + + assert stats.geocode_failed == 1 + assert stats.geocoded == 0 + assert stats.processed == 1 + assert stats.deals_updated == 0 + # No coords UPDATE… + assert coord_updates == [] + # …but geocode_tried_at IS stamped via the mark-tried UPDATE. + assert len(tried_updates) == 1 + assert tried_updates[0]["addr"] == "Екатеринбург, Несуществующая" + assert db.commit.call_count == 1 + + +async def test_mark_deals_tried_sql_stamps_tried_at_only(): + """The mark-tried writer sets geocode_tried_at and nothing else (no lat/lon).""" + db = MagicMock() + db.execute.return_value = MagicMock(rowcount=4) + n = _mark_deals_tried(db, address="Екатеринбург, X") + args, _kw = db.execute.call_args + sql_str = str(args[0]) + assert "UPDATE deals" in sql_str + assert "geocode_tried_at = NOW()" in sql_str + assert "SET lat" not in sql_str + assert "lat IS NULL" in sql_str # idempotency / no-clobber guard + assert n == 4 + + +async def test_run_backfill_geocode_raises_not_stamped(): + """A raised geocode error → counted failed but NOT stamped (retry sooner).""" + groups = [AddressGroup(address="Екатеринбург, Сетевая", deals_count=1)] + db, coord_updates, tried_updates = _make_db_mock() + + with patch(_GEOCODE_PATH, new_callable=AsyncMock, side_effect=RuntimeError("network")): + stats = await _run_backfill(db, groups, batch="b", dry_run=False) + + assert stats.geocode_failed == 1 + assert stats.geocoded == 0 + # Neither a coords UPDATE nor a mark-tried UPDATE — leave it for next run. + assert coord_updates == [] + assert tried_updates == [] + assert db.commit.call_count == 0 + + +# --------------------------------------------------------------------------- +# --dry-run — no DB writes +# --------------------------------------------------------------------------- + + +async def test_run_backfill_dry_run_issues_no_update(): + groups = [ + AddressGroup(address="Екатеринбург, Тестовая", deals_count=4), + AddressGroup(address="Екатеринбург, Пустая", deals_count=1), + ] + db, coord_updates, tried_updates = _make_db_mock() + + # First address resolves, second is a miss — neither must write in dry-run. + with patch( + _GEOCODE_PATH, + new_callable=AsyncMock, + side_effect=[_result("nominatim"), None], + ): + stats = await _run_backfill(db, groups, batch="dry", dry_run=True) + + assert stats.geocoded == 1 + assert stats.geocode_failed == 1 + assert stats.processed == 2 + assert coord_updates == [] + assert tried_updates == [] + assert db.commit.call_count == 0 + + +# --------------------------------------------------------------------------- +# dedup — one geocode call per distinct address +# --------------------------------------------------------------------------- + + +async def test_run_backfill_one_call_per_distinct_address(): + groups = [ + AddressGroup(address="Екатеринбург, А", deals_count=10), + AddressGroup(address="Екатеринбург, Б", deals_count=2), + ] + db, coord_updates, _ = _make_db_mock(update_rowcount=1) + + with patch( + _GEOCODE_PATH, new_callable=AsyncMock, return_value=_result("nominatim") + ) as mock_geo: + stats = await _run_backfill(db, groups, batch="b", dry_run=False) + + # 2 distinct addresses → exactly 2 geocode calls (not 12 = sum of deals). + assert mock_geo.call_count == 2 + assert stats.geocoded == 2 + assert len(coord_updates) == 2 + + +async def test_run_backfill_skips_too_short_address_without_geocoding(): + groups = [AddressGroup(address="ек", deals_count=1)] # < 3 chars + db, coord_updates, _ = _make_db_mock() + + with patch(_GEOCODE_PATH, new_callable=AsyncMock) as mock_geo: + stats = await _run_backfill(db, groups, batch="b", dry_run=False) + + assert stats.skipped == 1 + assert stats.processed == 0 + mock_geo.assert_not_called() + assert coord_updates == [] + + +# --------------------------------------------------------------------------- +# per-address SAVEPOINT — one bad UPDATE doesn't abort the batch +# --------------------------------------------------------------------------- + + +async def test_run_backfill_db_write_failure_isolated_to_address(): + groups = [ + AddressGroup(address="Екатеринбург, Первая", deals_count=1), + AddressGroup(address="Екатеринбург, Вторая", deals_count=1), + ] + db, coord_updates, _ = _make_db_mock(update_rowcount=1) + + # Make the FIRST coords UPDATE raise, the rest succeed. + call = {"n": 0} + real_side_effect = db.execute.side_effect + + def failing_execute(sql, params=None): + sql_str = str(sql) + if "UPDATE deals" in sql_str and "SET lat" in sql_str: + call["n"] += 1 + if call["n"] == 1: + raise RuntimeError("constraint blew up") + return real_side_effect(sql, params) + + db.execute.side_effect = failing_execute + + with patch(_GEOCODE_PATH, new_callable=AsyncMock, return_value=_result("nominatim")): + stats = await _run_backfill(db, groups, batch="b", dry_run=False) + + # Both addresses geocoded; one write failed but was isolated via SAVEPOINT. + assert stats.geocoded == 2 + assert db.rollback.call_count == 1 + # Only the second address's coords UPDATE was recorded. + assert [u["addr"] for u in coord_updates] == ["Екатеринбург, Вторая"] + # deals_updated only counts the successful write. + assert stats.deals_updated == 1 + + +# --------------------------------------------------------------------------- +# _update_deals_geocoded — bind shape + geom NOT set manually +# --------------------------------------------------------------------------- + + +def test_update_deals_geocoded_sets_lat_lon_tried_at_not_geom(): + db = MagicMock() + db.execute.return_value = MagicMock(rowcount=7) + n = _update_deals_geocoded(db, address="Екатеринбург, Y", lat=56.1, lon=60.2) + args, _kw = db.execute.call_args + sql_str = str(args[0]) + binds = args[1] + assert "UPDATE deals" in sql_str + assert "geocode_tried_at = NOW()" in sql_str + # geom must NOT be set manually — the deals_set_geom_trg trigger fills it. + assert "geom" not in sql_str + assert "lat IS NULL" in sql_str # no-clobber guard + assert binds == {"addr": "Екатеринбург, Y", "lat": 56.1, "lon": 60.2} + assert n == 7 + + +# --------------------------------------------------------------------------- +# resume filter — candidate SQL carries lat IS NULL + staleness predicate +# --------------------------------------------------------------------------- + + +def test_select_pending_addresses_filters_null_and_stale(): + """The candidate query must combine lat IS NULL with the staleness window + and bind --stale-days, so already-tried-recently rows are excluded.""" + db, _, _ = _make_db_mock( + address_rows=[{"address": "Екатеринбург, Z", "deals_count": 3}] + ) + + groups = _select_pending_addresses(db, limit=2000, stale_days=30) + + assert groups == [AddressGroup(address="Екатеринбург, Z", deals_count=3)] + # Inspect the SQL + binds of the SELECT. + select_call = next( + c for c in db.execute.call_args_list if "FROM deals" in str(c[0][0]) + ) + sql_str = str(select_call[0][0]) + binds = select_call[0][1] + assert "lat IS NULL" in sql_str + assert "geocode_tried_at IS NULL" in sql_str + assert "make_interval(days => CAST(:stale_days AS int))" in sql_str + assert "GROUP BY address" in sql_str + assert binds["stale_days"] == 30 + assert binds["limit"] == 2000 + + +def test_select_pending_addresses_passes_custom_stale_days(): + db, _, _ = _make_db_mock(address_rows=[]) + _select_pending_addresses(db, limit=500, stale_days=7) + select_call = next( + c for c in db.execute.call_args_list if "FROM deals" in str(c[0][0]) + ) + binds = select_call[0][1] + assert binds["stale_days"] == 7 + assert binds["limit"] == 500 + + +# --------------------------------------------------------------------------- +# main() — wiring, --limit, --dry-run, return value +# --------------------------------------------------------------------------- + + +async def test_main_geocodes_and_returns_count(): + address_rows = [ + {"address": "Екатеринбург, Малышева", "deals_count": 4}, + {"address": "Екатеринбург, Ленина", "deals_count": 2}, + ] + db, coord_updates, _ = _make_db_mock(address_rows=address_rows, update_rowcount=4) + + with ( + patch("scripts.geocode_deals_nominatim.SessionLocal", return_value=db), + patch(_GEOCODE_PATH, new_callable=AsyncMock, return_value=_result("nominatim")), + ): + n = await main(["--batch", "test_main"]) + + assert n == 2 # both distinct addresses geocoded + assert len(coord_updates) == 2 + assert {u["addr"] for u in coord_updates} == { + "Екатеринбург, Малышева", + "Екатеринбург, Ленина", + } + + +async def test_main_dry_run_writes_nothing(): + address_rows = [{"address": "Екатеринбург, Малышева", "deals_count": 3}] + db, coord_updates, tried_updates = _make_db_mock( + address_rows=address_rows, total_streets=1, total_rows=3 + ) + + with ( + patch("scripts.geocode_deals_nominatim.SessionLocal", return_value=db), + patch(_GEOCODE_PATH, new_callable=AsyncMock, return_value=_result("nominatim")), + ): + n = await main(["--dry-run"]) + + # dry-run returns the would-geocode count, writes nothing. + assert n == 1 + assert coord_updates == [] + assert tried_updates == [] + assert db.commit.call_count == 0 + + +async def test_main_respects_limit_bind(): + db, _, _ = _make_db_mock(address_rows=[]) + + with ( + patch("scripts.geocode_deals_nominatim.SessionLocal", return_value=db), + patch(_GEOCODE_PATH, new_callable=AsyncMock), + ): + await main(["--limit", "50"]) + + select_call = next( + c for c in db.execute.call_args_list if "FROM deals" in str(c[0][0]) + ) + assert select_call[0][1]["limit"] == 50 + + +async def test_main_no_pending_returns_zero(): + db, coord_updates, _ = _make_db_mock(address_rows=[]) + + with ( + patch("scripts.geocode_deals_nominatim.SessionLocal", return_value=db), + patch(_GEOCODE_PATH, new_callable=AsyncMock) as mock_geo, + ): + n = await main([]) + + assert n == 0 + assert coord_updates == [] + mock_geo.assert_not_called() + db.close.assert_called_once() + + +# --------------------------------------------------------------------------- +# Stats dataclass +# --------------------------------------------------------------------------- + + +def test_stats_defaults_zero(): + s = Stats() + assert s.processed == 0 + assert s.geocoded == 0 + assert s.geocode_failed == 0 + assert s.skipped == 0 + assert s.deals_updated == 0 + assert s.cache_hits == 0 + assert s.cache_misses == 0