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

"""Загрузчик резервов свободной мощности Россети Урал из xlsx-раскрытия (#2119).

Два xlsx-файла раскрытия информации ФАС (rosseti-ural.ru):
  (а) «свободная мощность по ЦП»                  → апдейт power_supply_centers
  (б) «свободная мощность для ТП по ПС и РП <35»  → UPSERT power_tp_rp_reserves

Источник ГЕО-БЛОКИРУЕТ non-RU IP → РАБОТАЕТ НА ПРОДЕ (Celery / docker exec).

Структуры xlsx ВЕРИФИЦИРОВАНЫ (см. docstring-и функций): фиксированные offset'ы
строк заголовков/нумерации/данных. openpyxl read_only для памяти на 10k+ строках.
Матч ЦП-резерва к power_supply_centers — по ОБЩЕМУ нормализатору 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_wfs_loader import normalize_sc_name

logger = logging.getLogger(__name__)

# Punycode-host не требуется — rosseti-ural.ru латиница.
DISCLOSURE_URL = "https://rosseti-ural.ru/company/disclosure/monopoly/characteristic/bandwidth/"

_HTTP_TIMEOUT = 60

# «н/д» / «-» / пусто → None (нет данных).
_NODATA_TOKENS = frozenset({"", "н/д", "нд", "n/a", "na", "-", "—", "–", "нет данных"})

# Дата актуальности «на 30.06.2026» из имени файла или заголовка листа.
_ASOF_RE = re.compile(r"на\s+(\d{1,2})[.\-/](\d{1,2})[.\-/](\d{4})", re.IGNORECASE)
# Дата в имени файла для выбора «свежайшего» (DD.MM.YYYY / DD-MM-YYYY / YYYY-MM-DD).
_FILE_DATE_RE = re.compile(
    r"(\d{1,2})[.\-](\d{1,2})[.\-](\d{4})|(\d{4})[.\-](\d{1,2})[.\-](\d{1,2})"
)

# Порог кВА-санитайза: у ТП установленная мощность в МВА физически <2.5;
# значение выше — почти наверняка забито в кВА (÷1000). РП не трогаем.
_TP_KVA_THRESHOLD_MVA = 2.5


def parse_reserve_number(value: object) -> float | None:
    """Парсит числовую ячейку резерва/мощности xlsx → float.

    «н/д»/пусто → None. Запятая-десятичный → точка. Сохраняет знак
    (резервы бывают отрицательными = дефицит). Нечисловой мусор → None.
    """
    if value is None:
        return None
    if isinstance(value, int | float):
        return float(value)
    s = str(value).strip().lower()
    if s in _NODATA_TOKENS:
        return None
    # Убираем пробелы-разделители тысяч (в т.ч. неразрывные) и нормализуем запятую.
    s = s.replace("\xa0", "").replace(" ", "").replace(",", ".")
    # Оставляем ведущий минус + цифры + точку.
    m = re.match(r"^-?\d+(?:\.\d+)?$", s)
    if not m:
        return None
    try:
        return float(s)
    except ValueError:
        return None


def parse_asof_date(*sources: str | None) -> date | None:
    """Ищет дату «на DD.MM.YYYY» в переданных строках (имя файла / заголовок листа).

    Возвращает первую найденную date. None — если нигде нет.
    """
    for src in sources:
        if not src:
            continue
        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 _file_date_key(filename: str) -> tuple[int, int, int]:
    """Ключ сортировки «свежести» по дате в имени файла. Нет даты → (0,0,0)."""
    m = _FILE_DATE_RE.search(filename)
    if not m:
        return (0, 0, 0)
    if m.group(1):  # DD-MM-YYYY
        return (int(m.group(3)), int(m.group(2)), int(m.group(1)))
    return (int(m.group(4)), int(m.group(5)), int(m.group(6)))  # YYYY-MM-DD


def sanitize_tp_capacity_mva(installed: float | None, is_rp: bool) -> tuple[float | None, bool]:
    """кВА-санитайз установленной мощности ТП.

    Часть строк ошибочно забивает кВА в колонку МВА. Если это ТП (не РП) и
    installed > порога (2.5 МВА физически невозможно для ТП) — считаем что это
    кВА и делим на 1000. Возвращает (значение_МВА, был_ли_санитайз).
    РП не трогаем (у РП мощность легитимно бывает высокой).
    """
    if installed is None or is_rp:
        return installed, False
    if installed > _TP_KVA_THRESHOLD_MVA:
        return installed / 1000.0, True
    return installed, False


def fetch_disclosure_links() -> dict[str, str]:
    """Скрейпит страницу раскрытия → ссылки на СВЕЖАЙШИЕ xlsx (ЦП + ТП/РП).

    RUN-ON-PROD (гео-блок). GET с follow_redirects, regex по href .xlsx, фильтр
    Свердловск-related, выбор LATEST по дате-в-имени для двух типов файлов.

    Returns: {"cp": url|'', "tp_rp": url|''} — пустые если не нашлось.
    """
    resp = httpx.get(DISCLOSURE_URL, timeout=_HTTP_TIMEOUT, follow_redirects=True)
    resp.raise_for_status()
    html = resp.text

    # href на .xlsx (относительные и абсолютные).
    hrefs = re.findall(r'href=["\']([^"\']+?\.xlsx)["\']', html, re.IGNORECASE)
    base = "https://rosseti-ural.ru"

    cp_candidates: list[str] = []
    tp_candidates: list[str] = []
    for href in hrefs:
        low = href.lower()
        # Свердловск-related фильтр (файлы бывают по разным филиалам).
        if "свердл" not in low and "sverdl" not in low and "свэс" not in low:
            # Если в href нет региона — пропускаем осторожно (может быть общий файл),
            # но для нашей задачи берём только явно свердловские.
            continue
        url = href if href.startswith("http") else base + href
        if "по цп" in low or "_цп" in low or "cp" in low:
            cp_candidates.append(url)
        if "тп" in low and ("рп" in low or "35" in low):
            tp_candidates.append(url)

    def _latest(cands: list[str]) -> str:
        return max(cands, key=_file_date_key) if cands else ""

    result = {"cp": _latest(cp_candidates), "tp_rp": _latest(tp_candidates)}
    logger.info(
        "rosseti disclosure links: cp=%r tp_rp=%r (from %d hrefs)",
        result["cp"],
        result["tp_rp"],
        len(hrefs),
    )
    return result


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


def load_cp_reserves(db: Session, xlsx_bytes: bytes, filename: str = "") -> dict[str, int]:
    """Апдейт power_supply_centers резервами по ЦП из xlsx-раскрытия.

    ВЕРИФИЦИРОВАННАЯ структура ЦП-файла: заголовки строки 5-6, строка 7 =
    нумерация столбцов, ДАННЫЕ С СТРОКИ 8. Колонки (1-based Excel → 0-based idx):
      B(1)=ЦП name, C(2)=ПО (баланс), E(4)=municipality, F(5)=voltage class,
      G(6)=installed МВА, H(7)=current load МВА, I(8)=reserve МВА.
    Матч к power_supply_centers по sc_name_norm (тот же нормализатор!) →
    installed/current/reserve/capacity_source='cp_35_110'. Unmatched → лог count.

    Returns: счётчики rows/matched/unmatched.
    """
    wb = load_workbook(io.BytesIO(xlsx_bytes), read_only=True, data_only=True)
    ws = wb.active
    asof = parse_asof_date(filename, ws.title if ws is not None else None)

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

    try:
        # min_row=8 → данные с 8-й строки (1-based). iter_rows отдаёт кортежи values.
        for row in ws.iter_rows(min_row=8, values_only=True):
            cp_name = _cell(row, 1)  # B
            if cp_name is None or not str(cp_name).strip():
                continue
            rows_seen += 1

            name_norm = normalize_sc_name(str(cp_name))
            installed = parse_reserve_number(_cell(row, 6))  # G
            current = parse_reserve_number(_cell(row, 7))  # H
            reserve = parse_reserve_number(_cell(row, 8))  # I
            municipality = _cell(row, 4)  # E

            try:
                with db.begin_nested():  # SAVEPOINT — битая строка не валит батч
                    res = db.execute(
                        text("""
                            UPDATE power_supply_centers
                               SET installed_capacity_mva = :installed,
                                   current_load_mva       = :current,
                                   reserve_mva            = :reserve,
                                   reserve_asof           = :asof,
                                   capacity_source        = 'cp_35_110',
                                   municipality = COALESCE(municipality, :municipality)
                             WHERE sc_name_norm = :name_norm
                        """),
                        {
                            "installed": installed,
                            "current": current,
                            "reserve": reserve,
                            "asof": asof,
                            "municipality": (str(municipality).strip() if municipality else None),
                            "name_norm": name_norm,
                        },
                    )
                if res.rowcount and res.rowcount > 0:
                    matched += res.rowcount
                else:
                    unmatched += 1
                    if len(unmatched_names) < 20:
                        unmatched_names.append(str(cp_name).strip())
            except Exception as e:
                logger.warning("cp_reserve update failed for %r: %s", cp_name, e)
                unmatched += 1

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

    logger.info(
        "load_cp_reserves done: rows=%d matched=%d unmatched=%d (sample unmatched=%s)",
        rows_seen,
        matched,
        unmatched,
        unmatched_names[:10],
    )
    return {"rows": rows_seen, "matched": matched, "unmatched": unmatched}


def _is_rp(name: str) -> bool:
    """Строка про РП (распределительный пункт), а не ТП? Влияет на кВА-санитайз."""
    return bool(re.search(r"\bрп\b|рп[\s-]?\d", name.lower()))


def load_tp_rp_reserves(db: Session, xlsx_bytes: bytes, filename: str = "") -> dict[str, int]:
    """UPSERT резервов ТП/РП <35 кВ в power_tp_rp_reserves из xlsx-раскрытия.

    ВЕРИФИЦИРОВАННАЯ структура: заголовки строки 6-7, нумерация строка 8,
    ДАННЫЕ С СТРОКИ 9 (~10 741 строк). Колонки (0-based idx):
      0=name (ТП-721…), 1=РЭС, 2=region, 3=municipality/settlement,
      4=voltage kV, 5=installed МВА, 6=load МВА, 7=reserve МВА.
    кВА-санитайз: у ТП (не РП) installed > 2.5 → делим на 1000 (лог count).
    «н/д» → NULL. UPSERT по UNIQUE(name_norm, municipality, voltage_kv).
    Per-row SAVEPOINT. Returns: счётчики.
    """
    wb = load_workbook(io.BytesIO(xlsx_bytes), read_only=True, data_only=True)
    ws = wb.active
    asof = parse_asof_date(filename, ws.title if ws is not None else None)

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

    try:
        for row in ws.iter_rows(min_row=9, values_only=True):
            name = _cell(row, 0)
            if name is None or not str(name).strip():
                continue
            name_str = str(name).strip()
            rows_seen += 1

            res_unit = _cell(row, 1)
            municipality = _cell(row, 3)
            voltage_kv = _cell(row, 4)
            installed = parse_reserve_number(_cell(row, 5))
            current = parse_reserve_number(_cell(row, 6))
            reserve = parse_reserve_number(_cell(row, 7))

            installed, sanitized = sanitize_tp_capacity_mva(installed, _is_rp(name_str))
            if sanitized:
                kva_sanitized += 1

            params = {
                "name": name_str,
                "name_norm": normalize_sc_name(name_str),
                "res_unit": str(res_unit).strip() if res_unit else None,
                "municipality": str(municipality).strip() if municipality else None,
                "voltage_kv": str(voltage_kv).strip() if voltage_kv else None,
                "installed": installed,
                "current": current,
                "reserve": reserve,
                "asof": asof,
            }

            try:
                with db.begin_nested():  # SAVEPOINT — битая строка не валит батч
                    result = db.execute(
                        # settlement намеренно = municipality: в xlsx Россетей это ОДНА
                        # колонка «МО/населённый пункт»; отдельного settlement источник
                        # не даёт (колонка в схеме — задел под Фазу B/геокод).
                        text("""
                            INSERT INTO power_tp_rp_reserves
                                (name, name_norm, res_unit, municipality, settlement,
                                 voltage_kv, installed_capacity_mva, current_load_mva,
                                 reserve_mva, reserve_asof, fetched_at)
                            VALUES (
                                :name, :name_norm, :res_unit, :municipality, :municipality,
                                :voltage_kv, :installed, :current,
                                :reserve, :asof, NOW()
                            )
                            ON CONFLICT (name_norm, municipality, voltage_kv) DO UPDATE
                               SET name                   = EXCLUDED.name,
                                   res_unit               = EXCLUDED.res_unit,
                                   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("tp_rp_reserve upsert failed for %r: %s", name_str, e)
                skipped += 1

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

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


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

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

    out: dict[str, dict] = {}
    try:
        links = fetch_disclosure_links()
        with httpx.Client(timeout=_HTTP_TIMEOUT, follow_redirects=True) as client:
            if links.get("cp"):
                try:
                    r = client.get(links["cp"])
                    r.raise_for_status()
                    out["cp"] = load_cp_reserves(db, r.content, links["cp"])
                except Exception as e:
                    logger.exception("load_all_reserves: cp file failed: %s", e)
                    out["cp"] = {"error": str(e)}
            else:
                out["cp"] = {"error": "no cp link found"}

            if links.get("tp_rp"):
                try:
                    r = client.get(links["tp_rp"])
                    r.raise_for_status()
                    out["tp_rp"] = load_tp_rp_reserves(db, r.content, links["tp_rp"])
                except Exception as e:
                    logger.exception("load_all_reserves: tp_rp file failed: %s", e)
                    out["tp_rp"] = {"error": str(e)}
            else:
                out["tp_rp"] = {"error": "no tp_rp link found"}
    finally:
        if owns_session:
            db.close()

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