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

"""SOAP (cargo3) Element serialization and parsing for CSE.

CSE exposes a SOAP/1C web service ("Карго") that transfers data through nested
universal ``Element`` structures (``Key``/``Value``/``ValueType``/``Fields``/
``List``/``Tables``/``Properties``). This module builds request envelopes and
parses responses into a plain Python representation that mappers can navigate.
"""

from __future__ import annotations

from dataclasses import dataclass, field
from typing import Any
from xml.etree import ElementTree as ET

CARGO_NS = "http://www.cargo3.ru"
SOAP_NS = "http://www.w3.org/2003/05/soap-envelope"

# Child tags of an Element that hold nested Element lists.
_LIST_TAGS = ("Fields", "List", "Tables", "Properties")
_SCALAR_TAGS = ("Key", "Value", "ValueType")
_BINARY_TAG = "BData"


@dataclass
class Element:
    """In-memory representation of a cargo3 ``Element`` structure."""

    key: str | None = None
    value: str | None = None
    value_type: str | None = None
    bdata: str | None = None
    fields: list["Element"] = field(default_factory=list)
    items: list["Element"] = field(default_factory=list)  # <List>
    tables: list["Element"] = field(default_factory=list)
    properties: list["Element"] = field(default_factory=list)

    def field_value(self, key: str) -> str | None:
        """Return the ``Value`` of the ``Fields`` entry with the given ``Key``."""

        return _find_value(self.fields, key)

    def property_value(self, key: str) -> str | None:
        """Return the ``Value`` of the ``Properties`` entry with the given ``Key``."""

        return _find_value(self.properties, key)


def _find_value(elements: list["Element"], key: str) -> str | None:
    for element in elements:
        if element.key == key:
            return element.value
    return None


def make_field(key: str, value: Any, value_type: str = "string") -> Element:
    """Build a leaf ``Element`` for a ``Fields`` entry."""

    return Element(key=key, value=_stringify(value), value_type=value_type)


def _stringify(value: Any) -> str:
    if isinstance(value, bool):
        return "true" if value else "false"
    return str(value)


def build_envelope(
    operation: str,
    *,
    login: str,
    password: str,
    body: dict[str, Element],
) -> bytes:
    """Build a SOAP 1.2 request envelope for the given CSE operation.

    ``body`` maps the operation parameter tag (e.g. ``data``/``parameters``) to
    the root ``Element`` placed under that tag.
    """

    envelope = ET.Element(f"{{{SOAP_NS}}}Envelope")
    soap_body = ET.SubElement(envelope, f"{{{SOAP_NS}}}Body")
    op_node = ET.SubElement(soap_body, f"{{{CARGO_NS}}}{operation}")

    login_node = ET.SubElement(op_node, f"{{{CARGO_NS}}}login")
    login_node.text = login
    password_node = ET.SubElement(op_node, f"{{{CARGO_NS}}}password")
    password_node.text = password

    for tag, element in body.items():
        _append_element(op_node, tag, element)

    ET.register_namespace("soap", SOAP_NS)
    ET.register_namespace("m", CARGO_NS)
    return ET.tostring(envelope, encoding="utf-8", xml_declaration=True)


def _append_element(parent: ET.Element, tag: str, element: Element) -> None:
    node = ET.SubElement(parent, f"{{{CARGO_NS}}}{tag}")
    if element.key is not None:
        _scalar(node, "Key", element.key)
    if element.value is not None:
        _scalar(node, "Value", element.value)
    if element.value_type is not None:
        _scalar(node, "ValueType", element.value_type)
    for child in element.fields:
        _append_element(node, "Fields", child)
    for child in element.items:
        _append_element(node, "List", child)
    for child in element.tables:
        _append_element(node, "Tables", child)
    for child in element.properties:
        _append_element(node, "Properties", child)


def _scalar(parent: ET.Element, tag: str, text: str) -> None:
    child = ET.SubElement(parent, f"{{{CARGO_NS}}}{tag}")
    child.text = text


def parse_response(xml_payload: str | bytes, operation: str) -> Element:
    """Parse a CSE SOAP response and return the ``return`` Element.

    Raises ``ValueError`` when the envelope is malformed or the expected
    ``<operation>Response`` / ``return`` nodes are missing.
    """

    try:
        root = ET.fromstring(xml_payload)
    except ET.ParseError as exc:
        raise ValueError("CSE response is not valid XML.") from exc

    response_tag = f"{{{CARGO_NS}}}{operation}Response"
    response_node = _find_descendant(root, response_tag)
    if response_node is None:
        raise ValueError(f"CSE response is missing <{operation}Response>.")

    return_node = _find_child(response_node, "return")
    if return_node is None:
        raise ValueError("CSE response is missing <return>.")

    return _parse_element(return_node)


def _parse_element(node: ET.Element) -> Element:
    element = Element()
    for child in node:
        local = _localname(child.tag)
        if local in _SCALAR_TAGS:
            text = (child.text or "").strip()
            if local == "Key":
                element.key = text
            elif local == "Value":
                element.value = text
            else:
                element.value_type = text
        elif local == "Fields":
            element.fields.append(_parse_element(child))
        elif local == "List":
            element.items.append(_parse_element(child))
        elif local == "Tables":
            element.tables.append(_parse_element(child))
        elif local == "Properties":
            element.properties.append(_parse_element(child))
        elif local == _BINARY_TAG:
            element.bdata = (child.text or "").strip()
    return element


def _find_descendant(root: ET.Element, tag: str) -> ET.Element | None:
    if root.tag == tag:
        return root
    return root.find(f".//{tag}")


def _find_child(node: ET.Element, local: str) -> ET.Element | None:
    for child in node:
        if _localname(child.tag) == local:
            return child
    return None


def _localname(tag: str) -> str:
    return tag.rsplit("}", 1)[-1]