吴让宇
53ea2d854d
- list_objects: maxrow/name/type 改为 operation=quickSearch/query/maxResults/objectType， 并剥离 NW 7.40/7.50 quickSearch name 里的本地化类型标签（如 "ZCL_FOO (Class)"） - query_table_data: HTTP 400 转抛 SapCliError，说明方言限制（WHERE 仅 IN/LIKE），不再打堆栈 - _query_transport_request: E071 查询 WHERE 改用 IN(...)，区分「无」与「无法获取」
202 lines
7.0 KiB
Python
202 lines
7.0 KiB
Python
"""SearchMixin — repository search, where-used, code search, diff helpers."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
import re
|
|
import xml.etree.ElementTree as ET
|
|
|
|
from sapcli.exceptions import SapCliError
|
|
|
|
logger = logging.getLogger("sapcli.client")
|
|
|
|
|
|
def _strip_type_label(name: str) -> str:
|
|
"""剥离 NW 7.40/7.50 quickSearch 名称里的本地化类型标签（如 ``ZCL_FOO (Class)``）。"""
|
|
return re.sub(r"\s*\(.*$", "", name).strip()
|
|
|
|
|
|
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] = {
|
|
"operation": "quickSearch",
|
|
"query": prefix or "*",
|
|
"maxResults": "200",
|
|
}
|
|
if obj_type:
|
|
params["objectType"] = obj_type
|
|
|
|
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 = _strip_type_label(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)
|