g2s-aggregator/app/adapters/delivery_providers/cse/order_mapper.py

"""CSE Calc, SaveDocuments and waybill payload mappers."""

import base64
import binascii
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, CSERequestError
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


@dataclass(frozen=True)
class CSEOrderInfo:
    order_number: str
    status_code: str | None
    waybill_number: str | None


def build_calc_body_for_calculation(
    request: DeliveryCalculationRequest,
    delivery_type: str = "",
    service_guid: str | None = None,
) -> 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)
    if service_guid:
        fields.append(make_field("Service", service_guid))
    return _calc_body(Element(key="Destination", fields=fields))


def build_calc_body_for_payment(
    request: InitPaymentRequest,
    delivery_type: str = "",
    service_guid: str | None = None,
) -> 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)
    if service_guid:
        fields.append(make_field("Service", service_guid))
    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, str]:
    """Split the CSE ``tariff_code`` into delivery type, service GUID, urgency.

    The expected format is ``"<DeliveryType>|<ServiceGuid>|<Urgency>"``.
    ``DeliveryType`` may be empty when CSE contract defaults are used.
    """

    parts = tariff_code.split("|")
    if len(parts) != 3 or not parts[1] or not parts[2]:
        raise CSERequestError("CSE tariff_code has invalid format.")
    return parts[0], parts[1], parts[2]


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"),
        make_field("CargoCost", str(request.content.declared_value), "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 build_tracking_body_for_order(order_number: str) -> dict[str, Element]:
    return {
        "documents": Element(
            key="Documents",
            items=[Element(key=order_number)],
        ),
        "parameters": Element(
            key="parameters",
            items=[
                make_field("DocumentType", "Order"),
                make_field("OnlySelectedType", True, "boolean"),
            ],
        ),
    }


def map_cse_tracking_response(root: Element) -> CSEOrderInfo:
    for document in root.items:
        order_number = document.property_value("Number") or document.key
        if not order_number:
            continue
        return CSEOrderInfo(
            order_number=order_number,
            status_code=_latest_tracking_status(document),
            waybill_number=_extract_waybill_number(document),
        )
    raise CSEMappingError("CSE Tracking response is missing document data.")


def build_waybill_print_form_body(waybill_number: str) -> dict[str, Element]:
    return {
        "documents": Element(
            key="Documents",
            items=[Element(key=waybill_number)],
        ),
        "parameters": Element(
            key="parameters",
            items=[
                make_field("DocumentType", "waybill"),
                make_field("Type", "print"),
                make_field(
                    "Name",
                    "Универсальная печатная "
                    "форма документа НАКЛАДНАЯ",
                ),
                make_field("Format", "pdf"),
                make_field("OnlySelectedType", True, "boolean"),
            ],
        ),
    }


def map_cse_waybill_print_form_response(root: Element) -> bytes:
    for document in root.items:
        bdata = document.bdata
        if not bdata:
            continue
        try:
            normalized_bdata = "".join(bdata.split())
            return base64.b64decode(normalized_bdata, validate=True)
        except (binascii.Error, ValueError) as exc:
            raise CSEMappingError(
                "CSE GetFormsForDocuments response contains invalid BData."
            ) from exc
    raise CSEMappingError("CSE GetFormsForDocuments response is missing BData.")


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")


def _latest_tracking_status(document: Element) -> str | None:
    dated: list[tuple[str, str]] = []
    for state in document.items:
        if not state.key:
            continue
        date_time = state.property_value("DateTime") or ""
        dated.append((date_time, state.key))
    if not dated:
        return None
    dated.sort(key=lambda item: item[0])
    return dated[-1][1]


def _extract_waybill_number(document: Element) -> str | None:
    for table in document.tables:
        if table.key != "Waybills":
            continue
        for waybill in table.items:
            document_type = waybill.property_value("DocumentType")
            if document_type is not None and document_type.lower() != "waybill":
                continue
            number = waybill.property_value("Number") or waybill.key
            if number:
                return number
    return None