kb-skills/kb_pipeline/errors.py

"""Structured error handling for KB Pipeline

Provides structured error types with context information for debugging.
All errors include step_name, input_summary, error_type, message, and suggestion.
"""

from enum import Enum
from typing import Optional, List, Any


class ErrorType(str, Enum):
    """错误类型分类"""
    NETWORK = "network"
    VALIDATION = "validation"
    SYSTEM = "system"
    AUTH = "auth"


class KBError(Exception):
    """KB 工作流基础错误

    所有工作流错误的基类，包含丰富的上下文信息便于调试。

    Attributes:
        message: 错误信息
        error_type: 错误类型
        step_name: 发生错误的步骤名称
        input_summary: 输入数据摘要
        suggestion: 可操作的建议
        cause: 原始异常（可选）
    """

    def __init__(
        self,
        message: str,
        error_type: ErrorType,
        step_name: str,
        input_summary: str,
        suggestion: str,
        cause: Optional[Exception] = None
    ):
        super().__init__(message)
        self.message = message
        self.error_type = error_type
        self.step_name = step_name
        self.input_summary = input_summary
        self.suggestion = suggestion
        self.cause = cause

    def to_dict(self) -> dict:
        """将错误转换为字典格式用于日志和状态记录

        Returns:
            包含所有错误上下文信息的字典
        """
        return {
            "error_type": self.error_type.value,
            "step_name": self.step_name,
            "input_summary": self.input_summary,
            "message": self.message,
            "suggestion": self.suggestion
        }


class NetworkError(KBError):
    """网络错误 (403/404/500/超时)

    用于表示网络请求失败，包括 HTTP 状态码错误和连接超时。

    Attributes:
        status_code: HTTP 状态码（可选）
    """

    def __init__(
        self,
        step_name: str,
        input_summary: str,
        status_code: Optional[int] = None,
        cause: Optional[Exception] = None
    ):
        if status_code:
            suggestion = f"检查网络连接，状态码: {status_code}"
        else:
            suggestion = "检查网络连接"
        super().__init__(
            message=f"网络请求失败: {status_code or 'unknown'}",
            error_type=ErrorType.NETWORK,
            step_name=step_name,
            input_summary=input_summary,
            suggestion=suggestion,
            cause=cause
        )
        self.status_code = status_code


class ValidationError(KBError):
    """验证错误 (Schema 验证失败)

    用于表示数据验证失败，包含详细的验证错误列表。

    Attributes:
        validation_errors: 验证错误列表
    """

    def __init__(
        self,
        step_name: str,
        input_summary: str,
        validation_errors: Optional[List[str]] = None,
        cause: Optional[Exception] = None
    ):
        super().__init__(
            message="Schema 验证失败",
            error_type=ErrorType.VALIDATION,
            step_name=step_name,
            input_summary=input_summary,
            suggestion="检查输入数据格式是否符合要求",
            cause=cause
        )
        self.validation_errors = validation_errors or []


class SystemError(KBError):
    """系统错误 (文件系统/内存/依赖缺失)

    用于表示系统级错误，如文件不存在、内存不足、依赖缺失等。

    Attributes:
        system_message: 系统错误消息
    """

    def __init__(
        self,
        step_name: str,
        input_summary: str,
        system_message: Optional[str] = None,
        cause: Optional[Exception] = None
    ):
        super().__init__(
            message=f"系统错误: {system_message or 'unknown'}",
            error_type=ErrorType.SYSTEM,
            step_name=step_name,
            input_summary=input_summary,
            suggestion="检查系统资源和依赖是否满足要求",
            cause=cause
        )
        self.system_message = system_message


class AuthError(KBError):
    """认证/配额错误 (API配额/Cookie过期/Token失效)

    用于表示认证失败或 API 配额超限。

    Attributes:
        auth_type: 认证类型（如 API Token, Cookie 等）
    """

    def __init__(
        self,
        step_name: str,
        input_summary: str,
        auth_type: Optional[str] = None,
        cause: Optional[Exception] = None
    ):
        super().__init__(
            message=f"认证失败: {auth_type or 'unknown'}",
            error_type=ErrorType.AUTH,
            step_name=step_name,
            input_summary=input_summary,
            suggestion="检查 API Token 或 Cookie 是否有效",
            cause=cause
        )
        self.auth_type = auth_type