Раис Юсупалиев
101 lines
3.3 KiB
Python
101 lines
3.3 KiB
Python
"""CDEK response mapper to unified delivery schema."""
|
|
|
|
from decimal import Decimal
|
|
from typing import Any
|
|
|
|
from app.schemas.response import DeliveryPrice
|
|
|
|
|
|
class CDEKMappingError(ValueError):
|
|
"""Raised when CDEK response cannot be mapped."""
|
|
|
|
|
|
def map_cdek_response(payload: dict[str, Any]) -> list[DeliveryPrice]:
|
|
tariffs = _get_tariffs(payload)
|
|
payload_currency = payload.get("currency")
|
|
return [
|
|
_map_tariff(tariff, payload_currency=payload_currency)
|
|
for tariff in tariffs
|
|
]
|
|
|
|
|
|
def map_cdek_response_for_tariff_code(
|
|
payload: dict[str, Any],
|
|
*,
|
|
tariff_code: str,
|
|
) -> DeliveryPrice | None:
|
|
tariff_codes = payload.get("tariff_codes")
|
|
if not isinstance(tariff_codes, list):
|
|
raise CDEKMappingError("CDEK response must include tariff_codes.")
|
|
|
|
payload_currency = payload.get("currency")
|
|
for tariff in tariff_codes:
|
|
if not isinstance(tariff, dict):
|
|
raise CDEKMappingError("CDEK tariff entry must be an object.")
|
|
if _tariff_code_matches(tariff.get("tariff_code"), tariff_code):
|
|
return _map_tariff(tariff, payload_currency=payload_currency)
|
|
return None
|
|
|
|
|
|
def _map_tariff(
|
|
tariff: dict[str, Any],
|
|
*,
|
|
payload_currency: object,
|
|
) -> DeliveryPrice:
|
|
service_name = tariff.get("tariff_name") or tariff.get("tariff_code")
|
|
if service_name is None:
|
|
raise CDEKMappingError("CDEK tariff_name is missing.")
|
|
|
|
raw_price = tariff.get("delivery_sum")
|
|
if raw_price is None:
|
|
raise CDEKMappingError("CDEK delivery_sum is missing.")
|
|
|
|
period_min = tariff.get("period_min")
|
|
if period_min is None:
|
|
raise CDEKMappingError("CDEK period_min is missing.")
|
|
period_max = tariff.get("period_max", period_min)
|
|
|
|
raw_currency = tariff.get("currency") or payload_currency or "RUB"
|
|
|
|
try:
|
|
return DeliveryPrice(
|
|
provider="cdek",
|
|
service_name=str(service_name),
|
|
price=Decimal(str(raw_price)),
|
|
currency=str(raw_currency).upper(),
|
|
delivery_days_min=int(period_min),
|
|
delivery_days_max=int(period_max),
|
|
tariff_code=_extract_tariff_code(tariff.get("tariff_code")),
|
|
bypass_parcel_type_filter=True,
|
|
)
|
|
except (ArithmeticError, TypeError, ValueError) as exc:
|
|
raise CDEKMappingError("CDEK response fields have invalid values.") from exc
|
|
|
|
|
|
def _get_tariffs(payload: dict[str, Any]) -> list[dict[str, Any]]:
|
|
tariff_codes = payload.get("tariff_codes")
|
|
if not isinstance(tariff_codes, list) or not tariff_codes:
|
|
raise CDEKMappingError("CDEK response must include non-empty tariff_codes.")
|
|
tariffs: list[dict[str, Any]] = []
|
|
for tariff in tariff_codes:
|
|
if not isinstance(tariff, dict):
|
|
raise CDEKMappingError("CDEK tariff entry must be an object.")
|
|
tariffs.append(tariff)
|
|
return tariffs
|
|
|
|
|
|
def _tariff_code_matches(value: object, expected_tariff_code: str) -> bool:
|
|
extracted = _extract_tariff_code(value)
|
|
if extracted is None:
|
|
return False
|
|
return extracted == expected_tariff_code
|
|
|
|
|
|
def _extract_tariff_code(value: object) -> str | None:
|
|
if isinstance(value, bool) or value is None:
|
|
return None
|
|
try:
|
|
return str(int(value))
|
|
except (TypeError, ValueError):
|
|
return None
|