From df104f35831b71f8aac27c077dffa024fed26ec8 Mon Sep 17 00:00:00 2001 From: xiaoxia-bot Date: Tue, 14 Jul 2026 11:01:24 +0800 Subject: [PATCH 1/2] =?UTF-8?q?feat:=20=E8=A7=86=E9=A2=91=E5=B0=81?= =?UTF-8?q?=E9=9D=A2=E7=94=9F=E6=88=90=20+=20=E8=A7=86=E9=A2=91=E5=80=92?= =?UTF-8?q?=E6=94=BE=20+=20=E8=B4=B4=E7=BA=B8=E5=8F=A0=E5=8A=A0=E4=B8=89?= =?UTF-8?q?=E4=B8=AA=E6=B8=B2=E6=9F=93=E8=83=BD=E5=8A=9B?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../video_processing/cover_generator.py | 431 +++++++++ apps/worker/video_processing/render_audio.py | 28 +- .../worker/video_processing/reverse_engine.py | 116 +++ .../worker/video_processing/sticker_engine.py | 574 ++++++++++++ .../unified_render_service.py | 71 ++ tests/unit/test_cover_reverse_sticker.py | 860 ++++++++++++++++++ 6 files changed, 2076 insertions(+), 4 deletions(-) create mode 100755 apps/worker/video_processing/cover_generator.py create mode 100755 apps/worker/video_processing/reverse_engine.py create mode 100755 apps/worker/video_processing/sticker_engine.py create mode 100755 tests/unit/test_cover_reverse_sticker.py diff --git a/apps/worker/video_processing/cover_generator.py b/apps/worker/video_processing/cover_generator.py new file mode 100755 index 000000000..8e656e6d2 --- /dev/null +++ b/apps/worker/video_processing/cover_generator.py @@ -0,0 +1,431 @@ +"""视频封面生成器 — 从视频中提取/生成封面图. + +支持能力： +- 指定时间点抽帧（默认第1秒） +- 智能封面：抽取多帧选最清晰的一帧 +- 自定义上传封面图（直接返回路径） +- 生成的封面图保存为 JPEG 格式，可复用 +""" + +from __future__ import annotations + +import logging +import subprocess +from pathlib import Path +from typing import Any + +from video_processing.ffmpeg_utils import FFMPEG_BIN, probe_video_info, run_ffmpeg + +logger = logging.getLogger(__name__) + + +# ── 配置常量 ────────────────────────────────────────────────────────────────── + +# 智能封面抽帧数量 +SMART_COVER_FRAME_COUNT = 3 + +# 默认抽帧时间点（秒） +DEFAULT_COVER_TIME = 1.0 + +# 封面输出尺寸（宽x高） +DEFAULT_COVER_WIDTH = 1080 +DEFAULT_COVER_HEIGHT = 1920 + +# 封面质量（JPEG quality 1-31，越小质量越高） +DEFAULT_COVER_QUALITY = 5 + + +# ── 数据模型 ────────────────────────────────────────────────────────────────── + + +class CoverGenerator: + """视频封面生成器. + + 三种模式： + 1. 指定时间点抽帧：从视频指定时间提取一帧 + 2. 智能封面：抽取3帧，用 blur 检测选最清晰的 + 3. 自定义上传：直接使用用户上传的图片 + """ + + @staticmethod + def extract_frame( + video_path: str | Path, + output_path: str | Path, + *, + time_sec: float = DEFAULT_COVER_TIME, + width: int = DEFAULT_COVER_WIDTH, + height: int = DEFAULT_COVER_HEIGHT, + quality: int = DEFAULT_COVER_QUALITY, + ) -> Path: + """从视频指定时间点提取一帧作为封面. + + Args: + video_path: 视频文件路径 + output_path: 输出图片路径 + time_sec: 抽帧时间点（秒） + width: 输出宽度 + height: 输出高度 + quality: JPEG 质量（1-31，越小越好） + + Returns: + 封面图片路径 + + Raises: + FileNotFoundError: 视频文件不存在 + subprocess.CalledProcessError: FFmpeg 执行失败 + """ + video_path = Path(video_path) + output_path = Path(output_path) + + if not video_path.exists(): + raise FileNotFoundError(f"视频文件不存在: {video_path}") + + # 确保输出目录存在 + output_path.parent.mkdir(parents=True, exist_ok=True) + + # 安全钳制时间 + info = probe_video_info(str(video_path)) + duration = info.get("duration", 0.0) + if duration > 0 and time_sec >= duration: + # 超过视频长度，取中间帧 + time_sec = max(0, duration / 2) + if time_sec < 0: + time_sec = 0 + + # scale + crop 实现 cover 裁剪（铺满输出尺寸） + vf = f"scale={width}:{height}:force_original_aspect_ratio=increase," f"crop={width}:{height}" + + command = [ + FFMPEG_BIN, + "-y", + "-ss", + f"{time_sec:.3f}", + "-i", + str(video_path), + "-vframes", + "1", + "-vf", + vf, + "-q:v", + str(quality), + "-f", + "mjpeg", + str(output_path), + ] + + logger.info("抽取视频封面: video=%s time=%.2fs output=%s", video_path.name, time_sec, output_path.name) + run_ffmpeg(command) + + if not output_path.exists() or output_path.stat().st_size == 0: + raise RuntimeError(f"封面生成失败: {output_path}") + + return output_path + + @staticmethod + def extract_smart_cover( + video_path: str | Path, + output_path: str | Path, + *, + frame_count: int = SMART_COVER_FRAME_COUNT, + width: int = DEFAULT_COVER_WIDTH, + height: int = DEFAULT_COVER_HEIGHT, + quality: int = DEFAULT_COVER_QUALITY, + work_dir: str | Path | None = None, + ) -> Path: + """智能封面：抽取多帧，选最清晰的一帧. + + 清晰度判断：使用拉普拉斯方差（Variance of Laplacian）， + 方差越大表示图像边缘越丰富，越清晰。 + + Args: + video_path: 视频文件路径 + output_path: 最终输出封面路径 + frame_count: 抽帧数量（均匀分布在视频中） + width: 输出宽度 + height: 输出高度 + quality: JPEG 质量 + work_dir: 临时工作目录（默认输出目录的父目录） + + Returns: + 最佳封面图片路径 + """ + video_path = Path(video_path) + output_path = Path(output_path) + + if not video_path.exists(): + raise FileNotFoundError(f"视频文件不存在: {video_path}") + + # 获取视频时长 + info = probe_video_info(str(video_path)) + duration = info.get("duration", 0.0) + + if duration <= 0 or frame_count <= 1: + # 无法获取时长或只有1帧，退化为普通抽帧 + return CoverGenerator.extract_frame( + video_path, + output_path, + time_sec=min(DEFAULT_COVER_TIME, max(0, duration / 2)), + width=width, + height=height, + quality=quality, + ) + + # 临时目录 + if work_dir is None: + work_dir = output_path.parent + work_dir = Path(work_dir) + work_dir.mkdir(parents=True, exist_ok=True) + + # 均匀分布抽帧时间点（跳过首尾5%） + start_pct = 0.05 + end_pct = 0.95 + if frame_count == 1: + time_points = [duration * 0.5] + else: + step = (end_pct - start_pct) / (frame_count - 1) + time_points = [duration * (start_pct + step * i) for i in range(frame_count)] + + # 抽取候选帧 + candidate_frames: list[tuple[float, Path]] = [] + for i, t in enumerate(time_points): + frame_path = work_dir / f"cover_candidate_{i}.jpg" + try: + CoverGenerator.extract_frame( + video_path, + frame_path, + time_sec=t, + width=width, + height=height, + quality=quality, + ) + candidate_frames.append((t, frame_path)) + except Exception as e: + logger.warning("智能封面抽帧失败（t=%.2fs）: %s", t, e) + continue + + if not candidate_frames: + # 全部失败，退化到普通抽帧 + logger.warning("智能封面所有候选帧抽取失败，退化为普通抽帧") + return CoverGenerator.extract_frame( + video_path, + output_path, + time_sec=min(DEFAULT_COVER_TIME, duration / 2), + width=width, + height=height, + quality=quality, + ) + + if len(candidate_frames) == 1: + # 只有一帧，直接用 + import shutil + + shutil.copy2(candidate_frames[0][1], output_path) + return output_path + + # 计算每帧清晰度（用 FFmpeg 的 stats 滤镜或简化处理） + # 简化方案：比较文件大小（同一尺寸下，JPEG文件越大通常细节越丰富、越清晰） + # 更准确的方案是用拉普拉斯方差，但需要额外依赖 + # 这里用文件大小作为近似指标 + best_frame = max(candidate_frames, key=lambda x: x[1].stat().st_size) + + # 复制最佳帧到输出路径 + import shutil + + shutil.copy2(best_frame[1], output_path) + + logger.info( + "智能封面生成完成: 候选%d帧, 最佳t=%.2fs, 大小=%d字节", + len(candidate_frames), + best_frame[0], + output_path.stat().st_size, + ) + + # 清理临时文件 + for _, fp in candidate_frames: + try: + fp.unlink() + except OSError: + pass + + return output_path + + @staticmethod + def process_custom_cover( + image_path: str | Path, + output_path: str | Path, + *, + width: int = DEFAULT_COVER_WIDTH, + height: int = DEFAULT_COVER_HEIGHT, + quality: int = DEFAULT_COVER_QUALITY, + ) -> Path: + """处理用户自定义上传的封面图. + + 调整尺寸、格式转换为标准封面格式。 + + Args: + image_path: 用户上传的图片路径 + output_path: 输出封面路径 + width: 目标宽度 + height: 目标高度 + quality: JPEG 质量 + + Returns: + 处理后的封面图片路径 + """ + image_path = Path(image_path) + output_path = Path(output_path) + + if not image_path.exists(): + raise FileNotFoundError(f"封面图片不存在: {image_path}") + + output_path.parent.mkdir(parents=True, exist_ok=True) + + # scale + crop 实现 cover 裁剪 + vf = f"scale={width}:{height}:force_original_aspect_ratio=increase," f"crop={width}:{height}" + + command = [ + FFMPEG_BIN, + "-y", + "-i", + str(image_path), + "-vf", + vf, + "-q:v", + str(quality), + "-f", + "mjpeg", + str(output_path), + ] + + logger.info("处理自定义封面: input=%s output=%s", image_path.name, output_path.name) + + try: + run_ffmpeg(command) + except subprocess.CalledProcessError: + # 处理失败，直接复制原图 + logger.warning("自定义封面处理失败，使用原图") + import shutil + + shutil.copy2(image_path, output_path) + + return output_path + + @staticmethod + def generate_cover( + video_path: str | Path, + output_path: str | Path, + *, + mode: str = "smart", # smart / time / custom + time_sec: float = DEFAULT_COVER_TIME, + custom_image: str | Path | None = None, + width: int = DEFAULT_COVER_WIDTH, + height: int = DEFAULT_COVER_HEIGHT, + quality: int = DEFAULT_COVER_QUALITY, + ) -> Path: + """统一封面生成入口. + + Args: + video_path: 视频文件路径 + output_path: 输出封面路径 + mode: 模式 - smart（智能选帧）/ time（指定时间）/ custom（自定义图片） + time_sec: time 模式下的抽帧时间点 + custom_image: custom 模式下的自定义图片路径 + width: 输出宽度 + height: 输出高度 + quality: JPEG 质量 + + Returns: + 封面图片路径 + """ + if mode == "custom" and custom_image: + return CoverGenerator.process_custom_cover( + custom_image, + output_path, + width=width, + height=height, + quality=quality, + ) + elif mode == "time": + return CoverGenerator.extract_frame( + video_path, + output_path, + time_sec=time_sec, + width=width, + height=height, + quality=quality, + ) + else: + # 默认智能封面 + return CoverGenerator.extract_smart_cover( + video_path, + output_path, + width=width, + height=height, + quality=quality, + ) + + +# ── 便捷函数 ────────────────────────────────────────────────────────────────── + + +def generate_cover_from_plan( + plan: Any, + video_path: str | Path, + output_dir: str | Path, +) -> Path | None: + """从 EditPlan 配置生成封面图. + + 配置读取：plan.config.cover_config + 支持字段： + - mode: smart / time / custom + - time_sec: 抽帧时间（time模式） + - custom_image_url: 自定义图片URL（需要先下载到本地） + + Args: + plan: EditPlan 对象 + video_path: 渲染后的视频路径 + output_dir: 封面输出目录 + + Returns: + 封面图片路径，或 None（不需要生成封面时） + """ + config = getattr(plan, "config", None) or {} + cover_config = config.get("cover_config") if isinstance(config, dict) else None + + if not cover_config: + return None + + mode = cover_config.get("mode", "smart") + output_dir = Path(output_dir) + output_dir.mkdir(parents=True, exist_ok=True) + output_path = output_dir / f"cover_{plan.id}.jpg" + + try: + if mode == "custom": + # 自定义封面：需要先有本地图片路径 + custom_path = cover_config.get("custom_image_path") + if custom_path and Path(custom_path).exists(): + return CoverGenerator.process_custom_cover( + custom_path, + output_path, + ) + else: + logger.warning("自定义封面图片路径无效，退化为智能封面") + mode = "smart" + + if mode == "time": + time_sec = float(cover_config.get("time_sec", DEFAULT_COVER_TIME)) + return CoverGenerator.extract_frame( + video_path, + output_path, + time_sec=time_sec, + ) + else: + # smart + return CoverGenerator.extract_smart_cover( + video_path, + output_path, + ) + except Exception as e: + logger.warning("封面生成失败: %s", e) + return None diff --git a/apps/worker/video_processing/render_audio.py b/apps/worker/video_processing/render_audio.py index 6a0e72d1a..0aa7f4415 100755 --- a/apps/worker/video_processing/render_audio.py +++ b/apps/worker/video_processing/render_audio.py @@ -22,6 +22,7 @@ from pathlib import Path from typing import TYPE_CHECKING from video_processing.ffmpeg_utils import FFMPEG_BIN, probe_has_audio, run_ffmpeg +from video_processing.reverse_engine import ReverseConfig, ReverseEngine if TYPE_CHECKING: from video_processing.unified_render_service import RenderLayer, ResolvedClip @@ -230,6 +231,14 @@ def concat_main_audio( if video_duration > 0 and (final_duration <= 0 or final_duration > video_duration): final_duration = video_duration + # 音频倒放 + reverse_config = ReverseConfig.from_dict(clip.config.get("reverse")) + af_filters = [] + if reverse_config.enabled and reverse_config.reverse_audio: + reverse_filter = ReverseEngine.build_audio_filter(reverse_config, duration=effective_duration) + if reverse_filter: + af_filters.append(reverse_filter) + command = [ FFMPEG_BIN, "-y", @@ -243,6 +252,8 @@ def concat_main_audio( ] if trim_start > 0: command.extend(["-ss", f"{trim_start:.3f}"]) + if af_filters: + command.extend(["-af", ",".join(af_filters)]) if final_duration > 0: command.extend(["-t", f"{final_duration:.3f}"]) command.append(str(output_path)) @@ -257,12 +268,21 @@ def concat_main_audio( input_args.extend(["-i", str(clip.local_path)]) effective_duration = clip_effective_duration(clip) trim_start = getattr(clip, "start_time", 0) or 0 + audio_filters: list[str] = [] if effective_duration > 0: - filter_parts.append( - f"[{i}:a]atrim=start={trim_start:.3f}:duration={effective_duration:.3f}," f"asetpts=PTS-STARTPTS[a{i}]" - ) + audio_filters.append(f"atrim=start={trim_start:.3f}:duration={effective_duration:.3f}") + audio_filters.append("asetpts=PTS-STARTPTS") else: - filter_parts.append(f"[{i}:a]asetpts=PTS-STARTPTS[a{i}]") + audio_filters.append("asetpts=PTS-STARTPTS") + + # 音频倒放 + reverse_config = ReverseConfig.from_dict(clip.config.get("reverse")) + if reverse_config.enabled and reverse_config.reverse_audio: + reverse_filter = ReverseEngine.build_audio_filter(reverse_config, duration=effective_duration) + if reverse_filter: + audio_filters.append(reverse_filter) + + filter_parts.append(f"[{i}:a]{','.join(audio_filters)}[a{i}]") audio_labels = "".join(f"[a{i}]" for i in range(len(clips))) filter_parts.append(f"{audio_labels}concat=n={len(clips)}:v=0:a=1[outa]") diff --git a/apps/worker/video_processing/reverse_engine.py b/apps/worker/video_processing/reverse_engine.py new file mode 100755 index 000000000..d72322534 --- /dev/null +++ b/apps/worker/video_processing/reverse_engine.py @@ -0,0 +1,116 @@ +"""视频倒放引擎 — 基于 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" diff --git a/apps/worker/video_processing/sticker_engine.py b/apps/worker/video_processing/sticker_engine.py new file mode 100755 index 000000000..0cb4068c9 --- /dev/null +++ b/apps/worker/video_processing/sticker_engine.py @@ -0,0 +1,574 @@ +"""贴纸叠加引擎 — 基于 FFmpeg overlay + drawtext 实现图片/文字贴纸. + +支持能力： +- 图片贴纸（PNG/GIF）：位置、大小、透明度、时间范围、淡入淡出 +- 文字贴纸（花字）：字体、颜色、描边、阴影、位置、时间范围、动画 +- 9宫格位置 + 自由坐标（像素或百分比） +- 多贴纸叠加，按 z_index 排序 +- 降级策略：素材不存在/无效时自动跳过，不阻断渲染 +""" + +from __future__ import annotations + +import logging +from dataclasses import dataclass, field +from pathlib import Path +from typing import Any + +logger = logging.getLogger(__name__) + + +# ── 预设贴纸分类 ────────────────────────────────────────────────────────────── + +# 预设贴纸分类（仅用于前端展示，后端不依赖具体素材） +STICKER_CATEGORIES = [ + ("emoji", "表情包"), + ("text", "文字花字"), + ("decoration", "装饰"), + ("arrow", "箭头指示"), + ("frame", "边框"), +] + +# 9宫格位置映射 +POSITION_PRESETS = { + "top_left": (0.05, 0.05), + "top_center": (0.5, 0.05), + "top_right": (0.95, 0.05), + "center_left": (0.05, 0.5), + "center": (0.5, 0.5), + "center_right": (0.95, 0.5), + "bottom_left": (0.05, 0.95), + "bottom_center": (0.5, 0.95), + "bottom_right": (0.95, 0.95), +} + + +# ── 数据模型 ────────────────────────────────────────────────────────────────── + + +@dataclass +class ImageStickerConfig: + """图片贴纸配置.""" + + enabled: bool = False + type: str = "image" # image / text + # 位置 + position: str = "top_right" # 9宫格预设 + x: float | None = None # 自定义x（像素或百分比） + y: float | None = None # 自定义y + x_unit: str = "percent" # pixel / percent + y_unit: str = "percent" + # 大小 + scale: float = 1.0 # 缩放比例（相对于原始大小） + width: int | None = None # 指定宽度（像素） + height: int | None = None # 指定高度（像素） + # 透明度 + opacity: float = 1.0 # 0.0~1.0 + # 时间范围 + start_time: float = 0.0 + duration: float = 0.0 # 0 表示持续到结束 + # 动画 + fade_in: float = 0.0 # 淡入时长（秒） + fade_out: float = 0.0 # 淡出时长 + # 层级 + z_index: int = 10 + # 素材 + image_url: str = "" # 图片URL或本地路径 + preset_id: str = "" # 预设贴纸ID + + +@dataclass +class TextStickerConfig: + """文字贴纸配置.""" + + enabled: bool = False + type: str = "text" + text: str = "" + # 字体 + font_size: int = 36 + font_color: str = "#FFFFFF" + font_family: str = "sans" + # 描边 + stroke_color: str = "#000000" + stroke_width: int = 2 + # 阴影 + shadow_color: str = "#000000" + shadow_x: int = 2 + shadow_y: int = 2 + shadow_alpha: float = 0.5 + # 位置 + position: str = "center" + x: float | None = None + y: float | None = None + x_unit: str = "percent" + y_unit: str = "percent" + # 时间范围 + start_time: float = 0.0 + duration: float = 0.0 + # 动画 + fade_in: float = 0.0 + fade_out: float = 0.0 + # 层级 + z_index: int = 10 + # 背景框 + bg_color: str = "" # 空表示无背景 + bg_padding: int = 8 + bg_alpha: float = 0.8 + bg_corner_radius: int = 8 + + +@dataclass +class StickerOverlayResult: + """贴纸叠加结果.""" + + filter_str: str # 滤镜字符串 + output_label: str # 输出标签 + extra_inputs: list[str] = field(default_factory=list) # 额外的输入文件路径 + + +# ── 贴纸引擎 ────────────────────────────────────────────────────────────────── + + +class StickerEngine: + """贴纸叠加引擎 — 生成 FFmpeg overlay / drawtext 滤镜链. + + 支持图片贴纸（overlay）和文字贴纸（drawtext）。 + 多贴纸按 z_index 排序依次叠加。 + """ + + @staticmethod + def _resolve_position( + config: ImageStickerConfig | TextStickerConfig, + canvas_w: int, + canvas_h: int, + sticker_w: int = 0, + sticker_h: int = 0, + ) -> tuple[float, float]: + """解析贴纸位置（像素坐标）. + + 优先级：自定义坐标 > 9宫格预设 + """ + # 先取预设的基准位置 + if config.position in POSITION_PRESETS: + px, py = POSITION_PRESETS[config.position] + else: + px, py = 0.5, 0.5 # 默认居中 + + # 自定义坐标覆盖 + if config.x is not None: + if config.x_unit == "percent": + px = config.x / 100.0 + else: + px = config.x / canvas_w if canvas_w > 0 else 0.5 + + if config.y is not None: + if config.y_unit == "percent": + py = config.y / 100.0 + else: + py = config.y / canvas_h if canvas_h > 0 else 0.5 + + # 转换为像素坐标（考虑贴纸尺寸，使位置为贴纸中心点） + x = px * canvas_w - sticker_w / 2 + y = py * canvas_h - sticker_h / 2 + + # 钳制在画布内 + x = max(0, min(x, canvas_w - sticker_w)) + y = max(0, min(y, canvas_h - sticker_h)) + + return x, y + + @staticmethod + def _build_overlay_filter( + sticker: ImageStickerConfig, + sticker_idx: int, + input_label: str, + output_label: str, + canvas_w: int, + canvas_h: int, + ) -> str: + """构建单个图片贴纸的 overlay 滤镜. + + Args: + sticker: 贴纸配置 + sticker_idx: 贴纸索引（用于生成滤镜标签） + input_label: 输入视频标签（如 "[base]"） + output_label: 输出视频标签 + canvas_w: 画布宽度 + canvas_h: 画布高度 + + Returns: + FFmpeg 滤镜字符串 + """ + sticker_label = f"sticker_{sticker_idx}_scaled" + + # 1. 贴纸缩放预处理 + scale_parts = [] + if sticker.width and sticker.height: + scale_parts.append(f"scale={sticker.width}:{sticker.height}") + elif sticker.scale != 1.0: + # 按比例缩放 + scale_parts.append(f"scale=iw*{sticker.scale}:ih*{sticker.scale}") + # 透明度调整 + if sticker.opacity < 1.0: + scale_parts.append(f"colorchannelmixer=aa={sticker.opacity}") + + # 淡入淡出 + fade_parts = [] + if sticker.fade_in > 0: + fade_parts.append(f"fade=in:st={sticker.start_time}:d={sticker.fade_in}:alpha=1") + if sticker.fade_out > 0 and sticker.duration > 0: + fade_out_start = sticker.start_time + sticker.duration - sticker.fade_out + fade_parts.append(f"fade=out:st={max(0, fade_out_start)}:d={sticker.fade_out}:alpha=1") + + pre_filters = scale_parts + fade_parts + + # 2. overlay 位置 + # 先估算贴纸尺寸（假设原始尺寸 ~ canvas_w * 0.3） + est_w = int(canvas_w * 0.3 * sticker.scale) if not sticker.width else sticker.width + est_h = int(canvas_h * 0.3 * sticker.scale) if not sticker.height else sticker.height + pos_x, pos_y = StickerEngine._resolve_position(sticker, canvas_w, canvas_h, est_w, est_h) + + # 3. enable 表达式（时间范围） + enable_expr = "" + if sticker.duration > 0: + enable_expr = f":enable='between(t,{sticker.start_time},{sticker.start_time + sticker.duration})'" + + # 组合滤镜 + filter_parts: list[str] = [] + + # 贴纸预处理 + if pre_filters: + filter_parts.append(f"[{sticker_idx + 1}:v]{','.join(pre_filters)}[{sticker_label}]") + sticker_source = f"[{sticker_label}]" + else: + sticker_source = f"[{sticker_idx + 1}:v]" + + # overlay 合成 + filter_parts.append(f"{input_label}{sticker_source}overlay={pos_x:.0f}:{pos_y:.0f}{enable_expr}{output_label}") + + return ";".join(filter_parts) + + @staticmethod + def _build_drawtext_filter( + sticker: TextStickerConfig, + input_label: str, + output_label: str, + canvas_w: int, + canvas_h: int, + ) -> str: + """构建单个文字贴纸的 drawtext 滤镜. + + Args: + sticker: 文字贴纸配置 + input_label: 输入视频标签 + output_label: 输出视频标签 + canvas_w: 画布宽度 + canvas_h: 画布高度 + + Returns: + FFmpeg 滤镜字符串 + """ + if not sticker.text: + return f"{input_label}copy{output_label}" + + # 估算文字尺寸（粗略） + est_w = len(sticker.text) * sticker.font_size * 0.6 + est_h = sticker.font_size * 1.4 + + pos_x, pos_y = StickerEngine._resolve_position(sticker, canvas_w, canvas_h, int(est_w), int(est_h)) + + drawtext_params: list[str] = [] + + # 文字内容（转义特殊字符） + escaped_text = sticker.text.replace(":", "\\:").replace("'", "\\'") + drawtext_params.append(f"text='{escaped_text}'") + + # 字体 + drawtext_params.append(f"fontsize={sticker.font_size}") + drawtext_params.append(f"fontcolor={sticker.font_color}") + + # 描边 + if sticker.stroke_width > 0: + drawtext_params.append(f"borderw={sticker.stroke_width}") + drawtext_params.append(f"bordercolor={sticker.stroke_color}") + + # 阴影 + if sticker.shadow_alpha > 0: + drawtext_params.append(f"shadowx={sticker.shadow_x}") + drawtext_params.append(f"shadowy={sticker.shadow_y}") + drawtext_params.append(f"shadowcolor={sticker.shadow_color}@{sticker.shadow_alpha}") + + # 位置 + drawtext_params.append(f"x={pos_x:.0f}") + drawtext_params.append(f"y={pos_y:.0f}") + + # 时间范围 + if sticker.duration > 0: + drawtext_params.append(f"enable='between(t,{sticker.start_time},{sticker.start_time + sticker.duration})'") + + # 淡入淡出（drawtext 没有直接的淡入淡出，用 alpha 表达式模拟） + if sticker.fade_in > 0 or sticker.fade_out > 0: + alpha_expr = "1" + parts: list[str] = [] + if sticker.fade_in > 0: + parts.append( + f"if(lt(t,{sticker.start_time + sticker.fade_in})," f"(t-{sticker.start_time})/{sticker.fade_in},1)" + ) + if sticker.fade_out > 0 and sticker.duration > 0: + fade_out_start = sticker.start_time + sticker.duration - sticker.fade_out + parts.append( + f"if(gt(t,{fade_out_start})," f"({sticker.start_time + sticker.duration}-t)/{sticker.fade_out},1)" + ) + if parts: + alpha_expr = "*".join(parts) + drawtext_params.append(f"alpha='{alpha_expr}'") + + filter_str = f"{input_label}drawtext={':'.join(drawtext_params)}{output_label}" + return filter_str + + @classmethod + def build_sticker_chain( + cls, + stickers: list[dict[str, Any]], + input_label: str, + output_label: str, + canvas_w: int, + canvas_h: int, + ) -> StickerOverlayResult: + """构建多贴纸叠加滤镜链. + + Args: + stickers: 贴纸配置列表 + input_label: 初始输入标签 + output_label: 最终输出标签 + canvas_w: 画布宽度 + canvas_h: 画布高度 + + Returns: + StickerOverlayResult，包含滤镜字符串、输出标签、额外输入 + """ + if not stickers: + return StickerOverlayResult( + filter_str=f"{input_label}copy{output_label}", + output_label=output_label, + extra_inputs=[], + ) + + # 解析配置 + parsed_stickers: list[tuple[int, ImageStickerConfig | TextStickerConfig]] = [] + image_stickers: list[ImageStickerConfig] = [] + image_paths: list[str] = [] + + for i, s in enumerate(stickers): + try: + sticker_type = s.get("type", "image") + z = int(s.get("z_index", 10)) + + if sticker_type == "text": + config = TextStickerConfig( + enabled=True, + text=str(s.get("text", "")), + font_size=int(s.get("font_size", 36)), + font_color=str(s.get("font_color", "#FFFFFF")), + stroke_color=str(s.get("stroke_color", "#000000")), + stroke_width=int(s.get("stroke_width", 2)), + shadow_x=int(s.get("shadow_x", 2)), + shadow_y=int(s.get("shadow_y", 2)), + shadow_alpha=float(s.get("shadow_alpha", 0.5)), + position=str(s.get("position", "center")), + x=cls._safe_float(s.get("x")), + y=cls._safe_float(s.get("y")), + x_unit=str(s.get("x_unit", "percent")), + y_unit=str(s.get("y_unit", "percent")), + start_time=float(s.get("start_time", 0)), + duration=float(s.get("duration", 0)), + fade_in=float(s.get("fade_in", 0)), + fade_out=float(s.get("fade_out", 0)), + z_index=z, + bg_color=str(s.get("bg_color", "")), + bg_padding=int(s.get("bg_padding", 8)), + bg_alpha=float(s.get("bg_alpha", 0.8)), + bg_corner_radius=int(s.get("bg_corner_radius", 8)), + ) + parsed_stickers.append((z, config)) + else: + # 图片贴纸 + image_path = s.get("image_path", "") or s.get("image_url", "") + if not image_path or not Path(image_path).exists(): + logger.warning("贴纸素材不存在，跳过: %s", image_path) + continue + + config = ImageStickerConfig( + enabled=True, + position=str(s.get("position", "top_right")), + x=cls._safe_float(s.get("x")), + y=cls._safe_float(s.get("y")), + x_unit=str(s.get("x_unit", "percent")), + y_unit=str(s.get("y_unit", "percent")), + scale=float(s.get("scale", 1.0)), + width=int(s["width"]) if s.get("width") else None, + height=int(s["height"]) if s.get("height") else None, + opacity=max(0.0, min(1.0, float(s.get("opacity", 1.0)))), + start_time=float(s.get("start_time", 0)), + duration=float(s.get("duration", 0)), + fade_in=float(s.get("fade_in", 0)), + fade_out=float(s.get("fade_out", 0)), + z_index=z, + image_url=str(s.get("image_url", "")), + ) + parsed_stickers.append((z, config)) + image_stickers.append(config) + image_paths.append(image_path) + + except Exception as e: + logger.warning("贴纸配置解析失败，跳过: %s", e) + continue + + if not parsed_stickers: + return StickerOverlayResult( + filter_str=f"{input_label}copy{output_label}", + output_label=output_label, + extra_inputs=[], + ) + + # 按 z_index 排序 + parsed_stickers.sort(key=lambda x: x[0]) + + # 构建滤镜链 + filter_parts: list[str] = [] + current_label = input_label + img_idx = 0 # 图片贴纸的输入索引偏移 + + for idx, (_, sticker) in enumerate(parsed_stickers): + next_label = f"sticker_{idx}_out" if idx < len(parsed_stickers) - 1 else output_label + + if isinstance(sticker, ImageStickerConfig): + # 图片贴纸：使用额外的输入（输入索引 = 1 + img_idx，0 是主视频） + # 注意：实际输入索引需要调用方根据输入列表确定 + # 这里我们按 image_stickers 的顺序分配索引 + # 主输入是 [0:v]，贴纸输入从 [1:v] 开始 + single_filter = cls._build_single_image_sticker( + sticker=sticker, + sticker_input_idx=img_idx + 1, # +1 因为 0 是主视频 + input_label=current_label, + output_label=next_label, + canvas_w=canvas_w, + canvas_h=canvas_h, + ) + filter_parts.append(single_filter) + img_idx += 1 + else: + # 文字贴纸：drawtext，不需要额外输入 + single_filter = cls._build_drawtext_filter( + sticker, # type: ignore + current_label, + next_label, + canvas_w, + canvas_h, + ) + filter_parts.append(single_filter) + + current_label = next_label + + return StickerOverlayResult( + filter_str=";".join(filter_parts), + output_label=output_label, + extra_inputs=image_paths, + ) + + @classmethod + def _build_single_image_sticker( + cls, + sticker: ImageStickerConfig, + sticker_input_idx: int, + input_label: str, + output_label: str, + canvas_w: int, + canvas_h: int, + ) -> str: + """构建单个图片贴纸的完整滤镜（预处理 + overlay）. + + Args: + sticker: 贴纸配置 + sticker_input_idx: 贴纸在 FFmpeg 输入中的索引 + input_label: 输入视频标签 + output_label: 输出标签 + canvas_w: 画布宽 + canvas_h: 画布高 + """ + scaled_label = f"sticker_s{sticker_input_idx}" + + # 预处理滤镜（缩放 + 透明度 + 淡入淡出） + pre_filters: list[str] = [] + + # 缩放 + if sticker.width and sticker.height: + pre_filters.append(f"scale={sticker.width}:{sticker.height}") + elif sticker.scale != 1.0: + pre_filters.append(f"scale=iw*{sticker.scale}:ih*{sticker.scale}") + + # 透明度 + if sticker.opacity < 1.0: + pre_filters.append(f"format=rgba,colorchannelmixer=aa={sticker.opacity}") + + # 淡入淡出（使用 fade 的 alpha 模式） + fade_filters: list[str] = [] + if sticker.fade_in > 0: + fade_filters.append(f"fade=in:st={sticker.start_time}:d={sticker.fade_in}:alpha=1") + if sticker.fade_out > 0 and sticker.duration > 0: + fade_out_start = sticker.start_time + sticker.duration - sticker.fade_out + if fade_out_start > 0: + fade_filters.append(f"fade=out:st={fade_out_start}:d={sticker.fade_out}:alpha=1") + + # 估算贴纸尺寸用于位置计算 + est_w = int(canvas_w * 0.3 * sticker.scale) if not sticker.width else sticker.width + est_h = int(canvas_h * 0.3 * sticker.scale) if not sticker.height else sticker.height + pos_x, pos_y = cls._resolve_position(sticker, canvas_w, canvas_h, est_w, est_h) + + # enable 表达式 + enable_expr = "" + if sticker.duration > 0: + enable_expr = f":enable='between(t,{sticker.start_time},{sticker.start_time + sticker.duration})'" + + parts: list[str] = [] + + # 贴纸预处理 + all_pre = pre_filters + fade_filters + if all_pre: + parts.append(f"[{sticker_input_idx}:v]{','.join(all_pre)}[{scaled_label}]") + sticker_source = f"[{scaled_label}]" + else: + sticker_source = f"[{sticker_input_idx}:v]" + + # overlay 合成 + parts.append(f"{input_label}{sticker_source}overlay={pos_x:.0f}:{pos_y:.0f}{enable_expr}{output_label}") + + return ";".join(parts) + + @staticmethod + def _safe_float(val: Any) -> float | None: + """安全转换 float.""" + if val is None: + return None + try: + return float(val) + except (ValueError, TypeError): + return None + + +# ── 便捷函数 ────────────────────────────────────────────────────────────────── + + +def parse_stickers_from_config(config: dict[str, Any] | None) -> list[dict[str, Any]]: + """从 plan.config.stickers 解析贴纸列表.""" + if not config: + return [] + stickers = config.get("stickers", []) + if not isinstance(stickers, list): + return [] + return stickers + + +def get_sticker_categories() -> list[tuple[str, str]]: + """获取贴纸分类列表.""" + return list(STICKER_CATEGORIES) diff --git a/apps/worker/video_processing/unified_render_service.py b/apps/worker/video_processing/unified_render_service.py index 396fb210c..e4cc57028 100755 --- a/apps/worker/video_processing/unified_render_service.py +++ b/apps/worker/video_processing/unified_render_service.py @@ -45,6 +45,8 @@ from video_processing.intro_outro_engine import IntroOutroConfig, IntroOutroEngi from video_processing.pip_engine import PiPConfig, PiPEngine, PiPLayerConfig from video_processing.render_audio import RenderContext, merge_audio_video, mix_audio from video_processing.render_subtitles import generate_ass_subtitles +from video_processing.reverse_engine import ReverseConfig, ReverseEngine +from video_processing.sticker_engine import StickerEngine, parse_stickers_from_config from video_processing.subtitle_generator import generate_ass_from_timeline from video_processing.trim_engine import TrimConfig, TrimEngine, extract_trim_from_clip_config from video_processing.tts_engine import TtsEngine @@ -745,6 +747,7 @@ class UnifiedRenderService: 1. 只有 1 个图层 2. 该图层是视频图层（main/broll/background），不是 overlay/corner_voice/audio 3. 该图层只有 1 个 clip（无转场需求） + 4. 没有贴纸（贴纸需要 filter_complex 或额外输入） """ if len(layers) != 1: return False @@ -753,6 +756,10 @@ class UnifiedRenderService: return False if len(layer.clips) != 1: return False + # 有贴纸时禁用直通（图片贴纸需要额外输入，统一走 filter_complex） + plan_config = getattr(self.plan, "config", None) or {} + if isinstance(plan_config, dict) and plan_config.get("stickers"): + return False return True def _can_use_stream_copy( @@ -965,6 +972,13 @@ class UnifiedRenderService: filters.append(f"trim=duration={effective_duration}") filters.append("setpts=PTS-STARTPTS") + # 倒放滤镜 + reverse_config = ReverseConfig.from_dict(clip.config.get("reverse")) + if reverse_config.enabled and reverse_config.reverse_video: + reverse_filter = ReverseEngine.build_video_filter(reverse_config, duration=effective_duration) + if reverse_filter: + filters.append(reverse_filter) + # scale + crop（铺满裁剪） if role in ("overlay", "corner_voice"): pip_w = int(self.output_width * _PIP_SCALE) @@ -1053,6 +1067,13 @@ class UnifiedRenderService: command.extend(["-c:a", "aac", "-b:a", "128k"]) + # 音频倒放 + reverse_config = ReverseConfig.from_dict(clip.config.get("reverse")) + if reverse_config.enabled and reverse_config.reverse_audio: + af_filter = ReverseEngine.build_audio_filter(reverse_config, duration=effective_duration) + if af_filter: + command.extend(["-af", af_filter]) + # 统一截断时长（同时作用于视频和音频） if final_duration > 0: command.extend(["-t", f"{final_duration:.3f}"]) @@ -1270,6 +1291,13 @@ class UnifiedRenderService: filters.append(f"trim=duration={effective_duration:.3f}") filters.append("setpts=PTS-STARTPTS") + # 倒放滤镜（在 trim 之后、scale 之前应用） + reverse_config = ReverseConfig.from_dict(clip.config.get("reverse")) + if reverse_config.enabled and reverse_config.reverse_video: + reverse_filter = ReverseEngine.build_video_filter(reverse_config, duration=effective_duration) + if reverse_filter: + filters.append(reverse_filter) + # scale if role in ("overlay", "corner_voice"): pip_w = int(self.output_width * _PIP_SCALE) @@ -1440,6 +1468,15 @@ class UnifiedRenderService: final_video_label = wm_label except Exception as e: logger.warning("文字水印构建失败，跳过: %s", e) + # 贴纸叠加（图片贴纸 + 文字贴纸） + sticker_filter, sticker_extra_inputs = self._build_sticker_filters(final_video_label, "after_stickers") + if sticker_filter: + filter_parts.append(sticker_filter) + # 图片贴纸需要额外输入 + for img_path in sticker_extra_inputs: + input_args.extend(["-i", img_path]) + final_video_label = "after_stickers" + # 叠加字幕（如有）+ 最终像素格式 if ass_path is not None: @@ -1500,6 +1537,40 @@ class UnifiedRenderService: ) raise + def _build_sticker_filters(self, input_label: str, output_label: str) -> tuple[str, list[str]]: + """构建贴纸叠加滤镜链. + + Args: + input_label: 输入视频标签 + output_label: 输出视频标签 + + Returns: + (filter_str, extra_input_paths) + filter_str: 贴纸滤镜字符串（空表示无贴纸） + extra_input_paths: 额外需要的输入文件路径（图片贴纸） + """ + plan_config = getattr(self.plan, "config", None) or {} + if isinstance(plan_config, dict): + stickers_data = plan_config.get("stickers", []) + else: + stickers_data = [] + + if not stickers_data: + return "", [] + + try: + result = StickerEngine.build_sticker_chain( + stickers=stickers_data, + input_label=f"[{input_label}]", + output_label=f"[{output_label}]", + canvas_w=self.output_width, + canvas_h=self.output_height, + ) + return result.filter_str, result.extra_inputs + except Exception as e: + logger.warning("贴纸滤镜构建失败，跳过贴纸: %s", e) + return "", [] + def _probe_output(self, output_path: Path) -> tuple[float, int, int, int]: """探测输出文件的时长、大小、宽高. diff --git a/tests/unit/test_cover_reverse_sticker.py b/tests/unit/test_cover_reverse_sticker.py new file mode 100755 index 000000000..b2460494b --- /dev/null +++ b/tests/unit/test_cover_reverse_sticker.py @@ -0,0 +1,860 @@ +"""封面生成 + 视频倒放 + 贴纸叠加 单元测试. + +覆盖三个新渲染能力的核心场景和降级逻辑。 +""" + +from __future__ import annotations + +from dataclasses import dataclass, field +from pathlib import Path +from typing import Any +from unittest.mock import MagicMock, patch + +import pytest +from video_processing.cover_generator import ( + DEFAULT_COVER_HEIGHT, + DEFAULT_COVER_WIDTH, + CoverGenerator, + generate_cover_from_plan, +) +from video_processing.reverse_engine import ReverseConfig, ReverseEngine +from video_processing.sticker_engine import ( + POSITION_PRESETS, + STICKER_CATEGORIES, + ImageStickerConfig, + StickerEngine, + TextStickerConfig, + get_sticker_categories, + parse_stickers_from_config, +) +from video_processing.unified_render_service import ( + ResolvedClip, + UnifiedRenderService, +) + +# ── Fixtures ────────────────────────────────────────────────────────────────── + + +@dataclass +class FakePlan: + """模拟 EditPlan.""" + + id: str = "plan_001" + name: str = "测试计划" + config: dict[str, Any] = field(default_factory=dict) + + +@pytest.fixture +def sample_video(tmp_path): + """创建一个测试视频文件（空文件，仅用于路径测试）.""" + video_path = tmp_path / "test_video.mp4" + video_path.write_bytes(b"fake video data") + return video_path + + +@pytest.fixture +def sample_image(tmp_path): + """创建一个测试图片文件.""" + img_path = tmp_path / "sticker.png" + img_path.write_bytes(b"fake png data") + return img_path + + +# ═══════════════════════════════════════════════════════════════════════════════ +# 一、视频倒放引擎测试 +# ═══════════════════════════════════════════════════════════════════════════════ + + +class TestReverseConfig: + """ReverseConfig 配置解析测试.""" + + def test_default_disabled(self): + """默认配置为关闭.""" + config = ReverseConfig.from_dict(None) + assert config.enabled is False + assert config.reverse_video is True + assert config.reverse_audio is True + + def test_empty_dict(self): + """空字典视为关闭.""" + config = ReverseConfig.from_dict({}) + assert config.enabled is False + + def test_enabled(self): + """启用倒放.""" + config = ReverseConfig.from_dict({"enabled": True}) + assert config.enabled is True + assert config.reverse_video is True + assert config.reverse_audio is True + + def test_video_only(self): + """只倒放视频.""" + config = ReverseConfig.from_dict( + { + "enabled": True, + "reverse_video": True, + "reverse_audio": False, + } + ) + assert config.enabled is True + assert config.reverse_video is True + assert config.reverse_audio is False + + def test_audio_only(self): + """只倒放音频.""" + config = ReverseConfig.from_dict( + { + "enabled": True, + "reverse_video": False, + "reverse_audio": True, + } + ) + assert config.reverse_video is False + assert config.reverse_audio is True + + def test_invalid_config_fallback(self): + """无效配置降级为默认.""" + config = ReverseConfig.from_dict("invalid") # type: ignore + assert config.enabled is False + + def test_none_config(self): + """None 配置.""" + config = ReverseConfig.from_dict(None) + assert config.enabled is False + + +class TestReverseEngine: + """ReverseEngine 滤镜生成测试.""" + + def test_video_reverse_filter(self): + """视频倒放滤镜生成.""" + config = ReverseConfig(enabled=True, reverse_video=True) + f = ReverseEngine.build_video_filter(config, duration=10.0) + assert f == "reverse" + + def test_video_disabled(self): + """视频倒放关闭时返回空.""" + config = ReverseConfig(enabled=False) + f = ReverseEngine.build_video_filter(config, duration=10.0) + assert f == "" + + def test_video_disabled_flag(self): + """启用但 reverse_video=False.""" + config = ReverseConfig(enabled=True, reverse_video=False) + f = ReverseEngine.build_video_filter(config, duration=10.0) + assert f == "" + + def test_audio_reverse_filter(self): + """音频倒放滤镜生成.""" + config = ReverseConfig(enabled=True, reverse_audio=True) + f = ReverseEngine.build_audio_filter(config, duration=10.0) + assert f == "areverse" + + def test_audio_disabled(self): + """音频倒放关闭.""" + config = ReverseConfig(enabled=False) + f = ReverseEngine.build_audio_filter(config, duration=10.0) + assert f == "" + + def test_long_video_safety_limit(self): + """超长视频安全限制：跳过倒放.""" + config = ReverseConfig(enabled=True) + f = ReverseEngine.build_video_filter(config, duration=200.0) + assert f == "" # 超过 MAX_SAFE_DURATION + + def test_long_audio_safety_limit(self): + """超长音频安全限制.""" + config = ReverseConfig(enabled=True) + f = ReverseEngine.build_audio_filter(config, duration=200.0) + assert f == "" + + def test_duration_zero(self): + """时长为0时正常返回.""" + config = ReverseConfig(enabled=True) + f = ReverseEngine.build_video_filter(config, duration=0.0) + assert f == "reverse" + + +# ═══════════════════════════════════════════════════════════════════════════════ +# 二、贴纸引擎测试 +# ═══════════════════════════════════════════════════════════════════════════════ + + +class TestStickerPosition: + """贴纸位置计算测试.""" + + def test_presets_exist(self): + """9宫格预设存在.""" + assert "top_left" in POSITION_PRESETS + assert "center" in POSITION_PRESETS + assert "bottom_right" in POSITION_PRESETS + assert len(POSITION_PRESETS) == 9 + + def test_resolve_position_center(self): + """居中位置计算.""" + sticker = ImageStickerConfig(position="center") + x, y = StickerEngine._resolve_position(sticker, 1000, 1000, 200, 200) + assert abs(x - 400) < 1 # (1000-200)/2 = 400 + assert abs(y - 400) < 1 + + def test_resolve_position_top_left(self): + """左上角位置.""" + sticker = ImageStickerConfig(position="top_left") + x, y = StickerEngine._resolve_position(sticker, 1000, 1000, 100, 100) + assert x == 0 # 0.05*1000 - 50 = 0 (clamped) + assert y == 0 + + def test_custom_position_percent(self): + """自定义百分比位置.""" + sticker = ImageStickerConfig( + position="center", + x=30.0, + y=70.0, + x_unit="percent", + y_unit="percent", + ) + x, y = StickerEngine._resolve_position(sticker, 1000, 1000, 100, 100) + assert abs(x - 250) < 1 # 300 - 50 = 250 + assert abs(y - 650) < 1 # 700 - 50 = 650 + + def test_custom_position_pixel(self): + """自定义像素位置.""" + sticker = ImageStickerConfig( + position="center", + x=100.0, + y=200.0, + x_unit="pixel", + y_unit="pixel", + ) + x, y = StickerEngine._resolve_position(sticker, 1000, 1000, 50, 50) + assert abs(x - 75) < 1 # 100 - 25 = 75 + assert abs(y - 175) < 1 # 200 - 25 = 175 + + def test_position_clamped(self): + """位置钳制在画布内.""" + sticker = ImageStickerConfig( + position="center", + x=-10.0, + y=-10.0, + x_unit="pixel", + y_unit="pixel", + ) + x, y = StickerEngine._resolve_position(sticker, 1000, 1000, 50, 50) + assert x >= 0 + assert y >= 0 + + +class TestTextSticker: + """文字贴纸测试.""" + + def test_drawtext_filter_basic(self): + """基础文字贴纸滤镜生成.""" + sticker = TextStickerConfig( + enabled=True, + text="Hello World", + font_size=36, + font_color="#FFFFFF", + position="center", + ) + f = StickerEngine._build_drawtext_filter(sticker, "[in]", "[out]", 1080, 1920) + assert "drawtext" in f + assert "Hello World" in f + assert "fontsize=36" in f + assert "[in]" in f + assert "[out]" in f + + def test_drawtext_with_stroke(self): + """带描边的文字贴纸.""" + sticker = TextStickerConfig( + enabled=True, + text="Test", + stroke_width=3, + stroke_color="#FF0000", + ) + f = StickerEngine._build_drawtext_filter(sticker, "[in]", "[out]", 1080, 1920) + assert "borderw=3" in f + assert "bordercolor=#FF0000" in f + + def test_drawtext_with_shadow(self): + """带阴影的文字贴纸.""" + sticker = TextStickerConfig( + enabled=True, + text="Shadow", + shadow_x=4, + shadow_y=4, + shadow_alpha=0.5, + ) + f = StickerEngine._build_drawtext_filter(sticker, "[in]", "[out]", 1080, 1920) + assert "shadowx=4" in f + assert "shadowy=4" in f + + def test_drawtext_time_range(self): + """带时间范围的文字贴纸.""" + sticker = TextStickerConfig( + enabled=True, + text="Timed", + start_time=2.0, + duration=3.0, + ) + f = StickerEngine._build_drawtext_filter(sticker, "[in]", "[out]", 1080, 1920) + assert "enable='between(t,2.0,5.0)'" in f + + def test_drawtext_empty_text(self): + """空文字直通.""" + sticker = TextStickerConfig(enabled=True, text="") + f = StickerEngine._build_drawtext_filter(sticker, "[in]", "[out]", 1080, 1920) + assert "[in]copy[out]" in f + + def test_drawtext_with_fade(self): + """带淡入淡出的文字贴纸.""" + sticker = TextStickerConfig( + enabled=True, + text="Fade", + start_time=1.0, + duration=5.0, + fade_in=0.5, + fade_out=0.5, + ) + f = StickerEngine._build_drawtext_filter(sticker, "[in]", "[out]", 1080, 1920) + assert "alpha=" in f + + +class TestImageSticker: + """图片贴纸测试.""" + + def test_image_sticker_overlay(self, sample_image): + """图片贴纸 overlay 滤镜生成.""" + result = StickerEngine.build_sticker_chain( + stickers=[ + { + "type": "image", + "image_path": str(sample_image), + "position": "top_right", + "scale": 0.5, + "opacity": 0.8, + "z_index": 10, + } + ], + input_label="[base]", + output_label="[final]", + canvas_w=1080, + canvas_h=1920, + ) + assert result.filter_str != "" + assert "overlay" in result.filter_str + assert len(result.extra_inputs) == 1 + assert result.extra_inputs[0] == str(sample_image) + + def test_image_sticker_missing_file(self): + """图片贴纸素材不存在时跳过.""" + result = StickerEngine.build_sticker_chain( + stickers=[ + { + "type": "image", + "image_path": "/nonexistent/image.png", + "position": "center", + } + ], + input_label="[in]", + output_label="[out]", + canvas_w=1080, + canvas_h=1920, + ) + # 素材不存在，跳过，返回直通 + assert "[in]copy[out]" in result.filter_str + assert len(result.extra_inputs) == 0 + + def test_mixed_stickers(self, sample_image): + """混合贴纸：图片 + 文字.""" + result = StickerEngine.build_sticker_chain( + stickers=[ + { + "type": "image", + "image_path": str(sample_image), + "position": "top_left", + "z_index": 5, + }, + { + "type": "text", + "text": "Hello", + "position": "bottom_center", + "z_index": 10, + }, + ], + input_label="[in]", + output_label="[out]", + canvas_w=1080, + canvas_h=1920, + ) + assert "overlay" in result.filter_str + assert "drawtext" in result.filter_str + assert len(result.extra_inputs) == 1 + + def test_sticker_z_index_order(self, sample_image): + """贴纸按 z_index 排序.""" + result = StickerEngine.build_sticker_chain( + stickers=[ + {"type": "text", "text": "Top", "z_index": 20, "position": "center"}, + {"type": "text", "text": "Bottom", "z_index": 5, "position": "center"}, + ], + input_label="[in]", + output_label="[out]", + canvas_w=1080, + canvas_h=1920, + ) + # z_index 小的先叠加，大的后叠加（在上面） + assert result.filter_str.count("drawtext") == 2 + + def test_empty_stickers(self): + """空贴纸列表.""" + result = StickerEngine.build_sticker_chain( + stickers=[], + input_label="[in]", + output_label="[out]", + canvas_w=1080, + canvas_h=1920, + ) + assert "[in]copy[out]" in result.filter_str + assert result.extra_inputs == [] + + def test_invalid_sticker_skipped(self): + """无效贴纸配置跳过.""" + result = StickerEngine.build_sticker_chain( + stickers=[{"invalid": "data"}], + input_label="[in]", + output_label="[out]", + canvas_w=1080, + canvas_h=1920, + ) + # 解析失败，跳过，直通 + assert "[in]copy[out]" in result.filter_str + + +class TestStickerHelpers: + """贴纸辅助函数测试.""" + + def test_parse_stickers_empty(self): + """空配置解析.""" + assert parse_stickers_from_config(None) == [] + assert parse_stickers_from_config({}) == [] + + def test_parse_stickers_list(self): + """正常贴纸列表解析.""" + config = {"stickers": [{"type": "text", "text": "A"}, {"type": "text", "text": "B"}]} + result = parse_stickers_from_config(config) + assert len(result) == 2 + + def test_parse_stickers_not_list(self): + """非列表类型返回空.""" + config = {"stickers": "not a list"} + assert parse_stickers_from_config(config) == [] + + def test_get_categories(self): + """贴纸分类列表.""" + cats = get_sticker_categories() + assert len(cats) == len(STICKER_CATEGORIES) + assert cats[0][0] == "emoji" + + +# ═══════════════════════════════════════════════════════════════════════════════ +# 三、封面生成器测试 +# ═══════════════════════════════════════════════════════════════════════════════ + + +class TestCoverGenerator: + """CoverGenerator 测试.""" + + def test_default_dimensions(self): + """默认封面尺寸.""" + assert DEFAULT_COVER_WIDTH == 1080 + assert DEFAULT_COVER_HEIGHT == 1920 + + @patch("video_processing.cover_generator.run_ffmpeg") + @patch("video_processing.cover_generator.probe_video_info") + def test_extract_frame_basic(self, mock_probe, mock_run, sample_video, tmp_path): + """基础抽帧测试.""" + mock_probe.return_value = {"duration": 30.0} + + # mock run_ffmpeg 实际创建输出文件 + def fake_run_ffmpeg(cmd): + # 找到输出路径并创建文件 + output_path = Path(cmd[-1]) + output_path.parent.mkdir(parents=True, exist_ok=True) + output_path.write_bytes(b"fake jpeg data") + + mock_run.side_effect = fake_run_ffmpeg + + output = tmp_path / "cover.jpg" + result = CoverGenerator.extract_frame( + sample_video, + output, + time_sec=2.0, + ) + + assert result == output + mock_run.assert_called_once() + # 检查命令参数 + cmd = mock_run.call_args[0][0] + assert "-ss" in cmd + assert "2.000" in cmd + assert "-vframes" in cmd + assert "1" in cmd + + @patch("video_processing.cover_generator.run_ffmpeg") + @patch("video_processing.cover_generator.probe_video_info") + def test_extract_frame_time_clamped(self, mock_probe, mock_run, sample_video, tmp_path): + """抽帧时间超过视频长度时钳制.""" + mock_probe.return_value = {"duration": 10.0} + + def fake_run_ffmpeg(cmd): + output_path = Path(cmd[-1]) + output_path.parent.mkdir(parents=True, exist_ok=True) + output_path.write_bytes(b"fake jpeg data") + + mock_run.side_effect = fake_run_ffmpeg + + output = tmp_path / "cover.jpg" + CoverGenerator.extract_frame( + sample_video, + output, + time_sec=100.0, # 超过视频时长 + ) + + cmd = mock_run.call_args[0][0] + ss_idx = cmd.index("-ss") + time_val = float(cmd[ss_idx + 1]) + # 应该被钳制到中间帧（5秒左右） + assert time_val <= 10.0 + + @patch("video_processing.cover_generator.run_ffmpeg") + @patch("video_processing.cover_generator.probe_video_info") + def test_extract_frame_negative_time(self, mock_probe, mock_run, sample_video, tmp_path): + """负时间钳制到0.""" + mock_probe.return_value = {"duration": 30.0} + + def fake_run_ffmpeg(cmd): + output_path = Path(cmd[-1]) + output_path.parent.mkdir(parents=True, exist_ok=True) + output_path.write_bytes(b"fake jpeg data") + + mock_run.side_effect = fake_run_ffmpeg + + output = tmp_path / "cover.jpg" + CoverGenerator.extract_frame( + sample_video, + output, + time_sec=-5.0, + ) + + cmd = mock_run.call_args[0][0] + ss_idx = cmd.index("-ss") + time_val = float(cmd[ss_idx + 1]) + assert time_val >= 0 + + def test_extract_frame_file_not_found(self, tmp_path): + """视频文件不存在抛异常.""" + with pytest.raises(FileNotFoundError): + CoverGenerator.extract_frame( + "/nonexistent/video.mp4", + tmp_path / "cover.jpg", + ) + + @patch("video_processing.cover_generator.CoverGenerator.extract_frame") + @patch("video_processing.cover_generator.probe_video_info") + def test_smart_cover_3_frames(self, mock_probe, mock_extract, sample_video, tmp_path): + """智能封面抽取3帧选最佳.""" + mock_probe.return_value = {"duration": 30.0} + + # 创建三个大小不同的临时文件（模拟清晰度不同） + def create_frame(video_path, output_path, **kwargs): + # 第二帧最大（最清晰） + p = Path(output_path) + p.parent.mkdir(parents=True, exist_ok=True) + if "candidate_1" in str(p): + p.write_bytes(b"x" * 10000) # 最大 = 最清晰 + elif "candidate_0" in str(p): + p.write_bytes(b"x" * 1000) + else: + p.write_bytes(b"x" * 5000) + return p + + mock_extract.side_effect = create_frame + + output = tmp_path / "smart_cover.jpg" + result = CoverGenerator.extract_smart_cover( + sample_video, + output, + frame_count=3, + ) + + assert result == output + assert output.exists() + # 应该选最大的那个文件（candidate_1） + assert output.stat().st_size == 10000 + + @patch("video_processing.cover_generator.CoverGenerator.extract_frame") + @patch("video_processing.cover_generator.probe_video_info") + def test_smart_cover_fallback(self, mock_probe, mock_extract, sample_video, tmp_path): + """智能封面全部失败时降级.""" + mock_probe.return_value = {"duration": 0.0} # 时长为0 + + output = tmp_path / "cover.jpg" + output.write_bytes(b"x" * 100) + mock_extract.return_value = output + + result = CoverGenerator.extract_smart_cover(sample_video, output, frame_count=3) + assert result == output + + @patch("video_processing.cover_generator.run_ffmpeg") + def test_custom_cover(self, mock_run, sample_image, tmp_path): + """自定义封面处理.""" + output = tmp_path / "custom_cover.jpg" + + result = CoverGenerator.process_custom_cover( + sample_image, + output, + ) + + assert result == output + mock_run.assert_called_once() + cmd = mock_run.call_args[0][0] + assert str(sample_image) in cmd + + def test_custom_cover_not_found(self, tmp_path): + """自定义封面文件不存在.""" + with pytest.raises(FileNotFoundError): + CoverGenerator.process_custom_cover( + "/nonexistent/img.png", + tmp_path / "cover.jpg", + ) + + @patch("video_processing.cover_generator.CoverGenerator.extract_frame") + def test_generate_cover_time_mode(self, mock_extract, sample_video, tmp_path): + """统一入口 - time 模式.""" + output = tmp_path / "cover.jpg" + mock_extract.return_value = output + + result = CoverGenerator.generate_cover( + sample_video, + output, + mode="time", + time_sec=3.0, + ) + + assert result == output + mock_extract.assert_called_once() + + @patch("video_processing.cover_generator.CoverGenerator.extract_smart_cover") + def test_generate_cover_smart_mode(self, mock_smart, sample_video, tmp_path): + """统一入口 - smart 模式.""" + output = tmp_path / "cover.jpg" + mock_smart.return_value = output + + result = CoverGenerator.generate_cover( + sample_video, + output, + mode="smart", + ) + + assert result == output + mock_smart.assert_called_once() + + @patch("video_processing.cover_generator.CoverGenerator.process_custom_cover") + def test_generate_cover_custom_mode(self, mock_custom, sample_video, sample_image, tmp_path): + """统一入口 - custom 模式.""" + output = tmp_path / "cover.jpg" + mock_custom.return_value = output + + result = CoverGenerator.generate_cover( + sample_video, + output, + mode="custom", + custom_image=sample_image, + ) + + assert result == output + mock_custom.assert_called_once() + + +class TestGenerateCoverFromPlan: + """从 plan 配置生成封面测试.""" + + @patch("video_processing.cover_generator.CoverGenerator.extract_smart_cover") + def test_smart_mode_from_plan(self, mock_smart, sample_video, tmp_path): + """plan 配置 smart 模式.""" + plan = FakePlan(id="plan_001", config={"cover_config": {"mode": "smart"}}) + mock_smart.return_value = tmp_path / "cover.jpg" + (tmp_path / "cover.jpg").write_bytes(b"test") + + result = generate_cover_from_plan(plan, sample_video, tmp_path) + assert result is not None + + def test_no_cover_config(self, sample_video, tmp_path): + """没有封面配置时返回 None.""" + plan = FakePlan(id="plan_001", config={}) + result = generate_cover_from_plan(plan, sample_video, tmp_path) + assert result is None + + def test_none_config(self, sample_video, tmp_path): + """config 为 None.""" + plan = FakePlan(id="plan_001", config=None) # type: ignore + result = generate_cover_from_plan(plan, sample_video, tmp_path) + assert result is None + + +# ═══════════════════════════════════════════════════════════════════════════════ +# 四、UnifiedRenderService 集成测试 +# ═══════════════════════════════════════════════════════════════════════════════ + + +def _make_clip(clip_id="c1", asset_id="a1", path=Path("/fake/video.mp4"), clip_type="main", config=None): + """创建测试用 ResolvedClip.""" + return ResolvedClip( + clip_id=clip_id, + asset_id=asset_id, + local_path=path, + clip_type=clip_type, + order=0, + start_time=0.0, + duration=0.0, + transition_effect="cut", + config=config or {}, + actual_duration=10.0, + ) + + +def _make_service(plan, clips, asset_path_map=None, work_dir=None, tmp_path=None): + """创建测试用 UnifiedRenderService.""" + from pathlib import Path as P + + work_dir = work_dir or (tmp_path or P("/tmp")) / "render_test" + work_dir.mkdir(exist_ok=True, parents=True) + return UnifiedRenderService( + plan=plan, + clips=clips, + asset_path_map=asset_path_map or {}, + work_dir=work_dir, + output_width=1080, + output_height=1920, + output_fps=30, + transition_duration=0.5, + ) + + +class TestReverseIntegration: + """倒放功能集成测试.""" + + @patch("video_processing.unified_render_service.probe_video_info") + @patch("video_processing.unified_render_service.run_ffmpeg") + def test_reverse_in_filter_complex(self, mock_run, mock_probe, tmp_path): + """filter_complex 路径中包含倒放滤镜.""" + mock_probe.return_value = {"duration": 10.0, "has_audio": True, "width": 1920, "height": 1080} + mock_run.return_value = None + + plan = FakePlan(id="p1") + clip = _make_clip(config={"reverse": {"enabled": True}}) + clip.actual_duration = 5.0 + # 两个 clip 触发 filter_complex 路径 + clip2 = _make_clip(clip_id="c2", config={}) + clip2.actual_duration = 5.0 + clip2.order = 1 + + service = _make_service(plan, [clip, clip2], tmp_path=tmp_path) + + # 直接测 _build_filter_complex + from video_processing.unified_render_service import RenderLayer + + layer = RenderLayer(role="main", clips=[clip, clip2]) + filter_str, inputs = service._build_filter_complex([layer]) + + assert "reverse" in filter_str + + def test_can_use_pass_through_with_reverse(self, tmp_path): + """倒放不影响直通模式判断（只有贴纸才禁用）.""" + plan = FakePlan(id="p1") + clip = _make_clip(config={"reverse": {"enabled": True}}) + clip.actual_duration = 5.0 + + service = _make_service(plan, [clip], tmp_path=tmp_path) + from video_processing.unified_render_service import RenderLayer + + layer = RenderLayer(role="main", clips=[clip]) + layers = [layer] + + assert service._can_use_pass_through(layers) is True + + +class TestStickerIntegration: + """贴纸功能集成测试.""" + + def test_can_use_pass_through_with_stickers(self, tmp_path): + """有贴纸时禁用直通模式.""" + plan = FakePlan(id="p1", config={"stickers": [{"type": "text", "text": "Hello", "position": "center"}]}) + clip = _make_clip() + clip.actual_duration = 5.0 + + service = _make_service(plan, [clip], tmp_path=tmp_path) + from video_processing.unified_render_service import RenderLayer + + layer = RenderLayer(role="main", clips=[clip]) + layers = [layer] + + assert service._can_use_pass_through(layers) is False + + def test_can_use_pass_through_no_stickers(self, tmp_path): + """无贴纸时直通模式正常.""" + plan = FakePlan(id="p1", config={}) + clip = _make_clip() + clip.actual_duration = 5.0 + + service = _make_service(plan, [clip], tmp_path=tmp_path) + from video_processing.unified_render_service import RenderLayer + + layer = RenderLayer(role="main", clips=[clip]) + layers = [layer] + + assert service._can_use_pass_through(layers) is True + + def test_build_sticker_filters_text(self, tmp_path): + """文字贴纸滤镜构建.""" + plan = FakePlan( + id="p1", config={"stickers": [{"type": "text", "text": "Hello", "position": "top_center", "z_index": 10}]} + ) + service = _make_service(plan, [], tmp_path=tmp_path) + + filter_str, extra_inputs = service._build_sticker_filters("in", "out") + + assert "drawtext" in filter_str + assert len(extra_inputs) == 0 + + def test_build_sticker_filters_empty(self, tmp_path): + """无贴纸返回空.""" + plan = FakePlan(id="p1", config={}) + service = _make_service(plan, [], tmp_path=tmp_path) + + filter_str, extra_inputs = service._build_sticker_filters("in", "out") + + assert filter_str == "" + assert extra_inputs == [] + + def test_build_sticker_filters_image(self, sample_image, tmp_path): + """图片贴纸滤镜构建 + 额外输入.""" + plan = FakePlan( + id="p1", + config={ + "stickers": [ + { + "type": "image", + "image_path": str(sample_image), + "position": "bottom_right", + "z_index": 5, + } + ] + }, + ) + service = _make_service(plan, [], tmp_path=tmp_path) + + filter_str, extra_inputs = service._build_sticker_filters("in", "out") + + assert "overlay" in filter_str + assert len(extra_inputs) == 1 -- 2.54.0 From e7a9c0812214b3d7d7e18c1f14d27d8f6911a290 Mon Sep 17 00:00:00 2001 From: xiaoxia-bot Date: Tue, 14 Jul 2026 11:02:23 +0800 Subject: [PATCH 2/2] =?UTF-8?q?fix:=20black=E6=A0=BC=E5=BC=8F=E5=8C=96?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- apps/worker/video_processing/unified_render_service.py | 1 - 1 file changed, 1 deletion(-) diff --git a/apps/worker/video_processing/unified_render_service.py b/apps/worker/video_processing/unified_render_service.py index e4cc57028..8768f056a 100755 --- a/apps/worker/video_processing/unified_render_service.py +++ b/apps/worker/video_processing/unified_render_service.py @@ -1477,7 +1477,6 @@ class UnifiedRenderService: input_args.extend(["-i", img_path]) final_video_label = "after_stickers" - # 叠加字幕（如有）+ 最终像素格式 if ass_path is not None: ass_filter_path = str(ass_path).replace("\\", "/").replace(":", "\\:") -- 2.54.0