mtsd/backend/app/core/menu_guard.py

"""菜单 code 权限守卫"""
from typing import Callable, Set

from fastapi import Depends, HTTPException
from sqlalchemy.orm import Session

from app.core.auth import get_current_user
from app.core.database import get_db
from app.models.models import SysUserMenuScope

ADMIN_ROLES = {"admin", "super_admin"}
SUPER_ADMIN_ROLE = "super_admin"


def require_admin_or_super_admin(current_user: dict = Depends(get_current_user)):
    """运营端角色守卫：admin / super_admin 可访问。"""
    role = current_user.get("role")
    if role not in ADMIN_ROLES:
        raise HTTPException(status_code=403, detail="需要运营人员权限")
    return current_user


def get_user_menu_codes(db: Session, user_id: str) -> Set[str]:
    """读取用户授权的菜单 code 集合。"""
    if not user_id:
        return set()
    rows = (
        db.query(SysUserMenuScope.menu_code)
        .filter(
            SysUserMenuScope.user_id == user_id,
            SysUserMenuScope.is_del == 0,
        )
        .all()
    )
    return {str(row[0]).strip() for row in rows if row and row[0]}


def require_menu_code(menu_code: str) -> Callable:
    """接口菜单权限守卫。super_admin 直接放行。"""
    normalized = (menu_code or "").strip()

    def _dep(
        current_user: dict = Depends(require_admin_or_super_admin),
        db: Session = Depends(get_db),
    ):
        role = current_user.get("role")
        if role == SUPER_ADMIN_ROLE:
            return current_user
        codes = get_user_menu_codes(db, current_user.get("user_id"))
        if normalized not in codes:
            raise HTTPException(status_code=403, detail="无菜单访问权限")
        return current_user

    return _dep