242 lines
11 KiB
Python
242 lines
11 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
|
||
|
||
|
||
@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
|