"""角色级 + 字段级权限中间件。 基于用户角色控制 API 访问权限和数据可见性。 """ from fastapi import Depends, HTTPException, status from sqlalchemy.ext.asyncio import AsyncSession from app.core.database import get_db from app.core.dependencies import get_current_user from app.models.user import User # 角色层级 ROLE_HIERARCHY = { "admin": 100, "investor": 50, "founder": 20, } def require_role(*allowed_roles: str): """角色级权限依赖工厂。 用法: @router.get("/admin-only", dependencies=[Depends(require_role("admin"))]) """ async def _check(user: User = Depends(get_current_user)) -> User: if user.role not in allowed_roles: raise HTTPException( status_code=status.HTTP_403_FORBIDDEN, detail=f"需要角色: {', '.join(allowed_roles)},当前角色: {user.role}", ) return user return _check def require_min_role(min_role: str): """最低角色层级权限依赖工厂。 用法: @router.get("/investor+", dependencies=[Depends(require_min_role("investor"))]) """ min_level = ROLE_HIERARCHY.get(min_role, 0) async def _check(user: User = Depends(get_current_user)) -> User: user_level = ROLE_HIERARCHY.get(user.role, 0) if user_level < min_level: raise HTTPException( status_code=status.HTTP_403_FORBIDDEN, detail=f"需要最低角色: {min_role},当前角色: {user.role}", ) return user return _check # 字段级权限:不同角色可见的字段 FIELD_VISIBILITY = { "founder": { "company": ["id", "name", "industry", "stage", "description", "website"], "report": ["id", "company_id", "period_year", "period_month", "status", "raw_content"], }, "investor": { "company": ["*"], # 全部可见 "report": ["*"], }, "admin": { "company": ["*"], "report": ["*"], }, } def filter_fields( resource: str, data: dict, user: User, ) -> dict: """根据用户角色过滤返回字段。 Args: resource: 资源名称(company / report 等) data: 原始数据字典 user: 当前用户 Returns: 过滤后的数据字典 """ allowed = FIELD_VISIBILITY.get(user.role, {}).get(resource, ["*"]) if "*" in allowed: return data return {k: v for k, v in data.items() if k in allowed}