From 2241a89fd17986d931f1c155a34dbbea4306b62e Mon Sep 17 00:00:00 2001 From: Hermes CI Fix Date: Mon, 31 Aug 2026 17:05:21 +0800 Subject: [PATCH] =?UTF-8?q?chore(scripts):=20=E6=8F=90=E4=BA=A4=E5=88=86?= =?UTF-8?q?=E8=A7=A3=E5=BC=B9=E7=AA=97E2E=E9=AA=8C=E8=AF=81=E8=BE=85?= =?UTF-8?q?=E5=8A=A9=E8=84=9A=E6=9C=AC=EF=BC=88cdp=5Fdriver=20+=20review?= =?UTF-8?q?=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- backend/scripts/cdp_driver.py | 93 +++++++++++++++++++ .../scripts/verify_decompose_dialog_review.py | 79 ++++++++++++++++ 2 files changed, 172 insertions(+) create mode 100644 backend/scripts/cdp_driver.py create mode 100644 backend/scripts/verify_decompose_dialog_review.py diff --git a/backend/scripts/cdp_driver.py b/backend/scripts/cdp_driver.py new file mode 100644 index 00000000..fd80f875 --- /dev/null +++ b/backend/scripts/cdp_driver.py @@ -0,0 +1,93 @@ +"""CDP e2e 续：预算管理页 → 年度分解弹窗实测""" +import json +import sys +import time +import urllib.request +import websocket + +CDP_HTTP = "http://127.0.0.1:9222" + + +def http_get(path): + with urllib.request.urlopen(CDP_HTTP + path, timeout=5) as r: + return json.loads(r.read().decode()) + + +class CDP: + def __init__(self, ws_url): + self.ws = websocket.create_connection(ws_url, timeout=30) + self.msg_id = 0 + + def call(self, method, params=None): + self.msg_id += 1 + mid = self.msg_id + self.ws.send(json.dumps({"id": mid, "method": method, "params": params or {}})) + while True: + resp = json.loads(self.ws.recv()) + if resp.get("id") == mid: + if "error" in resp: + raise RuntimeError(f"{method}: {resp['error']}") + return resp.get("result", {}) + if resp.get("method") in ("Page.loadEventFired", "Page.frameStoppedLoading"): + pass + + def eval(self, expr): + r = self.call("Runtime.evaluate", {"expression": expr, "returnByValue": True, "awaitPromise": True}) + if r.get("exceptionDetails"): + return "EXC: " + json.dumps(r["exceptionDetails"], ensure_ascii=False)[:200] + return r.get("result", {}).get("value") + + def close(self): + try: + self.ws.close() + except Exception: + pass + + +def new_page(url): + ver = http_get("/json/version") + bws = CDP(ver["webSocketDebuggerUrl"]) + t = bws.call("Target.createTarget", {"url": url}) + tid = t["targetId"] + bws.close() + time.sleep(2) + for p in http_get("/json"): + if p["id"] == tid: + return tid, p["webSocketDebuggerUrl"] + return tid, None + + +def main(): + tid, ws = new_page("https://cma.sxbh.ltd/") + if not ws: + print("FAIL: no ws") + sys.exit(1) + c = CDP(ws) + c.call("Page.enable") + c.call("Runtime.enable") + time.sleep(6) + + # 点击侧边栏"预算管理" + r = c.eval("""(() => { + const els = Array.from(document.querySelectorAll('a, li, span, div')); + const target = els.find(e => e.innerText && e.innerText.trim() === '预算管理' && e.offsetParent !== null); + if (!target) return 'NO_MENU'; + target.click(); + return 'CLICKED'; + })()""") + print("MENU_CLICK:", r) + time.sleep(4) + print("URL_NOW:", c.eval("location.href")) + body = c.eval("document.body ? document.body.innerText.slice(0,400) : ''") + print("BODY:", body.replace("\n", " | ")[:400]) + + # 检查是否有 预算管理 页面关键元素：年份选择 + 年度分解按钮 + btns = c.eval("""Array.from(document.querySelectorAll('button')).map((b,idx)=>({idx, text:(b.innerText||'').trim().slice(0,20)})).filter(x=>x.text)""") + print("BUTTONS:", json.dumps(btns, ensure_ascii=False)) + + c.close() + print("TARGET:", tid) + + +if __name__ == "__main__": + main() diff --git a/backend/scripts/verify_decompose_dialog_review.py b/backend/scripts/verify_decompose_dialog_review.py new file mode 100644 index 00000000..fe25fd8f --- /dev/null +++ b/backend/scripts/verify_decompose_dialog_review.py @@ -0,0 +1,79 @@ +"""budget-decompose-dialog-fix 独立复核：API 级实测 auto-decompose 全链路""" +import json +import sys +import urllib.request + +BASE = "http://127.0.0.1:8010" + + +def post(path, body, token=None, method="POST"): + req = urllib.request.Request( + BASE + path, + data=json.dumps(body).encode("utf-8"), + method=method, + headers={ + "Content-Type": "application/json", + "Authorization": f"Bearer {token}" if token else "", + }, + ) + try: + with urllib.request.urlopen(req, timeout=30) as resp: + return resp.status, json.loads(resp.read().decode("utf-8")) + except urllib.error.HTTPError as e: + try: + return e.code, json.loads(e.read().decode("utf-8")) + except Exception: + return e.code, {"detail": e.read().decode("utf-8", "ignore")} + + +def main(): + # 1. 登录（账套模式 entity_id=1） + status, login = post("/api/cma/auth/login", {"username": "admin", "password": "admin123", "entity_id": 1}) + token = login.get("token") or login.get("access_token") + if status != 200 or not token: + print("FAIL login:", status, login) + sys.exit(1) + print("PASS 登录成功, token 前缀:", token[:12], "...") + + # 2. 调用 auto-decompose（equal 均分） + status, r = post("/api/cma/budget/auto-decompose", {"year": 2026, "method": "equal", "version": "v1.0"}, token) + print("auto-decompose status:", status) + if status != 200: + print(" detail:", r.get("detail", r)) + print("FAIL auto-decompose 非200（可能该年无年度预算数据）") + sys.exit(2) + + print(" message:", r.get("message")) + results = r.get("results") or [] + print(" created:", r.get("created"), " results数:", len(results)) + for res in results[:5]: + print(" -", res.get("kpi_code"), res.get("kpi_name"), + "annual=", res.get("annual_budget"), "method=", res.get("method"), + "monthly_count=", len(res.get("monthly") or [])) + if not results: + print("FAIL results 为空") + sys.exit(3) + + # 3. 验证每条结果字段完整（前端表格依赖） + required = ["kpi_code", "kpi_name", "annual_budget", "method", "monthly"] + for res in results: + missing = [k for k in required if k not in res] + if missing: + print("FAIL 结果缺字段:", missing, res) + sys.exit(4) + if not res.get("monthly"): + print("FAIL monthly 为空:", res.get("kpi_code")) + sys.exit(5) + print("PASS 所有结果字段完整（kpi_code/kpi_name/annual_budget/method/monthly）") + + # 4. 幂等抽查：再调一次，结果一致 + status2, r2 = post("/api/cma/budget/auto-decompose", {"year": 2026, "method": "equal", "version": "v1.0"}, token) + snap1 = {res["kpi_id"]: tuple(res.get("monthly") or []) for res in results} + snap2 = {res["kpi_id"]: tuple(res.get("monthly") or []) for res in (r2.get("results") or [])} + print("PASS 二次调用幂等一致" if snap1 == snap2 else "WARN 二次调用结果不同（非幂等）") + + print("\nRESULT: API 全链路通过") + + +if __name__ == "__main__": + main()