zkxw/backend/app/core/menu_guard.py

"""菜单 code 权限守卫（V0.3：通过 sys_user_role → sys_role_menu → sys_menu.perms）"""
from typing import Any, Callable, Dict, Iterable, Set

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

from app.core.auth import get_current_user
from app.services.web_auth_roles import iter_user_menu_perm_codes

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]:
    """读取用户授权的菜单权限标识集合（与前端 meta.menuCode / perms 对齐）。"""
    return iter_user_menu_perm_codes(db, user_id or "")


def user_has_menu_perm(db: Session, user: dict, perm_code: str) -> bool:
    """判断用户是否拥有指定权限码（不拦截请求，用于字段/按钮级能力控制）。"""
    code = (perm_code or "").strip()
    if not code:
        return False
    if user.get("role") == SUPER_ADMIN_ROLE:
        return True
    user_id = (user.get("user_id") or "").strip()
    if not user_id:
        return False
    return code in get_user_menu_codes(db, user_id)


def strip_row_fields(rows: Iterable[Dict[str, Any]], *field_names: str) -> None:
    """从响应行中移除敏感字段（原地修改）。"""
    keys = [k for k in field_names if k]
    if not keys:
        return
    for row in rows:
        for key in keys:
            row.pop(key, None)


def require_menu_code(menu_code: str) -> Callable:
    """接口菜单权限守卫（已统一放行 menuCode 校验，仅保留运营端登录）。"""
    _ = menu_code  # 保留参数以兼容各路由 Depends(require_menu_code("..."))
    return require_admin_or_super_admin


def require_any_menu_code(*menu_codes: str) -> Callable:
    """接口菜单权限守卫（已统一放行 menuCode 校验，仅保留运营端登录）。"""
    _ = menu_codes
    return require_admin_or_super_admin