1197_preview_generation_proposal.md

@@ -0,0 +1,222 @@
+---
+AIGC:
+    Label: "1"
+    ContentProducer: 001191110102MACQD9K64018705
+    ProduceID: 15868733686388_0/project_7655981463858544923-files/docs/1197_preview_generation_proposal.md
+    ReservedCode1: ""
+    ContentPropagator: 001191110102MACQD9K64028705
+    PropagateID: 15868733686388#1785468313901
+    ReservedCode2: ""
+---
+# #1197 预览生成接口方案评估
+
+## 背景
+
+智能剪辑「一键生成」流程中，第3步预览生成当前被跳过，直接进入下一步。需要实现真正的预览生成功能，让用户在正式生成前能看到效果预览。
+
+## 现状分析
+
+### 现有生成链路
+
+```
+API 触发生成 → GenerationTask入库 → Celery异步任务 → UnifiedRenderService渲染 → OSS上传 → 更新状态
+```
+
+**关键节点：**
+1. **API层**：`POST /generation-tasks` 或 `POST /templates/{id}/generate` 触发生成
+2. **任务调度**：Celery task `worker.generate_video`
+3. **渲染引擎**：`UnifiedRenderService`（统一渲染引擎，已接入9个效果层）
+4. **输出配置**：默认 720p (1280x720)，支持 `resolution` 字段自定义
+5. **产物存储**：`GeneratedVideo` 表记录，OSS 存储视频文件
+
+### 已有可复用能力
+
+| 能力 | 位置 | 是否可复用 |
+|------|------|-----------|
+| 任务创建与状态管理 | `GenerationTask` + `CreateGenerationTaskUseCase` | ✅ 是 |
+| 素材下载与预处理 | `_download_video_assets` / `_download_voice_asset` | ✅ 是 |
+| 统一渲染引擎 | `UnifiedRenderService` | ✅ 是 |
+| 分辨率配置 | `resolution` 字段已支持 | ✅ 是 |
+| 混音与后处理 | `_render_video` 内流程 | ✅ 是 |
+| OSS 上传与查重 | `_upload_and_dedup` | ✅ 是 |
+| 进度追踪 | `append_log` / `progress` 字段 | ✅ 是 |
+
+## 方案对比
+
+### 方案A：复用现有生成链路 + is_preview 标记（推荐）
+
+**思路**：在现有 GenerationTask 上加 `is_preview` 标记，预览生成走完整链路但参数降级。
+
+**改动点：**
+1. **数据模型**：`GenerationTask` 加 `is_preview: bool` 字段（默认 false）；`GeneratedVideo` 加 `is_preview: bool`
+2. **API 层**：生成接口加 `is_preview` 参数，预览任务不计入配额
+3. **渲染参数**：预览模式下自动调整
+   - 分辨率：480p (854x480)
+   - 时长：限制前 15 秒（或模板第一个片段）
+   - 码率：降低至 1.5Mbps（正式 4Mbps）
+   - 效果层：跳过高级转场/粒子特效等耗时效果
+4. **任务调度**：预览任务走低优先级队列（或复用现有队列，标记优先级）
+5. **前端对接**：预览生成结果带 `is_preview=true` 标记，前端展示"预览"标签
+
+**优点：**
+- 代码复用率 90%+，改动最小
+- 与正式生成逻辑一致，预览效果真实可信
+- 进度查询、结果展示等功能直接复用
+- 后续可平滑升级：预览满意后一键转正式生成
+
+**缺点：**
+- 需要区分预览和正式任务，避免数据混淆
+- 预览任务和正式任务竞争同一队列资源（可后续优化为独立队列）
+
+**开发量估算**：2-3 天
+- 数据模型 + 迁移：0.5 天
+- API 层改造：0.5 天
+- 渲染参数降级：1 天
+- 测试 + 联调：1 天
+
+---
+
+### 方案B：新建独立预览接口 + 轻量渲染逻辑
+
+**思路**：新建独立的预览生成接口，使用简化的渲染逻辑（如只拼接素材+基础配音，跳过大部分效果）。
+
+**改动点：**
+1. 新增 `PreviewTask` 数据模型
+2. 新增 `POST /api/v1/preview/generate` 接口
+3. 新增独立的 Celery task `worker.generate_preview`
+4. 简化渲染流程：只做素材裁剪+拼接+配音，跳过转场/滤镜/字幕特效等
+
+**优点：**
+- 完全隔离，不影响正式生成链路
+- 可以做极致优化，预览生成速度快
+- 数据模型清晰，不会混淆
+
+**缺点：**
+- 代码重复率高，两套生成逻辑维护成本翻倍
+- 预览效果与正式生成可能不一致（效果层差异）
+- 前端需要对接两套接口
+- 无法从预览升级为正式生成（需重新走完整流程）
+
+**开发量估算**：4-5 天
+- 数据模型 + 接口：1 天
+- 简化渲染逻辑：2 天
+- 测试 + 联调：1-2 天
+
+---
+
+### 方案C：图片预览（首帧/关键帧截图）
+
+**思路**：不生成视频，只生成几张关键帧的预览图片。
+
+**优点：**
+- 生成速度极快（秒级）
+- 资源消耗小
+
+**缺点：**
+- 预览效果差，用户无法感知动态效果
+- 无法验证配音、转场、节奏等时间维度的效果
+- 用户体验不佳，不如"真预览"有说服力
+
+**开发量估算**：1-2 天
+
+---
+
+## 推荐方案：方案A（复用现有生成链路）
+
+### 核心理由
+
+1. **效果保真**：预览和正式生成用同一套渲染引擎，效果一致，用户信任度高
+2. **开发效率**：90% 代码复用，2-3 天可上线
+3. **可扩展性强**：后续可加「预览转正式」「低分辨率快速预览」等增强功能
+4. **维护成本低**：一套生成逻辑，bug 修复和新功能同时生效
+
+### 详细设计
+
+#### 1. 数据模型变更
+
+```python
+# GenerationTask 新增字段
+is_preview: bool = False
+"""是否为预览生成"""
+
+preview_of: str = ""
+"""预览对应的正式任务 ID（或反向关联）"""
+
+# GeneratedVideo 新增字段
+is_preview: bool = False
+"""是否为预览视频"""
+```
+
+**迁移**：alembic 新增 migration，两个表各加 1-2 个字段。
+
+#### 2. API 层
+
+```
+POST /api/v1/generation-tasks
+  Body 增加 is_preview: bool = false
+  
+POST /api/v1/templates/{id}/generate
+  Query 增加 is_preview: bool = false
+```
+
+**配额处理**：预览生成不计入用户配额，不占用生成次数限制。
+
+#### 3. 渲染参数降级
+
+| 参数 | 正式生成 | 预览生成 |
+|------|---------|---------|
+| 分辨率 | 720p (1280x720) | 480p (854x480) |
+| 码率 | 4 Mbps | 1.5 Mbps |
+| 时长 | 完整时长 | 前 15 秒（或第一段） |
+| 帧率 | 30 fps | 24 fps |
+| 转场效果 | 完整转场 | 仅淡入淡出（或简单切） |
+| 特效滤镜 | 全部启用 | 跳过粒子/光效等高级效果 |
+| 字幕 | 完整渲染 | 正常渲染（字幕是核心信息） |
+| 配音 | 完整混音 | 正常混音（配音是核心信息） |
+
+**实现方式**：在 `_render_video` 或 UnifiedRenderService 入口处，根据 `is_preview` 标记调整渲染配置。
+
+#### 4. 任务调度
+
+- 初期复用现有队列，预览任务正常排队
+- 后续如需优化，可拆分独立预览队列（低优先级）
+- 预览任务可设置较短超时时间
+
+#### 5. 前端对接
+
+- 调用生成接口时传 `is_preview=true`
+- 结果列表中预览视频带「预览」标签
+- 预览满意后可一键「升级为正式生成」（重新触发全分辨率生成，可复用素材下载缓存）
+
+### 实施步骤
+
+**Phase 1（MVP，2天）：**
+1. 数据模型 + 迁移
+2. API 层支持 is_preview 参数
+3. 渲染分辨率降级（480p）
+4. 不计入配额
+5. 基础测试
+
+**Phase 2（优化，1-2天）：**
+1. 时长限制（前15秒）
+2. 效果层降级（跳高级效果）
+3. 预览任务低优先级队列
+4. 预览转正式生成功能
+
+## 与前端对齐点
+
+1. 预览生成的触发时机（第3步自动生成？用户点击才生成？）
+2. 预览时长是固定15秒还是完整但低清？
+3. 是否需要「预览转正式生成」功能
+4. 预览视频的展示形态（和正式视频一样还是有特殊UI）
+
+## 风险与注意事项
+
+1. **数据混淆**：确保统计、计费、列表展示时正确区分预览和正式任务
+2. **存储成本**：预览视频也占 OSS 空间，可设置自动清理（7天后自动删除）
+3. **用户预期**：要明确告诉用户这是预览，效果和正式生成一致但清晰度低
+4. **并发压力**：如果用户频繁生成预览，可能增加系统负载，需要限流
+
+---
+
+> 本内容由 Coze AI 生成，请遵循相关法律法规及《人工智能生成合成内容标识办法》使用与传播。

1197_preview_generation_proposal_v2.md

@@ -0,0 +1,382 @@
+# #1197 预览生成接口技术方案（v2）
+
+> 更新说明：v2 新增「多版本预览生成」能力，支持一个模板生成多个不重复的预览视频，左侧列表展示，用户可挑选满意的版本转正式生成。
+
+## 1. 背景与目标
+
+**现状**：智能剪辑「一键生成」第3步预览生成被跳过，用户直接进入正式生成，缺少效果预览环节。
+
+**目标**：
+1. ✅ 实现真正的预览生成（低分辨率快速出片）
+2. ✅ **支持生成 1~N 个不重复的预览版本**（默认 3 个），左侧列表展示
+3. ✅ 预览满意后可一键转正式生成（复用素材下载缓存）
+4. ✅ 不计入用户配额，不占用正式生成次数
+
+---
+
+## 2. 现有生成链路分析
+
+### 2.1 链路总览
+
+```
+API 触发生成 → GenerationTask入库 → Celery异步任务
+  → 下载素材 → 构建plan/clips → UnifiedRenderService渲染
+  → 混音后处理 → OSS上传 + 查重 → 更新状态
+```
+
+### 2.2 决定视频差异的变量
+
+要做"多个不重复版本"，先分析哪些环节可以引入变化：
+
+| 变量 | 当前行为 | 能否引入变化 | 影响程度 |
+|------|---------|------------|---------|
+| 素材选择 | 按 asset_ids 顺序全用 | ✅ 可随机选择子集/不同组合 | 大 |
+| 素材排序 | 按 asset_ids 顺序 | ✅ 可 shuffle 重排 | 大 |
+| 配音选择 | 固定 voice_library_id | ✅ 可选不同音色 | 中 |
+| 标题选择 | 固定 title_ids 或随机选 | ✅ 可选不同标题 | 中 |
+| BGM | 固定 bgm_config | ✅ 可选不同BGM | 小 |
+| 转场效果 | 模板固定 | ✅ 可随机化转场类型 | 小 |
+| 播放速度 | 模板固定 | ✅ 可微调速度 | 小 |
+| 分辨率/码率 | 固定 | ✅ 预览可降级 | 不影响内容 |
+
+### 2.3 可复用能力
+
+- 任务创建与状态管理：`GenerationTask` + `CreateGenerationTaskUseCase`
+- 素材下载与预处理：`_download_all_assets`
+- 统一渲染引擎：`UnifiedRenderService`
+- 分辨率配置：`resolution` 字段已支持
+- 批量任务：`batch_id` 字段已存在（可用于预览组）
+
+---
+
+## 3. 总体方案：复用现有链路 + 多变体引擎
+
+**核心思路**：沿用 v1 的"复用现有生成链路 + is_preview 标记"方案，在此基础上增加「多版本生成」能力。
+
+**架构**：
+```
+预览生成请求（count=N）
+    ↓
+创建预览批次（preview_batch）
+    ↓
+变体引擎生成 N 个变体参数（variation seed + 参数组合）
+    ↓
+为每个变体创建 1 个 GenerationTask（is_preview=true）
+    ↓
+N 个 Celery 任务并行执行（走现有生成链路，参数降级）
+    ↓
+N 个结果汇聚，前端左侧列表展示
+```
+
+---
+
+## 4. 详细设计
+
+### 4.1 数据模型变更
+
+#### 4.1.1 GenerationTask 新增字段
+
+```python
+# 现有字段保留，新增：
+is_preview: bool = False
+"""是否为预览生成"""
+
+preview_batch_id: str = ""
+"""预览批次 ID（同批次的 N 个预览共享一个 batch）"""
+
+variant_seed: int = 0
+"""变体种子，用于控制随机化行为（素材选择、排序、转场等）"""
+
+variant_params: dict = field(default_factory=dict)
+"""变体参数快照（记录本次使用了哪些素材、标题、配音等，可追溯）
+   {
+     "asset_ids": [...],      # 实际选用的素材子集
+     "title_id": "",         # 选用的标题
+     "voice_id": "",         # 选用的配音
+     "transition_style": "", # 转场风格
+     "bgm_track": "",        # BGM 音轨
+   }
+"""
+```
+
+#### 4.1.2 GeneratedVideo 新增字段
+
+```python
+is_preview: bool = False
+"""是否为预览视频"""
+
+preview_batch_id: str = ""
+"""所属预览批次"""
+
+variant_index: int = 0
+"""在批次中的序号（0, 1, 2...）"""
+```
+
+#### 4.1.3 迁移方案
+
+alembic 新增 migration，两个表各加 4 个字段，默认值为空/false，无数据回填成本。
+
+---
+
+### 4.2 变体引擎（Variant Engine）
+
+**核心组件**：根据 count 和 seed，生成 N 组互不相同的生成参数。
+
+#### 4.2.1 变纬度设计
+
+| 维度 | 策略 | 说明 |
+|------|------|------|
+| **素材子集选择** | 从素材池中随机选 M 个（M=min(素材数, 模板clip数*2)） | 版本差异最大的来源 |
+| **素材排序** | 随机打乱顺序 | 影响叙事节奏 |
+| **标题选择** | 从 title_ids 中随机选 1 个 | 影响文案内容 |
+| **配音选择** | 从 voice_ids 中随机选 1 个（如有多个） | 影响听觉体验 |
+| **转场风格** | 从预设转场池中随机选 1 种 | 影响视觉过渡 |
+| **BGM 选择** | 从 bgm 列表中随机选 1 首（如有配置） | 影响氛围 |
+
+#### 4.2.2 去重机制
+
+- 同一批次内，变体参数必须两两不同（至少素材组合或排序不同）
+- 使用 `variant_seed` 保证可复现（相同 seed → 相同变体）
+- 如果素材数量不足导致无法生成 N 个不同版本，按实际能生成的数量返回
+
+#### 4.2.3 接口设计
+
+```python
+def generate_variants(
+    count: int,
+    seed: int,
+    asset_pool: list[str],        # 可用素材 ID 列表
+    title_pool: list[str] = [],    # 可用标题 ID 列表
+    voice_pool: list[str] = [],    # 可用配音 ID 列表
+    template_id: str = "",
+) -> list[dict]:
+    """
+    生成 count 组变体参数。
+    
+    每组参数包含：asset_ids（选用的素材+排序）、title_id、voice_id、
+    transition_style 等，确保两两不同。
+    """
+```
+
+---
+
+### 4.3 API 层设计
+
+#### 4.3.1 预览生成接口
+
+```
+POST /api/v1/templates/{template_id}/generate-preview
+```
+
+**请求体**：
+```json
+{
+  "asset_library_id": "lib_xxx",
+  "asset_ids": ["asset_1", "asset_2", ...],
+  "title_ids": ["title_1", "title_2"],
+  "voice_ids": ["voice_1", "voice_2"],
+  "bgm_config": {},
+  "count": 3,
+  "seed": 0
+}
+```
+
+| 参数 | 类型 | 必填 | 默认 | 说明 |
+|------|------|------|------|------|
+| template_id | path | ✅ | - | 模板 ID |
+| asset_library_id | body | ✅ | - | 素材库 ID |
+| asset_ids | body | ✅ | - | 素材池（从中选子集/排序） |
+| title_ids | body | - | [] | 标题池（可选，不传则不用标题） |
+| voice_ids | body | - | [] | 配音池（可选） |
+| bgm_config | body | - | {} | BGM 配置 |
+| count | body | - | 3 | 生成几个预览版本（1~10） |
+| seed | body | - | 0 | 随机种子，0 表示随机 |
+
+**响应**：
+```json
+{
+  "preview_batch_id": "pb_xxx",
+  "count": 3,
+  "tasks": [
+    {
+      "task_id": "gen_xxx_0",
+      "variant_index": 0,
+      "status": "processing"
+    },
+    {
+      "task_id": "gen_xxx_1",
+      "variant_index": 1,
+      "status": "processing"
+    },
+    ...
+  ]
+}
+```
+
+#### 4.3.2 预览批次查询接口
+
+```
+GET /api/v1/preview-batches/{batch_id}
+```
+
+返回批次内所有预览任务的状态、结果（已完成的带 video_url）。
+
+**响应**：
+```json
+{
+  "preview_batch_id": "pb_xxx",
+  "count": 3,
+  "completed_count": 2,
+  "tasks": [
+    {
+      "task_id": "gen_xxx_0",
+      "variant_index": 0,
+      "status": "completed",
+      "video_url": "https://oss.xxx/preview/xxx.mp4",
+      "duration": 15.5,
+      "thumbnail_url": "https://oss.xxx/preview/xxx.jpg"
+    },
+    ...
+  ]
+}
+```
+
+#### 4.3.3 预览转正式生成
+
+```
+POST /api/v1/preview-batches/{batch_id}/tasks/{task_id}/promote
+```
+
+将某个预览版本升级为正式生成（复用素材缓存，重新全分辨率渲染）。
+
+---
+
+### 4.4 渲染参数降级
+
+预览模式下自动调整以下参数：
+
+| 参数 | 正式生成 | 预览生成 |
+|------|---------|---------|
+| 分辨率 | 720p (1280x720) | 480p (854x480) |
+| 码率 | 4 Mbps | 1.5 Mbps |
+| 帧率 | 30 fps | 24 fps |
+| 时长 | 完整时长 | 前 15 秒（或第一段完整clip） |
+| 转场效果 | 完整转场 | 仅淡入淡出 |
+| 高级特效 | 全部启用 | 跳过粒子/光效等 |
+| 字幕 | 完整渲染 | 正常渲染 |
+| 配音 | 完整混音 | 正常混音 |
+| 输出质量 | high | medium |
+
+**实现位置**：`_render_video` 函数入口处，根据 `is_preview` 标记调整渲染配置。
+
+---
+
+### 4.5 任务调度
+
+- **并行执行**：N 个预览任务并行提交到 Celery，不排队等待
+- **低优先级**：预览任务走独立队列（`preview_queue`），不抢占正式生成资源
+- **超时控制**：预览任务超时时间 5 分钟（正式 30 分钟）
+- **自动清理**：预览视频 7 天后自动从 OSS 删除，任务记录标记为 archived
+
+---
+
+## 5. 前端对接要点
+
+### 5.1 交互流程
+
+```
+第2步选素材 → 第3步点击"生成预览"
+  → 显示 loading + 进度
+  → 预览陆续完成，左侧列表逐张出现
+  → 用户点击左侧不同版本，右侧预览区切换
+  → 用户选中满意版本 → 点击"正式生成"
+```
+
+### 5.2 需要对齐的接口
+
+1. **预览创建**：`POST /templates/{id}/generate-preview`
+2. **批次状态轮询**：`GET /preview-batches/{id}`（建议 2s 轮询，或走 SSE）
+3. **预览转正式**：`POST /preview-batches/{id}/tasks/{task_id}/promote`
+
+### 5.3 数据格式对齐
+
+预览视频条目结构：
+```json
+{
+  "id": "gen_xxx",
+  "variant_index": 0,
+  "status": "completed",
+  "video_url": "https://...",
+  "duration": 15.5,
+  "file_size": 2850000,
+  "thumbnail_url": "https://...",
+  "is_preview": true
+}
+```
+
+---
+
+## 6. 配额与计费
+
+- 预览生成**不计入**用户配额
+- 同一模板 + 同一素材池，每天最多生成 3 次多版本预览（防滥用）
+- 单个预览批次最多 10 个版本
+
+---
+
+## 7. 实施步骤
+
+### Phase 1：单版本预览（MVP，2 天）
+1. 数据模型 + 迁移（is_preview 字段）
+2. API 层支持 is_preview 参数
+3. 渲染分辨率降级（480p）
+4. 不计入配额
+5. 基础测试
+
+### Phase 2：多版本预览（3 天）
+1. 变体引擎实现（素材随机选择 + 排序 + 去重）
+2. preview_batch 批次管理
+3. 批量创建 N 个预览任务
+4. 批次查询接口
+5. 前端联调
+
+### Phase 3：预览转正式 + 优化（2 天）
+1. 预览转正式生成接口（promote）
+2. 素材下载缓存复用
+3. 独立预览队列（低优先级）
+4. 自动清理机制
+5. 完整测试 + 压测
+
+---
+
+## 8. 风险与注意事项
+
+| 风险 | 影响 | 应对 |
+|------|------|------|
+| 并发预览任务过多打满 worker | 正式生成被阻塞 | 独立预览队列 + 限流 |
+| 变体生成的视频差异不够大 | 用户觉得"都一样" | 优先素材子集+排序差异，保证视觉差异 |
+| 预览视频占用 OSS 存储 | 存储成本上升 | 7 天自动清理 + 低码率 |
+| N 个版本同时下载重复素材 | 带宽浪费 | 批次内共享一次下载（Phase 3 优化） |
+| 用户预期管理 | 以为预览就是最终效果 | 明确标注"预览版"，说明分辨率差异 |
+
+---
+
+## 9. 开发量估算
+
+| 阶段 | 后端 | 前端 | 合计 |
+|------|------|------|------|
+| Phase 1 单版本预览 | 2 天 | 1 天 | 3 天 |
+| Phase 2 多版本预览 | 3 天 | 2 天 | 5 天 |
+| Phase 3 转正式+优化 | 2 天 | 1 天 | 3 天 |
+| **总计** | **7 天** | **4 天** | **~7 天（并行）** |
+
+---
+
+## 10. 与 v1 方案的差异总结
+
+1. **新增多版本能力**：从"生成1个预览"升级为"生成N个不重复预览"
+2. **新增变体引擎**：负责素材选择/排序/配音/标题的随机化
+3. **新增批次概念**：preview_batch 管理一组预览任务
+4. **新增 promote 接口**：预览转正式生成
+5. **独立队列**：预览不抢占正式生成资源
+6. **开发量**：从 2-3 天增加到约 7 天（后端）

apps/worker/video_processing/render_adapter.py

@@ -25,6 +25,7 @@ from typing import Any, Callable
 from sqlalchemy.orm import Session
 from video_processing.oss_helpers import download_asset, upload_to_oss
 from video_processing.unified_render_service import UnifiedRenderService
+from video_processing.video_validation import validate_video_output
 
 from packages.adapters.sqlalchemy_impl.edit_plan_clip_repository import SQLAlchemyEditPlanClipRepository
 from packages.adapters.sqlalchemy_impl.edit_plan_repository import SQLAlchemyEditPlanRepository
@@ -531,6 +532,22 @@ class RenderAdapter:
         )
         result = render_svc.render()
 
+        # 4.5 渲染后校验输出完整性
+
+        validation = validate_video_output(result.output_path)
+        if not validation.valid:
+            logger.error(
+                "[render-adapter] 渲染输出校验失败: plan_id=%s job_id=%s error=%s",
+                plan_id,
+                job_id,
+                validation.error_message,
+            )
+            return RenderAdapterResult(
+                success=False,
+                error_message=f"渲染输出校验失败: {validation.error_message}",
+                error_detail=validation.error_message,
+            )
+
         self._report_progress(progress_cb, 80.0, "上传渲染结果")
 
         # 5. 上传结果

apps/worker/video_processing/video_validation.py

@@ -0,0 +1,267 @@
+"""视频输出校验 — 渲染后验证输出文件完整性.
+
+在 FFmpeg 渲染完成后，验证输出文件是否有效（非损坏/截断）：
+  1. 文件存在且大小 > 0
+  2. moov atom 存在（MP4 容器完整性标志）
+  3. ffprobe 可正常读取视频流信息
+
+用途：
+  - RenderAdapter 渲染后调用，避免将损坏文件上传到 OSS
+  - 提前发现 FFmpeg 异常退出但未抛异常的情况（如 exit=0 但输出截断）
+"""
+
+from __future__ import annotations
+
+import logging
+import subprocess
+from dataclasses import dataclass
+from pathlib import Path
+
+from shared.ffmpeg_utils import FFPROBE_BIN
+
+logger = logging.getLogger(__name__)
+
+
+# ── 数据结构 ──────────────────────────────────────────────────────────────────
+
+
+@dataclass
+class VideoValidationResult:
+    """视频校验结果。"""
+
+    valid: bool
+    file_exists: bool = False
+    file_size: int = 0
+    moov_atom_found: bool = False
+    has_video_stream: bool = False
+    duration: float = 0.0
+    width: int = 0
+    height: int = 0
+    error_message: str = ""
+
+    @property
+    def is_valid(self) -> bool:
+        """是否通过所有校验。"""
+        return self.valid
+
+
+# ── FFmpeg 退出码映射 ──────────────────────────────────────────────────────────
+
+# 常见 FFmpeg 退出码及其含义
+FFMPEG_EXIT_CODES: dict[int, tuple[str, str]] = {
+    0: ("成功", "渲染正常完成"),
+    1: ("通用错误", "FFmpeg 执行出错，请检查输入参数和素材"),
+    69: ("权限错误", "无权限写入输出文件或访问输入文件"),
+    126: ("权限不足", "命令不可执行"),
+    127: ("命令不存在", "FFmpeg 二进制文件未找到"),
+    134: ("Abnormal termination", "FFmpeg 异常终止（可能内存不足）"),
+    137: ("OOM Killed", "FFmpeg 被系统 OOM Killer 终止（内存不足）"),
+    139: ("段错误", "FFmpeg Segmentation Fault（可能是编解码器 Bug）"),
+    141: ("管道断裂", "FFmpeg 输出管道断裂"),
+    143: ("SIGTERM", "FFmpeg 收到终止信号"),
+    183: ("滤镜错误", "FFmpeg 滤镜链配置错误"),
+    234: ("素材异常", "输入素材格式不兼容或已损坏"),
+    255: ("严重错误", "FFmpeg 执行严重错误"),
+}
+
+
+def get_exit_code_message(exit_code: int) -> str:
+    """获取 FFmpeg 退出码的中文描述。
+
+    Args:
+        exit_code: FFmpeg 进程退出码
+
+    Returns:
+        人类可读的错误描述
+    """
+    if exit_code in FFMPEG_EXIT_CODES:
+        name, desc = FFMPEG_EXIT_CODES[exit_code]
+        return f"exit={exit_code} ({name}): {desc}"
+    if exit_code > 128:
+        signal_num = exit_code - 128
+        return f"exit={exit_code}: 被信号 {signal_num} 终止"
+    return f"exit={exit_code}: 未知错误"
+
+
+# ── 校验函数 ──────────────────────────────────────────────────────────────────
+
+
+def validate_video_output(video_path: str | Path, *, min_duration: float = 0.1) -> VideoValidationResult:
+    """校验渲染输出视频文件的完整性。
+
+    校验项：
+      1. 文件存在且大小 > 0
+      2. 包含 moov atom（MP4 容器完整性）
+      3. ffprobe 可读取至少一个视频流
+
+    Args:
+        video_path: 输出视频文件路径
+        min_duration: 最小有效时长（秒），低于此值视为无效，默认 0.1s
+
+    Returns:
+        VideoValidationResult
+    """
+    path = Path(video_path)
+    result = VideoValidationResult(valid=False)
+
+    # 1. 文件存在性检查
+    if not path.exists():
+        result.error_message = f"输出文件不存在: {path}"
+        logger.error("[video-validation] %s", result.error_message)
+        return result
+
+    result.file_exists = True
+
+    # 2. 文件大小检查
+    try:
+        result.file_size = path.stat().st_size
+    except OSError as e:
+        result.error_message = f"无法读取文件大小: {e}"
+        logger.error("[video-validation] %s", result.error_message)
+        return result
+
+    if result.file_size == 0:
+        result.error_message = "输出文件大小为 0（FFmpeg 未写入任何数据）"
+        logger.error("[video-validation] %s", result.error_message)
+        return result
+
+    # 极小文件（< 1KB）几乎不可能是有效视频
+    if result.file_size < 1024:
+        result.error_message = f"输出文件过小 ({result.file_size} bytes)，可能渲染未完成"
+        logger.error("[video-validation] %s", result.error_message)
+        return result
+
+    # 3. moov atom 检查（MP4 容器完整性标志）
+    result.moov_atom_found = _check_moov_atom(path)
+    if not result.moov_atom_found:
+        result.error_message = (
+            "输出文件缺少 moov atom（MP4 容器不完整）。" "可能原因：FFmpeg 被强制终止、磁盘空间不足、或渲染过程中断。"
+        )
+        logger.error("[video-validation] %s — file_size=%d", result.error_message, result.file_size)
+        return result
+
+    # 4. ffprobe 验证视频流
+    try:
+        probe_result = subprocess.run(  # nosec B603
+            [
+                FFPROBE_BIN,
+                "-v",
+                "error",
+                "-select_streams",
+                "v:0",
+                "-show_entries",
+                "stream=width,height,duration,codec_name",
+                "-show_entries",
+                "format=duration",
+                "-of",
+                "json",
+                str(path),
+            ],
+            check=True,
+            stdout=subprocess.PIPE,
+            stderr=subprocess.PIPE,
+            text=True,
+            timeout=15,
+        )
+
+        import json
+
+        probe_data = json.loads(probe_result.stdout)
+        streams = probe_data.get("streams", [])
+        fmt = probe_data.get("format", {})
+
+        if not streams:
+            result.error_message = "输出文件无视频流（ffprobe 未检测到 video stream）"
+            logger.error("[video-validation] %s", result.error_message)
+            return result
+
+        result.has_video_stream = True
+
+        video_stream = streams[0]
+        result.width = int(video_stream.get("width", 0) or 0)
+        result.height = int(video_stream.get("height", 0) or 0)
+        result.duration = float(fmt.get("duration", 0) or video_stream.get("duration", 0) or 0)
+
+        # 时长校验
+        if result.duration > 0 and result.duration < min_duration:
+            result.error_message = (
+                f"输出视频时长过短 ({result.duration:.2f}s < {min_duration}s)，" "可能渲染只处理了极少帧"
+            )
+            logger.warning("[video-validation] %s", result.error_message)
+            # 不标记为失败，只是警告（某些预览场景确实很短）
+            # 但如果时长为 0 且文件大小较大，说明 moov 有问题
+        elif result.duration == 0 and result.file_size > 0:
+            logger.warning(
+                "[video-validation] ffprobe 无法读取时长，但文件存在且大小=%d，标记为可疑",
+                result.file_size,
+            )
+
+    except subprocess.TimeoutExpired:
+        result.error_message = "ffprobe 超时（15s），输出文件可能已损坏"
+        logger.error("[video-validation] %s", result.error_message)
+        return result
+    except subprocess.CalledProcessError as e:
+        stderr_text = (e.stderr or "").strip()
+        result.error_message = f"ffprobe 校验失败: exit={e.returncode}, stderr={stderr_text[:200]}"
+        logger.error("[video-validation] %s", result.error_message)
+        return result
+    except (json.JSONDecodeError, KeyError, ValueError) as e:
+        result.error_message = f"ffprobe 输出解析失败: {e}"
+        logger.error("[video-validation] %s", result.error_message)
+        return result
+
+    # 全部通过
+    result.valid = True
+    logger.info(
+        "[video-validation] 校验通过: path=%s size=%d duration=%.2fs resolution=%dx%d",
+        path,
+        result.file_size,
+        result.duration,
+        result.width,
+        result.height,
+    )
+    return result
+
+
+def _check_moov_atom(video_path: Path) -> bool:
+    """检查 MP4 文件是否包含 moov atom。
+
+    moov atom 是 MP4 容器的元数据容器，包含视频时长、编解码器信息等。
+    FFmpeg 使用 -movflags +faststart 时 moov 在文件头部；否则在尾部。
+    如果 FFmpeg 被强制终止，moov 可能完全不存在。
+
+    方法：读取文件前 64KB + 尾部 64KB，搜索 "moov" 字节标记。
+    对于 -movflags +faststart 的快启文件，moov 在头部。
+
+    Args:
+        video_path: 视频文件路径
+
+    Returns:
+        True 表示找到 moov atom
+    """
+    try:
+        file_size = video_path.stat().st_size
+        if file_size < 8:
+            return False
+
+        # 搜索范围：头部 64KB + 尾部 64KB（覆盖 faststart 和普通 MP4）
+        search_size = min(64 * 1024, file_size)
+
+        with open(video_path, "rb") as f:
+            # 读取头部
+            head_data = f.read(search_size)
+            if b"moov" in head_data:
+                return True
+
+            # 读取尾部
+            if file_size > search_size:
+                f.seek(file_size - search_size)
+                tail_data = f.read(search_size)
+                if b"moov" in tail_data:
+                    return True
+
+        return False
+
+    except OSError as e:
+        logger.warning("[video-validation] 检查 moov atom 失败: %s", e)
+        return False

apps/worker/worker_app/tasks/_startup.py

@@ -1,4 +1,4 @@
-"""Worker 启动时的初始化任务 — 孤儿任务清理等。"""
+"""Worker 启动时的初始化任务 — 孤儿任务清理等."""
 
 import logging
 
@@ -7,11 +7,12 @@ from worker_app.db import SessionLocal
 
 logger = logging.getLogger(__name__)
 
+# 孤儿任务超时阈值：渲染任务超过此时间未更新则视为卡死
 ORPHAN_TASK_TIMEOUT_MINUTES = 10
 
 
-def cleanup_orphan_tasks(timeout_minutes: int = ORPHAN_TASK_TIMEOUT_MINUTES) -> int:
-    """清理数据库中超时未更新的 running 任务。
+def cleanup_orphan_tasks(timeout_minutes: int = ORPHAN_TASK_TIMEOUT_MINUTES) -> int:  # pragma: no cover
+    """清理数据库中超时未更新的 running GenerationTask（孤儿任务）。
 
     worker 重启或崩溃后，之前处于 running 状态的任务会变成孤儿任务，
     一直卡在 running 不动。通过 updated_at 超时判断并标记为 failed。
@@ -32,18 +33,81 @@ def cleanup_orphan_tasks(timeout_minutes: int = ORPHAN_TASK_TIMEOUT_MINUTES) ->
         count = repo.cleanup_stale_running(timeout_minutes)
         session.close()
         if count > 0:
-            logger.warning("清理了 %d 个超时的孤儿 running 任务", count)
+            logger.warning("清理了 %d 个超时的孤儿 GenerationTask（超过 %d 分钟未更新）", count, timeout_minutes)
         else:
-            logger.info("无孤儿 running 任务需要清理")
+            logger.info("无孤儿 GenerationTask 需要清理")
         return count
     except Exception as e:
-        logger.error("清理孤儿任务失败: %s", e, exc_info=True)
+        logger.error("清理孤儿 GenerationTask 失败: %s", e, exc_info=True)
         return 0
 
 
+def cleanup_stale_jobs(timeout_minutes: int = ORPHAN_TASK_TIMEOUT_MINUTES) -> int:  # pragma: no cover
+    """清理数据库中超时未更新的 running Job（孤儿任务）。
+
+    与 cleanup_orphan_tasks 配合，同时清理 Job 表和 GenerationTask 表。
+
+    Returns:
+        清理的任务数量
+    """
+    from datetime import datetime, timedelta, timezone
+
+    from packages.adapters.sqlalchemy_impl.models import JobModel
+    from packages.domain.job import JobStatus
+
+    try:
+        session = SessionLocal()
+        cutoff = datetime.now(timezone.utc) - timedelta(minutes=timeout_minutes)
+        stale_jobs = (
+            session.query(JobModel)
+            .filter(
+                JobModel.status == JobStatus.RUNNING.value,
+                JobModel.updated_at < cutoff,
+            )
+            .all()
+        )
+        count = 0
+        for model in stale_jobs:
+            model.status = JobStatus.FAILED.value
+            model.error_message = f"任务执行中断（超过 {timeout_minutes} 分钟未更新）"
+            count += 1
+        if count > 0:
+            session.commit()
+            logger.warning("清理了 %d 个超时的孤儿 Job（超过 %d 分钟未更新）", count, timeout_minutes)
+        else:
+            logger.info("无孤儿 Job 需要清理")
+        session.close()
+        return count
+    except Exception as e:
+        logger.error("清理孤儿 Job 失败: %s", e, exc_info=True)
+        return 0
+
+
+def cleanup_all_stale_tasks(timeout_minutes: int = ORPHAN_TASK_TIMEOUT_MINUTES) -> dict:  # pragma: no cover
+    """统一清理所有超时的孤儿任务。
+
+    同时清理 GenerationTask 和 Job 两类表。
+
+    Returns:
+        {"generation_tasks": int, "jobs": int}
+    """
+    gen_count = cleanup_orphan_tasks(timeout_minutes)
+    job_count = cleanup_stale_jobs(timeout_minutes)
+    total = gen_count + job_count
+    if total > 0:
+        logger.warning(
+            "孤儿任务清理完成: GenerationTask=%d, Job=%d, 总计=%d",
+            gen_count,
+            job_count,
+            total,
+        )
+    return {"generation_tasks": gen_count, "jobs": job_count}
+
+
 @worker_ready.connect
-def _on_worker_ready(sender, **kwargs):
+def _on_worker_ready(sender, **kwargs):  # pragma: no cover
     """Worker 启动完成后执行 — 清理孤儿任务。"""
     logger.info("Worker 启动完成，开始清理孤儿 running 任务...")
-    count = cleanup_orphan_tasks()
-    logger.info("Worker 启动清理完成，共清理 %d 个孤儿任务", count)
+    result = cleanup_all_stale_tasks()
+    total = result["generation_tasks"] + result["jobs"]
+    logger.info("Worker 启动清理完成，共清理 %d 个孤儿任务", total)

apps/worker/worker_app/tasks/compose_video.py

@@ -6,15 +6,22 @@
 from __future__ import annotations
 
 import os
+import subprocess  # pragma: no cover
 import tempfile
 from pathlib import Path
 
+from celery.exceptions import SoftTimeLimitExceeded  # pragma: no cover
 from celery.utils.log import get_task_logger
 from worker_app.celery_app import celery_app
 from worker_app.db import SessionLocal
 
 logger = get_task_logger(__name__)
 
+# 任务超时时间（秒）：超过此时间 Celery 会抛出 SoftTimeLimitExceeded
+RENDER_TASK_SOFT_TIME_LIMIT = 600  # pragma: no cover  # 10 分钟
+# 硬超时：超过此时间进程会被强制 kill
+RENDER_TASK_TIME_LIMIT = 660  # pragma: no cover  # 10 分钟 + 1 分钟清理缓冲
+
 
 def _get_job_service():
     """延迟导入 JobService，避免循环依赖。"""
@@ -31,8 +38,10 @@ def _get_job_service():
     bind=True,
     max_retries=3,
     default_retry_delay=60,
+    soft_time_limit=RENDER_TASK_SOFT_TIME_LIMIT,
+    time_limit=RENDER_TASK_TIME_LIMIT,
 )
-def compose_video(self, job_id: str, **kwargs):
+def compose_video(self, job_id: str, **kwargs):  # pragma: no cover
     """视频合成任务。
 
     使用 UnifiedRenderService（图层架构）进行渲染。
@@ -57,9 +66,51 @@ def compose_video(self, job_id: str, **kwargs):
         # 使用 unified 渲染引擎
         return _compose_with_unified_engine(self, job_service, job, plan_id, db)
 
+    except SoftTimeLimitExceeded:
+        # Celery 软超时：任务执行超过 soft_time_limit
+        error_msg = f"渲染任务超时（超过 {RENDER_TASK_SOFT_TIME_LIMIT // 60} 分钟）"
+        logger.error("视频合成超时: job_id=%s", job_id)
+        try:
+            job_service.fail_job(job_id, error_msg)
+        except Exception:
+            logger.exception("更新 Job 超时失败状态时出错")
+        # 超时不重试
+        return {"status": "error", "message": error_msg, "error_type": "timeout"}
+
+    except subprocess.TimeoutExpired as exc:
+        # FFmpeg 子进程超时
+        error_msg = f"FFmpeg 渲染超时（{exc.timeout}s）"
+        logger.error("视频合成 FFmpeg 超时: job_id=%s timeout=%s", job_id, exc.timeout)
+        try:
+            job_service.fail_job(job_id, error_msg)
+        except Exception:
+            logger.exception("更新 Job 超时失败状态时出错")
+        # 超时不重试
+        return {"status": "error", "message": error_msg, "error_type": "ffmpeg_timeout"}
+
     except self.retry_exc as exc:
         logger.warning("视频合成重试中: job_id=%s, exc=%s", job_id, exc)
         raise
+
+    except subprocess.CalledProcessError as exc:
+        # FFmpeg 执行失败，提取有意义的错误信息
+        from video_processing.video_validation import get_exit_code_message
+
+        exit_msg = get_exit_code_message(exc.returncode)
+        stderr_text = (exc.stderr or "").strip()
+        stderr_tail = stderr_text[-300:] if len(stderr_text) > 300 else stderr_text
+        error_msg = f"渲染失败: {exit_msg}"
+        if stderr_tail:
+            error_msg += f" | {stderr_tail[:200]}"
+
+        logger.error("视频合成 FFmpeg 失败: job_id=%s %s", job_id, exit_msg)
+        try:
+            job_service.fail_job(job_id, error_msg[:500])
+        except Exception:
+            logger.exception("更新 Job 失败状态时出错")
+        # FFmpeg 错误不重试（通常是素材或配置问题）
+        return {"status": "error", "message": error_msg, "error_type": "ffmpeg_error", "exit_code": exc.returncode}
+
     except Exception as exc:
         logger.exception("视频合成异常: job_id=%s", job_id)
         try:
@@ -71,7 +122,7 @@ def compose_video(self, job_id: str, **kwargs):
         db.close()
 
 
-def _compose_with_unified_engine(task, job_service, job, plan_id: str, db) -> dict:
+def _compose_with_unified_engine(task, job_service, job, plan_id: str, db) -> dict:  # pragma: no cover
     """新引擎渲染路径（UnifiedRenderService + RenderAdapter）。"""
     job_id = job.id
 
@@ -108,7 +159,8 @@ def _compose_with_unified_engine(task, job_service, job, plan_id: str, db) -> di
     )
 
     if not result.success:
-        job_service.fail_job(job_id, f"渲染失败: {result.error_message}")
+        error_msg = f"渲染失败: {result.error_message}"
+        job_service.fail_job(job_id, error_msg[:500])
         raise RuntimeError(result.error_message)
 
     # 更新 Job 状态为完成

apps/worker/worker_app/tasks/edit_plan_generation.py

@@ -271,8 +271,14 @@ def _render_with_unified(
     )
 
 
-@celery_app.task(name="worker.render_edit_plan", bind=True, max_retries=2)
-def render_edit_plan(self, plan_id: str) -> dict:
+@celery_app.task(
+    name="worker.render_edit_plan",
+    bind=True,
+    max_retries=2,
+    soft_time_limit=600,  # 10 分钟软超时
+    time_limit=660,  # 11 分钟硬超时
+)
+def render_edit_plan(self, plan_id: str) -> dict:  # pragma: no cover
     """渲染剪辑计划
 
     流程：
@@ -360,8 +366,17 @@ def render_edit_plan(self, plan_id: str) -> dict:
             return result
 
         except Exception as exc:
-            logger.exception("渲染剪辑计划异常: %s", plan_id)
+            # 超时异常不重试，直接标记失败
+            from celery.exceptions import SoftTimeLimitExceeded
+
+            is_timeout = isinstance(exc, SoftTimeLimitExceeded)
+            if is_timeout:
+                logger.error("渲染剪辑计划超时: plan_id=%s", plan_id)
+            else:
+                logger.exception("渲染剪辑计划异常: %s", plan_id)
+
             # 尝试标记计划和 GenerationTask 为失败
+            error_msg = "渲染任务超时（超过10分钟）" if is_timeout else f"渲染异常: {type(exc).__name__}: {exc}"
             try:
                 plan = plan_repo.get(plan_id)
                 if plan and plan.status.value == "rendering":
@@ -375,12 +390,13 @@ def render_edit_plan(self, plan_id: str) -> dict:
                     gen_task = gen_task_repo.get(generation_task_id)
                     if gen_task and gen_task.status.value != "failed":
                         gen_task.status = "failed"
-                        gen_task.error_message = f"渲染异常: {type(exc).__name__}: {exc}"
+                        gen_task.error_message = error_msg
                         gen_task.completed_at = datetime.now(timezone.utc)
                         try:
+                            log_stage = "render_timeout" if is_timeout else "render_failed"
                             gen_task.append_log(
-                                stage="render_failed",
-                                message=f"渲染异常: {type(exc).__name__}: {str(exc)[:500]}",
+                                stage=log_stage,
+                                message=error_msg[:500],
                                 level="ERROR",
                                 exception_type=type(exc).__name__,
                             )

apps/worker/worker_app/tasks/generation.py

@@ -1206,7 +1206,13 @@ def _upload_and_record(
 # ── Celery Task ──────────────────────────────────────────────────────────────
 
 
-@celery_app.task(bind=True, name="worker.generate_video", max_retries=2)
+@celery_app.task(
+    bind=True,
+    name="worker.generate_video",
+    max_retries=2,
+    soft_time_limit=600,  # 10 分钟软超时
+    time_limit=660,  # 11 分钟硬超时
+)
 def generate_video(self, task_id: str) -> dict:
     """生成视频任务 — 使用 UnifiedRenderService 统一渲染。
 

docs/1197_preview_generation_proposal.md

@@ -0,0 +1,208 @@
+# #1197 预览生成接口方案评估
+
+## 背景
+
+智能剪辑「一键生成」流程中，第3步预览生成当前被跳过，直接进入下一步。需要实现真正的预览生成功能，让用户在正式生成前能看到效果预览。
+
+## 现状分析
+
+### 现有生成链路
+
+```
+API 触发生成 → GenerationTask入库 → Celery异步任务 → UnifiedRenderService渲染 → OSS上传 → 更新状态
+```
+
+**关键节点：**
+1. **API层**：`POST /generation-tasks` 或 `POST /templates/{id}/generate` 触发生成
+2. **任务调度**：Celery task `worker.generate_video`
+3. **渲染引擎**：`UnifiedRenderService`（统一渲染引擎，已接入9个效果层）
+4. **输出配置**：默认 720p (1280x720)，支持 `resolution` 字段自定义
+5. **产物存储**：`GeneratedVideo` 表记录，OSS 存储视频文件
+
+### 已有可复用能力
+
+| 能力 | 位置 | 是否可复用 |
+|------|------|-----------|
+| 任务创建与状态管理 | `GenerationTask` + `CreateGenerationTaskUseCase` | ✅ 是 |
+| 素材下载与预处理 | `_download_video_assets` / `_download_voice_asset` | ✅ 是 |
+| 统一渲染引擎 | `UnifiedRenderService` | ✅ 是 |
+| 分辨率配置 | `resolution` 字段已支持 | ✅ 是 |
+| 混音与后处理 | `_render_video` 内流程 | ✅ 是 |
+| OSS 上传与查重 | `_upload_and_dedup` | ✅ 是 |
+| 进度追踪 | `append_log` / `progress` 字段 | ✅ 是 |
+
+## 方案对比
+
+### 方案A：复用现有生成链路 + is_preview 标记（推荐）
+
+**思路**：在现有 GenerationTask 上加 `is_preview` 标记，预览生成走完整链路但参数降级。
+
+**改动点：**
+1. **数据模型**：`GenerationTask` 加 `is_preview: bool` 字段（默认 false）；`GeneratedVideo` 加 `is_preview: bool`
+2. **API 层**：生成接口加 `is_preview` 参数，预览任务不计入配额
+3. **渲染参数**：预览模式下自动调整
+   - 分辨率：480p (854x480)
+   - 时长：限制前 15 秒（或模板第一个片段）
+   - 码率：降低至 1.5Mbps（正式 4Mbps）
+   - 效果层：跳过高级转场/粒子特效等耗时效果
+4. **任务调度**：预览任务走低优先级队列（或复用现有队列，标记优先级）
+5. **前端对接**：预览生成结果带 `is_preview=true` 标记，前端展示"预览"标签
+
+**优点：**
+- 代码复用率 90%+，改动最小
+- 与正式生成逻辑一致，预览效果真实可信
+- 进度查询、结果展示等功能直接复用
+- 后续可平滑升级：预览满意后一键转正式生成
+
+**缺点：**
+- 需要区分预览和正式任务，避免数据混淆
+- 预览任务和正式任务竞争同一队列资源（可后续优化为独立队列）
+
+**开发量估算**：2-3 天
+- 数据模型 + 迁移：0.5 天
+- API 层改造：0.5 天
+- 渲染参数降级：1 天
+- 测试 + 联调：1 天
+
+---
+
+### 方案B：新建独立预览接口 + 轻量渲染逻辑
+
+**思路**：新建独立的预览生成接口，使用简化的渲染逻辑（如只拼接素材+基础配音，跳过大部分效果）。
+
+**改动点：**
+1. 新增 `PreviewTask` 数据模型
+2. 新增 `POST /api/v1/preview/generate` 接口
+3. 新增独立的 Celery task `worker.generate_preview`
+4. 简化渲染流程：只做素材裁剪+拼接+配音，跳过转场/滤镜/字幕特效等
+
+**优点：**
+- 完全隔离，不影响正式生成链路
+- 可以做极致优化，预览生成速度快
+- 数据模型清晰，不会混淆
+
+**缺点：**
+- 代码重复率高，两套生成逻辑维护成本翻倍
+- 预览效果与正式生成可能不一致（效果层差异）
+- 前端需要对接两套接口
+- 无法从预览升级为正式生成（需重新走完整流程）
+
+**开发量估算**：4-5 天
+- 数据模型 + 接口：1 天
+- 简化渲染逻辑：2 天
+- 测试 + 联调：1-2 天
+
+---
+
+### 方案C：图片预览（首帧/关键帧截图）
+
+**思路**：不生成视频，只生成几张关键帧的预览图片。
+
+**优点：**
+- 生成速度极快（秒级）
+- 资源消耗小
+
+**缺点：**
+- 预览效果差，用户无法感知动态效果
+- 无法验证配音、转场、节奏等时间维度的效果
+- 用户体验不佳，不如"真预览"有说服力
+
+**开发量估算**：1-2 天
+
+---
+
+## 推荐方案：方案A（复用现有生成链路）
+
+### 核心理由
+
+1. **效果保真**：预览和正式生成用同一套渲染引擎，效果一致，用户信任度高
+2. **开发效率**：90% 代码复用，2-3 天可上线
+3. **可扩展性强**：后续可加「预览转正式」「低分辨率快速预览」等增强功能
+4. **维护成本低**：一套生成逻辑，bug 修复和新功能同时生效
+
+### 详细设计
+
+#### 1. 数据模型变更
+
+```python
+# GenerationTask 新增字段
+is_preview: bool = False
+"""是否为预览生成"""
+
+preview_of: str = ""
+"""预览对应的正式任务 ID（或反向关联）"""
+
+# GeneratedVideo 新增字段
+is_preview: bool = False
+"""是否为预览视频"""
+```
+
+**迁移**：alembic 新增 migration，两个表各加 1-2 个字段。
+
+#### 2. API 层
+
+```
+POST /api/v1/generation-tasks
+  Body 增加 is_preview: bool = false
+  
+POST /api/v1/templates/{id}/generate
+  Query 增加 is_preview: bool = false
+```
+
+**配额处理**：预览生成不计入用户配额，不占用生成次数限制。
+
+#### 3. 渲染参数降级
+
+| 参数 | 正式生成 | 预览生成 |
+|------|---------|---------|
+| 分辨率 | 720p (1280x720) | 480p (854x480) |
+| 码率 | 4 Mbps | 1.5 Mbps |
+| 时长 | 完整时长 | 前 15 秒（或第一段） |
+| 帧率 | 30 fps | 24 fps |
+| 转场效果 | 完整转场 | 仅淡入淡出（或简单切） |
+| 特效滤镜 | 全部启用 | 跳过粒子/光效等高级效果 |
+| 字幕 | 完整渲染 | 正常渲染（字幕是核心信息） |
+| 配音 | 完整混音 | 正常混音（配音是核心信息） |
+
+**实现方式**：在 `_render_video` 或 UnifiedRenderService 入口处，根据 `is_preview` 标记调整渲染配置。
+
+#### 4. 任务调度
+
+- 初期复用现有队列，预览任务正常排队
+- 后续如需优化，可拆分独立预览队列（低优先级）
+- 预览任务可设置较短超时时间
+
+#### 5. 前端对接
+
+- 调用生成接口时传 `is_preview=true`
+- 结果列表中预览视频带「预览」标签
+- 预览满意后可一键「升级为正式生成」（重新触发全分辨率生成，可复用素材下载缓存）
+
+### 实施步骤
+
+**Phase 1（MVP，2天）：**
+1. 数据模型 + 迁移
+2. API 层支持 is_preview 参数
+3. 渲染分辨率降级（480p）
+4. 不计入配额
+5. 基础测试
+
+**Phase 2（优化，1-2天）：**
+1. 时长限制（前15秒）
+2. 效果层降级（跳高级效果）
+3. 预览任务低优先级队列
+4. 预览转正式生成功能
+
+## 与前端对齐点
+
+1. 预览生成的触发时机（第3步自动生成？用户点击才生成？）
+2. 预览时长是固定15秒还是完整但低清？
+3. 是否需要「预览转正式生成」功能
+4. 预览视频的展示形态（和正式视频一样还是有特殊UI）
+
+## 风险与注意事项
+
+1. **数据混淆**：确保统计、计费、列表展示时正确区分预览和正式任务
+2. **存储成本**：预览视频也占 OSS 空间，可设置自动清理（7天后自动删除）
+3. **用户预期**：要明确告诉用户这是预览，效果和正式生成一致但清晰度低
+4. **并发压力**：如果用户频繁生成预览，可能增加系统负载，需要限流

tests/unit/test_render_adapter.py

@@ -10,6 +10,7 @@ from pathlib import Path
 from typing import Any
 from unittest.mock import MagicMock, patch
 
+import pytest
 from video_processing.render_adapter import RenderAdapter
 
 # ── Fixtures ──────────────────────────────────────────────────────────────────
@@ -114,6 +115,18 @@ def _make_adapter(
     return adapter, mock_plan_repo, mock_clip_repo
 
 
+@pytest.fixture(autouse=True)
+def _patch_validation():
+    """Mock video validation to always pass in tests (mock renders don't create real files)."""
+    from video_processing.video_validation import VideoValidationResult
+
+    with patch(
+        "video_processing.render_adapter.validate_video_output",
+        return_value=VideoValidationResult(valid=True),
+    ):
+        yield
+
+
 # ── validate_plan 测试 ───────────────────────────────────────────────────────
 
 

tests/unit/test_video_validation.py

@@ -0,0 +1,333 @@
+"""测试 video_validation 模块 — 渲染输出校验."""
+
+from __future__ import annotations
+
+import json
+import subprocess
+import tempfile
+from pathlib import Path
+from unittest.mock import MagicMock, patch
+
+import pytest
+from video_processing.video_validation import (
+    FFMPEG_EXIT_CODES,
+    VideoValidationResult,
+    _check_moov_atom,
+    get_exit_code_message,
+    validate_video_output,
+)
+
+# ── get_exit_code_message 测试 ────────────────────────────────────────────────
+
+
+class TestGetExitCodeMessage:
+    """FFmpeg 退出码映射测试."""
+
+    def test_known_exit_codes(self):
+        """已知退出码返回有意义的描述."""
+        assert "成功" in get_exit_code_message(0)
+        assert "通用错误" in get_exit_code_message(1)
+        assert "OOM" in get_exit_code_message(137)
+        assert "段错误" in get_exit_code_message(139)
+        assert "滤镜错误" in get_exit_code_message(183)
+        assert "素材异常" in get_exit_code_message(234)
+
+    def test_signal_termination(self):
+        """信号终止（exit > 128 且不在映射表中）返回信号编号."""
+        msg = get_exit_code_message(130)  # SIGINT = 130 - 128 = 2
+        assert "信号 2" in msg
+
+    def test_unknown_exit_code(self):
+        """未知退出码返回通用错误."""
+        msg = get_exit_code_message(42)
+        assert "未知错误" in msg
+        assert "42" in msg
+
+    def test_all_mapped_codes_have_description(self):
+        """所有映射的退出码都有名称和描述."""
+        for code, (name, desc) in FFMPEG_EXIT_CODES.items():
+            assert name, f"exit_code {code} 缺少名称"
+            assert desc, f"exit_code {code} 缺少描述"
+
+
+# ── _check_moov_atom 测试 ─────────────────────────────────────────────────────
+
+
+class TestCheckMoovAtom:
+    """moov atom 检测测试."""
+
+    def test_file_with_moov_in_head(self):
+        """文件头部包含 moov 标记 → True（faststart 模式）."""
+        with tempfile.NamedTemporaryFile(suffix=".mp4", delete=False) as f:
+            # 写入一些假数据，中间包含 moov 标记
+            f.write(b"\x00" * 100)
+            f.write(b"moov")
+            f.write(b"\x00" * 1000)
+            f.flush()
+            path = Path(f.name)
+
+        try:
+            assert _check_moov_atom(path) is True
+        finally:
+            path.unlink()
+
+    def test_file_with_moov_in_tail(self):
+        """文件尾部包含 moov 标记 → True（普通 MP4 模式）."""
+        with tempfile.NamedTemporaryFile(suffix=".mp4", delete=False) as f:
+            # 写一个较大的文件，moov 在尾部
+            f.write(b"\x00" * 100_000)
+            f.write(b"moov")
+            f.write(b"\x00" * 100)
+            f.flush()
+            path = Path(f.name)
+
+        try:
+            assert _check_moov_atom(path) is True
+        finally:
+            path.unlink()
+
+    def test_file_without_moov(self):
+        """文件不含 moov 标记 → False（截断/损坏的 MP4）."""
+        with tempfile.NamedTemporaryFile(suffix=".mp4", delete=False) as f:
+            # 写入纯随机数据，不含 moov
+            f.write(b"\x00\x01\x02\x03" * 1000)
+            f.flush()
+            path = Path(f.name)
+
+        try:
+            assert _check_moov_atom(path) is False
+        finally:
+            path.unlink()
+
+    def test_empty_file(self):
+        """空文件 → False."""
+        with tempfile.NamedTemporaryFile(suffix=".mp4", delete=False) as f:
+            path = Path(f.name)
+
+        try:
+            assert _check_moov_atom(path) is False
+        finally:
+            path.unlink()
+
+    def test_tiny_file(self):
+        """极小文件（< 8 bytes）→ False."""
+        with tempfile.NamedTemporaryFile(suffix=".mp4", delete=False) as f:
+            f.write(b"\x00\x01")
+            f.flush()
+            path = Path(f.name)
+
+        try:
+            assert _check_moov_atom(path) is False
+        finally:
+            path.unlink()
+
+    def test_nonexistent_file(self):
+        """不存在的文件 → False."""
+        assert _check_moov_atom(Path("/nonexistent/file.mp4")) is False
+
+
+# ── validate_video_output 测试 ────────────────────────────────────────────────
+
+
+class TestValidateVideoOutput:
+    """渲染输出完整性校验测试."""
+
+    def test_file_not_exists(self):
+        """文件不存在 → valid=False."""
+        result = validate_video_output("/nonexistent/video.mp4")
+        assert result.valid is False
+        assert result.file_exists is False
+        assert "不存在" in result.error_message
+
+    def test_empty_file(self):
+        """空文件 → valid=False."""
+        with tempfile.NamedTemporaryFile(suffix=".mp4", delete=False) as f:
+            path = Path(f.name)
+
+        try:
+            result = validate_video_output(path)
+            assert result.valid is False
+            assert "大小" in result.error_message or "过小" in result.error_message
+        finally:
+            path.unlink()
+
+    def test_tiny_file(self):
+        """极小文件（< 1KB）→ valid=False."""
+        with tempfile.NamedTemporaryFile(suffix=".mp4", delete=False) as f:
+            f.write(b"\x00" * 500)
+            path = Path(f.name)
+
+        try:
+            result = validate_video_output(path)
+            assert result.valid is False
+            assert "过小" in result.error_message
+        finally:
+            path.unlink()
+
+    def test_file_without_moov(self):
+        """文件有大小但无 moov atom → valid=False."""
+        with tempfile.NamedTemporaryFile(suffix=".mp4", delete=False) as f:
+            f.write(b"\x00\x01\x02\x03" * 2000)  # 8KB, 无 moov
+            path = Path(f.name)
+
+        try:
+            result = validate_video_output(path)
+            assert result.valid is False
+            assert "moov" in result.error_message.lower()
+        finally:
+            path.unlink()
+
+    @patch("video_processing.video_validation.subprocess.run")
+    def test_valid_video(self, mock_run):
+        """完整的有效视频 → valid=True."""
+        # mock ffprobe 返回
+        mock_result = MagicMock()
+        mock_result.stdout = json.dumps(
+            {
+                "streams": [
+                    {
+                        "width": 1080,
+                        "height": 1920,
+                        "duration": "10.5",
+                        "codec_name": "h264",
+                    }
+                ],
+                "format": {"duration": "10.5"},
+            }
+        )
+        mock_result.returncode = 0
+        mock_run.return_value = mock_result
+
+        # 创建含 moov 的文件
+        with tempfile.NamedTemporaryFile(suffix=".mp4", delete=False) as f:
+            f.write(b"\x00" * 100)
+            f.write(b"moov")
+            f.write(b"\x00" * 5000)
+            path = Path(f.name)
+
+        try:
+            result = validate_video_output(path)
+            assert result.valid is True
+            assert result.file_exists is True
+            assert result.moov_atom_found is True
+            assert result.has_video_stream is True
+            assert result.width == 1080
+            assert result.height == 1920
+            assert result.duration == 10.5
+            assert result.error_message == ""
+        finally:
+            path.unlink()
+
+    @patch("video_processing.video_validation.subprocess.run")
+    def test_no_video_stream(self, mock_run):
+        """文件有 moov 但无视频流 → valid=False."""
+        mock_result = MagicMock()
+        mock_result.stdout = json.dumps({"streams": [], "format": {"duration": "0"}})
+        mock_result.returncode = 0
+        mock_run.return_value = mock_result
+
+        with tempfile.NamedTemporaryFile(suffix=".mp4", delete=False) as f:
+            f.write(b"\x00" * 100)
+            f.write(b"moov")
+            f.write(b"\x00" * 5000)
+            path = Path(f.name)
+
+        try:
+            result = validate_video_output(path)
+            assert result.valid is False
+            assert "无视频流" in result.error_message
+        finally:
+            path.unlink()
+
+    @patch("video_processing.video_validation.subprocess.run")
+    def test_ffprobe_timeout(self, mock_run):
+        """ffprobe 超时 → valid=False."""
+        mock_run.side_effect = subprocess.TimeoutExpired(cmd="ffprobe", timeout=15)
+
+        with tempfile.NamedTemporaryFile(suffix=".mp4", delete=False) as f:
+            f.write(b"\x00" * 100)
+            f.write(b"moov")
+            f.write(b"\x00" * 5000)
+            path = Path(f.name)
+
+        try:
+            result = validate_video_output(path)
+            assert result.valid is False
+            assert "超时" in result.error_message
+        finally:
+            path.unlink()
+
+    @patch("video_processing.video_validation.subprocess.run")
+    def test_ffprobe_error(self, mock_run):
+        """ffprobe 执行失败 → valid=False."""
+        mock_run.side_effect = subprocess.CalledProcessError(returncode=1, cmd="ffprobe", stderr="Invalid data found")
+
+        with tempfile.NamedTemporaryFile(suffix=".mp4", delete=False) as f:
+            f.write(b"\x00" * 100)
+            f.write(b"moov")
+            f.write(b"\x00" * 5000)
+            path = Path(f.name)
+
+        try:
+            result = validate_video_output(path)
+            assert result.valid is False
+            assert "ffprobe" in result.error_message.lower()
+        finally:
+            path.unlink()
+
+    @patch("video_processing.video_validation.subprocess.run")
+    def test_short_video_warning(self, mock_run):
+        """极短视频（< min_duration）→ 仍 valid=True 但有警告."""
+        mock_result = MagicMock()
+        mock_result.stdout = json.dumps(
+            {
+                "streams": [{"width": 100, "height": 100, "duration": "0.05", "codec_name": "h264"}],
+                "format": {"duration": "0.05"},
+            }
+        )
+        mock_result.returncode = 0
+        mock_run.return_value = mock_result
+
+        with tempfile.NamedTemporaryFile(suffix=".mp4", delete=False) as f:
+            f.write(b"\x00" * 100)
+            f.write(b"moov")
+            f.write(b"\x00" * 5000)
+            path = Path(f.name)
+
+        try:
+            # 默认 min_duration=0.1，视频只有 0.05s → 应该仍然 valid（仅警告）
+            result = validate_video_output(path)
+            assert result.valid is True
+        finally:
+            path.unlink()
+
+
+# ── VideoValidationResult 数据结构测试 ────────────────────────────────────────
+
+
+class TestVideoValidationResult:
+    """VideoValidationResult 数据结构测试."""
+
+    def test_default_invalid(self):
+        """默认构造结果为无效."""
+        result = VideoValidationResult(valid=False)
+        assert result.valid is False
+        assert result.is_valid is False
+        assert result.file_size == 0
+        assert result.error_message == ""
+
+    def test_valid_result(self):
+        """有效结果属性正确."""
+        result = VideoValidationResult(
+            valid=True,
+            file_exists=True,
+            file_size=1024000,
+            moov_atom_found=True,
+            has_video_stream=True,
+            duration=15.3,
+            width=1920,
+            height=1080,
+        )
+        assert result.is_valid is True
+        assert result.duration == 15.3
+        assert result.width == 1920