iAOP/deploy/fba/seed_iaop_menus.py

#!/usr/bin/env python3
"""iAOP 菜单/角色播种到 FBA（Epic #159 · Phase 2）。

把 iAOP 的 6 个功能模块注册为 FBA sys_menu，并创建 engineer / viewer 角色、
按 PRD 8.2 三级权限分配菜单，使 Sider 菜单可由 FBA 服务端按角色授权驱动
（前端 web/shared/session.js 的 menuTree() 消费 /sys/menus/sidebar）。

命名约定（与 session.js 的匹配逻辑对应，改动需双侧同步）：
  · 模块菜单 name  = 模块 key（cockpit/chat/studio/admin/users/mobile）
  · 子菜单 name    = 子页 key（import/hyper/layout/version/models/kb/alerts/audit）
  · studio/admin 为目录（type=0）挂子菜单（type=1），其余为叶子菜单（type=1）

幂等：按 name 查重，已存在的菜单/角色跳过创建，只补齐缺失项；
     角色菜单关联每次按目标集合全量刷新（PUT /sys/roles/{id}/menus）。

用法（服务器上执行，仅需 Python 3.8+ 标准库）：
  python3 seed_iaop_menus.py \
      --base http://127.0.0.1:8001/api/v1 \
      --username admin --password '<FBA管理员密码>'

说明：登录走 /auth/login/swagger（HTTP Basic，免图形验证码），
专为运维/调试通道；不依赖 LOGIN_CAPTCHA_ENABLED 开关。
"""

import argparse
import base64
import getpass
import json
import sys
import urllib.error
import urllib.parse
import urllib.request

# ---- iAOP 模块注册表 v2（Epic #163 原生重构：type=1 原生菜单，component→views/iaop/*） ----
# path=/iaop/<key>，component=iaop/<key>/index（与 apps/web-antdv-next/src/views/iaop/* 对应）；
# icon 为 iconify 名；users 不注册（FBA 系统管理承载）、mobile 暂不注册（见 B9）。
MODULES = [
    {"key": "cockpit", "title": "配置化驾驶舱", "icon": "mdi:view-dashboard",
     "path": "/iaop/cockpit", "component": "iaop/cockpit/index"},
    {"key": "chat", "title": "对话助手", "icon": "mdi:chat-processing-outline",
     "path": "/iaop/chat", "component": "iaop/chat/index"},
    {"key": "studio", "title": "模板配置台", "icon": "mdi:widgets-outline",
     "path": "/iaop/studio", "component": "iaop/studio/index", "children": [
        {"key": "import", "title": "点位导入", "component": "iaop/studio/import"},
        {"key": "hyper", "title": "模型超参", "component": "iaop/studio/hyper"},
        {"key": "layout", "title": "驾驶舱编排", "component": "iaop/studio/layout"},
        {"key": "version", "title": "版本发布", "component": "iaop/studio/version"},
    ]},
    {"key": "admin", "title": "管理控制台", "icon": "mdi:cog-outline",
     "path": "/iaop/admin", "component": "iaop/admin/index", "children": [
        {"key": "models", "title": "模型管理", "component": "iaop/admin/models"},
        {"key": "kb", "title": "知识库管理", "component": "iaop/admin/kb"},
        {"key": "alerts", "title": "告警确认", "component": "iaop/admin/alerts"},
        {"key": "audit", "title": "审计查询", "component": "iaop/admin/audit"},
    ]},
]

# 角色 → 可见 iAOP 模块（PRD 8.2；admin 角色为新增业务管理员，非超管）
ROLE_MENUS = {
    "viewer": ["cockpit", "chat"],
    "engineer": ["cockpit", "chat", "studio"],
    "admin": ["cockpit", "chat", "studio", "admin"],
}

# admin 角色额外分配：FBA 原生管理菜单（存在即分配）
FBA_ADMIN_MENUS = ["system", "monitor", "log", "scheduler"]

# 退役菜单（issue #174 [B8]）：旧 #161 版注册的 users/mobile 改为隐藏（display=0），保留数据
# 另隐藏 fba init 测试数据自带的 概览(Dashboard) 三个菜单——其 component
# 指向的 /dashboard/analytics|workspace 视图在 UI 中不存在，点击必 404。
# 退役/不存在菜单：逻辑删除（deleted=1）
# - Analytics/Workspace/Dashboard（issue #181：FBA init 测试数据自带概览，视图不存在 → 404）
# - Document/Github/Apifox/Project（FBA 推广外链菜单，与业务无关）
# - users/mobile（issue #174 [B8] 退役）
# 注：FBA sidebar 查询仅过滤 deleted=0（不过滤 display/status），display=0 对超管无效，
#     必须删除才从 sidebar 消失；先删子菜单再删父，否则 409。
LEGACY_DELETE = ["Analytics", "Workspace", "Document", "Github", "Apifox",
                 "Dashboard", "Project", "users", "mobile"]


class FbaClient:
    def __init__(self, base: str, username: str, password: str):
        self.base = base.rstrip("/")
        cred = base64.b64encode(f"{username}:{password}".encode()).decode()
        # FastAPI >= 0.13x 将 HTTPBasicCredentials 绑定到 query 参数，
        # 登录需带 ?username=&password=（Basic 头仍保留，双通道兼容）。
        q = urllib.parse.urlencode({"username": username, "password": password})
        # 注意：/auth/login/swagger 返回裸 GetSwaggerToken（无 {code,msg} 包装），
        # 不能走 _req 的通用 code==200 校验。
        req = urllib.request.Request(
            f"{self.base}/auth/login/swagger?{q}",
            headers={"Content-Type": "application/json",
                     "Authorization": f"Basic {cred}"},
            method="POST")
        try:
            with urllib.request.urlopen(req, timeout=15) as r:
                d = json.loads(r.read().decode())
        except urllib.error.HTTPError as e:
            detail = e.read().decode(errors="replace")[:300]
            raise SystemExit(f"[fail] POST /auth/login/swagger → HTTP {e.code}: {detail}")
        self.token = d["access_token"]
        print(f"[ok] 登录成功：{username}（{d['user']['username']}）")

    def _req(self, method: str, path: str, body=None, headers=None):
        h = {"Content-Type": "application/json"}
        if getattr(self, "token", None):
            h["Authorization"] = f"Bearer {self.token}"
        if headers:
            h.update(headers)
        data = json.dumps(body).encode() if body is not None else None
        req = urllib.request.Request(self.base + path, data=data,
                                     headers=h, method=method)
        try:
            with urllib.request.urlopen(req, timeout=15) as r:
                d = json.loads(r.read().decode())
        except urllib.error.HTTPError as e:
            detail = e.read().decode(errors="replace")[:300]
            raise SystemExit(f"[fail] {method} {path} → HTTP {e.code}: {detail}")
        if d.get("code") != 200:
            raise SystemExit(f"[fail] {method} {path} → {d.get('msg')}")
        return d

    def get(self, path):
        return self._req("GET", path)["data"]

    def post(self, path, body):
        return self._req("POST", path, body)

    def put(self, path, body):
        return self._req("PUT", path, body)

    def delete(self, path):
        return self._req("DELETE", path, None)


def flatten_menu_names(tree, out=None):
    """菜单树 → {name: id}（含子级）"""
    out = out if out is not None else {}
    for node in tree or []:
        if node.get("name"):
            out[node["name"]] = node["id"]
        flatten_menu_names(node.get("children"), out)
    return out


def main():
    ap = argparse.ArgumentParser(description="iAOP 菜单/角色播种到 FBA")
    ap.add_argument("--base", default="http://127.0.0.1:8001/api/v1",
                    help="FBA API 根（直连容器端口用 8001；走 nginx 用 http://<host>/fba/api/v1）")
    ap.add_argument("--username", default="admin")
    ap.add_argument("--password", default=None, help="缺省时交互输入")
    args = ap.parse_args()
    password = args.password or getpass.getpass("FBA 管理员密码: ")

    cli = FbaClient(args.base, args.username, password)

    # ---- 1. 播种菜单（幂等：按 name 查重） --------------------------------
    existing = flatten_menu_names(cli.get("/sys/menus"))
    created, updated = 0, 0

    def upsert_menu(name, title, path, mtype, sort, parent_id=None,
                    component=None, icon=None, perms=None):
        nonlocal created, updated
        body = {
            "title": title, "name": name, "path": path,
            "parent_id": parent_id, "sort": sort, "icon": icon,
            "type": mtype,           # 0目录 1菜单
            "component": component, "perms": perms or f"iaop:{name}",
            "status": 1, "display": 1, "cache": 0,
            "link": None, "remark": "iAOP 原生菜单（seed_iaop_menus.py v2）",
        }
        if name in existing:
            # 幂等：已存在（含 #161 旧版 type=0/iframe 菜单）→ PUT 更新关键字段
            mid = existing[name]
            cli.put(f"/sys/menus/{mid}", body)
            updated += 1
            print(f"[upd] 更新菜单 {name}（{title}）")
            return mid
        cli.post("/sys/menus", body)
        created += 1
        # 创建后重新拉取以拿到 id（接口不回传 id）
        nonlocal_existing = flatten_menu_names(cli.get("/sys/menus"))
        existing.update(nonlocal_existing)
        print(f"[ok] 创建菜单 {name}（{title}）")
        return existing[name]

    for i, m in enumerate(MODULES):
        children = m.get("children")
        if children:
            pid = upsert_menu(m["key"], m["title"], m["path"], 0, i,
                              component=m.get("component"), icon=m.get("icon"))
            for j, c in enumerate(children):
                upsert_menu(c["key"], c["title"],
                            f"{m['path']}/{c['key']}", 1, j,
                            parent_id=pid, component=c.get("component"),
                            icon=c.get("icon"))
        else:
            upsert_menu(m["key"], m["title"], m["path"], 1, i,
                        component=m.get("component"), icon=m.get("icon"))
    print(f"[ok] 菜单：新建 {created}，更新 {updated}")

    # 退役/不存在菜单：逻辑删除（issue #174 [B8] + issue #181）
    for legacy in LEGACY_DELETE:
        if legacy in existing:
            mid = existing[legacy]
            cli.delete(f"/sys/menus/{mid}")
            print(f"[ok] 删除退役菜单 {legacy}（id={mid}，deleted=1）")

    # ---- 2. 播种角色并分配菜单 -------------------------------------------
    menu_ids = flatten_menu_names(cli.get("/sys/menus"))

    def module_menu_ids(mod_key):
        """模块目录 id + 其子菜单 id（叶子模块仅自身 id）"""
        ids = [menu_ids[mod_key]]
        mod = next(m for m in MODULES if m["key"] == mod_key)
        for c in mod.get("children", []):
            ids.append(menu_ids[c["key"]])
        return ids

    roles = {r["name"]: r["id"] for r in cli.get("/sys/roles/all")}
    for role_name, mod_keys in ROLE_MENUS.items():
        if role_name in roles:
            rid = roles[role_name]
            print(f"[ok] 角色已存在：{role_name}（id={rid}），刷新菜单关联")
        else:
            cli.post("/sys/roles", {
                "name": role_name, "status": 1,
                "is_filter_scopes": True,
                "remark": "iAOP 角色（seed_iaop_menus.py）",
            })
            roles = {r["name"]: r["id"] for r in cli.get("/sys/roles/all")}
            rid = roles[role_name]
            print(f"[ok] 创建角色：{role_name}（id={rid}）")
        ids = []
        for k in mod_keys:
            ids.extend(module_menu_ids(k))
        # admin 角色额外分配 FBA 原生管理菜单（system/monitor/log/scheduler 等）
        if role_name == "admin":
            for fba_name in FBA_ADMIN_MENUS:
                if fba_name in menu_ids:
                    ids.append(menu_ids[fba_name])
        cli.put(f"/sys/roles/{rid}/menus", {"menus": ids})
        print(f"[ok] 角色 {role_name} 菜单：{', '.join(mod_keys)}（{len(ids)} 项）")

    # ---- 3. 验证 sidebar ---------------------------------------------------
    side = cli.get("/sys/menus/sidebar")
    names = [n.get("name") for n in side or []]
    print(f"[ok] 当前管理员 sidebar 可见菜单：{names}")

    print("\n完成。后续：在 FBA 后台创建用户并分配 engineer/viewer 角色，"
          "前端 Sider 即按服务端授权渲染（admin 为超管可见全部）。")


if __name__ == "__main__":
    sys.exit(main())