shanxia2.0/backend/app/services/web_auth_roles.py

"""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


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 resolve_web_role_for_user(db: Session, user_id: str) -> str:
    """根据用户关联角色解析 Web 端 role（admin / super_admin）。"""
    if not user_id:
        return settings.WEB_AUTH_DEFAULT_ROLE
    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()
    )
    names = [str(r[0]).strip() for r in rows if r and r[0]]
    if not names:
        return settings.WEB_AUTH_DEFAULT_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）。"""
    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()