18 KiB

Raw Permalink Blame History

RFC 145: 批量对话配音生成接口

日期: 2026-02-13
状态: 已实施
类型: Feature
影响范围: AI Service, API, Celery Tasks, Storyboard Resources

📋 概述

新增 /api/v1/ai/generate-dialogue-voiceovers 接口，支持为多个分镜对话批量生成 AI 配音，并自动写入 storyboard_voiceovers 表。

核心目标

批量生成：一次请求为最多 50 个对话生成配音
自动写入：配音结果直接写入 storyboard_voiceovers 表，无需手动上传
部分失败容错：已成功的配音保留，失败的对话在结果中标记
完整参数支持：支持音色、语速、音量、音调等完整 TTS 参数

🎯 业务场景

现有流程的痛点

旧方案：

用户使用 /ai/generate-voice 生成配音（通用 TTS）
从 ai_generation_results 表获取音频 URL
手动调用 /storyboard-resources/dialogues/{id}/voiceovers 上传
需要逐个对话重复 3 次操作

问题：

❌ 操作繁琐，需要多次 API 调用
❌ 通用 TTS 接口与分镜业务逻辑脱节
❌ 无法批量处理，效率低

新方案优势

新流程：

用户调用 /ai/generate-dialogue-voiceovers，传入对话 ID 列表
系统自动：
- 读取对话文本
- 调用 TTS 生成配音
- 上传到 MinIO
- 写入 storyboard_voiceovers 表
返回 job_id，用户查询任务状态

优点：

✅ 一次请求完成所有操作
✅ 批量处理，性能优化
✅ 业务语义清晰
✅ 支持部分失败（容错性强）

🔧 技术设计

1. API 接口

端点: POST /api/v1/ai/generate-dialogue-voiceovers

请求参数:

{
  storyboardId: string;        // 分镜 ID（必填）
  dialogueIds: string[];       // 对话 ID 列表（1-50 个）
  voiceId: string;             // 音色 ID（通过 /voices 获取）
  voiceName?: string;          // 音色名称（可选）
  speed: number;               // 语速（0.25-4.0，默认 1.0）
  volume: number;              // 音量（0.0-2.0，默认 1.0）
  pitch: number;               // 音调（0.5-2.0，默认 1.0）
  isActive: boolean;           // 是否设为激活配音（默认 false）
}

注意:

参数使用 camelCase 格式（前端标准）
系统自动选择最佳音频模型，无需手动指定 model 参数

响应:

{
  "code": 200,
  "message": "批量配音生成任务创建成功",
  "data": {
    "jobId": "550e8400-e29b-41d4-a716-446655440000",
    "taskId": "celery-task-id",
    "status": "pending",
    "estimatedCredits": 150,
    "dialogueCount": 10
  }
}

任务完成后（查询 /jobs/{job_id}）:

{
  "outputData": {
    "successful_count": 9,
    "failed_count": 1,
    "successful_voiceovers": [
      {
        "dialogue_id": "d1",
        "voiceover_id": "v1",
        "audio_url": "https://minio.example.com/..."
      }
    ],
    "failed_dialogues": [
      {
        "dialogue_id": "d10",
        "error": "TTS 生成失败: timeout"
      }
    ]
  }
}

2. 数据模型

新增 AIJobType

# server/app/models/ai_job.py
class AIJobType(IntEnum):
    # ... 现有类型 ...
    DIALOGUE_VOICEOVER = 10  # 批量对话配音生成

写入目标表

表名: storyboard_voiceovers

字段:

voiceover_id: UUID           # 主键
dialogue_id: UUID            # 关联对话
storyboard_id: UUID          # 关联分镜
audio_url: str               # MinIO 音频 URL
status: int                  # 2=completed
is_active: bool              # 是否激活
voice_id: str                # 音色 ID
voice_name: Optional[str]    # 音色名称
speed: Decimal               # 语速
volume: Decimal              # 音量
pitch: Decimal               # 音调
duration: Optional[Decimal]  # 音频时长
file_size: Optional[int]     # 文件大小
format: Optional[str]        # 音频格式（mp3）
checksum: str                # SHA256 校验和
storage_provider: str        # minio
storage_path: str            # 存储路径
created_at: datetime         # 创建时间
completed_at: datetime       # 完成时间

3. 业务流程

3.1 AIService.generate_dialogue_voiceovers()

async def generate_dialogue_voiceovers(
    self,
    user_id: str,
    dialogue_ids: list[str],
    voice_id: str,
    voice_name: Optional[str] = None,
    speed: float = 1.0,
    volume: float = 1.0,
    pitch: float = 1.0,
    is_active: bool = False,
    model: Optional[str] = None,
    **kwargs
) -> Dict[str, Any]:
    """批量为对话生成 AI 配音"""
    
    # 1. 验证用户存在
    # 2. 验证对话数量（1-50）
    # 3. 获取所有对话内容
    # 4. 验证所有对话属于同一分镜
    # 5. 验证分镜权限（至少 viewer）
    # 6. 检查配额
    # 7. 获取模型配置（默认 AUDIO 类型）
    # 8. 计算所需积分（基于所有对话的总字符数）
    # 9. 预扣积分
    # 10. 创建 AI Job（job_type=10）
    # 11. 关联积分日志
    # 12. 启动 Celery 异步任务
    # 13. 返回 job_id

关键验证：

✅ 所有对话必须属于同一分镜
✅ 用户对分镜至少有 viewer 权限
✅ 对话总字符数不超过 10,000 字（避免超长任务）

3.2 Celery 任务

任务名称: generate_dialogue_voiceovers_task

执行流程:

@celery_app.task(base=AITask, bind=True, max_retries=3)
def generate_dialogue_voiceovers_task(
    self, job_id, user_id, dialogue_ids, voice_id, model, ...
):
    """批量对话配音生成任务"""
    
    successful_voiceovers = []
    failed_dialogues = []
    
    # 逐个处理对话
    for idx, dialogue_id in enumerate(dialogue_ids):
        try:
            # 1. 更新进度（5% - 85%）
            progress = 5 + int((idx / total_dialogues) * 80)
            
            # 2. 获取对话内容
            dialogue = await get_dialogue_by_id(dialogue_id)
            text = dialogue.content
            
            # 3. 调用 TTS Provider 生成配音
            result = await provider.generate_voice(
                text=text,
                voice_type=voice_id,
                speed=speed,
                ...
            )
            
            # 4. 上传音频到 MinIO
            metadata = await file_storage.upload_file(
                file_content=result['audio_data'],
                filename=f"dialogue_voice_{dialogue_id}.mp3",
                category='ai-generated/dialogue-voiceovers',
                ...
            )
            
            # 5. 写入 storyboard_voiceovers 表
            if is_active:
                await deactivate_all_voiceovers(dialogue_id)
            
            voiceover = StoryboardVoiceover(
                voiceover_id=generate_uuid(),
                dialogue_id=dialogue_id,
                storyboard_id=storyboard_id,
                audio_url=metadata.file_url,
                status=ResourceStatus.COMPLETED,
                is_active=is_active,
                voice_id=voice_id,
                voice_name=voice_name,
                speed=speed,
                volume=volume,
                pitch=pitch,
                ...
            )
            await create_voiceover(voiceover)
            
            successful_voiceovers.append({
                'dialogue_id': dialogue_id,
                'voiceover_id': str(voiceover.voiceover_id),
                'audio_url': metadata.file_url
            })
            
        except Exception as e:
            # 记录失败，继续处理下一个
            failed_dialogues.append({
                'dialogue_id': dialogue_id,
                'error': str(e)
            })
    
    # 6. 更新任务状态为完成（即使部分失败）
    await update_job_status(
        job_id,
        AIJobStatus.COMPLETED,
        progress=100,
        output_data={
            'successful_count': len(successful_voiceovers),
            'failed_count': len(failed_dialogues),
            'successful_voiceovers': successful_voiceovers,
            'failed_dialogues': failed_dialogues
        }
    )
    
    # 7. 确认积分消耗（不退款，部分失败也算完成）
    await confirm_credits(job_id, consumption_log_id, success=True)

容错策略：

✅ 部分失败继续：单个对话失败不影响其他对话
✅ 已成功的保留：已写入 storyboard_voiceovers 的记录不回滚
✅ 积分不退还：部分失败仍消耗积分（按总字符数预扣）

4. 积分计算

计费单位：字符数（char_count）

公式：

total_chars = sum(len(dialogue.content) for dialogue in dialogues)
credits_needed = calculate_credits(
    feature_type=FeatureType.VOICE_GENERATION,
    params={'model_name': model, 'char_count': total_chars}
)

示例：

10 个对话，每个 50 字 → 500 字 → 100 积分
即使 1 个失败，仍消耗 100 积分（不按成功数量退款）

5. 权限控制

检查项	要求
用户验证	用户必须存在
对话验证	所有对话 ID 必须有效
分镜归属	所有对话必须属于同一分镜
分镜权限	用户对分镜至少有 `viewer` 权限
配额检查	用户配额足够

📊 与现有接口对比

特性	`/generate-voice`（通用 TTS）	`/generate-dialogue-voiceovers`（批量对话）
用途	任意文本转语音	为分镜对话生成配音
输入	自由文本（`text`）	对话 ID 列表（`dialogue_ids`）
数据源	用户输入	`storyboard_dialogues` 表
写入表	`ai_generation_results`	`storyboard_voiceovers`
批量支持	❌ 单次单个文本	✅ 单次最多 50 个对话
分镜关联	可选（`storyboard_id`）	强制（验证对话归属）
权限验证	仅用户存在性	分镜权限验证
失败策略	全部失败	部分失败继续
参数	`voice_type`, `speed`, `language`	`voice_id`, `speed`, `volume`, `pitch`, `is_active`

保留原有接口的原因：

✅ /generate-voice 可用于通用 TTS（非分镜场景）
✅ 测试音色、预览效果
✅ 其他业务模块（如文档配音、通知语音等）

🛠️ 实现细节

文件改动

文件	改动
`server/app/models/ai_job.py`	新增 `AIJobType.DIALOGUE_VOICEOVER = 10`
`server/app/schemas/ai.py`	新增 `GenerateDialogueVoiceoversRequest`
`server/app/api/v1/ai.py`	新增 `/generate-dialogue-voiceovers` 路由
`server/app/services/ai_service.py`	新增 `generate_dialogue_voiceovers()` 方法
`server/app/tasks/ai_tasks.py`	新增 `generate_dialogue_voiceovers_task`

依赖关系

AIService.generate_dialogue_voiceovers()
  ├─> StoryboardResourceRepository.get_dialogue_by_id()  # 获取对话
  ├─> StoryboardService.get_storyboard()                # 验证分镜权限
  ├─> AIProviderFactory.create_provider()               # 创建 TTS Provider
  └─> generate_dialogue_voiceovers_task.delay()         # 启动 Celery 任务

generate_dialogue_voiceovers_task()
  ├─> provider.generate_voice()                          # 调用 TTS
  ├─> FileStorageService.upload_file()                   # 上传到 MinIO
  └─> StoryboardResourceRepository.create_voiceover()    # 写入数据库

🧪 测试计划

单元测试

# server/tests/unit/services/test_ai_service.py
class TestGenerateDialogueVoiceovers:
    async def test_validate_dialogue_ids_empty(self):
        """测试空对话列表验证"""
        with pytest.raises(ValidationError, match="对话 ID 列表不能为空"):
            await service.generate_dialogue_voiceovers(
                user_id=user_id,
                dialogue_ids=[],
                voice_id="alloy"
            )
    
    async def test_validate_dialogue_ids_exceed_limit(self):
        """测试对话数量超限"""
        dialogue_ids = [str(uuid4()) for _ in range(51)]
        with pytest.raises(ValidationError, match="最多支持 50 个对话"):
            await service.generate_dialogue_voiceovers(
                user_id=user_id,
                dialogue_ids=dialogue_ids,
                voice_id="alloy"
            )
    
    async def test_validate_dialogues_from_different_storyboards(self):
        """测试对话属于不同分镜"""
        with pytest.raises(ValidationError, match="必须属于同一个分镜"):
            await service.generate_dialogue_voiceovers(
                user_id=user_id,
                dialogue_ids=[dialogue1_id, dialogue2_id],  # 不同分镜
                voice_id="alloy"
            )
    
    async def test_create_job_success(self):
        """测试成功创建任务"""
        result = await service.generate_dialogue_voiceovers(
            user_id=user_id,
            dialogue_ids=[dialogue1_id, dialogue2_id],
            voice_id="alloy",
            speed=1.2
        )
        
        assert 'job_id' in result
        assert result['dialogue_count'] == 2
        assert result['status'] == 'pending'

集成测试

# server/tests/integration/test_ai_api.py
class TestGenerateDialogueVoiceoversAPI:
    async def test_create_task_success(self, async_client, test_user_token):
        """测试成功创建批量配音任务"""
        response = await async_client.post(
            '/api/v1/ai/generate-dialogue-voiceovers',
            json={
                'dialogue_ids': [str(dialogue1.dialogue_id), str(dialogue2.dialogue_id)],
                'voice_id': 'alloy',
                'speed': 1.0,
                'volume': 1.0,
                'pitch': 1.0,
                'is_active': False
            },
            headers={'Authorization': f'Bearer {test_user_token}'}
        )
        
        assert response.status_code == 200
        data = response.json()
        assert data['code'] == 200
        assert 'job_id' in data['data']
        assert data['data']['dialogue_count'] == 2

手动测试

# 1. 批量生成配音
curl -X POST "http://localhost:6160/api/v1/ai/generate-dialogue-voiceovers" \
  -H "Authorization: Bearer $TOKEN" \
  -H "Content-Type: application/json" \
  -d '{
    "dialogue_ids": ["d1", "d2", "d3"],
    "voice_id": "EXAVITQu4vr4xnSDxMaL",
    "voice_name": "Bella - Conversational, Calm, Reserved",
    "speed": 1.0,
    "volume": 1.0,
    "pitch": 1.0,
    "is_active": true
  }'

# 2. 查询任务状态
curl -X GET "http://localhost:6160/api/v1/ai/jobs/{job_id}" \
  -H "Authorization: Bearer $TOKEN"

# 3. 验证配音已写入
curl -X GET "http://localhost:6160/api/v1/storyboard-resources/dialogues/{dialogue_id}/voiceovers" \
  -H "Authorization: Bearer $TOKEN"

📝 使用示例

前端集成

// 1. 获取分镜的所有对话
const dialogues = await fetchDialogues(storyboardId);
const dialogueIds = dialogues.map(d => d.dialogueId);

// 2. 批量生成配音
const { data } = await fetch('/api/v1/ai/generate-dialogue-voiceovers', {
  method: 'POST',
  headers: { Authorization: `Bearer ${token}` },
  body: JSON.stringify({
    dialogue_ids: dialogueIds,
    voice_id: 'EXAVITQu4vr4xnSDxMaL',  // 从 /voices 接口获取
    voice_name: 'Bella',
    speed: 1.0,
    volume: 1.0,
    pitch: 1.0,
    is_active: true  // 自动激活
  })
});

console.log(`任务已创建: ${data.jobId}, 对话数量: ${data.dialogueCount}`);

// 3. 轮询任务状态
const jobId = data.jobId;
const interval = setInterval(async () => {
  const { data: job } = await fetch(`/api/v1/ai/jobs/${jobId}`);
  
  if (job.status === 3) {  // COMPLETED
    clearInterval(interval);
    console.log(`成功: ${job.outputData.successful_count}, 失败: ${job.outputData.failed_count}`);
    
    // 4. 刷新分镜对话列表（配音已自动关联）
    await refreshDialogues();
  }
}, 2000);

⚠️ 注意事项

1. 积分消耗

预扣全部积分：基于所有对话的总字符数一次性预扣
部分失败不退款：即使部分对话生成失败，积分不退还
建议：先测试少量对话，确认效果后再批量生成

2. 性能考虑

最大批量限制：50 个对话/请求（避免超长任务）
推荐批量大小：10-20 个对话（平衡性能与容错）
超时设置：每个对话最多 60 秒（TTS + 上传）

3. 失败处理

部分失败：已成功的配音保留，失败的在 output_data.failed_dialogues 中标记
重试策略：失败的对话可以单独重新提交（传入对应的 dialogue_ids）
查看失败原因：查询任务状态 /jobs/{job_id}，检查 outputData.failed_dialogues

4. 激活配音

is_active=true：每个对话的新配音会自动停用该对话的其他配音
is_active=false：生成配音但不激活，需手动调用 /voiceovers/{voiceover_id}/activate

🚀 后续优化

短期优化

并行生成：使用 asyncio.gather 并行调用多个 TTS Provider（当前串行）
进度细化：实时返回每个对话的生成进度
失败重试：自动重试失败的对话（最多 3 次）

长期规划

音色预设：为角色关联默认音色，自动应用
情绪映射：根据对话的 emotion 字段自动调整 TTS 参数
实时预览：生成前先预览单个对话的配音效果
批量导出：打包所有配音为 ZIP 下载

📚 相关文档

RFC 142: ElevenLabs Integration
RFC 144: AI Models Capability Config
ADR 006: 数据库时间戳使用 TIMESTAMPTZ
Changelog: 2026-02-13-dialogue-voiceovers-batch-generation.md

✅ 验收标准

API 接口实现完成
支持批量生成（1-50 个对话）
配音自动写入 storyboard_voiceovers 表
支持部分失败容错
完整的权限验证（用户、对话、分镜）
积分预扣和消耗记录
Celery 异步任务实现
文档完整（RFC + Changelog）
单元测试覆盖（待补充）
集成测试覆盖（待补充）

🎉 总结

本 RFC 实现了一个专业的批量对话配音生成接口，解决了现有流程的繁琐问题，提升了用户体验和开发效率。通过清晰的职责分离（通用 TTS vs 分镜配音），确保了系统的可维护性和可扩展性。

18 KiB Raw Permalink Blame History