60 lines
3.0 KiB
Python
60 lines
3.0 KiB
Python
"""只读检查明细接口，并用确定为空的商务筛选验证真实 XLSX 导出。"""
|
|
import argparse
|
|
import io
|
|
import json
|
|
import urllib.parse
|
|
import urllib.request
|
|
import zipfile
|
|
import xml.etree.ElementTree as ET
|
|
from pathlib import Path
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument("--base-url", default="http://127.0.0.1:18000")
|
|
parser.add_argument("--token-file", required=True, type=Path)
|
|
parser.add_argument("--output", required=True, type=Path)
|
|
args = parser.parse_args()
|
|
token = args.token_file.read_text(encoding="utf-8").strip()
|
|
checks = []
|
|
keyword = "__kra_empty_export_probe_20261104__"
|
|
|
|
def call(method, path, payload=None, expected=0, excel=False):
|
|
body = urllib.parse.urlencode(payload).encode() if payload is not None else None
|
|
request = urllib.request.Request(args.base_url + path, method=method, data=body,
|
|
headers={"x-token": token, "Content-Type": "application/x-www-form-urlencoded"})
|
|
with urllib.request.urlopen(request, timeout=30) as response:
|
|
raw = response.read()
|
|
item = {"method": method, "path": path, "http": response.status}
|
|
if excel:
|
|
with zipfile.ZipFile(io.BytesIO(raw)) as archive:
|
|
sheet = ET.fromstring(archive.read("xl/worksheets/sheet1.xml"))
|
|
rows = sheet.findall(".//{http://schemas.openxmlformats.org/spreadsheetml/2006/main}row")
|
|
item.update(xlsx=True, worksheet_rows=len(rows), passed=response.headers.get("success") == "true" and len(rows) == 1)
|
|
else:
|
|
value = json.loads(raw)
|
|
data = value.get("data", {})
|
|
item.update(code=value.get("code"), msg=value.get("msg"), passed=value.get("code") == expected)
|
|
if isinstance(data, dict) and "rows" in data:
|
|
item.update(total=data.get("total"), available=data.get("available"))
|
|
checks.append(item)
|
|
return item
|
|
|
|
for kind in ["register", "spend"]:
|
|
item = call("GET", f"/sw/{kind}/list?sourceDb=lx,hw,yq")
|
|
empty = call("GET", f"/sw/{kind}/list?" + urllib.parse.urlencode({"busierAccount": keyword}))
|
|
empty["passed"] &= empty.get("total") == 0 and empty.get("available") is True
|
|
call("POST", f"/sw/{kind}/export", {"busierAccount": keyword}, excel=True)
|
|
if item.get("available") is False:
|
|
call("POST", f"/sw/{kind}/export", {"sourceDb": "lx"}, expected=7)
|
|
call("GET", "/sw/register/list?beginRegisterTime=bad&endRegisterTime=2026-11-04", expected=7)
|
|
report = {"scope": "真实 sw7 会话，只读查询；导出使用确定为空的商务筛选，仅校验表头工作簿，不导出真实业务行。", "checks": checks, "passed": all(item["passed"] for item in checks)}
|
|
args.output.write_text(json.dumps(report, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
|
|
print(json.dumps(report, ensure_ascii=False))
|
|
if not report["passed"]:
|
|
raise SystemExit(1)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|