g2s-aggregator/app/adapters/delivery_providers/cse/client.py

"""CSE SOAP client and provider adapter."""

import asyncio
from collections.abc import Awaitable, Callable

import httpx
import structlog

from app.adapters.delivery_providers.base import DeliveryProvider
from app.adapters.delivery_providers.cse.constants import (
    CSE_PROVIDER_NAME,
    delivery_type_requires_pvz,
)
from app.adapters.delivery_providers.cse.errors import (
    CSEClientError,
    CSEMappingError,
    CSERequestError,
)
from app.adapters.delivery_providers.cse.mapper import (
    map_cse_calc_response,
    map_cse_calc_response_for_tariff_code,
)
from app.adapters.delivery_providers.cse.order_mapper import (
    CSEOrderInfo,
    CSEOrderRegistrationParams,
    CSEOrderRegistrationResult,
    build_calc_body_for_calculation,
    build_calc_body_for_payment,
    build_tracking_body_for_order,
    build_waybill_print_form_body,
    map_cse_tracking_response,
    map_cse_save_order_request,
    map_cse_save_order_response,
    map_cse_waybill_print_form_response,
    split_tariff_code,
)
from app.adapters.delivery_providers.cse.soap import (
    Element,
    build_envelope,
    make_field,
    parse_response,
)
from app.config import CSEDeliveryProviderConfig
from app.schemas.payment import InitPaymentRequest
from app.schemas.request import DeliveryCalculationRequest
from app.schemas.response import DeliveryPrice

log = structlog.get_logger(__name__)

_SOAP_HEADERS = {"Content-Type": "application/soap+xml; charset=utf-8"}


class CSEClient:
    def __init__(
        self,
        http_client: httpx.AsyncClient,
        *,
        base_url: str,
        login: str,
        password: str,
        registration_params: CSEOrderRegistrationParams,
        timeout_seconds: float = 10.0,
        retry_attempts: int = 2,
        retry_backoff_seconds: float = 0.2,
        sleep: Callable[[float], Awaitable[None]] = asyncio.sleep,
    ) -> None:
        self._http_client = http_client
        self._url = base_url
        self._login = login
        self._password = password
        self._registration_params = registration_params
        self._timeout_seconds = timeout_seconds
        self._retry_attempts = retry_attempts
        self._retry_backoff_seconds = retry_backoff_seconds
        self._sleep = sleep

    async def calc(self, body: dict[str, Element]) -> Element:
        return await self._post(
            "Calc",
            body,
            request_error_message="CSE calc request was rejected with status",
        )

    async def get_delivery_types(self) -> list[tuple[str, str]]:
        """Return available delivery schemes as (name, human label) pairs."""

        body = {
            "parameters": Element(
                key="parameters",
                items=[make_field("Reference", "DeliveryType")],
            )
        }
        root = await self._post(
            "GetReferenceData",
            body,
            request_error_message="CSE delivery types request was rejected with status",
        )
        result: list[tuple[str, str]] = []
        for entry in root.items:
            name = entry.value
            if name:
                result.append((name, entry.field_value("Information") or name))
        return result

    async def save_order(
        self, request: InitPaymentRequest, order_uuid: str
    ) -> CSEOrderRegistrationResult:
        body = map_cse_save_order_request(
            request, order_uuid, self._registration_params
        )
        root = await self._post(
            "SaveDocuments",
            body,
            request_error_message="CSE SaveDocuments request was rejected with status",
        )
        try:
            return map_cse_save_order_response(root)
        except CSEMappingError as exc:
            raise CSEClientError(str(exc)) from exc

    async def get_order(self, order_number: str) -> CSEOrderInfo:
        root = await self._post(
            "Tracking",
            build_tracking_body_for_order(order_number),
            request_error_message="CSE Tracking request was rejected with status",
        )
        try:
            return map_cse_tracking_response(root)
        except CSEMappingError as exc:
            raise CSEClientError(str(exc)) from exc

    async def download_waybill_pdf(self, waybill_number: str) -> bytes:
        root = await self._post(
            "GetFormsForDocuments",
            build_waybill_print_form_body(waybill_number),
            request_error_message=(
                "CSE GetFormsForDocuments request was rejected with status"
            ),
        )
        try:
            return map_cse_waybill_print_form_response(root)
        except CSEMappingError as exc:
            raise CSEClientError(str(exc)) from exc

    async def _post(
        self,
        operation: str,
        body: dict[str, Element],
        *,
        request_error_message: str,
    ) -> Element:
        envelope = build_envelope(
            operation,
            login=self._login,
            password=self._password,
            body=body,
        )
        for attempt in range(self._retry_attempts + 1):
            try:
                response = await self._http_client.post(
                    self._url,
                    content=envelope,
                    headers=_SOAP_HEADERS,
                    timeout=self._timeout_seconds,
                )
            except (httpx.TimeoutException, httpx.TransportError) as exc:
                if attempt < self._retry_attempts:
                    await self._sleep(self._retry_delay(attempt))
                    continue
                raise CSEClientError(
                    f"CSE {operation} request failed after retry attempts."
                ) from exc

            if self._should_retry(response.status_code):
                if attempt < self._retry_attempts:
                    await self._sleep(self._retry_delay(attempt))
                    continue
                log.warning(
                    "cse_request_server_error",
                    operation=operation,
                    status_code=response.status_code,
                    response_excerpt=_response_excerpt(response.text),
                )
                raise CSEClientError(
                    f"CSE {operation} request failed with status "
                    f"{response.status_code}."
                )

            if 400 <= response.status_code < 500:
                log.warning(
                    "cse_request_rejected",
                    operation=operation,
                    status_code=response.status_code,
                )
                raise CSERequestError(
                    f"{request_error_message} {response.status_code}."
                )

            try:
                root = parse_response(response.text, operation)
            except ValueError as exc:
                raise CSEClientError(
                    f"CSE {operation} response payload is invalid."
                ) from exc

            self._raise_for_response_error(root, operation, request_error_message)
            return root

        raise CSEClientError(f"CSE {operation} request failed unexpectedly.")

    @staticmethod
    def _raise_for_response_error(
        root: Element,
        operation: str,
        request_error_message: str,
    ) -> None:
        error_codes = _response_error_codes(root)
        if error_codes:
            log.warning(
                "cse_response_error",
                operation=operation,
                error_codes=error_codes,
            )
            raise CSERequestError(
                f"{request_error_message} application error {error_codes}."
            )

    @staticmethod
    def _should_retry(status_code: int) -> bool:
        return status_code == 429 or 500 <= status_code < 600

    def _retry_delay(self, attempt: int) -> float:
        return self._retry_backoff_seconds * (2**attempt)


def _response_excerpt(text: str, *, limit: int = 1000) -> str:
    return " ".join(text.split())[:limit]


def _response_error_codes(root: Element) -> list[str]:
    codes: list[str] = []
    _collect_response_error_codes(root, codes)
    return codes


def _collect_response_error_codes(element: Element, codes: list[str]) -> None:
    for prop in element.properties:
        if prop.key == "Error":
            codes.extend(item.value for item in prop.items if item.value)
    for child in (*element.items, *element.tables):
        _collect_response_error_codes(child, codes)


class CSEProvider(DeliveryProvider):
    name = CSE_PROVIDER_NAME

    def __init__(
        self,
        client: CSEClient,
        *,
        cache_ttl_seconds: int = 900,
        delivery_service_guids: tuple[str, ...] = (),
    ) -> None:
        self._client = client
        self.cache_ttl_seconds = cache_ttl_seconds
        self._delivery_service_guids = delivery_service_guids
        self._delivery_types: list[tuple[str, str]] | None = None

    @classmethod
    def from_config(
        cls,
        *,
        http_client: httpx.AsyncClient,
        config: CSEDeliveryProviderConfig,
    ) -> "CSEProvider":
        client = CSEClient(
            http_client=http_client,
            base_url=config.base_url,
            login=config.login,
            password=config.password,
            registration_params=CSEOrderRegistrationParams(
                payer=config.payer,
                payment_method=config.payment_method,
                shipping_method=config.shipping_method,
            ),
            timeout_seconds=config.timeout_seconds,
            retry_attempts=config.retry_attempts,
            retry_backoff_seconds=config.retry_backoff_seconds,
        )
        return cls(
            client=client,
            cache_ttl_seconds=config.cache_ttl_seconds,
            delivery_service_guids=tuple(config.delivery_service_guids),
        )

    async def get_prices(
        self, request: DeliveryCalculationRequest
    ) -> list[DeliveryPrice]:
        delivery_types = await self._resolve_delivery_types()

        results = await asyncio.gather(
            *(
                self._prices_for_delivery_type(request, name, label, service_guid)
                for name, label in delivery_types
                for service_guid in self._service_guids_for_price_calculation()
            ),
            return_exceptions=True,
        )

        prices: list[DeliveryPrice] = []
        errors: list[BaseException] = []
        for result in results:
            if isinstance(result, BaseException):
                errors.append(result)
            else:
                prices.extend(result)

        if not prices and errors:
            raise errors[0]
        return prices

    async def _prices_for_delivery_type(
        self,
        request: DeliveryCalculationRequest,
        delivery_type: str,
        delivery_type_label: str,
        service_guid: str | None,
    ) -> list[DeliveryPrice]:
        body = build_calc_body_for_calculation(request, delivery_type, service_guid)
        root = await self._client.calc(body)
        try:
            return map_cse_calc_response(
                root,
                delivery_type=delivery_type,
                delivery_type_label=delivery_type_label,
                service_guid=service_guid,
            )
        except CSEMappingError as exc:
            raise CSEClientError("CSE calc response payload is invalid.") from exc

    def _service_guids_for_price_calculation(self) -> tuple[str, ...]:
        if not self._delivery_service_guids:
            raise CSERequestError("CSE delivery service GUIDs are not configured.")
        return self._delivery_service_guids

    async def _resolve_delivery_types(self) -> list[tuple[str, str]]:
        if self._delivery_types is None:
            self._delivery_types = await self._client.get_delivery_types()
        # Exclude schemes that require a pickup point (PVZ) at registration,
        # since PVZ is not yet supported (only door-to-door is fulfillable).
        usable = [
            (name, label)
            for name, label in self._delivery_types
            if not delivery_type_requires_pvz(name, label)
        ]
        # Fall back to the contract-default scheme (empty delivery_type) so
        # calculation still works if no usable schemes are returned.
        return usable or [("", "")]

    async def get_payment_price(
        self,
        request: InitPaymentRequest,
    ) -> DeliveryPrice | None:
        tariff_code = request.system_data.tariff.tariff_code
        try:
            delivery_type, service_guid, _ = split_tariff_code(tariff_code)
            body = build_calc_body_for_payment(request, delivery_type, service_guid)
            root = await self._client.calc(body)
            return map_cse_calc_response_for_tariff_code(
                root,
                tariff_code=tariff_code,
            )
        except CSERequestError:
            raise
        except CSEMappingError as exc:
            raise CSEClientError(
                "CSE payment price validation response payload is invalid."
            ) from exc

    async def register_order(
        self, request: InitPaymentRequest, order_uuid: str
    ) -> CSEOrderRegistrationResult:
        return await self._client.save_order(request, order_uuid)

    async def get_order(self, order_number: str) -> CSEOrderInfo:
        return await self._client.get_order(order_number)

    async def download_waybill_pdf(self, waybill_number: str) -> bytes:
        return await self._client.download_waybill_pdf(waybill_number)