张翔
583 lines
18 KiB
Python
583 lines
18 KiB
Python
"""
|
|
错误诊断模块
|
|
提供详细的错误分析、分类、归因和恢复建议
|
|
"""
|
|
|
|
import traceback
|
|
import re
|
|
from typing import Dict, Any, List, Optional, Tuple
|
|
from dataclasses import dataclass
|
|
from enum import Enum
|
|
import json
|
|
|
|
|
|
class ErrorCategory(Enum):
|
|
"""错误分类"""
|
|
NETWORK_ERROR = "network_error"
|
|
AUTH_ERROR = "auth_error"
|
|
VALIDATION_ERROR = "validation_error"
|
|
SERVER_ERROR = "server_error"
|
|
TIMEOUT_ERROR = "timeout_error"
|
|
DATA_ERROR = "data_error"
|
|
CONFIG_ERROR = "config_error"
|
|
UNKNOWN_ERROR = "unknown_error"
|
|
|
|
|
|
class ErrorSeverity(Enum):
|
|
"""错误严重程度"""
|
|
LOW = "low"
|
|
MEDIUM = "medium"
|
|
HIGH = "high"
|
|
CRITICAL = "critical"
|
|
|
|
|
|
@dataclass
|
|
class ErrorAnalysis:
|
|
"""错误分析结果"""
|
|
error_type: str
|
|
error_category: ErrorCategory
|
|
error_severity: ErrorSeverity
|
|
error_message: str
|
|
stack_trace: str
|
|
context: Dict[str, Any]
|
|
possible_causes: List[str]
|
|
suggested_solutions: List[str]
|
|
recovery_actions: List[str]
|
|
related_tests: List[str]
|
|
|
|
|
|
class ErrorPattern:
|
|
"""错误模式"""
|
|
|
|
def __init__(self, pattern: str, category: ErrorCategory, severity: ErrorSeverity,
|
|
causes: List[str], solutions: List[str], recovery: List[str]):
|
|
"""
|
|
初始化错误模式
|
|
|
|
Args:
|
|
pattern: 错误模式（正则表达式）
|
|
category: 错误分类
|
|
severity: 错误严重程度
|
|
causes: 可能的原因
|
|
solutions: 建议的解决方案
|
|
recovery: 恢复操作
|
|
"""
|
|
self.pattern = re.compile(pattern, re.IGNORECASE)
|
|
self.category = category
|
|
self.severity = severity
|
|
self.causes = causes
|
|
self.solutions = solutions
|
|
self.recovery = recovery
|
|
|
|
def matches(self, error_message: str) -> bool:
|
|
"""
|
|
检查错误消息是否匹配此模式
|
|
|
|
Args:
|
|
error_message: 错误消息
|
|
|
|
Returns:
|
|
是否匹配
|
|
"""
|
|
return bool(self.pattern.search(error_message))
|
|
|
|
|
|
class ErrorDiagnoser:
|
|
"""错误诊断器"""
|
|
|
|
def __init__(self):
|
|
"""初始化错误诊断器"""
|
|
self.error_patterns = self._initialize_error_patterns()
|
|
self.error_history: List[ErrorAnalysis] = []
|
|
|
|
def _initialize_error_patterns(self) -> List[ErrorPattern]:
|
|
"""
|
|
初始化错误模式
|
|
|
|
Returns:
|
|
错误模式列表
|
|
"""
|
|
patterns = [
|
|
# 网络错误
|
|
ErrorPattern(
|
|
r"connection\s+(refused|timeout|reset|closed)",
|
|
ErrorCategory.NETWORK_ERROR,
|
|
ErrorSeverity.HIGH,
|
|
[
|
|
"网络连接被拒绝",
|
|
"服务器未启动或不可达",
|
|
"防火墙阻止连接",
|
|
"网络配置错误"
|
|
],
|
|
[
|
|
"检查服务器是否正在运行",
|
|
"验证网络连接",
|
|
"检查防火墙设置",
|
|
"确认服务器地址和端口正确"
|
|
],
|
|
[
|
|
"重启服务器",
|
|
"检查网络配置",
|
|
"重试连接"
|
|
]
|
|
),
|
|
ErrorPattern(
|
|
r"host\s+(unreachable|not\s+found)",
|
|
ErrorCategory.NETWORK_ERROR,
|
|
ErrorSeverity.HIGH,
|
|
[
|
|
"主机地址不存在",
|
|
"DNS解析失败",
|
|
"网络不可达"
|
|
],
|
|
[
|
|
"验证主机地址正确",
|
|
"检查DNS配置",
|
|
"确认网络连接"
|
|
],
|
|
[
|
|
"修正主机地址",
|
|
"重试连接"
|
|
]
|
|
),
|
|
|
|
# 认证错误
|
|
ErrorPattern(
|
|
r"(unauthorized|authentication\s+failed|invalid\s+(token|credentials))",
|
|
ErrorCategory.AUTH_ERROR,
|
|
ErrorSeverity.HIGH,
|
|
[
|
|
"认证令牌无效或过期",
|
|
"用户名或密码错误",
|
|
"权限不足"
|
|
],
|
|
[
|
|
"检查认证令牌",
|
|
"验证用户凭据",
|
|
"确认用户权限"
|
|
],
|
|
[
|
|
"重新登录",
|
|
"刷新认证令牌",
|
|
"联系管理员"
|
|
]
|
|
),
|
|
ErrorPattern(
|
|
r"(forbidden|access\s+denied)",
|
|
ErrorCategory.AUTH_ERROR,
|
|
ErrorSeverity.HIGH,
|
|
[
|
|
"访问被拒绝",
|
|
"权限不足",
|
|
"资源不存在"
|
|
],
|
|
[
|
|
"检查用户权限",
|
|
"验证资源是否存在",
|
|
"确认访问控制配置"
|
|
],
|
|
[
|
|
"联系管理员",
|
|
"申请相应权限"
|
|
]
|
|
),
|
|
|
|
# 验证错误
|
|
ErrorPattern(
|
|
r"(validation\s+failed|invalid\s+(parameter|input|data))",
|
|
ErrorCategory.VALIDATION_ERROR,
|
|
ErrorSeverity.MEDIUM,
|
|
[
|
|
"输入数据格式错误",
|
|
"参数验证失败",
|
|
"数据不符合要求"
|
|
],
|
|
[
|
|
"检查输入数据格式",
|
|
"验证参数类型和范围",
|
|
"参考API文档"
|
|
],
|
|
[
|
|
"修正输入数据",
|
|
"调整参数值"
|
|
]
|
|
),
|
|
ErrorPattern(
|
|
r"(required\s+field\s+missing|missing\s+required\s+parameter)",
|
|
ErrorCategory.VALIDATION_ERROR,
|
|
ErrorSeverity.MEDIUM,
|
|
[
|
|
"缺少必填字段",
|
|
"参数不完整"
|
|
],
|
|
[
|
|
"检查请求参数",
|
|
"确认必填字段",
|
|
"参考API文档"
|
|
],
|
|
[
|
|
"补充必填字段",
|
|
"修正请求参数"
|
|
]
|
|
),
|
|
|
|
# 服务器错误
|
|
ErrorPattern(
|
|
r"(internal\s+server\s+error|server\s+error)",
|
|
ErrorCategory.SERVER_ERROR,
|
|
ErrorSeverity.CRITICAL,
|
|
[
|
|
"服务器内部错误",
|
|
"服务器处理异常",
|
|
"服务器配置问题"
|
|
],
|
|
[
|
|
"检查服务器日志",
|
|
"验证服务器配置",
|
|
"检查数据库连接"
|
|
],
|
|
[
|
|
"联系技术支持",
|
|
"检查服务器状态",
|
|
"重启服务器"
|
|
]
|
|
),
|
|
ErrorPattern(
|
|
r"(database\s+(error|connection\s+failed|timeout))",
|
|
ErrorCategory.SERVER_ERROR,
|
|
ErrorSeverity.CRITICAL,
|
|
[
|
|
"数据库连接失败",
|
|
"数据库查询错误",
|
|
"数据库超时"
|
|
],
|
|
[
|
|
"检查数据库服务状态",
|
|
"验证数据库连接配置",
|
|
"检查SQL语句"
|
|
],
|
|
[
|
|
"重启数据库服务",
|
|
"修正连接配置",
|
|
"优化SQL查询"
|
|
]
|
|
),
|
|
|
|
# 超时错误
|
|
ErrorPattern(
|
|
r"(request\s+timeout|operation\s+timed\s+out)",
|
|
ErrorCategory.TIMEOUT_ERROR,
|
|
ErrorSeverity.MEDIUM,
|
|
[
|
|
"请求超时",
|
|
"服务器响应慢",
|
|
"网络延迟高"
|
|
],
|
|
[
|
|
"检查网络连接",
|
|
"增加超时时间",
|
|
"优化请求"
|
|
],
|
|
[
|
|
"重试请求",
|
|
"增加超时配置",
|
|
"优化网络环境"
|
|
]
|
|
),
|
|
|
|
# 数据错误
|
|
ErrorPattern(
|
|
r"(data\s+(not\s+found|does\s+not\s+exist)|record\s+not\s+found)",
|
|
ErrorCategory.DATA_ERROR,
|
|
ErrorSeverity.MEDIUM,
|
|
[
|
|
"数据不存在",
|
|
"记录未找到",
|
|
"数据已被删除"
|
|
],
|
|
[
|
|
"验证数据ID",
|
|
"检查数据是否存在",
|
|
"确认数据状态"
|
|
],
|
|
[
|
|
"使用正确的数据ID",
|
|
"重新创建数据"
|
|
]
|
|
),
|
|
ErrorPattern(
|
|
r"(duplicate\s+(key|entry|record)|constraint\s+violation)",
|
|
ErrorCategory.DATA_ERROR,
|
|
ErrorSeverity.MEDIUM,
|
|
[
|
|
"数据重复",
|
|
"唯一约束冲突",
|
|
"数据已存在"
|
|
],
|
|
[
|
|
"检查数据是否已存在",
|
|
"验证唯一字段",
|
|
"使用不同的值"
|
|
],
|
|
[
|
|
"删除重复数据",
|
|
"使用不同的值",
|
|
"更新现有数据"
|
|
]
|
|
),
|
|
|
|
# 配置错误
|
|
ErrorPattern(
|
|
r"(configuration\s+error|invalid\s+configuration|config\s+not\s+found)",
|
|
ErrorCategory.CONFIG_ERROR,
|
|
ErrorSeverity.HIGH,
|
|
[
|
|
"配置错误",
|
|
"配置文件缺失",
|
|
"配置参数无效"
|
|
],
|
|
[
|
|
"检查配置文件",
|
|
"验证配置参数",
|
|
"参考配置文档"
|
|
],
|
|
[
|
|
"修正配置文件",
|
|
"重置配置",
|
|
"重新加载配置"
|
|
]
|
|
)
|
|
]
|
|
|
|
return patterns
|
|
|
|
def diagnose_error(self, exception: Exception, context: Dict[str, Any] = None) -> ErrorAnalysis:
|
|
"""
|
|
诊断错误
|
|
|
|
Args:
|
|
exception: 异常对象
|
|
context: 上下文信息
|
|
|
|
Returns:
|
|
错误分析结果
|
|
"""
|
|
error_message = str(exception)
|
|
stack_trace = traceback.format_exc()
|
|
|
|
# 查找匹配的错误模式
|
|
matched_pattern = None
|
|
for pattern in self.error_patterns:
|
|
if pattern.matches(error_message):
|
|
matched_pattern = pattern
|
|
break
|
|
|
|
# 如果没有匹配的模式，使用默认分类
|
|
if matched_pattern is None:
|
|
matched_pattern = ErrorPattern(
|
|
r".*",
|
|
ErrorCategory.UNKNOWN_ERROR,
|
|
ErrorSeverity.MEDIUM,
|
|
["未知错误"],
|
|
["检查日志", "联系技术支持"],
|
|
["重试操作", "联系支持"]
|
|
)
|
|
|
|
# 分析上下文
|
|
context = context or {}
|
|
context.update({
|
|
"error_type": type(exception).__name__,
|
|
"error_message": error_message,
|
|
"timestamp": context.get("timestamp"),
|
|
"test_name": context.get("test_name"),
|
|
"url": context.get("url"),
|
|
"method": context.get("method")
|
|
})
|
|
|
|
# 创建错误分析结果
|
|
analysis = ErrorAnalysis(
|
|
error_type=type(exception).__name__,
|
|
error_category=matched_pattern.category,
|
|
error_severity=matched_pattern.severity,
|
|
error_message=error_message,
|
|
stack_trace=stack_trace,
|
|
context=context,
|
|
possible_causes=matched_pattern.causes,
|
|
suggested_solutions=matched_pattern.solutions,
|
|
recovery_actions=matched_pattern.recovery,
|
|
related_tests=self._find_related_tests(context)
|
|
)
|
|
|
|
# 记录错误历史
|
|
self.error_history.append(analysis)
|
|
|
|
return analysis
|
|
|
|
def _find_related_tests(self, context: Dict[str, Any]) -> List[str]:
|
|
"""
|
|
查找相关测试
|
|
|
|
Args:
|
|
context: 上下文信息
|
|
|
|
Returns:
|
|
相关测试列表
|
|
"""
|
|
related_tests = []
|
|
|
|
url = context.get("url", "")
|
|
method = context.get("method", "")
|
|
|
|
# 根据URL和方法查找相关测试
|
|
if "/auth/login" in url:
|
|
related_tests.extend(["用户登录", "认证测试", "权限测试"])
|
|
elif "/user/" in url:
|
|
related_tests.extend(["用户管理测试", "用户列表测试", "用户创建测试"])
|
|
elif "/role/" in url:
|
|
related_tests.extend(["角色管理测试", "角色列表测试", "角色创建测试"])
|
|
elif "/menu/" in url:
|
|
related_tests.extend(["菜单管理测试", "菜单列表测试"])
|
|
|
|
return related_tests
|
|
|
|
def get_error_statistics(self) -> Dict[str, Any]:
|
|
"""
|
|
获取错误统计信息
|
|
|
|
Returns:
|
|
错误统计信息
|
|
"""
|
|
if not self.error_history:
|
|
return {}
|
|
|
|
# 按分类统计
|
|
category_stats = {}
|
|
for analysis in self.error_history:
|
|
category = analysis.error_category.value
|
|
if category not in category_stats:
|
|
category_stats[category] = 0
|
|
category_stats[category] += 1
|
|
|
|
# 按严重程度统计
|
|
severity_stats = {}
|
|
for analysis in self.error_history:
|
|
severity = analysis.error_severity.value
|
|
if severity not in severity_stats:
|
|
severity_stats[severity] = 0
|
|
severity_stats[severity] += 1
|
|
|
|
# 按错误类型统计
|
|
type_stats = {}
|
|
for analysis in self.error_history:
|
|
error_type = analysis.error_type
|
|
if error_type not in type_stats:
|
|
type_stats[error_type] = 0
|
|
type_stats[error_type] += 1
|
|
|
|
return {
|
|
"total_errors": len(self.error_history),
|
|
"by_category": category_stats,
|
|
"by_severity": severity_stats,
|
|
"by_type": type_stats,
|
|
"most_common_errors": self._get_most_common_errors(5)
|
|
}
|
|
|
|
def _get_most_common_errors(self, limit: int = 5) -> List[Dict[str, Any]]:
|
|
"""
|
|
获取最常见的错误
|
|
|
|
Args:
|
|
limit: 返回数量限制
|
|
|
|
Returns:
|
|
最常见错误列表
|
|
"""
|
|
error_counts = {}
|
|
|
|
for analysis in self.error_history:
|
|
error_key = f"{analysis.error_type}: {analysis.error_message[:50]}"
|
|
if error_key not in error_counts:
|
|
error_counts[error_key] = 0
|
|
error_counts[error_key] += 1
|
|
|
|
sorted_errors = sorted(error_counts.items(), key=lambda x: x[1], reverse=True)
|
|
|
|
return [
|
|
{
|
|
"error": error[0],
|
|
"count": error[1]
|
|
}
|
|
for error in sorted_errors[:limit]
|
|
]
|
|
|
|
def generate_error_report(self, analysis: ErrorAnalysis) -> str:
|
|
"""
|
|
生成错误报告
|
|
|
|
Args:
|
|
analysis: 错误分析结果
|
|
|
|
Returns:
|
|
错误报告（Markdown格式）
|
|
"""
|
|
report = f"""# 错误诊断报告
|
|
|
|
## 错误信息
|
|
|
|
- **错误类型**: {analysis.error_type}
|
|
- **错误分类**: {analysis.error_category.value}
|
|
- **严重程度**: {analysis.error_severity.value}
|
|
- **错误消息**: {analysis.error_message}
|
|
|
|
## 错误堆栈
|
|
|
|
```
|
|
{analysis.stack_trace}
|
|
```
|
|
|
|
## 上下文信息
|
|
|
|
{self._format_context(analysis.context)}
|
|
|
|
## 可能的原因
|
|
|
|
{self._format_list(analysis.possible_causes)}
|
|
|
|
## 建议的解决方案
|
|
|
|
{self._format_list(analysis.suggested_solutions)}
|
|
|
|
## 恢复操作
|
|
|
|
{self._format_list(analysis.recovery_actions)}
|
|
|
|
## 相关测试
|
|
|
|
{self._format_list(analysis.related_tests)}
|
|
|
|
---
|
|
|
|
*报告生成时间: {self._get_current_timestamp()}*
|
|
"""
|
|
return report
|
|
|
|
def _format_context(self, context: Dict[str, Any]) -> str:
|
|
"""格式化上下文信息"""
|
|
lines = []
|
|
for key, value in context.items():
|
|
if value is not None:
|
|
lines.append(f"- **{key}**: {value}")
|
|
return "\n".join(lines)
|
|
|
|
def _format_list(self, items: List[str]) -> str:
|
|
"""格式化列表"""
|
|
return "\n".join([f"- {item}" for item in items])
|
|
|
|
def _get_current_timestamp(self) -> str:
|
|
"""获取当前时间戳"""
|
|
from datetime import datetime
|
|
return datetime.now().strftime("%Y-%m-%d %H:%M:%S")
|
|
|
|
def clear_history(self) -> None:
|
|
"""清空错误历史"""
|
|
self.error_history.clear() |