98 lines
3.2 KiB
Python
98 lines
3.2 KiB
Python
from datetime import datetime
|
||
|
||
from sqlalchemy import select
|
||
from sqlalchemy.orm import Session
|
||
|
||
from app.models import Permission, Role, User
|
||
|
||
|
||
DEFAULT_ROLES = {
|
||
"employee": "普通员工",
|
||
"admin": "业务管理员",
|
||
"system_admin": "系统管理员",
|
||
}
|
||
|
||
# 页面级权限：code -> (显示名, 模块)
|
||
DEFAULT_PAGE_PERMISSIONS: dict[str, tuple[str, str]] = {
|
||
"page.chat": ("智能问答", "page"),
|
||
"page.documents": ("知识文档", "page"),
|
||
"page.audit": ("问答审计", "page"),
|
||
"page.analytics": ("数据分析", "page"),
|
||
"page.settings": ("系统设置", "page"),
|
||
}
|
||
|
||
# 角色 -> 页面权限 code 列表
|
||
DEFAULT_ROLE_PAGE_PERMISSIONS: dict[str, list[str]] = {
|
||
"employee": ["page.chat"],
|
||
"admin": ["page.chat", "page.documents", "page.audit", "page.analytics"],
|
||
"system_admin": list(DEFAULT_PAGE_PERMISSIONS.keys()),
|
||
}
|
||
|
||
|
||
def _ensure_page_permissions(db: Session) -> dict[str, Permission]:
|
||
existing = {p.code: p for p in db.scalars(select(Permission)).all()}
|
||
for code, (name, module) in DEFAULT_PAGE_PERMISSIONS.items():
|
||
if code not in existing:
|
||
p = Permission(code=code, name=name, module=module)
|
||
db.add(p)
|
||
existing[code] = p
|
||
db.flush()
|
||
return existing
|
||
|
||
|
||
def ensure_roles(db: Session) -> None:
|
||
perms = _ensure_page_permissions(db)
|
||
existing_roles = {role.code: role for role in db.scalars(select(Role)).all()}
|
||
for code, name in DEFAULT_ROLES.items():
|
||
role = existing_roles.get(code)
|
||
if role is None:
|
||
role = Role(code=code, name=name)
|
||
db.add(role)
|
||
existing_roles[code] = role
|
||
db.flush()
|
||
# 只补齐默认页面权限，不覆盖管理员后续在后台自定义的权限
|
||
for code, page_codes in DEFAULT_ROLE_PAGE_PERMISSIONS.items():
|
||
role = existing_roles.get(code)
|
||
if role is None:
|
||
continue
|
||
current = {p.code for p in role.permissions}
|
||
for page_code in page_codes:
|
||
if page_code not in current and page_code in perms:
|
||
role.permissions.append(perms[page_code])
|
||
db.flush()
|
||
|
||
|
||
def ensure_default_admin(db: Session) -> User:
|
||
ensure_roles(db)
|
||
user = db.scalar(select(User).where(User.phone == "13800000000"))
|
||
if user:
|
||
return user
|
||
role = db.scalar(select(Role).where(Role.code == "system_admin"))
|
||
user = User(phone="13800000000", name="系统管理员", department="财务部", is_superuser=True)
|
||
if role:
|
||
user.roles.append(role)
|
||
db.add(user)
|
||
db.flush()
|
||
return user
|
||
|
||
|
||
def get_or_create_login_user(db: Session, phone: str) -> User:
|
||
ensure_roles(db)
|
||
user = db.scalar(select(User).where(User.phone == phone))
|
||
if user:
|
||
user.last_login_at = datetime.utcnow()
|
||
return user
|
||
role = db.scalar(select(Role).where(Role.code == "employee"))
|
||
user = User(phone=phone, name=f"用户{phone[-4:]}", status="active", last_login_at=datetime.utcnow())
|
||
if role:
|
||
user.roles.append(role)
|
||
db.add(user)
|
||
db.flush()
|
||
return user
|
||
|
||
|
||
def apply_roles(db: Session, user: User, role_codes: list[str]) -> None:
|
||
ensure_roles(db)
|
||
roles = db.scalars(select(Role).where(Role.code.in_(role_codes))).all() if role_codes else []
|
||
user.roles = list(roles)
|