API响应JSON数据解析的Python实现文档

import json
from typing import Any, Dict, Optional, Union
from dataclasses import dataclass, asdict
from datetime import datetime


@dataclass
class UserInfo:
    """用户信息数据类"""
    user_guid: str
    hostname: str
    zone: str

    @classmethod
    def from_dict(cls, data: Dict[str, Any]) -> 'UserInfo':
        """从字典创建UserInfo实例"""
        return cls(
            user_guid=data.get('userGuid', ''),
            hostname=data.get('hostname', ''),
            zone=data.get('zone', '')
        )


class APIResponse:
    """API响应处理类"""

    def __init__(self, response_data: Union[str, Dict]):
        """
        初始化API响应

        Args:
            response_data: API返回的数据，可以是JSON字符串或字典
        """
        self._raw_data = self._parse_response(response_data)
        self._parse_data()

    def _parse_response(self, response_data: Union[str, Dict]) -> Dict[str, Any]:
        """
        解析响应数据

        Returns:
            解析后的字典数据
        """
        if isinstance(response_data, str):
            try:
                return json.loads(response_data)
            except json.JSONDecodeError as e:
                raise ValueError(f"无效的JSON格式: {e}")
        elif isinstance(response_data, dict):
            return response_data
        else:
            raise TypeError(f"不支持的响应数据类型: {type(response_data)}")

    def _parse_data(self):
        """解析数据到各个属性"""
        # 基本字段
        self.success: bool = self._raw_data.get('success', False)
        self.message: str = self._raw_data.get('message', '')

        # 嵌套的data字段
        data_dict = self._raw_data.get('data', {})
        self.user_info: Optional[UserInfo] = None

        if data_dict:
            self.user_info = UserInfo.from_dict(data_dict)

        # 添加时间戳
        self.timestamp: datetime = datetime.now()

    @property
    def user_guid(self) -> str:
        """获取用户GUID（属性方式）"""
        return self.user_info.user_guid if self.user_info else ''

    @property
    def hostname(self) -> str:
        """获取主机名（属性方式）"""
        return self.user_info.hostname if self.user_info else ''

    @property
    def zone(self) -> str:
        """获取区域（属性方式）"""
        return self.user_info.zone if self.user_info else ''

    @property
    def is_success(self) -> bool:
        """检查是否成功（只读属性）"""
        return self.success

    def get_full_info(self) -> Dict[str, Any]:
        """获取完整的响应信息"""
        return {
            'success': self.success,
            'message': self.message,
            'user_info': asdict(self.user_info) if self.user_info else {},
            'timestamp': self.timestamp.isoformat(),
            'has_user_data': self.user_info is not None
        }

    def validate(self) -> bool:
        """验证响应数据是否完整有效"""
        if not self.success:
            return False

        if not self.user_info or not self.user_info.user_guid:
            return False

        return True

    def __str__(self) -> str:
        """字符串表示"""
        status = "✅ 成功" if self.success else "❌ 失败"
        return (
            f"API响应 [{status}]\n"
            f"消息: {self.message}\n"
            f"用户ID: {self.user_guid}\n"
            f"主机名: {self.hostname}\n"
            f"区域: {self.zone}\n"
            f"时间: {self.timestamp.strftime('%Y-%m-%d %H:%M:%S')}"
        )

    def __repr__(self) -> str:
        """repr表示"""
        return f"APIResponse(success={self.success}, user_guid={self.user_guid})"

    def to_dict(self) -> Dict[str, Any]:
        """转换为字典"""
        return self._raw_data.copy()

class APIClient:
    """模拟API客户端"""

    @staticmethod
    def login(username: str, password: str) -> APIResponse:
        """
        模拟登录API调用

        Args:
            username: 用户名
            password: 密码

        Returns:
            APIResponse实例
        """
        # 模拟API返回数据
        response_data = {
            'success': True,
            'data': {
                'userGuid': 'ccb94989-8804-4d0e-ab5d-441e7aca1240',
                'hostname': '4TRTVM9RFY',
                'zone': 'sr168.com'
            },
            'message': '登录成功'
        }

        return APIResponse(response_data)

    @staticmethod
    def get_user_info(token: str) -> APIResponse:
        """
        模拟获取用户信息API调用
        """
        # 模拟失败的响应
        response_data = {
            'success': False,
            'message': 'Token已过期',
            'data': None
        }

        return APIResponse(response_data)

class ResponseProcessor:
    """API响应处理器"""

    @staticmethod
    def process_login_response(response: APIResponse) -> Dict[str, Any]:
        """处理登录响应"""
        result = {
            'status': 'success' if response.success else 'error',
            'message': response.message,
            'user_data': {
                'guid': response.user_guid,
                'hostname': response.hostname,
                'zone': response.zone
            } if response.user_info else None,
            'processed_at': datetime.now().isoformat()
        }

        # 可以添加业务逻辑
        if response.success:
            result['session_valid'] = True
            result['welcome_message'] = f"欢迎回来，用户 {response.user_guid[:8]}..."

        return result