From 279515383cf554e64e2b6a87dbd966b0b34bcd63 Mon Sep 17 00:00:00 2001 From: bot-backend Date: Fri, 3 Jul 2026 08:04:01 +0500 Subject: [PATCH] =?UTF-8?q?feat(ekgas):=20=D0=B3=D0=B5=D0=BE=D0=BA=D0=BE?= =?UTF-8?q?=D0=B4=20=D1=82=D0=BE=D1=87=D0=B5=D0=BA=20=D0=B2=D1=8B=D1=85?= =?UTF-8?q?=D0=BE=D0=B4=D0=B0=20=D1=87=D0=B5=D1=80=D0=B5=D0=B7=20DaData=20?= =?UTF-8?q?(#2119=20B2,=20PR-3/4)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../site_finder/ekgas_outlet_loader.py | 215 ++++++++++++++++++ backend/tests/test_ekgas_outlet_loader.py | 177 +++++++++++++- 2 files changed, 391 insertions(+), 1 deletion(-) diff --git a/backend/app/services/site_finder/ekgas_outlet_loader.py b/backend/app/services/site_finder/ekgas_outlet_loader.py index 479a179f..a5a0c613 100644 --- a/backend/app/services/site_finder/ekgas_outlet_loader.py +++ b/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()) diff --git a/backend/tests/test_ekgas_outlet_loader.py b/backend/tests/test_ekgas_outlet_loader.py index d8052cf0..b88be402 100644 --- a/backend/tests/test_ekgas_outlet_loader.py +++ b/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