405 lines
13 KiB
Python
405 lines
13 KiB
Python
"""Web 管理端登录：基于 V0.3 sys_user / sys_user_role / sys_role 的角色与菜单权限解析
|
||
|
||
有效角色 = 用户直接角色 ∪ 经启用职位继承的角色（sys_post_role）。
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
from typing import List, Optional, Set, Collection
|
||
|
||
from sqlalchemy import and_
|
||
from sqlalchemy.orm import Session
|
||
|
||
from app.core.config import settings
|
||
from app.models.models import (
|
||
SysMenu,
|
||
SysPost,
|
||
SysPostRole,
|
||
SysRole,
|
||
SysRoleMenu,
|
||
SysUserPost,
|
||
SysUserRole,
|
||
)
|
||
|
||
# 无有效角色绑定时的 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 _dedupe_role_id_rows(rows) -> List[str]:
|
||
out: List[str] = []
|
||
seen: Set[str] = set()
|
||
for row in rows:
|
||
rid = str(row[0]).strip() if row and row[0] else ""
|
||
if rid and rid not in seen:
|
||
seen.add(rid)
|
||
out.append(rid)
|
||
return out
|
||
|
||
|
||
def get_direct_enabled_role_ids(
|
||
db: Session, user_id: str, tenant_no: Optional[str] = None
|
||
) -> List[str]:
|
||
"""用户直接绑定且启用的角色 ID（sys_user_role → sys_role）。"""
|
||
if not user_id:
|
||
return []
|
||
tenant = (tenant_no or settings.DEFAULT_TENANT_NO).strip()
|
||
rows = (
|
||
db.query(SysRole.id)
|
||
.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 _dedupe_role_id_rows(rows)
|
||
|
||
|
||
def get_post_inherited_role_ids(
|
||
db: Session, user_id: str, tenant_no: Optional[str] = None
|
||
) -> List[str]:
|
||
"""经启用职位继承的启用角色 ID（sys_user_post → sys_post → sys_post_role → sys_role）。"""
|
||
if not user_id:
|
||
return []
|
||
tenant = (tenant_no or settings.DEFAULT_TENANT_NO).strip()
|
||
rows = (
|
||
db.query(SysRole.id)
|
||
.join(SysPostRole, SysPostRole.role_id == SysRole.id)
|
||
.join(
|
||
SysPost,
|
||
and_(
|
||
SysPost.id == SysPostRole.post_id,
|
||
SysPost.del_flag == 0,
|
||
SysPost.status == "1",
|
||
SysPost.tenant_no == tenant,
|
||
),
|
||
)
|
||
.join(
|
||
SysUserPost,
|
||
and_(
|
||
SysUserPost.post_id == SysPost.id,
|
||
SysUserPost.user_id == user_id,
|
||
SysUserPost.del_flag == 0,
|
||
SysUserPost.tenant_no == tenant,
|
||
),
|
||
)
|
||
.filter(
|
||
SysPostRole.del_flag == 0,
|
||
SysPostRole.tenant_no == tenant,
|
||
SysRole.del_flag == 0,
|
||
SysRole.status == "1",
|
||
SysRole.tenant_no == tenant,
|
||
)
|
||
.order_by(SysRole.create_time.desc())
|
||
.all()
|
||
)
|
||
return _dedupe_role_id_rows(rows)
|
||
|
||
|
||
def get_effective_role_ids(
|
||
db: Session, user_id: str, tenant_no: Optional[str] = None
|
||
) -> List[str]:
|
||
"""用户有效角色 ID：直接绑定 ∪ 启用职位关联的启用角色（去重保序）。"""
|
||
if not user_id:
|
||
return []
|
||
out: List[str] = []
|
||
seen: Set[str] = set()
|
||
for rid in get_direct_enabled_role_ids(
|
||
db, user_id, tenant_no=tenant_no
|
||
) + get_post_inherited_role_ids(
|
||
db, user_id, tenant_no=tenant_no
|
||
):
|
||
if rid not in seen:
|
||
seen.add(rid)
|
||
out.append(rid)
|
||
return out
|
||
|
||
|
||
def _normalize_candidate_user_ids(
|
||
candidate_user_ids: Optional[Collection[str]],
|
||
) -> Optional[List[str]]:
|
||
if candidate_user_ids is None:
|
||
return None
|
||
return list(dict.fromkeys([str(u).strip() for u in candidate_user_ids if u and str(u).strip()]))
|
||
|
||
|
||
def list_user_ids_with_effective_role_name(
|
||
db: Session,
|
||
*,
|
||
tenant_no: str,
|
||
role_name: str,
|
||
candidate_user_ids: Optional[Collection[str]] = None,
|
||
) -> Set[str]:
|
||
"""拥有指定角色名的用户 ID（直接绑定 ∪ 职位继承）。"""
|
||
tenant = (tenant_no or "").strip()
|
||
rn = (role_name or "").strip()
|
||
if not tenant or not rn:
|
||
return set()
|
||
candidates = _normalize_candidate_user_ids(candidate_user_ids)
|
||
if candidates is not None and not candidates:
|
||
return set()
|
||
|
||
direct_q = (
|
||
db.query(SysUserRole.user_id)
|
||
.join(SysRole, SysRole.id == SysUserRole.role_id)
|
||
.filter(
|
||
SysUserRole.del_flag == 0,
|
||
SysUserRole.tenant_no == tenant,
|
||
SysRole.del_flag == 0,
|
||
SysRole.status == "1",
|
||
SysRole.tenant_no == tenant,
|
||
SysRole.role_name == rn,
|
||
)
|
||
)
|
||
post_q = (
|
||
db.query(SysUserPost.user_id)
|
||
.join(
|
||
SysPost,
|
||
and_(
|
||
SysPost.id == SysUserPost.post_id,
|
||
SysPost.del_flag == 0,
|
||
SysPost.status == "1",
|
||
SysPost.tenant_no == tenant,
|
||
),
|
||
)
|
||
.join(
|
||
SysPostRole,
|
||
and_(
|
||
SysPostRole.post_id == SysPost.id,
|
||
SysPostRole.del_flag == 0,
|
||
SysPostRole.tenant_no == tenant,
|
||
),
|
||
)
|
||
.join(
|
||
SysRole,
|
||
and_(
|
||
SysRole.id == SysPostRole.role_id,
|
||
SysRole.del_flag == 0,
|
||
SysRole.status == "1",
|
||
SysRole.tenant_no == tenant,
|
||
SysRole.role_name == rn,
|
||
),
|
||
)
|
||
.filter(SysUserPost.del_flag == 0, SysUserPost.tenant_no == tenant)
|
||
)
|
||
if candidates is not None:
|
||
direct_q = direct_q.filter(SysUserRole.user_id.in_(candidates))
|
||
post_q = post_q.filter(SysUserPost.user_id.in_(candidates))
|
||
|
||
out: Set[str] = set()
|
||
for row in direct_q.all():
|
||
uid = str(row[0]).strip() if row and row[0] else ""
|
||
if uid:
|
||
out.add(uid)
|
||
for row in post_q.all():
|
||
uid = str(row[0]).strip() if row and row[0] else ""
|
||
if uid:
|
||
out.add(uid)
|
||
return out
|
||
|
||
|
||
def list_user_ids_with_any_effective_role_ids(
|
||
db: Session,
|
||
*,
|
||
tenant_no: str,
|
||
role_ids: List[str],
|
||
candidate_user_ids: Optional[Collection[str]] = None,
|
||
) -> List[str]:
|
||
"""拥有任一指定角色 ID 的用户 ID 列表（直接绑定 ∪ 职位继承，去重）。"""
|
||
tenant = (tenant_no or "").strip()
|
||
rids = list(dict.fromkeys([str(r).strip() for r in (role_ids or []) if r and str(r).strip()]))
|
||
if not tenant or not rids:
|
||
return []
|
||
candidates = _normalize_candidate_user_ids(candidate_user_ids)
|
||
if candidates is not None and not candidates:
|
||
return []
|
||
|
||
direct_q = (
|
||
db.query(SysUserRole.user_id)
|
||
.join(SysRole, SysRole.id == SysUserRole.role_id)
|
||
.filter(
|
||
SysUserRole.del_flag == 0,
|
||
SysUserRole.tenant_no == tenant,
|
||
SysUserRole.role_id.in_(rids),
|
||
SysRole.del_flag == 0,
|
||
SysRole.status == "1",
|
||
SysRole.tenant_no == tenant,
|
||
)
|
||
)
|
||
post_q = (
|
||
db.query(SysUserPost.user_id)
|
||
.join(
|
||
SysPost,
|
||
and_(
|
||
SysPost.id == SysUserPost.post_id,
|
||
SysPost.del_flag == 0,
|
||
SysPost.status == "1",
|
||
SysPost.tenant_no == tenant,
|
||
),
|
||
)
|
||
.join(
|
||
SysPostRole,
|
||
and_(
|
||
SysPostRole.post_id == SysPost.id,
|
||
SysPostRole.del_flag == 0,
|
||
SysPostRole.tenant_no == tenant,
|
||
SysPostRole.role_id.in_(rids),
|
||
),
|
||
)
|
||
.join(
|
||
SysRole,
|
||
and_(
|
||
SysRole.id == SysPostRole.role_id,
|
||
SysRole.del_flag == 0,
|
||
SysRole.status == "1",
|
||
SysRole.tenant_no == tenant,
|
||
),
|
||
)
|
||
.filter(SysUserPost.del_flag == 0, SysUserPost.tenant_no == tenant)
|
||
)
|
||
if candidates is not None:
|
||
direct_q = direct_q.filter(SysUserRole.user_id.in_(candidates))
|
||
post_q = post_q.filter(SysUserPost.user_id.in_(candidates))
|
||
|
||
out: List[str] = []
|
||
seen: Set[str] = set()
|
||
for row in list(direct_q.all()) + list(post_q.all()):
|
||
uid = str(row[0]).strip() if row and row[0] else ""
|
||
if uid and uid not in seen:
|
||
seen.add(uid)
|
||
out.append(uid)
|
||
return out
|
||
|
||
|
||
def get_role_names_by_ids(
|
||
db: Session, role_ids: List[str], tenant_no: Optional[str] = None
|
||
) -> List[str]:
|
||
"""按给定角色 ID 返回启用角色名称（保序、去重）。"""
|
||
ordered = list(dict.fromkeys([str(r).strip() for r in (role_ids or []) if r]))
|
||
if not ordered:
|
||
return []
|
||
tenant = (tenant_no or settings.DEFAULT_TENANT_NO).strip()
|
||
rows = (
|
||
db.query(SysRole.id, SysRole.role_name)
|
||
.filter(
|
||
SysRole.id.in_(ordered),
|
||
SysRole.del_flag == 0,
|
||
SysRole.status == "1",
|
||
SysRole.tenant_no == tenant,
|
||
)
|
||
.all()
|
||
)
|
||
name_by_id = {
|
||
str(rid).strip(): str(name).strip()
|
||
for rid, name in rows
|
||
if rid and name
|
||
}
|
||
out: List[str] = []
|
||
seen: Set[str] = set()
|
||
for rid in ordered:
|
||
name = name_by_id.get(rid)
|
||
if name and name not in seen:
|
||
seen.add(name)
|
||
out.append(name)
|
||
return out
|
||
|
||
|
||
def _active_role_name_rows(
|
||
db: Session, user_id: str, tenant_no: Optional[str] = None
|
||
) -> list:
|
||
"""查询用户当前有效的 sys_role.role_name 列表（含职位继承）。"""
|
||
return get_role_names_by_ids(
|
||
db,
|
||
get_effective_role_ids(db, user_id, tenant_no=tenant_no),
|
||
tenant_no=tenant_no,
|
||
)
|
||
|
||
|
||
def user_has_active_roles(db: Session, user_id: str) -> bool:
|
||
"""用户是否至少拥有一个有效角色（直接或经职位继承）。"""
|
||
return bool(get_effective_role_ids(db, user_id))
|
||
|
||
|
||
def get_user_role_names(db: Session, user_id: str) -> list:
|
||
"""用户当前有效的 sys_role.role_name 列表（去重、保序）。"""
|
||
names = _active_role_name_rows(db, user_id)
|
||
return list(dict.fromkeys(names))
|
||
|
||
|
||
def resolve_web_role_for_user(
|
||
db: Session, user_id: str, tenant_no: Optional[str] = None
|
||
) -> str:
|
||
"""根据用户有效角色解析 Web 端 role（admin / super_admin / guest）。"""
|
||
names = _active_role_name_rows(db, user_id, tenant_no=tenant_no)
|
||
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, tenant_no: Optional[str] = None
|
||
) -> Set[str]:
|
||
"""用户通过有效角色关联得到的菜单权限标识（sys_menu.perms，空则退回菜单 id）。
|
||
|
||
注：不过滤 visible——F 类型按钮等 visible=0 的菜单仍可通过角色授权生效；
|
||
visible 仅控制侧栏/菜单树展示，不影响权限码解析。
|
||
一级菜单（menu:market 等）同样经本函数返回，用于前端顶部菜单权限控制。
|
||
"""
|
||
if not user_id:
|
||
return set()
|
||
tenant = (tenant_no or settings.DEFAULT_TENANT_NO).strip()
|
||
role_ids = get_effective_role_ids(db, user_id, tenant_no=tenant)
|
||
if not role_ids:
|
||
return set()
|
||
try:
|
||
q = (
|
||
db.query(SysMenu.perms, SysMenu.id)
|
||
.join(
|
||
SysRoleMenu,
|
||
and_(
|
||
SysRoleMenu.menu_id == SysMenu.id,
|
||
SysRoleMenu.del_flag == 0,
|
||
SysRoleMenu.tenant_no == tenant,
|
||
SysRoleMenu.role_id.in_(role_ids),
|
||
),
|
||
)
|
||
.filter(
|
||
SysMenu.del_flag == 0,
|
||
SysMenu.tenant_no == tenant,
|
||
SysMenu.status == "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()
|