g2s-aggregator/app/adapters/address_suggestions/yandex_geosuggest/client.py

"""Yandex Geosuggest address suggestion adapter."""

from typing import Any

import httpx

from app.adapters.address_suggestions.base import (
    AddressSuggestionClientError,
    AddressSuggestionProvider,
    AddressSuggestionRequestError,
)
from app.config import YandexGeosuggestAddressSuggestionsConfig
from app.schemas.request import AddressSuggestRequest
from app.schemas.response import AddressSuggestion

_STREET_KINDS = frozenset({"street"})
_HOUSE_KINDS = frozenset({"house"})
_FLAT_KINDS = frozenset({"apartment", "flat"})


class YandexGeosuggestClientError(AddressSuggestionClientError):
    """Raised when Yandex Geosuggest requests fail."""


class YandexGeosuggestRequestError(
    YandexGeosuggestClientError,
    AddressSuggestionRequestError,
):
    """Raised when Yandex Geosuggest rejects request parameters."""


class YandexGeosuggestAddressSuggestionProvider(AddressSuggestionProvider):
    name = "yandex_geosuggest"

    def __init__(
        self,
        http_client: httpx.AsyncClient,
        *,
        url: str,
        api_key: str,
        timeout_seconds: float = 10.0,
    ) -> None:
        self._http_client = http_client
        self._url = url
        self._api_key = api_key
        self._timeout_seconds = timeout_seconds

    @classmethod
    def from_config(
        cls,
        *,
        http_client: httpx.AsyncClient,
        config: YandexGeosuggestAddressSuggestionsConfig,
    ) -> "YandexGeosuggestAddressSuggestionProvider":
        return cls(
            http_client=http_client,
            url=config.url,
            api_key=config.api_key,
            timeout_seconds=config.timeout_seconds,
        )

    async def suggest(self, request: AddressSuggestRequest) -> list[AddressSuggestion]:
        params = self._build_query_params(request)
        try:
            response = await self._http_client.get(
                self._url,
                params=params,
                timeout=self._timeout_seconds,
            )
        except (httpx.TimeoutException, httpx.TransportError) as exc:
            raise YandexGeosuggestClientError(
                "Yandex Geosuggest request failed."
            ) from exc

        if response.status_code == 400:
            raise YandexGeosuggestRequestError(
                "Yandex Geosuggest request was rejected with status 400."
            )

        if response.status_code == 403 or response.status_code == 429:
            raise YandexGeosuggestClientError(
                "Yandex Geosuggest request failed with status "
                f"{response.status_code}."
            )

        if response.status_code >= 500 or 400 <= response.status_code < 500:
            raise YandexGeosuggestClientError(
                "Yandex Geosuggest request failed with status "
                f"{response.status_code}."
            )

        try:
            response.raise_for_status()
            raw_payload = response.json()
        except (httpx.HTTPError, TypeError, ValueError) as exc:
            raise YandexGeosuggestClientError(
                "Yandex Geosuggest returned invalid payload."
            ) from exc

        try:
            return _map_yandex_geosuggest_response(raw_payload)
        except ValueError as exc:
            raise YandexGeosuggestClientError(
                "Yandex Geosuggest response payload is invalid."
            ) from exc

    def _build_query_params(self, request: AddressSuggestRequest) -> dict[str, Any]:
        params: dict[str, Any] = {
            "apikey": self._api_key,
            "text": f"{request.city.strip()} {request.query.strip()}".strip(),
            "print_address": 1,
        }
        if request.limit is not None:
            params["results"] = request.limit
        return params


def _map_yandex_geosuggest_response(payload: object) -> list[AddressSuggestion]:
    if not isinstance(payload, dict):
        raise ValueError("Yandex Geosuggest response must be a JSON object.")

    raw_results = payload.get("results")
    if raw_results is None:
        return []
    if not isinstance(raw_results, list):
        raise ValueError("Yandex Geosuggest response must include results list.")

    return [_map_yandex_geosuggest_result(item) for item in raw_results]


def _map_yandex_geosuggest_result(payload: object) -> AddressSuggestion:
    if not isinstance(payload, dict):
        raise ValueError("Yandex Geosuggest result entry must be an object.")

    raw_address_payload = payload.get("address")
    if raw_address_payload is not None and not isinstance(raw_address_payload, dict):
        raise ValueError("Yandex Geosuggest address payload must be an object.")

    address_payload = raw_address_payload if isinstance(raw_address_payload, dict) else {}
    raw_components = address_payload.get("component")
    if raw_components is None:
        components: list[object] = []
    elif isinstance(raw_components, list):
        components = raw_components
    else:
        raise ValueError("Yandex Geosuggest address components must be a list.")

    address = _extract_formatted_address(address_payload) or _extract_title_text(payload)
    if address is None:
        raise ValueError("Yandex Geosuggest result has no valid address value.")

    return AddressSuggestion(
        address=address,
        street=_extract_component_value(components, _STREET_KINDS),
        house=_extract_component_value(components, _HOUSE_KINDS),
        flat=_extract_component_value(components, _FLAT_KINDS),
        postal_code=None,
    )


def _extract_formatted_address(payload: dict[str, Any]) -> str | None:
    value = payload.get("formatted_address")
    return _coerce_optional_text(value)


def _extract_title_text(payload: dict[str, Any]) -> str | None:
    raw_title = payload.get("title")
    if raw_title is None:
        return None
    if not isinstance(raw_title, dict):
        raise ValueError("Yandex Geosuggest title payload must be an object.")
    return _coerce_optional_text(raw_title.get("text"))


def _extract_component_value(
    components: list[object],
    expected_kinds: frozenset[str],
) -> str | None:
    for component in components:
        if not isinstance(component, dict):
            raise ValueError(
                "Yandex Geosuggest address component entry must be an object."
            )

        if not _component_has_kind(component, expected_kinds):
            continue

        raw_name = component.get("name")
        if raw_name is None:
            return None
        if not isinstance(raw_name, str):
            raise ValueError(
                "Yandex Geosuggest address component name must be a string."
            )
        return raw_name.strip() or None

    return None


def _component_has_kind(component: dict[str, Any], expected_kinds: frozenset[str]) -> bool:
    raw_kind = component.get("kind")
    if isinstance(raw_kind, str):
        kinds = [raw_kind]
    elif isinstance(raw_kind, list):
        kinds = raw_kind
    else:
        raise ValueError("Yandex Geosuggest address component kind must be provided.")

    normalized_kinds: set[str] = set()
    for kind in kinds:
        if not isinstance(kind, str):
            raise ValueError(
                "Yandex Geosuggest address component kind must be a string."
            )
        normalized_kind = kind.strip().lower()
        if normalized_kind:
            normalized_kinds.add(normalized_kind)

    return not expected_kinds.isdisjoint(normalized_kinds)


def _coerce_optional_text(value: object) -> str | None:
    if value is None:
        return None
    if not isinstance(value, str):
        raise ValueError("Yandex Geosuggest text field must be a string.")
    return value.strip() or None