116 lines
3.9 KiB
Python
116 lines
3.9 KiB
Python
"""Web 管理端登录：基于 V0.3 sys_user / sys_user_role / sys_role 的角色与菜单权限解析"""
|
||
from __future__ import annotations
|
||
|
||
from typing import Set
|
||
|
||
from sqlalchemy import and_
|
||
from sqlalchemy.orm import Session
|
||
|
||
from app.core.config import settings
|
||
from app.models.models import SysMenu, SysRole, SysRoleMenu, SysUserRole
|
||
|
||
# 无 sys_user_role 绑定时的 Web 端角色（不可为 super_admin，避免越权）
|
||
NO_ROLE_WEB_ROLE = "guest"
|
||
|
||
# 一级菜单权限码（用于顶部菜单权限控制）
|
||
TOP_MENU_PERMS = frozenset({"menu:market", "menu:production", "menu:quality", "menu:operations", "menu:system-manage"})
|
||
|
||
|
||
def _map_role_name_to_web(role_name: str) -> str:
|
||
"""将 sys_role.role_name 映射为 web-admin 使用的 role 字符串。"""
|
||
n = (role_name or "").strip()
|
||
if not n:
|
||
return "admin"
|
||
if "超级" in n or n in ("系统管理员", "超级管理员"):
|
||
return "super_admin"
|
||
if "管理" in n or "运营" in n:
|
||
return "admin"
|
||
return "admin"
|
||
|
||
|
||
def _active_role_name_rows(db: Session, user_id: str):
|
||
"""查询用户当前有效的 sys_role.role_name 列表。"""
|
||
if not user_id:
|
||
return []
|
||
tenant = settings.DEFAULT_TENANT_NO
|
||
rows = (
|
||
db.query(SysRole.role_name)
|
||
.join(SysUserRole, SysUserRole.role_id == SysRole.id)
|
||
.filter(
|
||
SysUserRole.user_id == user_id,
|
||
SysUserRole.del_flag == 0,
|
||
SysUserRole.tenant_no == tenant,
|
||
SysRole.del_flag == 0,
|
||
SysRole.status == "1",
|
||
SysRole.tenant_no == tenant,
|
||
)
|
||
.order_by(SysRole.create_time.desc())
|
||
.all()
|
||
)
|
||
return [str(r[0]).strip() for r in rows if r and r[0]]
|
||
|
||
|
||
def user_has_active_roles(db: Session, user_id: str) -> bool:
|
||
"""用户是否至少绑定一个有效角色（sys_user_role → sys_role）。"""
|
||
return bool(_active_role_name_rows(db, user_id))
|
||
|
||
|
||
def resolve_web_role_for_user(db: Session, user_id: str) -> str:
|
||
"""根据用户关联角色解析 Web 端 role（admin / super_admin / guest）。"""
|
||
names = _active_role_name_rows(db, user_id)
|
||
if not names:
|
||
return NO_ROLE_WEB_ROLE
|
||
mapped = [_map_role_name_to_web(n) for n in names]
|
||
if "super_admin" in mapped:
|
||
return "super_admin"
|
||
return "admin"
|
||
|
||
|
||
def iter_user_menu_perm_codes(db: Session, user_id: str) -> Set[str]:
|
||
"""用户通过角色关联得到的菜单权限标识（sys_menu.perms，空则退回菜单 id）。
|
||
|
||
注：一级菜单（market、production、quality、operations、system-manage）已在 sys_menu 表中，
|
||
其 menuCode（如 menu:market）会自动被本函数返回，可用于前端顶部菜单的权限控制。
|
||
"""
|
||
if not user_id:
|
||
return set()
|
||
tenant = settings.DEFAULT_TENANT_NO
|
||
try:
|
||
q = (
|
||
db.query(SysMenu.perms, SysMenu.id)
|
||
.join(
|
||
SysRoleMenu,
|
||
and_(
|
||
SysRoleMenu.menu_id == SysMenu.id,
|
||
SysRoleMenu.del_flag == 0,
|
||
SysRoleMenu.tenant_no == tenant,
|
||
),
|
||
)
|
||
.join(
|
||
SysUserRole,
|
||
and_(
|
||
SysUserRole.role_id == SysRoleMenu.role_id,
|
||
SysUserRole.user_id == user_id,
|
||
SysUserRole.del_flag == 0,
|
||
SysUserRole.tenant_no == tenant,
|
||
),
|
||
)
|
||
.filter(
|
||
SysMenu.del_flag == 0,
|
||
SysMenu.tenant_no == tenant,
|
||
SysMenu.status == "1",
|
||
SysMenu.visible == "1",
|
||
)
|
||
.distinct()
|
||
)
|
||
out: Set[str] = set()
|
||
for perms, mid in q.all():
|
||
p = (perms or "").strip()
|
||
if p:
|
||
out.add(p)
|
||
elif mid:
|
||
out.add(str(mid).strip())
|
||
return out
|
||
except Exception:
|
||
return set()
|