sap-cli-skill/assets/sapcli/client/_transport.py

"""TransportMixin — transport request management."""

from __future__ import annotations

import logging
import re
import xml.etree.ElementTree as ET

from sapcli.exceptions import CreateError, SapCliError

logger = logging.getLogger("sapcli.client")


class TransportMixin:
    """Transport-request CRUD and release."""

    def get_transport_request(self) -> str | None:
        url = f"{self.host}/sap/bc/adt/cts/transportrequests"
        hdrs: dict[str, str] = {
            **self._headers(),
            "Accept": "application/vnd.sap.adt.transportorganizer.v1+xml",
        }
        logger.info("GET TRANSPORT: GET %s", url)
        resp = self.session.get(url, headers=hdrs)
        logger.info("GET TRANSPORT RESPONSE: HTTP %s", resp.status_code)
        resp.raise_for_status()
        root = ET.fromstring(resp.content)
        tm_ns = "http://www.sap.com/cts/adt/tm"
        for req in root.findall(f".//{{{tm_ns}}}request"):
            num = req.attrib.get(f"{{{tm_ns}}}number", "")
            status = req.attrib.get(f"{{{tm_ns}}}status", "")
            if status == "D" and num:
                logger.info("Transport request: %s", num)
                return num
        return None

    def list_transport_requests(self) -> list[dict[str, str]]:
        """列出所有可修改的传输请求。

        Returns:
            列表，每项包含 number, description, owner, status 字段。
        """
        url = f"{self.host}/sap/bc/adt/cts/transportrequests"
        hdrs: dict[str, str] = {
            **self._headers(),
            "Accept": "application/vnd.sap.adt.transportorganizer.v1+xml",
        }
        logger.info("LIST TRANSPORT: GET %s", url)
        resp = self.session.get(url, headers=hdrs)
        logger.info("LIST TRANSPORT RESPONSE: HTTP %s", resp.status_code)
        resp.raise_for_status()
        root = ET.fromstring(resp.content)
        tm_ns = "http://www.sap.com/cts/adt/tm"
        results: list[dict[str, str]] = []
        for req in root.findall(f".//{{{tm_ns}}}request"):
            num = req.attrib.get(f"{{{tm_ns}}}number", "")
            status = req.attrib.get(f"{{{tm_ns}}}status", "")
            desc = req.attrib.get(f"{{{tm_ns}}}description", "")
            owner = req.attrib.get(f"{{{tm_ns}}}owner", "")
            if status == "D" and num:
                results.append(
                    {
                        "number": num,
                        "description": desc,
                        "owner": owner,
                        "status": status,
                    }
                )
                logger.info(
                    "Transport request: %s — %s (owner: %s)", num, desc, owner
                )
        return results

    def create_transport_request(self, description: str) -> str:
        """创建新的传输请求。

        Args:
            description: 传输请求描述。

        Returns:
            新创建的传输请求编号。
        """
        url = f"{self.host}/sap/bc/adt/cts/transportrequests"
        body = (
            '<?xml version="1.0" encoding="utf-8"?>'
            '<cts:request xmlns:cts="http://www.sap.com/adt/cts"'
            f' cts:type="K"'
            f' cts:description="{description}"/>'
        )
        hdrs = self._headers()
        logger.info("CREATE TRANSPORT: POST %s desc=%s", url, description)
        resp = self.session.post(url, headers=hdrs, data=body.encode("utf-8"))
        logger.info("CREATE TRANSPORT RESPONSE: HTTP %s", resp.status_code)
        if resp.status_code >= 400:
            logger.error("CREATE TRANSPORT ERROR: %s", resp.text[:2000])
            raise CreateError(
                f"创建传输请求失败: HTTP {resp.status_code} — {resp.text[:500]}"
            )

        # 从响应中解析新请求编号
        root = ET.fromstring(resp.content)
        tm_ns = "http://www.sap.com/cts/adt/tm"
        for req in root.findall(f".//{{{tm_ns}}}request"):
            num = req.attrib.get(f"{{{tm_ns}}}number", "")
            if num:
                logger.info("New transport request created: %s", num)
                return num

        # 尝试从 Location header 或响应体文本中提取编号
        location = resp.headers.get("Location", "")
        m = re.search(r"transportrequests/(\w+)", location)
        if m:
            logger.info("New transport request from Location: %s", m.group(1))
            return m.group(1)

        logger.warning("无法从响应中解析新传输请求编号")
        return ""

    def transport_info(self, corr_nr: str) -> dict[str, str]:
        """获取传输请求详情。

        Args:
            corr_nr: 传输请求编号。

        Returns:
            字典含 ``number``, ``description``, ``status``, ``owner`` 等。
        """
        url = f"{self.host}/sap/bc/adt/cts/transportrequests/{corr_nr}"
        hdrs: dict[str, str] = {
            **self._headers(),
            "Accept": "application/vnd.sap.adt.transportorganizer.v1+xml",
        }
        logger.info("TRANSPORT INFO: GET %s", url)
        resp = self.session.get(url, headers=hdrs)
        logger.info("TRANSPORT INFO RESPONSE: HTTP %s", resp.status_code)
        resp.raise_for_status()

        root = ET.fromstring(resp.content)
        tm_ns = "http://www.sap.com/cts/adt/tm"
        info: dict[str, str] = {
            "number": corr_nr,
            "description": "",
            "status": "",
            "owner": "",
        }
        for req in root.findall(f".//{{{tm_ns}}}request"):
            info["number"] = req.attrib.get(f"{{{tm_ns}}}number", corr_nr)
            info["description"] = req.attrib.get(f"{{{tm_ns}}}description", "")
            info["status"] = req.attrib.get(f"{{{tm_ns}}}status", "")
            info["owner"] = req.attrib.get(f"{{{tm_ns}}}owner", "")
            break
        return info

    def transport_release(self, corr_nr: str) -> bool:
        """释放传输请求。

        Args:
            corr_nr: 传输请求编号。

        Returns:
            是否成功。
        """
        url = f"{self.host}/sap/bc/adt/cts/transportrequests/{corr_nr}/%20"
        hdrs = self._headers()
        params: dict[str, str] = {"_action": "RELEASE"}
        logger.info("TRANSPORT RELEASE: POST %s corrNr=%s", url, corr_nr)
        resp = self.session.post(url, headers=hdrs, params=params, data="")
        logger.info("TRANSPORT RELEASE RESPONSE: HTTP %s", resp.status_code)
        if resp.status_code >= 400:
            logger.error("TRANSPORT RELEASE ERROR: %s", resp.text[:2000])
            raise SapCliError(
                f"释放传输请求失败: HTTP {resp.status_code} — {resp.text[:500]}"
            )
        return True

    def transport_objects(self, corr_nr: str) -> list[dict[str, str]]:
        """列出传输请求中的对象。

        Args:
            corr_nr: 传输请求编号。

        Returns:
            列表，每项含 ``name``, ``type``。
        """
        url = f"{self.host}/sap/bc/adt/cts/transportrequests/{corr_nr}"
        hdrs: dict[str, str] = {
            **self._headers(),
            "Accept": "application/vnd.sap.adt.transportorganizer.v1+xml",
        }
        params: dict[str, str] = {"withObjects": "true"}
        logger.info("TRANSPORT OBJECTS: GET %s", url)
        resp = self.session.get(url, headers=hdrs, params=params)
        logger.info("TRANSPORT OBJECTS RESPONSE: HTTP %s", resp.status_code)
        resp.raise_for_status()

        root = ET.fromstring(resp.content)
        results: list[dict[str, str]] = []
        core_ns = "http://www.sap.com/adt/core"
        for ref in root.findall(f".//{{{core_ns}}}objectReference"):
            obj_name = ref.attrib.get(f"{{{core_ns}}}name", "")
            obj_type = ref.attrib.get(f"{{{core_ns}}}type", "")
            results.append({"name": obj_name, "type": obj_type})
        return results