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 Optional, 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 _role_rows_for_user(db: Session, user_id: str, tenant_no: str):
    return (
        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_no,
            SysRole.del_flag == 0,
            SysRole.status == "1",
            SysRole.tenant_no == tenant_no,
        )
        .order_by(SysRole.create_time.desc())
        .all()
    )


def user_has_bound_roles(db: Session, user_id: str, tenant_no: Optional[str] = None) -> bool:
    """用户是否在 sys_user_role 中绑定了至少一个有效角色。"""
    if not user_id:
        return False
    tenant = (tenant_no or "").strip() or settings.DEFAULT_TENANT_NO
    rows = _role_rows_for_user(db, user_id, tenant)
    if not rows and tenant != settings.DEFAULT_TENANT_NO:
        rows = _role_rows_for_user(db, user_id, settings.DEFAULT_TENANT_NO)
    return bool(rows)


def resolve_web_role_for_user(db: Session, user_id: str, tenant_no: Optional[str] = None) -> str:
    """根据用户关联角色解析 Web 端 role（admin / super_admin）。"""
    if not user_id:
        return settings.WEB_AUTH_DEFAULT_ROLE
    tenant = (tenant_no or "").strip() or settings.DEFAULT_TENANT_NO
    rows = _role_rows_for_user(db, user_id, tenant)
    if not rows and tenant != settings.DEFAULT_TENANT_NO:
        rows = _role_rows_for_user(db, user_id, settings.DEFAULT_TENANT_NO)
    names = [str(r[0]).strip() for r in rows if r and r[0]]
    if not names:
        # 未绑定角色：视为普通 admin，无菜单权限（不可默认 super_admin 全菜单）
        return "admin"
    mapped = [_map_role_name_to_web(n) for n in names]
    if "super_admin" in mapped:
        return "super_admin"
    return "admin"


def _granted_menu_ids_for_user(db: Session, user_id: str, tenant: str) -> Set[str]:
    rows = (
        db.query(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()
        .all()
    )
    return {str(r[0]).strip() for r in rows if r and r[0]}


def _expand_with_ancestor_menu_ids(db: Session, menu_ids: Set[str], tenant: str) -> Set[str]:
    """角色仅勾选按钮（F）时，补齐祖先 C/M 菜单 id，便于前端匹配 :list 路由。"""
    expanded = set(menu_ids)
    frontier = list(menu_ids)
    while frontier:
        rows = (
            db.query(SysMenu.parent_id)
            .filter(
                SysMenu.id.in_(frontier),
                SysMenu.del_flag == 0,
                SysMenu.tenant_no == tenant,
            )
            .all()
        )
        next_frontier: list[str] = []
        for (parent_id,) in rows:
            pid = str(parent_id or "").strip()
            if pid and pid not in expanded:
                expanded.add(pid)
                next_frontier.append(pid)
        frontier = next_frontier
    return expanded


def iter_user_menu_perm_codes(
    db: Session, user_id: str, tenant_no: Optional[str] = None
) -> Set[str]:
    """用户通过角色关联得到的菜单权限标识（sys_menu.perms，与前端 meta.menuCode 对齐）。"""
    if not user_id:
        return set()
    if not user_has_bound_roles(db, user_id, tenant_no):
        return set()
    tenant = (tenant_no or "").strip() or settings.DEFAULT_TENANT_NO
    try:
        menu_ids = _granted_menu_ids_for_user(db, user_id, tenant)
        if not menu_ids and tenant != settings.DEFAULT_TENANT_NO:
            menu_ids = _granted_menu_ids_for_user(db, user_id, settings.DEFAULT_TENANT_NO)
        if not menu_ids:
            return set()
        all_menu_ids = _expand_with_ancestor_menu_ids(db, menu_ids, tenant)
        rows = (
            db.query(SysMenu.perms)
            .filter(
                SysMenu.id.in_(all_menu_ids),
                SysMenu.del_flag == 0,
                SysMenu.tenant_no == tenant,
                SysMenu.status == "1",
            )
            .all()
        )
        out: Set[str] = set()
        for (perms,) in rows:
            p = (perms or "").strip()
            if p:
                out.add(p)
        return out
    except Exception:
        return set()