feat: RBAC 权限机制重构——应用仓库定义 r:p 数据 + 统一导入脚本

build.sh

@@ -132,12 +132,12 @@ for d in sd_projects sd_iterations; do
 done
 
 
-# 11. Initialize RBAC path permissions (all modules' scripts/load_path.py)
-echo "=== RBAC permission init ==="
-if [ -f "$cdir/load_path.sh" ]; then
-    bash "$cdir/load_path.sh" || echo "  WARN: load_path.sh failed (DB not ready?)"
+# 11. Import RBAC permissions from conf/rp.json (r:p 数据 -> permission + rolepermission)
+echo "=== RBAC permission import ==="
+if [ -f "$cdir/scripts/import_rp.py" ]; then
+    "$cdir/py3/bin/python" "$cdir/scripts/import_rp.py" || echo "  WARN: import_rp.py failed (DB not ready?)"
 else
-    echo "  WARN: load_path.sh not found"
+    echo "  WARN: scripts/import_rp.py not found"
 fi
 
 chmod +x "$cdir/start.sh" "$cdir/stop.sh" 2>/dev/null || true

conf/rp.json

@@ -0,0 +1,38 @@
+{
+  "_comment": "pipeline-app RBAC 权限定义（r:p 数据）。角色 -> 路径模式。路径模式支持：精确匹配 /pipeline-sdlc/index.ui；前缀匹配 /product_management/** 或 /pricing/%（匹配该前缀下所有路径）。由 scripts/import_rp.py 导入 permission + rolepermission 表。",
+  "roles": {
+    "any": [
+      "/rbac/user/login.ui",
+      "/rbac/user/up_login.dspy",
+      "/rbac/user/userpassword_login.dspy",
+      "/rbac/user/userpassword_login.ui",
+      "/rbac/phone_login.dspy",
+      "/rbac/gen_sms_code.dspy",
+      "/rbac/qr_scan.ui",
+      "/i18n_getmsgs",
+      "/appbase/menu.ui",
+      "/product_management/menu.ui",
+      "/discount/menu.ui",
+      "/pricing/menu.ui",
+      "/unipay/usermenu.ui"
+    ],
+    "logined": [
+      "/pipeline-sdlc/**",
+      "/product_management/**",
+      "/discount/**",
+      "/pricing/**",
+      "/unipay/**",
+      "/appbase/**",
+      "/showcase/**",
+      "/tenant/**",
+      "/pipeline_core/**",
+      "/pipeline_ops/**",
+      "/pipeline_dist/**",
+      "/pipeline_task/**",
+      "/app_audit/**"
+    ],
+    "owner.superuser": [
+      "/rbac/**"
+    ]
+  }
+}

scripts/import_rp.py

@@ -0,0 +1,56 @@
+#!/usr/bin/env python3
+"""统一 RBAC 权限导入脚本（r:p 数据 -> 数据库）。
+
+读取 conf/rp.json 的角色->路径模式，写入 permission + rolepermission 表。
+幂等：path 已存在则复用 permid，roleid+permid 已存在则跳过。
+
+用法：
+  py3/bin/python scripts/import_rp.py
+"""
+import sys, os, json, asyncio
+
+SCRIPT_DIR = os.path.dirname(os.path.abspath(__file__))
+ROOT_DIR = os.path.dirname(SCRIPT_DIR)
+sys.path.insert(0, os.path.join(ROOT_DIR, 'py3', 'lib', 'python3.10', 'site-packages'))
+sys.path.insert(0, ROOT_DIR)
+
+from sqlor.dbpools import DBPools
+from appPublic.jsonConfig import getConfig
+from appPublic.folderUtils import ProgramPath
+from appPublic.uniqueID import getID
+from ahserver.serverenv import ServerEnv
+from ahserver.globalEnv import initEnv
+
+
+async def main():
+    config = getConfig(ROOT_DIR, NS={'workdir': ROOT_DIR, 'ProgramPath': ProgramPath()})
+    DBPools(config.databases)
+    initEnv()
+    env = ServerEnv()
+    env.get_module_dbname = lambda m: 'pipeline' if 'pipeline' in m else 'sage'
+
+    rp_file = os.path.join(ROOT_DIR, 'conf', 'rp.json')
+    rp = json.load(open(rp_file, encoding='utf-8'))
+    roles = rp['roles']
+
+    async with DBPools().sqlorContext('pipeline') as sor:
+        n_perm = 0
+        n_rp = 0
+        for role, paths in roles.items():
+            for path in paths:
+                recs = await sor.R('permission', {'path': path})
+                if not recs:
+                    permid = getID()
+                    await sor.C('permission', {'id': permid, 'path': path})
+                    n_perm += 1
+                else:
+                    permid = recs[0].id
+                rp_rec = await sor.R('rolepermission', {'roleid': role, 'permid': permid})
+                if not rp_rec:
+                    await sor.C('rolepermission', {'id': getID(), 'roleid': role, 'permid': permid})
+                    n_rp += 1
+    print(f'import_rp: {len(roles)} roles, {n_perm} new permissions, {n_rp} new role-permissions')
+
+
+if __name__ == '__main__':
+    asyncio.run(main())