cma-management/backend/scripts/verify_suggestioncenter_unapplied_ui.py

#!/usr/bin/env python3
"""UI实证: AI建议中心(SuggestionCenter) 待批列表默认展示 unapplied + 类型过滤 (P0-② M3)

前置: 后端已部署新代码(8010) + 前端 dist 已 cp /var/www/cma
      Chrome 已启动: --remote-debugging-port=9225 --host-resolver-rules="MAP cma.sxbh.ltd 127.0.0.1" --ignore-certificate-errors
      存在一条 unapplied 建议(title含 E2E-验证), 如 SUG_ID=300
用法: /opt/hermes-agent/bin/python3.12 scripts/verify_suggestioncenter_unapplied_ui.py
"""
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:9225"
TITLE_NEEDLE = "E2E-验证 建议中心UI实证"


def http_json(path):
    with urllib.request.urlopen(CDP_HTTP + path, timeout=8) 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", {})
            if resp.get("method") == "Page.loadEventFired":
                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)[:300]
        return r.get("result", {}).get("value")

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


def main():
    ver = http_json("/json/version")
    browser = CDP(ver["webSocketDebuggerUrl"])
    t = browser.call("Target.createTarget", {"url": "https://cma.sxbh.ltd/login"})
    tid = t["targetId"]
    browser.close()
    time.sleep(3)
    ws_url = None
    for p in http_json("/json"):
        if p["id"] == tid:
            ws_url = p["webSocketDebuggerUrl"]
            break
    if not ws_url:
        print("FAIL: no page ws")
        sys.exit(1)
    c = CDP(ws_url)
    c.call("Page.enable")
    c.call("Runtime.enable")
    # 显式导航到登录页（Target.createTarget 偶发停在 about:blank）
    c.call("Page.navigate", {"url": "https://cma.sxbh.ltd/login"})
    time.sleep(6)

    # 同源API登录 → localStorage → 跳 /ai-suggestions
    js_login = r"""
    (async () => {
      const d = await (await fetch('/api/cma/auth/login', {method:'POST',
        headers:{'Content-Type':'application/json'},
        body: JSON.stringify({username:'__CMA_ADMIN_USER__', password:'__CMA_ADMIN_PASS__', entity_id:1})
      })).json();
      if (!d.token) return 'LOGIN_FAIL: ' + JSON.stringify(d).slice(0,200);
      localStorage.setItem('cma_token', d.token);
      localStorage.setItem('cma_entity_id', String(d.entity_id));
      localStorage.setItem('cma_entity_name', d.entity_name || '');
      localStorage.setItem('cma_user', JSON.stringify(d.user || {}));
      location.href = '/ai-suggestions';
      return 'LOGIN_OK entity=' + d.entity_id;
    })()
    """.replace("__CMA_ADMIN_USER__", ADMIN_USER).replace("__CMA_ADMIN_PASS__", ADMIN_PASS)
    print("LOGIN:", c.eval(js_login))
    time.sleep(6)
    print("URL:", c.eval("location.href"))
    body = c.eval("document.body ? document.body.innerText : ''") or ""
    print("BODY_SNIP:", body.replace("\n", " | ")[:800])

    checks = []
    checks.append(("页面含AI建议中心标题", "AI决策建议中心" in body))
    checks.append(("未应用筛选默认选中(radio aria/class)", 'class="el-radio-button__original-radio" checked' in body or '未应用' in body))
    checks.append(("类型过滤芯片存在", "全部类型" in body and "改KPI目标" in body and "调预算" in body))
    checks.append(("待批列表可见unapplied建议", TITLE_NEEDLE in body))

    # 结构性断言(不依赖文本顺序): 卡片存在 + 卡片含建议标题
    dom = c.eval("""(() => {
      const cards = Array.from(document.querySelectorAll('.sug-card'));
      const hit = cards.map(x => (x.innerText || '')).filter(t => t.indexOf('""" + TITLE_NEEDLE + """') >= 0);
      const radios = Array.from(document.querySelectorAll('.el-radio-button')).map(x => (x.innerText||'').trim());
      return {card_hit: hit.length, card_text: hit[0] ? hit[0].slice(0, 300) : '', radios: radios};
    })()""")
    print("DOM_STRUCT:", json.dumps(dom, ensure_ascii=False)[:900])
    if isinstance(dom, dict):
        checks.append(("卡片DOM命中", dom.get("card_hit", 0) >= 1))
        radios = dom.get("radios", [])
        checks.append(("类型radio组含全部类型/三类型", all(x in radios for x in ("全部类型", "改KPI目标", "调预算", "建行动方案"))))
        checks.append(("状态radio组含未应用/已应用", "未应用" in radios and "已应用" in radios))

    print()
    fails = [n for n, ok in checks if not ok]
    for n, ok in checks:
        print(("  [PASS] " if ok else "  [FAIL] ") + n)
    c.close()
    print()
    if fails:
        print("UI RESULT: FAIL:", fails)
        sys.exit(1)
    print("UI RESULT: PASS — 待批列表默认展示 unapplied 建议, 类型过滤可用")


if __name__ == "__main__":
    main()