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

"""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 "<lon>,<lat>" (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: "<lon> <lat>" — 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,
    )