This commit is contained in:
parent
c438c9a9e8
commit
cc9516ab3d
2 changed files with 391 additions and 1 deletions
|
|
@ -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())
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue