"""CSE Calc and SaveDocuments payload mappers.""" from dataclasses import dataclass from app.adapters.delivery_providers.cse.constants import ( resolve_cse_geography, resolve_type_of_cargo, ) from app.adapters.delivery_providers.cse.errors import CSEMappingError from app.adapters.delivery_providers.cse.soap import Element, make_field from app.schemas.payment import Address, InitPaymentRequest from app.schemas.request import DeliveryCalculationRequest _DATETIME_FORMAT = "%Y-%m-%dT%H:%M:%S" @dataclass(frozen=True) class CSEOrderRegistrationParams: """Contract-specific required parameters for SaveDocuments. ``Urgency`` is not here: it comes from the tariff the client selected (``system_data.tariff.tariff_code`` holds the urgency GUID). """ payer: str payment_method: str shipping_method: str @dataclass(frozen=True) class CSEOrderRegistrationResult: order_number: str def build_calc_body_for_calculation( request: DeliveryCalculationRequest, delivery_type: str = "", ) -> dict[str, Element]: fields = [ make_field("SenderGeography", resolve_cse_geography(request.from_city)), make_field("RecipientGeography", resolve_cse_geography(request.to_city)), make_field("TypeOfCargo", resolve_type_of_cargo(_parcel_type(request))), make_field("Weight", _format_decimal(request.weight_kg), "float"), make_field("Qty", "1", "int"), ] _append_delivery_type(fields, delivery_type) return _calc_body(Element(key="Destination", fields=fields)) def build_calc_body_for_payment( request: InitPaymentRequest, delivery_type: str = "", ) -> dict[str, Element]: system_data = request.system_data fields = [ make_field( "SenderGeography", resolve_cse_geography(request.sender_address.city_id), ), make_field( "RecipientGeography", resolve_cse_geography(request.receiver_address.city_id), ), make_field("TypeOfCargo", resolve_type_of_cargo(system_data.parcel_type)), make_field("Weight", system_data.weight, "float"), make_field("Qty", "1", "int"), ] _append_delivery_type(fields, delivery_type) return _calc_body(Element(key="Destination", fields=fields)) def _append_delivery_type(fields: list[Element], delivery_type: str) -> None: if delivery_type: fields.append(make_field("DeliveryType", delivery_type)) def split_tariff_code(tariff_code: str) -> tuple[str, str]: """Split the CSE ``tariff_code`` into (delivery_type, urgency). The unified tariff_code encodes both dimensions as ``"|"``. """ delivery_type, _, urgency = tariff_code.partition("|") return delivery_type, urgency def _calc_body(destination: Element) -> dict[str, Element]: return { "data": Element(key="Destinations", items=[destination]), "parameters": Element( key="parameters", items=[make_field("ipaddress", "0.0.0.0")], ), } def map_cse_save_order_request( request: InitPaymentRequest, order_uuid: str, params: CSEOrderRegistrationParams, ) -> dict[str, Element]: system_data = request.system_data take_date = request.pickup_date.strftime(_DATETIME_FORMAT) delivery_type, urgency = split_tariff_code(system_data.tariff.tariff_code) fields = [ make_field("TakeDate", take_date, "dateTime"), make_field("TakeDateOf", take_date, "dateTime"), make_field("Sender", request.sender_contact.full_name), make_field( "SenderGeography", resolve_cse_geography(request.sender_address.city_id), ), make_field("SenderAddress", _compose_address(request.sender_address)), make_field("SenderPhone", request.sender_contact.phone), make_field("Recipient", request.receiver_contact.full_name), make_field( "RecipientGeography", resolve_cse_geography(request.receiver_address.city_id), ), make_field("RecipientAddress", _compose_address(request.receiver_address)), make_field("RecipientPhone", request.receiver_contact.phone), make_field("Urgency", urgency), make_field("Payer", params.payer, "float"), make_field("PaymentMethod", params.payment_method, "float"), make_field("ShippingMethod", params.shipping_method), make_field("TypeOfCargo", resolve_type_of_cargo(system_data.parcel_type)), make_field("Weight", system_data.weight, "float"), make_field("CargoPackageQty", "1", "float"), ] if delivery_type: fields.append(make_field("DeliveryOfCargo", delivery_type)) if request.delivery_date is not None: fields.append( make_field( "DeliveryDate", request.delivery_date.strftime(_DATETIME_FORMAT), "dateTime", ) ) if request.content.description: fields.append(make_field("CargoDescription", request.content.description)) fields.append(make_field("Comment", request.content.description)) if request.sender_contact.email: fields.append(make_field("SenderEMail", request.sender_contact.email)) if request.receiver_contact.email: fields.append(make_field("RecipientEMail", request.receiver_contact.email)) if system_data.parcel_type == "parcel" and system_data.dimensions is not None: dimensions = system_data.dimensions fields.append(make_field("Length", dimensions.length, "float")) fields.append(make_field("Width", dimensions.width, "float")) fields.append(make_field("Height", dimensions.height, "float")) order = Element( key="Order", fields=fields, properties=[make_field("ClientNumber", order_uuid)], ) return { "data": Element(key="Documents", items=[order]), "parameters": Element( key="parameters", items=[make_field("DocumentType", "Order")], ), } def map_cse_save_order_response(root: Element) -> CSEOrderRegistrationResult: for document in root.items: number = document.property_value("Number") if number: return CSEOrderRegistrationResult(order_number=number) raise CSEMappingError("CSE SaveDocuments response is missing document Number.") def _parcel_type(request: DeliveryCalculationRequest) -> str | None: return request.parcel_type.value if request.parcel_type is not None else None def _compose_address(address: Address) -> str: parts = [address.city, address.street, address.house] line = ", ".join(part for part in parts if part) if address.apartment: line = f"{line}, кв. {address.apartment}" return line def _format_decimal(value: float) -> str: return format(value, "g")