yumoqing
3c9fdcf444
- Add SDLC dashboard, pipeline editor, ops center UI - build.sh: clone business modules to pkgs/, xls2ui CRUD, fix created_by - global_func.py: password_encode None-safe wrapper - pipeline_app.py: permission cache warmup removed - load_path.py: RBAC permission registration for all modules - scripts/merge_i18n.py: i18n merge tool - bin/init_perms.py, bin/init_data.py: init scripts - set_role_perm.py: single permission registration - Model uitype fields set for form editing - pipeline_core/pipeline_ops load_path.py scripts
63 lines
2.0 KiB
Python
63 lines
2.0 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Register a single RBAC permission for pipeline modules.
|
|
Called by init_perms.py or directly.
|
|
|
|
Usage:
|
|
python set_role_perm.py <role> <path>
|
|
python set_role_perm.py admin /pipeline_core/pipelines/*.dspy
|
|
"""
|
|
import sys, os, asyncio
|
|
|
|
SCRIPT_DIR = os.path.dirname(os.path.abspath(__file__))
|
|
ROOT_DIR = os.path.dirname(SCRIPT_DIR) if os.path.basename(SCRIPT_DIR) == 'bin' else 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(role, path):
|
|
config = getConfig(os.path.join(ROOT_DIR, 'conf', 'config.json'),
|
|
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'
|
|
|
|
# Determine module from path
|
|
parts = path.strip('/').split('/')
|
|
module = parts[0] if parts else 'app'
|
|
|
|
async with DBPools().sqlorContext('sage') as sor:
|
|
existing = await sor.sqlExe(
|
|
"SELECT id FROM role_path WHERE module=${m}$ AND path=${p}$ AND role_name=${r}$",
|
|
{'m': module, 'p': path, 'r': role})
|
|
if existing:
|
|
print(f'Permission exists: {role} {path}')
|
|
return
|
|
|
|
await sor.C('role_path', {
|
|
'id': getID(),
|
|
'module': module,
|
|
'path': path,
|
|
'role_name': role,
|
|
'description': f'Auto-registered: {role} -> {path}'
|
|
})
|
|
print(f'Registered: {role} -> {path}')
|
|
|
|
|
|
if __name__ == '__main__':
|
|
if len(sys.argv) != 3:
|
|
print(f"Usage: {sys.argv[0]} <role> <path>")
|
|
print(f" roles: guest, logined, admin")
|
|
print(f" path example: /pipeline_core/index.ui")
|
|
sys.exit(1)
|
|
|
|
asyncio.run(main(sys.argv[1], sys.argv[2]))
|