From d5607a118aecabd6b84b55a0dfc5a1f9f769a5b0 Mon Sep 17 00:00:00 2001 From: bot-backend Date: Fri, 3 Jul 2026 07:10:05 +0500 Subject: [PATCH] =?UTF-8?q?feat(site-finder):=20=D0=BB=D0=BE=D0=B0=D0=B4?= =?UTF-8?q?=D0=B5=D1=80=20=D1=82=D0=BE=D1=87=D0=B5=D0=BA=20=D0=B2=D1=8B?= =?UTF-8?q?=D1=85=D0=BE=D0=B4=D0=B0=20=D0=93=D0=A0=D0=A1=20=C2=AB=D0=95?= =?UTF-8?q?=D0=BA=D0=B0=D1=82=D0=B5=D1=80=D0=B8=D0=BD=D0=B1=D1=83=D1=80?= =?UTF-8?q?=D0=B3=D0=B3=D0=B0=D0=B7=C2=BB=20(#2119=20B2,=20PR-2/4)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../site_finder/ekgas_outlet_loader.py | 557 ++++++++++++++++++ backend/tests/test_ekgas_outlet_loader.py | 457 ++++++++++++++ 2 files changed, 1014 insertions(+) create mode 100644 backend/app/services/site_finder/ekgas_outlet_loader.py create mode 100644 backend/tests/test_ekgas_outlet_loader.py diff --git a/backend/app/services/site_finder/ekgas_outlet_loader.py b/backend/app/services/site_finder/ekgas_outlet_loader.py new file mode 100644 index 00000000..ec0dbdef --- /dev/null +++ b/backend/app/services/site_finder/ekgas_outlet_loader.py @@ -0,0 +1,557 @@ +"""Загрузчик адресных точек выхода газораспределительной сети (#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 меняется + с каждым файлом → 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. +_HEADER_ROW = 10 # строка подписей («Точка входа»/«Точка выхода»/…) +_DATA_START_ROW = 12 # первая строка данных (после нумерации row 11) + +# Колонки формы 6 (0-indexed для списка ячеек строки). +_COL_ENTRY = 0 # «Точка входа» (ГРС, block-marker → forward-fill) +_COL_OUTLET = 1 # «Точка выхода» (свободный текст) +_COL_TYPE = 2 # тип потребителя (котельная/ТЭЦ/…) +_COL_GROUP = 3 # группа потребления +_COL_REQUESTED = 4 # объём заявок, млн. м³/мес +_COL_APPROVED = 5 # удовлетворено, млн. м³/мес +_COL_FREE = 6 # свободная мощность, млн. м³/мес (float ИЛИ «0.000*») + +# «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}]. + + По каждой -ссылке, похожей на помесячную форму 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]]) -> list[dict]: + """Строки данных → записи с forward-fill «Точки входа» (block-marker ГРС). + + «Точка входа» (col0) заполнена ТОЛЬКО в первой строке блока ГРС → протягиваем + последнее непустое значение вниз (5 блоков). Строки до первого block-marker (нет + ГРС) и single-cell строки (сноска «*» в конце листа) пропускаются. + """ + 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(cells[_COL_ENTRY]) if len(cells) > _COL_ENTRY else "" + if entry: + current_grs = entry + outlet = _clean_text(cells[_COL_OUTLET]) if len(cells) > _COL_OUTLET else "" + if not current_grs or not outlet: + continue # строка до первого block-marker ИЛИ без точки выхода → скип + free_raw = cells[_COL_FREE] if len(cells) > _COL_FREE else None + free, needs_calc = _parse_free_capacity(free_raw) + records.append( + { + "grs_name": current_grs, + "outlet_name": outlet, + "consumer_type": _clean_text(cells[_COL_TYPE]) or None + if len(cells) > _COL_TYPE + else None, + "consumption_group": _clean_text(cells[_COL_GROUP]) or None + if len(cells) > _COL_GROUP + else None, + "volume_requested_mln_m3": parse_reserve_number( + cells[_COL_REQUESTED] if len(cells) > _COL_REQUESTED else None + ), + "volume_approved_mln_m3": parse_reserve_number( + cells[_COL_APPROVED] if len(cells) > _COL_APPROVED else None + ), + "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 ГРС + дедуп имён). + + РЕАЛЬНАЯ сетка (снята с прода): подписи row 10 (1-indexed), нумерация 1..7 row 11, + данные с row 12; col1=ГРС (block-marker, forward-fill), col2=точка выхода, + col3=тип, col4=группа, col5=заявки, col6=удовлетворено, col7=свободная мощность + (float ИЛИ «0.000*» → needs_calc). Single-cell сноска про «*» в конце листа → скип. + + Каждая запись: {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: + data_rows: list[list[object]] = [] + for i, row in enumerate(ws.iter_rows(values_only=True), start=1): + if i < _DATA_START_ROW: + continue # подписи (row 10) + нумерация (row 11) пропускаем + data_rows.append(list(row)) + finally: + wb.close() + + records = _forward_fill_grs(data_rows) + 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 diff --git a/backend/tests/test_ekgas_outlet_loader.py b/backend/tests/test_ekgas_outlet_loader.py new file mode 100644 index 00000000..83ca5ce7 --- /dev/null +++ b/backend/tests/test_ekgas_outlet_loader.py @@ -0,0 +1,457 @@ +"""Unit-тесты загрузчика точек выхода ГРС «Екатеринбурггаз» (ekgas.ru) (#2119 B2, PR-2). + +Pure / mock-based — без реальной сети и БД. Покрывает: +- extract_form6_links + select_latest_file: выбор свежайшего файла из синтетического + каталога с ПРИОРИТЕТОМ ФАКТА («за») над прогнозом («на») — даже когда «на» новее по + (год, месяц) (отдельный тест); годовой .xls / не-форма-6 отфильтрованы; +- parse_outlet_sheet: РЕАЛЬНАЯ сетка (подписи row 10, нумерация row 11, данные с row 12), + forward-fill block-marker ГРС, «0.000*»→needs_calc (НЕ ноль), дефицит −0.084 сохранён, + сноска-скип single-cell; +- extract_address: скобочный/хвостовой адрес извлечён, юрлицо без адреса → None; +- _dedup_outlet_names: суффикс « (N)»; _upsert_outlet_rows: счётчики (mock-DB). +""" + +from __future__ import annotations + +import io +from contextlib import contextmanager +from datetime import date +from typing import Any + +import openpyxl + +from app.services.site_finder import ekgas_outlet_loader as ek + +# ── Каталог: синтетический HTML со ссылками на формы 6 ──────────────────────── + + +def _form6_name(kind: str, month: str, year: int) -> str: + """Имя файла формы 6 приложения №4 в реальном формате ekgas.""" + return ( + f"Форма 6 приложения № 4 к Приказу ФАС России № 960 от 08.12.2022г. " + f"{kind} {month} {year}г..xlsx" + ) + + +def _catalog_link(kind: str, month: str, year: int, hex_dir: str, asof: str = "") -> str: + """ каталога: путь /uploads/default//<имя> + опц. дата публикации рядом.""" + name = _form6_name(kind, month, year) + href = f"/uploads/default/{hex_dir}/{name}" + asof_html = f"{asof}" if asof else "" + return f'
{name}{asof_html}
' + + +def test_extract_form6_links_parses_period_and_asof() -> None: + """Период «(за|на) <месяц> <год>» + asof (DD.MM.YYYY рядом) извлекаются; URL к apex.""" + html = ( + "" + + _catalog_link("за", "май", 2026, "a1b2c3", asof="03.06.2026") + + "" + ) + + links = ek.extract_form6_links(html) + + assert len(links) == 1 + link = links[0] + assert link["period_month"] == date(2026, 5, 1) + assert link["is_forecast"] is False # «за» = факт + assert link["asof"] == date(2026, 6, 3) + assert link["url"].startswith("https://ekgas.ru/uploads/default/a1b2c3/") + + +def test_extract_form6_links_na_is_forecast() -> None: + """«на <месяц>» → is_forecast=True (прогноз).""" + html = "" + _catalog_link("на", "июль", 2026, "deadbeef") + "" + links = ek.extract_form6_links(html) + assert len(links) == 1 + assert links[0]["is_forecast"] is True + assert links[0]["period_month"] == date(2026, 7, 1) + + +def test_extract_form6_links_skips_yearly_xls_and_foreign() -> None: + """Годовой «Форма №6 (Приложение №2) … план на <год>» (.xls) и прочие файлы — скип.""" + html = ( + "" + # годовой план — приложение №2, .xls → НЕ форма 6 приложения №4 + '' + "годовой план" + # посторонний файл + 'прайс' + # валидная помесячная форма 6 + + _catalog_link("за", "апрель", 2026, "cc22") + + "" + ) + + links = ek.extract_form6_links(html) + + assert len(links) == 1 + assert links[0]["period_month"] == date(2026, 4, 1) + + +def test_select_latest_file_fact_beats_newer_forecast() -> None: + """ПРИОРИТЕТ ФАКТА: «за май 2026» выбирается, хотя «на июль 2026» новее по (год,месяц). + + Документирует правило select_latest_file: если есть хоть один файл «за» (факт), + берём максимальный (год, месяц) СРЕДИ «за», даже если прогноз «на» новее. Прогноз + берётся ТОЛЬКО когда файлов «за» нет вообще. + """ + html = ( + "" + + _catalog_link("за", "апрель", 2026, "h1") + + _catalog_link("за", "май", 2026, "h2") # свежайший ФАКТ + + _catalog_link("на", "июль", 2026, "h3") # прогноз новее по (год,месяц) + + "" + ) + + chosen = ek.select_latest_file(ek.extract_form6_links(html)) + + assert chosen is not None + assert chosen["period_month"] == date(2026, 5, 1) # «за май», не «на июль» + assert chosen["is_forecast"] is False + + +def test_select_latest_file_forecast_only_fallback() -> None: + """Файлов «за» нет вообще → берём свежайший «на» (прогноз, is_forecast=True).""" + html = ( + "" + + _catalog_link("на", "июнь", 2026, "f1") + + _catalog_link("на", "август", 2026, "f2") # свежайший прогноз + + "" + ) + + chosen = ek.select_latest_file(ek.extract_form6_links(html)) + + assert chosen is not None + assert chosen["period_month"] == date(2026, 8, 1) + assert chosen["is_forecast"] is True + + +def test_select_latest_file_empty_returns_none() -> None: + assert ek.select_latest_file([]) is None + + +# ── parse_outlet_sheet: РЕАЛЬНАЯ сетка (row 10 подписи, 11 нумерация, 12+ данные) ── + + +def _pad(cells: dict[int, object], width: int = 7) -> list[object]: + """Строка листа из width колонок: dict{col_index: value} → выровненный ряд.""" + row: list[object] = [None] * width + for idx, value in cells.items(): + row[idx] = value + return row + + +def _outlet_xlsx_bytes() -> bytes: + """XLSX формы 6 — сокращённая РЕАЛЬНАЯ сетка (собрана openpyxl'ом как на проде). + + Строки 1-9 пустые/шапка, row 10 = подписи, row 11 = нумерация 1..7, данные с row 12: + * 2 блока ГРС (block-marker в col0 только в 1-й строке блока → forward-fill); + * «0.000*» в col6 → needs_calc (НЕ ноль); + * отрицательный free (дефицит −0.084) сохраняется знаком; + * адрес в «Точке выхода» → address_extracted; юрлицо без адреса → None; + * последняя строка — сноска про «*» (single-cell) → скип. + """ + wb = openpyxl.Workbook() + ws = wb.active + ws.title = "Форма 6" + + # Строки 1-9: пустая шапка формы (title/период/пусто) — до подписей. + for _ in range(9): + ws.append(_pad({})) + # row 10 — подписи колонок. + ws.append( + _pad( + { + 0: "Точка входа", + 1: "Точка выхода", + 2: "Тип потребителя", + 3: "Группа потребления", + 4: "Объём заявок, млн.куб.м", + 5: "Удовлетворено, млн.куб.м", + 6: "Свободная мощность, млн.куб.м", + } + ) + ) + # row 11 — нумерация столбцов 1..7. + ws.append(_pad({0: 1, 1: 2, 2: 3, 3: 4, 4: 5, 5: 6, 6: 7})) + + # row 12 — блок 1, первая строка (block-marker ГРС в col0). + ws.append( + _pad( + { + 0: "ГРС Арамиль", + 1: 'ООО "Тепло" (г.Екатеринбург, Складской проезд,4а)', + 2: "котельная", + 3: "5 - 0,1-1млн.куб.м", + 4: 1.2, + 5: 0.8, + 6: 0.4, + } + ) + ) + # row 13 — блок 1, вторая строка (col0 ПУСТ → forward-fill «ГРС Арамиль»), + # свободная мощность «0.000*» → needs_calc. + ws.append( + _pad( + { + 1: "ИП Рошканяну Владимир Антонович", # юрлицо без адреса → address None + 2: "промплощадка", + 4: 0.05, + 5: 0.05, + 6: "0.000*", # НЕ ноль — требует гидрорасчёта + } + ) + ) + # row 14 — блок 2, первая строка (новый block-marker), дефицит −0.084. + ws.append( + _pad( + { + 0: "ГРС Кольцово", + 1: 'ЗАО "Луч", пер. Базовый, 17', # адрес хвостом + 2: "ТЭЦ", + 3: "4", + 4: 2.5, + 5: 2.5, + 6: -0.084, # дефицит — знак сохраняется + } + ) + ) + # row 15 — сноска про «*» (single-cell) → скип. + ws.append(_pad({0: "* требуется гидравлический расчёт свободной мощности"})) + + buf = io.BytesIO() + wb.save(buf) + return buf.getvalue() + + +def test_parse_outlet_sheet_forward_fill_and_markers() -> None: + """Forward-fill ГРС, «0.000*»→needs_calc, дефицит сохранён, адрес извлечён, сноска-скип.""" + records = ek.parse_outlet_sheet(_outlet_xlsx_bytes()) + + # 3 строки данных (сноска single-cell отброшена). + assert len(records) == 3 + + r0, r1, r2 = records + + # Блок 1, строка 1. + assert r0["grs_name"] == "ГРС Арамиль" + assert r0["free_capacity_mln_m3"] == 0.4 + assert r0["free_capacity_needs_calc"] is False + assert r0["address_extracted"] == "г.Екатеринбург, Складской проезд,4а" + + # Блок 1, строка 2 — forward-fill той же ГРС + «0.000*» → needs_calc, NULL free. + assert r1["grs_name"] == "ГРС Арамиль" # col0 был пуст → протянут + assert r1["free_capacity_mln_m3"] is None + assert r1["free_capacity_needs_calc"] is True # «0.000*» — НЕ ноль + assert r1["address_extracted"] is None # ИП без адреса + + # Блок 2 — новый block-marker + дефицит −0.084 (знак сохранён) + адрес хвостом. + assert r2["grs_name"] == "ГРС Кольцово" + assert r2["free_capacity_mln_m3"] == -0.084 + assert r2["free_capacity_needs_calc"] is False + assert r2["address_extracted"] == "пер. Базовый, 17" + + # raw_cells — сырая строка списком (для jsonb raw). + assert r0["raw_cells"][0] == "ГРС Арамиль" + + +def test_parse_outlet_sheet_volumes_and_meta() -> None: + """Объёмы заявок/удовлетворено и тип/группа парсятся из своих колонок.""" + records = ek.parse_outlet_sheet(_outlet_xlsx_bytes()) + r0 = records[0] + assert r0["volume_requested_mln_m3"] == 1.2 + assert r0["volume_approved_mln_m3"] == 0.8 + assert r0["consumer_type"] == "котельная" + assert r0["consumption_group"] == "5 - 0,1-1млн.куб.м" + + +# ── extract_address: консервативный адрес-экстрактор ───────────────────────── + + +def test_extract_address_parenthetical() -> None: + """«(г.Екатеринбург, Складской проезд,4а)» → адрес извлечён.""" + name = 'ООО "Тепло" (г.Екатеринбург, Складской проезд,4а)' + assert ek.extract_address(name) == "г.Екатеринбург, Складской проезд,4а" + + +def test_extract_address_trailing() -> None: + """«, пер. Базовый, 17» хвостом → адрес извлечён.""" + assert ek.extract_address('ЗАО "Луч", пер. Базовый, 17') == "пер. Базовый, 17" + + +def test_extract_address_street_with_house() -> None: + """«г.…, ул. К.Маркса,31» → адрес с улицей и домом.""" + got = ek.extract_address("Котельная, г. Реж, ул. К.Маркса,31") + assert got is not None + assert "К.Маркса" in got + assert got.rstrip().endswith("31") + + +def test_extract_address_legal_entity_returns_none() -> None: + """Юрлицо без адресного сигнала → None (не выдумываем).""" + assert ek.extract_address("ИП Рошканяну Владимир Антонович") is None + + +def test_extract_address_no_street_signal_returns_none() -> None: + """Название с числом, но без улицы/города («ООО ТЭЦ-2») → None (число ≠ дом).""" + assert ek.extract_address('ООО "ТЭЦ-2"') is None + assert ek.extract_address("") is None + + +# ── _parse_free_capacity: «0.000*» vs настоящий 0 vs дефицит ───────────────── + + +def test_parse_free_capacity_needs_calc_marker() -> None: + """«0.000*» → (None, True). Вариации с запятой/пробелом тоже маркер.""" + assert ek._parse_free_capacity("0.000*") == (None, True) + assert ek._parse_free_capacity("0,000 *") == (None, True) + + +def test_parse_free_capacity_real_zero_and_deficit() -> None: + """Настоящий 0 (без «*») → (0.0, False); дефицит −0.084 → (−0.084, False).""" + assert ek._parse_free_capacity(0.0) == (0.0, False) + assert ek._parse_free_capacity(-0.084) == (-0.084, False) + assert ek._parse_free_capacity("—") == (None, False) + + +# ── _dedup_outlet_names: повторы outlet_name внутри файла ───────────────────── + + +def test_dedup_outlet_names_suffix() -> None: + """Одинаковые outlet_name у одной ГРС → 2-му и далее суффикс « (N)» (не схлопывать).""" + records = [ + {"grs_name": "ГРС Арти", "outlet_name": "Котельная №1"}, + {"grs_name": "ГРС Арти", "outlet_name": "Котельная №1"}, # дубль + {"grs_name": "ГРС Арти", "outlet_name": "Котельная №1"}, # ещё дубль + {"grs_name": "ГРС Кольцово", "outlet_name": "Котельная №1"}, # другая ГРС → не дубль + ] + + ek._dedup_outlet_names(records) + + assert [r["outlet_name"] for r in records] == [ + "Котельная №1", + "Котельная №1 (2)", + "Котельная №1 (3)", + "Котельная №1", # другая ГРС — суффикс не нужен + ] + + +# ── Fake DB (SAVEPOINT + execute), как в test_eias_heat_loader.py ───────────── + + +class _FakeResult: + def __init__(self, scalar_value: Any = True) -> None: + self._scalar = scalar_value + + def scalar(self) -> Any: + return self._scalar + + +class _FakeSession: + """Мини-заглушка Session: собирает execute-параметры, поддерживает begin_nested.""" + + def __init__(self, scalar_value: Any = True) -> None: + self.calls: list[tuple[str, Any]] = [] + self._scalar = scalar_value + + @contextmanager + def begin_nested(self): # type: ignore[no-untyped-def] + yield + + def execute(self, stmt: Any, params: Any = None) -> _FakeResult: + self.calls.append((str(stmt), params)) + return _FakeResult(self._scalar) + + +def _record( + grs: str, + outlet: str, + free: float | None, + needs_calc: bool = False, + address: str | None = None, +) -> dict: + return { + "grs_name": grs, + "outlet_name": outlet, + "consumer_type": "котельная", + "consumption_group": "4", + "volume_requested_mln_m3": 1.0, + "volume_approved_mln_m3": 0.9, + "free_capacity_mln_m3": free, + "free_capacity_needs_calc": needs_calc, + "address_extracted": address, + "raw_cells": [grs, outlet, "котельная"], + } + + +_META = { + "period_month": date(2026, 5, 1), + "is_forecast": False, + "asof": date(2026, 6, 3), + "source_url": "https://ekgas.ru/uploads/default/h1/f.xlsx", +} + + +def test_upsert_outlet_rows_counters_and_params() -> None: + """Счётчики inserted/needs_calc/with_address; params корректны; CAST jsonb + ON CONFLICT.""" + db = _FakeSession(scalar_value=True) # xmax=0 → insert + records = [ + _record("ГРС Арамиль", "ООО Тепло", 0.4, address="пер. Базовый, 17"), + _record("ГРС Арамиль", "ИП Рошканяну", None, needs_calc=True), + _record("ГРС Кольцово", "ЗАО Луч", -0.084), # дефицит: знак сохраняется + ] + + counts = ek._upsert_outlet_rows(db, records, {**_META, "source_url": _META["source_url"]}) + + assert counts == { + "rows": 3, + "inserted": 3, + "updated": 0, + "skipped": 0, + "needs_calc": 1, + "with_address": 1, + } + assert len(db.calls) == 3 + + sql, first = db.calls[0] + assert first["grs_name"] == "ГРС Арамиль" + assert first["grs_name_norm"] == "грс арамиль" # нормализация из B1 + assert first["free"] == 0.4 + assert first["period_month"] == date(2026, 5, 1) + assert first["is_forecast"] is False + assert first["address"] == "пер. Базовый, 17" + assert '"cells"' in first["raw"] + assert "CAST(:raw AS jsonb)" in sql + assert "ON CONFLICT (grs_name_norm, outlet_name, period_month, is_forecast)" in sql + # needs_calc-строка: free NULL, needs_calc True прокинуты. + assert db.calls[1][1]["free"] is None + assert db.calls[1][1]["needs_calc"] is True + # дефицит сохранён знаком. + assert db.calls[2][1]["free"] == -0.084 + + +def test_upsert_outlet_rows_update_counter() -> None: + """scalar()=False (xmax≠0) → строка учтена как update, не insert.""" + db = _FakeSession(scalar_value=False) + counts = ek._upsert_outlet_rows(db, [_record("ГРС X", "outlet", 1.0)], _META) + assert counts["updated"] == 1 + assert counts["inserted"] == 0 + + +def test_upsert_outlet_rows_skip_on_error() -> None: + """execute бросает → строка skipped, счётчики insert/update не растут.""" + + class _RaisingSession(_FakeSession): + def execute(self, stmt: Any, params: Any = None) -> _FakeResult: + raise RuntimeError("db boom") + + db = _RaisingSession() + counts = ek._upsert_outlet_rows(db, [_record("ГРС Y", "outlet", 2.0)], _META) + assert counts == { + "rows": 1, + "inserted": 0, + "updated": 0, + "skipped": 1, + "needs_calc": 0, + "with_address": 0, + } -- 2.45.3