xiaoxia-saas/docs/video-editing-modes-design.md

四种视频剪辑模式 — 技术设计方案

状态: 草案（待代码审计审查） Issue: #22 作者: 后端查重 日期: 2026-06-29

---

1. 背景与目标

1.1 问题

当前系统已有 4 种 EditingMode（one_take、pip、voice_over、voice_pip），定义了视频的组装方式。但缺少对生成视频进行降重处理的能力——即通过画面变换使同一素材生成的视频在指纹层面（pHash、颜色直方图、MD5）产生差异，降低平台查重命中率。

1.2 目标

设计 4 种降重剪辑模式（DeduplicationMode），每种组合不同的降重技术子集，让用户在创建生成任务时选择。这是与 EditingMode 正交的维度：EditingMode 控制"怎么拼"，DeduplicationMode 控制"怎么变"。

1.3 六种降重技术（来自产品规划）

编号	技术	FFmpeg 实现	效果
T1	素材随机排列	shuffle / Python random.shuffle	改变帧序列顺序
T2	画面镜像翻转	hflip / vflip	水平/垂直镜像
T3	随机变速 (0.9x~1.1x)	setpts=PTS*k + atempo	微调播放速度
T4	随机裁剪区域	crop=w:h:x:y	裁掉边缘像素
T5	叠加滤镜/色调	curves / colorbalance / eq	改变颜色分布
T6	不同转场效果	xfade=transition=T	改变片段衔接方式

---

2. 四种降重模式设计

2.1 模式总览

模式	标识	降重强度	技术组合	适用场景
极简模式	minimal	★☆☆	T1	追求原始画质，仅打乱顺序
标准模式	standard	★★☆	T1 + T6	日常使用，轻度降重
创意模式	creative	★★★	T1 + T2 + T3 + T4 + T5 + T6	最大降重，全技术叠加
批量变体模式	batch_variant	★★☆	T1 + T2 + T3 + T5（可选 T4/T6）	同一素材生成多个不同视频

2.2 极简模式 (minimal)

降重策略: 仅随机排列素材顺序。

处理流程:

1. 接收素材列表 [v1, v2, ..., vn]
2. 使用种子随机打乱顺序 → [v3, v1, v5, ...]
3. 按现有 EditingMode 逻辑组装（无额外滤镜）

FFmpeg 滤镜链: 无额外滤镜，直接交给 EditingModeProcessor 处理。

性能: 零额外开销，处理速度与当前一致。

降重效果: 仅改变片段顺序，pHash 整体变化较小，但 MD5 指纹完全不同。

2.3 标准模式 (standard)

降重策略: 随机排列 + 随机转场。

处理流程:

1. 随机排列素材顺序
2. 为每对相邻片段随机选择转场效果
3. 按 EditingMode 组装，在片段间插入 xfade 转场

可选转场池:

TRANSITION_POOL = [
    "fade", "wipeleft", "wiperight", "wipeup", "wipedown",
    "slideleft", "slideright", "slideup", "slidedown",
    "circlecrop", "rectcrop", "distancefade", "smoothleft",
    "smoothright", "smoothup", "smoothdown",
]

FFmpeg 滤镜链（以 2 段视频为例）:

[0:v][1:v]xfade=transition={random_T}:duration={0.3~0.8}:offset={calc}[v]

性能: 轻微额外开销（xfade 已在 one_take 模式中使用）。

降重效果: 片段顺序 + 转场类型双重变化，pHash 中等程度改变。

2.4 创意模式 (creative)

降重策略: 全技术叠加，最大降重。

处理流程:

1. T1 随机排列: 打乱素材顺序
2. T2 镜像翻转: 对每个素材随机决定是否 hflip（50% 概率）
3. T3 随机变速: 对每个素材生成 0.9~1.1 的随机速度因子
4. T4 随机裁剪: 裁掉 2%~8% 的随机边缘
5. T5 滤镜色调: 从预设滤镜池中随机选择一个应用
6. T6 转场效果: 同标准模式

滤镜色调池:

COLOR_FILTER_POOL = [
    # 暖色调
    "colorbalance=rs=0.15:gs=0.05:bs=-0.1:rm=0.1:gm=0.05:bm=-0.08",
    # 冷色调
    "colorbalance=rs=-0.1:gs=0.05:bs=0.15:rm=-0.08:gm=0.05:bm=0.1",
    # 高对比
    "eq=contrast=1.15:brightness=0.02:saturation=1.2",
    # 低饱和（电影感）
    "eq=saturation=0.7:contrast=1.1",
    # 复古色调
    "curves=preset=cross_process",
    # 自然增强
    "eq=brightness=0.03:saturation=1.15:contrast=1.05",
]

FFmpeg 滤镜链（单个素材预处理）:

# 每个素材独立处理
hflip (50%概率)
→ crop=iw*(1-rand_crop_pct):ih*(1-rand_crop_pct):rand_x:rand_y
→ setpts=PTS*{speed_factor}
→ {color_filter}
→ scale=1280:720:force_original_aspect_ratio=decrease,
  pad=1280:720:(ow-iw)/2:(oh-ih)/2

# 片段间转场
[0:v][1:v]xfade=transition={T}:duration={d}:offset={o}[v]

性能: 显著额外开销。每个素材需独立预处理（约增加 30%~50% 处理时间）。

降重效果: pHash、颜色直方图、MD5 均大幅改变，降重效果最强。

2.5 批量变体模式 (batch_variant)

降重策略: 同一素材生成 N 个视觉不同的视频变体。

处理流程:

1. 接收素材列表 + 变体数量 N（默认 3）
2. 对每个变体 i（0..N-1）:
   • 使用种子 seed = base_seed + i 确保可复现
   • 应用 T1（随机排列）
   • 应用 T2（镜像翻转，种子控制）
   • 应用 T3（随机变速，种子控制）
   • 应用 T5（色调滤镜，种子控制）
   • 可选：T4（裁剪）、T6（转场）
3. 生成 N 个独立输出文件

种子机制:

import hashlib

def variant_seed(base_seed: str, variant_index: int) -> int:
    """为每个变体生成确定性种子"""
    h = hashlib.md5(f"{base_seed}:{variant_index}".encode())
    return int(h.hexdigest()[:8], 16)

API 行为: 一个 GenerationTask 产生 N 个 GeneratedVideo 记录，每个有不同的 generation_params（包含变体索引和种子）。

性能: 线性增长，N 个变体 ≈ N 倍处理时间。可通过并行 Celery 子任务加速。

---

3. 数据模型设计

3.1 新增 DeduplicationMode 枚举

文件: packages/domain/deduplication_mode.py

from enum import StrEnum

class DeduplicationMode(StrEnum):
    """降重剪辑模式"""
    MINIMAL = "minimal"           # 极简：仅随机排列
    STANDARD = "standard"         # 标准：排列 + 转场
    CREATIVE = "creative"         # 创意：全技术叠加
    BATCH_VARIANT = "batch_variant"  # 批量变体：多种变体

    @property
    def techniques(self) -> list[str]:
        """该模式使用的降重技术列表"""
        return _MODE_TECHNIQUES[self]

_MODE_TECHNIQUES = {
    DeduplicationMode.MINIMAL: ["shuffle"],
    DeduplicationMode.STANDARD: ["shuffle", "transition"],
    DeduplicationMode.CREATIVE: [
        "shuffle", "flip", "speed", "crop", "color_filter", "transition",
    ],
    DeduplicationMode.BATCH_VARIANT: [
        "shuffle", "flip", "speed", "color_filter",
    ],
}

3.2 GenerationTask 模型扩展

ORM 模型 (packages/adapters/sqlalchemy_impl/models.py):

class GenerationTaskModel(Base):
    # ... 现有字段 ...
    deduplication_mode = Column(
        String(20),
        default="standard",
        nullable=False,
    )
    dedup_seed = Column(String(64), nullable=True)  # 可选的确定性种子
    variant_count = Column(Integer, default=1)       # batch_variant 模式的变体数
    variant_index = Column(Integer, default=0)       # 当前变体索引

Alembic 迁移: 新增迁移脚本，添加上述 3 列，默认值兼容现有数据。

3.3 GenerationParams 扩展

GeneratedVideo.generation_params 字典新增字段:

{
    "deduplication_mode": "creative",
    "dedup_seed": "a1b2c3d4",
    "variant_index": 0,           # batch_variant 时 > 0
    "applied_techniques": ["shuffle", "flip", "speed", "crop", "color_filter", "transition"],
    "technique_params": {         # 记录实际应用的参数，便于复现
        "flip": "hflip",
        "speed_factor": 0.97,
        "crop_percent": 0.04,
        "color_filter": "warm_tone",
        "transition": "wipeleft",
    },
}

---

4. API 设计

4.1 创建任务接口扩展

POST /api/v1/tasks 请求体新增字段:

class CreateGenerationTaskRequest(BaseModel):
    project_id: str
    asset_library_id: str
    strategy_id: str | None = None
    voice_library_id: str | None = None
    # 新增字段
    deduplication_mode: DeduplicationMode = DeduplicationMode.STANDARD
    dedup_seed: str | None = None          # 不传则自动生成
    variant_count: int = Field(default=1, ge=1, le=10)  # batch_variant 用

4.2 查询任务接口扩展

GET /api/v1/tasks/{task_id} 响应新增:

class GenerationTaskResponse(BaseModel):
    # ... 现有字段 ...
    deduplication_mode: str
    variant_count: int
    variants: list[VariantInfo] | None  # batch_variant 时返回各变体状态

class VariantInfo(BaseModel):
    variant_index: int
    status: str
    video_id: str | None

4.3 新增查询模式列表接口

GET /api/v1/deduplication-modes (公开接口，无需鉴权):

class DeduplicationModeInfo(BaseModel):
    mode: str
    name: str
    description: str
    techniques: list[str]
    estimated_processing_time: str  # "fast" / "normal" / "slow"

# 返回:
[
    {
        "mode": "minimal",
        "name": "极简模式",
        "description": "仅随机排列素材顺序，保持原始画质",
        "techniques": ["素材随机排列"],
        "estimated_processing_time": "fast",
    },
    {
        "mode": "standard",
        "name": "标准模式",
        "description": "随机排列 + 随机转场，轻度降重",
        "techniques": ["素材随机排列", "转场效果"],
        "estimated_processing_time": "fast",
    },
    {
        "mode": "creative",
        "name": "创意模式",
        "description": "全技术叠加：排列、翻转、变速、裁剪、滤镜、转场",
        "techniques": ["素材随机排列", "画面翻转", "随机变速", "随机裁剪", "滤镜色调", "转场效果"],
        "estimated_processing_time": "slow",
    },
    {
        "mode": "batch_variant",
        "name": "批量变体模式",
        "description": "同一素材生成多个不同视频，适合批量投放",
        "techniques": ["素材随机排列", "画面翻转", "随机变速", "滤镜色调"],
        "estimated_processing_time": "normal",
    },
]

---

5. Worker 实现方案

5.1 架构概览

┌─────────────────────────────────────────────────────────┐
│                  generate_video (Celery Task)            │
│                                                          │
│  1. Load GenerationTask                                  │
│  2. Download assets from OSS                             │
│  3. ┌──────────────────────────────────────────────┐    │
│     │  DeduplicationProcessor (NEW)                 │    │
│     │  ├─ shuffle_assets()                          │    │
│     │  ├─ apply_per_asset_transforms()              │    │
│     │  │   ├─ flip()                                │    │
│     │  │   ├─ speed_change()                        │    │
│     │  │   ├─ random_crop()                         │    │
│     │  │   └─ color_filter()                        │    │
│     │  └─ apply_transitions()                       │    │
│     └──────────────────────────────────────────────┘    │
│  4. EditingModeProcessor (EXISTING, unchanged)          │
│  5. Upload result to OSS                                │
│  6. Create GeneratedVideo record                        │
└─────────────────────────────────────────────────────────┘

5.2 DeduplicationProcessor 类设计

文件: apps/worker/video_processing/deduplication_processor.py

@dataclass
class DeduplicationConfig:
    mode: DeduplicationMode
    seed: str                    # 确定性随机种子
    variant_count: int = 1
    variant_index: int = 0
    # 可覆盖的默认参数范围
    speed_range: tuple[float, float] = (0.9, 1.1)
    crop_range: tuple[float, float] = (0.02, 0.08)  # 裁剪百分比
    transition_duration_range: tuple[float, float] = (0.3, 0.8)


class DeduplicationProcessor:
    """降重处理器 — 在 EditingModeProcessor 之前对素材进行预处理"""

    def __init__(self, config: DeduplicationConfig, work_dir: str):
        self.config = config
        self.work_dir = work_dir
        self.rng = self._init_rng()

    def _init_rng(self) -> random.Random:
        """基于种子初始化确定性随机数生成器"""
        seed_int = int(hashlib.md5(
            f"{self.config.seed}:{self.config.variant_index}".encode()
        ).hexdigest()[:8], 16)
        return random.Random(seed_int)

    def process(self, video_paths: list[str]) -> list[str]:
        """
        对素材列表应用降重处理，返回处理后的临时文件路径列表。
        后续的 EditingModeProcessor 对这些路径进行组装。
        """
        paths = list(video_paths)

        # T1: 随机排列（所有模式都有）
        paths = self._shuffle(paths)

        # T2/T3/T4/T5: 逐素材预处理（creative 和 batch_variant）
        if self.config.mode in (DeduplicationMode.CREATIVE, DeduplicationMode.BATCH_VARIANT):
            paths = self._per_asset_transforms(paths)

        # T6: 转场信息传递给 EditingModeProcessor（standard 及以上）
        # 转场在 EditingModeProcessor 中应用，这里记录转场参数
        self._prepare_transition_params(len(paths))

        return paths

    def _shuffle(self, paths: list[str]) -> list[str]:
        """T1: 随机排列素材"""
        shuffled = list(paths)
        self.rng.shuffle(shuffled)
        return shuffled

    def _per_asset_transforms(self, paths: list[str]) -> list[str]:
        """T2+T3+T4+T5: 对每个素材独立应用变换"""
        result = []
        for path in paths:
            current = path
            techniques = self.config.mode.techniques

            if "flip" in techniques and self.rng.random() < 0.5:
                current = self._apply_flip(current)

            if "speed" in techniques:
                factor = self.rng.uniform(*self.config.speed_range)
                current = self._apply_speed(current, factor)

            if "crop" in techniques:
                pct = self.rng.uniform(*self.config.crop_range)
                current = self._apply_crop(current, pct)

            if "color_filter" in techniques:
                filter_choice = self.rng.choice(COLOR_FILTER_POOL)
                current = self._apply_color_filter(current, filter_choice)

            result.append(current)
        return result

    def _apply_flip(self, path: str) -> str:
        """T2: 水平翻转"""
        output = self._temp_path(path, "flipped")
        # ffmpeg -i input -vf hflip -c:v libx264 -crf 23 output
        ...
        return output

    def _apply_speed(self, path: str, factor: float) -> str:
        """T3: 变速 (setpts + atempo)"""
        output = self._temp_path(path, "speed")
        pts_factor = 1.0 / factor
        # ffmpeg -i input -vf "setpts={pts_factor}*PTS"
        #        -af "atempo={factor}" -c:v libx264 -crf 23 output
        ...
        return output

    def _apply_crop(self, path: str, pct: float) -> str:
        """T4: 随机裁剪"""
        output = self._temp_path(path, "cropped")
        # 先获取视频尺寸，计算裁剪参数
        # ffmpeg -i input -vf "crop=iw*(1-pct):ih*(1-pct):rand_x:rand_y"
        #        + scale 回原始尺寸
        ...
        return output

    def _apply_color_filter(self, path: str, filter_expr: str) -> str:
        """T5: 色调滤镜"""
        output = self._temp_path(path, "filtered")
        # ffmpeg -i input -vf "{filter_expr}" -c:v libx264 -crf 23 output
        ...
        return output

    def get_transition_params(self) -> list[dict]:
        """T6: 返回每对相邻片段的转场参数"""
        return self._transition_params

5.3 EditingModeProcessor 适配

在 EditingModeProcessor.process() 中增加转场参数的接收：

def process(
    self,
    video_paths: list[str],
    audio_path: Optional[str] = None,
    output_path: Optional[str] = None,
    transition_params: Optional[list[dict]] = None,  # 新增
) -> str:

当 transition_params 不为 None 时，_one_take_with_xfade 使用指定的转场类型而非默认 fade。

5.4 Celery Task 适配

generate_video task 的修改:

@celery_app.task(bind=True, name="worker.generate_video", max_retries=2)
def generate_video(self, task_id: str) -> dict:
    # ... 现有逻辑 ...

    # 新增: 降重处理
    dedup_mode = gen_task.deduplication_mode or "standard"
    dedup_config = DeduplicationConfig(
        mode=DeduplicationMode(dedup_mode),
        seed=gen_task.dedup_seed or uuid4().hex,
        variant_count=gen_task.variant_count or 1,
        variant_index=gen_task.variant_index or 0,
    )
    dedup_processor = DeduplicationProcessor(dedup_config, work_dir)
    processed_paths = dedup_processor.process(local_video_paths)

    # 现有: EditingMode 处理（使用降重后的路径）
    processor = create_processor(mode, work_dir=work_dir)
    output = processor.process(
        processed_paths,
        audio_path=local_audio_path,
        transition_params=dedup_processor.get_transition_params(),
    )

    # ... 上传 + 记录 generation_params ...

5.5 batch_variant 模式的并行策略

当 deduplication_mode == "batch_variant" 且 variant_count > 1 时:

方案 A（推荐）: 在 API 层拆分为多个子任务

# API 层: 创建 N 个 GenerationTask，每个 variant_index 不同
for i in range(variant_count):
    task = GenerationTaskModel(
        ...,
        deduplication_mode="batch_variant",
        dedup_seed=base_seed,
        variant_count=variant_count,
        variant_index=i,
    )
    db.add(task)
    generate_video.delay(task.id)

方案 B: 在 Worker 层串行处理

在单个 Celery task 内循环生成 N 个变体。简单但慢。

推荐方案 A：利用 Celery 并行能力，N 个变体同时处理。

---

6. 文件变更清单

文件	变更类型	说明
packages/domain/deduplication_mode.py	新增	DeduplicationMode 枚举
packages/adapters/sqlalchemy_impl/models.py	修改	GenerationTaskModel 新增 3 列
alembic/versions/xxx_add_dedup_mode.py	新增	数据库迁移
apps/api/app/api/routes/generation_tasks.py	修改	请求/响应模型新增字段
apps/api/app/schemas/generation.py	修改	Schema 新增字段
apps/api/app/api/routes/deduplication_modes.py	新增	查询模式列表接口
apps/worker/video_processing/deduplication_processor.py	新增	降重处理器核心实现
apps/worker/video_processing/editing_modes.py	修改	支持自定义转场参数
apps/worker/worker_app/tasks/generation.py	修改	集成 DeduplicationProcessor

---

7. 实施计划

Phase 1: 基础设施（预计 2 天）

1. 新增 DeduplicationMode 枚举
2. 数据库迁移（新增列）
3. DeduplicationProcessor 核心实现 + 单元测试

Phase 2: API 集成（预计 1 天）

4. 扩展创建任务 API
5. 新增查询模式列表 API
6. 扩展查询任务 API

Phase 3: Worker 集成（预计 2 天）

7. EditingModeProcessor 适配转场参数
8. generate_video task 集成降重处理
9. batch_variant 并行策略实现

Phase 4: 测试与优化（预计 1 天）

10. 端到端集成测试
11. 降重效果验证（pHash / 直方图对比）
12. 性能基准测试

---

8. 风险与缓解

风险	影响	缓解措施
创意模式处理时间过长	用户体验差	设置超时告警；提供进度回调
降重后画质下降明显	用户不满	crop 限制在 8% 以内；CRF 保持 23
FFmpeg 滤镜链复杂度增加	调试困难	每个变换步骤独立临时文件，便于排查
batch_variant 并行任务过多	Celery 队列拥堵	限制 variant_count ≤ 10；优先级队列
随机种子不可复现	无法重现结果	所有随机操作基于确定性 RNG

---

9. 与现有系统的兼容性

• 向后兼容: deduplication_mode 默认值为 "standard"，现有任务不受影响
• EditingMode 正交: 降重模式与组装模式独立，可自由组合（如 one_take + creative）
• 查重系统: 降重后的视频通过现有 VideoDeduplicator 验证降重效果
• 指纹记录: GeneratedVideo.video_fingerprint 记录降重后的指纹，用于后续查重

---

附录 A: FFmpeg 滤镜参考

变速 (setpts + atempo)

# 1.05x 加速
ffmpeg -i input.mp4 \
  -vf "setpts=PTS/1.05" \
  -af "atempo=1.05" \
  -c:v libx264 -crf 23 output.mp4

# 注意: atempo 只接受 [0.5, 2.0] 范围，0.9~1.1 安全

随机裁剪

# 裁掉 5% 边缘（从左上角随机偏移）
ffmpeg -i input.mp4 \
  -vf "crop=iw*0.95:ih*0.95:((iw-iw*0.95)/2):((ih-ih*0.95)/2),scale=1280:720" \
  -c:v libx264 -crf 23 output.mp4

色调调整

# 暖色调
ffmpeg -i input.mp4 \
  -vf "colorbalance=rs=0.15:gs=0.05:bs=-0.1:rm=0.1:gm=0.05:bm=-0.08" \
  -c:v libx264 -crf 23 output.mp4

xfade 转场

# wipeleft 转场
ffmpeg -i seg1.mp4 -i seg2.mp4 \
  -filter_complex "[0:v][1:v]xfade=transition=wipeleft:duration=0.5:offset=4.5[v]" \
  -map "[v]" -c:v libx264 -crf 23 output.mp4