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

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

Источник: https://okn-mk.mkrf.ru/Maps/searchMap
Метод: POST, Content-Type: application/x-www-form-urlencoded
Аутентификация: не требуется (открытый endpoint).

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

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

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

from __future__ import annotations

import logging
import urllib.parse
from typing import Any

import httpx

logger = logging.getLogger(__name__)

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

# Таймаут на один HTTP-запрос к ЕГРКН (секунды)
_REQUEST_TIMEOUT = 30.0

# 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",
}


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 (сырые свойства из ЕГРКН, часть полей требует авторизации).
        Пустой список при ошибке или пустом ответе.

    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)
    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,
) -> list[dict[str, Any]]:
    """Выполнить HTTP POST к searchMap. При SSL-ошибке с verify=True повторяет с verify=False."""
    # 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)
    try:
        with httpx.Client(timeout=_REQUEST_TIMEOUT, verify=verify, headers=_HEADERS) as client:
            resp = client.post(
                _SEARCHMAP_URL,
                content=body,
            )
            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)
        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


__all__ = ["fetch_okn_points"]