feat(ekgas): геокод точек выхода через DaData (#2119 B2, PR-3/4)

backend/app/services/site_finder/ekgas_outlet_loader.py

@@ -41,10 +41,12 @@ None, знак сохраняется: дефицит −0.084 хранится
 jsonb (будущие/неизвестные колонки не теряются).
 """
 
+import argparse
 import io
 import json
 import logging
 import re
+import sys
 from datetime import date
 from urllib.parse import urljoin
 
@@ -55,6 +57,7 @@ from sqlalchemy import text
 from sqlalchemy.orm import Session
 
 from app.core.db import SessionLocal
+from app.services.dadata_client import clean_address
 from app.services.site_finder.gazprom_grs_loader import _ru_ssl_context, normalize_grs_name
 from app.services.site_finder.rosseti_reserve_loader import parse_reserve_number
 
@@ -67,6 +70,25 @@ CATALOG_URL = "https://ekgas.ru/raskrytie-informacii/raskrytie-informatsii-1"
 
 _HTTP_TIMEOUT = 60
 
+# ── Геокод address_extracted → geom (PR-3, #2119 B2) ──────────────────────────
+# Cap вызовов DaData за один прогон (адресов на проде ~164 → влезает; страхует
+# квоту от разбухшего файла). Тот же порядок, что GEO_DADATA_CALL_LIMIT в
+# objective_backfill.
+_GEO_DADATA_CAP = 200
+
+# Префикс города: address_extracted — адресный ХВОСТ без города («ул. Стрелочников,
+# д. 2»). DaData на строке без города отдаёт qc_geo=5 (нет координат) — прод-урок
+# objective_backfill 2026-07-03. Точки выхода ЕКБ-газа привязаны к Екатеринбургу и
+# окрестностям → безопасно префиксуем городом, если его в строке нет.
+_EKB_PREFIX = "г Екатеринбург, "
+
+# Sanity-bbox Екатеринбурга и ближних окрестностей (Арамиль/Реж/Берёзовский/Кольцово):
+# DaData на кривом/усечённом адресе может «дорисовать» другой город — точку вне bbox
+# НЕ пишем (честность: лучше NULL geom, чем ложная координата в Москве). Границы
+# грубые с запасом: lat 56.5..57.2, lon 60.0..61.2.
+_EKB_BBOX_LAT = (56.5, 57.2)
+_EKB_BBOX_LON = (60.0, 61.2)
+
 # Целевой лист XLSX (единственный) — берём активный, если имя не совпало.
 # 1-indexed строки листа: подписи колонок = row 10, нумерация 1..7 = row 11, данные с 12.
 # Сетка НЕ хардкодится по номерам строк/колонок: живой файл за май 2026 имеет
@@ -497,6 +519,10 @@ def _upsert_outlet_rows(db: Session, records: list[dict], meta: dict) -> dict[st
                         )
                         ON CONFLICT (grs_name_norm, outlet_name, period_month, is_forecast)
                         DO UPDATE
+                           -- geom НАМЕРЕННО НЕ в SET: geocode_gas_outlets (PR-3)
+                           -- пишет его отдельным UPDATE. Помесячный рефреш INSERT-ит
+                           -- без geom → EXCLUDED.geom = NULL; включив его в SET, мы бы
+                           -- ЗАТЁРЛИ уже геокоженную точку в NULL. Оставляем как есть.
                            SET grs_name                 = EXCLUDED.grs_name,
                                consumer_type            = EXCLUDED.consumer_type,
                                consumption_group        = EXCLUDED.consumption_group,
@@ -587,3 +613,192 @@ def load_gas_outlet_points(db: Session | None = None) -> dict:
     counts["is_forecast"] = chosen["is_forecast"]
     logger.info("load_gas_outlet_points done: %s", counts)
     return counts
+
+
+# ── Геокод address_extracted → geom (PR-3) ────────────────────────────────────
+
+# Кандидаты геокода: строки последнего period_month с адресом, но БЕЗ geom.
+# MAX(period_month) в подзапросе — работаем только по свежайшему снапшоту (старые
+# месяцы не трогаем). geom стабилен между месяцами (UPSERT его не затирает — см.
+# DO UPDATE выше), так что уже геокоженные строки прошлых месяцев не пере-геокодим.
+_GEOCODE_CANDIDATES_SQL = text("""
+    SELECT id, address_extracted
+    FROM gas_grs_outlet_points
+    WHERE address_extracted IS NOT NULL
+      AND geom IS NULL
+      AND period_month = (SELECT MAX(period_month) FROM gas_grs_outlet_points)
+    ORDER BY id
+""")
+
+# UPDATE geom = точка WGS84. CAST(:lon/:lat AS float) — psycopg3-канон (backend.md:
+# НИКОГДА :lon::float). Пишем ТОЛЬКО если geom всё ещё NULL (idempotent при повторе).
+_GEOCODE_UPDATE_SQL = text("""
+    UPDATE gas_grs_outlet_points
+       SET geom = ST_SetSRID(
+                      ST_MakePoint(CAST(:lon AS float), CAST(:lat AS float)),
+                      4326
+                  )
+     WHERE id = CAST(:id AS bigint)
+       AND geom IS NULL
+""")
+
+
+def _ekb_prefixed(address: str) -> str:
+    """address_extracted → строка для DaData с городом-префиксом при его отсутствии.
+
+    «ул. Стрелочников, д. 2» → «г Екатеринбург, ул. Стрелочников, д. 2». Если город
+    уже есть в строке («г.Екатеринбург, …» / «г. Реж, …») — не дублируем.
+    """
+    if "екатеринбург" in address.lower() or _HAS_CITY_RE.search(address):
+        return address
+    return f"{_EKB_PREFIX}{address}"
+
+
+def _in_ekb_bbox(lat: float, lon: float) -> bool:
+    """Точка в bbox Екатеринбурга и окрестностей? (sanity-гейт против чужого города)."""
+    return (
+        _EKB_BBOX_LAT[0] <= lat <= _EKB_BBOX_LAT[1] and _EKB_BBOX_LON[0] <= lon <= _EKB_BBOX_LON[1]
+    )
+
+
+def geocode_gas_outlets(db: Session | None = None, *, apply: bool = False) -> dict:
+    """Геокодит address_extracted точек выхода → geom (последний period_month).
+
+    Дефолт dry-run (apply=False): считает счётчики + печатает первые 20 пар
+    адрес→(lat,lon), НО В БД НЕ ПИШЕТ. apply=True — UPDATE geom.
+
+    Алгоритм:
+      1. SELECT строки последнего period_month c address_extracted IS NOT NULL AND
+         geom IS NULL (уже геокоженные не трогаем — geom стабилен между месяцами).
+      2. Каждый адрес: префикс «г Екатеринбург, » при отсутствии города → clean_address.
+      3. Sanity-bbox ЕКБ+окрестности: координата вне bbox → failed (DaData «дорисовал»
+         чужой город на кривом адресе) — geom НЕ пишем (честность).
+      4. Кэш в памяти (адреса повторяются) + cap _GEO_DADATA_CAP вызовов DaData.
+         Per-row SAVEPOINT (битая строка/UPDATE не валит батч, backend.md).
+
+    Returns счётчики {candidates, geocoded, failed, dadata_calls, cap_hit, applied}.
+    """
+    owns_session = db is None
+    if db is None:
+        db = SessionLocal()
+
+    candidates: list[tuple[int, str]] = [
+        (int(row[0]), str(row[1])) for row in db.execute(_GEOCODE_CANDIDATES_SQL).all()
+    ]
+
+    counts = {
+        "candidates": len(candidates),
+        "geocoded": 0,
+        "failed": 0,
+        "dadata_calls": 0,
+        "cap_hit": False,
+        "applied": apply,
+    }
+    # Кэш прогона: address → (lat, lon) | None (None = «пробовали, DaData не дал/вне
+    # bbox» — кэшируем и его, чтобы не жечь квоту на повтор одного адреса).
+    geo_cache: dict[str, tuple[float, float] | None] = {}
+    sample: list[tuple[str, tuple[float, float]]] = []
+
+    try:
+        for row_id, address in candidates:
+            if address in geo_cache:
+                coords = geo_cache[address]
+            elif counts["dadata_calls"] >= _GEO_DADATA_CAP:
+                counts["cap_hit"] = True
+                counts["failed"] += 1
+                continue
+            else:
+                counts["dadata_calls"] += 1
+                raw = clean_address(_ekb_prefixed(address))
+                # bbox-гейт ДО кэша: вне ЕКБ → трактуем как «нет координат» (None).
+                if raw is not None and not _in_ekb_bbox(raw[0], raw[1]):
+                    logger.info(
+                        "ekgas geocode: %r вне bbox ЕКБ (lat=%.4f lon=%.4f) → failed",
+                        address[:60],
+                        raw[0],
+                        raw[1],
+                    )
+                    raw = None
+                coords = raw
+                geo_cache[address] = coords
+
+            if coords is None:
+                counts["failed"] += 1
+                continue
+
+            lat, lon = coords
+            if len(sample) < 20:
+                sample.append((address, (lat, lon)))
+
+            if not apply:
+                counts["geocoded"] += 1
+                continue
+
+            try:
+                with db.begin_nested():  # SAVEPOINT — откат только этой строки
+                    db.execute(_GEOCODE_UPDATE_SQL, {"id": row_id, "lat": lat, "lon": lon})
+                counts["geocoded"] += 1
+            except Exception as e:
+                logger.warning("ekgas geocode UPDATE failed for id=%s %r: %s", row_id, address, e)
+                counts["failed"] += 1
+
+        if apply:
+            db.commit()
+    except Exception as e:
+        if apply:
+            db.rollback()
+        logger.exception("geocode_gas_outlets: rolled back: %s", e)
+        raise
+    finally:
+        if owns_session:
+            db.close()
+
+    logger.info("geocode_gas_outlets done (apply=%s): %s", apply, counts)
+    if not apply:
+        for addr, (lat, lon) in sample:
+            logger.info("  sample: %r → (%.6f, %.6f)", addr[:80], lat, lon)
+    return counts
+
+
+def _main(argv: list[str] | None = None) -> int:
+    """CLI: помесячная загрузка формы 6 (--load) и геокод адресов → geom (--geocode-outlets).
+
+    RUN-ON-PROD (гео-блок источника + DaData-ключи на проде). Пример:
+        python -m app.services.site_finder.ekgas_outlet_loader --geocode-outlets
+        python -m app.services.site_finder.ekgas_outlet_loader --geocode-outlets --apply
+    """
+    parser = argparse.ArgumentParser(
+        description="Загрузчик точек выхода ГРС «Екатеринбурггаз» (#2119 B2): "
+        "load формы 6 + геокод address_extracted → geom."
+    )
+    parser.add_argument(
+        "--load",
+        action="store_true",
+        help="Загрузить свежайшую форму 6 из каталога ekgas → UPSERT точек выхода.",
+    )
+    parser.add_argument(
+        "--geocode-outlets",
+        action="store_true",
+        help="Геокодить address_extracted (последний period_month, geom IS NULL) → geom.",
+    )
+    parser.add_argument(
+        "--apply",
+        action="store_true",
+        help="Реально писать в БД (для --geocode-outlets). Без флага — dry-run.",
+    )
+    args = parser.parse_args(argv)
+
+    logging.basicConfig(level=logging.INFO, format="%(levelname)s %(name)s: %(message)s")
+
+    if not args.load and not args.geocode_outlets:
+        parser.error("нужен --load и/или --geocode-outlets")
+
+    if args.load:
+        logger.info("load result: %s", load_gas_outlet_points())
+    if args.geocode_outlets:
+        logger.info("geocode result: %s", geocode_gas_outlets(apply=args.apply))
+    return 0
+
+
+if __name__ == "__main__":  # pragma: no cover
+    sys.exit(_main())

backend/tests/test_ekgas_outlet_loader.py

@@ -8,7 +8,9 @@ Pure / mock-based — без реальной сети и БД. Покрывае
   forward-fill block-marker ГРС, «0.000*»→needs_calc (НЕ ноль), дефицит −0.084 сохранён,
   сноска-скип single-cell;
 - extract_address: скобочный/хвостовой адрес извлечён, юрлицо без адреса → None;
-- _dedup_outlet_names: суффикс « (N)»; _upsert_outlet_rows: счётчики (mock-DB).
+- _dedup_outlet_names: суффикс « (N)»; _upsert_outlet_rows: счётчики (mock-DB);
+- geocode_gas_outlets (PR-3): префикс города, bbox-гейт (Москва → failed), кэш
+  повторного адреса, dry-run не пишет, apply пишет geom, cap вызовов DaData.
 """
 
 from __future__ import annotations
@@ -19,6 +21,7 @@ from datetime import date
 from typing import Any
 
 import openpyxl
+import pytest
 
 from app.services.site_finder import ekgas_outlet_loader as ek
 
@@ -512,3 +515,175 @@ def test_upsert_outlet_rows_skip_on_error() -> None:
         "needs_calc": 0,
         "with_address": 0,
     }
+
+
+# ── Геокод address_extracted → geom (PR-3) ────────────────────────────────────
+
+# Координаты внутри bbox ЕКБ (центр — пл. 1905 года ≈ 56.838, 60.597).
+_EKB_COORDS = (56.838, 60.597)
+# Москва — заведомо вне bbox ЕКБ (DaData «дорисовал» чужой город на кривом адресе).
+_MSK_COORDS = (55.7558, 37.6173)
+
+
+class _GeocodeSession:
+    """Fake Session для geocode_gas_outlets: отдаёт кандидатов на SELECT, собирает UPDATE.
+
+    execute(SELECT) → .all() = список (id, address); execute(UPDATE) → пишет params
+    в updates. begin_nested — no-op SAVEPOINT, commit — считает вызовы.
+    """
+
+    def __init__(self, candidates: list[tuple[int, str]]) -> None:
+        self._candidates = candidates
+        self.updates: list[dict] = []
+        self.commits = 0
+
+    @contextmanager
+    def begin_nested(self):  # type: ignore[no-untyped-def]
+        yield
+
+    def execute(self, stmt: Any, params: Any = None) -> Any:
+        sql = str(stmt)
+        if "UPDATE gas_grs_outlet_points" in sql:
+            self.updates.append(params)
+            return _FakeResult(None)
+        # SELECT кандидатов.
+        rows = [(cid, addr) for cid, addr in self._candidates]
+        return _SelectResult(rows)
+
+    def commit(self) -> None:
+        self.commits += 1
+
+    def rollback(self) -> None:  # pragma: no cover - defensive
+        pass
+
+
+class _SelectResult:
+    def __init__(self, rows: list[tuple[int, str]]) -> None:
+        self._rows = rows
+
+    def all(self) -> list[tuple[int, str]]:
+        return self._rows
+
+
+def _patch_geocoder(monkeypatch: pytest.MonkeyPatch, mapping: dict) -> list[str]:
+    """Мокает clean_address: query→coords по mapping (default _EKB_COORDS).
+
+    Возвращает список реально переданных в DaData query-строк (для проверки префикса
+    города и cap-счётчика вызовов).
+    """
+    calls: list[str] = []
+
+    def _fake(query: str) -> tuple[float, float] | None:
+        calls.append(query)
+        if query in mapping:
+            return mapping[query]
+        return _EKB_COORDS
+
+    monkeypatch.setattr(ek, "clean_address", _fake)
+    return calls
+
+
+def test_geocode_prefixes_city(monkeypatch: pytest.MonkeyPatch) -> None:
+    """Адрес без города → DaData-запрос префиксован «г Екатеринбург, »."""
+    calls = _patch_geocoder(monkeypatch, {})
+    db = _GeocodeSession([(1, "ул. Стрелочников, д. 2")])
+
+    counts = ek.geocode_gas_outlets(db, apply=True)
+
+    assert calls == ["г Екатеринбург, ул. Стрелочников, д. 2"]
+    assert counts["geocoded"] == 1
+    assert counts["failed"] == 0
+
+
+def test_geocode_keeps_existing_city(monkeypatch: pytest.MonkeyPatch) -> None:
+    """Адрес с городом («г.Екатеринбург, …») → префикс НЕ дублируется."""
+    calls = _patch_geocoder(monkeypatch, {})
+    db = _GeocodeSession([(1, "г.Екатеринбург, Складской проезд, 4а")])
+
+    ek.geocode_gas_outlets(db, apply=True)
+
+    assert calls == ["г.Екатеринбург, Складской проезд, 4а"]
+
+
+def test_geocode_bbox_gate_rejects_moscow(monkeypatch: pytest.MonkeyPatch) -> None:
+    """Координата вне bbox ЕКБ (Москва) → failed, geom НЕ пишем."""
+    query = "г Екатеринбург, ул. Кривая, 1"
+    _patch_geocoder(monkeypatch, {query: _MSK_COORDS})
+    db = _GeocodeSession([(1, "ул. Кривая, 1")])
+
+    counts = ek.geocode_gas_outlets(db, apply=True)
+
+    assert counts["geocoded"] == 0
+    assert counts["failed"] == 1
+    assert db.updates == []  # geom не тронут
+
+
+def test_geocode_caches_repeated_address(monkeypatch: pytest.MonkeyPatch) -> None:
+    """Один и тот же адрес в 2 строках → DaData дёрнут ОДИН раз, geom у обеих."""
+    calls = _patch_geocoder(monkeypatch, {})
+    db = _GeocodeSession([(1, "ул. Общая, 5"), (2, "ул. Общая, 5")])
+
+    counts = ek.geocode_gas_outlets(db, apply=True)
+
+    assert len(calls) == 1  # кэш: второй адрес не жёг квоту
+    assert counts["dadata_calls"] == 1
+    assert counts["geocoded"] == 2
+    assert {u["id"] for u in db.updates} == {1, 2}
+
+
+def test_geocode_dry_run_does_not_write(monkeypatch: pytest.MonkeyPatch) -> None:
+    """Dry-run (apply=False): считает geocoded, но НЕ UPDATE-ит и НЕ commit-ит."""
+    _patch_geocoder(monkeypatch, {})
+    db = _GeocodeSession([(1, "ул. Стрелочников, д. 2")])
+
+    counts = ek.geocode_gas_outlets(db, apply=False)
+
+    assert counts["geocoded"] == 1
+    assert counts["applied"] is False
+    assert db.updates == []
+    assert db.commits == 0
+
+
+def test_geocode_apply_writes_geom(monkeypatch: pytest.MonkeyPatch) -> None:
+    """apply=True: UPDATE с CAST(:lon/:lat AS float) + ST_SetSRID(ST_MakePoint...)."""
+    _patch_geocoder(monkeypatch, {})
+    db = _GeocodeSession([(7, "ул. Стрелочников, д. 2")])
+
+    counts = ek.geocode_gas_outlets(db, apply=True)
+
+    assert counts["geocoded"] == 1
+    assert db.commits == 1
+    assert len(db.updates) == 1
+    upd = db.updates[0]
+    assert upd["id"] == 7
+    assert upd["lat"] == _EKB_COORDS[0]
+    assert upd["lon"] == _EKB_COORDS[1]
+
+
+def test_geocode_update_sql_uses_cast_and_postgis() -> None:
+    """SQL UPDATE: CAST(:lon/:lat AS float) (psycopg3-канон), ST_SetSRID+ST_MakePoint, 4326."""
+    sql = str(ek._GEOCODE_UPDATE_SQL)
+    assert "CAST(:lon AS float)" in sql
+    assert "CAST(:lat AS float)" in sql
+    assert "ST_SetSRID" in sql
+    assert "ST_MakePoint" in sql
+    assert "4326" in sql
+    assert "geom IS NULL" in sql  # idempotent — не переписываем уже геокоженное
+    # никакого :param::type (psycopg3 игнорит) в этом файле.
+    assert "::" not in sql
+
+
+def test_geocode_cap_limits_dadata_calls(monkeypatch: pytest.MonkeyPatch) -> None:
+    """Превышение cap вызовов DaData → cap_hit, остаток failed без обращения к DaData."""
+    calls = _patch_geocoder(monkeypatch, {})
+    monkeypatch.setattr(ek, "_GEO_DADATA_CAP", 2)
+    # 3 УНИКАЛЬНЫХ адреса, cap=2 → 3-й не геокодится.
+    db = _GeocodeSession([(1, "ул. A, 1"), (2, "ул. B, 2"), (3, "ул. C, 3")])
+
+    counts = ek.geocode_gas_outlets(db, apply=True)
+
+    assert counts["dadata_calls"] == 2
+    assert counts["geocoded"] == 2
+    assert counts["failed"] == 1
+    assert counts["cap_hit"] is True
+    assert len(calls) == 2