Раис Юсупалиев
151 lines
4.7 KiB
Python
151 lines
4.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 = "",
|
|
service_guid: str | None = None,
|
|
) -> list[DeliveryPrice]:
|
|
"""Map a parsed ``Calc`` ``return`` Element into unified delivery prices."""
|
|
|
|
prices: list[DeliveryPrice] = []
|
|
for tariff in _iter_tariffs(root):
|
|
if service_guid is not None and tariff.value != service_guid:
|
|
continue
|
|
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 tariff matching ``"<DeliveryType>|<ServiceGuid>|<Urgency>"``."""
|
|
|
|
delivery_type, tariff_guid, urgency = _split_tariff_code(tariff_code)
|
|
for tariff in _iter_tariffs(root):
|
|
if _tariff_matches(tariff, tariff_guid, urgency):
|
|
price = _map_tariff(tariff, delivery_type, delivery_type_label)
|
|
if price is not None:
|
|
return price
|
|
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 _tariff_matches(tariff: Element, tariff_guid: str, urgency: str) -> bool:
|
|
if _tariff_urgency(tariff) != urgency:
|
|
return False
|
|
return tariff.value == tariff_guid
|
|
|
|
|
|
def _split_tariff_code(tariff_code: str) -> tuple[str, str, str]:
|
|
parts = tariff_code.split("|")
|
|
if len(parts) != 3 or not parts[1] or not parts[2]:
|
|
raise CSEMappingError("CSE tariff_code has invalid format.")
|
|
return parts[0], parts[1], parts[2]
|
|
|
|
|
|
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:
|
|
if _is_additional_service_tariff(tariff):
|
|
return None
|
|
|
|
urgency = _tariff_urgency(tariff)
|
|
if not urgency:
|
|
return None
|
|
if not tariff.value:
|
|
return None
|
|
# Unified tariff_code carries the delivery scheme, CSE service GUID and
|
|
# urgency so payment validation can recalculate the exact selected service.
|
|
tariff_code = f"{delivery_type}|{tariff.value}|{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 _is_additional_service_tariff(tariff: Element) -> bool:
|
|
value = tariff.field_value("AdditionalService")
|
|
if value is None:
|
|
return False
|
|
return value.strip().lower() == "true"
|
|
|
|
|
|
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
|