"""Backfill deals.lat/lon via a per-street centroid join against houses.

Issue #569, Step 2. The `deals` table (49,791 Rosreestr ДКП sale records) has
100% NULL lat/lon/geom, so `estimator._fetch_deals()` — which matches via
`ST_DWithin(geom, point, radius)` — never returns a single deal. The "real
deals" comparable feature is silently dead.

Approach — **street-centroid join, zero external geocoder calls**:
  1. Build a per-street centroid map from `houses WHERE geom IS NOT NULL`
     (~8,600 rows already geocoded): derive a normalized street key from
     `houses.address`, group by it, compute the centroid as AVG(lat)/AVG(lon).
  2. For each `deals` row with lat IS NULL: derive the SAME street key from
     `deals.address` ('Екатеринбург, <street>', street-only — no house number),
     look up the centroid and `UPDATE deals SET lat, lon, geocode_tried_at=NOW()`.

Street-level precision is acceptable — the estimator search radius is
1000-2000 m, so all deals on the same street land inside the same comparable
window regardless of which house we used for the centroid.

Why a street key (bare street name) and not `normalize_address`:
  `deals.address` carries NO house number and sometimes NO street-type word
  ('Екатеринбург, Малышева'), while `houses.address` is full and varied
  ('Свердловская обл., Екатеринбург, ул. Большакова, 17'). `normalize_address`
  keeps the house number and the type word, so the two sides would never
  collide. `_street_key` strips city/region prefix, the street-type token AND
  the house number, leaving just the lowercased street name ('малышева',
  '8 марта') — the only token both sides reliably share.

The `deals_set_geom_trg` BEFORE INSERT OR UPDATE OF lat, lon trigger
(002_core_tables.sql) auto-fills `geom` from lat/lon via listings_set_geom(),
so this script sets lat/lon ONLY — never geom directly.

Resume-safe: only `deals WHERE lat IS NULL` are processed (matches the
`deals_geocode_pending_idx` partial index from 005_geocode_tracking.sql); a
successful UPDATE drops the row out of the candidate set on the next run.

Usage:
    DATABASE_URL=postgresql+psycopg://... \\
    python -m scripts.geocode_deals_from_houses --dry-run

    # real backfill, default cap 5000 rows/run
    python -m scripts.geocode_deals_from_houses --batch 2026-05-28

Flags:
    --dry-run      report coverage projection + top-10 unmatched streets,
                   no DB writes.
    --limit N      max deals to process this run (default 5000).
    --batch LABEL  log label (default `deals_geo_YYYY-MM-DD`).
"""

from __future__ import annotations

import argparse
import logging
import math
import re
import unicodedata
from collections import Counter
from dataclasses import dataclass, field
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_from_houses` (preferred,
# matches the pattern from backfill_houses_dadata.py) and as a stand-alone file.
try:
    from app.core.db import SessionLocal  # type: ignore[import-not-found]
    from app.services.regions import REGIONS  # type: ignore[import-not-found]
except ImportError:  # pragma: no cover — fallback for adhoc invocation
    import sys

    sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
    from app.core.db import SessionLocal
    from app.services.regions import REGIONS

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s %(levelname)s %(name)s %(message)s",
)
logger = logging.getLogger("geocode_deals_from_houses")

# Default per-run cap. The full deals backlog is ~50k; a SAVEPOINT-per-row
# UPDATE loop over 50k is fine in one pass, but a cap keeps canary runs cheap
# and lets the caller chunk if they want.
_DEFAULT_LIMIT = 5000

# Progress log cadence.
_LOG_EVERY = 1000

# A street key shorter than this is almost certainly a parse failure (an empty
# string, a stray house number, a one-letter token) — skip it on both the
# houses (centroid) and deals (lookup) side so garbage never anchors a match.
_MIN_KEY_LEN = 3

# Default region. 66 (Свердловская обл.) — единственный регион, на котором
# скрипт исторически работал; остаётся умолчанием, чтобы существующие вызовы
# без флага вели себя ровно как раньше.
_DEFAULT_REGION_CODE = 66

# Sanity-порог разброса домов ВНУТРИ одного ключа (км). Ключ — (регион, НП,
# улица), так что легитимный разброс — это длина одной улицы: даже проспект в
# большом городе редко длиннее 10-15 км, а типичная улица — 1-3 км. Разброс
# больше порога означает, что в ведро слиплись ОДНОИМЁННЫЕ улицы разных
# населённых пунктов (или один НП распознан двумя написаниями) — среднее таких
# точек даёт правдоподобную координату ПОСЕРЕДИНЕ между городами, и ошибка
# тихая. Такое ведро выбрасывается целиком: дыра в покрытии честнее, чем
# сделка, посаженная в поле между Клином и Серпуховом.
_DEFAULT_MAX_SPREAD_KM = 5.0

# Классификация comma-сегмента адреса (см. `_split_place_street`).
# Тип населённого пункта в начале сегмента: «г Химки», «пос. Голубое»,
# «д Сабурово», «рп Оболенск», «снт Заря». За типом ОБЯЗАН идти буквенный
# токен — иначе «д 5» (номер дома) распознался бы как деревня.
_LOCALITY_TYPE_RE = re.compile(
    r"^(?:"
    r"г|гор|город|пгт|рп|дп|снт|днп|тер"
    r"|п|пос|поселок|посёлок|п/ст"
    r"|д|дер|деревня|с|село|сл|слобода|ст|станция|х|хутор|аул"
    r")\.?\s+(?=[а-яё])",
    flags=re.UNICODE,
)

# Сегмент — номер дома/квартиры/корпуса, а не имя: «125», «д 5», «кв 12»,
# «корп 3», «литера А». Такие сегменты не могут быть ни НП, ни улицей.
_NUMERIC_SEGMENT_RE = re.compile(
    r"^(?:"
    r"д|дом|к|кор|корп|корпус|стр|строение|кв|квартира|литера?|уч|участок"
    r"|пом|помещение|оф|офис|бокс|гараж"
    # Хвост `$` обязателен: сегмент считается номером, только если КРОМЕ
    # числа в нём ничего нет. Иначе «8 марта» (числовое имя улицы) было бы
    # принято за номер дома и улица потерялась бы целиком.
    r")?\.?\s*\d+[а-яё]?(?:\s*[/-]\s*\d+[а-яё]?)?$",
    flags=re.UNICODE,
)


# ---------------------------------------------------------------------------
# Street-key normalization — the crux of match rate
# ---------------------------------------------------------------------------

# Comma-separated admin chunks to drop wholesale: «Россия», «РФ», a region
# («Свердловская область» / «...обл.»), a district («... р-н» / «... район»),
# a city («г. Екатеринбург» / «Екатеринбург»). Each pattern matches a WHOLE
# comma-segment (anchored ^…$ against the segment) so it can never bite a
# partial word — segments that don't match are kept verbatim. Order doesn't
# matter; we test every leading segment until one fails to match.
_ADMIN_SEGMENT_RES = [
    re.compile(r"^(?:россия|рф|российская\s+федерация)$", flags=re.UNICODE),
    re.compile(r"^[а-яё][а-яё\s.-]*\bобл(?:асть|\.)?$", flags=re.UNICODE),
    re.compile(
        r"^[а-яё][а-яё\s.-]*\b(?:р-н|район|округ|край|республика)$",
        flags=re.UNICODE,
    ),
]
# NB: сегменты-НАСЕЛЁННЫЕ ПУНКТЫ («г. Екатеринбург», «екатеринбург») здесь
# больше НЕ выбрасываются — они несут вторую половину ключа и разбираются
# `_locality_name` по реестру регионов. Хардкод `^екатеринбург$` уехал туда же:
# список городов региона живёт в `REGIONS[region_code].cities`, а не здесь.

# Street-type token at the START of the street segment. Stripped because
# deals.address sometimes omits it entirely ('Екатеринбург, Малышева'), so the
# type word must NOT be part of the key or '<type> малышева' vs 'малышева'
# would diverge. Hyphenated forms first (longest-match), then short variants.
# `\b` after the token forbids matching the head of a real name (e.g. the 'ал'
# of 'алмазная' or the 'пр' of 'пришвина').
_STREET_TYPE_RE = re.compile(
    r"^(?:"
    r"пр-кт|пр-т|пр-д|б-р|кв-л"
    r"|улица|проспект|переулок|бульвар|шоссе|набережная|проезд|тракт"
    r"|площадь|микрорайон|тупик|аллея|квартал"
    r"|ул|пр|пер|наб|пл|мкр|туп"
    r")(?:\.|\b)\s+",
    flags=re.UNICODE,
)

# District / apartment / building tail INSIDE the street segment. Anchored on a
# left boundary ((?<=^)|(?<=[\s,])) so the keyword is a standalone token, never
# the middle of a word ('к' must not bite 'катеринбург'). Drops everything from
# the marker to end-of-segment.
_DISTRICT_SUFFIX_RE = re.compile(r"\s*[·|].*$", flags=re.UNICODE)
# Right side requires a DIGIT after the token (`кв 12`, `корп. 2`, `стр 10`) so the
# alternation can't eat real street names that merely START with these letters
# («Строителей», «Офицеров», «Корпусная», «Помолова» → would collapse to 'ул.'
# garbage key → wrong centroid → wrong coords). Caught in pre-push review.
_APT_SUFFIX_RE = re.compile(
    r"(?:^|(?<=[\s,]))(?:кв|корп|оф|пом|стр|строение|подъезд)\.?\s*\d.*$",
    flags=re.UNICODE,
)

# Whitespace collapse.
_WS_RE = re.compile(r"\s+")

# A street name that legitimately STARTS with a number followed by a word —
# '8 марта', '1905 года', '40 лет октября'. We must keep these intact while
# still stripping a pure house number like '125' or '44-а'.
_NUMERIC_STREET_RE = re.compile(r"^\d+\s+[а-яё]", flags=re.UNICODE)


def _strip_house_tail(segment: str) -> str:
    """Remove a trailing house-number / building token from a street segment.

    'малышева 125'      → 'малышева'
    'большакова, 17'    → 'большакова'
    'крауля 48/2'       → 'крауля'
    'репина 75/2 стр.'  → 'репина'   (apt suffix already stripped upstream)
    '8 марта 100'       → '8 марта'  (leading numeric street preserved)
    '8 марта'           → '8 марта'
    'малышева'          → 'малышева' (no number → unchanged)

    Strategy: walk tokens left→right, keeping tokens until we hit one that
    starts with a digit AND is not the leading numeric-street token. A token
    is "house-like" if it starts with a digit; the only digit-leading token we
    keep is position 0 of a recognized numeric street name ('8 марта').
    """
    seg = segment.replace(",", " ")
    seg = _WS_RE.sub(" ", seg).strip()
    if not seg:
        return ""
    tokens = seg.split(" ")
    keep: list[str] = []
    numeric_street = bool(_NUMERIC_STREET_RE.match(seg))
    for i, tok in enumerate(tokens):
        if tok and tok[0].isdigit():
            # First token of a numeric street ('8' in '8 марта') is kept; any
            # later digit-leading token is a house number → stop here.
            if i == 0 and numeric_street:
                keep.append(tok)
                continue
            break
        keep.append(tok)
    return " ".join(keep).strip()


def _is_admin_segment(seg: str) -> bool:
    """Сегмент — страна/регион/район (выбрасывается целиком)."""
    return any(rx.match(seg) for rx in _ADMIN_SEGMENT_RES)


def _is_numeric_segment(seg: str) -> bool:
    """Сегмент — номер дома/квартиры/корпуса («125», «д 5», «кв 12»)."""
    return bool(_NUMERIC_SEGMENT_RE.match(seg))


def _normalize_place(value: str | None) -> str:
    """Нормализованное имя НП: lower, ё→е, без типа («г. Химки» → «химки»)."""
    if not value:
        return ""
    v = unicodedata.normalize("NFC", value).lower().replace("ё", "е")
    v = _WS_RE.sub(" ", v).strip()
    v = _LOCALITY_TYPE_RE.sub("", v)
    return _WS_RE.sub(" ", v).strip(" .,")


def _locality_name(seg: str, cities: frozenset[str]) -> str | None:
    """Имя НП, если сегмент — населённый пункт, иначе None.

    Два признака: явный тип («г Химки», «д Сабурово», «снт Заря») — имя берём
    как есть; ИЛИ голое имя, которое реестр региона знает как город
    (`REGIONS[region_code].cities`). Голое незнакомое имя здесь НЕ считается
    НП — иначе «Малышева, 125» прочиталось бы как НП «Малышева»; такой случай
    ловит позиционный fallback в `_split_place_street`.
    """
    m = _LOCALITY_TYPE_RE.match(seg)
    if m:
        name = seg[m.end() :]
    else:
        if _STREET_TYPE_RE.match(seg) or _is_numeric_segment(seg):
            return None
        if _WS_RE.sub(" ", seg).strip() not in cities:
            return None
        name = seg
    name = _WS_RE.sub(" ", name).strip(" .,")
    return name or None


def _split_place_street(
    address: str | None, region_code: int = _DEFAULT_REGION_CODE
) -> tuple[str, str]:
    """Разложить адрес на (населённый пункт, улица) — обе половины ключа.

    ПОЧЕМУ НП обязан быть в ключе. Раньше функция звалась `_street_key` и
    возвращала ГОЛОЕ имя улицы, выбрасывая НП. Для одного города (скрипт жил
    ЕКБ-only, с хардкодом `^екатеринбург$`) это безвредно. Для Московской
    области — тихая катастрофа: «Ленина» / «Центральная» / «Советская» есть
    почти в каждом из ~970 НП области, дома со всех таких улиц слиплись бы в
    одно ведро, а среднее их координат — правдоподобная точка ПОСЕРЕДИНЕ между
    городами. Сделка уезжает на десятки километров, и ни одна проверка этого не
    замечает: координата валидная, внутри области, рядом есть дома.

    Разбор:
      1. NFC, lower, ё→е, снять хвост-район (' · р-н ...', ' | ...').
      2. Порезать на comma-сегменты, выбросить страну/регион/район.
      3. Первый сегмент-НП (`_locality_name`) → place, улицу ищем ПОСЛЕ него.
      4. Улица — первый не-числовой сегмент из остатка.
      5. Fallback «НП без типа и вне реестра» («Сабурово, Луговая» — ровно
         формат deals.address по области): если НП не нашёлся, а не-числовых
         сегментов >= 2 и первый не начинается с типа улицы — первый считается
         НП, второй улицей.
      6. Улицу чистим как раньше: квартира/корпус, тип улицы, номер дома.

    Неизвестный `region_code` -> KeyError реестра, не молчаливый ''.
    """
    cities = REGIONS[region_code].cities
    if not address:
        return "", ""
    s = unicodedata.normalize("NFC", address).lower().replace("ё", "е")
    s = _WS_RE.sub(" ", s).strip()
    if not s:
        return "", ""
    s = _DISTRICT_SUFFIX_RE.sub("", s)

    segments = [seg.strip() for seg in s.split(",") if seg.strip()]
    non_admin = [seg for seg in segments if not _is_admin_segment(seg)]
    if not non_admin:
        return "", ""

    place = ""
    rest = non_admin
    for i, seg in enumerate(non_admin):
        name = _locality_name(seg, cities)
        if name:
            place = name
            rest = non_admin[i + 1 :]
            break

    street_seg = next((seg for seg in rest if not _is_numeric_segment(seg)), "")

    if not place:
        usable = [seg for seg in non_admin if not _is_numeric_segment(seg)]
        if len(usable) >= 2 and not _STREET_TYPE_RE.match(usable[0]):
            place = _WS_RE.sub(" ", usable[0]).strip(" .,")
            street_seg = usable[1]

    if not street_seg:
        return place, ""

    street_seg = _APT_SUFFIX_RE.sub("", street_seg).strip()
    street_seg = _STREET_TYPE_RE.sub("", street_seg).strip()
    street_seg = _strip_house_tail(street_seg)
    return place, _WS_RE.sub(" ", street_seg).strip()


def _street_key(address: str | None, region_code: int = _DEFAULT_REGION_CODE) -> str:
    """Только уличная половина ключа. Как ключ join'а САМА ПО СЕБЕ не годится
    (одноимённые улицы разных НП) — см. `_address_key`; оставлена для отчётов
    и тестов нормализации улицы.

    Примеры (region 66):
      'Екатеринбург, ул. Малышева, 125'   -> 'малышева'
      'г Екатеринбург, улица Малышева'    -> 'малышева'
      'Свердловская обл., Екатеринбург, ул. Большакова, 17' -> 'большакова'
      'улица Яскина, 12 · р-н Октябрьский' -> 'яскина'
      'Екатеринбург, ул. 8 Марта, 100'    -> '8 марта'
    """
    return _split_place_street(address, region_code)[1]


def _address_key(
    address: str | None,
    region_code: int = _DEFAULT_REGION_CODE,
    city: str | None = None,
) -> tuple[int, str, str] | None:
    """Составной ключ join'а: (регион, населённый пункт, улица). None — мусор.

    `city` (у deals колонка заполнена на 100% и надёжнее текста адреса)
    переопределяет НП, разобранный из адреса.

    Пустой НП: у региона С городом-ядром (66) подставляется `city_token` —
    ровно историческое поведение «всё, что без города, это Екатеринбург». У
    региона БЕЗ ядра (50) подставлять нечего, и ключ отбрасывается: сделка без
    распознанного НП лучше останется без координат, чем сядет в случайный
    город области.
    """
    place, street = _split_place_street(address, region_code)
    from_city = _normalize_place(city)
    if from_city:
        place = from_city
    if len(street) < _MIN_KEY_LEN:
        return None
    if not place:
        region = REGIONS[region_code]
        if not region.has_city_core:
            return None
        place = region.city_token
    if len(place) < _MIN_KEY_LEN:
        return None
    return (region_code, place, street)


def _spread_km(points: list[tuple[float, float]], lat_c: float, lon_c: float) -> float:
    """Максимальное удаление точки ведра от его центроида, км (equirectangular).

    Проекция плоская — на масштабе одного НП (единицы километров) ошибка
    сотые доли процента, а формула дешевле haversine на каждом из десятков
    тысяч домов.
    """
    worst = 0.0
    cos_lat = math.cos(math.radians(lat_c))
    for lat, lon in points:
        dy = (lat - lat_c) * 111.32
        dx = (lon - lon_c) * 111.32 * cos_lat
        worst = max(worst, math.hypot(dx, dy))
    return worst


# ---------------------------------------------------------------------------
# Domain types
# ---------------------------------------------------------------------------


@dataclass
class Centroid:
    """Центроид одного ключа (регион, НП, улица) по домам этого ключа."""

    lat: float
    lon: float
    house_count: int
    # Максимальное удаление дома ведра от центроида, км (sanity-чек склейки).
    spread_km: float = 0.0


@dataclass
class DealRow:
    """Minimal deals fields needed for the centroid lookup."""

    id: int
    address: str | None
    # deals.city — росреестровая колонка; по области заполнена у 100% строк без
    # geom и надёжнее текста адреса, поэтому переопределяет НП из адреса.
    city: str | None = None


@dataclass
class Stats:
    """Final-summary counters."""

    processed: int = 0
    geocoded: int = 0
    no_street_match: int = 0
    failed: int = 0
    # Вёдер выброшено sanity-чеком разброса (склейка нескольких НП).
    dropped_spread: int = 0
    # street_key → count of deals that had no house centroid (dry-run report).
    unmatched_streets: Counter[str] = field(default_factory=Counter)


# ---------------------------------------------------------------------------
# Source queries
# ---------------------------------------------------------------------------


def _build_centroid_map(
    db: Session,
    region_code: int = _DEFAULT_REGION_CODE,
    max_spread_km: float = _DEFAULT_MAX_SPREAD_KM,
) -> dict[tuple[int, str, str], Centroid]:
    """Центроиды по ключу (регион, НП, улица) из houses с координатами.

    Фильтр региона обязателен: без него карта строится по домам ВСЕХ регионов,
    и прогон по одному региону тихо тащит чужие координаты.

    `houses.region_code` добавлен поздней миграцией (272_houses_region_code.sql)
    и у части строк NULL. Отбрасывать их нельзя (это ударило бы по покрытию 66),
    поэтому строка без региона принимается по географии — если её координаты
    внутри `bbox_region` реестра. Это именно гео-проверка, а не догадка о
    происхождении строки.

    Агрегация в Python, а не в SQL: вывод ключа должен быть ОДНИМ И ТЕМ ЖЕ
    кодом на обеих сторонах join'а, дублировать регексы в plpgsql — верный
    дрейф.
    """
    region = REGIONS[region_code]
    lat_min, lat_max, lon_min, lon_max = region.bbox_region
    rows = (
        db.execute(
            text(
                "SELECT address, lat, lon "
                "FROM houses "
                "WHERE geom IS NOT NULL "
                "  AND lat IS NOT NULL "
                "  AND lon IS NOT NULL "
                "  AND address IS NOT NULL "
                "  AND length(trim(address)) > 0 "
                "  AND ( region_code = CAST(:rc AS smallint) "
                "     OR ( region_code IS NULL "
                "          AND lat BETWEEN CAST(:lat_min AS double precision) "
                "                      AND CAST(:lat_max AS double precision) "
                "          AND lon BETWEEN CAST(:lon_min AS double precision) "
                "                      AND CAST(:lon_max AS double precision) ) )"
            ),
            {
                "rc": region_code,
                "lat_min": lat_min,
                "lat_max": lat_max,
                "lon_min": lon_min,
                "lon_max": lon_max,
            },
        )
        .mappings()
        .all()
    )

    acc: dict[tuple[int, str, str], list[tuple[float, float]]] = {}
    for r in rows:
        key = _address_key(r["address"], region_code)
        if key is None:
            continue
        acc.setdefault(key, []).append((float(r["lat"]), float(r["lon"])))

    out: dict[tuple[int, str, str], Centroid] = {}
    dropped = 0
    for key, points in acc.items():
        n = len(points)
        lat_c = sum(lat for lat, _ in points) / n
        lon_c = sum(lon for _, lon in points) / n
        spread = _spread_km(points, lat_c, lon_c)
        if spread > max_spread_km:
            # Одна «улица» шириной в десятки километров — это не улица, а
            # слипшиеся одноимённые улицы разных НП (или один НП, записанный
            # двумя способами). Среднее таких точек — координата в поле между
            # городами; отдавать её сделке нельзя, ведро выбрасывается.
            dropped += 1
            logger.warning(
                "centroid bucket dropped: key=%s houses=%d spread=%.1f km > %.1f km",
                key,
                n,
                spread,
                max_spread_km,
            )
            continue
        out[key] = Centroid(lat=lat_c, lon=lon_c, house_count=n, spread_km=spread)

    if dropped:
        logger.warning(
            "sanity: %d/%d вёдер выброшено по разбросу > %.1f km",
            dropped,
            len(acc),
            max_spread_km,
        )
    return out


def _region_predicate(column: str = "region_code") -> str:
    """SQL-предикат «строка принадлежит региону :rc» для deals.

    `deals.region_code` проставлен миграцией 177_deals_city_region.sql; строки,
    существовавшие ДО неё, по построению екатеринбургские (в скрипт импорта
    был зашит префикс 'Екатеринбург, '), поэтому NULL засчитывается региону 66
    и только ему. Для любого другого региона NULL — не кандидат.
    """
    return f"( {column} = CAST(:rc AS int)   OR ( {column} IS NULL AND CAST(:rc AS int) = 66 ) )"


def _select_deals_without_coords(
    db: Session, limit: int, region_code: int = _DEFAULT_REGION_CODE
) -> list[DealRow]:
    """deals needing coords (lat IS NULL) в пределах ОДНОГО региона.

    Matches `deals_geocode_pending_idx` (WHERE lat IS NULL). A successful
    UPDATE sets lat NOT NULL, dropping the row out on the next run.
    """
    rows = (
        db.execute(
            text(
                "SELECT id, address, city "
                "FROM deals "
                "WHERE lat IS NULL "
                "  AND address IS NOT NULL "
                "  AND length(trim(address)) > 0 "
                "  AND " + _region_predicate() + " "
                "ORDER BY id "
                "LIMIT CAST(:lim AS int)"
            ),
            {"lim": limit, "rc": region_code},
        )
        .mappings()
        .all()
    )
    return [DealRow(id=r["id"], address=r["address"], city=r.get("city")) for r in rows]


# ---------------------------------------------------------------------------
# DB writer
# ---------------------------------------------------------------------------


def _update_deal_coords(
    db: Session,
    *,
    deal_id: int,
    lat: float,
    lon: float,
    region_code: int = _DEFAULT_REGION_CODE,
) -> None:
    """UPDATE deals SET lat/lon + geocode_tried_at=NOW(); geom auto-fills.

    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. Setting geocode_tried_at marks
    the row processed for the partial index / future cron passes.
    """
    db.execute(
        text(
            "UPDATE deals "
            "   SET lat = CAST(:lat AS double precision), "
            "       lon = CAST(:lon AS double precision), "
            "       geocode_tried_at = NOW() "
            " WHERE id = CAST(:id AS bigint) "
            "   AND " + _region_predicate()
        ),
        {"id": deal_id, "lat": lat, "lon": lon, "rc": region_code},
    )


# ---------------------------------------------------------------------------
# Main loop
# ---------------------------------------------------------------------------


def _run_backfill(
    db: Session,
    deals: list[DealRow],
    centroids: dict[tuple[int, str, str], Centroid],
    *,
    batch: str,
    dry_run: bool,
    region_code: int = _DEFAULT_REGION_CODE,
) -> Stats:
    """For each deal, look up its street centroid and UPDATE lat/lon.

    Per-row SAVEPOINT (`db.begin_nested()`) so one bad UPDATE can't abort the
    batch (backend.md SAVEPOINT rule). Per-row commit on success → a crash
    mid-run leaves already-geocoded rows persisted, and resume picks up the
    rest (lat IS NULL filter).
    """
    stats = Stats()

    for i, deal in enumerate(deals, start=1):
        key = _address_key(deal.address, region_code, city=deal.city)
        centroid = centroids.get(key) if key is not None else None

        if centroid is None:
            stats.no_street_match += 1
            # Track the raw key (or a sentinel) so the dry-run report can show
            # which streets we're missing. Empty key → '<no-street-parsed>'.
            stats.unmatched_streets["/".join(key[1:]) if key else "<no-street-parsed>"] += 1
            stats.processed += 1
            if dry_run and i % _LOG_EVERY == 0:
                logger.info(
                    "DRY-RUN deal_id=%s addr=%r key=%r → no centroid",
                    deal.id,
                    (deal.address or "")[:60],
                    key,
                )
            continue

        if dry_run:
            stats.geocoded += 1
            stats.processed += 1
        else:
            try:
                with db.begin_nested():
                    _update_deal_coords(
                        db,
                        deal_id=deal.id,
                        lat=centroid.lat,
                        lon=centroid.lon,
                        region_code=region_code,
                    )
                # Per-row commit so resume picks up exactly where we crashed.
                db.commit()
                stats.geocoded += 1
                stats.processed += 1
            except Exception as exc:  # defensive — isolate one bad UPDATE
                db.rollback()
                stats.failed += 1
                logger.warning("db_write failed for deal_id=%s: %s", deal.id, exc)

        if i % _LOG_EVERY == 0:
            logger.info(
                "batch=%s progress %d/%d geocoded=%d no_match=%d failed=%d",
                batch,
                i,
                len(deals),
                stats.geocoded,
                stats.no_street_match,
                stats.failed,
            )

    return stats


# ---------------------------------------------------------------------------
# Dry-run reporting
# ---------------------------------------------------------------------------


def _report_dry_run(
    stats: Stats,
    centroids: dict[tuple[int, str, str], Centroid],
    *,
    total_deals_null: int,
    candidates: int,
) -> None:
    """Log coverage projection + the top-10 unmatched deal streets.

    `candidates` is how many lat-IS-NULL deals we actually scanned this run
    (capped by --limit); `total_deals_null` is the full backlog so the
    projected % extrapolates honestly when --limit < backlog.
    """
    distinct_streets = len(centroids)
    matched = stats.geocoded
    scanned = candidates

    # Coverage on the scanned slice, then projected onto the full backlog.
    match_rate = (matched / scanned) if scanned else 0.0
    projected = round(match_rate * total_deals_null)

    logger.info("─" * 60)
    logger.info("DRY-RUN SUMMARY (no DB writes)")
    logger.info("distinct (region, НП, улица) keys with a house centroid: %d", distinct_streets)
    logger.info("deals scanned this run (lat IS NULL, capped by --limit): %d", scanned)
    logger.info("deals matched to a centroid: %d", matched)
    logger.info("deals with no street match: %d", stats.no_street_match)
    logger.info("match rate on scanned slice: %.1f%%", match_rate * 100.0)
    logger.info(
        "full lat-IS-NULL backlog: %d → projected matched ≈ %d (%.1f%%)",
        total_deals_null,
        projected,
        match_rate * 100.0,
    )
    logger.info("top-10 unmatched deal keys НП/улица (by row count):")
    for street, cnt in stats.unmatched_streets.most_common(10):
        logger.info("    %6d  %s", cnt, street)
    logger.info("─" * 60)


# ---------------------------------------------------------------------------
# CLI
# ---------------------------------------------------------------------------


def _count_deals_null(db: Session, region_code: int = _DEFAULT_REGION_CODE) -> int:
    """Full count of deals WHERE lat IS NULL в этом регионе — знаменатель."""
    row = db.execute(
        text(
            "SELECT count(*) AS n FROM deals "
            "WHERE lat IS NULL AND address IS NOT NULL AND length(trim(address)) > 0 "
            "  AND " + _region_predicate()
        ),
        {"rc": region_code},
    ).first()
    return int(row[0]) if row else 0


def _parse_args(argv: list[str] | None = None) -> argparse.Namespace:
    """argparse setup, factored out for testability."""
    p = argparse.ArgumentParser(
        description=(
            "Issue #569 Step 2 — backfill deals.lat/lon from per-street house "
            "centroids (no external geocoder)."
        ),
    )
    p.add_argument(
        "--limit",
        type=int,
        default=_DEFAULT_LIMIT,
        help=f"Max deals to process this run (default {_DEFAULT_LIMIT}).",
    )
    p.add_argument(
        "--batch",
        default=f"deals_geo_{date.today().isoformat()}",
        help="Log batch label (does not affect DB filters — logs only).",
    )
    p.add_argument(
        "--region-code",
        type=int,
        default=_DEFAULT_REGION_CODE,
        choices=sorted(REGIONS),
        help=(
            "Регион прогона (default %(default)s). Отбирает и обновляет ТОЛЬКО "
            "строки этого региона; входит в ключ join'а."
        ),
    )
    p.add_argument(
        "--max-spread-km",
        type=float,
        default=_DEFAULT_MAX_SPREAD_KM,
        help=(
            "Порог разброса домов внутри ключа (default %(default)s км). Ведро "
            "с большим разбросом — склейка нескольких НП, оно выбрасывается."
        ),
    )
    p.add_argument(
        "--dry-run",
        action="store_true",
        help=(
            "Report distinct house streets, projected coverage %% of the "
            "lat-IS-NULL backlog, and the top-10 unmatched deal streets. "
            "No DB writes."
        ),
    )
    return p.parse_args(argv)


def main(argv: list[str] | None = None) -> int:
    """CLI entry point. Returns the number of deals geocoded this run."""
    args = _parse_args(argv)
    logger.info(
        "starting batch=%s region=%s limit=%s max_spread_km=%s dry_run=%s",
        args.batch,
        args.region_code,
        args.limit,
        args.max_spread_km,
        args.dry_run,
    )

    db = SessionLocal()
    try:
        centroids = _build_centroid_map(
            db, region_code=args.region_code, max_spread_km=args.max_spread_km
        )
        logger.info(
            "built centroid map: %d distinct keys (region %s)", len(centroids), args.region_code
        )
        if not centroids:
            logger.warning(
                "no house centroids for region %s — houses has no geocoded rows there; "
                "nothing to do",
                args.region_code,
            )
            return 0

        deals = _select_deals_without_coords(db, args.limit, region_code=args.region_code)
        logger.info("loaded deals without coords: %d", len(deals))
        if not deals:
            logger.info("nothing to do — no deals with lat IS NULL and an address")
            return 0

        stats = _run_backfill(
            db,
            deals,
            centroids,
            batch=args.batch,
            dry_run=args.dry_run,
            region_code=args.region_code,
        )

        if args.dry_run:
            total_null = _count_deals_null(db, region_code=args.region_code)
            _report_dry_run(
                stats,
                centroids,
                total_deals_null=total_null,
                candidates=len(deals),
            )

        logger.info(
            "done: batch=%s processed=%d geocoded=%d no_street_match=%d failed=%d",
            args.batch,
            stats.processed,
            stats.geocoded,
            stats.no_street_match,
            stats.failed,
        )
        return stats.geocoded
    finally:
        db.close()


if __name__ == "__main__":  # pragma: no cover
    raise SystemExit(0 if main() >= 0 else 1)