gendesign/backend/app/services/site_finder/eesk_reserve_loader.py

"""Загрузчик резервов свободной мощности АО «ЕЭСК» из xlsx-раскрытия (#2119 B2).

Хаб eesk.ru «Наличие объёма свободной для технологического присоединения
мощности» — статический HTML со ссылками на xlsx двух серий:
  (а) ПС 35-220 кВ («резерв мощности по центрам питания», текст ссылки БЕЗ «ниже»
      / содержит «35-220») → апдейт ``power_supply_centers`` (резерв в МВт!);
  (б) ТП/РП «ниже 35» кВ (текст ссылки содержит «ниже 35»)
      → UPSERT ``power_tp_rp_reserves``.

Свежайший файл каждой серии выбирается по max(YYYY, N-квартал) из ТЕКСТА ссылки
(«…N квартал(а) YYYY…»). period='YYYY-QN', reserve_asof = последний день квартала.

Источник — обычный SSL, БЕЗ гео-блока, но лоадер всё равно RUN-ON-PROD рядом с
rosseti/gas/vodokanal (единый weekly-воркер). httpx с явным таймаутом, per-row
SAVEPOINT при UPSERT (битая строка не валит батч).

Числовой парс/дата — переиспользуют ``parse_reserve_number`` / ``parse_asof_date``
из ``rosseti_reserve_loader``; матч ЦП — общий ``normalize_sc_name``.
"""

import io
import logging
import re
from datetime import date

import httpx
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.rosseti_reserve_loader import (
    parse_asof_date,  # noqa: F401 — переэкспорт для соседей/тестов симметрии
    parse_reserve_number,
)
from app.services.site_finder.rosseti_wfs_loader import normalize_sc_name

logger = logging.getLogger(__name__)

# Хаб раскрытия (статический HTML). ЕЭСК — латиница, punycode не требуется.
HUB_URL = (
    "https://www.eesk.ru/Raskritie_informacii/standarty_raskrytia_info/"
    "Ob_osnovnih_potrebitelskih_harakteristik/"
    "Informacija_o_tehnicheskom_sostojanii_se/"
    "Nalichie_obema_svobodnoj_dlja_tehnologic"
)

_HTTP_TIMEOUT = 60

# Ссылки на xlsx-файлы в загрузках сайта: /upload/site1/document_file/<hash>.xls[x].
# Захватываем href + видимый ТЕКСТ ссылки (в тексте — серия и квартал/год).
_LINK_RE = re.compile(
    r'<a\b[^>]*href=["\'](?P<href>[^"\']*/upload/site1/document_file/'
    r'[^"\']+?\.xlsx?)["\'][^>]*>(?P<text>.*?)</a>',
    re.IGNORECASE | re.DOTALL,
)

# «N квартал[а] YYYY» в тексте ссылки → (N, YYYY). Слово «квартал» в любой форме.
_QUARTER_RE = re.compile(r"(\d)\s*квартал\w*\s*(\d{4})", re.IGNORECASE)

_BASE = "https://www.eesk.ru"

# Последний день квартала (месяц, день) по номеру квартала.
_QUARTER_LAST_DAY: dict[int, tuple[int, int]] = {
    1: (3, 31),
    2: (6, 30),
    3: (9, 30),
    4: (12, 31),
}


def _strip_tags(html_fragment: str) -> str:
    """Убирает html-теги/entity-мусор из текста ссылки → плоская строка."""
    text_only = re.sub(r"<[^>]+>", " ", html_fragment)
    text_only = text_only.replace("&nbsp;", " ").replace("\xa0", " ")
    return re.sub(r"\s+", " ", text_only).strip()


def _classify_series(link_text_low: str) -> str | None:
    """Классифицирует ссылку по её тексту: 'ps_35_220' | 'tp_rp_sub35' | None.

    Различаем серии ТОЛЬКО по тексту ссылки:
      (a) ПС 35-220 — содержит «35-220» ИЛИ «резерв мощности по центрам питания»,
          при этом НЕ содержит «ниже» (иначе это серия б);
      (b) ТП/РП <35 — содержит «ниже 35».
    """
    has_below = "ниже 35" in link_text_low or "ниже35" in link_text_low
    if has_below:
        return "tp_rp_sub35"
    if "35-220" in link_text_low or "по центрам питания" in link_text_low:
        return "ps_35_220"
    return None


def _quarter_from_text(link_text: str) -> tuple[int, int] | None:
    """Извлекает (year, quarter) из текста ссылки «…N квартал(а) YYYY…». Нет → None."""
    m = _QUARTER_RE.search(link_text)
    if not m:
        return None
    quarter = int(m.group(1))
    year = int(m.group(2))
    if quarter not in _QUARTER_LAST_DAY:
        return None
    return (year, quarter)


def quarter_period(year: int, quarter: int) -> str:
    """(2026, 2) → '2026-Q2'."""
    return f"{year}-Q{quarter}"


def quarter_last_day(year: int, quarter: int) -> date:
    """Последний календарный день квартала → date (для reserve_asof)."""
    month, day = _QUARTER_LAST_DAY[quarter]
    return date(year, month, day)


def fetch_hub_links() -> dict[str, dict]:
    """Скрейпит хаб → СВЕЖАЙШИЙ xlsx каждой серии (ps_35_220 + tp_rp_sub35).

    RUN-ON-PROD. GET с follow_redirects, regex по <a href …upload…xlsx>ТЕКСТ</a>,
    классификация серии + квартал по ТЕКСТУ ссылки, выбор max(year, quarter).

    Returns: {series: {"url": str, "year": int, "quarter": int}} — только найденные.
    """
    resp = httpx.get(HUB_URL, timeout=_HTTP_TIMEOUT, follow_redirects=True)
    resp.raise_for_status()
    return select_freshest_links(resp.text)


def select_freshest_links(html: str) -> dict[str, dict]:
    """Из HTML-хаба выбирает свежайшую ссылку каждой серии (чистая, тестируемая).

    Свежайший = max((year, quarter)) среди ссылок одной серии. Ссылки без
    распознанного квартала игнорируются (нельзя датировать).
    """
    best: dict[str, dict] = {}
    n_links = 0
    for m in _LINK_RE.finditer(html):
        n_links += 1
        link_text = _strip_tags(m.group("text"))
        series = _classify_series(link_text.lower())
        if series is None:
            continue
        yq = _quarter_from_text(link_text)
        if yq is None:
            continue
        year, quarter = yq
        href = m.group("href")
        url = href if href.startswith("http") else _BASE + href
        prev = best.get(series)
        if prev is None or (year, quarter) > (prev["year"], prev["quarter"]):
            best[series] = {"url": url, "year": year, "quarter": quarter}

    logger.info(
        "eesk hub links: ps_35_220=%r tp_rp_sub35=%r (from %d xlsx links)",
        best.get("ps_35_220"),
        best.get("tp_rp_sub35"),
        n_links,
    )
    return best


def _cell(row: tuple, idx: int) -> object:
    """Безопасно достаёт ячейку row по 0-based индексу (None если за границей)."""
    return row[idx] if idx < len(row) else None


def _pct_share_to_percent(value: object) -> float | None:
    """Доля загрузки (0.41) → проценты (41.0). Уже-проценты (>1) не трогаем.

    В xlsx ЕЭСК степень загрузки хранится ДОЛЕЙ (0..1).

    #2464-B: продакшен-вызывающих у функции СЕЙЧАС НЕТ. Значение колонки E
    раньше писалось в `load_index`, но это категориальная колонка
    ('open'|'limited'|'closed'|NULL) — число в ней фронт отбрасывает в
    «неизвестно» и плодит мусорный бакет в `power_summary.by_load_index`.
    Функцию оставляю с тестами: она описывает формат листа, и она понадобится
    в тот момент, когда под процент загрузки заведут числовую колонку.
    Если такого решения не будет — удалить вместе с тестом, а не держать молча.

    None/мусор → None.
    """
    num = parse_reserve_number(value)
    if num is None:
        return None
    # Доля 0..1 → *100. Значения >1 считаем уже процентами (защита от смешанного ввода).
    return round(num * 100.0, 2) if -1.0 <= num <= 1.0 else num


def load_ps_35_220(db: Session, xlsx_bytes: bytes, reserve_asof: date | None) -> dict[str, int]:
    """Апдейт power_supply_centers резервом ЕЭСК ПС 35-220 из xlsx.

    ВЕРИФИЦИРОВАННАЯ структура (лист ``Лист1``, снята с живого файла): шапка строка
    4 (1-based), данные строки 7-63 и 65-83, между секциями строки-разделители.
    Скипаем строки где колонка C (имя ПС) пуста или D (мощность) не парсится числом.
    Колонки 0-based: A=№, B=адм. район, C=имя ПС, D=макс полная мощность МВА,
    E=% загрузки ДОЛЯ (0.41→41.0), F=степень МУ250 (в raw), G=свободная МВт.

    Матч к power_supply_centers по normalize_sc_name(C) AND dzo_name ILIKE '%ЕЭСК%'.
    reserve_mva=G, reserve_unit='МВт', installed_capacity_mva=D, district=B,
    capacity_source='eesk_35_220', reserve_asof. Unmatched имена — warning-лог.

    Returns: счётчики rows/matched/unmatched.
    """
    wb = load_workbook(io.BytesIO(xlsx_bytes), read_only=True, data_only=True)
    ws = wb["Лист1"] if "Лист1" in wb.sheetnames else wb.active

    rows_seen = 0
    matched = 0
    unmatched = 0
    unmatched_names: list[str] = []

    try:
        # min_row=7 (1-based) → данные с 7-й строки; секции 7-63 и 65-83 идут подряд,
        # строки-разделители отсеиваются проверкой C пусто / D не число.
        for row in ws.iter_rows(min_row=7, values_only=True):
            sc_name = _cell(row, 2)  # C
            installed = parse_reserve_number(_cell(row, 3))  # D
            # Строка-разделитель между секциями: нет имени ПС ИЛИ мощность не число.
            if sc_name is None or not str(sc_name).strip() or installed is None:
                continue
            rows_seen += 1

            district = _cell(row, 1)  # B
            # Колонку E (степень загрузки ЦП долей) НЕ читаем и не храним: места
            # под неё в power_supply_centers нет — load_index категориальный,
            # current_load_mva в мегавольт-амперах (#2464-B, см. UPDATE ниже).
            reserve = parse_reserve_number(_cell(row, 6))  # G (свободная МВт)
            name_norm = normalize_sc_name(str(sc_name))

            params = {
                "installed": installed,
                "reserve": reserve,
                "asof": reserve_asof,
                "district": str(district).strip() if district else None,
                "name_norm": name_norm,
            }

            try:
                with db.begin_nested():  # SAVEPOINT — битая строка не валит батч
                    res = db.execute(
                        text("""
                            UPDATE power_supply_centers
                               SET reserve_mva            = :reserve,
                                   reserve_unit           = 'МВт',
                                   installed_capacity_mva = :installed,
                                   district               = :district,
                                   -- #2464-B: сюда БОЛЬШЕ НЕ пишем степень загрузки:
                                   -- load_index — категориальная колонка, её заполняет
                                   -- rosseti_wfs_loader._map_load_index. Историю см. в
                                   -- git log этого файла.
                                   --
                                   -- ВАЖНО: в этом комментарии НЕЛЬЗЯ упоминать
                                   -- бинд-параметры в синтаксисе «двоеточие + имя».
                                   -- SQLAlchemy text() парсит бинды и внутри
                                   -- SQL-комментариев: упоминание снятого параметра
                                   -- «(двоеточие)load_pct» в тексте комментария
                                   -- сделало его ОБЯЗАТЕЛЬНЫМ, все 71 UPDATE падали
                                   -- с 18.08 по 02.09, а per-row except глотал это
                                   -- как «битую строку» — задача оставалась зелёной.
                                   capacity_source        = 'eesk_35_220',
                                   reserve_asof           = :asof
                             WHERE sc_name_norm = :name_norm
                               AND dzo_name ILIKE '%ЕЭСК%'
                        """),
                        params,
                    )
                if res.rowcount and res.rowcount > 0:
                    matched += res.rowcount
                else:
                    unmatched += 1
                    if len(unmatched_names) < 50:
                        unmatched_names.append(str(sc_name).strip())
            except Exception as e:
                logger.warning("eesk ps_35_220 update failed for %r: %s", sc_name, e)
                unmatched += 1

        db.commit()
    except Exception as e:
        db.rollback()
        logger.exception("load_ps_35_220: outer tx rolled back: %s", e)
        raise
    finally:
        wb.close()

    if unmatched_names:
        logger.warning(
            "eesk ps_35_220 unmatched (%d) sc_name список: %s",
            unmatched,
            unmatched_names,
        )
    logger.info(
        "load_ps_35_220 done: rows=%d matched=%d unmatched=%d",
        rows_seen,
        matched,
        unmatched,
    )
    return {"rows": rows_seen, "matched": matched, "unmatched": unmatched}


def load_tp_rp_sub35(db: Session, xlsx_bytes: bytes, reserve_asof: date | None) -> dict[str, int]:
    """UPSERT резервов ЕЭСК ТП/РП «ниже 35» кВ в power_tp_rp_reserves из xlsx.

    ВЕРИФИЦИРОВАННАЯ структура (лист ``Публикация``): строка 1 — title, строка 3 —
    шапка, ДАННЫЕ С СТРОКИ 4 (~2014 строк). Колонки 0-based: A=номер ТП/РП
    («10003»), B=предельно допустимая загрузка МВт, C=загрузка по 4 замерам МВт,
    D=текущий резерв МВт.

    UPSERT: name=«ТП/РП <A>», name_norm=str(A).strip().lower(),
    municipality='Екатеринбург', voltage_kv=NULL, installed=B, current=C,
    reserve=D, reserve_asof. ON CONFLICT (name_norm, municipality, voltage_kv) DO
    UPDATE. БЕЗ кВА-санитайза. Per-row SAVEPOINT.

    Returns: счётчики rows/inserted/updated/skipped.
    """
    wb = load_workbook(io.BytesIO(xlsx_bytes), read_only=True, data_only=True)
    ws = wb["Публикация"] if "Публикация" in wb.sheetnames else wb.active

    rows_seen = 0
    inserted = 0
    updated = 0
    skipped = 0

    try:
        # min_row=4 (1-based) → данные с 4-й строки; 1 title, 3 шапка выше.
        for row in ws.iter_rows(min_row=4, values_only=True):
            num = _cell(row, 0)  # A
            if num is None or not str(num).strip():
                continue
            # Номер ТП/РП бывает float из Excel (10003.0) — нормализуем к целому виду.
            num_raw = num
            if isinstance(num, float) and num.is_integer():
                num_raw = int(num)
            num_str = str(num_raw).strip()
            rows_seen += 1

            installed = parse_reserve_number(_cell(row, 1))  # B
            current = parse_reserve_number(_cell(row, 2))  # C
            reserve = parse_reserve_number(_cell(row, 3))  # D

            params = {
                "name": f"ТП/РП {num_str}",
                "name_norm": num_str.lower(),
                "municipality": "Екатеринбург",
                "installed": installed,
                "current": current,
                "reserve": reserve,
                "asof": reserve_asof,
            }

            try:
                with db.begin_nested():  # SAVEPOINT — битая строка не валит батч
                    result = db.execute(
                        # voltage_kv намеренно NULL: у ЕЭСК-серии «ниже 35» отдельного
                        # класса напряжения нет; settlement=municipality (одна колонка).
                        text("""
                            INSERT INTO power_tp_rp_reserves
                                (name, name_norm, municipality, settlement,
                                 voltage_kv, installed_capacity_mva, current_load_mva,
                                 reserve_mva, reserve_asof, fetched_at)
                            VALUES (
                                :name, :name_norm, :municipality, :municipality,
                                NULL, :installed, :current,
                                :reserve, :asof, NOW()
                            )
                            ON CONFLICT (name_norm, municipality, voltage_kv) DO UPDATE
                               SET name                   = EXCLUDED.name,
                                   installed_capacity_mva = EXCLUDED.installed_capacity_mva,
                                   current_load_mva       = EXCLUDED.current_load_mva,
                                   reserve_mva            = EXCLUDED.reserve_mva,
                                   reserve_asof           = EXCLUDED.reserve_asof,
                                   fetched_at             = NOW()
                            RETURNING (xmax = 0) AS is_insert
                        """),
                        params,
                    ).scalar()
                if result:
                    inserted += 1
                else:
                    updated += 1
            except Exception as e:
                logger.warning("eesk tp_rp_sub35 upsert failed for %r: %s", num_str, e)
                skipped += 1

        db.commit()
    except Exception as e:
        db.rollback()
        logger.exception("load_tp_rp_sub35: outer tx rolled back: %s", e)
        raise
    finally:
        wb.close()

    result_dict = {
        "rows": rows_seen,
        "inserted": inserted,
        "updated": updated,
        "skipped": skipped,
    }
    logger.info("load_tp_rp_sub35 done: %s", result_dict)
    return result_dict


def load_eesk_reserves(db: Session | None = None) -> dict[str, dict]:
    """Полный прогон ЕЭСК: discovery ссылок + скачивание + загрузка ПС и ТП/РП.

    RUN-ON-PROD. Owns-session как соседи (rosseti/gas/vodokanal). Каждый шаг
    graceful: сбой одного файла не валит другой.
    Returns: {"ps": {...}, "tp_rp": {...}} со счётчиками (или {"error": ...}).
    """
    owns_session = db is None
    if db is None:
        db = SessionLocal()

    out: dict[str, dict] = {}
    try:
        links = fetch_hub_links()
        with httpx.Client(timeout=_HTTP_TIMEOUT, follow_redirects=True) as client:
            ps = links.get("ps_35_220")
            if ps:
                try:
                    r = client.get(ps["url"])
                    r.raise_for_status()
                    asof = quarter_last_day(ps["year"], ps["quarter"])
                    out["ps"] = load_ps_35_220(db, r.content, asof)
                    out["ps"]["period"] = quarter_period(ps["year"], ps["quarter"])
                except Exception as e:
                    logger.exception("load_eesk_reserves: ps file failed: %s", e)
                    out["ps"] = {"error": str(e)}
            else:
                out["ps"] = {"error": "no ps_35_220 link found"}

            tp = links.get("tp_rp_sub35")
            if tp:
                try:
                    r = client.get(tp["url"])
                    r.raise_for_status()
                    asof = quarter_last_day(tp["year"], tp["quarter"])
                    out["tp_rp"] = load_tp_rp_sub35(db, r.content, asof)
                    out["tp_rp"]["period"] = quarter_period(tp["year"], tp["quarter"])
                except Exception as e:
                    logger.exception("load_eesk_reserves: tp_rp file failed: %s", e)
                    out["tp_rp"] = {"error": str(e)}
            else:
                out["tp_rp"] = {"error": "no tp_rp_sub35 link found"}
    finally:
        if owns_session:
            db.close()

    logger.info("load_eesk_reserves done: %s", out)
    return out