mirror of
https://github.com/zhouxiaoka/autoclip.git
synced 2026-05-06 22:13:00 +08:00
主要改进: 1. 清理重复文件和冗余代码 - 删除重复的API文件 (upload_backup.py, upload_new.py等) - 删除旧版本的B站服务文件 (bilibili_upload_v2.py到v6.py) - 统一B站服务接口 2. 优化数据存储架构 - 创建OptimizedStorageService解决双重存储问题 - 数据库只存储元数据,文件系统存储实际文件 - 提供数据迁移脚本和一致性检查工具 3. 统一错误处理机制 - 创建统一错误处理中间件 - 提供错误处理装饰器和上下文管理器 - 统一错误响应格式 4. 完善配置管理 - 创建UnifiedConfig统一配置系统 - 整合所有配置源(环境变量、配置文件、默认值) - 提供配置迁移脚本 5. 增强进度系统 - 创建EnhancedProgressService - 支持Redis缓存、数据库持久化和内存缓存 - 提供完整的进度跟踪和状态管理 新增文件: - backend/services/optimized_storage_service.py - backend/core/error_middleware.py - backend/core/unified_config.py - backend/services/enhanced_progress_service.py - scripts/migrate_to_optimized_storage.py - scripts/check_data_consistency.py - scripts/migrate_config.py - docs/ERROR_HANDLING_GUIDE.md - docs/PROGRESS_SYSTEM_GUIDE.md 修复文件: - backend/services/bilibili_service.py (统一上传接口) - backend/main.py (使用新的错误处理中间件)
9.0 KiB
9.0 KiB
统一错误处理指南
📋 概述
本项目已实现统一的错误处理机制,提供一致的错误响应格式和自动错误处理功能。
🏗️ 错误处理架构
错误分类
class ErrorCategory(Enum):
CONFIGURATION = "CONFIGURATION" # 配置错误
NETWORK = "NETWORK" # 网络错误
API = "API" # API错误
FILE_IO = "FILE_IO" # 文件IO错误
PROCESSING = "PROCESSING" # 处理错误
VALIDATION = "VALIDATION" # 验证错误
SYSTEM = "SYSTEM" # 系统错误
错误级别
class ErrorLevel(Enum):
DEBUG = "DEBUG"
INFO = "INFO"
WARNING = "WARNING"
ERROR = "ERROR"
CRITICAL = "CRITICAL"
🚀 使用方法
1. 抛出自定义异常
from backend.utils.error_handler import AutoClipsException, ErrorCategory
# 抛出配置错误
raise AutoClipsException(
message="API密钥未配置",
category=ErrorCategory.CONFIGURATION,
details={"config_key": "DASHSCOPE_API_KEY"}
)
# 抛出文件错误
raise AutoClipsException(
message="文件不存在",
category=ErrorCategory.FILE_IO,
details={"file_path": "/path/to/file.mp4"}
)
2. 使用错误处理装饰器
from backend.core.error_middleware import handle_errors
from backend.utils.error_handler import ErrorCategory
@handle_errors(ErrorCategory.PROCESSING)
async def process_video(video_path: str):
# 函数内的任何异常都会被自动转换为AutoClipsException
if not os.path.exists(video_path):
raise FileNotFoundError("视频文件不存在")
# 处理逻辑...
return result
3. 使用错误上下文管理器
from backend.core.error_middleware import error_context
from backend.utils.error_handler import ErrorCategory
def upload_file(file_path: str):
with error_context(ErrorCategory.FILE_IO, {"file_path": file_path}):
# 在这个上下文中抛出的任何异常都会被转换为AutoClipsException
with open(file_path, 'r') as f:
content = f.read()
return content
4. 在API路由中使用
from fastapi import APIRouter, HTTPException
from backend.utils.error_handler import AutoClipsException, ErrorCategory
router = APIRouter()
@router.get("/projects/{project_id}")
async def get_project(project_id: str):
try:
# 业务逻辑
project = await get_project_from_db(project_id)
if not project:
raise AutoClipsException(
message=f"项目不存在: {project_id}",
category=ErrorCategory.VALIDATION,
details={"project_id": project_id}
)
return project
except AutoClipsException:
# 重新抛出,让全局异常处理器处理
raise
except Exception as e:
# 其他异常会被转换为AutoClipsException
raise AutoClipsException(
message="获取项目失败",
category=ErrorCategory.SYSTEM,
original_exception=e
)
📊 错误响应格式
所有错误响应都遵循统一格式:
{
"error": {
"code": "AUTOCLIPS_VALIDATION",
"message": "项目不存在: abc123",
"details": {
"project_id": "abc123"
},
"request_id": "req_123456",
"timestamp": 1640995200.0
}
}
字段说明
code: 错误代码,格式为AUTOCLIPS_{CATEGORY}或HTTP_{STATUS_CODE}message: 错误消息,用户友好的描述details: 错误详情,包含调试信息request_id: 请求ID,用于追踪timestamp: 错误发生时间戳
🔧 HTTP状态码映射
| 错误分类 | HTTP状态码 | 说明 |
|---|---|---|
| CONFIGURATION | 500 | 配置错误 |
| NETWORK | 503 | 网络错误 |
| API | 502 | API错误 |
| FILE_IO | 500 | 文件IO错误 |
| PROCESSING | 500 | 处理错误 |
| VALIDATION | 400 | 验证错误 |
| SYSTEM | 500 | 系统错误 |
📝 最佳实践
1. 错误消息编写
# ✅ 好的错误消息
raise AutoClipsException(
message="视频文件格式不支持,请使用MP4格式",
category=ErrorCategory.VALIDATION,
details={"supported_formats": ["mp4", "avi", "mov"]}
)
# ❌ 不好的错误消息
raise AutoClipsException(
message="Error: Invalid file",
category=ErrorCategory.VALIDATION
)
2. 错误详情包含
# ✅ 包含有用的调试信息
raise AutoClipsException(
message="处理视频失败",
category=ErrorCategory.PROCESSING,
details={
"project_id": project_id,
"step": "video_cutting",
"error_code": "FFMPEG_ERROR",
"file_size": file_size
}
)
3. 错误分类选择
# ✅ 根据错误性质选择正确的分类
if not api_key:
raise AutoClipsException(
message="API密钥未配置",
category=ErrorCategory.CONFIGURATION # 配置问题
)
if response.status_code == 429:
raise AutoClipsException(
message="API调用频率超限",
category=ErrorCategory.API # API问题
)
if not os.path.exists(file_path):
raise AutoClipsException(
message="文件不存在",
category=ErrorCategory.FILE_IO # 文件问题
)
4. 异常链保持
# ✅ 保持原始异常信息
try:
result = some_risky_operation()
except Exception as e:
raise AutoClipsException(
message="操作失败",
category=ErrorCategory.SYSTEM,
original_exception=e # 保持原始异常
)
🧪 测试错误处理
1. 测试自定义异常
import pytest
from backend.utils.error_handler import AutoClipsException, ErrorCategory
def test_custom_exception():
with pytest.raises(AutoClipsException) as exc_info:
raise AutoClipsException(
message="测试错误",
category=ErrorCategory.VALIDATION
)
assert exc_info.value.category == ErrorCategory.VALIDATION
assert exc_info.value.message == "测试错误"
2. 测试API错误响应
from fastapi.testclient import TestClient
from backend.main import app
client = TestClient(app)
def test_api_error_response():
response = client.get("/api/v1/projects/nonexistent")
assert response.status_code == 400
assert "error" in response.json()
assert response.json()["error"]["code"] == "AUTOCLIPS_VALIDATION"
🔍 错误监控和日志
1. 错误日志格式
所有错误都会自动记录到日志,格式如下:
2024-01-01 12:00:00 - ERROR - 未处理的异常: AutoClipsException: 项目不存在: abc123
request_id: req_123456
path: /api/v1/projects/abc123
method: GET
traceback: [完整的堆栈跟踪]
2. 错误统计
可以通过日志分析工具统计错误:
# 统计错误类型
grep "AUTOCLIPS_" backend.log | cut -d' ' -f4 | sort | uniq -c
# 统计错误频率
grep "ERROR" backend.log | wc -l
🚨 常见错误处理场景
1. 文件操作错误
@handle_errors(ErrorCategory.FILE_IO)
async def save_file(file_path: str, content: bytes):
try:
with open(file_path, 'wb') as f:
f.write(content)
except PermissionError:
raise AutoClipsException(
message="没有文件写入权限",
category=ErrorCategory.FILE_IO,
details={"file_path": file_path}
)
except OSError as e:
raise AutoClipsException(
message="文件系统错误",
category=ErrorCategory.FILE_IO,
details={"file_path": file_path, "os_error": str(e)}
)
2. API调用错误
@handle_errors(ErrorCategory.API)
async def call_external_api(url: str, data: dict):
try:
async with aiohttp.ClientSession() as session:
async with session.post(url, json=data) as response:
if response.status == 429:
raise AutoClipsException(
message="API调用频率超限",
category=ErrorCategory.API,
details={"url": url, "status": 429}
)
return await response.json()
except aiohttp.ClientError as e:
raise AutoClipsException(
message="网络请求失败",
category=ErrorCategory.NETWORK,
details={"url": url, "error": str(e)}
)
3. 数据处理错误
@handle_errors(ErrorCategory.PROCESSING)
async def process_video_data(video_path: str):
try:
# 处理逻辑
result = await video_processor.process(video_path)
return result
except VideoProcessingError as e:
raise AutoClipsException(
message="视频处理失败",
category=ErrorCategory.PROCESSING,
details={
"video_path": video_path,
"error_code": e.code,
"step": e.step
},
original_exception=e
)