feat: 迁移注册充值明细和筛选导出

2026-11-04 00:32:49 +08:00
parent f5b7ce21b9
commit 6703a66f0d
34 changed files with 1964 additions and 9 deletions
@@ -0,0 +1,59 @@
+"""只读检查明细接口，并用确定为空的商务筛选验证真实 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()