bot-backend
9e6176c486
All checks were successful
CI Trade-In / changes (pull_request) Successful in 7s
CI / changes (pull_request) Successful in 7s
CI Trade-In / backend-tests (pull_request) Has been skipped
CI Trade-In / frontend-checks (pull_request) Has been skipped
CI / frontend-tests (pull_request) Has been skipped
CI / openapi-codegen-check (pull_request) Successful in 1m51s
CI / backend-tests (pull_request) Successful in 14m5s
589 lines
32 KiB
Python
589 lines
32 KiB
Python
"""Загрузчик адресных точек выхода газораспределительной сети (#2119 Phase B2, PR-2).
|
||
|
||
Источник — раскрытие информации АО «Екатеринбурггаз» (ГРО), форма 6 приложения №4
|
||
к Приказу ФАС России №960 от 08.12.2022 «О порядке раскрытия информации…»: построчная
|
||
выгрузка «точек выхода» ГРС (~1037 строк/мес) — объём заявок, удовлетворено, свободная
|
||
мощность точки выхода (млн. м³/МЕСЯЦ — НЕ тыс. м³/ч как в gas_grs_capacity!). UPSERT-ит
|
||
в ``gas_grs_outlet_points``.
|
||
|
||
Источник ГЕО-БЛОКИРУЕТ non-RU IP → загрузчик РАБОТАЕТ НА ПРОДЕ (Celery monthly / manual
|
||
docker exec) — как gas_grs_capacity (B1, ГТЕ) и rosseti-/vodokanal-/eias-лоадеры. Код не
|
||
предполагает локальный сетевой доступ. ТОЛЬКО apex-домен ``ekgas.ru`` — ``www.ekgas.ru``
|
||
даёт ``SSLV3_ALERT_HANDSHAKE_FAILURE``. SSL через общий ``_ru_ssl_context`` из
|
||
gazprom_grs_loader (Russian Trusted CA Минцифры), httpx с явным таймаутом.
|
||
|
||
Данные лежат за 2 GET-переходами:
|
||
|
||
1. Каталог раскрытия (``/raskrytie-informacii/raskrytie-informatsii-1``, ~800KB HTML,
|
||
585 ссылок на файлы). Целевой — «Форма 6 приложения № 4 к Приказу ФАС России № 960
|
||
… (на|за) <месяц> <год>г..xlsx», путь ``/uploads/default/<hex>/<имя>``. Hex меняется
|
||
с каждым файлом → URL НЕ ЯКОРИМ: парсим каталог, вытаскиваем период «(за|на) <месяц>
|
||
<год>» + дату публикации (asof, DD.MM.YYYY рядом со ссылкой). Годовой «Форма №6
|
||
(Приложение №2) … план на <год>» (.xls) — СКИП (не адресный, xlrd в стеке нет).
|
||
|
||
Выбор свежайшего: приоритет ФАКТА над прогнозом. Берём максимальный (год, месяц)
|
||
среди «за» (is_forecast=False); если файлов «за» НЕТ вообще — максимальный «на»
|
||
(is_forecast=True, прогноз). ВНИМАНИЕ: даже если «на» новее по (год, месяц), чем
|
||
любой «за», всё равно берём «за» (факт важнее прогноза). См. select_latest_file.
|
||
|
||
2. XLSX (~91КБ, 1 лист): подписи в row 10 (1-indexed), нумерация 1..7 в row 11, данные
|
||
с row 12 (~1037 строк, последняя — сноска про «*», single-cell → скип). Колонки
|
||
(1-indexed): 1=«Точка входа» (ГРС, block-marker — заполнена ТОЛЬКО в первой строке
|
||
блока → forward-fill; 5 блоков), 2=«Точка выхода» (свободный текст: юрлицо/объект,
|
||
иногда адрес внутри), 3=тип (котельная/ТЭЦ/…), 4=группа потребления, 5=объём заявок,
|
||
6=удовлетворено, 7=свободная мощность — float ИЛИ текст-маркер «0.000*» (НЕ ноль! →
|
||
free_capacity_mln_m3=NULL, free_capacity_needs_calc=TRUE).
|
||
|
||
grs_name_norm нормализуется ТЕМ ЖЕ ``normalize_grs_name`` из gazprom_grs_loader (B1) —
|
||
иначе JOIN «точка выхода → свободная мощность ГРС» на gas_grs_capacity порвётся.
|
||
Числовой парс — общий ``parse_reserve_number`` из rosseti_reserve_loader (запятая, «—» →
|
||
None, знак сохраняется: дефицит −0.084 хранится знаком). Вся строка целиком → ``raw``
|
||
jsonb (будущие/неизвестные колонки не теряются).
|
||
"""
|
||
|
||
import io
|
||
import json
|
||
import logging
|
||
import re
|
||
from datetime import date
|
||
from urllib.parse import urljoin
|
||
|
||
import httpx
|
||
from bs4 import BeautifulSoup
|
||
from openpyxl import load_workbook
|
||
from sqlalchemy import text
|
||
from sqlalchemy.orm import Session
|
||
|
||
from app.core.db import SessionLocal
|
||
from app.services.site_finder.gazprom_grs_loader import _ru_ssl_context, normalize_grs_name
|
||
from app.services.site_finder.rosseti_reserve_loader import parse_reserve_number
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
# ТОЛЬКО apex-домен — www.ekgas.ru отдаёт SSLV3_ALERT_HANDSHAKE_FAILURE (разведка
|
||
# 2026-07-03). Каталог раскрытия информации со ссылками на помесячные формы 6.
|
||
BASE_URL = "https://ekgas.ru"
|
||
CATALOG_URL = "https://ekgas.ru/raskrytie-informacii/raskrytie-informatsii-1"
|
||
|
||
_HTTP_TIMEOUT = 60
|
||
|
||
# Целевой лист XLSX (единственный) — берём активный, если имя не совпало.
|
||
# 1-indexed строки листа: подписи колонок = row 10, нумерация 1..7 = row 11, данные с 12.
|
||
# Сетка НЕ хардкодится по номерам строк/колонок: живой файл за май 2026 имеет
|
||
# header на row 11 (не 10 из разведки) и колонку «№ п/п» слева (данные в col 2-8,
|
||
# сдвиг +1) — первый прод-прогон распарсил 5 строк вместо 1037. Урок воды/тепла:
|
||
# опорные строки и колонки ищем ПО ПОДПИСЯМ (_find_layout), скан первых N строк.
|
||
_LAYOUT_SCAN_ROWS = 25
|
||
|
||
# Колонки формы 6 (0-indexed для списка ячеек строки).
|
||
# Подписи колонок header-строки → внутренние ключи (substring-матч, lower).
|
||
_COLUMN_LABELS: dict[str, str] = {
|
||
"точка входа": "entry",
|
||
"точка выхода": "outlet",
|
||
"наименование потребителя": "consumer_type",
|
||
"номер группы": "group",
|
||
"поступивш": "requested",
|
||
"удовлетво": "approved",
|
||
"свободная мощность": "free",
|
||
}
|
||
|
||
|
||
def _find_layout(rows: list[list[object]]) -> tuple[int, dict[str, int]] | None:
|
||
"""Ищет header-строку по подписям в первых _LAYOUT_SCAN_ROWS строках.
|
||
|
||
Returns (data_start_idx, {key: col_idx}) — данные начинаются через строку
|
||
после header (между ними строка-нумерация «1..7»). None — сетка не распознана
|
||
(graceful: 0 записей + warning, не мусор — урок vodokanal).
|
||
"""
|
||
for i, row in enumerate(rows[:_LAYOUT_SCAN_ROWS]):
|
||
cols: dict[str, int] = {}
|
||
for j, cell in enumerate(row):
|
||
text = _clean_text(cell).lower()
|
||
if not text:
|
||
continue
|
||
for label, key in _COLUMN_LABELS.items():
|
||
if label in text and key not in cols:
|
||
cols[key] = j
|
||
if "entry" in cols and "outlet" in cols and "free" in cols:
|
||
return (i + 2, cols) # +1 строка-нумерация, +1 → первая строка данных
|
||
return None
|
||
|
||
|
||
# «0.000*» (и вариации с запятой/пробелами) в колонке свободной мощности = НЕ ноль, а
|
||
# «требуется гидравлический расчёт» → free_capacity_mln_m3=NULL, needs_calc=True.
|
||
# Маркер = число, за которым идёт «*» (звёздочка). Настоящий 0 без «*» парсится как 0.0.
|
||
_NEEDS_CALC_RE = re.compile(r"^\s*[\d.,\s\xa0]+\*\s*$")
|
||
|
||
# Русские месяцы → номер (родительный падеж как в «за январЯ»? нет — в именах файлов
|
||
# именительный: «за январь 2026»). Ключи в нижнем регистре.
|
||
_MONTHS: dict[str, int] = {
|
||
"январь": 1,
|
||
"февраль": 2,
|
||
"март": 3,
|
||
"апрель": 4,
|
||
"май": 5,
|
||
"июнь": 6,
|
||
"июль": 7,
|
||
"август": 8,
|
||
"сентябрь": 9,
|
||
"октябрь": 10,
|
||
"ноябрь": 11,
|
||
"декабрь": 12,
|
||
}
|
||
_MONTHS_ALT = "|".join(_MONTHS)
|
||
|
||
# Период в имени файла формы 6: «… (за|на) <месяц> <год>г..xlsx». «за» = факт
|
||
# (is_forecast=False), «на» = прогноз (is_forecast=True). Год — 4 цифры.
|
||
_PERIOD_RE = re.compile(rf"\b(за|на)\s+({_MONTHS_ALT})\s+(\d{{4}})", re.IGNORECASE)
|
||
|
||
# Целевой файл — форма 6 приложения №4 (адресная, помесячная, .xlsx). Годовой «Форма №6
|
||
# (Приложение №2) … план на <год>» (.xls) сюда НЕ подходит: другое приложение + .xls.
|
||
_FORM6_MARKER_RE = re.compile(r"форма\s*6.*приложени\w*\s*№?\s*4", re.IGNORECASE | re.DOTALL)
|
||
_XLSX_RE = re.compile(r"\.xlsx(?:$|\?)", re.IGNORECASE)
|
||
|
||
# Дата публикации рядом со ссылкой (DD.MM.YYYY) → asof.
|
||
_ASOF_RE = re.compile(r"\b(\d{1,2})\.(\d{1,2})\.(\d{4})\b")
|
||
|
||
# Адрес-экстрактор: улица-токен + номер дома. Город опционален. Консервативно —
|
||
# нет уверенного адреса → None (не выдумываем). Паттерны из реальных строк:
|
||
# «(г.Екатеринбург, Складской проезд,4а)» — в скобках;
|
||
# «, пер. Базовый, 17» — хвостом;
|
||
# «г.…, ул. К.Маркса,31».
|
||
# Улица-токен: ул|пер|пр|просп|проезд|б-р|ш|тракт|наб (с точкой или без, регистронезав.).
|
||
_STREET_TOKEN = r"(?:ул|пер|пр|просп|проспект|проезд|б-р|бульвар|ш|шоссе|тракт|наб)"
|
||
# Опциональный город впереди: «г.Екатеринбург,» / «г. Реж,».
|
||
_CITY_OPT = r"(?:г\.?\s*[А-ЯЁ][а-яё\-]+\s*,\s*)?"
|
||
# Название улицы: слова с кириллицей/точками/дефисами (напр. «К.Маркса», «Базовый»,
|
||
# «Складской проезд») — до номера дома. Номер: цифры + опц. буква/дробь («4а», «17»,
|
||
# «31/2»). Токен улицы может стоять ПЕРЕД названием («ул. Базовый») или ПОСЛЕ
|
||
# («Складской проезд»).
|
||
_ADDRESS_RE = re.compile(
|
||
rf"{_CITY_OPT}"
|
||
rf"(?:{_STREET_TOKEN}\.?\s*)?" # ведущий токен («ул. …», «пер. …») — опционален
|
||
r"[А-ЯЁ][А-Яа-яЁё.\- ]{1,40}?" # название улицы (капитализировано)
|
||
rf"(?:\s+{_STREET_TOKEN})?" # хвостовой токен («… проезд», «… тракт») — опционален
|
||
r"\s*,?\s*"
|
||
r"(?:д\.?\s*)?" # опц. «д.»
|
||
r"(\d+[а-яё]?(?:[/\-]\d+[а-яё]?)?)", # номер дома («4а», «17», «31/2»)
|
||
re.IGNORECASE,
|
||
)
|
||
# Чтобы отсечь ложные срабатывания на юрлицах («ИП Рошканяну Владимир Антонович») —
|
||
# требуем ЯВНЫЙ адресный сигнал: либо улица-токен, либо «г.<Город>» перед номером.
|
||
_HAS_STREET_TOKEN_RE = re.compile(rf"\b{_STREET_TOKEN}\b\.?", re.IGNORECASE)
|
||
_HAS_CITY_RE = re.compile(r"\bг\.?\s*[А-ЯЁ][а-яё\-]+", re.IGNORECASE)
|
||
|
||
_MULTISPACE_RE = re.compile(r"\s+")
|
||
|
||
|
||
def _clean_text(value: object) -> str:
|
||
"""Ячейка/текст → strip-строка со схлопнутыми пробелами (\\xa0→пробел). None → ''."""
|
||
if value is None:
|
||
return ""
|
||
return _MULTISPACE_RE.sub(" ", str(value).replace("\xa0", " ")).strip()
|
||
|
||
|
||
def _parse_period(name: str) -> tuple[int, int, bool] | None:
|
||
"""Имя файла → (year, month, is_forecast). «за»→факт(False), «на»→прогноз(True).
|
||
|
||
Нет распознанного «(за|на) <месяц> <год>» → None. Месяц вне словаря → None.
|
||
"""
|
||
m = _PERIOD_RE.search(name)
|
||
if not m:
|
||
return None
|
||
kind = m.group(1).lower()
|
||
month = _MONTHS.get(m.group(2).lower())
|
||
if month is None:
|
||
return None
|
||
year = int(m.group(3))
|
||
return (year, month, kind == "на")
|
||
|
||
|
||
def _is_form6_monthly(link_text: str, href: str) -> bool:
|
||
"""Ссылка — помесячная адресная форма 6 приложения №4 (.xlsx)?
|
||
|
||
Требуем: маркер «форма 6 … приложение № 4» (в тексте ссылки ИЛИ href), расширение
|
||
.xlsx и распознаваемый период. Годовой «Форма №6 (Приложение №2) … план на <год>»
|
||
(.xls) сюда НЕ проходит (другое приложение + не .xlsx).
|
||
"""
|
||
haystack = f"{link_text} {href}"
|
||
if not _FORM6_MARKER_RE.search(haystack):
|
||
return False
|
||
if not _XLSX_RE.search(href) and not _XLSX_RE.search(link_text):
|
||
return False
|
||
return _parse_period(haystack) is not None
|
||
|
||
|
||
def extract_form6_links(catalog_html: str) -> list[dict]:
|
||
"""Из HTML каталога → список кандидатов формы 6: [{url, period_month, is_forecast, asof}].
|
||
|
||
По каждой <a>-ссылке, похожей на помесячную форму 6 приложения №4 (.xlsx),
|
||
вытаскиваем период «(за|на) <месяц> <год>» из текста/href → period_month (первое
|
||
число месяца) + is_forecast, и дату публикации (DD.MM.YYYY рядом со ссылкой) → asof.
|
||
Годовые/не-xlsx/без периода ссылки игнорируются. URL резолвится к apex-домену.
|
||
"""
|
||
soup = BeautifulSoup(catalog_html, "html.parser")
|
||
out: list[dict] = []
|
||
for a in soup.find_all("a", href=True):
|
||
href = a["href"].strip()
|
||
link_text = _clean_text(a.get_text(" "))
|
||
if not _is_form6_monthly(link_text, href):
|
||
continue
|
||
period = _parse_period(f"{link_text} {href}")
|
||
if period is None: # защита (уже проверено в _is_form6_monthly)
|
||
continue
|
||
year, month, is_forecast = period
|
||
# asof — дата публикации рядом со ссылкой (в тексте ссылки или соседних узлах).
|
||
asof = _extract_asof_near(a, link_text)
|
||
out.append(
|
||
{
|
||
"url": urljoin(BASE_URL, href),
|
||
"period_month": date(year, month, 1),
|
||
"is_forecast": is_forecast,
|
||
"asof": asof,
|
||
}
|
||
)
|
||
logger.info("ekgas: кандидатов формы 6 в каталоге: %d", len(out))
|
||
return out
|
||
|
||
|
||
def _extract_asof_near(anchor: object, link_text: str) -> date | None:
|
||
"""Дата публикации (DD.MM.YYYY) рядом со ссылкой → date. Нет — None.
|
||
|
||
Ищем сначала в тексте самой ссылки, затем в тексте родительского контейнера
|
||
(в каталоге дата обычно стоит в той же строке/ячейке, что и ссылка).
|
||
"""
|
||
candidates = [link_text]
|
||
parent = getattr(anchor, "parent", None)
|
||
if parent is not None:
|
||
candidates.append(_clean_text(parent.get_text(" ")))
|
||
for src in candidates:
|
||
m = _ASOF_RE.search(src)
|
||
if m:
|
||
day, month, year = int(m.group(1)), int(m.group(2)), int(m.group(3))
|
||
try:
|
||
return date(year, month, day)
|
||
except ValueError:
|
||
continue
|
||
return None
|
||
|
||
|
||
def select_latest_file(candidates: list[dict]) -> dict | None:
|
||
"""Выбирает свежайший файл: ПРИОРИТЕТ ФАКТА («за») над прогнозом («на»).
|
||
|
||
Правило (см. модульный docstring): если есть хоть один файл «за» (is_forecast=False),
|
||
берём максимальный (year, month) СРЕДИ «за» — даже если какой-то «на» новее по
|
||
(year, month). Файлов «за» нет вообще → максимальный «на» (is_forecast=True).
|
||
Пусто → None.
|
||
"""
|
||
if not candidates:
|
||
return None
|
||
facts = [c for c in candidates if not c["is_forecast"]]
|
||
pool = facts if facts else candidates
|
||
return max(pool, key=lambda c: (c["period_month"].year, c["period_month"].month))
|
||
|
||
|
||
def _parse_free_capacity(value: object) -> tuple[float | None, bool]:
|
||
"""Ячейка свободной мощности → (free_capacity_mln_m3, needs_calc).
|
||
|
||
«0.000*» (число + звёздочка, любой вариант «0,000*»/«0.000 *») = НЕ ноль, а
|
||
«требуется гидравлический расчёт» → (None, True). Иначе — обычный числовой парс:
|
||
настоящий 0 (без «*») → (0.0, False); дефицит −0.084 сохраняется знаком; «—»/текст
|
||
→ (None, False).
|
||
"""
|
||
if isinstance(value, str) and _NEEDS_CALC_RE.match(value):
|
||
return None, True
|
||
return parse_reserve_number(value), False
|
||
|
||
|
||
def extract_address(outlet_name: str) -> str | None:
|
||
"""Извлекает адресный хвост из «Точки выхода» для геокода DaData (PR-3). Нет — None.
|
||
|
||
Консервативно: возвращаем адрес ТОЛЬКО при явном сигнале (улица-токен ул|пер|пр|
|
||
проезд|б-р|ш|тракт|наб ИЛИ «г.<Город>») + номер дома. Юрлицо без адреса
|
||
(«ИП Рошканяну Владимир Антонович») → None (не выдумываем).
|
||
|
||
Examples:
|
||
«…(г.Екатеринбург, Складской проезд,4а)» → «г.Екатеринбург, Складской проезд,4а»
|
||
«…, пер. Базовый, 17» → «пер. Базовый, 17»
|
||
«ИП Рошканяну Владимир Антонович» → None
|
||
"""
|
||
if not outlet_name:
|
||
return None
|
||
# Требуем явный адресный сигнал во всей строке — иначе номер в названии юрлица/объекта
|
||
# («ООО ТЭЦ-2», «цех №5») ложно распознался бы как дом.
|
||
if not (_HAS_STREET_TOKEN_RE.search(outlet_name) or _HAS_CITY_RE.search(outlet_name)):
|
||
return None
|
||
m = _ADDRESS_RE.search(outlet_name)
|
||
if not m:
|
||
return None
|
||
address = _MULTISPACE_RE.sub(" ", m.group(0)).strip().strip("(),; ")
|
||
return address or None
|
||
|
||
|
||
def _forward_fill_grs(rows: list[list[object]], cols: dict[str, int]) -> list[dict]:
|
||
"""Строки данных → записи с forward-fill «Точки входа» (block-marker ГРС).
|
||
|
||
«Точка входа» заполнена ТОЛЬКО в первой строке блока ГРС → протягиваем
|
||
последнее непустое значение вниз (5 блоков). Индексы колонок — из
|
||
_find_layout (label-детект). Строки до первого block-marker (нет ГРС) и
|
||
single-cell строки (сноска «*» в конце листа) пропускаются.
|
||
"""
|
||
|
||
def _cell(cells: list[object], key: str) -> object:
|
||
idx = cols.get(key, -1)
|
||
return cells[idx] if 0 <= idx < len(cells) else None
|
||
|
||
records: list[dict] = []
|
||
current_grs: str | None = None
|
||
for cells in rows:
|
||
# Single-cell строка (сноска про «*» / служебная) → скип.
|
||
non_empty = [c for c in cells if _clean_text(c)]
|
||
if len(non_empty) <= 1:
|
||
continue
|
||
entry = _clean_text(_cell(cells, "entry"))
|
||
if entry:
|
||
current_grs = entry
|
||
outlet = _clean_text(_cell(cells, "outlet"))
|
||
if not current_grs or not outlet:
|
||
continue # строка до первого block-marker ИЛИ без точки выхода → скип
|
||
free_raw = _cell(cells, "free")
|
||
free, needs_calc = _parse_free_capacity(free_raw)
|
||
records.append(
|
||
{
|
||
"grs_name": current_grs,
|
||
"outlet_name": outlet,
|
||
"consumer_type": _clean_text(_cell(cells, "consumer_type")) or None,
|
||
"consumption_group": _clean_text(_cell(cells, "group")) or None,
|
||
"volume_requested_mln_m3": parse_reserve_number(_cell(cells, "requested")),
|
||
"volume_approved_mln_m3": parse_reserve_number(_cell(cells, "approved")),
|
||
"free_capacity_mln_m3": free,
|
||
"free_capacity_needs_calc": needs_calc,
|
||
"address_extracted": extract_address(outlet),
|
||
"raw_cells": _cells_to_jsonable(cells),
|
||
}
|
||
)
|
||
return records
|
||
|
||
|
||
def _cells_to_jsonable(cells: list[object]) -> list[object]:
|
||
"""Ячейки строки → JSON-совместимый вид (datetime/прочее → str, числа/None as-is)."""
|
||
out: list[object] = []
|
||
for value in cells:
|
||
if value is None or isinstance(value, bool | int | float | str):
|
||
out.append(value)
|
||
else:
|
||
out.append(str(value))
|
||
return out
|
||
|
||
|
||
def _dedup_outlet_names(records: list[dict]) -> list[dict]:
|
||
"""Дедуп повторов outlet_name В ПРЕДЕЛАХ файла: 2-му и далее — суффикс « (N)».
|
||
|
||
Один и тот же outlet_name у одной ГРС может встречаться несколько раз (разные
|
||
группы потребления / объекты одного юрлица). UPSERT по (grs_name_norm, outlet_name,
|
||
period_month, is_forecast) схлопнул бы их с потерей данных (урок тепла/воды) —
|
||
добавляем порядковый суффикс. Ключ дедупа — (grs_name_norm, outlet_name).
|
||
"""
|
||
seen: dict[tuple[str, str], int] = {}
|
||
for rec in records:
|
||
key = (normalize_grs_name(rec["grs_name"]), rec["outlet_name"])
|
||
seen[key] = seen.get(key, 0) + 1
|
||
if seen[key] > 1:
|
||
rec["outlet_name"] = f"{rec['outlet_name']} ({seen[key]})"
|
||
return records
|
||
|
||
|
||
def parse_outlet_sheet(xlsx_bytes: bytes) -> list[dict]:
|
||
"""Парсит лист формы 6 → записи точек выхода (forward-fill ГРС + дедуп имён).
|
||
|
||
Сетка детектится ПО ПОДПИСЯМ (_find_layout): живой файл за май 2026 — header
|
||
row 11, слева колонка «№ п/п», данные с row 13 (разведка видела r10/r12 без
|
||
«№ п/п» — формат плавает). Свободная мощность: float ИЛИ «0.000*» → needs_calc.
|
||
Single-cell сноска про «*» в конце листа → скип. Сетка не распознана → 0 записей
|
||
+ warning (не мусор — урок vodokanal).
|
||
|
||
Каждая запись: {grs_name, outlet_name, consumer_type, consumption_group,
|
||
volume_requested_mln_m3, volume_approved_mln_m3, free_capacity_mln_m3,
|
||
free_capacity_needs_calc, address_extracted, raw_cells}. outlet_name дедуплен
|
||
суффиксом « (N)» в пределах файла.
|
||
"""
|
||
wb = load_workbook(io.BytesIO(xlsx_bytes), read_only=True, data_only=True)
|
||
ws = wb.active
|
||
try:
|
||
all_rows: list[list[object]] = [list(row) for row in ws.iter_rows(values_only=True)]
|
||
finally:
|
||
wb.close()
|
||
|
||
layout = _find_layout(all_rows)
|
||
if layout is None:
|
||
logger.warning(
|
||
"ekgas: сетка листа не распознана (нет header с подписями) — 0 записей; "
|
||
"первые 5 строк: %s",
|
||
[[_clean_text(c) for c in r[:9]] for r in all_rows[:5]],
|
||
)
|
||
return []
|
||
data_start, cols = layout
|
||
records = _forward_fill_grs(all_rows[data_start:], cols)
|
||
records = _dedup_outlet_names(records)
|
||
logger.info("ekgas: распознано точек выхода=%d", len(records))
|
||
return records
|
||
|
||
|
||
def fetch_catalog_html() -> str:
|
||
"""Тянет HTML каталога раскрытия. RUN-ON-PROD (гео-блок). httpx с таймаутом, apex-SSL."""
|
||
resp = httpx.get(
|
||
CATALOG_URL, timeout=_HTTP_TIMEOUT, follow_redirects=True, verify=_ru_ssl_context()
|
||
)
|
||
resp.raise_for_status()
|
||
return resp.text
|
||
|
||
|
||
def fetch_xlsx(url: str) -> bytes:
|
||
"""Тянет XLSX формы 6 по URL. RUN-ON-PROD (гео-блок). httpx с таймаутом, apex-SSL."""
|
||
resp = httpx.get(url, timeout=_HTTP_TIMEOUT, follow_redirects=True, verify=_ru_ssl_context())
|
||
resp.raise_for_status()
|
||
return resp.content
|
||
|
||
|
||
def _upsert_outlet_rows(db: Session, records: list[dict], meta: dict) -> dict[str, int]:
|
||
"""UPSERT точек выхода в gas_grs_outlet_points. Per-row SAVEPOINT (битая строка не валит батч).
|
||
|
||
meta = {period_month: date, is_forecast: bool, asof: date|None, source_url: str}.
|
||
ON CONFLICT (grs_name_norm, outlet_name, period_month, is_forecast) DO UPDATE —
|
||
обновляем объёмы/free/needs_calc/address/raw/fetched_at. Returns счётчики
|
||
inserted/updated/skipped/needs_calc/with_address.
|
||
"""
|
||
inserted = 0
|
||
updated = 0
|
||
skipped = 0
|
||
needs_calc = 0
|
||
with_address = 0
|
||
for rec in records:
|
||
grs_name = rec["grs_name"]
|
||
params = {
|
||
"grs_name": grs_name,
|
||
"grs_name_norm": normalize_grs_name(grs_name),
|
||
"outlet_name": rec["outlet_name"],
|
||
"consumer_type": rec["consumer_type"],
|
||
"consumption_group": rec["consumption_group"],
|
||
"requested": rec["volume_requested_mln_m3"],
|
||
"approved": rec["volume_approved_mln_m3"],
|
||
"free": rec["free_capacity_mln_m3"],
|
||
"needs_calc": rec["free_capacity_needs_calc"],
|
||
"address": rec["address_extracted"],
|
||
"period_month": meta["period_month"],
|
||
"is_forecast": meta["is_forecast"],
|
||
"asof": meta.get("asof"),
|
||
"source_url": meta.get("source_url"),
|
||
"raw": json.dumps({"cells": rec["raw_cells"]}, ensure_ascii=False),
|
||
}
|
||
try:
|
||
with db.begin_nested(): # SAVEPOINT — откат только этой строки
|
||
result = db.execute(
|
||
text("""
|
||
INSERT INTO gas_grs_outlet_points
|
||
(grs_name, grs_name_norm, outlet_name, consumer_type,
|
||
consumption_group, volume_requested_mln_m3,
|
||
volume_approved_mln_m3, free_capacity_mln_m3,
|
||
free_capacity_needs_calc, address_extracted,
|
||
period_month, is_forecast, asof, source_url, raw, fetched_at)
|
||
VALUES (
|
||
:grs_name, :grs_name_norm, :outlet_name, :consumer_type,
|
||
:consumption_group, :requested,
|
||
:approved, :free,
|
||
:needs_calc, :address,
|
||
:period_month, :is_forecast, :asof, :source_url,
|
||
CAST(:raw AS jsonb), NOW()
|
||
)
|
||
ON CONFLICT (grs_name_norm, outlet_name, period_month, is_forecast)
|
||
DO UPDATE
|
||
SET grs_name = EXCLUDED.grs_name,
|
||
consumer_type = EXCLUDED.consumer_type,
|
||
consumption_group = EXCLUDED.consumption_group,
|
||
volume_requested_mln_m3 = EXCLUDED.volume_requested_mln_m3,
|
||
volume_approved_mln_m3 = EXCLUDED.volume_approved_mln_m3,
|
||
free_capacity_mln_m3 = EXCLUDED.free_capacity_mln_m3,
|
||
free_capacity_needs_calc = EXCLUDED.free_capacity_needs_calc,
|
||
address_extracted = EXCLUDED.address_extracted,
|
||
asof = EXCLUDED.asof,
|
||
source_url = EXCLUDED.source_url,
|
||
raw = EXCLUDED.raw,
|
||
fetched_at = NOW()
|
||
RETURNING (xmax = 0) AS is_insert
|
||
"""),
|
||
params,
|
||
).scalar()
|
||
if result:
|
||
inserted += 1
|
||
else:
|
||
updated += 1
|
||
if rec["free_capacity_needs_calc"]:
|
||
needs_calc += 1
|
||
if rec["address_extracted"]:
|
||
with_address += 1
|
||
except Exception as e:
|
||
logger.warning("ekgas upsert failed for %r/%r: %s", grs_name, rec["outlet_name"], e)
|
||
skipped += 1
|
||
return {
|
||
"rows": len(records),
|
||
"inserted": inserted,
|
||
"updated": updated,
|
||
"skipped": skipped,
|
||
"needs_calc": needs_calc,
|
||
"with_address": with_address,
|
||
}
|
||
|
||
|
||
def load_gas_outlet_points(db: Session | None = None) -> dict:
|
||
"""Тянет каталог ekgas → свежайшая форма 6 → парс XLSX → UPSERT точек выхода.
|
||
|
||
RUN-ON-PROD (гео-блок источника). Owns-session как соседи (gas/rosseti/eias): ``db``
|
||
None → своя SessionLocal (fetch + запись sync). Выбор файла — приоритет факта («за»)
|
||
над прогнозом («на»), см. select_latest_file.
|
||
|
||
Returns счётчики {rows, inserted, updated, skipped, needs_calc, with_address,
|
||
period, is_forecast} или {"error": ...} если подходящий файл в каталоге не найден.
|
||
"""
|
||
catalog_html = fetch_catalog_html()
|
||
candidates = extract_form6_links(catalog_html)
|
||
chosen = select_latest_file(candidates)
|
||
if chosen is None:
|
||
logger.warning("ekgas: подходящая форма 6 в каталоге не найдена (0 кандидатов)")
|
||
return {"error": "no form6 file found in catalog"}
|
||
|
||
meta = {
|
||
"period_month": chosen["period_month"],
|
||
"is_forecast": chosen["is_forecast"],
|
||
"asof": chosen["asof"],
|
||
"source_url": chosen["url"],
|
||
}
|
||
logger.info(
|
||
"ekgas: выбран файл period=%s is_forecast=%s asof=%s url=%s",
|
||
chosen["period_month"],
|
||
chosen["is_forecast"],
|
||
chosen["asof"],
|
||
chosen["url"],
|
||
)
|
||
|
||
xlsx_bytes = fetch_xlsx(chosen["url"])
|
||
records = parse_outlet_sheet(xlsx_bytes)
|
||
|
||
owns_session = db is None
|
||
if db is None:
|
||
db = SessionLocal()
|
||
|
||
try:
|
||
counts = _upsert_outlet_rows(db, records, meta)
|
||
db.commit()
|
||
except Exception as e:
|
||
db.rollback()
|
||
logger.exception("load_gas_outlet_points: outer tx rolled back: %s", e)
|
||
raise
|
||
finally:
|
||
if owns_session:
|
||
db.close()
|
||
|
||
counts["period"] = chosen["period_month"].isoformat()
|
||
counts["is_forecast"] = chosen["is_forecast"]
|
||
logger.info("load_gas_outlet_points done: %s", counts)
|
||
return counts
|