428b3565b2
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_014aUaQeLnwbb1zTpN7kHeat
101 lines
4.9 KiB
Python
101 lines
4.9 KiB
Python
"""Direct ADT HTTP access. Only for teardown (deletion). The model never uses it.
|
|
|
|
Credentials come from the environment (or .env): A4H_URL, A4H_USER, A4H_PASSWORD, A4H_CLIENT.
|
|
"""
|
|
import base64
|
|
import http.cookiejar
|
|
import os
|
|
import re
|
|
import urllib.request
|
|
from xml.sax.saxutils import quoteattr
|
|
|
|
DEL_REQ = "application/vnd.sap.adt.deletion.request.v1+xml"
|
|
DEL_RES = "application/vnd.sap.adt.deletion.response.v1+xml"
|
|
|
|
|
|
def load_env(path=".env"):
|
|
if os.path.exists(path):
|
|
for line in open(path):
|
|
line = line.strip()
|
|
if line and not line.startswith("#") and "=" in line:
|
|
k, v = line.split("=", 1)
|
|
os.environ.setdefault(k.strip(), v.strip())
|
|
|
|
|
|
class AdtClient:
|
|
def __init__(self):
|
|
load_env()
|
|
self.base = os.environ["A4H_URL"].rstrip("/")
|
|
self.client = os.environ.get("A4H_CLIENT", "001")
|
|
auth = f'{os.environ["A4H_USER"]}:{os.environ["A4H_PASSWORD"]}'
|
|
self.auth = "Basic " + base64.b64encode(auth.encode()).decode()
|
|
self.opener = urllib.request.build_opener(
|
|
urllib.request.HTTPCookieProcessor(http.cookiejar.CookieJar()))
|
|
self.csrf = None
|
|
|
|
def _req(self, method, path, body=None, headers=None):
|
|
url = f"{self.base}{path}{'&' if '?' in path else '?'}sap-client={self.client}"
|
|
h = {"Authorization": self.auth}
|
|
if self.csrf:
|
|
h["x-csrf-token"] = self.csrf
|
|
h.update(headers or {})
|
|
req = urllib.request.Request(url, body.encode() if body else None, h, method=method)
|
|
with self.opener.open(req, timeout=300) as r:
|
|
return r.status, dict(r.headers), r.read().decode()
|
|
|
|
def fetch_csrf(self):
|
|
_, hdr, _ = self._req("GET", "/sap/bc/adt/discovery",
|
|
headers={"x-csrf-token": "fetch", "Accept": "*/*"})
|
|
self.csrf = hdr.get("x-csrf-token") or hdr.get("X-CSRF-Token")
|
|
|
|
def delete(self, uris):
|
|
"""Delete objects by ADT URI. Returns {uri: (deleted, message)}."""
|
|
if not uris:
|
|
return {}
|
|
if not self.csrf:
|
|
self.fetch_csrf()
|
|
objs = "".join(f"<del:object adtcore:uri={quoteattr(u)}><del:transportNumber/></del:object>"
|
|
for u in uris)
|
|
body = ('<?xml version="1.0" encoding="UTF-8"?>'
|
|
'<del:deletionRequest xmlns:del="http://www.sap.com/adt/deletion" '
|
|
'xmlns:adtcore="http://www.sap.com/adt/core">' + objs + '</del:deletionRequest>')
|
|
_, _, text = self._req("POST", "/sap/bc/adt/deletion/delete", body,
|
|
{"Content-Type": DEL_REQ, "Accept": DEL_RES})
|
|
out = {}
|
|
for m in re.finditer(r'<del:object\b([^>]*)>(.*?)</del:object>', text, re.S):
|
|
attrs, inner = m.group(1), m.group(2)
|
|
uri = re.search(r'adtcore:uri="([^"]+)"', attrs)
|
|
ok = re.search(r'del:isDeleted="([^"]+)"', attrs)
|
|
msg = re.search(r'<del:text>(.*?)</del:text>', inner, re.S)
|
|
out[uri.group(1) if uri else "?"] = (ok is not None and ok.group(1) == "true",
|
|
msg.group(1) if msg else "")
|
|
return out
|
|
|
|
def write_activate(self, uri, name, source):
|
|
"""Lock, write source/main, unlock, activate. Fallback for EPOD: a second sap_push_source on a
|
|
PROG or FUNC writes an inactive version that is not in the inactive list, so no activation runs
|
|
(probe Z0ZZZ999_PROBE_PROG, 2026-10-03). Returns (activated, [(severity, text)])."""
|
|
if not self.csrf:
|
|
self.fetch_csrf()
|
|
st = {"X-sap-adt-sessiontype": "stateful"}
|
|
_, _, t = self._req("POST", uri + "?_action=LOCK&accessMode=MODIFY", None,
|
|
dict(st, Accept="application/vnd.sap.as+xml;charset=UTF-8;"
|
|
"dataname=com.sap.adt.lock.result"))
|
|
lh = re.search(r"<LOCK_HANDLE>(.*?)</LOCK_HANDLE>", t).group(1)
|
|
try:
|
|
self._req("PUT", f"{uri}/source/main?lockHandle={lh}", source,
|
|
dict(st, **{"Content-Type": "text/plain; charset=utf-8"}))
|
|
finally:
|
|
self._req("POST", f"{uri}?_action=UNLOCK&lockHandle={lh}", None, st)
|
|
body = ('<?xml version="1.0" encoding="UTF-8"?><adtcore:objectReferences '
|
|
'xmlns:adtcore="http://www.sap.com/adt/core">'
|
|
f'<adtcore:objectReference adtcore:uri={quoteattr(uri)} adtcore:name={quoteattr(name)}/>'
|
|
'</adtcore:objectReferences>')
|
|
_, _, t = self._req("POST", "/sap/bc/adt/activation?method=activate&preauditRequested=true", body,
|
|
{"Content-Type": "application/xml", "Accept": "application/xml"})
|
|
msgs = [(m.group(1), re.sub(r"<[^>]+>", "", m.group(2)).strip())
|
|
for m in re.finditer(r'<msg\b[^>]*\btype="(\w)"[^>]*>(.*?)</msg>', t, re.S)]
|
|
done = 'activationExecuted="true"' in t and not any(sev in ("E", "A") for sev, _ in msgs)
|
|
return done, msgs
|
|
|