gendesign/backend/app/services/scrapers/okn_egrkn_client.py

"""Клиент АИС ЕГРКН Минкультуры — поиск ОКН-объектов на карте + detail-card (#1066, #1159).

Источник:
- list: ``POST https://okn-mk.mkrf.ru/Maps/searchMap`` (Content-Type: form-urlencoded).
- detail: ``GET  https://okn-mk.mkrf.ru/Show/Show?id={source_id}`` (HTML-карточка).

Аутентификация: не требуется (открытый endpoint).

Ответ searchMap: GeoJSON FeatureCollection.
Каждая фича: {id, geometry: {type: "Point", coordinates: [lat, lon]}, properties: {id}}.

ВНИМАНИЕ: ЕГРКН использует Яндекс-порядок координат [lat, lon], НЕ [lon, lat].
Для PostGIS: ST_MakePoint(coordinates[1], coordinates[0]) → (lon, lat).

searchMap отдаёт в properties **только {id}** — название/адрес/датировка приходят
отдельным detail-endpoint'ом ``Show/Show?id=`` (HTML-карточка ОКН с метаданными).
Парсим название/адрес/категорию охраны через BeautifulSoup, ошибки detail-парсинга
не валят прогон (raw_props сохраняется без detail-полей, баг #1159 acceptance 2).

TLS: mkrf.ru использует сертификат, верификация по умолчанию включена.
При SSLError автоматический fallback на verify=False (аналогично ekburg_permits #242).

WAF-hygiene (#2445 item D3): sibling-скрапер без backoff словил бан IP по WAF
DOM.РФ 24.05.2026 (#2443) — okn-mk.mkrf.ru имеет тот же risk shape (гос. CMS,
десятки последовательных запросов без паузы). Идиома пэйсинга/backoff/ретраев
портирована из ``gisogd66.py`` (``_RETRY_BACKOFF_S`` + сон между попытками) и
``objective.py`` (явная обработка 429/Retry-After): rate-limit между запросами
(``_wait_rate_limit``), экспоненциальный backoff на 429/5xx с капом
(``_do_request``/``_do_detail_request``), уважение заголовка Retry-After.
"""

from __future__ import annotations

import logging
import re
import time
import urllib.parse
from typing import Any

import httpx
from bs4 import BeautifulSoup

logger = logging.getLogger(__name__)

_SEARCHMAP_URL = "https://okn-mk.mkrf.ru/Maps/searchMap"
_DETAIL_URL = "https://okn-mk.mkrf.ru/Show/Show"

# Таймаут на один HTTP-запрос к ЕГРКН (секунды).
# Detail-вызов короче (10 s) — карточки лёгкие, длинный timeout растягивал бы прогон
# на сотни секунд при ~700 объектах.
_REQUEST_TIMEOUT = 30.0
_DETAIL_TIMEOUT = 10.0

# ── WAF-hygiene: пэйсинг + backoff (#2445 item D3) ────────────────────────────
# Вежливая пауза между последовательными запросами (мс) — тот же порядок величины,
# что и у objective.py (_DEFAULT_RATE_MS=500) / nspd_client.py (DEFAULT_RATE_MS=600).
_RATE_LIMIT_MS = 400
# Ретраи на 429/5xx (helper делает суммарно до 1 + _MAX_RETRIES попыток).
_MAX_RETRIES = 3
# Экспоненциальный backoff на 429/5xx: base * 2**attempt, капается _MAX_BACKOFF_S
# (аналогично objective.py: 30 * 2**attempt, cap 300s — здесь эндпоинт легче,
# поэтому база/потолок меньше, но идиома та же).
_RETRY_BACKOFF_BASE_S = 2.0
_MAX_BACKOFF_S = 60.0

# process-локальная метка времени последнего запроса (обеих эндпоинтов) —
# rate-limit общий, чтобы 33 list-запроса и до 1500 detail-запросов не долбили
# WAF без пауз даже при чередовании вызовов.
_last_request_ts: float = 0.0


def _wait_rate_limit() -> None:
    """Выдержать паузу ``_RATE_LIMIT_MS`` с последнего запроса к okn-mk.mkrf.ru.

    Общий для searchMap и detail — обе цели один и тот же WAF-хост.
    """
    global _last_request_ts
    elapsed_ms = (time.time() - _last_request_ts) * 1000.0
    if elapsed_ms < _RATE_LIMIT_MS:
        time.sleep((_RATE_LIMIT_MS - elapsed_ms) / 1000.0)
    _last_request_ts = time.time()


def _backoff_seconds(attempt: int, retry_after_hdr: str | None) -> float:
    """Секунды ожидания перед повтором после 429/5xx.

    Retry-After (если валиден) уважается первым, но капается _MAX_BACKOFF_S —
    аналогично objective.py. Иначе экспоненциальный backoff от attempt (0-based).
    """
    if retry_after_hdr and retry_after_hdr.isdigit():
        return min(float(retry_after_hdr), _MAX_BACKOFF_S)
    return min(_RETRY_BACKOFF_BASE_S * (2**attempt), _MAX_BACKOFF_S)


# Browser headers -- required: without User-Agent the server returns 403.
# Content-Type включён явно: при content= (raw bytes/str) httpx не выставляет его
# автоматически — нужно прописать вручную. Конфликта с data= нет, т.к. data= не используется.
_HEADERS: dict[str, str] = {
    "User-Agent": (
        "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 "
        "(KHTML, like Gecko) Chrome/125.0.0.0 Safari/537.36"
    ),
    "X-Requested-With": "XMLHttpRequest",
    "Referer": "https://okn-mk.mkrf.ru/Maps",
    "Origin": "https://okn-mk.mkrf.ru",
    "Content-Type": "application/x-www-form-urlencoded; charset=UTF-8",
}

# Detail-карточка отдаёт HTML, не JSON — Content-Type form-urlencoded убираем,
# X-Requested-With оставляем для совместимости (на случай SSR-условного рендера).
_DETAIL_HEADERS: dict[str, str] = {
    "User-Agent": _HEADERS["User-Agent"],
    "Referer": "https://okn-mk.mkrf.ru/Maps",
    "Accept": "text/html,application/xhtml+xml",
    "Accept-Language": "ru-RU,ru;q=0.9",
}


def _build_form_data(address: str, category_type: int) -> list[tuple[str, str]]:
    """Построить form-data в формате jQuery serializeArray для поиска по карте.

    Формат: повторяющиеся пары data[i][name] / data[i][value].
    httpx сохраняет порядок и дублирующиеся имена при передаче списком кортежей.
    """
    return [
        ("data[0][name]", "CultureObjects[cob_address]"),
        ("data[0][value]", address),
        ("data[1][name]", "CultureObjects[cob_category_type]"),
        ("data[1][value]", str(category_type)),
    ]


def fetch_okn_points(address: str, category_type: int) -> list[dict[str, Any]]:
    """Запросить список ОКН-объектов из ЕГРКН по адресу и категории.

    Args:
        address: строка поиска (например «Екатеринбург» или «Свердловская»).
        category_type: 1 — федеральный, 2 — региональный, 4 — местного значения.

    Returns:
        Список feature-dict'ов из GeoJSON FeatureCollection.
        Каждый dict содержит поля: id, geometry (с coordinates=[lat, lon]),
        properties (обычно {id} — название/адрес/датировку нужно догружать
        через ``fetch_okn_detail``).
        Пустой список при ошибке или пустом ответе.

    Note:
        ЕГРКН возвращает coordinates в порядке [lat, lon] (Яндекс-конвенция).
        Для PostGIS-вставки: lon=coordinates[1], lat=coordinates[0].
    """
    form_data = _build_form_data(address, category_type)
    features = _do_request(form_data, verify=True, attempt=0)
    return features


def _is_ssl_error(exc: Exception) -> bool:
    """Проверить, является ли исключение SSL-ошибкой.

    httpx >= 0.28 не экспортирует отдельный SSLError — SSL-ошибки приходят
    как httpx.ConnectError с SSL-сообщением в тексте или через цепочку __cause__.
    """
    msg = str(exc).lower()
    ssl_keywords = ("ssl", "certificate", "cert verify", "tls", "handshake")
    if any(kw in msg for kw in ssl_keywords):
        return True
    cause = getattr(exc, "__cause__", None) or getattr(exc, "__context__", None)
    if cause is not None:
        return any(kw in str(cause).lower() for kw in ssl_keywords)
    return False


def _do_request(
    form_data: list[tuple[str, str]],
    *,
    verify: bool,
    attempt: int = 0,
) -> list[dict[str, Any]]:
    """Выполнить HTTP POST к searchMap. При SSL-ошибке с verify=True повторяет с verify=False.

    WAF-hygiene: rate-limit перед каждой попыткой (``_wait_rate_limit``), на 429/5xx —
    экспоненциальный backoff и retry до ``_MAX_RETRIES`` раз (аналогично objective.py).
    """
    # urllib.parse.urlencode гарантированно кодирует список кортежей с повторяющимися ключами
    # (data[0][name]=...&data[0][value]=...&data[1][name]=...). Использование data= с list[tuple]
    # конфликтует с ручным Content-Type → httpx не urlencode-ит, бросает
    # «expected bytes, tuple found». content= передаёт тело как raw str.
    body = urllib.parse.urlencode(form_data)
    _wait_rate_limit()
    try:
        with httpx.Client(timeout=_REQUEST_TIMEOUT, verify=verify, headers=_HEADERS) as client:
            resp = client.post(
                _SEARCHMAP_URL,
                content=body,
            )
            if resp.status_code == 429 or resp.status_code >= 500:
                if attempt < _MAX_RETRIES:
                    wait = _backoff_seconds(attempt, resp.headers.get("Retry-After"))
                    logger.warning(
                        "okn_egrkn_client: HTTP %d от ЕГРКН (попытка %d/%d), retry через %.1fs",
                        resp.status_code,
                        attempt + 1,
                        _MAX_RETRIES,
                        wait,
                    )
                    time.sleep(wait)
                    return _do_request(form_data, verify=verify, attempt=attempt + 1)
            resp.raise_for_status()
    except httpx.HTTPStatusError as exc:
        logger.error("okn_egrkn_client: HTTP %d от ЕГРКН: %s", exc.response.status_code, exc)
        return []
    except httpx.ConnectError as exc:
        if verify and _is_ssl_error(exc):
            logger.warning(
                "okn_egrkn_client: SSL-ошибка при verify=True, повторяем с verify=False: %s", exc
            )
            return _do_request(form_data, verify=False, attempt=attempt)
        logger.error("okn_egrkn_client: ошибка подключения к ЕГРКН: %s", exc)
        return []
    except Exception as exc:
        logger.error("okn_egrkn_client: ошибка запроса к ЕГРКН: %s", exc)
        return []

    try:
        payload: dict[str, Any] = resp.json()
    except Exception as exc:
        logger.error("okn_egrkn_client: не удалось распарсить JSON-ответ ЕГРКН: %s", exc)
        return []

    features: list[dict[str, Any]] = payload.get("features") or []
    if not isinstance(features, list):
        logger.warning("okn_egrkn_client: неожиданный тип features: %s", type(features).__name__)
        return []

    logger.info("okn_egrkn_client: получено %d фич (verify=%s)", len(features), verify)
    return features


# ── Detail-карточка ──────────────────────────────────────────────────────────


# Поля карточки ЕГРКН: «Наименование», «Адрес», «Категория историко-культурного значения»,
# «Вид объекта», «Датировка», «Документ о постановке на гос. охрану», «Регистрационный номер».
# HTML-структура — labelled rows, label слева, value справа в одном tr/td-блоке.
# Используем устойчивые keys-маркеры (lowercased substring match) для тегов <dt>/<th>/<label>.
_FIELD_MARKERS: dict[str, list[str]] = {
    "name": ["наименование"],
    "address": ["адрес"],
    "protection_category": ["категория", "историко-культурного значения"],
    "object_kind": ["вид объекта"],
    "dating": ["датировка"],
    "protection_doc": ["документ о постановке"],
    "regnumber": ["регистрационный номер"],
}

# Лёгкая нормализация пробелов в извлечённом тексте.
_WS_RE = re.compile(r"\s+")


def _norm_text(s: str | None) -> str | None:
    """Сжать whitespace и trim. Пустую строку → None."""
    if not s:
        return None
    value = _WS_RE.sub(" ", s).strip()
    return value or None


def _match_marker(label: str, markers: list[str]) -> bool:
    """Все маркеры присутствуют в lowercased label (substring AND)."""
    lab = label.lower()
    return all(m in lab for m in markers)


def _extract_label_value_pairs(soup: BeautifulSoup) -> dict[str, str]:
    """Извлечь пары (label → value) из карточки ОКН.

    HTML ЕГРКН построен на стандартных Bootstrap-«row»'ах: первая колонка — label,
    вторая — value. Также встречаются <dl><dt>/<dd>. Парсер собирает пары по обоим
    шаблонам; пары накапливаются в порядке появления.

    Возвращает {raw_label_lower: value}, без интерпретации.
    """
    pairs: dict[str, str] = {}

    # 1) <dl><dt>...</dt><dd>...</dd>
    for dt in soup.find_all("dt"):
        dd = dt.find_next_sibling("dd")
        if dd is None:
            continue
        label = _norm_text(dt.get_text(" "))
        value = _norm_text(dd.get_text(" "))
        if label and value:
            pairs.setdefault(label.lower(), value)

    # 2) Bootstrap-row / table-row: пара дочерних элементов в одном контейнере.
    for row in soup.find_all(["tr", "div"]):
        kids = [c for c in row.find_all(recursive=False) if c.name in {"td", "th", "div", "span"}]
        if len(kids) < 2:
            continue
        label = _norm_text(kids[0].get_text(" "))
        value = _norm_text(kids[1].get_text(" "))
        if label and value and len(label) < 120:
            pairs.setdefault(label.lower(), value)

    return pairs


def _parse_detail_html(html: str) -> dict[str, Any]:
    """Распарсить HTML-карточку ОКН в {name, address, …}.

    Невалидный HTML или отсутствие распознанных полей → пустой dict. Не raise.
    """
    if not html or not html.strip():
        return {}
    try:
        soup = BeautifulSoup(html, "html.parser")
    except Exception as exc:
        logger.warning("okn_egrkn_client: BS4 parse failed: %s", exc)
        return {}

    pairs = _extract_label_value_pairs(soup)
    if not pairs:
        return {}

    out: dict[str, Any] = {}
    for field, markers in _FIELD_MARKERS.items():
        for label, value in pairs.items():
            if _match_marker(label, markers):
                out[field] = value
                break
    return out


def fetch_okn_detail(source_id: str) -> dict[str, Any]:
    """Запросить detail-карточку ОКН и распарсить метаданные.

    Args:
        source_id: id фичи из searchMap (он же id записи в ЕГРКН-реестре).

    Returns:
        dict с подмножеством {name, address, protection_category, object_kind,
        dating, protection_doc, regnumber}. Пустой dict при ошибке (любая)
        — never raise, ingest продолжается с базовыми полями.
    """
    if not source_id:
        return {}
    return _do_detail_request(source_id, verify=True, attempt=0)


def _do_detail_request(source_id: str, *, verify: bool, attempt: int = 0) -> dict[str, Any]:
    """HTTP GET карточки + парсинг. SSL-fallback аналогичен searchMap.

    WAF-hygiene: rate-limit перед каждой попыткой (общий с searchMap — один WAF-хост),
    на 429/5xx — экспоненциальный backoff и retry до ``_MAX_RETRIES`` раз.
    """
    params = {"id": source_id}
    _wait_rate_limit()
    try:
        with httpx.Client(
            timeout=_DETAIL_TIMEOUT, verify=verify, headers=_DETAIL_HEADERS
        ) as client:
            resp = client.get(_DETAIL_URL, params=params)
            if resp.status_code == 429 or resp.status_code >= 500:
                if attempt < _MAX_RETRIES:
                    wait = _backoff_seconds(attempt, resp.headers.get("Retry-After"))
                    logger.warning(
                        "okn_egrkn_client: detail HTTP %d source_id=%s "
                        "(попытка %d/%d), retry через %.1fs",
                        resp.status_code,
                        source_id,
                        attempt + 1,
                        _MAX_RETRIES,
                        wait,
                    )
                    time.sleep(wait)
                    return _do_detail_request(source_id, verify=verify, attempt=attempt + 1)
            resp.raise_for_status()
    except httpx.HTTPStatusError as exc:
        logger.warning(
            "okn_egrkn_client: detail HTTP %d source_id=%s",
            exc.response.status_code,
            source_id,
        )
        return {}
    except httpx.ConnectError as exc:
        if verify and _is_ssl_error(exc):
            logger.warning(
                "okn_egrkn_client: SSL-ошибка detail с verify=True, retry verify=False: %s", exc
            )
            return _do_detail_request(source_id, verify=False, attempt=attempt)
        logger.warning("okn_egrkn_client: detail connect error source_id=%s: %s", source_id, exc)
        return {}
    except Exception as exc:
        logger.warning("okn_egrkn_client: detail request error source_id=%s: %s", source_id, exc)
        return {}

    return _parse_detail_html(resp.text)


__all__ = ["fetch_okn_detail", "fetch_okn_points"]