feat(site-finder): лоадер точек выхода ГРС «Екатеринбурггаз» (#2119 B2, PR-2/4)

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>/<имя>``. 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}].
+
+    По каждой <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]]) -> 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

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:
+    """<a> каталога: путь /uploads/default/<hex>/<имя> + опц. дата публикации рядом."""
+    name = _form6_name(kind, month, year)
+    href = f"/uploads/default/{hex_dir}/{name}"
+    asof_html = f"<span>{asof}</span>" if asof else ""
+    return f'<div class="row"><a href="{href}">{name}</a>{asof_html}</div>'
+
+
+def test_extract_form6_links_parses_period_and_asof() -> None:
+    """Период «(за|на) <месяц> <год>» + asof (DD.MM.YYYY рядом) извлекаются; URL к apex."""
+    html = (
+        "<html><body>"
+        + _catalog_link("за", "май", 2026, "a1b2c3", asof="03.06.2026")
+        + "</body></html>"
+    )
+
+    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 = "<html>" + _catalog_link("на", "июль", 2026, "deadbeef") + "</html>"
+    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 = (
+        "<html>"
+        # годовой план — приложение №2, .xls → НЕ форма 6 приложения №4
+        '<a href="/uploads/default/ff00/Форма №6 (Приложение №2) план на 2026г..xls">'
+        "годовой план</a>"
+        # посторонний файл
+        '<a href="/uploads/default/aa11/Прайс-лист.pdf">прайс</a>'
+        # валидная помесячная форма 6
+        + _catalog_link("за", "апрель", 2026, "cc22")
+        + "</html>"
+    )
+
+    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 = (
+        "<html>"
+        + _catalog_link("за", "апрель", 2026, "h1")
+        + _catalog_link("за", "май", 2026, "h2")  # свежайший ФАКТ
+        + _catalog_link("на", "июль", 2026, "h3")  # прогноз новее по (год,месяц)
+        + "</html>"
+    )
+
+    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 = (
+        "<html>"
+        + _catalog_link("на", "июнь", 2026, "f1")
+        + _catalog_link("на", "август", 2026, "f2")  # свежайший прогноз
+        + "</html>"
+    )
+
+    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,
+    }