iot-device-management-service/app/yudao_compat.py

"""
芋道平台兼容层

保持与芋道主平台一致的接口规范，便于后续无缝对接。

设计原则：
- 响应格式符合芋道标准: {"code": 0, "data": ..., "msg": "success"}
- 分页参数: pageNo, pageSize
- HTTP 状态码始终 200，业务错误通过 code 区分
"""

from typing import Any, Optional
from fastapi import HTTPException, Header, Request
from fastapi.responses import JSONResponse

from app.config import settings


class YudaoResponse:
    """芋道标准响应格式"""

    @staticmethod
    def success(data: Any = None, msg: str = "success") -> dict:
        """成功响应"""
        return {"code": 0, "data": data, "msg": msg}

    @staticmethod
    def error(code: int, msg: str) -> dict:
        """错误响应"""
        return {"code": code, "data": None, "msg": msg}

    @staticmethod
    def page(list_data: list, total: int, page_no: int, page_size: int) -> dict:
        """
        分页响应格式

        Args:
            list_data: 数据列表
            total: 总记录数
            page_no: 当前页码
            page_size: 每页大小
        """
        return {
            "code": 0,
            "data": {
                "list": list_data,
                "total": total,
                "pageNo": page_no,
                "pageSize": page_size
            },
            "msg": "success"
        }


async def yudao_exception_handler(request: Request, exc: HTTPException) -> JSONResponse:
    """
    统一异常处理器

    芋道规范：HTTP 状态码始终返回 200，业务错误通过 code 区分
    """
    return JSONResponse(
        status_code=200,
        content=YudaoResponse.error(exc.status_code, exc.detail)
    )


def get_current_user(authorization: Optional[str] = Header(None, alias="Authorization")) -> dict:
    """
    获取当前用户

    测试阶段 (dev_mode=True)：
        - 跳过 Token 验证
        - 返回模拟的超级管理员用户
        - 拥有 ["*:*:*"] 全部权限

    生产阶段 (dev_mode=False)：
        - 验证 Authorization Header
        - 调用芋道主平台验证 Token
        - 返回真实用户信息

    Returns:
        dict: 用户信息，包含 userId, username, permissions
    """
    if settings.app.dev_mode:
        # 测试阶段：返回模拟的超级管理员
        return {
            "userId": 1,
            "username": "admin",
            "nickname": "超级管理员",
            "permissions": ["*:*:*"]
        }

    # 生产阶段：验证 Token
    if not authorization or not authorization.startswith("Bearer "):
        raise HTTPException(status_code=401, detail="请先登录")

    token = authorization.replace("Bearer ", "")

    # TODO: 生产阶段调用芋道主平台验证 Token
    # response = requests.get(
    #     f"{settings.yudao.base_url}/system/auth/check-token",
    #     headers={"Authorization": f"Bearer {token}"}
    # )
    # if response.status_code != 200:
    #     raise HTTPException(status_code=401, detail="Token 无效或已过期")
    # return response.json()["data"]["user"]

    # 临时：生产模式未实现时返回模拟用户
    return {
        "userId": 1,
        "username": "admin",
        "nickname": "超级管理员",
        "permissions": ["*:*:*"]
    }


def check_permission(required_permission: str):
    """
    权限检查装饰器

    测试阶段：始终通过（超级用户拥有 *:*:* 权限）
    生产阶段：检查用户是否拥有指定权限

    Usage:
        @router.get("/alert/page")
        @check_permission("ai-alert:alert:query")
        async def get_alert_page(...):
            ...
    """
    def decorator(func):
        async def wrapper(*args, current_user: dict = None, **kwargs):
            if current_user is None:
                raise HTTPException(status_code=401, detail="请先登录")

            permissions = current_user.get("permissions", [])

            # 超级权限检查
            if "*:*:*" in permissions:
                return await func(*args, current_user=current_user, **kwargs)

            # 具体权限检查
            if required_permission not in permissions:
                raise HTTPException(status_code=403, detail="没有操作权限")

            return await func(*args, current_user=current_user, **kwargs)

        return wrapper

    return decorator