Раис Юсупалиев
107 lines
3.3 KiB
Python
107 lines
3.3 KiB
Python
"""CDEK order registration payload mappers."""
|
|
|
|
from typing import Any
|
|
|
|
from app.schemas.payment import (
|
|
InitPaymentRequest,
|
|
PaymentPackage,
|
|
PaymentParty,
|
|
)
|
|
|
|
|
|
class CDEKOrderMappingError(ValueError):
|
|
"""Raised when CDEK order payload cannot be mapped."""
|
|
|
|
|
|
def map_cdek_order_request(request: InitPaymentRequest) -> dict[str, Any]:
|
|
payload: dict[str, Any] = {
|
|
"number": request.order_uuid,
|
|
"type": request.type,
|
|
"tariff_code": request.tariff_code,
|
|
"sender": _map_order_party(request.sender),
|
|
"recipient": _map_order_party(request.recipient),
|
|
"from_location": request.from_location.model_dump(mode="python"),
|
|
"to_location": request.to_location.model_dump(mode="python"),
|
|
"packages": [_map_order_package(package) for package in request.packages],
|
|
}
|
|
if request.comment is not None:
|
|
payload["comment"] = request.comment
|
|
if request.services is not None:
|
|
payload["services"] = [
|
|
service.model_dump(mode="python") for service in request.services
|
|
]
|
|
return payload
|
|
|
|
|
|
def map_cdek_order_response(payload: dict[str, Any]) -> str:
|
|
entity = payload.get("entity")
|
|
if not isinstance(entity, dict):
|
|
raise CDEKOrderMappingError("CDEK order response must include entity object.")
|
|
|
|
order_uuid = entity.get("uuid")
|
|
if not isinstance(order_uuid, str) or not order_uuid:
|
|
raise CDEKOrderMappingError("CDEK order response must include entity.uuid.")
|
|
|
|
return order_uuid
|
|
|
|
|
|
_CDEK_DUPLICATE_ORDER_ERROR_CODES = frozenset({"v2_entity_already_exists"})
|
|
|
|
|
|
def map_cdek_existing_order_response(payload: object) -> str | None:
|
|
if not isinstance(payload, dict):
|
|
return None
|
|
|
|
if not _contains_cdek_duplicate_error_code(payload):
|
|
return None
|
|
|
|
try:
|
|
return map_cdek_order_response(payload)
|
|
except CDEKOrderMappingError:
|
|
return _find_first_uuid(payload)
|
|
|
|
|
|
def _map_order_party(party: PaymentParty) -> dict[str, Any]:
|
|
return {
|
|
"name": party.name,
|
|
"email": party.email,
|
|
"phones": [party.phone.model_dump(mode="python")],
|
|
}
|
|
|
|
|
|
def _map_order_package(package: PaymentPackage) -> dict[str, Any]:
|
|
payload = package.model_dump(mode="python", exclude_none=True)
|
|
payload["weight"] = package.weight * 1000
|
|
return payload
|
|
|
|
|
|
def _contains_cdek_duplicate_error_code(payload: object) -> bool:
|
|
if isinstance(payload, dict):
|
|
code = payload.get("code")
|
|
if isinstance(code, str) and code in _CDEK_DUPLICATE_ORDER_ERROR_CODES:
|
|
return True
|
|
return any(
|
|
_contains_cdek_duplicate_error_code(value)
|
|
for value in payload.values()
|
|
)
|
|
if isinstance(payload, list):
|
|
return any(_contains_cdek_duplicate_error_code(item) for item in payload)
|
|
return False
|
|
|
|
|
|
def _find_first_uuid(payload: object) -> str | None:
|
|
if isinstance(payload, dict):
|
|
value = payload.get("uuid")
|
|
if isinstance(value, str) and value:
|
|
return value
|
|
for nested_value in payload.values():
|
|
nested_uuid = _find_first_uuid(nested_value)
|
|
if nested_uuid is not None:
|
|
return nested_uuid
|
|
if isinstance(payload, list):
|
|
for item in payload:
|
|
nested_uuid = _find_first_uuid(item)
|
|
if nested_uuid is not None:
|
|
return nested_uuid
|
|
return None
|