selfrelease
7ec4fb0747
- LLM 客户端：全部 SSE 流式输出，兼容 OpenAI 接口
- AI 月报解析：SSE 流式端点 POST /reports/{id}/parse
- 健康度计算引擎：四维评分（财务/经营/AI商业化/AI成本）
- 风险自动检测引擎：6 条规则自动检测指标越界
- AI Copilot：SSE 流式对话 POST /copilot/chat
- 权限中间件：角色级 + 字段级权限控制
- 测试：21 个新测试（健康度 8 + 风险检测 8 + 权限 5），总计 64 passed
92 lines
2.5 KiB
Python
92 lines
2.5 KiB
Python
"""角色级 + 字段级权限中间件。
|
|
|
|
基于用户角色控制 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}
|