From d5c2e2833a02d1919985ac25d7fa5dcad133dc9a Mon Sep 17 00:00:00 2001 From: bot-backend Date: Fri, 3 Jul 2026 10:06:20 +0500 Subject: [PATCH] =?UTF-8?q?feat(site-finder):=20OSM-=D0=B3=D0=B5=D0=BE?= =?UTF-8?q?=D0=BA=D0=BE=D0=B4=20=D0=B3=D0=BE=D1=80=D0=BE=D0=B4=D1=81=D0=BA?= =?UTF-8?q?=D0=B8=D1=85=20=D0=A2=D0=9F/=D0=A0=D0=9F=20=D0=95=D0=9A=D0=91?= =?UTF-8?q?=20=E2=86=92=20geom=20(#2119=20B2)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Открытых координат для 2012 городских ТП нет (разведка: WFS портал-тп.рф слой ТП не публикует; в xlsx ЕЭСК только номер+мощности) — потолок OSM ~119 точек (5.9%). Матч name_norm (чистый номер) ↔ OSM substation ref/name через Overpass (bbox ЕКБ, kumi.systems — overpass-api.de даёт 406); консервативная экстракция (ТП/РП-токен или голый ref, вольтаж мимо), дубль-номера → ambiguous-скип, same-point дедуп ~11м, bbox-sanity, UPDATE geom+geocode_status='osm' WHERE geom IS NULL (idempotent, SAVEPOINT). Dry-run дефолт + CLI --apply; Celery вт 05:45 МСК. Полное покрытие — только B2B-запрос реестра в ЕЭСК (у Anton). --- .../services/site_finder/osm_tp_geocoder.py | 319 ++++++++++++++++++ backend/app/workers/beat_schedule.py | 10 + .../workers/tasks/connection_capacity_sync.py | 26 ++ backend/tests/test_osm_tp_geocoder.py | 286 ++++++++++++++++ 4 files changed, 641 insertions(+) create mode 100644 backend/app/services/site_finder/osm_tp_geocoder.py create mode 100644 backend/tests/test_osm_tp_geocoder.py diff --git a/backend/app/services/site_finder/osm_tp_geocoder.py b/backend/app/services/site_finder/osm_tp_geocoder.py new file mode 100644 index 00000000..d7ee418a --- /dev/null +++ b/backend/app/services/site_finder/osm_tp_geocoder.py @@ -0,0 +1,319 @@ +"""OSM-геокод городских ТП/РП ЕКБ → geom для power_tp_rp_reserves (#2119 B2, финал). + +Хвост §3 «свободная мощность»: у ~2012 ЕКБ-строк power_tp_rp_reserves (ЕЭСК, +«ниже 35» кВ) name_norm = чистый номер («10003»), но geom NULL. OSM в bbox ЕКБ +несёт ~821 substation; часть подписана номером ТП/РП в теге ``ref`` или ``name`` +(«РП-101», «ТП-180», «ТП 1591») — по нему матчим наши строки и заливаем координату. + +Разведка 2026-07-03: матч по номеру даёт ~119 наших ТП, 0 неоднозначностей в срезе. +Консервативно: берём число только если это похоже на номер ТП/РП (ТП/РП/RP/TP-токен +рядом ИЛИ голый ref-число) — «110 кВ» из voltage мимо. Дубль-номера (>1 OSM-фичи на +один номер) честно скипаем в ambiguous. + +RUN-ON-PROD-friendly, но источник (Overpass) без гео-блока — можно и локально. +httpx с явным таймаутом, per-row SAVEPOINT (backend.md), idempotent UPDATE +(geom IS NULL + geocode_status='osm'). Dry-run по умолчанию; CLI ``--apply``. + +Overpass endpoint: overpass.kumi.systems (overpass-api.de даёт 406 на наш POST; +area-запрос роняет соединение — используем ТОЛЬКО bbox). +""" + +from __future__ import annotations + +import argparse +import logging +import re + +import httpx +from sqlalchemy import text +from sqlalchemy.orm import Session + +from app.core.db import SessionLocal + +logger = logging.getLogger(__name__) + +# Overpass-инстанс: overpass-api.de отдаёт 406 на наш POST, kumi стабилен (разведка). +OVERPASS_URL = "https://overpass.kumi.systems/api/interpreter" + +# bbox ЕКБ (south, west, north, east) для Overpass. area-запрос роняет соединение — +# ТОЛЬКО bbox. Тот же прямоугольник используем и для sanity-гейта координаты. +_EKB_BBOX = (56.70, 60.40, 56.97, 60.80) + +_HTTP_TIMEOUT = 60 + +# Overpass QL: node/way/relation с power=substation в bbox, вывод с центроидом (center) +# и тегами. `out center tags;` — для way/relation отдаёт center-точку, для node — саму. +_OVERPASS_QL = ( + "[out:json][timeout:{timeout}];" + "(" + 'node["power"="substation"]({s},{w},{n},{e});' + 'way["power"="substation"]({s},{w},{n},{e});' + 'relation["power"="substation"]({s},{w},{n},{e});' + ");" + "out center tags;" +) + +# Токен «ТП/РП» (кириллица и латиница-двойники) + число 2-5 цифр после него. +# Примеры: «РП-101» → 101, «ТП-180» → 180, «ТП 1591» → 1591, «TP-42» → 42. +# Разделитель между токеном и числом — дефис/пробел/точка/двоеточие (или отсутствует). +_TP_TOKEN_NUM_RE = re.compile( + r"(?:тп|рп|tp|rp)\s*[-–—.:№]?\s*(\d{2,5})\b", + re.IGNORECASE, +) + +# Голый ref-номер целиком: только цифры 2-5 знаков (напр. ref='101' без ТП-токена). +_BARE_NUM_RE = re.compile(r"^\s*(\d{2,5})\s*$") + + +def fetch_osm_substations() -> list[dict]: + """Тянет OSM-подстанции ЕКБ (power=substation) в bbox через Overpass → element-list. + + POST data= (GET area роняет соединение). Скип ``substation=traction`` (тяговые + ж/д подстанции — не городские ТП/РП). httpx с явным таймаутом. Возвращает + сырые Overpass-элементы ``{"type","id","lat"/"center","tags"}``. + """ + ql = _OVERPASS_QL.format( + timeout=_HTTP_TIMEOUT, + s=_EKB_BBOX[0], + w=_EKB_BBOX[1], + n=_EKB_BBOX[2], + e=_EKB_BBOX[3], + ) + resp = httpx.post(OVERPASS_URL, data={"data": ql}, timeout=_HTTP_TIMEOUT) + resp.raise_for_status() + data = resp.json() + elements: list[dict] = (data or {}).get("elements") or [] + + kept: list[dict] = [] + for el in elements: + tags = el.get("tags") or {} + # Тяговые ж/д подстанции — не городские ТП/РП, выкидываем. + if tags.get("substation") == "traction": + continue + kept.append(el) + + logger.info( + "osm_tp_geocoder: Overpass отдал %d substation, после traction-скипа %d", + len(elements), + len(kept), + ) + return kept + + +def extract_tp_number(tags: dict) -> str | None: + """Извлекает номер ТП/РП из тегов OSM-фичи (``ref`` приоритетнее ``name``). + + Консервативно — только когда число выглядит номером ТП/РП, не воль­тажем: + - ``ref`` = голое число 2-5 цифр («101») → берём; + - в ``ref``/``name`` есть ТП/РП/TP/RP-токен + число («РП-101», «ТП 1591») → берём число; + - «110 кВ» / «110 kV» БЕЗ ТП-токена → None (это класс напряжения, не номер). + + Приоритет источника: ref, затем name. Возвращает строку-номер или None. + """ + ref = (tags.get("ref") or "").strip() + name = (tags.get("name") or "").strip() + + # 1. Голый ref-номер (наиболее надёжный сигнал: оператор проставил ref явно). + if ref: + m_bare = _BARE_NUM_RE.match(ref) + if m_bare: + return m_bare.group(1) + + # 2. ТП/РП-токен + число — сперва в ref, потом в name. + for source in (ref, name): + if not source: + continue + m = _TP_TOKEN_NUM_RE.search(source) + if m: + return m.group(1) + + # «110 кВ» и прочий вольтаж без ТП-токена и без голого ref — намеренно None. + return None + + +def _center_coords(el: dict) -> tuple[float, float] | None: + """(lat, lon) из Overpass-элемента: node → lat/lon, way/relation → center. None — нет.""" + if el.get("lat") is not None and el.get("lon") is not None: + return float(el["lat"]), float(el["lon"]) + center = el.get("center") or {} + if center.get("lat") is not None and center.get("lon") is not None: + return float(center["lat"]), float(center["lon"]) + return None + + +def _in_ekb_bbox(lat: float, lon: float) -> bool: + """Координата в bbox ЕКБ? Overpass и так ограничен bbox'ом — проверка дёшева/страховка.""" + s, w, n, e = _EKB_BBOX + return s <= lat <= n and w <= lon <= e + + +def _index_osm_by_number(elements: list[dict]) -> tuple[dict[str, tuple[float, float]], set[str]]: + """Строит {номер → (lat,lon)} по OSM-фичам; номера с >1 фичей → ambiguous-set. + + Номер извлекается ``extract_tp_number``; берётся только фича с валидной координатой + в bbox ЕКБ. Если один номер несут ≥2 разных фичи (разные координаты) — он попадает + в ambiguous и НЕ матчится (честность: не угадываем, какая из ТП «та самая»). + """ + by_number: dict[str, tuple[float, float]] = {} + seen_coords: dict[str, set[tuple[float, float]]] = {} + ambiguous: set[str] = set() + + for el in elements: + tags = el.get("tags") or {} + number = extract_tp_number(tags) + if number is None: + continue + coords = _center_coords(el) + if coords is None: + continue + lat, lon = coords + if not _in_ekb_bbox(lat, lon): + logger.info( + "osm_tp_geocoder: номер %s вне bbox ЕКБ (lat=%.4f lon=%.4f) → скип", + number, + lat, + lon, + ) + continue + + prev = seen_coords.setdefault(number, set()) + # Округляем до ~11 м, чтобы дубли-теги одной ТП (way+node) не считались коллизией. + coord_key = (round(lat, 4), round(lon, 4)) + prev.add(coord_key) + if len(prev) > 1: + ambiguous.add(number) + by_number.pop(number, None) + continue + if number not in ambiguous: + by_number[number] = (lat, lon) + + return by_number, ambiguous + + +# UPDATE geom + geocode_status='osm'. CAST(:x AS ...) — psycopg3-канон (backend.md, +# НИКОГДА :x::float). Пишем ТОЛЬКО ЕКБ + geom всё ещё NULL → idempotent при повторе. +_UPDATE_SQL = text(""" + UPDATE power_tp_rp_reserves + SET geom = ST_SetSRID( + ST_MakePoint(CAST(:lon AS float), CAST(:lat AS float)), + 4326 + ), + geocode_status = 'osm' + WHERE name_norm = CAST(:name_norm AS text) + AND municipality = 'Екатеринбург' + AND geom IS NULL +""") + +# Кандидаты на геокод: ЕКБ-строки без координаты. name_norm = чистый номер. +_CANDIDATES_SQL = text(""" + SELECT name_norm + FROM power_tp_rp_reserves + WHERE municipality = 'Екатеринбург' + AND geom IS NULL + AND name_norm IS NOT NULL + ORDER BY name_norm +""") + + +def geocode_tp_from_osm(db: Session | None = None, *, apply: bool = False) -> dict: + """Матч name_norm ЕКБ-ТП/РП ↔ OSM-номер → UPDATE geom + geocode_status='osm'. + + Дефолт dry-run (apply=False): считает счётчики + логирует первые 20 пар + номер→(lat,lon), НО В БД НЕ ПИШЕТ. apply=True — реально пишет. + + Алгоритм: + 1. fetch OSM-substation ЕКБ (traction уже отсеян). + 2. индекс {номер → координата}; дубль-номер → ambiguous (не матчим). + 3. SELECT ЕКБ-строк с geom IS NULL; matched, если номер есть в индексе. + 4. UPDATE geom WHERE name_norm=... AND municipality='Екатеринбург' AND geom IS NULL + (idempotent). Per-row SAVEPOINT — битый UPDATE не валит батч. + + Returns счётчики {osm_features, numbered, matched, updated, ambiguous, applied}. + """ + owns_session = db is None + if db is None: + db = SessionLocal() + + counts = { + "osm_features": 0, + "numbered": 0, + "matched": 0, + "updated": 0, + "ambiguous": 0, + "applied": apply, + } + sample: list[tuple[str, tuple[float, float]]] = [] + + try: + elements = fetch_osm_substations() + counts["osm_features"] = len(elements) + + by_number, ambiguous = _index_osm_by_number(elements) + counts["numbered"] = len(by_number) + counts["ambiguous"] = len(ambiguous) + + candidates: list[str] = [str(row[0]) for row in db.execute(_CANDIDATES_SQL).all()] + + for name_norm in candidates: + coords = by_number.get(name_norm) + if coords is None: + continue + counts["matched"] += 1 + lat, lon = coords + if len(sample) < 20: + sample.append((name_norm, (lat, lon))) + + if not apply: + continue + + try: + with db.begin_nested(): # SAVEPOINT — откат только этой строки + res = db.execute(_UPDATE_SQL, {"name_norm": name_norm, "lat": lat, "lon": lon}) + if res.rowcount and res.rowcount > 0: + counts["updated"] += res.rowcount + except Exception as e: + logger.warning("osm_tp_geocoder UPDATE failed for %r: %s", name_norm, e) + + if apply: + db.commit() + except Exception as e: + if apply: + db.rollback() + logger.exception("geocode_tp_from_osm: rolled back: %s", e) + raise + finally: + if owns_session: + db.close() + + logger.info("geocode_tp_from_osm done (apply=%s): %s", apply, counts) + if not apply: + for number, (lat, lon) in sample: + logger.info(" sample: ТП/РП %s → (%.6f, %.6f)", number, lat, lon) + return counts + + +def _main(argv: list[str] | None = None) -> int: + """CLI: OSM-геокод городских ТП/РП ЕКБ → geom. Dry-run по умолчанию. + + Примеры: + python -m app.services.site_finder.osm_tp_geocoder # dry-run + python -m app.services.site_finder.osm_tp_geocoder --apply # пишет в БД + """ + parser = argparse.ArgumentParser( + description="OSM-геокод городских ТП/РП ЕКБ (#2119 B2): матч name_norm ↔ " + "OSM substation ref/name → geom + geocode_status='osm'." + ) + parser.add_argument( + "--apply", + action="store_true", + help="Реально писать в БД. Без флага — dry-run (только счётчики + сэмпл).", + ) + args = parser.parse_args(argv) + + logging.basicConfig(level=logging.INFO, format="%(levelname)s %(name)s: %(message)s") + logger.info("osm_tp_geocoder result: %s", geocode_tp_from_osm(apply=args.apply)) + return 0 + + +if __name__ == "__main__": # pragma: no cover + raise SystemExit(_main()) diff --git a/backend/app/workers/beat_schedule.py b/backend/app/workers/beat_schedule.py index 958d2783..c24b3b62 100644 --- a/backend/app/workers/beat_schedule.py +++ b/backend/app/workers/beat_schedule.py @@ -333,6 +333,16 @@ def build_beat_schedule() -> dict: "schedule": _parse_cron("30 5 * * tue"), # 05:30 МСК, вторник "options": {"queue": "celery"}, } + # OSM-геокод городских ТП/РП ЕКБ → geom (Forgejo #2119 Фаза B2, финал). Матч + # name_norm (чистый номер) ↔ OSM substation ref/name через Overpass → + # power_tp_rp_reserves.geom. Источник (Overpass) БЕЗ гео-блока, но держим в том же + # вторник-кластере рядом с ЕЭСК. 05:45 МСК — после газа/ЕЭСК (05:30), до тепла + # (06:00); идемпотентно (geom IS NULL-гейт), вне monday-Overpass-нагрузки. + schedule["osm-tp-geocode-weekly"] = { + "task": "tasks.connection_capacity_sync.sync_osm_tp_geocode", + "schedule": _parse_cron("45 5 * * tue"), # 05:45 МСК, вторник + "options": {"queue": "celery"}, + } # Тепло-резервы (Forgejo #2119 Фаза B2): резерв мощности источников тепла из # реестра раскрытия ФАС (ri.eias.ru, ЕТК + Екатеринбургэнерго). Источник ГЕО- # БЛОКИРУЕТ non-RU IP + российский УЦ → RUN-ON-PROD. 06:00 МСК — через 30 мин diff --git a/backend/app/workers/tasks/connection_capacity_sync.py b/backend/app/workers/tasks/connection_capacity_sync.py index 336fed7e..7d0a6b1d 100644 --- a/backend/app/workers/tasks/connection_capacity_sync.py +++ b/backend/app/workers/tasks/connection_capacity_sync.py @@ -145,6 +145,32 @@ def sync_eesk_reserves() -> dict: raise +@celery_app.task( + name="tasks.connection_capacity_sync.sync_osm_tp_geocode", + queue="celery", +) +def sync_osm_tp_geocode() -> dict: + """OSM-геокод городских ТП/РП ЕКБ → geom для power_tp_rp_reserves. #2119 B2 (финал). + + RUN-ON-PROD рядом с ЕЭСК (тот же вторник-кластер). Матч name_norm (чистый номер) + ↔ OSM substation ref/name → UPDATE geom + geocode_status='osm'. Идемпотентно + (geom IS NULL-гейт): не-новые точки просто пере-скипаются без эффекта. apply=True. + Источник (Overpass) без гео-блока, но держим на проде в едином weekly-кластере. + """ + _log_breadcrumb("osm_tp_geocode_sync", "info", "task started") + try: + from app.services.site_finder.osm_tp_geocoder import geocode_tp_from_osm + + result = geocode_tp_from_osm(apply=True) + logger.info("sync_osm_tp_geocode done: %s", result) + _log_breadcrumb("osm_tp_geocode_sync", "info", f"done: {result}") + return result + except Exception as e: + logger.exception("sync_osm_tp_geocode failed: %s", e) + _log_breadcrumb("osm_tp_geocode_sync", "error", f"{type(e).__name__}: {e}") + raise + + @celery_app.task( name="tasks.connection_capacity_sync.sync_gas_outlet_points", queue="celery", diff --git a/backend/tests/test_osm_tp_geocoder.py b/backend/tests/test_osm_tp_geocoder.py new file mode 100644 index 00000000..60b6f41e --- /dev/null +++ b/backend/tests/test_osm_tp_geocoder.py @@ -0,0 +1,286 @@ +"""Unit-тесты OSM-геокода городских ТП/РП ЕКБ (#2119 B2, финал). + +Pure / mock-based — без реальной сети и БД. Покрывает: +- extract_tp_number: «РП-101»→101, «ТП 1591»→1591, голый ref='101'→101, + «110 кВ» БЕЗ ТП-токена/ref→None, traction-скип на уровне fetch. +- fetch_osm_substations: скип substation=traction (mock httpx.post). +- _index_osm_by_number: дубль-номер (разные координаты) → ambiguous, не матчится; + дубль-теги одной точки (way+node, те же координаты) → НЕ ambiguous. +- geocode_tp_from_osm: dry-run НЕ пишет; apply=True шлёт UPDATE с CAST-формами + и geom IS NULL-гейтом; sanity-bbox отсекает координату вне ЕКБ. +- Celery-регистрация таска sync_osm_tp_geocode + beat-строка. +""" + +from __future__ import annotations + +from contextlib import contextmanager +from typing import Any + +from app.services.site_finder import osm_tp_geocoder as og + +# ── extract_tp_number ──────────────────────────────────────────────────────── + + +def test_extract_tp_number_rp_token() -> None: + """«РП-101» в name → 101 (ТП-токен + число).""" + assert og.extract_tp_number({"name": "РП-101"}) == "101" + + +def test_extract_tp_number_tp_hyphen() -> None: + """«ТП-180» → 180.""" + assert og.extract_tp_number({"name": "ТП-180"}) == "180" + + +def test_extract_tp_number_tp_space() -> None: + """«ТП 1591» (пробел-разделитель) → 1591.""" + assert og.extract_tp_number({"name": "ТП 1591"}) == "1591" + + +def test_extract_tp_number_bare_ref() -> None: + """Голый ref='101' (без ТП-токена) → 101 — оператор проставил ref явно.""" + assert og.extract_tp_number({"ref": "101"}) == "101" + + +def test_extract_tp_number_ref_priority_over_name() -> None: + """ref приоритетнее name: ref='42', name='ТП-999' → 42.""" + assert og.extract_tp_number({"ref": "42", "name": "ТП-999"}) == "42" + + +def test_extract_tp_number_voltage_without_token_is_none() -> None: + """«110 кВ» БЕЗ ТП-токена и без ref → None (это класс напряжения, не номер).""" + assert og.extract_tp_number({"name": "Подстанция 110 кВ"}) is None + assert og.extract_tp_number({"name": "110 кВ", "voltage": "110000"}) is None + + +def test_extract_tp_number_empty_tags_is_none() -> None: + """Нет ни ref, ни name → None.""" + assert og.extract_tp_number({}) is None + + +def test_extract_tp_number_latin_tp_token() -> None: + """Латинский токен «TP-42» → 42.""" + assert og.extract_tp_number({"name": "TP-42"}) == "42" + + +# ── fetch_osm_substations: traction-скип ───────────────────────────────────── + + +class _FakeResp: + def __init__(self, payload: dict) -> None: + self._payload = payload + + def raise_for_status(self) -> None: + pass + + def json(self) -> dict: + return self._payload + + +def test_fetch_skips_traction(monkeypatch: Any) -> None: + """substation=traction (тяговые ж/д) отсеиваются; обычные ТП остаются.""" + payload = { + "elements": [ + {"type": "node", "id": 1, "lat": 56.8, "lon": 60.6, "tags": {"name": "ТП-1"}}, + { + "type": "node", + "id": 2, + "lat": 56.8, + "lon": 60.6, + "tags": {"substation": "traction", "name": "Тяговая-2"}, + }, + ] + } + + def _fake_post(url: str, data: Any = None, timeout: Any = None) -> _FakeResp: + return _FakeResp(payload) + + monkeypatch.setattr(og.httpx, "post", _fake_post) + kept = og.fetch_osm_substations() + assert len(kept) == 1 + assert kept[0]["id"] == 1 + + +# ── _index_osm_by_number: ambiguous ────────────────────────────────────────── + + +def _node(el_id: int, lat: float, lon: float, tags: dict) -> dict: + return {"type": "node", "id": el_id, "lat": lat, "lon": lon, "tags": tags} + + +def test_index_ambiguous_duplicate_number() -> None: + """Один номер несут 2 фичи с РАЗНЫМИ координатами → ambiguous, из матча выпадает.""" + elements = [ + _node(1, 56.80, 60.60, {"name": "ТП-100"}), + _node(2, 56.85, 60.65, {"name": "ТП-100"}), # тот же номер, другая точка + _node(3, 56.82, 60.62, {"name": "ТП-200"}), # уникальный + ] + by_number, ambiguous = og._index_osm_by_number(elements) + assert "100" in ambiguous + assert "100" not in by_number + assert "200" in by_number + + +def test_index_same_point_not_ambiguous() -> None: + """Дубль-теги ОДНОЙ точки (way+node на тех же координатах) → НЕ ambiguous.""" + elements = [ + _node(1, 56.80, 60.60, {"name": "ТП-100"}), + {"type": "way", "id": 2, "center": {"lat": 56.80, "lon": 60.60}, "tags": {"ref": "100"}}, + ] + by_number, ambiguous = og._index_osm_by_number(elements) + assert "100" not in ambiguous + assert by_number["100"] == (56.80, 60.60) + + +def test_index_out_of_bbox_skipped() -> None: + """Координата вне bbox ЕКБ (sanity-гейт) → не индексируется.""" + elements = [_node(1, 55.75, 37.61, {"name": "ТП-300"})] # Москва + by_number, ambiguous = og._index_osm_by_number(elements) + assert by_number == {} + assert ambiguous == set() + + +def test_index_way_center_coords() -> None: + """way без lat/lon, но с center → берём center.""" + elements = [ + {"type": "way", "id": 9, "center": {"lat": 56.83, "lon": 60.63}, "tags": {"ref": "555"}} + ] + by_number, _ = og._index_osm_by_number(elements) + assert by_number["555"] == (56.83, 60.63) + + +# ── Fake DB session (execute + begin_nested), как в test_eesk_reserve_loader ── + + +class _FakeResult: + def __init__(self, rows: list[Any] | None = None, rowcount: int = 1) -> None: + self._rows = rows or [] + self.rowcount = rowcount + + def all(self) -> list[Any]: + return self._rows + + +class _FakeSession: + """Мини-заглушка Session: собирает execute-вызовы, поддерживает begin_nested.""" + + def __init__(self, candidate_norms: list[str], rowcount: int = 1) -> None: + self.calls: list[tuple[str, Any]] = [] + self._candidates = [(n,) for n in candidate_norms] + self._rowcount = rowcount + self.committed = False + + @contextmanager + def begin_nested(self): # type: ignore[no-untyped-def] + yield + + def execute(self, stmt: Any, params: Any = None) -> _FakeResult: + sql = str(stmt) + self.calls.append((sql, params)) + if "SELECT" in sql and "power_tp_rp_reserves" in sql: + return _FakeResult(rows=self._candidates) + return _FakeResult(rowcount=self._rowcount) + + def commit(self) -> None: + self.committed = True + + def rollback(self) -> None: + pass + + +def _patch_fetch(monkeypatch: Any, elements: list[dict]) -> None: + monkeypatch.setattr(og, "fetch_osm_substations", lambda: elements) + + +# ── geocode_tp_from_osm ────────────────────────────────────────────────────── + + +def test_geocode_dry_run_does_not_write(monkeypatch: Any) -> None: + """Dry-run (apply=False): matched считается, но UPDATE не шлётся и commit не зовётся.""" + _patch_fetch(monkeypatch, [_node(1, 56.80, 60.60, {"name": "ТП-101"})]) + db = _FakeSession(candidate_norms=["101"]) + result = og.geocode_tp_from_osm(db=db, apply=False) + + assert result["matched"] == 1 + assert result["updated"] == 0 + assert result["applied"] is False + assert not db.committed + # Только SELECT-кандидатов, никаких UPDATE. + assert all("UPDATE" not in sql for sql, _ in db.calls) + + +def test_geocode_apply_updates_with_cast(monkeypatch: Any) -> None: + """apply=True: UPDATE шлётся с CAST-формами, geom IS NULL-гейтом, commit зовётся.""" + _patch_fetch(monkeypatch, [_node(1, 56.80, 60.60, {"name": "ТП-101"})]) + db = _FakeSession(candidate_norms=["101"], rowcount=1) + result = og.geocode_tp_from_osm(db=db, apply=True) + + assert result["matched"] == 1 + assert result["updated"] == 1 + assert result["applied"] is True + assert db.committed + + update_calls = [(sql, p) for sql, p in db.calls if "UPDATE" in sql] + assert len(update_calls) == 1 + sql, params = update_calls[0] + # CAST-формы (psycopg3-канон), НИКОГДА :x::type. + assert "CAST(:lon AS float)" in sql + assert "CAST(:lat AS float)" in sql + assert "CAST(:name_norm AS text)" in sql + assert "::float" not in sql and "::text" not in sql + # Идемпотентность + скоуп ЕКБ. + assert "geom IS NULL" in sql + assert "geocode_status = 'osm'" in sql + assert "municipality = 'Екатеринбург'" in sql + assert params == {"name_norm": "101", "lat": 56.80, "lon": 60.60} + + +def test_geocode_ambiguous_not_matched(monkeypatch: Any) -> None: + """Дубль-номер (ambiguous) не матчится: candidate '100' есть, но geom не пишется.""" + _patch_fetch( + monkeypatch, + [ + _node(1, 56.80, 60.60, {"name": "ТП-100"}), + _node(2, 56.85, 60.65, {"name": "ТП-100"}), # коллизия + ], + ) + db = _FakeSession(candidate_norms=["100"], rowcount=1) + result = og.geocode_tp_from_osm(db=db, apply=True) + + assert result["ambiguous"] == 1 + assert result["matched"] == 0 + assert result["updated"] == 0 + assert all("UPDATE" not in sql for sql, _ in db.calls) + + +def test_geocode_unmatched_candidate_skipped(monkeypatch: Any) -> None: + """Кандидат без OSM-номера просто скипается (matched=0), без падения.""" + _patch_fetch(monkeypatch, [_node(1, 56.80, 60.60, {"name": "ТП-777"})]) + db = _FakeSession(candidate_norms=["101"], rowcount=1) # 101 нет в OSM + result = og.geocode_tp_from_osm(db=db, apply=True) + assert result["matched"] == 0 + assert result["updated"] == 0 + + +# ── Celery / beat регистрация ──────────────────────────────────────────────── + + +def test_celery_task_registered() -> None: + """Таск sync_osm_tp_geocode регистрируется при импорте модуля тасков. + + Модуль в celery_app.include — на worker-старте импортируется и таск попадает в + celery_app.tasks. В тесте импорт делаем явно (Celery не бутит воркер). + """ + import app.workers.tasks.connection_capacity_sync # noqa: F401 — импорт регистрирует таск + from app.workers.celery_app import celery_app + + assert "tasks.connection_capacity_sync.sync_osm_tp_geocode" in celery_app.tasks + + +def test_beat_schedule_entry() -> None: + """Beat содержит еженедельную вторник-05:45 запись osm-tp-geocode.""" + from app.workers.beat_schedule import build_beat_schedule + + schedule = build_beat_schedule() + assert "osm-tp-geocode-weekly" in schedule + entry = schedule["osm-tp-geocode-weekly"] + assert entry["task"] == "tasks.connection_capacity_sync.sync_osm_tp_geocode" -- 2.45.3