140 lines
5.2 KiB
Python
140 lines
5.2 KiB
Python
from datetime import datetime
|
||
|
||
from sqlalchemy import select
|
||
from sqlalchemy.orm import Session
|
||
|
||
from app.models import JobTitle, Permission, Role, User
|
||
|
||
|
||
DEFAULT_ROLES = {
|
||
"employee": "普通员工",
|
||
"admin": "业务管理员",
|
||
"system_admin": "系统管理员",
|
||
}
|
||
|
||
# 业务职级预设。用户在个人资料里选的头衔。跟制度里的分组名不一定字面一致，
|
||
# 字面不一致时必须在 JOB_TITLE_GROUP_ALIASES 里加映射，否则 chat.py 拿 job_title
|
||
# 做子串匹配会全部落空，LLM 反问澄清。
|
||
DEFAULT_JOB_TITLES: list[str] = [
|
||
"董事长",
|
||
"总经理",
|
||
"领导班子副职",
|
||
"总助级",
|
||
"中层干部正职及相当职级人员",
|
||
"中层干部副职及相当职级人员",
|
||
"教授级高工及相当职称人员",
|
||
"普通员工",
|
||
]
|
||
|
||
# 业务职级 -> 制度里"按职级分组"分组名的关键词列表。
|
||
# 制度里分组常写作"总院领导班子正职"这种复合词，用户选的却是"董事长"这种具体头衔，
|
||
# chat.py 会拿这里的关键词代替原 job_title 做子串规则（"包含任一关键词"）。
|
||
# 未在此表的 job_title 回退成拿它本身做子串匹配（老行为）。
|
||
# 加/改字典项时，同步维护此表；关键词要够特异，避免误命中相邻分组
|
||
# （例如"正职"太泛，会同时命中"领导班子正职"和"中层正职"）。
|
||
JOB_TITLE_GROUP_ALIASES: dict[str, list[str]] = {
|
||
"董事长": ["领导班子正职"],
|
||
"总经理": ["领导班子正职"],
|
||
"领导班子副职": ["领导班子副职"],
|
||
"总助级": ["总助级"],
|
||
"中层干部正职及相当职级人员": ["中层正职", "中层干部正职"],
|
||
"中层干部副职及相当职级人员": ["中层副职", "中层干部副职"],
|
||
"教授级高工及相当职称人员": ["教授级高工"],
|
||
"普通员工": ["普通员工"],
|
||
}
|
||
|
||
# 页面级权限：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_job_titles(db: Session) -> None:
|
||
"""把 DEFAULT_JOB_TITLES 里没入库的补进去；已存在的保留，不覆盖启用状态/顺序。"""
|
||
existing = {jt.name for jt in db.scalars(select(JobTitle)).all()}
|
||
for idx, name in enumerate(DEFAULT_JOB_TITLES):
|
||
if name in existing:
|
||
continue
|
||
db.add(JobTitle(name=name, sort_order=idx, is_active=True))
|
||
db.flush()
|
||
|
||
|
||
def ensure_default_admin(db: Session) -> User:
|
||
ensure_roles(db)
|
||
ensure_job_titles(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)
|