112 lines
3.8 KiB
Python
112 lines
3.8 KiB
Python
"""RBAC 种子：权限、角色与默认账号。"""
|
|
from __future__ import annotations
|
|
|
|
from sqlalchemy.orm import Session
|
|
|
|
from models import Permission, Role, RolePermission, User, UserRole
|
|
from services.security import hash_password
|
|
|
|
|
|
PERMISSIONS = [
|
|
("learning:use", "使用学习功能", "learning"),
|
|
("users:manage", "用户管理", "admin"),
|
|
("catalog:manage", "教材/课程管理", "admin"),
|
|
("question-bank:manage", "题库管理", "admin"),
|
|
("llm:manage", "模型配置管理", "admin"),
|
|
("knowledge:manage", "知识图谱管理", "admin"),
|
|
]
|
|
|
|
ROLES = {
|
|
"admin": {
|
|
"name": "管理员",
|
|
"description": "拥有用户、模型、教材课程与题库的全部管理权限",
|
|
# 管理员是纯管理账号，不参与学习，因此不授予 learning:use
|
|
"permissions": [
|
|
code for code, _, _ in PERMISSIONS if code != "learning:use"
|
|
],
|
|
},
|
|
"student": {
|
|
"name": "普通用户",
|
|
"description": "使用学习总览、练习、错题、分析等学习功能",
|
|
"permissions": ["learning:use"],
|
|
},
|
|
}
|
|
|
|
DEFAULT_USERS = [
|
|
("admin", "admin123", "系统管理员", "admin"),
|
|
("student", "student123", "演示学生", "student"),
|
|
]
|
|
|
|
|
|
def _permission_map(db: Session) -> dict[str, Permission]:
|
|
rows = {row.code: row for row in db.query(Permission).all()}
|
|
for code, name, group in PERMISSIONS:
|
|
if code not in rows:
|
|
row = Permission(code=code, name=name, group=group)
|
|
db.add(row)
|
|
db.flush()
|
|
rows[code] = row
|
|
return rows
|
|
|
|
|
|
def ensure_rbac(db: Session) -> None:
|
|
"""幂等创建权限、角色与默认账号，并同步旧库中已过期的角色权限。"""
|
|
permissions = _permission_map(db)
|
|
roles = {row.code: row for row in db.query(Role).all()}
|
|
for code, meta in ROLES.items():
|
|
if code not in roles:
|
|
role = Role(code=code, name=meta["name"], description=meta["description"])
|
|
db.add(role)
|
|
db.flush()
|
|
roles[code] = role
|
|
expected = set(meta["permissions"])
|
|
existing_rows = (
|
|
db.query(RolePermission, Permission.code)
|
|
.join(Permission, Permission.id == RolePermission.permission_id)
|
|
.filter(RolePermission.role_id == roles[code].id)
|
|
.all()
|
|
)
|
|
for role_permission, permission_code in existing_rows:
|
|
if permission_code not in expected:
|
|
db.delete(role_permission)
|
|
for permission_code in meta["permissions"]:
|
|
exists = (
|
|
db.query(RolePermission)
|
|
.filter(
|
|
RolePermission.role_id == roles[code].id,
|
|
RolePermission.permission_id == permissions[permission_code].id,
|
|
)
|
|
.first()
|
|
)
|
|
if exists is None:
|
|
db.add(
|
|
RolePermission(
|
|
role_id=roles[code].id,
|
|
permission_id=permissions[permission_code].id,
|
|
)
|
|
)
|
|
db.flush()
|
|
|
|
for username, password, display_name, role_code in DEFAULT_USERS:
|
|
user = db.query(User).filter(User.username == username).first()
|
|
if user is None:
|
|
user = User(
|
|
username=username,
|
|
password_hash=hash_password(password),
|
|
display_name=display_name,
|
|
active=True,
|
|
)
|
|
db.add(user)
|
|
db.flush()
|
|
assignment = (
|
|
db.query(UserRole)
|
|
.filter(
|
|
UserRole.user_id == user.id,
|
|
UserRole.role_id == roles[role_code].id,
|
|
)
|
|
.first()
|
|
)
|
|
if assignment is None:
|
|
db.add(UserRole(user_id=user.id, role_id=roles[role_code].id))
|
|
db.commit()
|