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

"""DdicMixin — object CRUD, DDIC helpers, CDS, packages, ATC, pretty-print."""

from __future__ import annotations

import logging
import xml.etree.ElementTree as ET
from typing import Any

from sapcli.exceptions import CreateError, DeleteError
from sapcli.types import ObjectTypeConfig, get_type_config

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


class DdicMixin:
    """Object creation / deletion, DDIC operations, CDS, packages, ATC, pretty-print."""

    # ------------------------------------------------------------------
    #  Object create / delete
    # ------------------------------------------------------------------

    def delete_object(
        self,
        obj_uri: str,
        corr_nr: str | None = None,
    ) -> tuple[bool, str]:
        lock_handle, _ = self.lock(obj_uri, corr_nr)
        url = f"{self.host}{obj_uri}"
        params: dict[str, str] = {"lockHandle": lock_handle}
        if corr_nr:
            params["corrNr"] = corr_nr
        hdrs = self._headers()
        logger.info("DELETE: DELETE %s", url)
        resp = self.session.delete(url, headers=hdrs, params=params)
        logger.info("DELETE RESPONSE: HTTP %s", resp.status_code)
        if resp.status_code in (200, 204):
            logger.info("删除成功")
            return True, ""
        error_text = resp.text[:500] if resp.text else f"HTTP {resp.status_code}"
        logger.error("删除失败: %s", error_text)
        raise DeleteError(f"删除失败: {error_text}")

    def create_object(
        self,
        obj_type: str,
        name: str,
        description: str | None = None,
        corr_nr: str | None = None,
        source: str | None = None,
        package: str = "$TMP",
    ) -> tuple[str, str | None]:
        config = get_type_config(obj_type)
        params: dict[str, str] = {}
        if corr_nr:
            params["corrNr"] = corr_nr

        if obj_type == "function":
            if "/" not in name:
                raise CreateError("function 类型需要'函数组名/函数模块名' 格式")
            group, fm = name.split("/", 1)
            collection_url = config.format_collection_uri(group=group.lower())
            obj_name = fm.upper()
            obj_uri = config.format_obj_uri(fm.lower(), group=group.lower())
            src_uri = config.format_src_uri(fm.lower(), group=group.lower())
        else:
            collection_url = config.collection_uri
            obj_name = name.upper()
            obj_uri = config.format_obj_uri(name.lower())
            src_uri = config.format_src_uri(name.lower())

        body = self._build_create_body(obj_type, obj_name, description, package)
        url = f"{self.host}{collection_url}"
        hdrs = self._headers(config.create_content_type)
        logger.info("CREATE: POST %s name=%s", url, obj_name)
        resp = self.session.post(
            url, headers=hdrs, params=params, data=body.encode("utf-8")
        )
        logger.info("CREATE RESPONSE: HTTP %s", resp.status_code)
        if resp.status_code >= 400:
            logger.error("CREATE ERROR: %s", resp.text[:2000])
            raise CreateError(
                f"创建对象失败: HTTP {resp.status_code} — {resp.text[:500]}"
            )

        logger.info("对象已创建: %s", obj_uri)

        if source and src_uri:
            lock_handle, _ = self.lock(obj_uri, corr_nr)
            try:
                self.set_source(src_uri, source, lock_handle, corr_nr)
            finally:
                self.unlock(obj_uri, lock_handle)
            self.activate(obj_name, obj_uri, corr_nr)

        return obj_uri, src_uri

    def _build_create_body(
        self,
        obj_type: str,
        name: str,
        description: str | None,
        package: str = "$TMP",
    ) -> str:
        desc = description or name
        if obj_type == "report":
            return (
                '<?xml version="1.0" encoding="UTF-8"?>'
                '<program:abapProgram xmlns:program="http://www.sap.com/adt/programs/programs"'
                ' xmlns:adtcore="http://www.sap.com/adt/core"'
                f' adtcore:name="{name}"'
                ' adtcore:type="PROG/P"'
                f' adtcore:description="{desc}">'
                f'<adtcore:packageRef adtcore:name="{package}"/>'
                "</program:abapProgram>"
            )
        elif obj_type == "class":
            return (
                '<?xml version="1.0" encoding="UTF-8"?>'
                '<class:abapClass xmlns:class="http://www.sap.com/adt/oo/classes"'
                ' xmlns:adtcore="http://www.sap.com/adt/core"'
                f' adtcore:name="{name}"'
                ' adtcore:type="CLAS/OC"'
                f' adtcore:description="{desc}">'
                f'<adtcore:packageRef adtcore:name="{package}"/>'
                "</class:abapClass>"
            )
        elif obj_type == "function":
            return (
                '<?xml version="1.0" encoding="UTF-8"?>'
                '<fm:abapFunctionModule xmlns:fm="http://www.sap.com/adt/functions/fmodules"'
                ' xmlns:adtcore="http://www.sap.com/adt/core"'
                f' adtcore:name="{name}"'
                ' adtcore:type="FUNC/F"'
                f' adtcore:description="{desc}">'
                f'<adtcore:packageRef adtcore:name="{package}"/>'
                "</fm:abapFunctionModule>"
            )
        elif obj_type == "functiongroup":
            return (
                '<?xml version="1.0" encoding="UTF-8"?>'
                '<fg:abapFunctionGroup xmlns:fg="http://www.sap.com/adt/functions/groups"'
                ' xmlns:adtcore="http://www.sap.com/adt/core"'
                f' adtcore:name="{name}"'
                ' adtcore:type="FUGR/F"'
                f' adtcore:description="{desc}">'
                f'<adtcore:packageRef adtcore:name="{package}"/>'
                "</fg:abapFunctionGroup>"
            )
        elif obj_type == "interface":
            return (
                '<?xml version="1.0" encoding="UTF-8"?>'
                '<intf:abapInterface xmlns:intf="http://www.sap.com/adt/oo/interfaces"'
                ' xmlns:adtcore="http://www.sap.com/adt/core"'
                f' adtcore:name="{name}"'
                ' adtcore:type="INTF/OI"'
                f' adtcore:description="{desc}">'
                f'<adtcore:packageRef adtcore:name="{package}"/>'
                "</intf:abapInterface>"
            )
        elif obj_type == "domain":
            return (
                '<?xml version="1.0" encoding="UTF-8"?>'
                '<doma:domain xmlns:doma="http://www.sap.com/dictionary/domain"'
                ' xmlns:adtcore="http://www.sap.com/adt/core"'
                f' adtcore:name="{name}"'
                f' adtcore:type="DOMA/DD"'
                f' adtcore:description="{desc}"'
                ' adtcore:language="EN">'
                f'<adtcore:packageRef adtcore:name="{package}"/>'
                "</doma:domain>"
            )
        elif obj_type == "dataelement":
            return (
                '<?xml version="1.0" encoding="UTF-8"?>'
                '<dtel:dataElement xmlns:dtel="http://www.sap.com/adt/dictionary/dataelements"'
                ' xmlns:adtcore="http://www.sap.com/adt/core"'
                f' adtcore:name="{name}"'
                f' adtcore:type="DTEL/DE"'
                f' adtcore:description="{desc}"'
                ' adtcore:language="EN">'
                f'<adtcore:packageRef adtcore:name="{package}"/>'
                "</dtel:dataElement>"
            )
        elif obj_type == "table":
            return (
                '<?xml version="1.0" encoding="UTF-8"?>'
                '<tabl:table xmlns:tabl="http://www.sap.com/adt/dictionary/tables"'
                ' xmlns:adtcore="http://www.sap.com/adt/core"'
                f' adtcore:name="{name}"'
                f' adtcore:type="TABL/TT"'
                f' adtcore:description="{desc}"'
                ' adtcore:language="EN">'
                f'<adtcore:packageRef adtcore:name="{package}"/>'
                "</tabl:table>"
            )
        elif obj_type == "tabletype":
            return (
                '<?xml version="1.0" encoding="UTF-8"?>'
                '<ttyp:abapTableType xmlns:ttyp="http://www.sap.com/adt/ddic/tabletypes"'
                ' xmlns:adtcore="http://www.sap.com/adt/core"'
                f' adtcore:name="{name}"'
                ' adtcore:type="TTYP"'
                f' adtcore:description="{desc}">'
                f'<adtcore:packageRef adtcore:name="{package}"/>'
                "</ttyp:abapTableType>"
            )
        raise ValueError(f"不支持的对象类型: {obj_type}")

    # ------------------------------------------------------------------
    #  DDIC helpers
    # ------------------------------------------------------------------

    def create_ddic_object(
        self,
        obj_type: str,
        name: str,
        definition_body: str,
        corr_nr: str | None = None,
    ) -> tuple[str, str | None]:
        config = get_type_config(obj_type)
        obj_uri = config.format_obj_uri(name.lower())
        src_uri = config.format_src_uri(name.lower())

        if obj_type in ("table", "structure"):
            self._put_ddl_source(src_uri, definition_body, "", corr_nr)
        else:
            obj_name = name.upper()
            body = self._build_create_body(obj_type, obj_name, obj_name)
            collection_url = config.collection_uri
            params: dict[str, str] = {}
            if corr_nr:
                params["corrNr"] = corr_nr
            hdrs = self._headers(config.create_content_type)
            logger.info(
                "CREATE DDIC: POST %s name=%s (with full definition)",
                collection_url,
                obj_name,
            )
            resp = self.session.post(
                f"{self.host}{collection_url}",
                headers=hdrs,
                params=params,
                data=definition_body.encode("utf-8"),
            )
            logger.info("CREATE DDIC RESPONSE: HTTP %s", resp.status_code)
            if resp.status_code >= 400:
                raise CreateError(
                    f"创建 DDIC 对象失败: HTTP {resp.status_code} — {resp.text[:500]}"
                )

        obj_name = name.upper()
        success, messages = self.activate(obj_name, obj_uri, corr_nr)
        if not success:
            errors = [m for m in messages if m["type"] == "E"]
            if errors:
                raise CreateError(
                    f"DDIC 对象激活失败: {'; '.join(e['text'] for e in errors)}"
                )

        logger.info("DDIC 对象已创建并激活: %s", obj_uri)
        return obj_uri, src_uri

    def _put_ddic_xml(
        self,
        obj_uri: str,
        xml_body: str,
        lock_handle: str = "",
        corr_nr: str | None = None,
    ) -> bool:
        url = f"{self.host}{obj_uri}"
        params: dict[str, str] = {}
        if lock_handle:
            params["lockHandle"] = lock_handle
        if corr_nr:
            params["corrNr"] = corr_nr
        hdrs = self._headers()
        hdrs["Accept"] = "*/*"
        logger.info("PUT DDIC XML: PUT %s (%d chars)", url, len(xml_body))
        resp = self.session.put(
            url, headers=hdrs, params=params, data=xml_body.encode("utf-8")
        )
        logger.info("PUT DDIC XML RESPONSE: HTTP %s", resp.status_code)
        if resp.status_code >= 400:
            raise CreateError(
                f"写入 DDIC XML 失败: HTTP {resp.status_code} — {resp.text[:500]}"
            )
        return True

    def _put_ddl_with_auto_lock(
        self,
        src_uri: str,
        ddl_body: str,
        obj_uri: str,
        corr_nr: str | None = None,
    ) -> bool:
        try:
            lock_handle, _ = self.lock(obj_uri, corr_nr)
        except Exception:
            logger.debug("自动锁定失败，使用空锁句柄继续", exc_info=True)
            lock_handle = ""
        try:
            return self.set_source(src_uri, ddl_body, lock_handle, corr_nr)
        finally:
            if lock_handle:
                self.unlock(obj_uri, lock_handle)

    def _put_ddl_source(
        self,
        src_uri: str,
        ddl_body: str,
        lock_handle: str,
        corr_nr: str | None = None,
    ) -> bool:
        return self.set_source(src_uri, ddl_body, lock_handle, corr_nr)

    def get_object_status(self, obj_uri: str) -> dict[str, Any]:
        """查询对象在 SAP 系统中的状态。

        Returns:
            {"exists": bool, "status": str, "corr_nr": str|None}
            status: "active" / "inactive" / "not_exists"
        """
        url = f"{self.host}{obj_uri}"
        hdrs = self._headers()
        hdrs["Accept"] = "*/*"
        logger.info("GET OBJECT STATUS: GET %s", url)
        resp = self.session.get(url, headers=hdrs)
        logger.info("GET OBJECT STATUS RESPONSE: HTTP %s", resp.status_code)

        if resp.status_code == 404:
            return {"exists": False, "status": "not_exists", "corr_nr": None}

        if resp.status_code != 200:
            logger.warning(
                "Unexpected status %d for object status check: %s",
                resp.status_code,
                obj_uri,
            )
            return {"exists": False, "status": "not_exists", "corr_nr": None}

        # 解析 XML 获取 version（active/inactive）
        status = "active"
        corr_nr: str | None = None
        try:
            root = ET.fromstring(resp.content)
            ns = {"adtcore": "http://www.sap.com/adt/core"}
            version = root.attrib.get(f"{{{ns['adtcore']}}}version", "")
            if version == "inactive":
                status = "inactive"
            elif version and version != "active":
                status = version
        except ET.ParseError:
            pass

        # 尝试获取 corr_nr：通过快速 lock → unlock 探测
        try:
            lock_handle, detected_corr = self.lock(obj_uri)
            corr_nr = detected_corr
            self.unlock(obj_uri, lock_handle)
        except Exception:
            # 锁定失败也正常（可能权限问题），corr_nr 保持 None
            logger.debug("探测 corr_nr 失败", exc_info=True)

        return {"exists": True, "status": status, "corr_nr": corr_nr}

    # ------------------------------------------------------------------
    #  Function-group helpers
    # ------------------------------------------------------------------

    def function_group_exists(self, group_name: str) -> bool:
        url = f"{self.host}/sap/bc/adt/functions/groups/{group_name.lower()}"
        hdrs = self._headers()
        logger.info("CHECK FG EXISTS: GET %s", url)
        resp = self.session.get(url, headers=hdrs)
        logger.info("CHECK FG EXISTS RESPONSE: HTTP %s", resp.status_code)
        return resp.status_code == 200

    def create_function_group(
        self,
        group_name: str,
        description: str | None = None,
        corr_nr: str | None = None,
    ) -> bool:
        config = get_type_config("functiongroup")
        params: dict[str, str] = {"groupname": group_name.upper()}
        if corr_nr:
            params["corrNr"] = corr_nr
        desc = description or group_name
        body = self._build_create_body("functiongroup", group_name.upper(), desc)
        url = f"{self.host}{config.collection_uri}"
        hdrs = self._headers(config.create_content_type)
        logger.info("CREATE FG: POST %s name=%s", url, group_name.upper())
        resp = self.session.post(
            url, headers=hdrs, params=params, data=body.encode("utf-8")
        )
        logger.info("CREATE FG RESPONSE: HTTP %s", resp.status_code)
        if resp.status_code >= 400:
            logger.error("CREATE FG ERROR: %s", resp.text[:2000])
            raise CreateError(
                f"创建函数组失败: HTTP {resp.status_code} — {resp.text[:500]}"
            )
        return True

    # ------------------------------------------------------------------
    #  CDS View
    # ------------------------------------------------------------------

    def get_cds_source(self, name: str) -> str:
        """读取 CDS View DDL 源码。

        Args:
            name: CDS 名称。

        Returns:
            DDL 源码字符串。
        """
        url = f"{self.host}/sap/bc/adt/ddic/ddlsources/{name.lower()}/source/main"
        hdrs = self._headers()
        hdrs["Accept"] = "text/plain"
        logger.info("GET CDS SOURCE: GET %s", url)
        resp = self.session.get(url, headers=hdrs)
        logger.info("GET CDS SOURCE RESPONSE: HTTP %s", resp.status_code)
        resp.raise_for_status()
        return resp.text

    def create_cds(
        self,
        name: str,
        description: str,
        ddl_source: str,
    ) -> tuple[str, str]:
        """创建 CDS View 并写入 DDL 源码。

        Args:
            name: CDS 名称。
            description: 描述。
            ddl_source: DDL 源码。

        Returns:
            ``(obj_uri, src_uri)`` 元组。
        """
        obj_uri = f"/sap/bc/adt/ddic/ddlsources/{name.lower()}"
        src_uri = f"/sap/bc/adt/ddic/ddlsources/{name.lower()}/source/main"

        # 1) 创建 CDS 对象
        create_url = f"{self.host}/sap/bc/adt/ddic/ddlsources"
        desc = description or name
        body = (
            '<?xml version="1.0" encoding="UTF-8"?>'
            '<ddlsrc:abapSource xmlns:ddlsrc="http://www.sap.com/adt/ddic/ddlsources"'
            ' xmlns:adtcore="http://www.sap.com/adt/core"'
            f' adtcore:name="{name.lower()}"'
            ' adtcore:type="DDLS/DF"'
            f' adtcore:description="{desc}">'
            '<adtcore:packageRef adtcore:name="$TMP"/>'
            "</ddlsrc:abapSource>"
        )
        hdrs = self._headers("application/xml")
        params: dict[str, str] = {"name": name.lower()}
        logger.info("CREATE CDS: POST %s name=%s", create_url, name)
        resp = self.session.post(
            create_url,
            headers=hdrs,
            params=params,
            data=body.encode("utf-8"),
        )
        logger.info("CREATE CDS RESPONSE: HTTP %s", resp.status_code)
        if resp.status_code >= 400:
            logger.error("CREATE CDS ERROR: %s", resp.text[:2000])
            raise CreateError(
                f"创建 CDS 失败: HTTP {resp.status_code} — {resp.text[:500]}"
            )

        # 2) 写入 DDL 源码
        lock_handle, corr_nr = self.lock(obj_uri)
        try:
            self.set_source(src_uri, ddl_source, lock_handle, corr_nr)
        finally:
            self.unlock(obj_uri, lock_handle)

        return obj_uri, src_uri

    # ------------------------------------------------------------------
    #  Package
    # ------------------------------------------------------------------

    def create_package(
        self,
        name: str,
        description: str | None = None,
        superpackage: str | None = None,
    ) -> bool:
        """创建 ABAP 包。

        Args:
            name: 包名。
            description: 描述。
            superpackage: 上级包名。

        Returns:
            是否成功。
        """
        url = f"{self.host}/sap/bc/adt/packages"
        desc = description or name
        body = (
            '<?xml version="1.0" encoding="UTF-8"?>'
            '<pkg:package xmlns:pkg="http://www.sap.com/adt/packages"'
            ' xmlns:adtcore="http://www.sap.com/adt/core"'
            f' adtcore:name="{name}"'
            f' adtcore:description="{desc}"'
        )
        if superpackage:
            body += f'><adtcore:packageRef adtcore:name="{superpackage}"/></pkg:package>'
        else:
            body += "/>"

        hdrs = self._headers("application/xml")
        logger.info("CREATE PACKAGE: POST %s name=%s", url, name)
        resp = self.session.post(url, headers=hdrs, data=body.encode("utf-8"))
        logger.info("CREATE PACKAGE RESPONSE: HTTP %s", resp.status_code)
        if resp.status_code >= 400:
            logger.error("CREATE PACKAGE ERROR: %s", resp.text[:2000])
            raise CreateError(
                f"创建包失败: HTTP {resp.status_code} — {resp.text[:500]}"
            )
        return True

    def get_package_info(self, name: str) -> dict[str, str]:
        """获取 ABAP 包信息。

        Args:
            name: 包名。

        Returns:
            字典含 ``name``, ``description``, ``owner``, ``superpackage`` 等。
        """
        url = f"{self.host}/sap/bc/adt/packages/{name}"
        hdrs = self._headers()
        hdrs["Accept"] = "application/xml"
        logger.info("GET PACKAGE INFO: GET %s", url)
        resp = self.session.get(url, headers=hdrs)
        logger.info("GET PACKAGE INFO RESPONSE: HTTP %s", resp.status_code)
        resp.raise_for_status()

        root = ET.fromstring(resp.content)
        core_ns = "http://www.sap.com/adt/core"
        info: dict[str, str] = {
            "name": name,
            "description": "",
            "owner": "",
            "superpackage": "",
        }
        # 尝试从根元素属性提取
        info["description"] = root.attrib.get(f"{{{core_ns}}}description", "")
        info["owner"] = root.attrib.get(f"{{{core_ns}}}owner", "")
        # 上级包引用
        pkg_ref = root.find(f".//{{{core_ns}}}packageRef")
        if pkg_ref is not None:
            info["superpackage"] = pkg_ref.attrib.get(f"{{{core_ns}}}name", "")
        return info

    # ------------------------------------------------------------------
    #  ATC / Quality
    # ------------------------------------------------------------------

    def atc_check(
        self,
        name: str,
        obj_uri: str,
        variant: str | None = None,
    ) -> tuple[bool, list[dict[str, str]]]:
        """执行 ATC 代码检查。

        Args:
            name: 对象名称。
            obj_uri: 对象 ADT URI。
            variant: 检查变体名称。

        Returns:
            ``(success, findings)`` — success 表示无严重错误，
            findings 是发现项列表，每项含 ``type``, ``line``, ``text`` 等。
        """
        url = f"{self.host}/sap/bc/adt/atos/checks"
        params: dict[str, str] = {"context": obj_uri}
        if variant:
            params["variant"] = variant

        body = (
            '<?xml version="1.0" encoding="UTF-8"?>'
            '<adtcore:objectReferences xmlns:adtcore="http://www.sap.com/adt/core">'
            f'<adtcore:objectReference adtcore:uri="{obj_uri}" adtcore:name="{name}"/>'
            "</adtcore:objectReferences>"
        )

        hdrs = self._headers()
        logger.info("ATC CHECK: POST %s name=%s", url, name)
        resp = self.session.post(
            url, headers=hdrs, params=params, data=body.encode("utf-8")
        )
        logger.info("ATC CHECK RESPONSE: HTTP %s", resp.status_code)
        resp.raise_for_status()

        findings: list[dict[str, str]] = []
        has_error = False

        root = ET.fromstring(resp.content)
        core_ns = "http://www.sap.com/adt/core"
        for item in root.iter():
            severity = item.attrib.get("severity", item.attrib.get("type", ""))
            line = item.attrib.get("line", "")
            text = item.attrib.get("message", item.text or "")
            if severity or text:
                findings.append({"type": severity, "line": line, "text": text})
                if severity in ("E", "1"):
                    has_error = True

        return not has_error, findings

    # ------------------------------------------------------------------
    #  Pretty printer
    # ------------------------------------------------------------------

    def pretty_print(self, source: str) -> str:
        """调用 ABAP Pretty Printer 格式化源码。

        Args:
            source: 原始 ABAP 源码。

        Returns:
            格式化后的源码。
        """
        url = f"{self.host}/sap/bc/adt/prettyprinter"
        hdrs = self._headers("text/plain; charset=utf-8")
        hdrs["Accept"] = "text/plain"
        logger.info("PRETTY PRINT: POST %s (%d chars)", url, len(source))
        resp = self.session.post(url, headers=hdrs, data=source.encode("utf-8"))
        logger.info("PRETTY PRINT RESPONSE: HTTP %s", resp.status_code)
        resp.raise_for_status()
        return resp.text

    # ------------------------------------------------------------------
    #  Table structure / data query
    # ------------------------------------------------------------------

    def get_table_fields(self, table_name: str) -> list[dict[str, str]]:
        """查询 DDIC 表的字段结构。

        Args:
            table_name: 表名（不区分大小写）。

        Returns:
            字段列表，每项含 name, type, length, description, key_attribute。
        """
        url = f"{self.host}/sap/bc/adt/datapreview/ddic/{table_name.lower()}/metadata"
        hdrs = self._headers()
        hdrs["Accept"] = "*/*"
        logger.info("GET TABLE FIELDS: GET %s", url)
        resp = self.session.get(url, headers=hdrs)
        logger.info("GET TABLE FIELDS RESPONSE: HTTP %s", resp.status_code)
        resp.raise_for_status()

        root = ET.fromstring(resp.content)
        ns = {"dp": "http://www.sap.com/adt/dataPreview"}
        fields = []
        for col in root.findall(".//dp:columns/dp:metadata", ns):
            field = {
                "name": col.attrib.get(f"{{{ns['dp']}}}name", ""),
                "type": col.attrib.get(f"{{{ns['dp']}}}type", ""),
                "length": col.attrib.get(f"{{{ns['dp']}}}length", ""),
                "description": col.attrib.get(f"{{{ns['dp']}}}description", ""),
                "key_attribute": col.attrib.get(f"{{{ns['dp']}}}keyAttribute", "false"),
            }
            fields.append(field)
        return fields

    def query_table_data(self, sql: str, max_rows: int = 200) -> dict:
        """通过 ADT freestyle SQL 查询表数据。

        Args:
            sql: SELECT SQL 语句。
            max_rows: 最大返回行数。

        Returns:
            {"columns": [...字段名...], "rows": [[值1, 值2, ...], ...], "total_rows": int, "execution_time": str}
        """
        url = f"{self.host}/sap/bc/adt/datapreview/freestyle"
        hdrs = self._headers("text/plain; charset=utf-8")
        hdrs["Accept"] = "*/*"
        params = {"rowNumber": str(max_rows)}
        logger.info("QUERY TABLE DATA: POST %s sql=%s maxRows=%d", url, sql[:80], max_rows)
        resp = self.session.post(url, headers=hdrs, params=params, data=sql.encode("utf-8"))
        logger.info("QUERY TABLE DATA RESPONSE: HTTP %s", resp.status_code)
        resp.raise_for_status()

        root = ET.fromstring(resp.content)
        ns = {"dp": "http://www.sap.com/adt/dataPreview"}

        # 提取列名（从第一组 columns/metadata）
        columns = []
        for col in root.findall(".//dp:columns/dp:metadata", ns):
            name = col.attrib.get(f"{{{ns['dp']}}}name", "")
            if name:
                columns.append(name)

        # 提取数据 — ADT 按列存储（每个 columns 包含一列的 dataSet）
        # 需要转置为按行返回
        col_data: list[list[str]] = []
        for col_group in root.findall(".//dp:columns", ns):
            dataset = col_group.find("dp:dataSet", ns)
            if dataset is None:
                col_data.append([])
                continue
            values = [v.text or "" for v in dataset.findall("dp:data", ns)]
            col_data.append(values)

        # 转置：列数据 → 行数据
        max_len = max((len(c) for c in col_data), default=0)
        rows = []
        for i in range(max_len):
            row = []
            for c in col_data:
                row.append(c[i] if i < len(c) else "")
            rows.append(row)

        # 元数据
        total_el = root.find("dp:totalRows", ns)
        time_el = root.find("dp:queryExecutionTime", ns)

        return {
            "columns": columns,
            "rows": rows,
            "total_rows": int(total_el.text) if total_el is not None and total_el.text else len(rows),
            "execution_time": time_el.text if time_el is not None else "",
        }

    # ------------------------------------------------------------------
    #  Run program
    # ------------------------------------------------------------------

    def run_program(self, program_name: str) -> str:
        """远程执行 ABAP 程序并返回输出。

        Args:
            program_name: 程序名（不区分大小写）。

        Returns:
            程序的标准输出文本（text/plain）。
        """
        url = f"{self.host}/sap/bc/adt/programs/programrun/{program_name.lower()}"
        hdrs = self._headers("application/xml")
        hdrs["Accept"] = "*/*"
        logger.info("RUN PROGRAM: POST %s", url)
        resp = self.session.post(url, headers=hdrs)
        logger.info("RUN PROGRAM RESPONSE: HTTP %s (%d bytes)", resp.status_code, len(resp.content))
        resp.raise_for_status()
        return resp.text