57 lines
2.0 KiB
Python
57 lines
2.0 KiB
Python
#!/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())
|