#!/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 '' 说明：登录走 /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/，component=iaop//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"] 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 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:///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}") # ---- 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())