wurangyu
1e39b6da88
- assets/: sap-cli source (v2.1.0, 26 commands, 16 object types) - references/: tool constraints + error handling (self-contained) - scripts/setup.py: one-click install/config/verify - SKILL.md: full command reference + dual-platform install guide - VERSION: 2.1.0 Built from D:/Codespace/sap-cli via scripts/pack_skill.py
405 lines
17 KiB
Python
405 lines
17 KiB
Python
"""SourceMixin — source code read/write, lock/unlock, activate, syntax-check."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
import re
|
|
import xml.etree.ElementTree as ET
|
|
|
|
from sapcli.exceptions import (
|
|
ActivationError,
|
|
LockError,
|
|
SyntaxCheckError,
|
|
)
|
|
|
|
logger = logging.getLogger("sapcli.client")
|
|
|
|
|
|
class SourceMixin:
|
|
"""Source-code read/write and lock management."""
|
|
|
|
# ------------------------------------------------------------------
|
|
# Source read / write
|
|
# ------------------------------------------------------------------
|
|
|
|
def get_source(self, src_uri: str) -> str:
|
|
url = f"{self.host}{src_uri}"
|
|
hdrs = self._headers()
|
|
hdrs["Accept"] = "text/plain"
|
|
logger.info("GET SOURCE: GET %s", url)
|
|
resp = self.session.get(url, headers=hdrs)
|
|
logger.info(
|
|
"GET SOURCE RESPONSE: HTTP %s, %d bytes",
|
|
resp.status_code,
|
|
len(resp.content) if resp.content else 0,
|
|
)
|
|
resp.raise_for_status()
|
|
return resp.text
|
|
|
|
def set_source(
|
|
self,
|
|
src_uri: str,
|
|
source: str,
|
|
lock_handle: str,
|
|
corr_nr: str | None = None,
|
|
) -> bool:
|
|
params: dict[str, str] = {}
|
|
if lock_handle:
|
|
params["lockHandle"] = lock_handle
|
|
if corr_nr:
|
|
params["corrNr"] = corr_nr
|
|
url = f"{self.host}{src_uri}"
|
|
hdrs = self._headers("text/plain; charset=utf-8")
|
|
logger.info("SET SOURCE: PUT %s (%d chars)", url, len(source))
|
|
resp = self.session.put(
|
|
url, headers=hdrs, params=params, data=source.encode("utf-8")
|
|
)
|
|
logger.info("SET SOURCE RESPONSE: HTTP %s", resp.status_code)
|
|
if resp.status_code >= 400:
|
|
logger.error("SET SOURCE ERROR BODY: %s", resp.text[:2000])
|
|
existing_nr = self._extract_locked_corrnr(resp.text)
|
|
if existing_nr and existing_nr != corr_nr:
|
|
logger.info("使用 corrNr=%s 重试 SET SOURCE", existing_nr)
|
|
params["corrNr"] = existing_nr
|
|
resp = self.session.put(
|
|
url, headers=hdrs, params=params, data=source.encode("utf-8")
|
|
)
|
|
logger.info("SET SOURCE RETRY RESPONSE: HTTP %s", resp.status_code)
|
|
if resp.status_code >= 400:
|
|
resp.raise_for_status()
|
|
return True
|
|
|
|
# ------------------------------------------------------------------
|
|
# Lock / unlock
|
|
# ------------------------------------------------------------------
|
|
|
|
def lock(
|
|
self,
|
|
obj_uri: str,
|
|
corr_nr: str | None = None,
|
|
accept: str | None = None,
|
|
) -> tuple[str, str | None]:
|
|
"""锁定对象。
|
|
|
|
Returns:
|
|
(lock_handle, effective_corr_nr) — 实际使用的传输请求号。
|
|
如果对象已绑定在某请求中，effective_corrnr 为该请求号；
|
|
否则为传入的 corr_nr（可能为 None）。
|
|
"""
|
|
self._stateful = True
|
|
url = f"{self.host}{obj_uri}"
|
|
params: dict[str, str] = {"_action": "LOCK", "accessMode": "MODIFY"}
|
|
effective_corr_nr = corr_nr
|
|
if corr_nr:
|
|
params["corrNr"] = corr_nr
|
|
hdrs = self._headers()
|
|
hdrs["Accept"] = accept or "*/*"
|
|
logger.info("LOCK: POST %s (corrNr=%s)", url, corr_nr or "none")
|
|
resp = self.session.post(url, headers=hdrs, params=params)
|
|
logger.info("LOCK RESPONSE: HTTP %s", resp.status_code)
|
|
if resp.status_code == 500:
|
|
existing_nr = self._extract_locked_corrnr(resp.text)
|
|
if existing_nr:
|
|
effective_corr_nr = existing_nr
|
|
logger.info("对象已锁定在请求 %s 中，使用该请求重试", existing_nr)
|
|
params["corrNr"] = existing_nr
|
|
resp = self.session.post(url, headers=hdrs, params=params, data="")
|
|
logger.info("LOCK RETRY RESPONSE: HTTP %s", resp.status_code)
|
|
if resp.status_code >= 400:
|
|
raise LockError(f"锁定失败: HTTP {resp.status_code}", obj_uri=obj_uri)
|
|
root = ET.fromstring(resp.content)
|
|
handle = root.findtext(".//LOCK_HANDLE")
|
|
if not handle:
|
|
for el in root.iter():
|
|
if "HANDLE" in el.tag.upper():
|
|
handle = el.text
|
|
break
|
|
if not handle:
|
|
raise LockError("锁定失败: 未获取到 lock handle", obj_uri=obj_uri)
|
|
|
|
# 从成功响应中提取 corrNr（对象可能已绑定传输请求）
|
|
if not effective_corr_nr:
|
|
extracted = self._extract_corrnr_from_lock_response(root)
|
|
if extracted:
|
|
effective_corr_nr = extracted
|
|
logger.info("从 lock 响应中检测到 corrNr: %s", extracted)
|
|
|
|
logger.info(
|
|
"Lock handle: %s (corrNr=%s)",
|
|
handle[:20] if len(handle) > 20 else handle,
|
|
effective_corr_nr or "none",
|
|
)
|
|
return handle, effective_corr_nr
|
|
|
|
def _extract_locked_corrnr(self, error_body: str) -> str | None:
|
|
try:
|
|
root = ET.fromstring(error_body)
|
|
for entry in root.iter():
|
|
if entry.attrib.get("key") == "corrNr":
|
|
corrnr = entry.text
|
|
if corrnr and corrnr != "*":
|
|
return corrnr
|
|
for el in root.iter():
|
|
if el.tag.endswith("message") and el.text:
|
|
m = re.search(r"request\s+(\w+)", el.text, re.IGNORECASE)
|
|
if m:
|
|
return m.group(1)
|
|
except Exception:
|
|
logger.debug("XML 解析提取 locked corrnr 失败", exc_info=True)
|
|
return None
|
|
|
|
def _extract_corrnr_from_lock_response(self, root: ET.Element) -> str | None:
|
|
"""从成功的 lock 响应 XML 中提取传输请求号。
|
|
|
|
常见格式:
|
|
<lock:lock ...>
|
|
<lock:LOCK_HANDLE>...</lock:LOCK_HANDLE>
|
|
<lock:CORRN>DEVK901362</lock:CORRN>
|
|
</lock:lock>
|
|
也可能是:
|
|
<adtcore:property adtcore:key="corrNr">DEVK901362</adtcore:property>
|
|
"""
|
|
# 策略 1: 直接找 CORRN 标签
|
|
for tag_name in ("CORRN", "corrNr", "corr_nr"):
|
|
text = root.findtext(f".//{tag_name}")
|
|
if text and text.strip() and text.strip() != "*":
|
|
return text.strip()
|
|
|
|
# 策略 2: 遍历所有元素，找标签或属性包含 corrNr 的
|
|
for el in root.iter():
|
|
# 检查属性
|
|
for attr_key in el.attrib:
|
|
if "corr" in attr_key.lower() and "nr" in attr_key.lower():
|
|
val = el.attrib[attr_key]
|
|
if val and val.strip() and val.strip() != "*":
|
|
return val.strip()
|
|
# 检查标签名包含 CORRN
|
|
if el.text and "corr" in el.tag.lower() and el.text.strip() and el.text.strip() != "*":
|
|
return el.text.strip()
|
|
|
|
# 策略 3: 找 key="corrNr" 的 property 元素
|
|
for el in root.iter():
|
|
if el.attrib.get("key") == "corrNr" and el.text:
|
|
val = el.text.strip()
|
|
if val and val != "*":
|
|
return val
|
|
|
|
return None
|
|
|
|
def unlock(self, obj_uri: str, lock_handle: str) -> bool:
|
|
url = f"{self.host}{obj_uri}"
|
|
params = {"_action": "UNLOCK", "lockHandle": lock_handle}
|
|
hdrs = self._headers("text/plain; charset=utf-8")
|
|
logger.info("UNLOCK: POST %s", url)
|
|
resp = self.session.post(url, headers=hdrs, params=params, data="")
|
|
logger.info("UNLOCK RESPONSE: HTTP %s", resp.status_code)
|
|
self._stateful = False
|
|
return resp.status_code == 200
|
|
|
|
# ------------------------------------------------------------------
|
|
# Activate / syntax check
|
|
# ------------------------------------------------------------------
|
|
|
|
def activate(
|
|
self,
|
|
name: str,
|
|
obj_uri: str,
|
|
corr_nr: str | None = None,
|
|
) -> tuple[bool, list[dict[str, str]]]:
|
|
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>"
|
|
)
|
|
self._stateful = False
|
|
url = f"{self.host}/sap/bc/adt/activation"
|
|
params: dict[str, str] = {"method": "activate"}
|
|
if corr_nr:
|
|
params["corrNr"] = corr_nr
|
|
hdrs = self._headers()
|
|
logger.info("ACTIVATE: POST %s name=%s corrNr=%s", url, name, corr_nr or "none")
|
|
resp = self.session.post(url, headers=hdrs, params=params, data=body)
|
|
logger.info(
|
|
"ACTIVATE RESPONSE: HTTP %s CT=%s",
|
|
resp.status_code,
|
|
resp.headers.get("content-type", "")[:80],
|
|
)
|
|
|
|
if resp.status_code != 200:
|
|
logger.error("Activation HTTP %s", resp.status_code)
|
|
raise ActivationError(f"激活失败: HTTP {resp.status_code}")
|
|
|
|
ct = resp.headers.get("content-type", "")
|
|
if "inactivectsobjects" in ct:
|
|
root = ET.fromstring(resp.content)
|
|
ioc_ns = "http://www.sap.com/abapxml/inactiveCtsObjects"
|
|
inactive_objects: list[str] = []
|
|
for entry in root.findall(f".//{{{ioc_ns}}}entry"):
|
|
obj_elem = entry.find(f"{{{ioc_ns}}}object")
|
|
if obj_elem is not None:
|
|
ref = obj_elem.find(".//{http://www.sap.com/adt/core}ref")
|
|
if ref is None:
|
|
for child in obj_elem:
|
|
if child.tag.endswith("}ref"):
|
|
ref = child
|
|
break
|
|
if ref is not None:
|
|
inactive_name = ref.attrib.get(
|
|
"{http://www.sap.com/adt/core}name", ""
|
|
)
|
|
inactive_type = ref.attrib.get(
|
|
"{http://www.sap.com/adt/core}type", ""
|
|
)
|
|
if inactive_name:
|
|
inactive_objects.append(f"{inactive_name} ({inactive_type})")
|
|
if inactive_objects:
|
|
logger.warning("激活返回未激活对象: %s", ", ".join(inactive_objects))
|
|
msg_text = f"Objects still inactive: {', '.join(inactive_objects)}"
|
|
messages: list[dict[str, str]] = [
|
|
{"type": "E", "line": "?", "text": msg_text, "href": ""}
|
|
]
|
|
logger.info("Activation result: FAILED (inactive objects)")
|
|
return False, messages
|
|
logger.info("Activation result: SUCCESS (inactiveObjects 响应但无对象列出)")
|
|
return True, []
|
|
|
|
if not resp.content or not resp.text.strip():
|
|
# NW 7.40 返回空响应但可能并未真正激活
|
|
# 验证对象实际状态
|
|
logger.info("Activation returned empty response, verifying object status...")
|
|
try:
|
|
verify_resp = self.session.get(
|
|
f"{self.host}{obj_uri}",
|
|
headers=self._headers(),
|
|
)
|
|
if verify_resp.status_code == 200:
|
|
vroot = ET.fromstring(verify_resp.content)
|
|
version = vroot.attrib.get("{http://www.sap.com/adt/core}version", "")
|
|
if version == "inactive":
|
|
# NW 7.40: first activation may only "stage" the change.
|
|
# Retry activation once, then re-verify.
|
|
logger.info("Object still inactive, retrying activation (NW 7.40 double-activate workaround)...")
|
|
retry_resp = self.session.post(
|
|
url, headers=hdrs, params=params, data=body,
|
|
)
|
|
logger.info(
|
|
"RETRY ACTIVATE: HTTP %s len=%s",
|
|
retry_resp.status_code, len(retry_resp.content),
|
|
)
|
|
if retry_resp.status_code == 200:
|
|
# Re-verify after retry
|
|
verify_resp2 = self.session.get(
|
|
f"{self.host}{obj_uri}",
|
|
headers=self._headers(),
|
|
)
|
|
if verify_resp2.status_code == 200:
|
|
vroot2 = ET.fromstring(verify_resp2.content)
|
|
version2 = vroot2.attrib.get(
|
|
"{http://www.sap.com/adt/core}version", ""
|
|
)
|
|
if version2 == "active":
|
|
logger.info("Retry activation succeeded!")
|
|
return True, []
|
|
logger.warning("Object still inactive after retry (NW 7.40)")
|
|
messages = [
|
|
{
|
|
"type": "W",
|
|
"line": "?",
|
|
"text": "ADT 激活返回空响应，对象仍为 inactive（NW 7.40 已知限制）。请在 SAP GUI SE09 手动激活。",
|
|
"href": "",
|
|
}
|
|
]
|
|
return False, messages
|
|
elif version == "active":
|
|
logger.info("Verified: object is active after activation")
|
|
except Exception as e:
|
|
logger.warning("Could not verify activation status: %s", e)
|
|
logger.info("Activation success (空响应)")
|
|
return True, []
|
|
|
|
root = ET.fromstring(resp.content)
|
|
messages = []
|
|
for msg in root.iter():
|
|
msg_type = msg.attrib.get("type", "")
|
|
if msg_type in ("E", "W", "I", "S"):
|
|
line = msg.attrib.get("line", "?")
|
|
href = msg.attrib.get("href", "")
|
|
txt = ""
|
|
for child in msg.iter():
|
|
if child.text and child.tag.endswith("}txt"):
|
|
txt = child.text
|
|
break
|
|
if not txt:
|
|
for child in msg.iter():
|
|
if child.text and len(child.text.strip()) > 3:
|
|
txt = child.text.strip()
|
|
break
|
|
messages.append({"type": msg_type, "line": line, "text": txt, "href": href})
|
|
logger.info("Activation msg [%s] line=%s: %s", msg_type, line, txt)
|
|
|
|
errors = [m for m in messages if m["type"] == "E"]
|
|
success = len(errors) == 0
|
|
logger.info(
|
|
"Activation result: %s (%d errors)",
|
|
"SUCCESS" if success else "FAILED",
|
|
len(errors),
|
|
)
|
|
return success, messages
|
|
|
|
def syntax_check(
|
|
self,
|
|
name: str,
|
|
obj_uri: str,
|
|
) -> tuple[bool, list[dict[str, str]]]:
|
|
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>"
|
|
)
|
|
self._stateful = False
|
|
url = f"{self.host}/sap/bc/adt/activation"
|
|
params = {"method": "check"}
|
|
hdrs = self._headers()
|
|
logger.info("SYNTAX CHECK: POST %s name=%s", url, name)
|
|
resp = self.session.post(url, headers=hdrs, params=params, data=body)
|
|
logger.info("SYNTAX CHECK RESPONSE: HTTP %s", resp.status_code)
|
|
|
|
if resp.status_code != 200:
|
|
logger.error("Syntax check HTTP %s", resp.status_code)
|
|
raise SyntaxCheckError(f"语法检查失败: HTTP {resp.status_code}")
|
|
|
|
if not resp.content or not resp.text.strip():
|
|
logger.info("Syntax check: OK (空响应)")
|
|
return True, []
|
|
|
|
root = ET.fromstring(resp.content)
|
|
messages: list[dict[str, str]] = []
|
|
for msg in root.iter():
|
|
msg_type = msg.attrib.get("type", "")
|
|
if msg_type in ("E", "W", "I", "S"):
|
|
line = msg.attrib.get("line", "?")
|
|
href = msg.attrib.get("href", "")
|
|
txt = ""
|
|
for child in msg.iter():
|
|
if child.text and child.tag.endswith("}txt"):
|
|
txt = child.text
|
|
break
|
|
if not txt:
|
|
for child in msg.iter():
|
|
if child.text and len(child.text.strip()) > 3:
|
|
txt = child.text.strip()
|
|
break
|
|
messages.append({"type": msg_type, "line": line, "text": txt, "href": href})
|
|
logger.info("Syntax check msg [%s] line=%s: %s", msg_type, line, txt)
|
|
|
|
errors = [m for m in messages if m["type"] == "E"]
|
|
success = len(errors) == 0
|
|
logger.info(
|
|
"Syntax check result: %s (%d errors)",
|
|
"OK" if success else "ERRORS",
|
|
len(errors),
|
|
)
|
|
return success, messages
|