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())

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 мин

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",

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"