zkxw/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

# 无 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 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) -> 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）。

    注：不过滤 visible——F 类型按钮等 visible=0 的菜单仍可通过角色授权生效；
    visible 仅控制侧栏/菜单树展示，不影响权限码解析。
    一级菜单（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",
            )
            .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()