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_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