Раис Юсупалиев
121 lines
3.7 KiB
Python
121 lines
3.7 KiB
Python
"""CSE Calc response mapper to the unified delivery schema."""
|
|
|
|
from decimal import Decimal, InvalidOperation
|
|
|
|
from app.adapters.delivery_providers.cse.constants import (
|
|
CSE_PROVIDER_NAME,
|
|
normalize_currency,
|
|
)
|
|
from app.adapters.delivery_providers.cse.errors import CSEMappingError
|
|
from app.adapters.delivery_providers.cse.soap import Element
|
|
from app.schemas.response import DeliveryPrice
|
|
|
|
_TARIFF_KEY = "Tariff"
|
|
|
|
|
|
def map_cse_calc_response(
|
|
root: Element,
|
|
*,
|
|
delivery_type: str = "",
|
|
delivery_type_label: str = "",
|
|
) -> list[DeliveryPrice]:
|
|
"""Map a parsed ``Calc`` ``return`` Element into unified delivery prices."""
|
|
|
|
prices: list[DeliveryPrice] = []
|
|
for tariff in _iter_tariffs(root):
|
|
price = _map_tariff(tariff, delivery_type, delivery_type_label)
|
|
if price is not None:
|
|
prices.append(price)
|
|
return prices
|
|
|
|
|
|
def map_cse_calc_response_for_tariff_code(
|
|
root: Element,
|
|
*,
|
|
tariff_code: str,
|
|
delivery_type_label: str = "",
|
|
) -> DeliveryPrice | None:
|
|
"""Return the tariff matching ``tariff_code`` (``"<DeliveryType>|<Urgency>"``)."""
|
|
|
|
delivery_type, _, urgency = tariff_code.partition("|")
|
|
for tariff in _iter_tariffs(root):
|
|
if _tariff_urgency(tariff) == urgency:
|
|
return _map_tariff(tariff, delivery_type, delivery_type_label)
|
|
return None
|
|
|
|
|
|
def _tariff_urgency(tariff: Element) -> str | None:
|
|
"""Urgency GUID of a Calc tariff (fallback: the tariff record GUID)."""
|
|
|
|
return tariff.field_value("Urgency") or tariff.value
|
|
|
|
|
|
def _iter_tariffs(root: Element):
|
|
for destination in root.items:
|
|
for tariff in destination.items:
|
|
if tariff.key == _TARIFF_KEY:
|
|
yield tariff
|
|
|
|
|
|
def _map_tariff(
|
|
tariff: Element,
|
|
delivery_type: str,
|
|
delivery_type_label: str,
|
|
) -> DeliveryPrice | None:
|
|
urgency = _tariff_urgency(tariff)
|
|
if not urgency:
|
|
return None
|
|
# Unified tariff_code carries both the delivery scheme and the urgency so
|
|
# registration (SaveDocuments) can set DeliveryOfCargo and Urgency.
|
|
tariff_code = f"{delivery_type}|{urgency}"
|
|
|
|
raw_total = tariff.field_value("Total")
|
|
if raw_total is None:
|
|
raise CSEMappingError("CSE tariff is missing Total.")
|
|
|
|
try:
|
|
price = Decimal(raw_total)
|
|
except (InvalidOperation, TypeError, ValueError) as exc:
|
|
raise CSEMappingError("CSE tariff Total is not a valid decimal.") from exc
|
|
if not price.is_finite() or price <= 0:
|
|
return None
|
|
|
|
base_name = tariff.field_value("Service") or tariff.field_value("UrgencyName")
|
|
if not base_name:
|
|
base_name = urgency
|
|
service_name = (
|
|
f"{base_name} — {delivery_type_label}" if delivery_type_label else base_name
|
|
)
|
|
|
|
currency = normalize_currency(tariff.field_value("CurrencyName"))
|
|
|
|
min_days = _to_int(tariff.field_value("MinPeriod"))
|
|
max_days = _to_int(tariff.field_value("MaxPeriod"))
|
|
if min_days is None:
|
|
min_days = 0
|
|
if max_days is None or max_days < min_days:
|
|
max_days = min_days
|
|
|
|
try:
|
|
return DeliveryPrice(
|
|
provider=CSE_PROVIDER_NAME,
|
|
service_name=str(service_name),
|
|
price=price,
|
|
currency=currency,
|
|
delivery_days_min=min_days,
|
|
delivery_days_max=max_days,
|
|
tariff_code=tariff_code,
|
|
bypass_parcel_type_filter=True,
|
|
)
|
|
except (ArithmeticError, TypeError, ValueError) as exc:
|
|
raise CSEMappingError("CSE tariff fields have invalid values.") from exc
|
|
|
|
|
|
def _to_int(value: str | None) -> int | None:
|
|
if value is None or value == "":
|
|
return None
|
|
try:
|
|
return int(Decimal(value))
|
|
except (InvalidOperation, TypeError, ValueError):
|
|
return None
|