355 lines
17 KiB
Python
355 lines
17 KiB
Python
"""Загрузчик свободной мощности ГРС «Газпром трансгаз Екатеринбург» (#2119 B1).
|
||
|
||
No-B2B верифицированный источник: страница раскрытия ГТЕ содержит ИНЛАЙН
|
||
HTML-таблицу «Информация о технической возможности…» (~316 строк) со свободной
|
||
мощностью газораспределительных станций (ГРС) — проектная / фактическая загрузка /
|
||
свободная мощность (тыс. м³/ч) + срок расшивки. UPSERT-ит в ``gas_grs_capacity``.
|
||
|
||
Источник ГЕО-БЛОКИРУЕТ non-RU IP → загрузчик РАБОТАЕТ НА ПРОДЕ (Celery weekly /
|
||
manual docker exec) — как rosseti-/vodokanal-лоадеры. Код не предполагает
|
||
локальный сетевой доступ.
|
||
|
||
Парсинг DEFENSIVE: колонки мапятся ПО ЗАГОЛОВКАМ (не по фикс-позиции — ГТЕ может
|
||
переставить/добавить столбцы), имя ГРС с rowspan («выход 1/2/3») forward-fill'ится
|
||
на строки-выходы (идея из vodokanal_reserve_loader.parse_docx_table_rows). Числа —
|
||
через ОБЩИЙ ``parse_reserve_number`` из rosseti_reserve_loader (запятая, «н/д»,
|
||
знак). Неизвестные колонки не теряются — падают в ``raw`` jsonb.
|
||
"""
|
||
|
||
import json
|
||
import logging
|
||
import re
|
||
|
||
import httpx
|
||
from bs4 import BeautifulSoup
|
||
from sqlalchemy import text
|
||
from sqlalchemy.orm import Session
|
||
|
||
from app.core.db import SessionLocal
|
||
from app.services.site_finder.rosseti_reserve_loader import parse_reserve_number
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
# Страница расклытия «Газпром трансгаз Екатеринбург» с инлайн HTML-таблицей.
|
||
# Punycode не требуется — латинский хост.
|
||
DISCLOSURE_URL = "https://ekaterinburg-tr.gazprom.ru/about/informatsiya-o-tekhnicheskoj-vozmo/"
|
||
|
||
_HTTP_TIMEOUT = 60
|
||
|
||
_MULTISPACE_RE = re.compile(r"\s+")
|
||
|
||
# «н/д» / «-» / прочерк / пусто в текстовой ячейке (напр. срок расшивки) → None.
|
||
_NODATA_TEXT_TOKENS = frozenset(
|
||
{"", "н/д", "нд", "n/a", "na", "-", "—", "–", "нет данных", "х", "x"}
|
||
)
|
||
|
||
# Заголовок-матчинг: подстрока в тексте <th>/первой строки → логическая колонка.
|
||
# Порядок важен — «свободн» проверяем раньше «загрузк», а «%» отдельным ключом.
|
||
# Каждая запись: (набор подстрок-триггеров, ключ). Первый матч на колонку выигрывает.
|
||
_HEADER_PATTERNS: list[tuple[tuple[str, ...], str]] = [
|
||
(("выход",), "output_name"),
|
||
(("наименование", "грс", "станц"), "grs_name"),
|
||
(("проектн",), "design"),
|
||
(("свободн", "резерв"), "free"),
|
||
(("загрузк", "фактич", "текущ"), "load"),
|
||
(("срок", "расшивк", "мероприят"), "upgrade_due"),
|
||
]
|
||
|
||
# Region-колонка (если ГТЕ добавит регион в таблицу) → фильтр Свердловской области.
|
||
_REGION_HEADER_TOKENS = ("регион", "область", "субъект")
|
||
_SVERDLOVSK_TOKEN = "свердлов"
|
||
|
||
|
||
def normalize_grs_name(name: str | None) -> str:
|
||
"""Нормализует имя ГРС для матча/дедупа.
|
||
|
||
Правила: lower, ё→е, схлопывание пробелов и дефисов (все виды тире → одиночный
|
||
дефис без пробелов вокруг), удаление кавычек. Префиксы «ГРС»/«АРП» СОХРАНЯЮТСЯ —
|
||
они различают станции. Пусто → ''.
|
||
|
||
Examples:
|
||
«ГРС Кольцово» → «грс кольцово»
|
||
«ГРС Берёзовский» → «грс березовский»
|
||
«ГРС Северная — 2» → «грс северная-2»
|
||
"""
|
||
if not name:
|
||
return ""
|
||
s = name.strip().lower().replace("ё", "е")
|
||
s = s.replace("«", "").replace("»", "").replace('"', "").replace("'", "")
|
||
# Все виды тире (—, –, -) в одиночный дефис, пробелы вокруг убираем.
|
||
s = re.sub(r"\s*[—–-]\s*", "-", s)
|
||
s = _MULTISPACE_RE.sub(" ", s).strip()
|
||
return s
|
||
|
||
|
||
def _clean_text(value: object | None) -> str | None:
|
||
"""Текст ячейки → strip + collapse пробелов. Пусто / «н/д»-токен → None."""
|
||
if value is None:
|
||
return None
|
||
s = _MULTISPACE_RE.sub(" ", str(value).replace("\xa0", " ")).strip()
|
||
if not s or s.lower() in _NODATA_TEXT_TOKENS:
|
||
return None
|
||
return s
|
||
|
||
|
||
def _classify_header(header_text: str) -> str | None:
|
||
"""Заголовок колонки → логический ключ (grs_name/output_name/design/…) или None."""
|
||
low = header_text.lower()
|
||
for tokens, key in _HEADER_PATTERNS:
|
||
if any(tok in low for tok in tokens):
|
||
return key
|
||
return None
|
||
|
||
|
||
def _is_pct_header(header_text: str) -> bool:
|
||
"""Колонка про проценты свободной мощности?"""
|
||
low = header_text.lower()
|
||
return "%" in low or "процент" in low
|
||
|
||
|
||
def _is_region_header(header_text: str) -> bool:
|
||
"""Колонка про регион/область (для Свердловск-фильтра)?"""
|
||
low = header_text.lower()
|
||
return any(tok in low for tok in _REGION_HEADER_TOKENS)
|
||
|
||
|
||
def _extract_table(html: str) -> object | None:
|
||
"""Находит наиболее «содержательную» <table> (с макс. числом строк). None — нет."""
|
||
soup = BeautifulSoup(html, "html.parser")
|
||
tables = soup.find_all("table")
|
||
if not tables:
|
||
return None
|
||
# Основная таблица — с наибольшим числом строк (шапки/навигация мелкие).
|
||
return max(tables, key=lambda t: len(t.find_all("tr")))
|
||
|
||
|
||
def _row_cells(tr: object) -> list[object]:
|
||
"""Ячейки строки (<th> и <td> в порядке появления)."""
|
||
return tr.find_all(["th", "td"]) # type: ignore[union-attr]
|
||
|
||
|
||
def _build_column_map(header_cells: list[object]) -> tuple[dict[int, str], dict[int, str]]:
|
||
"""Из ячеек строки-заголовка строит col_idx→logical_key + col_idx→raw_header.
|
||
|
||
Возвращает (mapping, headers): mapping только для распознанных колонок
|
||
(grs_name/output_name/design/load/free/pct/upgrade_due/region); headers — сырые
|
||
тексты ВСЕХ колонок (для raw + region-детекта). «%»-колонка → ключ 'pct'.
|
||
"""
|
||
mapping: dict[int, str] = {}
|
||
headers: dict[int, str] = {}
|
||
for idx, cell in enumerate(header_cells):
|
||
htext = _clean_text(cell.get_text(" ")) or "" # type: ignore[union-attr]
|
||
headers[idx] = htext
|
||
if _is_region_header(htext):
|
||
mapping[idx] = "region"
|
||
continue
|
||
if _is_pct_header(htext):
|
||
mapping[idx] = "pct"
|
||
continue
|
||
key = _classify_header(htext)
|
||
if key is not None:
|
||
mapping[idx] = key
|
||
return mapping, headers
|
||
|
||
|
||
def parse_grs_table(html: str) -> list[dict]:
|
||
"""Парсит инлайн HTML-таблицу ГТЕ → список записей ГРС (по выходам).
|
||
|
||
DEFENSIVE:
|
||
- основная <table> = с макс. числом строк;
|
||
- первая строка = заголовок, колонки мапятся ПО ЗАГОЛОВКАМ (contains);
|
||
- rowspan имени ГРС → forward-fill на строки-выходы (несколько выходов у ГРС).
|
||
ДОПУЩЕНИЕ: offset-коррекция работает только когда колонка имени — ПЕРВАЯ
|
||
(в текущей таблице ГТЕ так); rowspan не-первой колонки даст пропуск строк
|
||
(не порчу данных — безымянные строки скипаются, raw сохраняется);
|
||
- числа через parse_reserve_number (запятая, «н/д», знак);
|
||
- неизвестные колонки → raw (не теряются).
|
||
|
||
Каждая запись:
|
||
{grs_name, output_name, region, design_capacity_th_m3_h, current_load_th_m3_h,
|
||
free_capacity_th_m3_h, free_capacity_pct, upgrade_due, raw}.
|
||
grs_name пустой (нет forward-fill и нет своего имени) → строка пропускается.
|
||
"""
|
||
table = _extract_table(html)
|
||
if table is None:
|
||
logger.warning("gazprom_grs: <table> не найдена в HTML")
|
||
return []
|
||
|
||
rows = table.find_all("tr") # type: ignore[union-attr]
|
||
if not rows:
|
||
return []
|
||
|
||
mapping, headers = _build_column_map(_row_cells(rows[0]))
|
||
if "grs_name" not in mapping.values():
|
||
logger.warning("gazprom_grs: колонка имени ГРС не распознана по заголовкам (%s)", headers)
|
||
return []
|
||
|
||
records: list[dict] = []
|
||
# forward-fill имени ГРС по rowspan: если ячейка имени отсутствует в строке
|
||
# (schrunk из-за rowspan предыдущей), наследуем последнее известное имя.
|
||
last_grs_name: str | None = None
|
||
|
||
for tr in rows[1:]:
|
||
cells = _row_cells(tr)
|
||
if not cells:
|
||
continue
|
||
|
||
# Матчим ячейки к логическим ключам по позиции колонки-заголовка. При
|
||
# rowspan имени в строке-выходе ячейки имени НЕТ → индексы сдвигаются; мы
|
||
# определяем это по числу ячеек < числу заголовков и forward-fill'им имя.
|
||
values: dict[str, str | None] = {}
|
||
raw_cells: list[str] = []
|
||
|
||
name_col = next(i for i, k in mapping.items() if k == "grs_name")
|
||
shifted = len(cells) < len(headers) and name_col == 0
|
||
# Если строка короче заголовков и имя-колонка первая — считаем что
|
||
# rowspan «съел» первую ячейку: логические индексы сдвигаем на +1.
|
||
offset = 1 if shifted else 0
|
||
|
||
for idx, cell in enumerate(cells):
|
||
# raw_cells — сырой текст ячейки (в т.ч. «н/д»), чтобы raw jsonb не терял
|
||
# исходное значение; values — очищенный (_clean_text: «н/д» → None).
|
||
raw_text = _MULTISPACE_RE.sub(
|
||
" ",
|
||
cell.get_text(" ").replace("\xa0", " "), # type: ignore[union-attr]
|
||
).strip()
|
||
raw_cells.append(raw_text)
|
||
logical_idx = idx + offset
|
||
key = mapping.get(logical_idx)
|
||
if key is not None:
|
||
values[key] = _clean_text(raw_text)
|
||
|
||
# values уже очищены (_clean_text применён при заполнении: str | None).
|
||
grs_name = values.get("grs_name")
|
||
if grs_name:
|
||
last_grs_name = grs_name # type: ignore[assignment]
|
||
else:
|
||
grs_name = last_grs_name # forward-fill (rowspan имени ГРС)
|
||
if not grs_name:
|
||
continue # шапка/итоговая строка без имени и без forward-fill
|
||
|
||
rec = {
|
||
"grs_name": grs_name,
|
||
"output_name": values.get("output_name"),
|
||
"region": values.get("region"),
|
||
"design_capacity_th_m3_h": parse_reserve_number(values.get("design")),
|
||
"current_load_th_m3_h": parse_reserve_number(values.get("load")),
|
||
"free_capacity_th_m3_h": parse_reserve_number(values.get("free")),
|
||
"free_capacity_pct": parse_reserve_number(values.get("pct")),
|
||
"upgrade_due": values.get("upgrade_due"),
|
||
"raw": {"headers": list(headers.values()), "cells": raw_cells},
|
||
}
|
||
records.append(rec)
|
||
|
||
logger.info("gazprom_grs: распознано записей ГРС=%d", len(records))
|
||
return records
|
||
|
||
|
||
def _passes_region_filter(records: list[dict]) -> list[dict]:
|
||
"""Если хоть у одной записи есть region — фильтруем по «свердлов»; иначе всё.
|
||
|
||
В таблице ГТЕ обычно НЕТ колонки региона (весь ГТЕ = Свердловская + сопред.
|
||
области) → грузим всё с region=NULL. Если ГТЕ добавит регион — оставляем только
|
||
свердловские строки.
|
||
"""
|
||
if any(r.get("region") for r in records):
|
||
return [r for r in records if (r.get("region") or "").lower().find(_SVERDLOVSK_TOKEN) >= 0]
|
||
return records
|
||
|
||
|
||
def fetch_grs_html() -> str:
|
||
"""Тянет HTML страницы раскрытия ГТЕ. RUN-ON-PROD (гео-блок). httpx с таймаутом."""
|
||
resp = httpx.get(DISCLOSURE_URL, timeout=_HTTP_TIMEOUT, follow_redirects=True)
|
||
resp.raise_for_status()
|
||
return resp.text
|
||
|
||
|
||
def _upsert_grs_rows(db: Session, records: list[dict]) -> dict[str, int]:
|
||
"""UPSERT записей ГРС в gas_grs_capacity. Per-row SAVEPOINT (битая строка не валит батч).
|
||
|
||
ON CONFLICT (grs_name_norm, output_name) — NULLS NOT DISTINCT в схеме (output_name
|
||
nullable). Returns: счётчики rows/inserted/updated/skipped.
|
||
"""
|
||
inserted = 0
|
||
updated = 0
|
||
skipped = 0
|
||
for rec in records:
|
||
grs_name = rec["grs_name"]
|
||
params = {
|
||
"grs_name": grs_name,
|
||
"grs_name_norm": normalize_grs_name(grs_name),
|
||
"output_name": rec["output_name"],
|
||
"region": rec["region"],
|
||
"design": rec["design_capacity_th_m3_h"],
|
||
"load": rec["current_load_th_m3_h"],
|
||
"free": rec["free_capacity_th_m3_h"],
|
||
"pct": rec["free_capacity_pct"],
|
||
"upgrade_due": rec["upgrade_due"],
|
||
"raw": json.dumps(rec["raw"], ensure_ascii=False),
|
||
}
|
||
try:
|
||
with db.begin_nested(): # SAVEPOINT — откат только этой строки
|
||
result = db.execute(
|
||
text("""
|
||
INSERT INTO gas_grs_capacity
|
||
(grs_name, grs_name_norm, output_name, region,
|
||
design_capacity_th_m3_h, current_load_th_m3_h,
|
||
free_capacity_th_m3_h, free_capacity_pct,
|
||
upgrade_due, raw, fetched_at)
|
||
VALUES (
|
||
:grs_name, :grs_name_norm, :output_name, :region,
|
||
:design, :load, :free, :pct,
|
||
:upgrade_due, CAST(:raw AS jsonb), NOW()
|
||
)
|
||
ON CONFLICT (grs_name_norm, output_name) DO UPDATE
|
||
SET grs_name = EXCLUDED.grs_name,
|
||
region = EXCLUDED.region,
|
||
design_capacity_th_m3_h = EXCLUDED.design_capacity_th_m3_h,
|
||
current_load_th_m3_h = EXCLUDED.current_load_th_m3_h,
|
||
free_capacity_th_m3_h = EXCLUDED.free_capacity_th_m3_h,
|
||
free_capacity_pct = EXCLUDED.free_capacity_pct,
|
||
upgrade_due = EXCLUDED.upgrade_due,
|
||
raw = EXCLUDED.raw,
|
||
fetched_at = NOW()
|
||
RETURNING (xmax = 0) AS is_insert
|
||
"""),
|
||
params,
|
||
).scalar()
|
||
if result:
|
||
inserted += 1
|
||
else:
|
||
updated += 1
|
||
except Exception as e:
|
||
logger.warning("gas_grs upsert failed for %r/%r: %s", grs_name, rec["output_name"], e)
|
||
skipped += 1
|
||
return {"rows": len(records), "inserted": inserted, "updated": updated, "skipped": skipped}
|
||
|
||
|
||
def load_grs_capacity(db: Session | None = None) -> dict[str, int]:
|
||
"""Тянет страницу ГТЕ + парсит HTML-таблицу + UPSERT свободной мощности ГРС.
|
||
|
||
RUN-ON-PROD (гео-блок источника). ``db`` для совместимости сигнатуры; None →
|
||
своя SessionLocal (fetch + запись sync, как rosseti_wfs_loader).
|
||
|
||
Returns: счётчики {rows, inserted, updated, skipped}.
|
||
"""
|
||
html = fetch_grs_html()
|
||
records = _passes_region_filter(parse_grs_table(html))
|
||
|
||
owns_session = db is None
|
||
if db is None:
|
||
db = SessionLocal()
|
||
|
||
try:
|
||
counts = _upsert_grs_rows(db, records)
|
||
db.commit()
|
||
except Exception as e:
|
||
db.rollback()
|
||
logger.exception("load_grs_capacity: outer tx rolled back: %s", e)
|
||
raise
|
||
finally:
|
||
if owns_session:
|
||
db.close()
|
||
|
||
logger.info("load_grs_capacity done: %s", counts)
|
||
return counts
|