197 lines
10 KiB
Python
197 lines
10 KiB
Python
"""Read-only TEST template audit. Credentials and workbook bytes stay in memory."""
|
||
import hashlib
|
||
import io
|
||
import json
|
||
import re
|
||
import sys
|
||
import urllib.error
|
||
import urllib.parse
|
||
import urllib.request
|
||
from datetime import datetime, timezone
|
||
from pathlib import Path
|
||
from openpyxl import load_workbook
|
||
|
||
|
||
def require(value, message):
|
||
if not value:
|
||
raise AssertionError(message)
|
||
|
||
|
||
def labels(values, labeler):
|
||
result = []
|
||
for item in values:
|
||
base = labeler(item)
|
||
label, n = base, 2
|
||
while label in result:
|
||
label = f"{base}（同名选项 {n}）"
|
||
n += 1
|
||
result.append(label)
|
||
return result
|
||
|
||
|
||
def label(name, code):
|
||
return code if not name or name.lower() == code.lower() else f"{name}｜{code}"
|
||
|
||
|
||
def compare_options(actual, expected):
|
||
return {"expectedCount": len(expected), "actualCount": len(actual),
|
||
"missingCount": len(set(expected) - set(actual)),
|
||
"extraCount": len(set(actual) - set(expected)),
|
||
"duplicateCount": len(actual) - len(set(actual)),
|
||
"passed": sorted(actual) == sorted(expected),
|
||
"actualSha256": hashlib.sha256(json.dumps(sorted(actual), ensure_ascii=False).encode()).hexdigest()}
|
||
|
||
|
||
def inspect_template(content, expected):
|
||
book = load_workbook(io.BytesIO(content), data_only=False)
|
||
try:
|
||
require(set(book.sheetnames) == {"订单导入", "选项数据", "填写说明"}, "Unexpected workbook sheets")
|
||
require(book["选项数据"].sheet_state == "hidden", "Options sheet must be hidden")
|
||
sheet = book["订单导入"]
|
||
headers = {str(cell.value).removesuffix(" *"): cell.column for cell in sheet[2] if cell.value}
|
||
checks = {}
|
||
for title, values in expected.items():
|
||
require(title in headers, "Missing expected header")
|
||
col = headers[title]
|
||
validations = [v for v in sheet.data_validations.dataValidation
|
||
if any(r.min_col <= col <= r.max_col and r.min_row <= 3 <= r.max_row for r in v.sqref.ranges)]
|
||
if not values:
|
||
require(not validations, "Empty customer directory unexpectedly has choices")
|
||
checks[title] = compare_options([], [])
|
||
continue
|
||
require(len(validations) == 1, "Missing or overlapping dropdown validation")
|
||
validation = validations[0]
|
||
require(validation.type == "list" and validation.showErrorMessage, "Dropdown must reject invalid choices")
|
||
require(any(r.min_row == 3 and r.max_row == 52 and r.min_col == col and r.max_col == col
|
||
for r in validation.sqref.ranges), "Dropdown must cover all 50 input rows")
|
||
name = book.defined_names.get(validation.formula1)
|
||
require(name is not None, "Dropdown name does not resolve")
|
||
destinations = list(name.destinations)
|
||
require(len(destinations) == 1 and destinations[0][0] == "选项数据", "Invalid dropdown destination")
|
||
actual = [cell.value for row in book[destinations[0][0]][destinations[0][1]] for cell in row]
|
||
require(all(isinstance(v, str) and v for v in actual), "Dropdown contains blank or invalid entries")
|
||
checks[title] = compare_options(actual, values)
|
||
return {"checks": checks, "passed": all(v["passed"] for v in checks.values()),
|
||
"workbookSha256": hashlib.sha256(content).hexdigest(), "headerCount": len(headers),
|
||
"dropdowns": len(sheet.data_validations.dataValidation)}
|
||
finally:
|
||
book.close()
|
||
|
||
|
||
def main(config):
|
||
require(config.get("environment") == "TEST" and re.fullmatch(r"[A-Za-z0-9_-]{1,60}", config.get("runId", "")), "Explicit TEST run required")
|
||
directory = Path(__file__).resolve().parent.parent / "reports" / ("client-template-" + config["runId"])
|
||
directory.mkdir() # Never overwrite an earlier run.
|
||
report_path = directory / "report.json"
|
||
report = {"startedAt": datetime.now(timezone.utc).isoformat(), "readOnlyBusiness": True,
|
||
"success": False, "sessionClosed": False, "requests": []}
|
||
token, refresh = "", ""
|
||
|
||
def save():
|
||
report_path.write_text(json.dumps(report, ensure_ascii=False, indent=2), encoding="utf-8")
|
||
|
||
def request(method, path, body=None, binary=False):
|
||
require(method == "GET" or method == "POST" and path in ["/auth/login", "/auth/logout"], "Business writes forbidden")
|
||
parsed = urllib.parse.urlsplit(path)
|
||
query = urllib.parse.parse_qs(parsed.query)
|
||
record = {"method": method, "path": parsed.path, "outcome": "PENDING",
|
||
"page": query.get("page"), "size": query.get("size")}
|
||
report["requests"].append(record)
|
||
save()
|
||
headers = {"Accept": "application/json"}
|
||
if token:
|
||
headers["Authorization"] = "Bearer " + token
|
||
if body is not None:
|
||
headers["Content-Type"] = "application/json"
|
||
req = urllib.request.Request(config["base"] + path, method=method, headers=headers,
|
||
data=json.dumps(body).encode() if body is not None else None)
|
||
try:
|
||
with urllib.request.urlopen(req, timeout=30) as response:
|
||
data = response.read()
|
||
record.update(outcome="RECEIVED", http=response.status,
|
||
requestId=response.headers.get("X-Request-ID"), traceId=response.headers.get("X-Trace-ID"))
|
||
if binary:
|
||
require(data[:2] == b"PK", "Template response is not XLSX")
|
||
record["sha256"] = hashlib.sha256(data).hexdigest()
|
||
save()
|
||
return data
|
||
payload = json.loads(data)
|
||
record.update(code=payload.get("code"), requestId=payload.get("requestId") or record["requestId"],
|
||
traceId=payload.get("traceId") or record["traceId"])
|
||
save()
|
||
require(str(payload.get("code")) == "200", "API business response rejected")
|
||
return payload["data"]
|
||
except urllib.error.HTTPError as error:
|
||
record.update(outcome="RECEIVED", http=error.code)
|
||
save()
|
||
raise AssertionError("HTTP response rejected") from None
|
||
except Exception:
|
||
if record["outcome"] == "PENDING":
|
||
record["outcome"] = "UNKNOWN"
|
||
save()
|
||
raise
|
||
|
||
try:
|
||
auth = request("POST", "/auth/login", {"loginId": config["username"], "password": config["password"]})
|
||
token, refresh = auth.get("accessToken", ""), auth.get("refreshToken", "")
|
||
require(token and refresh, "Incomplete login")
|
||
profile = request("GET", "/merchant/v1/iam/tenant/profile")
|
||
require(profile.get("id") is not None, "Missing authenticated customer identity")
|
||
report["customerIdentitySha256"] = hashlib.sha256(str(profile["id"]).encode()).hexdigest()
|
||
warehouses, total, page_no = [], None, 1
|
||
while True:
|
||
page = request("GET", f"/merchant/v1/warehouses?status=ACTIVE&page={page_no}&size=200")
|
||
if total is None:
|
||
total = int(page["total"])
|
||
require(int(page["page"]) == page_no and int(page["size"]) == 200 and int(page["total"]) == total
|
||
and int(page["pages"]) == (total + 199) // 200, "Warehouse pagination changed")
|
||
require(len(page["records"]) == min(200, max(0, total - (page_no - 1) * 200)), "Warehouse page incomplete")
|
||
warehouses.extend(page["records"])
|
||
if page_no >= int(page["pages"]):
|
||
break
|
||
require(page_no < 1000, "Warehouse page limit exceeded")
|
||
page_no += 1
|
||
require(len({str(w["id"]) for w in warehouses}) == total and all(w["status"] == "ACTIVE" for w in warehouses), "Invalid warehouse set")
|
||
expected = {}
|
||
for kind, title in [("DOMESTIC", "国内收货仓"), ("FBA", "FBA目的仓"), ("OVERSEAS", "海外仓目的仓")]:
|
||
expected[title] = labels([w for w in warehouses if w["warehouseType"] == kind], lambda w: label(w["warehouseName"], w["warehouseCode"]))
|
||
channels = []
|
||
for warehouse in warehouses:
|
||
if warehouse["warehouseType"] != "DOMESTIC":
|
||
continue
|
||
values = request("GET", "/merchant/v1/first-leg/pricing/channels?originWarehouseId=" + str(warehouse["id"]))
|
||
channels.extend(label(c["channelName"], c["channelCode"]) for c in values)
|
||
expected["运输渠道"] = list(dict.fromkeys(channels))
|
||
for kind, title in [("PICKUP", "取货地址"), ("PRIVATE_DESTINATION", "私人目的仓地址")]:
|
||
addresses = request("GET", "/merchant/v1/orders/first-leg/addresses?addressType=" + kind)
|
||
expected[title] = labels(addresses, lambda a: "｜".join(a[key] for key in ["addressName", "contactName", "city", "addressLine"]))
|
||
content = request("GET", "/merchant/v1/orders/first-leg/batch-imports/template", binary=True)
|
||
report["template"] = inspect_template(content, expected)
|
||
report["warehousePages"] = page_no
|
||
report["warehouseTotal"] = total
|
||
report["success"] = report["template"]["passed"]
|
||
report["limitations"] = ["No import submission or Excel desktop interaction performed",
|
||
"Second-customer address rejection requires separate evidence"]
|
||
except Exception as error:
|
||
report["failure"] = str(error) if isinstance(error, AssertionError) else type(error).__name__
|
||
finally:
|
||
if token:
|
||
try:
|
||
request("POST", "/auth/logout", {"refreshToken": refresh})
|
||
report["sessionClosed"] = True
|
||
except Exception:
|
||
report["logoutFailure"] = True
|
||
report["success"] = report["success"] and report["sessionClosed"]
|
||
report["completedAt"] = datetime.now(timezone.utc).isoformat()
|
||
save()
|
||
print(json.dumps({"success": report["success"], "sessionClosed": report["sessionClosed"], "report": str(report_path)}))
|
||
return 0 if report["success"] else 1
|
||
|
||
|
||
if __name__ == "__main__":
|
||
try:
|
||
sys.exit(main(json.load(sys.stdin)))
|
||
except Exception:
|
||
print("Template audit did not start; verify the new report directory and runtime.", file=sys.stderr)
|
||
sys.exit(1)
|