From dc89b14479eb37d0e8f61c780c5f05bb11e395a2 Mon Sep 17 00:00:00 2001
From: Hermes CI Fix <hermes-ci@sxbh.ltd>
Date: Mon, 31 Aug 2026 15:57:54 +0800
Subject: [PATCH] =?UTF-8?q?chore(scripts):=20=E6=8F=90=E4=BA=A4=E5=B7=A5?=
 =?UTF-8?q?=E4=BD=9C=E5=8F=B0=E9=9A=94=E7=A6=BB=E9=AA=8C=E8=AF=81=E8=84=9A?=
 =?UTF-8?q?=E6=9C=AC=20verify=5Fdashboard=5Fisolation.py=EF=BC=88dashboard?=
 =?UTF-8?q?-optimize=20=E9=81=97=E7=95=99=EF=BC=89?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 backend/scripts/verify_dashboard_isolation.py | 85 +++++++++++++++++++
 1 file changed, 85 insertions(+)
 create mode 100644 backend/scripts/verify_dashboard_isolation.py

diff --git a/backend/scripts/verify_dashboard_isolation.py b/backend/scripts/verify_dashboard_isolation.py
new file mode 100644
index 00000000..12adef57
--- /dev/null
+++ b/backend/scripts/verify_dashboard_isolation.py
@@ -0,0 +1,85 @@
+#!/usr/bin/env python3
+# -*- coding: utf-8 -*-
+"""my-dashboard 隔离补漏验证: 登录 entity1 -> 工作台 -> 回查DB归属全为 entity1"""
+import json
+import urllib.request
+import pymysql
+
+BASE = "http://127.0.0.1:8010"
+
+
+def post(path, data):
+    req = urllib.request.Request(
+        BASE + path,
+        data=json.dumps(data).encode(),
+        headers={"Content-Type": "application/json"},
+    )
+    with urllib.request.urlopen(req, timeout=30) as resp:
+        return json.loads(resp.read().decode())
+
+
+def get(path, token):
+    req = urllib.request.Request(
+        BASE + path, headers={"Authorization": "Bearer " + token}
+    )
+    with urllib.request.urlopen(req, timeout=30) as resp:
+        return json.loads(resp.read().decode())
+
+
+def main():
+    # 1. 登录 entity1
+    r = post("/api/cma/auth/login", {"username": "admin", "password": "admin123", "entity_id": 1})
+    token = r.get("access_token") or r.get("token")
+    assert token, f"登录失败: {r}"
+    print("1. 登录 entity1 OK")
+
+    # 2. 工作台
+    d = get("/api/cma/dashboard/my-dashboard", token)
+    plans = d.get("action_plans", [])
+    reminders = d.get("reminders", [])
+    print(f"2. my-dashboard OK: action_plans={len(plans)} 条, reminders={len(reminders)} 条")
+
+    conn = pymysql.connect(host="127.0.0.1", user="cma_user", password="cma_pass_2026",
+                           database="cma", charset="utf8mb4")
+    cur = conn.cursor()
+
+    # 3. action_plans 全部归属 entity1
+    ids = [p["id"] for p in plans]
+    if ids:
+        fmt = ",".join(["%s"] * len(ids))
+        cur.execute(f"SELECT id, entity_id FROM action_plans WHERE id IN ({fmt})", ids)
+        rows = cur.fetchall()
+        bad = [r for r in rows if r[1] != 1]
+        print(f"3. action_plans 回查DB归属: 非entity1 = {bad if bad else '无'}")
+    else:
+        print("3. action_plans 返回 0 条 (跳过)")
+
+    # 4. reminders 中 action_plan 的 related_id 归属
+    plan_rids = [r["related_id"] for r in reminders if r.get("related_type") == "action_plan"]
+    isolation_bad = False
+    if plan_rids:
+        fmt = ",".join(["%s"] * len(plan_rids))
+        cur.execute(f"SELECT id, entity_id FROM action_plans WHERE id IN ({fmt})", plan_rids)
+        rows = cur.fetchall()
+        bad = [r for r in rows if r[1] != 1]
+        isolation_bad = bool(bad)
+        print(f"4. reminders.action_plan related_id 共 {len(plan_rids)} 个, 非entity1 = {bad if bad else '无'}")
+    else:
+        print("4. reminders 无 action_plan 类型 (跳过)")
+
+    # 5. reminders 类型/严重度分布 (前端标签/排序数据源)
+    dist = {}
+    for r in reminders:
+        key = (r.get("related_type"), r.get("type"), r.get("severity"))
+        dist[key] = dist.get(key, 0) + 1
+    print("5. reminders 类型分布:", dist)
+
+    cur.close()
+    conn.close()
+
+    assert not isolation_bad, "发现跨账套数据泄漏"
+
+
+if __name__ == "__main__":
+    main()
+    print("PASS: entity1 工作台数据无跨账套泄漏")