cma-management/backend/scripts/verify_budget_versions_ui_browser.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""2026-09-07 预算版本管理UI修复 - 浏览器页面实测
流程: 登录博海(entity2) → /budget 版本管理 → 提取行DOM断言 → switch-entity 到酣客(entity1) → 再提取断言
验收: 博海 v2026H2 行: 状态草稿/条目数25/总额462/最后更新非空/提交审批按钮可见
      酣客 active版本有提交审批; 2026H2 approved 显示已批准 无提交按钮
"""
import json
import sys
import time
import urllib.request
import websocket

import os  # noqa: E402  (凭据加载器需要)
# ---- 管理员凭据同源加载（2026-09-14 cma-defaultcred-code-fix）----
# 生产凭据只存 600 密钥文件；禁止在脚本/文档/日志里硬编码口令。
# 顺序：$CMA_ADMIN_ENV → wecom-ops 主副本（ops 轮换落点，权威） → 本 profile 副本。
def _admin_creds():
    """返回 (user, password, 来源文件)；未找到 → 显式报错退出（不静默退回默认口令）"""
    for p in (os.environ.get("CMA_ADMIN_ENV"),
              "/root/.hermes/profiles/wecom-ops/.secrets/cma-admin.env",
              "/root/.hermes/profiles/wecom-fullstack/.secrets/cma-admin.env"):
        if p and os.path.isfile(p):
            kv = {}
            for line in open(p, encoding="utf-8"):
                line = line.strip()
                if "=" in line and not line.startswith("#"):
                    k, v = line.split("=", 1)
                    kv[k.strip()] = v.strip().strip('"').strip("'")
            if kv.get("CMA_ADMIN_USER") and kv.get("CMA_ADMIN_PASS"):
                return kv["CMA_ADMIN_USER"], kv["CMA_ADMIN_PASS"], p
    raise SystemExit("[错误] 未找到 CMA 管理员凭据：请设 $CMA_ADMIN_ENV，或准备 600 权限密钥文件 "
                     ".secrets/cma-admin.env（键名 CMA_ADMIN_USER / CMA_ADMIN_PASS）")


ADMIN_USER, ADMIN_PASS, CRED_SRC = _admin_creds()
print(f"[creds] 来源={CRED_SRC} 账号={ADMIN_USER}（口令不回显）")

CDP_HTTP = "http://127.0.0.1:9224"
BASE = "https://cma.sxbh.ltd/api/cma"


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=40)
        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", {})

    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)[:300]
        return r.get("result", {}).get("value")

    def close(self):
        try:
            self.ws.close()
        except Exception:
            pass


def open_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.5)
    for p in http_get("/json"):
        if p["id"] == tid:
            return tid, p["webSocketDebuggerUrl"]
    return tid, None


def wait_for(c, expr, timeout=25, interval=1.0):
    """轮询直到 expr 为 truthy，返回其值"""
    for _ in range(int(timeout / interval)):
        v = c.eval(expr)
        if v:
            return v
        time.sleep(interval)
    return None


def login_via_fetch(c, entity_id):
    """同源 fetch 登录并写 localStorage，返回状态文本"""
    expr = f"""
    (async () => {{
      const r = await fetch('{BASE}/auth/login', {{
        method: 'POST', headers: {{'Content-Type': 'application/json'}},
        body: JSON.stringify({{username:'{ADMIN_USER}', password:'{ADMIN_PASS}', entity_id:{entity_id}}})
      }});
      const j = await r.json();
      if (!r.ok) return 'LOGIN_FAIL:' + JSON.stringify(j);
      localStorage.setItem('cma_token', j.token);
      localStorage.setItem('cma_entity_id', String(j.entity_id));
      localStorage.setItem('cma_entity_name', j.entity_name || '');
      localStorage.setItem('cma_user', JSON.stringify(j.user || {{}}));
      return 'LOGIN_OK:' + j.entity_name + ':' + j.token.slice(0, 8);
    }})()
    """
    return c.eval(expr)


JS_GO_BUDGET = """
(() => {
  location.href = '/budget';
  return 'NAV';
})()
"""

JS_EXTRACT = r"""
(() => {
  // 等表格渲染完
  const rows = Array.from(document.querySelectorAll('.el-table__body-wrapper tbody tr'));
  if (!rows.length) return 'NO_ROWS';
  const out = [];
  for (const tr of rows) {
    const tds = Array.from(tr.querySelectorAll('td'));
    if (tds.length < 6) continue;
    const getTxt = (td) => (td.innerText || '').trim().replace(/\s+/g, ' ');
    const version = getTxt(tds[0]);
    const status = getTxt(tds[1]);
    const planCount = getTxt(tds[2]);
    const total = getTxt(tds[3]);
    const updated = getTxt(tds[4]);
    const opTxt = (tds[5] ? tds[5].innerText || '' : '').trim().replace(/\s+/g, ' ');
    const btns = Array.from(tds[5].querySelectorAll('button')).map(b => (b.innerText || '').trim().replace(/\s+/g, ''));
    const checkboxCount = tds[5].querySelectorAll('.el-checkbox').length;
    out.push({ version, status, planCount, total, updated, opTxt, btns, checkboxCount });
  }
  return JSON.stringify(out);
})()
"""

JS_CLICK_VERSION_TAB = """
(() => {
  const tabs = Array.from(document.querySelectorAll('[role="tab"]'));
  const t = tabs.find(e => (e.innerText || '').replace(/\\s/g, '') === '版本管理');
  if (!t) return 'NO_TAB:' + tabs.map(x => x.innerText).join('|');
  t.click();
  return 'TAB_CLICKED';
})()
"""


def main():
    tid, ws = open_page("https://cma.sxbh.ltd/login")
    if not ws:
        print("FAIL: no ws for new page")
        sys.exit(1)
    c = CDP(ws)
    c.call("Page.enable")
    c.call("Runtime.enable")
    # 等待页面导航到 cma 域完成（约 blank → login 竞态修复，2026-09-07 复跑实测：
    # Target.createTarget 后偶发停留在 about:blank → 访问 localStorage 抛 SecurityError）
    nav = wait_for(c, "location.hostname === 'cma.sxbh.ltd'", timeout=25)
    if not nav:
        print("FAIL: 页面未导航到 cma.sxbh.ltd")
        sys.exit(1)
    time.sleep(3)

    # 1) 登录博海 entity=2
    r = login_via_fetch(c, 2)
    print("LOGIN_E2:", r)
    if not (isinstance(r, str) and r.startswith("LOGIN_OK")):
        print("FAIL login entity2")
        sys.exit(1)
    c.eval(JS_GO_BUDGET)
    time.sleep(5)
    print("URL1:", c.eval("location.href"))
    # 确认当前账套
    print("ENTITY1:", c.eval("localStorage.getItem('cma_entity_id')"))
    tab = wait_for(c, JS_CLICK_VERSION_TAB, timeout=15)
    print("TAB1:", tab)
    time.sleep(4)
    rows_json = wait_for(c, JS_EXTRACT, timeout=20)
    print("ROWS_E2_RAW:", rows_json)
    rows_e2 = json.loads(rows_json) if isinstance(rows_json, str) and rows_json.startswith("[") else []
    for row in rows_e2:
        print("  E2", json.dumps(row, ensure_ascii=False))

    # 断言博海
    ok = True
    target = [r for r in rows_e2 if r.get("version", "").strip().startswith("v2026H2")]
    if not target:
        print("FAIL: 博海未见 v2026H2")
        ok = False
    else:
        r = target[0]
        assert "草稿" in r["status"], f"状态应含草稿, got {r['status']}"
        assert r["planCount"].replace(" ", "") == "25", f"条目数应25, got {r['planCount']}"
        assert "462" in r["total"].replace(",", ""), f"总额应462, got {r['total']}"
        assert r["updated"], "最后更新应非空"
        assert "提交审批" in r["opTxt"], f"应有提交审批按钮, got {r['btns']}"
        print("E2_ASSERT_OK:", json.dumps(r, ensure_ascii=False))

    # 2) 切换酣客 entity=1（同页面 fetch switch-entity 或直接重新登录）
    r = login_via_fetch(c, 1)
    print("LOGIN_E1:", r)
    c.eval(JS_GO_BUDGET)
    time.sleep(5)
    print("ENTITY2:", c.eval("localStorage.getItem('cma_entity_id')"))
    tab = wait_for(c, JS_CLICK_VERSION_TAB, timeout=15)
    print("TAB2:", tab)
    time.sleep(4)
    rows_json = wait_for(c, JS_EXTRACT, timeout=20)
    print("ROWS_E1_RAW:", rows_json)
    rows_e1 = json.loads(rows_json) if isinstance(rows_json, str) and rows_json.startswith("[") else []
    for row in rows_e1:
        print("  E1", json.dumps(row, ensure_ascii=False))

    ok2 = True
    for row in rows_e1:
        ver = row.get("version", "").strip()
        if ver == "2026H2":
            if "已批准" not in row["status"] or "提交审批" in row["opTxt"]:
                print(f"FAIL: 酣客2026H2 approved 应只读已批准无提交, got {json.dumps(row, ensure_ascii=False)}")
                ok2 = False
        elif row["status"].replace(" ", "").startswith("草稿"):
            if "提交审批" not in row["opTxt"]:
                print(f"FAIL: 酣客{ver} active/草稿 应有提交审批, got {json.dumps(row, ensure_ascii=False)}")
                ok2 = False
    if not rows_e1:
        ok2 = False
        print("FAIL: 酣客无行")

    print("\n" + ("ALL_PASS" if (ok and ok2) else "HAS_FAIL"))
    c.close()
    sys.exit(0 if (ok and ok2) else 1)


if __name__ == "__main__":
    main()