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