95 lines
3.9 KiB
Python
95 lines
3.9 KiB
Python
"""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
|