gendesign/tradein-mvp/backend/scripts/geocode_deals_nominatim.py

"""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 ('<city>, <street>'), 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()))