citu_vanna/common/result.py

# 给dataops对话助手返回结果
from datetime import datetime
from .messages import MessageTemplate, ErrorType

def success(data=None, message="操作成功", code=200):
    """
    Return a standardized success response
    
    Args:
        data: The data to return
        message: A success message
        code: HTTP status code
        
    Returns:
        dict: A standardized success response
    """
    return {
        "code": code,
        "success": True,
        "message": message,
        "data": data
    }

def failed(message="操作失败", code=500, data=None):
    """
    Return a standardized error response
    
    Args:
        message: An error message
        code: HTTP status code
        data: Optional data to return
        
    Returns:
        dict: A standardized error response
    """
    return {
        "code": code,
        "success": False,
        "message": message,
        "data": data
    }

# ===== 标准化响应方法 =====

def success_response(response_text=None, data=None, message=MessageTemplate.SUCCESS, code=200):
    """
    标准化成功响应，将具体内容放在data.response中
    
    Args:
        response_text: 用户看到的具体内容
        data: 其他业务数据
        message: 高层级描述信息
        code: HTTP状态码
        
    Returns:
        dict: 标准化成功响应
    """
    response_data = data or {}
    if response_text is not None:
        response_data["response"] = response_text
    response_data["timestamp"] = datetime.now().isoformat()
    
    return {
        "code": code,
        "success": True,
        "message": message,
        "data": response_data
    }

def error_response(response_text, error_type=None, message=MessageTemplate.PROCESSING_FAILED, 
                  code=500, data=None, can_retry=None):
    """
    标准化错误响应，将具体错误信息放在data.response中
    
    Args:
        response_text: 用户看到的具体错误信息
        error_type: 错误类型标识
        message: 高层级描述信息
        code: HTTP状态码
        data: 其他业务数据
        can_retry: 是否可以重试
        
    Returns:
        dict: 标准化错误响应
    """
    response_data = data or {}
    response_data["response"] = response_text
    response_data["timestamp"] = datetime.now().isoformat()
    
    if error_type:
        response_data["error_type"] = error_type
    if can_retry is not None:
        response_data["can_retry"] = can_retry
    
    return {
        "code": code,
        "success": False,
        "message": message,
        "data": response_data
    }

# ===== Ask Agent API 专用响应方法 =====

def agent_success_response(response_type, conversation_id=None, execution_path=None, 
                          classification_info=None, agent_version="langgraph_v1", **kwargs):
    """
    Ask Agent API 成功响应格式
    
    Args:
        response_type: 响应类型 ("DATABASE" 或 "CHAT")
        conversation_id: 对话ID
        execution_path: 执行路径
        classification_info: 分类信息
        agent_version: Agent版本
        **kwargs: 其他字段 (response, sql, query_result, summary等)
        
    Returns:
        dict: Ask Agent API 标准成功响应
    """
    data = {
        "type": response_type,
        "conversation_id": conversation_id,
        "execution_path": execution_path or [],
        "classification_info": classification_info or {},
        "agent_version": agent_version,
        "timestamp": datetime.now().isoformat()
    }
    
    # 添加其他字段
    for key, value in kwargs.items():
        if value is not None:  # 只添加非None值
            data[key] = value
    
    return {
        "code": 200,
        "success": True,
        "message": MessageTemplate.SUCCESS,
        "data": data
    }

def agent_error_response(response_text, error_type=None, message=MessageTemplate.PROCESSING_FAILED,
                        code=500, conversation_id=None, execution_path=None, 
                        classification_info=None, agent_version="langgraph_v1", **kwargs):
    """
    Ask Agent API 错误响应格式
    
    Args:
        response_text: 用户看到的具体错误信息
        error_type: 错误类型标识
        message: 高层级描述信息
        code: HTTP状态码
        conversation_id: 对话ID
        execution_path: 执行路径
        classification_info: 分类信息
        agent_version: Agent版本
        **kwargs: 其他错误相关字段
        
    Returns:
        dict: Ask Agent API 标准错误响应
    """
    data = {
        "response": response_text,
        "conversation_id": conversation_id,
        "execution_path": execution_path or [],
        "classification_info": classification_info or {},
        "agent_version": agent_version,
        "timestamp": datetime.now().isoformat()
    }
    
    if error_type:
        data["error_type"] = error_type
    
    # 添加其他错误相关字段
    for key, value in kwargs.items():
        if value is not None:
            data[key] = value
    
    return {
        "code": code,
        "success": False,
        "message": message,
        "data": data
    }

# ===== 健康检查专用响应方法 =====

def health_success_response(status="healthy", test_result=True, **kwargs):
    """
    健康检查成功响应格式
    
    Args:
        status: 健康状态
        test_result: 测试结果
        **kwargs: 其他健康检查数据
        
    Returns:
        dict: 健康检查成功响应
    """
    data = {
        "status": status,
        "test_result": test_result,
        "response": "Agent健康检查完成",
        "timestamp": datetime.now().isoformat()
    }
    
    # 添加其他字段
    for key, value in kwargs.items():
        if value is not None:
            data[key] = value
    
    return {
        "code": 200,
        "success": True,
        "message": MessageTemplate.SUCCESS,
        "data": data
    }

def health_error_response(status="unhealthy", test_result=False, message=MessageTemplate.SERVICE_UNAVAILABLE, **kwargs):
    """
    健康检查错误响应格式
    
    Args:
        status: 健康状态
        test_result: 测试结果
        message: 高层级错误描述
        **kwargs: 其他健康检查数据
        
    Returns:
        dict: 健康检查错误响应
    """
    data = {
        "status": status,
        "test_result": test_result,
        "response": "部分组件异常" if status == "degraded" else "Agent健康检查失败",
        "timestamp": datetime.now().isoformat()
    }
    
    # 添加其他字段
    for key, value in kwargs.items():
        if value is not None:
            data[key] = value
    
    return {
        "code": 503,
        "success": False,
        "message": message,
        "data": data
    }

# ===== 便捷方法 =====

def bad_request_response(response_text, missing_params=None):
    """请求参数错误响应"""
    data = {}
    if missing_params:
        data["missing_params"] = missing_params
    
    return error_response(
        response_text=response_text,
        error_type=ErrorType.MISSING_REQUIRED_PARAMS,
        message=MessageTemplate.BAD_REQUEST,
        code=400,
        data=data
    )

def validation_failed_response(response_text):
    """参数验证失败响应"""
    return error_response(
        response_text=response_text,
        error_type=ErrorType.INVALID_PARAMS,
        message=MessageTemplate.VALIDATION_FAILED,
        code=422
    )

def internal_error_response(response_text, can_retry=True):
    """系统内部错误响应"""
    return error_response(
        response_text=response_text,
        error_type=ErrorType.DATABASE_ERROR,
        message=MessageTemplate.INTERNAL_ERROR,
        code=500,
        can_retry=can_retry
    )

def service_unavailable_response(response_text, can_retry=True):
    """服务不可用响应"""
    return error_response(
        response_text=response_text,
        error_type=ErrorType.AGENT_INITIALIZATION_FAILED,
        message=MessageTemplate.SERVICE_UNAVAILABLE,
        code=503,
        can_retry=can_retry
    )

def not_found_response(response_text):
    """资源未找到响应"""
    return error_response(
        response_text=response_text,
        error_type=ErrorType.RESOURCE_NOT_FOUND,
        message=MessageTemplate.NOT_FOUND,
        code=404
    )