gendesign/backend/app/workers/tasks/connection_capacity_sync.py

"""Celery tasks: еженедельная синхронизация точек подключения со свободной
мощностью — электро (Россети WFS + xlsx-резервы) и вода (Водоканал docx). #2119.

Источники ГЕО-БЛОКИРУЮТ non-RU IP → таски выполняются НА ПРОДЕ (Celery worker
на VPS). Persistent breadcrumbs в nspd_geo_log (как utility/poi/noise sync).
Import-light: тяжёлые загрузчики импортируются лениво внутри тасков.
"""

import logging

from sqlalchemy import text

from app.workers.celery_app import celery_app

logger = logging.getLogger(__name__)


def _log_breadcrumb(stage: str, level: str, message: str) -> None:
    """Записать строку в nspd_geo_log (job_id=NULL) — диагностика без SSH."""
    try:
        from app.core.db import SessionLocal

        db = SessionLocal()
        try:
            db.execute(
                text(
                    "INSERT INTO nspd_geo_log (job_id, level, stage, message) "
                    "VALUES (NULL, :lvl, :stage, :msg)"
                ),
                {"lvl": level, "stage": stage, "msg": message[:500]},
            )
            db.commit()
        finally:
            db.close()
    except Exception as e:
        logger.warning("connection_capacity breadcrumb failed: %s", e)


@celery_app.task(
    name="tasks.connection_capacity_sync.sync_power_supply_centers",
    queue="celery",
)
def sync_power_supply_centers() -> dict:
    """WFS центры питания + xlsx-резервы (ЦП + ТП/РП) → power_supply_centers / ТП-РП.

    RUN-ON-PROD (гео-блок источников). Сначала WFS (координаты + напряжение +
    индекс загрузки), затем резервы из xlsx-раскрытия. Каждый шаг graceful —
    сбой резервов не откатывает загруженные WFS-точки.
    """
    _log_breadcrumb("power_capacity_sync", "info", "task started")
    result: dict = {}
    try:
        from app.services.site_finder.rosseti_wfs_loader import load_power_supply_centers

        result["wfs"] = load_power_supply_centers()
    except Exception as e:
        logger.exception("sync_power_supply_centers: WFS failed: %s", e)
        _log_breadcrumb("power_capacity_sync", "error", f"wfs {type(e).__name__}: {e}")
        result["wfs"] = {"error": str(e)}

    try:
        from app.services.site_finder.rosseti_reserve_loader import load_all_reserves

        result["reserves"] = load_all_reserves()
    except Exception as e:
        logger.exception("sync_power_supply_centers: reserves failed: %s", e)
        _log_breadcrumb("power_capacity_sync", "error", f"reserves {type(e).__name__}: {e}")
        result["reserves"] = {"error": str(e)}

    logger.info("sync_power_supply_centers done: %s", result)
    _log_breadcrumb("power_capacity_sync", "info", f"done: {result}")
    return result


@celery_app.task(
    name="tasks.connection_capacity_sync.sync_water_reserves",
    queue="celery",
)
def sync_water_reserves() -> dict:
    """Резервы ЦСВ/ЦСК Водоканала ЕКБ (docx) → water_supply_reserves.

    RUN-ON-PROD (гео-блок). Обе страницы (водоснабжение/водоотведение) graceful.
    """
    _log_breadcrumb("water_capacity_sync", "info", "task started")
    try:
        from app.services.site_finder.vodokanal_reserve_loader import load_water_reserves

        result = load_water_reserves()
        logger.info("sync_water_reserves done: %s", result)
        _log_breadcrumb("water_capacity_sync", "info", f"done: {result}")
        return result
    except Exception as e:
        logger.exception("sync_water_reserves failed: %s", e)
        _log_breadcrumb("water_capacity_sync", "error", f"{type(e).__name__}: {e}")
        raise


@celery_app.task(
    name="tasks.connection_capacity_sync.sync_gas_grs_capacity",
    queue="celery",
)
def sync_gas_grs_capacity() -> dict:
    """Свободная мощность ГРС «Газпром трансгаз Екатеринбург» (HTML) → gas_grs_capacity.

    RUN-ON-PROD (гео-блок источника). Инлайн HTML-таблица раскрытия ГТЕ → UPSERT
    свободной мощности ГРС (по выходам). #2119 Фаза B1.
    """
    _log_breadcrumb("gas_grs_sync", "info", "task started")
    try:
        from app.services.site_finder.gazprom_grs_loader import load_grs_capacity

        result = load_grs_capacity()
        logger.info("sync_gas_grs_capacity done: %s", result)
        _log_breadcrumb("gas_grs_sync", "info", f"done: {result}")
        return result
    except Exception as e:
        logger.exception("sync_gas_grs_capacity failed: %s", e)
        _log_breadcrumb("gas_grs_sync", "error", f"{type(e).__name__}: {e}")
        raise


@celery_app.task(
    name="tasks.connection_capacity_sync.sync_eesk_reserves",
    queue="celery",
)
def sync_eesk_reserves() -> dict:
    """Резервы АО «ЕЭСК» (xlsx-раскрытие) → power_supply_centers (ПС 35-220, МВт) +
    power_tp_rp_reserves (ТП/РП «ниже 35»). #2119 Фаза B2.

    RUN-ON-PROD рядом с rosseti/gas/vodokanal (единый weekly-кластер). discovery
    свежайшего xlsx + загрузка ПС и ТП/РП — каждый файл graceful (сбой одного не
    валит другой; лоадер сам изолирует per-row SAVEPOINT-ами).
    """
    _log_breadcrumb("eesk_reserves_sync", "info", "task started")
    try:
        from app.services.site_finder.eesk_reserve_loader import load_eesk_reserves

        result = load_eesk_reserves()
        logger.info("sync_eesk_reserves done: %s", result)
        _log_breadcrumb("eesk_reserves_sync", "info", f"done: {result}")
        return result
    except Exception as e:
        logger.exception("sync_eesk_reserves failed: %s", e)
        _log_breadcrumb("eesk_reserves_sync", "error", f"{type(e).__name__}: {e}")
        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",
)
def sync_gas_outlet_points() -> dict:
    """Точки выхода ГРС АО «Екатеринбурггаз» (форма 6 ФАС №960) → gas_grs_outlet_points,
    затем догеокод новых адресных строк → geom. #2119 Фаза B2, PR-4.

    RUN-ON-PROD (гео-блок источника ekgas.ru + DaData-ключи на проде). Два шага:
      1. load_gas_outlet_points() — свежайшая форма 6 из каталога → UPSERT (идемпотентно:
         помесячный UPSERT по (grs_name_norm, outlet_name, period_month, is_forecast),
         geom НЕ затирается в DO UPDATE);
      2. geocode_gas_outlets(apply=True) — новые адресные строки последнего period_month
         с geom IS NULL догеокодятся через DaData (in-memory кэш + cap внутри лоадера;
         уже геокоженные строки прошлых месяцев не трогаются).

    Источник обновляется ~5-го числа месяца, но расписание еженедельное (вторник) —
    не-новый период просто пере-UPSERT-ится без эффекта (geom сохраняется). Геокод
    graceful: сбой шага 2 не откатывает загруженные строки (шаг 1 уже закоммичен своей
    сессией). Каждый шаг логируется breadcrumb-ом в nspd_geo_log.
    """
    _log_breadcrumb("gas_outlet_sync", "info", "task started")
    result: dict = {}
    try:
        from app.services.site_finder.ekgas_outlet_loader import load_gas_outlet_points

        result["load"] = load_gas_outlet_points()
    except Exception as e:
        logger.exception("sync_gas_outlet_points: load failed: %s", e)
        _log_breadcrumb("gas_outlet_sync", "error", f"load {type(e).__name__}: {e}")
        result["load"] = {"error": str(e)}

    try:
        from app.services.site_finder.ekgas_outlet_loader import geocode_gas_outlets

        result["geocode"] = geocode_gas_outlets(apply=True)
    except Exception as e:
        logger.exception("sync_gas_outlet_points: geocode failed: %s", e)
        _log_breadcrumb("gas_outlet_sync", "error", f"geocode {type(e).__name__}: {e}")
        result["geocode"] = {"error": str(e)}

    logger.info("sync_gas_outlet_points done: %s", result)
    _log_breadcrumb("gas_outlet_sync", "info", f"done: {result}")
    return result


@celery_app.task(
    name="tasks.connection_capacity_sync.sync_heat_reserves",
    queue="celery",
)
def sync_heat_reserves() -> dict:
    """Резервы тепловой мощности ФАС (ri.eias.ru) → heat_system_reserves. #2119 B2.

    RUN-ON-PROD (гео-блок non-RU + российский УЦ). По всем организациям-держателям
    (ЕТК, Екатеринбургэнерго): карточка ФАС → публикации свежайшего квартала →
    выгрузки xlsx → UPSERT. Каждая организация graceful (падение одной не валит вторую).
    """
    _log_breadcrumb("heat_reserves_sync", "info", "task started")
    try:
        from app.services.site_finder.eias_heat_loader import load_heat_reserves

        result = load_heat_reserves()
        logger.info("sync_heat_reserves done: %s", result)
        _log_breadcrumb("heat_reserves_sync", "info", f"done: {result}")
        return result
    except Exception as e:
        logger.exception("sync_heat_reserves failed: %s", e)
        _log_breadcrumb("heat_reserves_sync", "error", f"{type(e).__name__}: {e}")
        raise