add-video-driver

name	add-video-driver
description	新增视频驱动专家，指导如何在本项目中添加新的视频生成模型驱动（如 Happy Horse、Sora、Kling 等）。当需要接入新的视频生成 API 时使用。
allowed-tools	Read, Write, Terminal

新增视频驱动专家

角色定位

你是一位视频驱动集成专家，负责在本项目中添加新的视频生成模型驱动（如 Happy Horse、Sora、Kling 等）。

项目结构

项目根目录/
├── task/visual_drivers/           # 视频驱动目录
│   ├── base_video_driver.py       # 抽象基类
│   ├── driver_factory.py          # 驱动工厂（注册中心）
│   ├── vidu_default_driver.py     # Vidu 驱动示例
│   └── ...
├── config/
│   ├── unified_config.py          # 统一配置（任务类型、驱动实现常量）
│   └── default_configs.py         # 热更新配置定义
├── alembic/versions/              # 数据库迁移脚本
└── tests/base/                    # 驱动单元测试

核心概念

三层映射架构

任务类型(ID) → 业务驱动名称 → 实现驱动名称 → 驱动类实例
     ↓              ↓                ↓              ↓
    14    →  vidu_i2v  →  vidu_default  →  ViduDefaultDriver()

配置位置：config/unified_config.py 的 ALL_TASK_CONFIGS 列表

任务分类

• TEXT_TO_VIDEO - 文生视频（t2v）
• IMAGE_TO_VIDEO - 图生视频（i2v）
• TEXT_TO_IMAGE - 文生图
• IMAGE_EDIT - 图片编辑
• DIGITAL_HUMAN - 数字人

新增视频驱动完整流程

第一步：定义驱动常量和任务配置

文件：config/unified_config.py

1.1 添加驱动实现常量

在 DriverImplementation 类中添加新驱动的实现名称：

class DriverImplementation:
    """驱动实现类名常量"""
    # ... 已有驱动
    
    # Happy Horse (阿里云百炼)
    HAPPY_HORSE_BAILIAN_I2V_V1 = 'happy_horse_bailian_i2v_v1'  # 图生视频
    HAPPY_HORSE_BAILIAN_R2V_V1 = 'happy_horse_bailian_r2v_v1'  # 参考生视频
    HAPPY_HORSE_BAILIAN_T2V_V1 = 'happy_horse_bailian_t2v_v1'  # 文生视频

1.2 添加驱动实现 ID

在 DriverImplementationId 类中添加对应的数字 ID（用于数据库存储）：

class DriverImplementationId:
    """驱动实现 ID 常量"""
    # ... 已有 ID
    
    HAPPY_HORSE_BAILIAN_I2V_V1 = 50  # 选择未使用的 ID
    HAPPY_HORSE_BAILIAN_R2V_V1 = 51
    HAPPY_HORSE_BAILIAN_T2V_V1 = 52

1.3 注册任务配置

在 ALL_TASK_CONFIGS 列表中添加新任务配置：

ALL_TASK_CONFIGS = [
    # ... 已有任务
    
    # Happy Horse 图生视频
    UnifiedTaskConfig(
        id=30,  # 任务类型 ID（对应 ai_tools 表的 type 字段）
        name="happy_horse_i2v",
        display_name="Happy Horse 图生视频",
        category=TaskCategory.IMAGE_TO_VIDEO,
        driver_name="happy_horse_i2v",  # 业务驱动名称
        provider=TaskProvider.BAILIAN,  # 供应商（需在 TaskProvider 中定义）
        enabled=True,
        description="阿里云百炼 Happy Horse 图生视频",
        computing_power=0, #为0 可以根据 实现方的算力决定，不为0 会直接根据这个算力值计算
        
        # 实现方配置
        implementations=[
            ImplementationConfig(
                name=DriverImplementation.HAPPY_HORSE_BAILIAN_I2V_V1,
                display_name="阿里云百炼",
                driver_class="HappyHorseBailianI2vV1Driver",
                default_computing_power={5: 10, 10: 20},  # {时长: 算力}
                enabled=True,
                description="阿里云百炼官方 API",
                sort_order=1.0,
                required_config_keys=["bailian.api_key"]  # 依赖的配置键
            ),
        ],
        
        # 支持的参数
        supported_params={
            "prompt": {"required": True, "type": "string"},
            "image_path": {"required": True, "type": "string"},
            "duration": {"required": True, : , : [, ]},
            : {: , : , : [, , ]},
        },
        
        
        default_params={
            : ,
            : ,
        },
    ),
]

关键字段说明：

• id - 任务类型 ID，对应数据库 ai_tools.type 字段
• driver_name - 业务驱动名称，用于第一层映射
• implementations - 实现方列表，每个实现方对应一个具体的驱动类
• default_computing_power - 算力配置，可以是固定值或按时长的字典
• required_config_keys - 依赖的配置键，用于检查配置是否完整

1.5 同步实现方映射（DRIVER_IMPLEMENTATION_MAPPING）⚠️ 易漏

文件：config/constant.py

⚠️ 这是最容易遗漏的一步：项目里有两套独立的实现方映射，必须同时改：

配置	文件	作用
UnifiedTaskConfig.implementations	config/unified_config.py（见 1.4）	运行时驱动选择（用户提交任务时决定用哪个实现）
DRIVER_IMPLEMENTATION_MAPPING	config/constant.py	admin「实现方管理」页面的分组渲染

在 config/constant.py 的 DRIVER_IMPLEMENTATION_MAPPING 字典中，把新实现方加入对应 DriverKey 的列表：

DriverKey.SEEDANCE_2_0_IMAGE_TO_VIDEO: [
    DriverImplementation.SEEDANCE_2_0_VOLCENGINE_V1,           # 火山引擎国内版
    DriverImplementation.SEEDANCE_2_0_VOLCENGINE_OVERSEA_V1,   # 火山引擎海外版
    DriverImplementation.SEEDANCE_2_0_KKIDC_V1,                # kkidc 网关（新增）
],

只改 unified_config.py 而漏改 constant.py 的后果：

• ✅ 任务能正常运行、用户能切换实现
• ❌ admin「实现方管理」页面不显示该实现方（后端按 DriverKey 分组时拿不到映射，实现方被跳过）

1.5 仅适用于「新增实现方作为现有任务的备选」场景。如果是新建独立任务（新 DriverKey），也需在此为新 DriverKey 建立映射条目。

第二步：实现驱动类

文件：task/visual_drivers/{driver_name}_driver.py

2.1 创建驱动文件

创建 task/visual_drivers/happy_horse_bailian_i2v_v1_driver.py：

"""
Happy Horse 阿里云百炼图生视频驱动
"""
from typing import Dict, Any, Optional
import requests
import traceback
from .base_video_driver import BaseVideoDriver, ImageMode
from config.config_util import get_dynamic_config_value
from utils.sentry_util import SentryUtil, AlertLevel
from utils.image_upload_utils import upload_local_images_to_cdn_sync


class HappyHorseBailianI2vV1Driver(BaseVideoDriver):
    """
    Happy Horse 阿里云百炼图生视频驱动
    支持图生视频（i2v）模式
    """

    def __init__(self):
        super().__init__(driver_name="happy_horse_bailian_i2v", driver_type=30)

        # 加载配置
        self._api_key = get_dynamic_config_value("bailian", "api_key", default="")
        self._base_url = get_dynamic_config_value(
            "bailian", "base_url", 
            default="https://dashscope.aliyuncs.com/api/v1/services/aigc/video-generation/generation"
        )
        self._timeout = get_dynamic_config_value("timeout", "request_timeout", default=30)

        # 验证必需配置
        self._validate_required({
            "阿里云百炼 API Key": self._api_key,
        })

     () -> [, ]:
        
        :
            
            image_url = ._upload_image_to_cdn(ai_tool.image_path)
            
            
            request_data = {
                : ,
                : {
                    : ai_tool.prompt,
                    : image_url,
                },
                : {
                    : ai_tool.duration,
                }
            }
            
            
             (ai_tool, )  ai_tool.ratio:
                request_data[][] = ai_tool.ratio
            
            
            headers = {
                : ,
                : ,
            }
            
            .logger.info()
            
            response = requests.post(
                ._base_url,
                json=request_data,
                headers=headers,
                timeout=._timeout
            )
            
            
            result = response.json()
            
            
            is_valid, error_msg = ._validate_submit_response(result)
              is_valid:
                ._send_alert(
                    ,
                    ,
                    {: result}
                )
                 {
                    : ,
                    : ,
                    : 
                }
            
            
               result:
                error_code = result[].get(, )
                error_msg = result[].get(, )
                 {
                    : ,
                    : ,
                    : error_msg
                }
            
            
             {
                : ,
                : result[][],
                : result[][],
            }
            
         requests.exceptions.Timeout:
             ._handle_timeout_error()
         requests.exceptions.RequestException  e:
             ._handle_network_error(e)
         Exception  e:
             ._handle_unexpected_error(e, traceback.format_exc())

     () -> [, ]:
        
        :
            
            query_url = 
            
            headers = {
                : ,
            }
            
            
            response = requests.get(
                query_url,
                headers=headers,
                timeout=._timeout
            )
            
            result = response.json()
            
            
            is_valid, error_msg = ._validate_status_response(result)
              is_valid:
                ._send_alert(
                    ,
                    ,
                    {: result}
                )
                 {
                    : ,
                    : 
                }
            
            
            status = result[][]
            
            
             status  [, ]:
                 {: , : }
             status == :
                 {
                    : ,
                    : ,
                    : result[][],
                }
             status == :
                error_msg = result[].get(, )
                 {
                    : ,
                    : ,
                    : error_msg,
                }
            :
                 {
                    : ,
                    : 
                }
                
         Exception  e:
             ._handle_unexpected_error(e, traceback.format_exc())

     () -> [, []]:
        
          (result, ):
             , 
        
           result:
             ,   
        
            result:
             , 
        
        output = result[]
            output:
             , 
        
         , 

     () -> [, []]:
        
          (result, ):
             , 
        
            result:
             , 
        
        output = result[]
            output:
             , 
        
         , 

     () -> :
        
        cdn_urls = upload_local_images_to_cdn_sync([image_path])
          cdn_urls   cdn_urls[]:
             ValueError()
         cdn_urls[]

     ():
        
        SentryUtil.send_alert(
            alert_type=alert_type,
            message=message,
            level=AlertLevel.ERROR,
            context=context
        )

关键实现要点：

1. 继承 BaseVideoDriver - 必须实现 submit_task 和 check_status 方法
2. 配置加载 - 使用 get_dynamic_config_value 支持热更新
3. 参数验证 - 使用 _validate_required 检查必需配置
4. 响应验证 - 实现 _validate_submit_response 和 _validate_status_response
5. 错误处理 - 区分网络错误、超时错误、业务错误和系统错误
6. 日志记录 - 使用 self.logger 记录关键操作
7. 报警机制 - 使用 SentryUtil 上报系统级错误

第三步：注册驱动到工厂

文件：task/visual_drivers/driver_factory.py

在文件末尾的注册区域添加新驱动的注册代码：

# 在 driver_factory.py 的注册区域（约 600 行之后）

try:
    from .happy_horse_bailian_i2v_v1_driver import HappyHorseBailianI2vV1Driver
    # 注册 Happy Horse 图生视频驱动
    VideoDriverFactory.register_driver(
        DriverImplementation.HAPPY_HORSE_BAILIAN_I2V_V1, 
        HappyHorseBailianI2vV1Driver
    )
except ImportError as e:
    logger.warning(f"Failed to import HappyHorseBailianI2vV1Driver: {e}")

注意事项：

• 使用 try-except 包裹，避免单个驱动导入失败影响整个系统
• 驱动名称必须与 DriverImplementation 中定义的常量一致
• 按供应商或功能分组注册，便于维护

第四步：配置文件与热更新

新增视频驱动后，需要在配置文件和热更新定义中添加对应的配置项，以便管理员通过后台配置 API Key 等参数。

4.1 添加示例配置

文件：config.example.yml 和 config_prod.base.yaml

在配置文件中添加新供应商的配置段：

# 阿里云百炼配置
bailian:
  api_key: ""                    # 必填，阿里云百炼 API Key
  base_url: "https://dashscope.aliyuncs.com/api/v1/services/aigc/video-generation/generation"  # 可选

4.2 添加热更新配置

文件：config/default_configs.py

在 DEFAULT_CONFIGS 列表中添加热更新配置项：

# ==================== 阿里云百炼配置 ====================
{
    'key': 'bailian.api_key',
    'value_type': 'string',
    'description': '阿里云百炼 API Key',
    'editable': True,
    'is_sensitive': True,
    'quick_config': True
},
{
    'key': 'bailian.base_url',
    'value_type': 'string',
    'description': '阿里云百炼 API 基础URL（默认 https://dashscope.aliyuncs.com/...）',
    'editable': True,
    'is_sensitive': False,
    'quick_config': False
},

配置说明：

• quick_config: True - 配置项出现在管理后台的快速配置弹窗中
• is_sensitive: True - 敏感信息，日志中会脱敏显示
• editable: True - 允许管理员修改

4.3 添加供应商常量（如需要）

文件：config/unified_config.py

如果是新供应商，需要在 TaskProvider 类中添加常量：

class TaskProvider:
    """任务供应商常量"""
    DUOMI = 'duomi'
    RUNNINGHUB = 'runninghub'
    VIDU = 'vidu'
    VOLCENGINE = 'volcengine'
    BAILIAN = 'bailian'  # 新增：阿里云百炼
    # ...

第五步：编写单元测试

文件：tests/base/test_{driver_name}_driver.py

创建驱动的单元测试文件：

"""
Happy Horse 阿里云百炼驱动单元测试
"""
import pytest
from unittest.mock import Mock, patch, MagicMock
from task.visual_drivers.happy_horse_bailian_i2v_v1_driver import HappyHorseBailianI2vV1Driver


class TestHappyHorseBailianI2vV1Driver:
    """Happy Horse 图生视频驱动测试"""

    @pytest.fixture
    def driver(self):
        """创建驱动实例"""
        with patch('task.visual_drivers.happy_horse_bailian_i2v_v1_driver.get_dynamic_config_value') as mock_config:
            # Mock 配置值
            def config_side_effect(section, key, default=None):
                config_map = {
                    ('bailian', 'api_key'): 'test_api_key_123',
                    ('bailian', 'base_url'): 'https://test.api.com',
                    ('timeout', 'request_timeout'): 30,
                }
                return config_map.get((section, key), default)
            
            mock_config.side_effect = config_side_effect
            driver = HappyHorseBailianI2vV1Driver()
            return driver

    @pytest.fixture
    def mock_ai_tool(self):
        """创建 Mock AITool 对象"""
        ai_tool = Mock()
        ai_tool.prompt = "一只可爱的猫咪在草地上奔跑"
        ai_tool.image_path = "/path/to/image.jpg"
        ai_tool.duration = 5
        ai_tool.ratio = 
         ai_tool

     ():
        
         patch()  mock_upload, \
             patch()  mock_post:
            
            
            mock_upload.return_value = []
            
            
            mock_response = Mock()
            mock_response.json.return_value = {
                : {
                    : ,
                    : 
                }
            }
            mock_post.return_value = mock_response
            
            
            result = driver.submit_task(mock_ai_tool)
            
            
             result[]  
             result[] == 
             result[] == 
            
            
            mock_post.assert_called_once()
            call_args = mock_post.call_args
             call_args[][][] == 
             call_args[][][][] == mock_ai_tool.prompt

     ():
        
         patch()  mock_upload, \
             patch()  mock_post:
            
            mock_upload.return_value = []
            
            
            mock_response = Mock()
            mock_response.json.return_value = {
                : {
                    : ,
                    : 
                }
            }
            mock_post.return_value = mock_response
            
            result = driver.submit_task(mock_ai_tool)
            
             result[]  
               result[]

     ():
        
         patch()  mock_get:
            
            mock_response = Mock()
            mock_response.json.return_value = {
                : {
                    : ,
                    : 
                }
            }
            mock_get.return_value = mock_response
            
            result = driver.check_status()
            
             result[]  
             result[] == 
             result[] == 

     ():
        
         patch()  mock_get:
            mock_response = Mock()
            mock_response.json.return_value = {
                : {
                    : 
                }
            }
            mock_get.return_value = mock_response
            
            result = driver.check_status()
            
             result[]  
             result[] == 

     ():
        
         patch()  mock_get:
            mock_response = Mock()
            mock_response.json.return_value = {
                : {
                    : ,
                    : 
                }
            }
            mock_get.return_value = mock_response
            
            result = driver.check_status()
            
             result[]  
             result[] == 
               result[]

     ():
        
        
        valid_response = {
            : {
                : ,
                : 
            }
        }
        is_valid, error = driver._validate_submit_response(valid_response)
         is_valid  
         error  
        
        
        invalid_response = {: }
        is_valid, error = driver._validate_submit_response(invalid_response)
         is_valid  
           error
        
        
        error_response = {
            : {
                : ,
                : 
            }
        }
        is_valid, error = driver._validate_submit_response(error_response)
         is_valid    

测试要点：

1. Mock 配置 - 使用 patch Mock 配置读取
2. Mock 外部依赖 - Mock HTTP 请求、CDN 上传等
3. 覆盖关键场景 - 成功、失败、超时、异常等
4. 验证响应格式 - 测试响应验证逻辑
5. 验证 API 调用 - 检查请求参数是否正确

运行测试：

pytest tests/base/test_happy_horse_bailian_i2v_v1_driver.py -v

第六步：更新文档（可选）

如果驱动有特殊说明或使用注意事项，可以更新 task/visual_drivers/README.md：

## 支持的驱动列表

### 阿里云百炼
- **happy_horse_bailian_i2v_v1** - Happy Horse 图生视频 ✅
- **happy_horse_bailian_r2v_v1** - Happy Horse 参考生视频 ✅
- **happy_horse_bailian_t2v_v1** - Happy Horse 文生视频 ✅

**配置要求**：
- `bailian.api_key` - 必填，阿里云百炼 API Key
- `bailian.base_url` - 可选，默认使用官方 API 地址

**获取 API Key**：访问 [阿里云百炼控制台](https://bailian.console.aliyun.com/)

检查清单

完成新增视频驱动后，按顺序确认以下事项：

第一步：常量和配置

• config/unified_config.py - 已在 DriverImplementation 中添加驱动实现常量
• config/unified_config.py - 已在 DriverImplementationId 中添加对应 ID
• config/unified_config.py - 已在 ALL_TASK_CONFIGS 中添加任务配置
• config/unified_config.py - 已在 TaskProvider 中添加供应商常量（如需要）
• config/constant.py - 已在 DRIVER_IMPLEMENTATION_MAPPING 中为新实现方加入对应 DriverKey（⚠️ 否则 admin「实现方管理」页面不显示）

第二步：驱动实现

• task/visual_drivers/{driver}_driver.py - 已创建驱动文件
• 驱动类已继承 BaseVideoDriver
• 已实现 submit_task 方法
• 已实现 check_status 方法
• 已实现 _validate_submit_response 方法
• 已实现 _validate_status_response 方法
• 已添加完整的错误处理（超时、网络、业务、系统错误）
• 已添加 Sentry 报警机制

第三步：驱动注册

• task/visual_drivers/driver_factory.py - 已导入驱动类
• task/visual_drivers/driver_factory.py - 已调用 register_driver 注册
• 使用 try-except 包裹导入和注册代码

第四步：配置文件

• config.example.yml - 已添加供应商配置段
• config_prod.base.yaml - 已添加供应商配置段（与 example 一致）
• config/default_configs.py - 已添加热更新配置项（api_key 等）

第五步：单元测试

• tests/base/test_{driver}_driver.py - 已创建测试文件
• 已测试 submit_task 成功场景
• 已测试 submit_task 失败场景（API 错误、网络错误）
• 已测试 check_status 各种状态（processing、success、failed）
• 已测试响应格式验证逻辑
• 所有测试通过

第六步：文档更新（可选）

• task/visual_drivers/README.md - 已添加驱动说明
• 已说明配置要求和获取方式

常见问题

Q1: 如何确定任务类型 ID（driver_type）？

A: 任务类型 ID 对应数据库 ai_tools 表的 type 字段。查询现有 ID：

SELECT type, name FROM ai_tools ORDER BY type;

选择一个未使用的 ID（建议从 30 开始递增）。

Q2: 算力配置如何计算？

A: 算力配置基于成本和时长：

• 固定算力：不区分时长，如 default_computing_power=10
• 按时长计费：字典格式，如 {5: 10, 10: 20} 表示 5 秒消耗 10 点算力，10 秒消耗 20 点

计算公式：

算力 = (API 成本 / 0.04元) × 利润率 (一般1.1)

例如：API 成本 0.4 元，利润率 1.1，则算力 = ceil(0.4 / 0.04 * 1.1) = 11 点

Q3: 如何区分用户错误和系统错误？

A: 错误分类原则：

• 用户错误：参数错误、余额不足、内容违规等 → 返回 user_message，不发送 Sentry 报警
• 系统错误：API 格式异常、网络超时、未预期异常等 → 发送 Sentry 报警

示例：

# 用户错误
if "InvalidParameter" in error_code:
    return {
        "success": False,
        "error": f"参数错误: {error_msg}",
        "user_message": "请检查输入参数是否正确"
    }

# 系统错误
if not is_valid_format:
    self._send_alert("INVALID_RESPONSE", "API 响应格式异常", {"response": result})
    return {
        "success": False,
        "error": "API 响应格式异常",
        "user_message": "服务暂时不可用，请稍后重试"
    }

Q4: 如何支持多种图片输入模式？

A: 在任务配置中指定 image_modes：

UnifiedTaskConfig(
    # ...
    image_modes=[
        ImageMode.FIRST_LAST_FRAME,      # 首尾帧模式
        ImageMode.MULTI_REFERENCE,       # 多参考图模式
        ImageMode.FIRST_LAST_WITH_REF,   # 首尾帧+参考图模式
    ],
)

驱动实现中根据 ai_tool.image_mode 判断：

def submit_task(self, ai_tool):
    if ai_tool.image_mode == ImageMode.FIRST_LAST_FRAME:
        # 处理首尾帧模式
        first_frame = ai_tool.image_path
        last_frame = ai_tool.image_path_2
    elif ai_tool.image_mode == ImageMode.MULTI_REFERENCE:
        # 处理多参考图模式
        reference_images = [ai_tool.image_path, ai_tool.image_path_2, ...]

Q5: 如何实现同步 API 驱动？

A: 对于同步阻塞的 API（如某些本地模型），需要：

1. 在 ImplementationConfig 中设置 sync_mode=True
2. 系统会自动将同步驱动放入独立进程池处理，避免阻塞异步事件循环

ImplementationConfig(
    name="local_model_v1",
    # ...
    sync_mode=True,  # 标记为同步模式
)

Q6: 如何测试驱动是否正常工作？

A: 三种测试方法：

1. 单元测试：

   pytest tests/base/test_{driver}_driver.py -v
   

2. 集成测试：通过 API 提交真实任务

   curl -X POST http://localhost:8000/api/ai-tools/submit \
     -H "Content-Type: application/json" \
     -d '{"type": 30, "prompt": "测试", "image_path": "/path/to/image.jpg"}'
   

3. 工厂测试：验证驱动是否正确注册

   from task.visual_drivers.driver_factory import VideoDriverFactory
   driver = VideoDriverFactory.create_driver_by_type(30)
   assert driver is not None
   

Q7: 为什么新增实现方后，admin「实现方管理」页面不显示？

A: 按顺序排查：

1. 检查 config/constant.py 的 DRIVER_IMPLEMENTATION_MAPPING（最常见原因）：是否已把新实现方加入对应 DriverKey 的列表？admin「实现方管理」页面按 DriverKey 分组，分组依据来自此字典。漏改会导致实现方被完全跳过、不显示。

   • 注意：config/unified_config.py 的 UnifiedTaskConfig.implementations 列表不能替代此映射 —— 前者管运行时驱动选择，后者管页面分组渲染，两者必须同步。

2. 检查 required_config_keys 配置是否已填写：后端渲染实现方列表时会实例化驱动校验配置，若该实现的 required_config_keys（如 kkidc.api_key）在系统配置里为空，实例化抛 DriverConfigError，实现方会被过滤掉。先到后台填好对应的 API Key。

注意事项

1. 配置键命名规范

• 使用小写字母和下划线：bailian.api_key
• 分层结构：{vendor}.{key}
• 避免使用中文或特殊字符

2. 驱动命名规范

• 格式：{model}_{provider}_{version}_driver.py
• 示例：happy_horse_bailian_i2v_v1_driver.py
• 类名：驼峰命名，如 HappyHorseBailianI2vV1Driver

3. 错误处理最佳实践

try:
    # API 调用
    response = requests.post(url, json=data, timeout=self._timeout)
    result = response.json()
    
except requests.exceptions.Timeout:
    return self._handle_timeout_error()
    
except requests.exceptions.RequestException as e:
    return self._handle_network_error(e)
    
except Exception as e:
    return self._handle_unexpected_error(e, traceback.format_exc())

4. 响应格式验证

必须验证：

• 响应是否为字典类型
• 必需字段是否存在（task_id、status 等）
• 字段类型是否正确

目的：

• 及时发现 API 变更
• 通过 Sentry 报警通知开发者
• 避免下游代码异常

5. CDN 上传注意事项

# 同步上传（阻塞）
cdn_urls = upload_local_images_to_cdn_sync([image_path])

# 异步上传（非阻塞）
cdn_urls = await upload_local_images_to_cdn_async([image_path])

选择原则：

• 驱动的 submit_task 方法是同步的 → 使用 sync 版本
• 如果驱动标记为 sync_mode=True → 可以使用阻塞调用

6. 日志记录规范

# 关键操作
self.logger.info(f"[{self.driver_name}] 提交任务: task_id={task_id}")

# 调试信息
self.logger.debug(f"[{self.driver_name}] 请求参数: {request_data}")

# 错误信息
self.logger.error(f"[{self.driver_name}] API 错误: {error_msg}")

7. 兼容性考虑

• 向后兼容：修改驱动时不要破坏现有 API 接口
• 版本管理：重大变更时创建新版本驱动（如 _v2）
• 优雅降级：配置缺失时提供友好提示，而非直接崩溃

总结

新增视频驱动的核心步骤：

1. 定义常量 → unified_config.py 中添加驱动实现和任务配置
2. 实现驱动 → 继承 BaseVideoDriver，实现 submit_task 和 check_status
3. 注册驱动 → 在 driver_factory.py 中注册
4. 配置文件 → 添加热更新配置和示例配置（如需要新配置键）
5. 单元测试 → 编写完整的测试用例
6. 文档更新 → 更新 README 和使用说明（可选）

关键原则：

• ✅ 统一接口，便于维护
• ✅ 完整错误处理，区分用户错误和系统错误
• ✅ 响应格式验证，及时发现 API 变更
• ✅ 单元测试覆盖，保证代码质量
• ✅ 配置热更新，无需重启服务