"""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, city) — 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 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. Grouping is by the PAIR (address, city), not by address alone (#2603, same shape as #2601 fixed in app/tasks/geocode_missing.py): the same address text in two different cities must not collapse into one geocode call, and the UPDATE must not spill onto the other city's rows — hence `city IS NOT DISTINCT FROM` (plain `=` never matches a NULL city, so a NULL-city group would update nothing). City hint (and why it is filtered) ---------------------------------- `deals.city` comes from Rosreestr (migration 177) and is filled on 100% of the rows, so a hint is available for every group — but its long tail holds values that are not cities at all ('Бессонова', 'Бердюгина', 'Билейский рыбопитомник'). A non-EKB hint is a HARD signal inside the geocoder: it closes the EKB-only local tiers (`_ekb_local_tiers_allowed`) and gets prefixed into the provider query, so a junk hint makes the result strictly WORSE than no hint at all. We therefore pass the hint only for values that are recognised cities of oblast 66 — `geocoder.known_city_hint`, shared with the other DB-column callers (`app/api/v1/admin.py`, `app/tasks/geocode_missing.py`); everything else degrades to the previous behaviour (no hint, address text only). 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, known_city_hint, ) except ImportError: # pragma: no cover import sys sys.path.insert(0, str(Path(__file__).resolve().parents[1])) from app.services.geocoder import GeocodeResult, geocode, known_city_hint 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, deals.city) pair + its lat-IS-NULL row count. `city` is the raw Rosreestr value (may be NULL for non-rosreestr sources, and may be junk — see `geocoder.known_city_hint`), kept verbatim so the UPDATE can target exactly the rows this group came from. """ address: str deals_count: int city: str | None = None @dataclass class Stats: """Final-summary counters. - processed distinct (address, city) groups 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 (address, city) pairs still needing coords — resume-safe 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-group 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, city for a deterministic resume order (the window ORDER BY must match the outer one, otherwise the running_rows cap slices a different ordering). GROUP BY address, city — NOT address alone (#2603, the shape #2601 fixed in app/tasks/geocode_missing.py): the same street name in two cities is two geocode calls with two different hints, not one call whose result lands on both. SQL groups NULL cities together, so a NULL-city group stays its own group rather than merging into an arbitrary city. """ rows = ( db.execute( text( "SELECT address, city, deals_count FROM (" " SELECT address, city, " " COUNT(*) AS deals_count, " " SUM(COUNT(*)) OVER (" " ORDER BY COUNT(*) DESC, address ASC, city ASC NULLS FIRST" " ) 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, city " ") g " "WHERE running_rows - deals_count < CAST(:limit AS int) " "ORDER BY deals_count DESC, address ASC, city ASC NULLS FIRST" ), {"limit": limit, "stale_days": stale_days}, ) .mappings() .all() ) return [ AddressGroup( address=r["address"], city=r.get("city"), deals_count=r["deals_count"], ) for r in rows ] def _count_pending_total(db: Session, *, stale_days: int) -> tuple[int, int]: """Full backlog: (distinct (address, city) pairs, total rows) this pass. Denominators for the dry-run projection — counts every lat-IS-NULL deal that passes the staleness filter, ignoring --limit. Counts PAIRS, matching what `_select_pending_addresses` actually feeds the geocoder (#2603) — counting distinct addresses here would understate the denominator and the projection would read above 100%. COALESCE(city, '') keeps a NULL-city pair countable: `count()` skips NULL inputs, and a bare row expression with a NULL field invites exactly that argument — the empty string can't collide with a real city name, so the pair count stays honest either way. """ row = db.execute( text( "SELECT COUNT(DISTINCT (address, COALESCE(city, ''))) 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, city: str | None, lat: float, lon: float ) -> int: """UPDATE every lat-IS-NULL deal on (address, city); geom 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. `city IS NOT DISTINCT FROM` (not `=`) scopes the write to the group the geocode was made for: plain `=` is never true for a NULL city, so a NULL-city group would update zero rows and re-run forever; `IS NOT DISTINCT FROM` treats NULL=NULL as a match while staying strict for a real city, so the same street text in another city keeps its own coords (#2603). 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 city IS NOT DISTINCT FROM CAST(:city AS text) " " AND lat IS NULL" ), {"addr": address, "city": city, "lat": lat, "lon": lon}, ) return result.rowcount def _mark_deals_tried(db: Session, *, address: str, city: str | None) -> 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. Scoped to the (address, city) pair for the same reason as the coords write — a miss in one city must not defer the other city's retry (#2603). 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 city IS NOT DISTINCT FROM CAST(:city AS text) " " AND lat IS NULL" ), {"addr": address, "city": city}, ) return result.rowcount # --------------------------------------------------------------------------- # Main loop # --------------------------------------------------------------------------- async def _run_backfill( db: Session, groups: list[AddressGroup], *, batch: str, dry_run: bool, ) -> Stats: """For each (address, city) pair: 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 city = group.city # Only a recognised oblast-66 city is fed to the geocoder; junk Rosreestr # values degrade to None (geocoder.known_city_hint — shared with the # other DB-column callers). The raw `city` is still used for the UPDATE # scope — it identifies the group either way. hint = known_city_hint(city) # 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, city_hint=hint) 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 city=%r (%d deals): %s", address[:60], city, 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 city=%r (%d deals) → NOT FOUND (would stamp tried)", address[:60], city, group.deals_count, ) else: try: with db.begin_nested(): _mark_deals_tried(db, address=address, city=city) db.commit() except Exception as exc: # defensive — isolate one bad UPDATE db.rollback() logger.warning( "mark_tried failed for addr=%r city=%r: %s", address[:60], city, 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 city=%r → (%.5f, %.5f) provider=%s would update %d deals", address[:60], city, 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, city=city, 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 city=%r: %s", address[:60], city, 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 (address, city) groups (%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()))