zkxw/backend/app/services/web_auth_roles.py

"""Web 管理端登录：基于 V0.3 sys_user / sys_user_role / sys_role 的角色与菜单权限解析

有效角色 = 用户直接角色 ∪ 经启用职位继承的角色（sys_post_role）。
"""
from __future__ import annotations

from typing import List, Set

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 get_effective_role_ids(db: Session, user_id: str) -> List[str]:
    """用户有效角色 ID：直接绑定 ∪ 启用职位关联的启用角色（去重保序）。"""
    if not user_id:
        return []
    tenant = settings.DEFAULT_TENANT_NO

    direct_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()
    )

    post_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()
    )

    out: List[str] = []
    seen: Set[str] = set()
    for row in list(direct_rows) + list(post_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 _active_role_name_rows(db: Session, user_id: str) -> list:
    """查询用户当前有效的 sys_role.role_name 列表（含职位继承）。"""
    role_ids = get_effective_role_ids(db, user_id)
    if not role_ids:
        return []
    tenant = settings.DEFAULT_TENANT_NO
    rows = (
        db.query(SysRole.role_name)
        .filter(
            SysRole.id.in_(role_ids),
            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:
    """用户是否至少拥有一个有效角色（直接或经职位继承）。"""
    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) -> 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()
    role_ids = get_effective_role_ids(db, user_id)
    if not role_ids:
        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,
                    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()