71 lines
2.4 KiB
Python
71 lines
2.4 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""初始化/更新 admin 管理员账号（RBAC 标准接口，幂等 upsert）。
|
||
|
||
用法: python scripts/init_admin_user.py
|
||
- 走 rbac 标准用户表 rbac_user（密码经 ahserver password_encode 统一编码）
|
||
- 不存在 → 创建；存在 → 更新密码/启用状态（幂等，可重复执行）
|
||
- 为 admin 绑定 logined 角色（rbac_user_role），使登录后能访问业务接口
|
||
- 凭证来源 .env：SCENSE_ADMIN_USER / SCENSE_ADMIN_PASSWORD（默认 admin/admin123）
|
||
"""
|
||
import os
|
||
import sys
|
||
|
||
# 允许直接运行：python scripts/init_admin_user.py
|
||
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
||
|
||
from ahserver.serverenv import ServerEnv
|
||
from ahserver.globalEnv import password_encode
|
||
from sqlor import sor
|
||
|
||
DBNAME = os.environ.get('MYSQL_DATABASE', 'scense')
|
||
|
||
|
||
def upsert_admin(env):
|
||
username = os.environ.get('SCENSE_ADMIN_USER', 'admin')
|
||
password = os.environ.get('SCENSE_ADMIN_PASSWORD', 'admin123')
|
||
enc = password_encode(password)
|
||
|
||
rows = sor.R(DBNAME, 'rbac_user', {'username': username})
|
||
if rows:
|
||
sor.U(DBNAME, 'rbac_user', {'username': username},
|
||
{'password': enc, 'status': 1})
|
||
print('admin user updated: ' + username)
|
||
else:
|
||
sor.C(DBNAME, 'rbac_user', {
|
||
'username': username,
|
||
'password': enc,
|
||
'nickname': 'Administrator',
|
||
'status': 1,
|
||
})
|
||
print('admin user created: ' + username)
|
||
return username
|
||
|
||
|
||
def bind_logined_role(username):
|
||
"""绑定 logined 角色（幂等），确保登录后具备业务接口访问权限。"""
|
||
users = sor.R(DBNAME, 'rbac_user', {'username': username})
|
||
if not users:
|
||
return
|
||
uid = users[0].get('id')
|
||
if uid is None:
|
||
return
|
||
roles = sor.R(DBNAME, 'rbac_role', {'code': 'logined'})
|
||
if not roles:
|
||
sor.C(DBNAME, 'rbac_role', {'code': 'logined', 'name': '已登录用户'})
|
||
binds = sor.R(DBNAME, 'rbac_user_role', {'user_id': uid, 'role_code': 'logined'})
|
||
if not binds:
|
||
sor.C(DBNAME, 'rbac_user_role', {'user_id': uid, 'role_code': 'logined'})
|
||
print('rbac role logined bound for ' + username)
|
||
|
||
|
||
def main():
|
||
env = ServerEnv()
|
||
env.get_module_dbname = env.get_module_dbname or (lambda m: DBNAME)
|
||
username = upsert_admin(env)
|
||
bind_logined_role(username)
|
||
print('OK: admin initialized (username=' + username + ', db=' + DBNAME + ')')
|
||
|
||
|
||
if __name__ == '__main__':
|
||
main()
|