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

"""SearchMixin — repository search, where-used, code search, diff helpers."""

from __future__ import annotations

import logging
import xml.etree.ElementTree as ET

from sapcli.exceptions import SapCliError

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


class SearchMixin:
    """Object search, where-used listing, and source-code search."""

    def list_objects(
        self,
        obj_type: str | None = None,
        package: str | None = None,
        prefix: str | None = None,
    ) -> list[dict[str, str]]:
        """通过 ADT Information System 搜索对象。

        Args:
            obj_type: ADT 类型代码，例如 ``PROG/P``, ``CLAS/OC``。
            package: 包名过滤。
            prefix:  名称前缀过滤（如 ``Z*``）。

        Returns:
            列表，每项含 ``name``, ``type``, ``description``, ``package``。
        """
        url = f"{self.host}/sap/bc/adt/repository/informationsystem/search"
        params: dict[str, str] = {"maxrow": "200"}
        if obj_type:
            params["type"] = obj_type
        if prefix:
            params["name"] = prefix
        else:
            params["name"] = "*"
        if package:
            params["pkg"] = package

        hdrs = self._headers()
        hdrs["Accept"] = "application/xml"
        logger.info("LIST OBJECTS: GET %s params=%s", url, params)
        resp = self.session.get(url, headers=hdrs, params=params)
        logger.info("LIST 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"):
            name = ref.attrib.get(f"{{{core_ns}}}name", "")
            otype = ref.attrib.get(f"{{{core_ns}}}type", "")
            desc = ref.attrib.get(f"{{{core_ns}}}description", "")
            pkg = ref.attrib.get(f"{{{core_ns}}}package", "")
            results.append(
                {
                    "name": name,
                    "type": otype,
                    "description": desc,
                    "package": pkg,
                }
            )
        return results

    def where_used(
        self,
        name: str,
        obj_uri: str,
        adt_type: str | None = None,
    ) -> list[dict[str, str]]:
        """Where-Used 引用查询。

        Args:
            name: 对象名称。
            obj_uri: 对象 ADT URI。
            adt_type: ADT 类型代码。

        Returns:
            列表，每项含 ``name``, ``type``, ``package``, ``uri``。
        """
        url = f"{self.host}/sap/bc/adt/usage/whereusedlist"
        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}"'
        )
        if adt_type:
            body += f' adtcore:type="{adt_type}"'
        body += "/></adtcore:objectReferences>"

        hdrs = self._headers()
        logger.info("WHERE USED: POST %s name=%s", url, name)
        resp = self.session.post(url, headers=hdrs, data=body.encode("utf-8"))
        logger.info("WHERE USED 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"):
            ref_name = ref.attrib.get(f"{{{core_ns}}}name", "")
            ref_type = ref.attrib.get(f"{{{core_ns}}}type", "")
            ref_pkg = ref.attrib.get(f"{{{core_ns}}}package", "")
            ref_uri = ref.attrib.get(f"{{{core_ns}}}uri", "")
            results.append(
                {
                    "name": ref_name,
                    "type": ref_type,
                    "package": ref_pkg,
                    "uri": ref_uri,
                }
            )
        return results

    def search_code(
        self,
        query: str,
        obj_type: str | None = None,
    ) -> list[dict[str, str]]:
        """源代码搜索。

        Args:
            query: 搜索关键词。
            obj_type: ADT 类型代码过滤。

        Returns:
            列表，每项含 ``name``, ``type``, ``description``。
        """
        url = f"{self.host}/sap/bc/adt/repository/structuredsearch"
        body = (
            '<?xml version="1.0" encoding="UTF-8"?>'
            '<ssearch:search xmlns:ssearch="http://www.sap.com/adt/structuredsearch"'
            ' ssearch:maxResults="200">'
            "<ssearch:queries>"
            "<ssearch:query>"
            f"<ssearch:selection>"
            f'<ssearch:attribute ssearch:attribute="name" ssearch:operator="contains"'
            f' ssearch:value="{query}"/>'
        )
        if obj_type:
            body += (
                f'<ssearch:attribute ssearch:attribute="type" ssearch:operator="equals"'
                f' ssearch:value="{obj_type}"/>'
            )
        body += (
            "</ssearch:selection>"
            "</ssearch:query>"
            "</ssearch:queries>"
            "</ssearch:search>"
        )

        hdrs = self._headers()
        hdrs["Accept"] = "application/xml"
        logger.info("SEARCH CODE: POST %s query=%s", url, query)
        resp = self.session.post(url, headers=hdrs, data=body.encode("utf-8"))
        logger.info("SEARCH CODE 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_val = ref.attrib.get(f"{{{core_ns}}}type", "")
            desc = ref.attrib.get(f"{{{core_ns}}}description", "")
            results.append(
                {
                    "name": obj_name,
                    "type": obj_type_val,
                    "description": desc,
                }
            )
        return results

    # ------------------------------------------------------------------
    #  Diff helper
    # ------------------------------------------------------------------

    def read_source_for_diff(self, name: str, obj_type: str) -> str:
        """读取对象源码用于 diff 对比。

        Args:
            name: 对象名称。
            obj_type: 对象类型键（如 ``report``, ``class`` 等）。

        Returns:
            源代码字符串。
        """
        from sapcli.types import parse_object_name

        parsed = parse_object_name(name, obj_type)
        if parsed.src_uri is None:
            raise SapCliError(f"{obj_type} 对象没有源代码 URI")
        return self.get_source(parsed.src_uri)