Files
xiaoxia-saas/apps/worker/video_processing/reverse_engine.py
T
CI Bot 11a4d83cdb feat: 视频封面生成 + 视频倒放 + 贴纸叠加三个渲染能力
- cover_generator.py: 三种封面生成模式(指定时间抽帧/智能封面/自定义上传)
  - 指定时间点抽帧,默认第1秒,支持时间钳制
  - 智能封面:均匀抽3帧,选最清晰的一帧
  - 自定义封面:自动调整尺寸到标准格式
  - 失败降级:不阻断主流程

- reverse_engine.py: 视频/音频倒放引擎
  - 基于 FFmpeg reverse + areverse 滤镜
  - 每个 clip 独立配置,支持只倒视频或只倒音频
  - 安全限制:超过120s自动跳过,防止内存溢出
  - 配置从 clip.config.reverse 读取,零侵入数据模型

- sticker_engine.py: 贴纸叠加引擎
  - 图片贴纸:位置/大小/透明度/时间范围/淡入淡出
  - 文字贴纸(花字):字体/颜色/描边/阴影/动画
  - 9宫格预设 + 自由坐标(像素/百分比)
  - 多贴纸按 z_index 排序叠加
  - 5个预设贴纸分类
  - 降级:素材不存在自动跳过,不阻断渲染

- UnifiedRenderService 全链路接入
  - 倒放:Step 1 预处理(trim之后scale之前),filter_complex + 直通双路径
  - 贴纸:最终合成后字幕之前叠加,图片贴纸需额外输入
  - 有贴纸时禁用直通模式(需要filter_complex)
  - 音频倒放:render_audio 单clip + 多clip双路径接入

- 59个新增单测,覆盖核心场景 + 降级逻辑 + 集成测试
2026-07-14 10:29:29 +08:00

117 lines
3.9 KiB
Python
Executable File

"""视频倒放引擎 — 基于 FFmpeg reverse + areverse 滤镜实现视频/音频倒放.
支持能力:
- 视频倒放(reverse 滤镜)
- 音频倒放(areverse 滤镜)
- 按 clip 分段倒放,每个 clip 独立配置
- 降级策略:不支持时跳过,不阻断渲染
"""
from __future__ import annotations
import logging
from dataclasses import dataclass
from typing import Any
logger = logging.getLogger(__name__)
# ── 数据模型 ──────────────────────────────────────────────────────────────────
@dataclass
class ReverseConfig:
"""视频倒放配置.
从 clip.config.reverse 读取,零侵入数据模型.
"""
enabled: bool = False
reverse_video: bool = True # 是否倒放视频
reverse_audio: bool = True # 是否倒放音频
@classmethod
def from_dict(cls, data: dict[str, Any] | None) -> "ReverseConfig":
"""从字典解析配置."""
if not data:
return cls(enabled=False)
try:
if not data.get("enabled", False):
return cls(enabled=False)
return cls(
enabled=True,
reverse_video=bool(data.get("reverse_video", True)),
reverse_audio=bool(data.get("reverse_audio", True)),
)
except (AttributeError, TypeError) as e:
logger.warning("倒放配置解析失败: %s,使用默认配置", e)
return cls(enabled=False)
# ── 倒放引擎 ──────────────────────────────────────────────────────────────────
class ReverseEngine:
"""视频倒放引擎 — 生成 FFmpeg 倒放滤镜.
视频倒放:reverse 滤镜
音频倒放:areverse 滤镜
注意事项:
- reverse 滤镜需要将整个视频帧加载到内存,长视频可能占用大量内存
- 建议对单 clip 时长做限制(如 < 60s),超长视频建议降级
"""
# 安全限制:单 clip 超过此时长不启用倒放(防止内存溢出)
MAX_SAFE_DURATION = 120.0 # 秒
@staticmethod
def build_video_filter(config: ReverseConfig, duration: float = 0.0) -> str:
"""构建视频倒放滤镜字符串.
Args:
config: 倒放配置
duration: clip 时长(秒),用于安全检查
Returns:
FFmpeg 滤镜字符串,如 "reverse";无效果返回空字符串
"""
if not config.enabled or not config.reverse_video:
return ""
# 安全检查:超长视频不启用倒放
if duration > ReverseEngine.MAX_SAFE_DURATION:
logger.warning(
"视频倒放安全限制:clip 时长 %.1fs 超过上限 %.1fs,跳过倒放",
duration,
ReverseEngine.MAX_SAFE_DURATION,
)
return ""
return "reverse"
@staticmethod
def build_audio_filter(config: ReverseConfig, duration: float = 0.0) -> str:
"""构建音频倒放滤镜字符串.
Args:
config: 倒放配置
duration: clip 时长(秒),用于安全检查
Returns:
FFmpeg 音频滤镜字符串,如 "areverse";无效果返回空字符串
"""
if not config.enabled or not config.reverse_audio:
return ""
# 安全检查:超长音频不启用倒放
if duration > ReverseEngine.MAX_SAFE_DURATION:
logger.warning(
"音频倒放安全限制:clip 时长 %.1fs 超过上限 %.1fs,跳过倒放",
duration,
ReverseEngine.MAX_SAFE_DURATION,
)
return ""
return "areverse"