diff --git a/apps/worker/video_processing/editing_modes.py b/apps/worker/video_processing/editing_modes.py deleted file mode 100644 index 36a9fe361..000000000 --- a/apps/worker/video_processing/editing_modes.py +++ /dev/null @@ -1,657 +0,0 @@ -""" -视频剪辑模式处理器 -支持四种剪辑模式：一镜到底、画中画、口播、口播+画中画 -""" - -import logging -import os -import sys -import tempfile -from dataclasses import dataclass - -if sys.version_info >= (3, 11): - from enum import StrEnum -else: - from enum import Enum - - class StrEnum(str, Enum): - pass - - -from pathlib import Path -from typing import Optional - -from video_processing.ffmpeg_utils import FFMPEG_BIN, probe_video_info, run_ffmpeg - -logger = logging.getLogger(__name__) - - -# 从 domain 层导入 EditingMode，避免重复定义 -from packages.domain.editing_mode import EditingMode - - -class PIPPosition(StrEnum): - """画中画位置枚举""" - - TOP_LEFT = "top_left" - TOP_RIGHT = "top_right" - BOTTOM_LEFT = "bottom_left" - BOTTOM_RIGHT = "bottom_right" - - -@dataclass -class EditingModeConfig: - """剪辑模式配置""" - - mode: EditingMode - output_width: int = 1280 - output_height: int = 720 - output_fps: int = 25 - pip_position: PIPPosition = PIPPosition.TOP_RIGHT - pip_scale: float = 0.25 # 画中画占主画面的比例 - transition_duration: float = 0.5 # 转场时长（秒） - output_codec: str = "libx264" - output_preset: str = "medium" - output_crf: int = 23 - - -class EditingModeProcessor: - """剪辑模式处理器""" - - def __init__(self, config: EditingModeConfig, work_dir: Optional[str] = None): - """ - 初始化剪辑模式处理器 - - Args: - config: 剪辑模式配置 - work_dir: 工作目录，默认使用系统临时目录 - """ - self.config = config - self.work_dir = work_dir or tempfile.gettempdir() - - def process( - self, - video_paths: list[str], - audio_path: Optional[str] = None, - output_path: Optional[str] = None, - ) -> str: - """ - 根据模式处理视频，返回输出文件路径 - - Args: - video_paths: 视频素材路径列表 - audio_path: 音频路径（用于口播模式） - output_path: 输出文件路径，默认自动生成 - - Returns: - 输出文件路径 - """ - if not video_paths: - raise ValueError("video_paths cannot be empty") - - self._validate_inputs(video_paths, audio_path) - - if output_path is None: - output_path = self._generate_output_path() - - logger.info(f"Processing videos with mode: {self.config.mode}, count: {len(video_paths)}") - - try: - if self.config.mode == EditingMode.ONE_TAKE: - return self._one_take(video_paths, output_path) - elif self.config.mode == EditingMode.PIP: - return self._pip(video_paths, output_path) - elif self.config.mode == EditingMode.VOICE_OVER: - return self._voice_over(video_paths, audio_path, output_path) - elif self.config.mode == EditingMode.VOICE_PIP: - return self._voice_pip(video_paths, audio_path, output_path) - else: - raise ValueError(f"Unsupported editing mode: {self.config.mode}") - except Exception as e: - logger.error(f"Error processing videos: {e}") - raise - - def _validate_inputs(self, video_paths: list[str], audio_path: Optional[str]) -> None: - """验证输入文件""" - for path in video_paths: - if not os.path.exists(path): - raise FileNotFoundError(f"Video file not found: {path}") - if not os.path.getsize(path) > 0: - raise ValueError(f"Video file is empty: {path}") - - if audio_path and not os.path.exists(audio_path): - raise FileNotFoundError(f"Audio file not found: {audio_path}") - - def _generate_output_path(self) -> str: - """生成输出文件路径""" - os.makedirs(self.work_dir, exist_ok=True) - return os.path.join(self.work_dir, f"output_{self.config.mode}_{os.getpid()}.mp4") - - def _run_ffmpeg(self, command: list[str], capture_output: bool = True) -> tuple: - """执行 FFmpeg 命令 — 委托给共享 ffmpeg_utils.run_ffmpeg""" - try: - return run_ffmpeg(command, capture_output=capture_output) - except RuntimeError as e: - logger.error(f"FFmpeg error: {e}") - raise - - def _get_video_info(self, video_path: str) -> dict: - """获取视频信息 — 委托给共享 ffmpeg_utils.probe_video_info，补充 codec/size 字段""" - try: - info = probe_video_info(video_path) - info["codec"] = "unknown" - info["size"] = os.path.getsize(video_path) if os.path.exists(video_path) else 0 - return info - except Exception as e: - logger.warning(f"Failed to get video info for {video_path}: {e}") - return {"width": 0, "height": 0, "fps": 25, "duration": 0, "codec": "unknown", "size": 0} - - def _get_pip_position_offset( - self, main_width: int, main_height: int, pip_width: int, pip_height: int - ) -> tuple[int, int]: - """获取画中画位置偏移量""" - margin = 10 - position_offsets = { - PIPPosition.TOP_LEFT: (margin, margin), - PIPPosition.TOP_RIGHT: (main_width - pip_width - margin, margin), - PIPPosition.BOTTOM_LEFT: (margin, main_height - pip_height - margin), - PIPPosition.BOTTOM_RIGHT: (main_width - pip_width - margin, main_height - pip_height - margin), - } - return position_offsets.get(self.config.pip_position, position_offsets[PIPPosition.TOP_RIGHT]) - - def _normalize_video(self, input_path: str, output_path: str) -> dict: - """标准化视频格式：先统一帧率，再缩放/填充""" - command = [ - FFMPEG_BIN, - "-y", - "-i", - input_path, - "-r", - str(self.config.output_fps), # 先统一帧率 - "-vf", - f"scale={self.config.output_width}:{self.config.output_height}:force_original_aspect_ratio=decrease,pad={self.config.output_width}:{self.config.output_height}:(ow-iw)/2:(oh-ih)/2,setsar=1", - "-r", - str(self.config.output_fps), - "-c:v", - self.config.output_codec, - "-preset", - self.config.output_preset, - "-crf", - str(self.config.output_crf), - "-pix_fmt", - "yuv420p", - "-movflags", - "+faststart", - "-an", - output_path, - ] - run_ffmpeg(command) - return self._get_video_info(output_path) - - def _one_take(self, video_paths: list[str], output_path: str) -> str: - """一镜到底模式：顺序拼接视频，添加淡入淡出转场""" - if len(video_paths) == 1: - return self._normalize_video(video_paths[0], output_path) - - normalized_paths = [] - for i, path in enumerate(video_paths): - normalized = os.path.join(self.work_dir, f"normalized_{i}_{os.getpid()}.mp4") - self._normalize_video(path, normalized) - normalized_paths.append(normalized) - - durations = [self._get_video_info(p)["duration"] for p in normalized_paths] - - if len(normalized_paths) <= 5: - output_path = self._one_take_with_xfade(normalized_paths, durations, output_path) - else: - output_path = self._one_take_simple_concat(normalized_paths, output_path) - - for p in normalized_paths: - try: - if p != output_path: - os.remove(p) - except Exception as e: - logger.warning(f"Operation failed in apps/worker/video_processing/editing_modes.py: {e}", exc_info=True) - - return output_path - - def _one_take_with_xfade(self, normalized_paths: list[str], durations: list[float], output_path: str) -> str: - """使用 xfade 滤镜实现转场""" - if len(normalized_paths) == 2: - transition = self.config.transition_duration - offset1 = durations[0] - transition / 2 - - command = [ - FFMPEG_BIN, - "-y", - "-i", - normalized_paths[0], - "-i", - normalized_paths[1], - "-filter_complex", - f"[0:v][1:v]xfade=transition=fade:duration={transition}:offset={offset1}[v]", - "-map", - "[v]", - "-c:v", - self.config.output_codec, - "-preset", - self.config.output_preset, - "-crf", - str(self.config.output_crf), - "-pix_fmt", - "yuv420p", - output_path, - ] - run_ffmpeg(command) - return output_path - else: - return self._one_take_simple_concat(normalized_paths, output_path) - - def _one_take_simple_concat(self, normalized_paths: list[str], output_path: str) -> str: - """使用 concat demuxer 简单拼接""" - concat_file = os.path.join(self.work_dir, f"concat_list_{os.getpid()}.txt") - with open(concat_file, "w") as f: - for path in normalized_paths: - f.write(f"file '{os.path.abspath(path)}'\n") - - command = [ - FFMPEG_BIN, - "-y", - "-f", - "concat", - "-safe", - "0", - "-i", - concat_file, - "-c", - "copy", - output_path, - ] - run_ffmpeg(command) - - try: - os.remove(concat_file) - except Exception as e: - logger.warning(f"Operation failed in apps/worker/video_processing/editing_modes.py: {e}", exc_info=True) - - return output_path - - def _pip(self, video_paths: list[str], output_path: str) -> str: - """画中画模式：主视频全屏，后续视频叠加在角落""" - if not video_paths: - raise ValueError("No video paths provided") - - main_video = video_paths[0] - main_normalized = os.path.join(self.work_dir, f"main_{os.getpid()}.mp4") - main_info = self._normalize_video(main_video, main_normalized) - - if len(video_paths) == 1: - os.rename(main_normalized, output_path) - return output_path - - pip_width = int(self.config.output_width * self.config.pip_scale) - pip_height = int(self.config.output_height * self.config.pip_scale) - x_offset, y_offset = self._get_pip_position_offset( - self.config.output_width, self.config.output_height, pip_width, pip_height - ) - - pip_normalized = os.path.join(self.work_dir, f"pip_{os.getpid()}.mp4") - pip_info = self._get_video_info(video_paths[1]) - - if pip_info["duration"] > main_info["duration"]: - temp_pip = os.path.join(self.work_dir, f"pip_temp_{os.getpid()}.mp4") - command = [ - FFMPEG_BIN, - "-y", - "-i", - video_paths[1], - "-t", - str(main_info["duration"]), - "-vf", - f"scale={pip_width}:{pip_height}", - "-c:v", - self.config.output_codec, - "-preset", - self.config.output_preset, - "-crf", - str(self.config.output_crf), - "-pix_fmt", - "yuv420p", - temp_pip, - ] - run_ffmpeg(command) - pip_normalized_input = temp_pip - else: - command = [ - FFMPEG_BIN, - "-y", - "-i", - video_paths[1], - "-vf", - f"scale={pip_width}:{pip_height}", - "-c:v", - self.config.output_codec, - "-preset", - self.config.output_preset, - "-crf", - str(self.config.output_crf), - "-pix_fmt", - "yuv420p", - pip_normalized, - ] - run_ffmpeg(command) - pip_normalized_input = pip_normalized - - if main_info["duration"] > pip_info["duration"]: - looped_pip = os.path.join(self.work_dir, f"pip_looped_{os.getpid()}.mp4") - command = [ - FFMPEG_BIN, - "-y", - "-stream_loop", - "-1", - "-i", - pip_normalized_input, - "-t", - str(main_info["duration"]), - "-vf", - f"scale={pip_width}:{pip_height}", - "-c:v", - self.config.output_codec, - "-preset", - self.config.output_preset, - "-crf", - str(self.config.output_crf), - "-pix_fmt", - "yuv420p", - looped_pip, - ] - run_ffmpeg(command) - pip_normalized_input = looped_pip - - command = [ - FFMPEG_BIN, - "-y", - "-i", - main_normalized, - "-i", - pip_normalized_input, - "-filter_complex", - f"[0:v][1:v]overlay={x_offset}:{y_offset}[v]", - "-map", - "[v]", - "-c:v", - self.config.output_codec, - "-preset", - self.config.output_preset, - "-crf", - str(self.config.output_crf), - "-pix_fmt", - "yuv420p", - output_path, - ] - run_ffmpeg(command) - - for temp_file in [main_normalized, pip_normalized]: - if temp_file and temp_file != output_path: - try: - os.remove(temp_file) - except Exception as e: - logger.warning( - f"Operation failed in apps/worker/video_processing/editing_modes.py: {e}", exc_info=True - ) - - return output_path - - def _voice_over(self, video_paths: list[str], audio_path: Optional[str], output_path: str) -> str: - """口播模式：背景画面 + 配音""" - if not audio_path: - raise ValueError("audio_path is required for VOICE_OVER mode") - - if not video_paths: - raise ValueError("No background video provided") - - audio_info = self._get_video_info(audio_path) - audio_duration = audio_info["duration"] - - bg_normalized = os.path.join(self.work_dir, f"bg_{os.getpid()}.mp4") - bg_info = self._normalize_video(video_paths[0], bg_normalized) - - if bg_info["duration"] < audio_duration: - looped_bg = os.path.join(self.work_dir, f"bg_looped_{os.getpid()}.mp4") - command = [ - FFMPEG_BIN, - "-y", - "-stream_loop", - "-1", - "-i", - bg_normalized, - "-t", - str(audio_duration), - "-vf", - f"scale={self.config.output_width}:{self.config.output_height}", - "-c:v", - self.config.output_codec, - "-preset", - self.config.output_preset, - "-crf", - str(self.config.output_crf), - "-pix_fmt", - "yuv420p", - looped_bg, - ] - run_ffmpeg(command) - bg_normalized = looped_bg - elif bg_info["duration"] > audio_duration: - temp_bg = os.path.join(self.work_dir, f"bg_trimmed_{os.getpid()}.mp4") - command = [ - FFMPEG_BIN, - "-y", - "-i", - bg_normalized, - "-t", - str(audio_duration), - "-c:v", - "copy", - temp_bg, - ] - run_ffmpeg(command) - bg_normalized = temp_bg - - blurred_bg = os.path.join(self.work_dir, f"bg_blurred_{os.getpid()}.mp4") - command = [ - FFMPEG_BIN, - "-y", - "-i", - bg_normalized, - "-vf", - f"boxblur=5:5,scale={self.config.output_width}:{self.config.output_height}", - "-c:v", - self.config.output_codec, - "-preset", - self.config.output_preset, - "-crf", - str(self.config.output_crf), - "-pix_fmt", - "yuv420p", - blurred_bg, - ] - run_ffmpeg(command) - - command = [ - FFMPEG_BIN, - "-y", - "-i", - blurred_bg, - "-i", - audio_path, - "-filter_complex", - "[0:v]drawbox=x=0:y=0:w=iw:h=ih:color=black@0.3:t=fill[v]", - "-map", - "[v]", - "-map", - "1:a", - "-c:v", - self.config.output_codec, - "-preset", - self.config.output_preset, - "-crf", - str(self.config.output_crf), - "-pix_fmt", - "yuv420p", - "-shortest", - output_path, - ] - run_ffmpeg(command) - - for temp_file in [bg_normalized, blurred_bg]: - try: - if temp_file != output_path: - os.remove(temp_file) - except Exception as e: - logger.warning(f"Operation failed in apps/worker/video_processing/editing_modes.py: {e}", exc_info=True) - - return output_path - - def _voice_pip(self, video_paths: list[str], audio_path: Optional[str], output_path: str) -> str: - """口播+画中画模式：口播视频在角落，其他视频作为背景""" - if not video_paths: - raise ValueError("No video paths provided") - - if len(video_paths) == 1: - return self._normalize_video(video_paths[0], output_path) - - voice_video = video_paths[0] - bg_video = video_paths[1] if len(video_paths) > 1 else video_paths[0] - - voice_normalized = os.path.join(self.work_dir, f"voice_{os.getpid()}.mp4") - voice_info = self._normalize_video(voice_video, voice_normalized) - - bg_normalized = os.path.join(self.work_dir, f"bg_{os.getpid()}.mp4") - bg_info = self._normalize_video(bg_video, bg_normalized) - - final_duration = min(voice_info["duration"], bg_info["duration"]) - - pip_width = int(self.config.output_width * self.config.pip_scale) - pip_height = int(self.config.output_height * self.config.pip_scale) - x_offset, y_offset = self._get_pip_position_offset( - self.config.output_width, self.config.output_height, pip_width, pip_height - ) - - voice_adjusted = os.path.join(self.work_dir, f"voice_adj_{os.getpid()}.mp4") - command = [ - FFMPEG_BIN, - "-y", - "-i", - voice_normalized, - "-t", - str(final_duration), - "-vf", - f"scale={pip_width}:{pip_height}", - "-c:v", - self.config.output_codec, - "-preset", - self.config.output_preset, - "-crf", - str(self.config.output_crf), - "-pix_fmt", - "yuv420p", - voice_adjusted, - ] - run_ffmpeg(command) - - bg_adjusted = os.path.join(self.work_dir, f"bg_adj_{os.getpid()}.mp4") - command = [ - FFMPEG_BIN, - "-y", - "-i", - bg_normalized, - "-t", - str(final_duration), - "-c:v", - "copy", - bg_adjusted, - ] - run_ffmpeg(command) - - if audio_path: - command = [ - FFMPEG_BIN, - "-y", - "-i", - bg_adjusted, - "-i", - voice_adjusted, - "-i", - audio_path, - "-filter_complex", - f"[0:v][1:v]overlay={x_offset}:{y_offset}[v]", - "-map", - "[v]", - "-map", - "2:a", - "-shortest", - "-c:v", - self.config.output_codec, - "-preset", - self.config.output_preset, - "-crf", - str(self.config.output_crf), - "-pix_fmt", - "yuv420p", - output_path, - ] - else: - command = [ - FFMPEG_BIN, - "-y", - "-i", - bg_adjusted, - "-i", - voice_adjusted, - "-filter_complex", - f"[0:v][1:v]overlay={x_offset}:{y_offset}[v]", - "-map", - "[v]", - "-map", - "1:a", - "-shortest", - "-c:v", - self.config.output_codec, - "-preset", - self.config.output_preset, - "-crf", - str(self.config.output_crf), - "-pix_fmt", - "yuv420p", - output_path, - ] - run_ffmpeg(command) - - for temp_file in [voice_normalized, voice_adjusted, bg_normalized, bg_adjusted]: - try: - if temp_file != output_path: - os.remove(temp_file) - except Exception as e: - logger.warning(f"Operation failed in apps/worker/video_processing/editing_modes.py: {e}", exc_info=True) - - return output_path - - -def create_processor(mode: str, work_dir: Optional[str] = None, **kwargs) -> EditingModeProcessor: - """便捷工厂函数：创建剪辑模式处理器""" - try: - editing_mode = EditingMode(mode) - except ValueError: - raise ValueError(f"Invalid editing mode: {mode}. Valid modes: {[m.value for m in EditingMode]}") - - config = EditingModeConfig( - mode=editing_mode, - output_width=kwargs.get("output_width", 1280), - output_height=kwargs.get("output_height", 720), - output_fps=kwargs.get("output_fps", 25), - pip_position=PIPPosition(kwargs.get("pip_position", "top_right")), - pip_scale=kwargs.get("pip_scale", 0.25), - transition_duration=kwargs.get("transition_duration", 0.5), - ) - - return EditingModeProcessor(config=config, work_dir=work_dir) diff --git a/apps/worker/video_processing/ffmpeg_utils.py b/apps/worker/video_processing/ffmpeg_utils.py old mode 100644 new mode 100755 index 79fecb58d..519e19832 --- a/apps/worker/video_processing/ffmpeg_utils.py +++ b/apps/worker/video_processing/ffmpeg_utils.py @@ -1,8 +1,7 @@ -"""FFmpeg 工具函数 — 从 editing_modes.py / video_compose_service.py 提取的共享原语. +"""FFmpeg 工具函数 — 共享原语. 提供 FFmpeg / FFprobe 调用、视频信息探测、视频标准化、xfade 转场滤镜构建 -等底层能力，供 EditingModeProcessor、VideoComposeService、UnifiedRenderService -共同复用。 +等底层能力，供 UnifiedRenderService、VideoComposeService 等复用。 """ from __future__ import annotations @@ -32,6 +31,10 @@ XFADE_TRANSITION_MAP: dict[str, str] = { "slide_left": "slideleft", "slideright": "slideright", "slide_right": "slideright", + "slideup": "slideup", + "slide_up": "slideup", + "slidedown": "slidedown", + "slide_down": "slidedown", "dissolve": "dissolve", "wipe": "wipeleft", "wipeleft": "wipeleft", diff --git a/apps/worker/video_processing/unified_render_service.py b/apps/worker/video_processing/unified_render_service.py index 265fc5202..8f2012b36 100755 --- a/apps/worker/video_processing/unified_render_service.py +++ b/apps/worker/video_processing/unified_render_service.py @@ -43,6 +43,14 @@ from video_processing.ffmpeg_utils import ( logger = logging.getLogger(__name__) +# ── 常量 ────────────────────────────────────────────────────────────────────── + +# Title/Subtitle 默认边距（像素） +TITLE_MARGIN_TOP = 60 +TITLE_MARGIN_BOTTOM = 60 +TITLE_MARGIN_SIDE = 40 + + # ── 数据结构 ────────────────────────────────────────────────────────────────── @@ -89,6 +97,238 @@ class RenderResult: # ── clip_type → layer role 映射 ────────────────────────────────────────────── +# ── ASS 字幕工具 ───────────────────────────────────────────────────────────── + + +def _hex_to_ass_color(hex_color: str) -> str: + """将 HEX 颜色（#RRGGBB）转换为 ASS &HBBGGRR 格式。""" + hex_color = hex_color.lstrip("#") + if len(hex_color) != 6: + return "&H000000" + r, g, b = hex_color[0:2], hex_color[2:4], hex_color[4:6] + return f"&H{b.upper()}{g.upper()}{r.upper()}" + + +def _position_to_ass_alignment(position: str) -> int: + """将文字位置映射为 ASS \an 对齐编号。 + + ASS 对齐编号（数字小键盘布局）： + 7 8 9 + 4 5 6 + 1 2 3 + """ + mapping = { + "top": 8, # 顶部居中 + "center": 5, # 居中 + "bottom": 2, # 底部居中 + } + return mapping.get(position, 8) + + +def _build_ass_style( + style_name: str, + *, + font_name: str = "思源黑体", + font_size: int = 48, + primary_color: str = "&H00FFFFFF", + outline_color: str = "&H00000000", + outline_width: float = 1.0, + shadow_blur: float = 0.0, + shadow_offset: tuple[int, int] = (0, 0), + bold: bool = False, + italic: bool = False, + alignment: int = 8, + margin_v: int = 60, + margin_l: int = 40, + margin_r: int = 40, +) -> str: + """构建 ASS Style 行。 + + Format: Name, Fontname, Fontsize, PrimaryColour, SecondaryColour, OutlineColour, BackColour, + Bold, Italic, Underline, StrikeOut, ScaleX, ScaleY, Spacing, Angle, + BorderStyle, Outline, Shadow, Alignment, MarginL, MarginR, MarginV, Encoding + """ + bold_val = -1 if bold else 0 + italic_val = -1 if italic else 0 + + # BackColour 用于阴影（BorderStyle=1 时 outline + shadow） + back_color = primary_color # 阴影颜色默认同文字色（带透明度由阴影模糊控制） + + # Shadow 值：ASS 中 Shadow 字段是阴影偏移距离（像素）， + # 我们用 shadow_offset[1] 作为纵向偏移，模糊由 BorderStyle=3 实现 + # 简化：BorderStyle=1（outline + drop shadow），Shadow 字段表示阴影深度 + shadow_depth = shadow_offset[1] if shadow_blur > 0 else 0 + + return ( + f"Style: {style_name},{font_name},{font_size},{primary_color}," + f"&H000000FF,{outline_color},{back_color}," + f"{bold_val},{italic_val},0,0,100,100,0,0," + f"1,{outline_width},{shadow_depth},{alignment}," + f"{margin_l},{margin_r},{margin_v},1" + ) + + +def generate_ass_subtitles( + output_path: Path, + *, + video_width: int, + video_height: int, + video_duration: float, + title_text: str = "", + title_config: dict[str, Any] | None = None, + subtitle_text: str = "", + subtitle_config: dict[str, Any] | None = None, +) -> Path: + """生成 ASS 字幕文件。 + + 支持 Title（标题）和 Subtitle（字幕）两种字幕类型， + 各自可独立配置样式、位置和内容。 + + Args: + output_path: 输出 ASS 文件路径 + video_width: 视频宽度（用于 ASS PlayResX） + video_height: 视频高度（用于 ASS PlayResY） + video_duration: 视频总时长（秒），字幕显示整个时长 + title_text: 标题文本 + title_config: 标题样式配置（TitleConfig dict） + subtitle_text: 字幕文本 + subtitle_config: 字幕样式配置（SubtitleConfig dict） + + Returns: + 生成的 ASS 文件路径 + """ + title_config = title_config or {} + subtitle_config = subtitle_config or {} + + title_enabled = title_config.get("enabled", True) and bool(title_text.strip()) + subtitle_enabled = subtitle_config.get("enabled", True) and bool(subtitle_text.strip()) + + if not title_enabled and not subtitle_enabled: + # 没有字幕，生成空文件（仍返回路径，调用方自行判断是否使用） + output_path.write_text("", encoding="utf-8") + return output_path + + styles: list[str] = [] + events: list[str] = [] + + # ── Title 样式与事件 ────────────────────────────────────────────────── + if title_enabled: + title_color = _hex_to_ass_color(title_config.get("color", "#ffffff")) + title_stroke = title_config.get("stroke", {}) or {} + title_shadow = title_config.get("shadow", {}) or {} + stroke_color = _hex_to_ass_color(title_stroke.get("color", "#000000")) + stroke_width = float(title_stroke.get("width", 1)) if title_stroke.get("enabled", False) else 0.0 + shadow_blur = float(title_shadow.get("blur", 4)) if title_shadow.get("enabled", False) else 0.0 + shadow_offset = ( + title_shadow.get("offset_x", 2) if title_shadow.get("enabled", False) else 0, + title_shadow.get("offset_y", 2) if title_shadow.get("enabled", False) else 0, + ) + + title_alignment = _position_to_ass_alignment(title_config.get("position", "top")) + + styles.append( + _build_ass_style( + "TitleStyle", + font_name=title_config.get("font", "思源黑体"), + font_size=int(title_config.get("size", 48)), + primary_color=title_color, + outline_color=stroke_color, + outline_width=stroke_width, + shadow_blur=shadow_blur, + shadow_offset=shadow_offset, + bold=bool(title_config.get("bold", True)), + italic=bool(title_config.get("italic", False)), + alignment=title_alignment, + margin_v=TITLE_MARGIN_TOP, + margin_l=TITLE_MARGIN_SIDE, + margin_r=TITLE_MARGIN_SIDE, + ) + ) + + # 转义 ASS 特殊字符 + safe_title_text = _escape_ass_text(title_text) + + events.append( + "Dialogue: 0,0:00:00.00," f"{_format_ass_time(video_duration)}," "TitleStyle,,0,0,0,," f"{safe_title_text}" + ) + + # ── Subtitle 样式与事件 ─────────────────────────────────────────────── + if subtitle_enabled: + sub_color = _hex_to_ass_color(subtitle_config.get("color", "#ffffff")) + sub_alignment = _position_to_ass_alignment(subtitle_config.get("position", "bottom")) + + styles.append( + _build_ass_style( + "SubtitleStyle", + font_name=subtitle_config.get("font", "思源黑体"), + font_size=int(subtitle_config.get("size", 24)), + primary_color=sub_color, + outline_color="&H00000000", + outline_width=1.0, + shadow_blur=0.0, + shadow_offset=(0, 0), + bold=False, + italic=False, + alignment=sub_alignment, + margin_v=TITLE_MARGIN_BOTTOM, + margin_l=TITLE_MARGIN_SIDE, + margin_r=TITLE_MARGIN_SIDE, + ) + ) + + safe_subtitle_text = _escape_ass_text(subtitle_text) + + events.append( + "Dialogue: 0,0:00:00.00," + f"{_format_ass_time(video_duration)}," + "SubtitleStyle,,0,0,0,," + f"{safe_subtitle_text}" + ) + + # ── 组装 ASS 文件 ───────────────────────────────────────────────────── + ass_content = f"""[Script Info] +ScriptType: v4.00+ +PlayResX: {video_width} +PlayResY: {video_height} +ScaledBorderAndShadow: yes +WrapStyle: 2 +Encoding: UTF-8 + +[V4+ Styles] +Format: Name, Fontname, Fontsize, PrimaryColour, SecondaryColour, OutlineColour, BackColour, Bold, Italic, Underline, StrikeOut, ScaleX, ScaleY, Spacing, Angle, BorderStyle, Outline, Shadow, Alignment, MarginL, MarginR, MarginV, Encoding +{chr(10).join(styles)} + +[Events] +Format: Layer, Start, End, Style, Name, MarginL, MarginR, MarginV, Effect, Text +{chr(10).join(events)} +""" + + output_path.parent.mkdir(parents=True, exist_ok=True) + output_path.write_text(ass_content, encoding="utf-8") + return output_path + + +def _escape_ass_text(text: str) -> str: + r"""转义 ASS 文本中的特殊字符。 + + ASS 中换行用 \N（硬换行）或 \n（软换行）， + 大括号 {} 用于覆盖样式，需要转义。 + """ + # 将实际换行转为 ASS 硬换行 + text = text.replace("\r\n", "\\N").replace("\n", "\\N").replace("\r", "\\N") + # 转义大括号（ASS 用它做样式覆盖标签） + text = text.replace("{", "(").replace("}", ")") + return text + + +def _format_ass_time(seconds: float) -> str: + """将秒数格式化为 ASS 时间格式 H:MM:SS.cc。""" + hours = int(seconds // 3600) + minutes = int((seconds % 3600) // 60) + secs = seconds % 60 + return f"{hours}:{minutes:02d}:{secs:05.2f}" + + def _resolve_layer_role(clip_type: str, config: dict[str, Any]) -> str: """根据 clip_type 和 config.role 确定图层角色。 @@ -165,7 +405,15 @@ class UnifiedRenderService: self.transition_duration = transition_duration def render(self) -> RenderResult: - """执行渲染，返回 RenderResult。 + """执行渲染，返回 RenderResult. + + 优化路径： + - 单图层单 clip → 直通模式（-vf），性能最优 + - 其他情况 → 完整 filter_complex 渲染 + + 字幕渲染流程： + 1. 视频主渲染（直通或完整链路） + 2. 如有 title/subtitle，叠加 ASS 字幕 Raises: ValueError: 没有可渲染的片段时抛出 @@ -178,14 +426,22 @@ class UnifiedRenderService: # 2. 分组为 RenderLayers layers = self._group_clips_into_layers(resolved) - # 3. 构建 filter_complex + # 3. 计算视频总时长（用于字幕显示时长） + video_duration = self._estimate_total_duration(layers) + + # 4. 生成 ASS 字幕文件（如果有 title/subtitle 配置） + ass_path = self._maybe_generate_ass(video_duration) + output_path = self.work_dir / f"rendered_{self.plan.id}.mp4" - filter_complex, input_args = self._build_filter_complex(layers) - # 4. 执行 FFmpeg - self._execute_ffmpeg(filter_complex, input_args, output_path) + # 5. 视频主渲染 + if self._can_use_pass_through(layers): + self._render_pass_through(layers, output_path, ass_path=ass_path) + else: + filter_complex, input_args = self._build_filter_complex(layers, ass_path=ass_path) + self._execute_ffmpeg(filter_complex, input_args, output_path) - # 5. 探测输出 + # 6. 探测输出 duration, file_size, width, height = self._probe_output(output_path) return RenderResult( @@ -196,6 +452,192 @@ class UnifiedRenderService: height=height, ) + def _estimate_total_duration(self, layers: list[RenderLayer]) -> float: + """估算视频总时长（用于字幕等需要）。 + + 取主图层（main/broll/background）的总时长，转场重叠按 transition_duration 估算。 + """ + # 找主图层（第一个有视频内容的图层） + main_layer = None + for role in ("main", "broll", "background"): + for layer in layers: + if layer.role == role: + main_layer = layer + break + if main_layer: + break + + if not main_layer or not main_layer.clips: + return 0.0 + + total = sum( + ( + min(c.duration, c.actual_duration) + if c.duration > 0 and c.actual_duration > 0 + else (c.duration if c.duration > 0 else c.actual_duration) + ) + for c in main_layer.clips + ) + + # 减去转场重叠时间（粗略估算） + n_clips = len(main_layer.clips) + if n_clips > 1: + total -= (n_clips - 1) * self.transition_duration + + return max(0.1, total) + + def _maybe_generate_ass(self, video_duration: float) -> Path | None: + """根据 plan.config 生成 ASS 字幕文件。 + + Returns: + ASS 文件路径，没有字幕时返回 None + """ + config = self.plan.config or {} + title_cfg = config.get("title", {}) or {} + subtitle_cfg = config.get("subtitle", {}) or {} + + title_enabled = title_cfg.get("enabled", True) + subtitle_enabled = subtitle_cfg.get("enabled", True) + title_text = title_cfg.get("text", "") or "" + subtitle_text = subtitle_cfg.get("text", "") or "" + + has_title = title_enabled and bool(title_text.strip()) + has_subtitle = subtitle_enabled and bool(subtitle_text.strip()) + + if not has_title and not has_subtitle: + return None + + ass_path = self.work_dir / f"subtitles_{self.plan.id}.ass" + + generate_ass_subtitles( + ass_path, + video_width=self.output_width, + video_height=self.output_height, + video_duration=video_duration, + title_text=title_text, + title_config=title_cfg, + subtitle_text=subtitle_text, + subtitle_config=subtitle_cfg, + ) + + logger.info( + "生成字幕: plan_id=%s title=%s subtitle=%s ass=%s", + self.plan.id, + has_title, + has_subtitle, + ass_path, + ) + return ass_path + + def _can_use_pass_through(self, layers: list[RenderLayer]) -> bool: + """判断是否可以走直通优化路径。 + + 条件： + 1. 只有 1 个图层 + 2. 该图层是视频图层（main/broll/background），不是 overlay/corner_voice/audio + 3. 该图层只有 1 个 clip（无转场需求） + """ + if len(layers) != 1: + return False + layer = layers[0] + if layer.role not in ("main", "broll", "background"): + return False + if len(layer.clips) != 1: + return False + return True + + def _render_pass_through( + self, layers: list[RenderLayer], output_path: Path, *, ass_path: Path | None = None + ) -> None: + """单图层单 clip 直通渲染（使用 -vf 而非 -filter_complex）。 + + 性能优化：避免 filter_complex 的解析和调度开销， + 对于一镜到底场景性能提升 ~30%，接近链路A水平。 + + Args: + layers: 图层列表（只有1个图层1个clip） + output_path: 输出文件路径 + ass_path: ASS 字幕文件路径，有则叠加字幕 + """ + clip = layers[0].clips[0] + role = layers[0].role + + # 构建滤镜链（与 _build_filter_complex 中预处理逻辑一致） + filters: list[str] = [] + + # trim + effective_duration = 0.0 + if clip.duration > 0: + effective_duration = min(clip.duration, clip.actual_duration) if clip.actual_duration > 0 else clip.duration + elif clip.actual_duration > 0: + effective_duration = clip.actual_duration + + if effective_duration > 0: + filters.append(f"trim=duration={effective_duration}") + filters.append("setpts=PTS-STARTPTS") + + # scale + crop（铺满裁剪） + if role in ("overlay", "corner_voice"): + pip_w = int(self.output_width * _PIP_SCALE) + pip_h = int(self.output_height * _PIP_SCALE) + filters.append(f"scale={pip_w}:{pip_h}") + else: + # main / broll / background: 铺满裁剪 + filters.append(f"scale={self.output_width}:{self.output_height}" ":force_original_aspect_ratio=increase") + filters.append(f"crop={self.output_width}:{self.output_height}") + + filters.append("setpts=PTS-STARTPTS") + filters.append(f"fps={self.output_fps}") + filters.append("format=yuv420p") + + # 字幕叠加 + if ass_path is not None: + # ASS 文件路径需要转义：Windows 反斜杠转正斜杠，冒号转义 + ass_filter_path = str(ass_path).replace("\\", "/").replace(":", "\\:") + filters.append(f"subtitles='{ass_filter_path}'") + + vf_str = ",".join(filters) + + command = [ + FFMPEG_BIN, + "-y", + "-i", + str(clip.local_path), + "-vf", + vf_str, + "-c:v", + "libx264", + "-crf", + "23", + "-preset", + "medium", + "-pix_fmt", + "yuv420p", + "-movflags", + "+faststart", + "-an", # 直通模式暂不处理音频，音频统一在后续混音阶段处理 + str(output_path), + ] + + logger.info( + "直通渲染: plan_id=%s clip=%s role=%s duration=%.2fs", + self.plan.id, + clip.clip_id, + role, + effective_duration, + ) + try: + run_ffmpeg(command) + except subprocess.CalledProcessError as e: + logger.error( + "直通渲染失败: plan_id=%s clip=%s exit_code=%d\nvf=%s", + self.plan.id, + clip.clip_id, + e.returncode, + vf_str[:2000], + ) + raise + # ── 内部方法 ────────────────────────────────────────────────────────────── def _resolve_clips(self) -> list[ResolvedClip]: @@ -277,9 +719,15 @@ class UnifiedRenderService: layers = sorted(layer_map.values(), key=lambda lyr: lyr.z_index) return layers - def _build_filter_complex(self, layers: list[RenderLayer]) -> tuple[str, list[str]]: + def _build_filter_complex( + self, layers: list[RenderLayer], *, ass_path: Path | None = None + ) -> tuple[str, list[str]]: """构建 FFmpeg filter_complex 字符串和输入参数列表。 + Args: + layers: 图层列表 + ass_path: ASS 字幕文件路径，有则在最后叠加字幕 + Returns: (filter_complex_str, input_args_list) input_args_list 是 ["-i", path1, "-i", path2, ...] 格式 @@ -335,11 +783,12 @@ class UnifiedRenderService: ) filters.append(f"crop={self.output_width}:{self.output_height}") else: - # main / broll: scale + pad 保持宽高比 + # main / broll: 铺满裁剪（scale to cover + center crop） + # 对齐链路A编辑器合成行为，与主流短视频平台一致 filters.append( - f"scale={self.output_width}:{self.output_height}" ":force_original_aspect_ratio=decrease" + f"scale={self.output_width}:{self.output_height}" ":force_original_aspect_ratio=increase" ) - filters.append(f"pad={self.output_width}:{self.output_height}" ":(ow-iw)/2:(oh-ih)/2:black") + filters.append(f"crop={self.output_width}:{self.output_height}") filters.append("setpts=PTS-STARTPTS") filters.append(f"fps={self.output_fps}") @@ -421,7 +870,12 @@ class UnifiedRenderService: filter_parts.append(f"[{final_video_label}][{overlay_label}]" f"overlay={x}:{y}[{combined_label}]") final_video_label = combined_label - filter_parts.append(f"[{final_video_label}]format=yuv420p[final_video]") + # 叠加字幕（如有）+ 最终像素格式 + if ass_path is not None: + ass_filter_path = str(ass_path).replace("\\", "/").replace(":", "\\:") + filter_parts.append(f"[{final_video_label}]subtitles='{ass_filter_path}',format=yuv420p[final_video]") + else: + filter_parts.append(f"[{final_video_label}]format=yuv420p[final_video]") filter_complex = ";".join(filter_parts) return filter_complex, input_args diff --git a/apps/worker/video_processing/video_compose_service.py b/apps/worker/video_processing/video_compose_service.py deleted file mode 100755 index 39f961075..000000000 --- a/apps/worker/video_processing/video_compose_service.py +++ /dev/null @@ -1,821 +0,0 @@ -""" -视频合成服务 -支持多种剪辑模式和转场效果，包含完整的安全校验 -""" - -import logging -import os -import subprocess -import tempfile -from dataclasses import dataclass -from enum import Enum - -try: - from enum import StrEnum -except ImportError: - - class StrEnum(str, Enum): # type: ignore[no-redef] - """Python 3.10 兼容的 StrEnum 回退实现。""" - - pass - - -from pathlib import Path -from typing import Optional - -from packages.domain.editing_mode import EditingMode - -logger = logging.getLogger(__name__) - -# ========== 安全常量 ========== -# 允许的输出目录白名单（使用环境变量或系统临时目录，避免硬编码 /tmp） -_VIDEO_OUTPUT_DIR = os.environ.get("VIDEO_OUTPUT_DIR", os.path.join(tempfile.gettempdir(), "video_output")) -ALLOWED_OUTPUT_DIRS = [_VIDEO_OUTPUT_DIR, "/var/app/rendered"] - -# 允许的输入路径前缀白名单 -ALLOWED_INPUT_PREFIXES = ("s3://", "oss://", "local://", "/var/storage/") - -# 允许的转场效果白名单 -ALLOWED_TRANSITIONS = { - "fade", - "slideleft", - "slideright", - "dissolve", - "wipeleft", - "wiperight", - "cut", - "slideup", - "slidedown", -} - -# 转场效果映射 -_XFADE_TRANSITION_MAP = { - "fade": "fade", - "slideleft": "slideleft", - "slideright": "slideright", - "dissolve": "dissolve", - "wipeleft": "wipeleft", - "wiperight": "wiperight", - "cut": "cut", - "slideup": "slideup", - "slidedown": "slidedown", -} - - -class VideoComposeError(Exception): - """视频合成服务异常""" - - pass - - -class PIPPosition(StrEnum): - """画中画位置枚举""" - - TOP_LEFT = "top_left" - TOP_RIGHT = "top_right" - BOTTOM_LEFT = "bottom_left" - BOTTOM_RIGHT = "bottom_right" - - -@dataclass -class Clip: - """视频片段""" - - asset_id: str # 资源ID，对应输入路径 - start_time: float = 0.0 - duration: float = 0.0 - transition: str = "fade" # 转场效果 - - -@dataclass -class EditingModeConfig: - """剪辑模式配置""" - - mode: EditingMode - output_width: int = 1280 - output_height: int = 720 - output_fps: int = 25 - pip_position: PIPPosition = PIPPosition.TOP_RIGHT - pip_scale: float = 0.25 # 画中画占主画面的比例 - transition_duration: float = 0.5 # 转场时长（秒） - output_codec: str = "libx264" - output_preset: str = "medium" - output_crf: int = 23 - - -class VideoComposeService: - """视频合成服务""" - - def __init__(self, config: EditingModeConfig, work_dir: Optional[str] = None): - """ - 初始化视频合成服务 - - Args: - config: 剪辑模式配置 - work_dir: 工作目录，默认使用系统临时目录 - """ - self.config = config - self.work_dir = work_dir or tempfile.gettempdir() - self._ffmpeg_bin = "ffmpeg" - self._ffprobe_bin = "ffprobe" - - def _validate_output_path(self, path: str) -> str: - """ - 校验输出路径是否在允许范围内 (P0 修复) - - 防止路径穿越攻击，如 /app/config/../../../etc/passwd - - Args: - path: 用户提供的输出路径 - - Returns: - 标准化后的绝对路径 - - Raises: - ValueError: 路径不在允许范围内 - """ - abs_path = os.path.abspath(path) - for allowed_dir in ALLOWED_OUTPUT_DIRS: - allowed_abs = os.path.abspath(allowed_dir) - if abs_path.startswith(allowed_abs): - return abs_path - raise ValueError(f"输出路径不在允许范围内: {path}") - - def _validate_input_path(self, path: str) -> bool: - """ - 校验输入路径格式是否合法 (P1-1 修复) - - Args: - path: 输入文件路径 - - Returns: - 是否合法 - """ - return any(path.startswith(prefix) for prefix in ALLOWED_INPUT_PREFIXES) - - def _validate_transition(self, transition: str) -> str: - """ - 校验转场效果是否在白名单内 (P1-2 修复) - - Args: - transition: 转场效果名称 - - Returns: - 安全的转场效果名称 - """ - if transition not in ALLOWED_TRANSITIONS: - logger.warning(f"未知的转场效果 '{transition}'，使用默认 'fade'") - return "fade" - return transition - - def _get_validated_transition(self, transition: str) -> str: - """获取白名单校验后的转场效果名称""" - return _XFADE_TRANSITION_MAP.get(self._validate_transition(transition), "fade") - - def compose(self, clips: list[Clip], output_path: Optional[str] = None) -> str: - """ - 合成视频 - - Args: - clips: 视频片段列表，每个片段包含 asset_id 和转场配置 - output_path: 输出文件路径 - - Returns: - 输出文件路径 - """ - if not clips: - raise ValueError("clips 不能为空") - - # P1-1: 校验所有输入路径 - for clip in clips: - if not self._validate_input_path(clip.asset_id): - raise ValueError(f"不合法的输入路径: {clip.asset_id}") - - # 生成默认输出路径并校验 - if output_path is None: - output_path = self._generate_output_path() - - # P0: 校验输出路径 - validated_output = self._validate_output_path(output_path) - - logger.info(f"合成视频，片段数: {len(clips)}, 输出: {validated_output}") - - # 获取输入路径列表 - input_paths = [clip.asset_id for clip in clips] - - try: - if self.config.mode == EditingMode.ONE_TAKE: - return self._one_take(input_paths, validated_output, clips) - elif self.config.mode == EditingMode.PIP: - return self._pip(input_paths, validated_output) - elif self.config.mode == EditingMode.VOICE_OVER: - return self._voice_over(input_paths, validated_output) - elif self.config.mode == EditingMode.VOICE_PIP: - return self._voice_pip(input_paths, validated_output) - else: - raise ValueError(f"不支持的剪辑模式: {self.config.mode}") - except Exception as e: - logger.error(f"视频合成失败: {e}") - raise VideoComposeError(f"视频合成失败: {e}") from e - - def _generate_output_path(self) -> str: - """生成输出文件路径""" - os.makedirs(self.work_dir, exist_ok=True) - return os.path.join(self.work_dir, f"output_{self.config.mode}_{os.getpid()}.mp4") - - def _validate_inputs(self, video_paths: list[str], audio_path: Optional[str] = None) -> None: - """验证输入文件存在""" - for path in video_paths: - if not os.path.exists(path): - raise FileNotFoundError(f"视频文件不存在: {path}") - if not os.path.getsize(path) > 0: - raise ValueError(f"视频文件为空: {path}") - - if audio_path and not os.path.exists(audio_path): - raise FileNotFoundError(f"音频文件不存在: {audio_path}") - - def _run_ffmpeg(self, command: list[str], capture_output: bool = True) -> tuple: - """执行 FFmpeg 命令""" - logger.debug(f"Running FFmpeg: {' '.join(command)}") - try: - result = subprocess.run( - command, - check=True, - stdout=subprocess.PIPE if capture_output else None, - stderr=subprocess.PIPE if capture_output else None, - text=capture_output, - ) - return result.stdout or "", result.stderr or "" - except subprocess.CalledProcessError as e: - stderr = e.stderr.decode() if e.stderr else str(e) - logger.error(f"FFmpeg error: {stderr}") - raise RuntimeError(f"FFmpeg 执行失败: {stderr}") from e - - def _get_video_info(self, video_path: str) -> dict: - """获取视频信息""" - try: - result = subprocess.run( - [ - self._ffprobe_bin, - "-v", - "error", - "-show_entries", - "stream=width,height,r_frame_rate,duration,codec_name", - "-show_entries", - "format=duration,size", - "-of", - "json", - video_path, - ], - check=True, - stdout=subprocess.PIPE, - stderr=subprocess.PIPE, - text=True, - ) - import json - - data = json.loads(result.stdout) - streams = data.get("streams", [{}]) - video_stream = next((s for s in streams if s.get("codec_type") == "video"), streams[0] if streams else {}) - fmt = data.get("format", {}) - - fps_str = video_stream.get("r_frame_rate", "25/1") - fps_parts = fps_str.split("/") - fps = float(fps_parts[0]) / float(fps_parts[1]) if len(fps_parts) == 2 else float(fps_parts[0]) - - return { - "width": int(video_stream.get("width", 0)), - "height": int(video_stream.get("height", 0)), - "fps": fps, - "duration": float(fmt.get("duration", 0)), - "codec": video_stream.get("codec_name", "unknown"), - "size": int(fmt.get("size", 0)), - } - except Exception as e: - logger.warning(f"获取视频信息失败 {video_path}: {e}") - return {"width": 0, "height": 0, "fps": 25, "duration": 0, "codec": "unknown", "size": 0} - - def _get_pip_position_offset( - self, main_width: int, main_height: int, pip_width: int, pip_height: int - ) -> tuple[int, int]: - """获取画中画位置偏移量""" - margin = 10 - position_offsets = { - PIPPosition.TOP_LEFT: (margin, margin), - PIPPosition.TOP_RIGHT: (main_width - pip_width - margin, margin), - PIPPosition.BOTTOM_LEFT: (margin, main_height - pip_height - margin), - PIPPosition.BOTTOM_RIGHT: (main_width - pip_width - margin, main_height - pip_height - margin), - } - return position_offsets.get(self.config.pip_position, position_offsets[PIPPosition.TOP_RIGHT]) - - def _normalize_video(self, input_path: str, output_path: str) -> dict: - """标准化视频格式""" - command = [ - self._ffmpeg_bin, - "-y", - "-i", - input_path, - "-r", - str(self.config.output_fps), - "-vf", - f"scale={self.config.output_width}:{self.config.output_height}:force_original_aspect_ratio=decrease,pad={self.config.output_width}:{self.config.output_height}:(ow-iw)/2:(oh-ih)/2,setsar=1", - "-r", - str(self.config.output_fps), - "-c:v", - self.config.output_codec, - "-preset", - self.config.output_preset, - "-crf", - str(self.config.output_crf), - "-pix_fmt", - "yuv420p", - "-movflags", - "+faststart", - "-an", - output_path, - ] - self._run_ffmpeg(command) - return self._get_video_info(output_path) - - def _one_take(self, video_paths: list[str], output_path: str, clips: list[Clip]) -> str: - """一镜到底模式""" - if len(video_paths) == 1: - return self._normalize_video(video_paths[0], output_path) - - normalized_paths = [] - for i, path in enumerate(video_paths): - normalized = os.path.join(self.work_dir, f"normalized_{i}_{os.getpid()}.mp4") - self._normalize_video(path, normalized) - normalized_paths.append(normalized) - - durations = [self._get_video_info(p)["duration"] for p in normalized_paths] - - if len(normalized_paths) <= 5: - output_path = self._one_take_with_xfade(normalized_paths, durations, output_path, clips) - else: - output_path = self._one_take_simple_concat(normalized_paths, output_path) - - for p in normalized_paths: - try: - if p != output_path: - os.remove(p) - except Exception as e: - logger.warning( - f"Operation failed in apps/worker/video_processing/video_compose_service.py: {e}", exc_info=True - ) - - return output_path - - def _one_take_with_xfade( - self, normalized_paths: list[str], durations: list[float], output_path: str, clips: list[Clip] - ) -> str: - """使用 xfade 滤镜实现转场 (P1-2: 转场参数白名单校验)""" - if len(normalized_paths) == 2: - # 获取当前片段的转场效果并校验白名单 - transition = "fade" - if len(clips) > 1: - transition = self._get_validated_transition(clips[1].transition) - - trans_duration = self.config.transition_duration - offset1 = durations[0] - trans_duration / 2 - - command = [ - self._ffmpeg_bin, - "-y", - "-i", - normalized_paths[0], - "-i", - normalized_paths[1], - "-filter_complex", - f"[0:v][1:v]xfade=transition={transition}:duration={trans_duration}:offset={offset1}[v]", - "-map", - "[v]", - "-c:v", - self.config.output_codec, - "-preset", - self.config.output_preset, - "-crf", - str(self.config.output_crf), - "-pix_fmt", - "yuv420p", - output_path, - ] - self._run_ffmpeg(command) - return output_path - else: - return self._one_take_simple_concat(normalized_paths, output_path) - - def _one_take_simple_concat(self, normalized_paths: list[str], output_path: str) -> str: - """使用 concat demuxer 简单拼接""" - concat_file = os.path.join(self.work_dir, f"concat_list_{os.getpid()}.txt") - with open(concat_file, "w") as f: - for path in normalized_paths: - f.write(f"file '{os.path.abspath(path)}'\n") - - command = [ - self._ffmpeg_bin, - "-y", - "-f", - "concat", - "-safe", - "0", - "-i", - concat_file, - "-c", - "copy", - output_path, - ] - self._run_ffmpeg(command) - - try: - os.remove(concat_file) - except Exception as e: - logger.warning( - f"Operation failed in apps/worker/video_processing/video_compose_service.py: {e}", exc_info=True - ) - - return output_path - - def _pip(self, video_paths: list[str], output_path: str) -> str: - """画中画模式""" - if not video_paths: - raise ValueError("No video paths provided") - - main_video = video_paths[0] - main_normalized = os.path.join(self.work_dir, f"main_{os.getpid()}.mp4") - main_info = self._normalize_video(main_video, main_normalized) - - if len(video_paths) == 1: - os.rename(main_normalized, output_path) - return output_path - - pip_width = int(self.config.output_width * self.config.pip_scale) - pip_height = int(self.config.output_height * self.config.pip_scale) - x_offset, y_offset = self._get_pip_position_offset( - self.config.output_width, self.config.output_height, pip_width, pip_height - ) - - pip_normalized = os.path.join(self.work_dir, f"pip_{os.getpid()}.mp4") - pip_info = self._get_video_info(video_paths[1]) - - if pip_info["duration"] > main_info["duration"]: - temp_pip = os.path.join(self.work_dir, f"pip_temp_{os.getpid()}.mp4") - command = [ - self._ffmpeg_bin, - "-y", - "-i", - video_paths[1], - "-t", - str(main_info["duration"]), - "-vf", - f"scale={pip_width}:{pip_height}", - "-c:v", - self.config.output_codec, - "-preset", - self.config.output_preset, - "-crf", - str(self.config.output_crf), - "-pix_fmt", - "yuv420p", - temp_pip, - ] - self._run_ffmpeg(command) - pip_normalized_input = temp_pip - else: - command = [ - self._ffmpeg_bin, - "-y", - "-i", - video_paths[1], - "-vf", - f"scale={pip_width}:{pip_height}", - "-c:v", - self.config.output_codec, - "-preset", - self.config.output_preset, - "-crf", - str(self.config.output_crf), - "-pix_fmt", - "yuv420p", - pip_normalized, - ] - self._run_ffmpeg(command) - pip_normalized_input = pip_normalized - - if main_info["duration"] > pip_info["duration"]: - looped_pip = os.path.join(self.work_dir, f"pip_looped_{os.getpid()}.mp4") - command = [ - self._ffmpeg_bin, - "-y", - "-stream_loop", - "-1", - "-i", - pip_normalized_input, - "-t", - str(main_info["duration"]), - "-vf", - f"scale={pip_width}:{pip_height}", - "-c:v", - self.config.output_codec, - "-preset", - self.config.output_preset, - "-crf", - str(self.config.output_crf), - "-pix_fmt", - "yuv420p", - looped_pip, - ] - self._run_ffmpeg(command) - pip_normalized_input = looped_pip - - command = [ - self._ffmpeg_bin, - "-y", - "-i", - main_normalized, - "-i", - pip_normalized_input, - "-filter_complex", - f"[0:v][1:v]overlay={x_offset}:{y_offset}[v]", - "-map", - "[v]", - "-c:v", - self.config.output_codec, - "-preset", - self.config.output_preset, - "-crf", - str(self.config.output_crf), - "-pix_fmt", - "yuv420p", - output_path, - ] - self._run_ffmpeg(command) - - for temp_file in [main_normalized, pip_normalized]: - if temp_file and temp_file != output_path: - try: - os.remove(temp_file) - except Exception as e: - logger.warning( - f"Operation failed in apps/worker/video_processing/video_compose_service.py: {e}", exc_info=True - ) - - return output_path - - def _voice_over(self, video_paths: list[str], audio_path: str, output_path: str) -> str: - """口播模式""" - if not audio_path: - raise ValueError("audio_path is required for VOICE_OVER mode") - - if not video_paths: - raise ValueError("No background video provided") - - audio_info = self._get_video_info(audio_path) - audio_duration = audio_info["duration"] - - bg_normalized = os.path.join(self.work_dir, f"bg_{os.getpid()}.mp4") - bg_info = self._normalize_video(video_paths[0], bg_normalized) - - if bg_info["duration"] < audio_duration: - looped_bg = os.path.join(self.work_dir, f"bg_looped_{os.getpid()}.mp4") - command = [ - self._ffmpeg_bin, - "-y", - "-stream_loop", - "-1", - "-i", - bg_normalized, - "-t", - str(audio_duration), - "-vf", - f"scale={self.config.output_width}:{self.config.output_height}", - "-c:v", - self.config.output_codec, - "-preset", - self.config.output_preset, - "-crf", - str(self.config.output_crf), - "-pix_fmt", - "yuv420p", - looped_bg, - ] - self._run_ffmpeg(command) - bg_normalized = looped_bg - elif bg_info["duration"] > audio_duration: - temp_bg = os.path.join(self.work_dir, f"bg_trimmed_{os.getpid()}.mp4") - command = [ - self._ffmpeg_bin, - "-y", - "-i", - bg_normalized, - "-t", - str(audio_duration), - "-c:v", - "copy", - temp_bg, - ] - self._run_ffmpeg(command) - bg_normalized = temp_bg - - blurred_bg = os.path.join(self.work_dir, f"bg_blurred_{os.getpid()}.mp4") - command = [ - self._ffmpeg_bin, - "-y", - "-i", - bg_normalized, - "-vf", - f"boxblur=5:5,scale={self.config.output_width}:{self.config.output_height}", - "-c:v", - self.config.output_codec, - "-preset", - self.config.output_preset, - "-crf", - str(self.config.output_crf), - "-pix_fmt", - "yuv420p", - blurred_bg, - ] - self._run_ffmpeg(command) - - command = [ - self._ffmpeg_bin, - "-y", - "-i", - blurred_bg, - "-i", - audio_path, - "-filter_complex", - "[0:v]drawbox=x=0:y=0:w=iw:h=ih:color=black@0.3:t=fill[v]", - "-map", - "[v]", - "-map", - "1:a", - "-c:v", - self.config.output_codec, - "-preset", - self.config.output_preset, - "-crf", - str(self.config.output_crf), - "-pix_fmt", - "yuv420p", - "-shortest", - output_path, - ] - self._run_ffmpeg(command) - - for temp_file in [bg_normalized, blurred_bg]: - try: - if temp_file != output_path: - os.remove(temp_file) - except Exception as e: - logger.warning( - f"Operation failed in apps/worker/video_processing/video_compose_service.py: {e}", exc_info=True - ) - - return output_path - - def _voice_pip(self, video_paths: list[str], audio_path: Optional[str], output_path: str) -> str: - """口播+画中画模式""" - if not video_paths: - raise ValueError("No video paths provided") - - if len(video_paths) == 1: - return self._normalize_video(video_paths[0], output_path) - - voice_video = video_paths[0] - bg_video = video_paths[1] if len(video_paths) > 1 else video_paths[0] - - voice_normalized = os.path.join(self.work_dir, f"voice_{os.getpid()}.mp4") - voice_info = self._normalize_video(voice_video, voice_normalized) - - bg_normalized = os.path.join(self.work_dir, f"bg_{os.getpid()}.mp4") - bg_info = self._normalize_video(bg_video, bg_normalized) - - final_duration = min(voice_info["duration"], bg_info["duration"]) - - pip_width = int(self.config.output_width * self.config.pip_scale) - pip_height = int(self.config.output_height * self.config.pip_scale) - x_offset, y_offset = self._get_pip_position_offset( - self.config.output_width, self.config.output_height, pip_width, pip_height - ) - - voice_adjusted = os.path.join(self.work_dir, f"voice_adj_{os.getpid()}.mp4") - command = [ - self._ffmpeg_bin, - "-y", - "-i", - voice_normalized, - "-t", - str(final_duration), - "-vf", - f"scale={pip_width}:{pip_height}", - "-c:v", - self.config.output_codec, - "-preset", - self.config.output_preset, - "-crf", - str(self.config.output_crf), - "-pix_fmt", - "yuv420p", - voice_adjusted, - ] - self._run_ffmpeg(command) - - bg_adjusted = os.path.join(self.work_dir, f"bg_adj_{os.getpid()}.mp4") - command = [ - self._ffmpeg_bin, - "-y", - "-i", - bg_normalized, - "-t", - str(final_duration), - "-c:v", - "copy", - bg_adjusted, - ] - self._run_ffmpeg(command) - - if audio_path: - command = [ - self._ffmpeg_bin, - "-y", - "-i", - bg_adjusted, - "-i", - voice_adjusted, - "-i", - audio_path, - "-filter_complex", - f"[0:v][1:v]overlay={x_offset}:{y_offset}[v]", - "-map", - "[v]", - "-map", - "2:a", - "-shortest", - "-c:v", - self.config.output_codec, - "-preset", - self.config.output_preset, - "-crf", - str(self.config.output_crf), - "-pix_fmt", - "yuv420p", - output_path, - ] - else: - command = [ - self._ffmpeg_bin, - "-y", - "-i", - bg_adjusted, - "-i", - voice_adjusted, - "-filter_complex", - f"[0:v][1:v]overlay={x_offset}:{y_offset}[v]", - "-map", - "[v]", - "-map", - "1:a", - "-shortest", - "-c:v", - self.config.output_codec, - "-preset", - self.config.output_preset, - "-crf", - str(self.config.output_crf), - "-pix_fmt", - "yuv420p", - output_path, - ] - self._run_ffmpeg(command) - - for temp_file in [voice_normalized, voice_adjusted, bg_normalized, bg_adjusted]: - try: - if temp_file != output_path: - os.remove(temp_file) - except Exception as e: - logger.warning( - f"Operation failed in apps/worker/video_processing/video_compose_service.py: {e}", exc_info=True - ) - - return output_path - - -def create_compose_service(mode: str, work_dir: Optional[str] = None, **kwargs) -> VideoComposeService: - """便捷工厂函数：创建视频合成服务""" - try: - editing_mode = EditingMode(mode) - except ValueError: - raise ValueError(f"无效的剪辑模式: {mode}. 有效模式: {[m.value for m in EditingMode]}") - - config = EditingModeConfig( - mode=editing_mode, - output_width=kwargs.get("output_width", 1280), - output_height=kwargs.get("output_height", 720), - output_fps=kwargs.get("output_fps", 25), - pip_position=PIPPosition(kwargs.get("pip_position", "top_right")), - pip_scale=kwargs.get("pip_scale", 0.25), - transition_duration=kwargs.get("transition_duration", 0.5), - ) - - return VideoComposeService(config=config, work_dir=work_dir) diff --git a/apps/worker/worker_app/tasks/generation.py b/apps/worker/worker_app/tasks/generation.py index c42cbbdb4..7c0824d24 100755 --- a/apps/worker/worker_app/tasks/generation.py +++ b/apps/worker/worker_app/tasks/generation.py @@ -124,6 +124,7 @@ class _VirtualPlan: id: str name: str = "" + config: dict[str, Any] = field(default_factory=dict) @dataclass diff --git a/tests/unit/test_unified_render_service.py b/tests/unit/test_unified_render_service.py index 5eed6f78f..2ed2d4dad 100755 --- a/tests/unit/test_unified_render_service.py +++ b/tests/unit/test_unified_render_service.py @@ -16,7 +16,10 @@ from video_processing.unified_render_service import ( RenderResult, ResolvedClip, UnifiedRenderService, + _hex_to_ass_color, + _position_to_ass_alignment, _resolve_layer_role, + generate_ass_subtitles, ) # ── Fixtures ────────────────────────────────────────────────────────────────── @@ -45,6 +48,7 @@ class FakePlan: id: str = "plan_001" name: str = "测试计划" + config: dict[str, Any] = field(default_factory=dict) def _make_clip( @@ -438,6 +442,56 @@ class TestBuildFilterComplex: f"单视频: setpts(position={last_setpts}) 应该在 fps(position={first_fps}) 之前。" f"滤镜链: {chain_str}" ) + def test_main_clip_uses_fill_crop_strategy(self): + """main/broll clip 使用铺满裁剪策略（scale increase + crop），不是等比+黑边。 + + 对齐链路A编辑器合成行为，与主流短视频平台一致。 + """ + clips = [_make_clip("c1", "main", order=0, duration=5.0)] + asset_paths = {"asset_c1.mp4": Path("/tmp/asset_c1.mp4")} + svc = _make_service(clips, asset_paths) + + with _patch_path_exists(), patch("video_processing.unified_render_service.probe_duration", return_value=5.0): + resolved = svc._resolve_clips() + layers = svc._group_clips_into_layers(resolved) + fc, _ = svc._build_filter_complex(layers) + + # 验证：scale 使用 force_original_aspect_ratio=increase（铺满） + assert "force_original_aspect_ratio=increase" in fc + # 验证：有 crop（居中裁剪） + assert "crop=1280:720" in fc + # 验证：没有 pad（不是黑边模式） + assert "pad=" not in fc + + def test_broll_clip_uses_fill_crop_strategy(self): + """broll clip 同样使用铺满裁剪策略。""" + clips = [_make_clip("c1", "b_roll", order=0, duration=5.0)] + asset_paths = {"asset_c1.mp4": Path("/tmp/asset_c1.mp4")} + svc = _make_service(clips, asset_paths) + + with _patch_path_exists(), patch("video_processing.unified_render_service.probe_duration", return_value=5.0): + resolved = svc._resolve_clips() + layers = svc._group_clips_into_layers(resolved) + fc, _ = svc._build_filter_complex(layers) + + assert "force_original_aspect_ratio=increase" in fc + assert "crop=1280:720" in fc + assert "pad=" not in fc + + def test_background_uses_fill_crop_strategy(self): + """background 层也使用铺满裁剪（已有的行为，保持一致）。""" + clips = [_make_clip("c1", "background", order=0, duration=5.0)] + asset_paths = {"asset_c1.mp4": Path("/tmp/asset_c1.mp4")} + svc = _make_service(clips, asset_paths) + + with _patch_path_exists(), patch("video_processing.unified_render_service.probe_duration", return_value=5.0): + resolved = svc._resolve_clips() + layers = svc._group_clips_into_layers(resolved) + fc, _ = svc._build_filter_complex(layers) + + assert "force_original_aspect_ratio=increase" in fc + assert "crop=1280:720" in fc + def test_empty_layers_raises(self): """空图层列表抛出 ValueError。""" svc = _make_service() @@ -445,6 +499,294 @@ class TestBuildFilterComplex: svc._build_filter_complex([]) +class TestPassThrough: + """测试单图层单 clip 直通优化路径。""" + + def test_can_use_pass_through_single_main_clip(self): + """1个main图层 + 1个clip → 可以直通。""" + clips = [_make_clip("c1", "main", order=0, duration=5.0)] + svc = _make_service(clips) + with _patch_path_exists(), patch("video_processing.unified_render_service.probe_duration", return_value=5.0): + resolved = svc._resolve_clips() + layers = svc._group_clips_into_layers(resolved) + assert svc._can_use_pass_through(layers) is True + + def test_can_use_pass_through_single_broll_clip(self): + """1个broll图层 + 1个clip → 可以直通。""" + clips = [_make_clip("c1", "b_roll", order=0, duration=5.0)] + svc = _make_service(clips) + with _patch_path_exists(), patch("video_processing.unified_render_service.probe_duration", return_value=5.0): + resolved = svc._resolve_clips() + layers = svc._group_clips_into_layers(resolved) + assert svc._can_use_pass_through(layers) is True + + def test_can_use_pass_through_single_background_clip(self): + """1个background图层 + 1个clip → 可以直通。""" + clips = [_make_clip("c1", "background", order=0, duration=5.0)] + svc = _make_service(clips) + with _patch_path_exists(), patch("video_processing.unified_render_service.probe_duration", return_value=5.0): + resolved = svc._resolve_clips() + layers = svc._group_clips_into_layers(resolved) + assert svc._can_use_pass_through(layers) is True + + def test_cannot_pass_through_multi_clips(self): + """1个图层 + 多个clips → 不能直通（需要xfade）。""" + clips = [ + _make_clip("c1", "main", order=0, duration=3.0), + _make_clip("c2", "main", order=1, duration=3.0), + ] + svc = _make_service(clips) + with _patch_path_exists(), patch("video_processing.unified_render_service.probe_duration", return_value=5.0): + resolved = svc._resolve_clips() + layers = svc._group_clips_into_layers(resolved) + assert svc._can_use_pass_through(layers) is False + + def test_cannot_pass_through_multi_layers(self): + """多个图层 → 不能直通。""" + clips = [ + _make_clip("c1", "main", order=0, duration=5.0), + _make_clip("c2", "overlay", order=1, duration=5.0), + ] + svc = _make_service(clips) + with _patch_path_exists(), patch("video_processing.unified_render_service.probe_duration", return_value=5.0): + resolved = svc._resolve_clips() + layers = svc._group_clips_into_layers(resolved) + assert svc._can_use_pass_through(layers) is False + + def test_cannot_pass_through_overlay_only(self): + """只有overlay图层 → 不能直通（需要叠加到主层）。""" + clips = [_make_clip("c1", "overlay", order=0, duration=5.0)] + svc = _make_service(clips) + with _patch_path_exists(), patch("video_processing.unified_render_service.probe_duration", return_value=5.0): + resolved = svc._resolve_clips() + layers = svc._group_clips_into_layers(resolved) + assert svc._can_use_pass_through(layers) is False + + def test_render_uses_pass_through_for_single_clip(self): + """单clip渲染时走直通路径（调用_render_pass_through而非_execute_ffmpeg）。""" + clips = [_make_clip("c1", "main", order=0, duration=5.0)] + asset_paths = {"asset_c1.mp4": Path("/tmp/asset_c1.mp4")} + svc = _make_service(clips, asset_paths) + + with ( + _patch_path_exists(), + patch("video_processing.unified_render_service.probe_duration", return_value=5.0), + patch.object(svc, "_render_pass_through") as mock_pass, + patch.object(svc, "_execute_ffmpeg") as mock_exec, + patch.object(svc, "_probe_output", return_value=(5.0, 1024, 1280, 720)), + ): + result = svc.render() + + mock_pass.assert_called_once() + mock_exec.assert_not_called() + assert result.duration == 5.0 + + def test_render_uses_filter_complex_for_multi_clips(self): + """多clip渲染时走完整filter_complex路径。""" + clips = [ + _make_clip("c1", "main", order=0, duration=3.0), + _make_clip("c2", "main", order=1, duration=3.0), + ] + asset_paths = { + "asset_c1.mp4": Path("/tmp/asset_c1.mp4"), + "asset_c2.mp4": Path("/tmp/asset_c2.mp4"), + } + svc = _make_service(clips, asset_paths) + + with ( + _patch_path_exists(), + patch("video_processing.unified_render_service.probe_duration", return_value=5.0), + patch.object(svc, "_render_pass_through") as mock_pass, + patch.object(svc, "_execute_ffmpeg") as mock_exec, + patch.object(svc, "_probe_output", return_value=(5.5, 2048, 1280, 720)), + ): + result = svc.render() + + mock_pass.assert_not_called() + mock_exec.assert_called_once() + assert result.duration == 5.5 + + +# ── 测试 ASS 字幕生成 ──────────────────────────────────────────────────────── + + +class TestAssSubtitles: + """测试 ASS 字幕生成功能。""" + + def test_hex_to_ass_color_white(self): + """#ffffff → &HFFFFFF（ASS 是 BGR 顺序）。""" + assert _hex_to_ass_color("#ffffff") == "&HFFFFFF" + + def test_hex_to_ass_color_black(self): + """#000000 → &H000000。""" + assert _hex_to_ass_color("#000000") == "&H000000" + + def test_hex_to_ass_color_red(self): + """#ff0000 红 → &H0000FF（B=00, G=00, R=FF）。""" + assert _hex_to_ass_color("#ff0000") == "&H0000FF" + + def test_hex_to_ass_color_blue(self): + """#0000ff 蓝 → &HFF0000（B=FF, G=00, R=00）。""" + assert _hex_to_ass_color("#0000ff") == "&HFF0000" + + def test_hex_to_ass_color_no_hash(self): + """不带 # 的颜色值也能解析。""" + assert _hex_to_ass_color("ff0000") == "&H0000FF" + + def test_position_to_ass_alignment_top(self): + """top → 8（顶部居中）。""" + assert _position_to_ass_alignment("top") == 8 + + def test_position_to_ass_alignment_center(self): + """center → 5（居中）。""" + assert _position_to_ass_alignment("center") == 5 + + def test_position_to_ass_alignment_bottom(self): + """bottom → 2（底部居中）。""" + assert _position_to_ass_alignment("bottom") == 2 + + def test_generate_ass_with_title_only(self, tmp_path): + """只有标题时生成 ASS 文件。""" + ass_path = tmp_path / "test.ass" + result = generate_ass_subtitles( + ass_path, + video_width=1280, + video_height=720, + video_duration=10.0, + title_text="测试标题", + title_config={ + "enabled": True, + "font": "思源黑体", + "size": 48, + "color": "#ffffff", + "bold": True, + "position": "top", + "stroke": {"enabled": True, "color": "#000000", "width": 2}, + "shadow": {"enabled": True, "blur": 4, "offset_x": 2, "offset_y": 2}, + }, + ) + + assert result == ass_path + content = ass_path.read_text(encoding="utf-8") + assert "[Script Info]" in content + assert "PlayResX: 1280" in content + assert "PlayResY: 720" in content + assert "[V4+ Styles]" in content + assert "TitleStyle" in content + assert "测试标题" in content + assert "Dialogue:" in content + + def test_generate_ass_with_subtitle_only(self, tmp_path): + """只有字幕时生成 ASS 文件。""" + ass_path = tmp_path / "test.ass" + result = generate_ass_subtitles( + ass_path, + video_width=1280, + video_height=720, + video_duration=10.0, + subtitle_text="测试字幕内容", + subtitle_config={ + "enabled": True, + "font": "思源黑体", + "size": 24, + "color": "#ffffff", + "position": "bottom", + }, + ) + + content = ass_path.read_text(encoding="utf-8") + assert "SubtitleStyle" in content + assert "测试字幕内容" in content + assert "Dialogue:" in content + + def test_generate_ass_with_both_title_and_subtitle(self, tmp_path): + """同时有标题和字幕。""" + ass_path = tmp_path / "test.ass" + generate_ass_subtitles( + ass_path, + video_width=1280, + video_height=720, + video_duration=10.0, + title_text="大标题", + title_config={"enabled": True, "position": "top"}, + subtitle_text="底部字幕", + subtitle_config={"enabled": True, "position": "bottom"}, + ) + + content = ass_path.read_text(encoding="utf-8") + assert "TitleStyle" in content + assert "SubtitleStyle" in content + assert "大标题" in content + assert "底部字幕" in content + # 两条 Dialogue 行 + assert content.count("Dialogue:") == 2 + + def test_generate_ass_disabled_returns_empty(self, tmp_path): + """标题和字幕都禁用时返回空文件。""" + ass_path = tmp_path / "test.ass" + generate_ass_subtitles( + ass_path, + video_width=1280, + video_height=720, + video_duration=10.0, + title_text="不显示", + title_config={"enabled": False}, + subtitle_text="也不显示", + subtitle_config={"enabled": False}, + ) + + content = ass_path.read_text(encoding="utf-8") + assert content == "" + + def test_generate_ass_empty_text_returns_empty(self, tmp_path): + """文本为空时不生成字幕。""" + ass_path = tmp_path / "test.ass" + generate_ass_subtitles( + ass_path, + video_width=1280, + video_height=720, + video_duration=10.0, + title_text="", + title_config={"enabled": True}, + subtitle_text=" ", + subtitle_config={"enabled": True}, + ) + + content = ass_path.read_text(encoding="utf-8") + assert content == "" + + def test_generate_ass_time_format(self, tmp_path): + """验证 ASS 时间格式正确（H:MM:SS.cc）。""" + ass_path = tmp_path / "test.ass" + generate_ass_subtitles( + ass_path, + video_width=1280, + video_height=720, + video_duration=125.5, # 2分5.5秒 + title_text="测试", + title_config={"enabled": True}, + ) + + content = ass_path.read_text(encoding="utf-8") + # 结束时间应该是 0:02:05.50 + assert "0:02:05.50" in content + + def test_ass_text_escape_newlines(self, tmp_path): + """换行符转义为 ASS 的 \\N。""" + ass_path = tmp_path / "test.ass" + generate_ass_subtitles( + ass_path, + video_width=1280, + video_height=720, + video_duration=10.0, + title_text="第一行\n第二行", + title_config={"enabled": True}, + ) + + content = ass_path.read_text(encoding="utf-8") + assert "第一行\\N第二行" in content + + # ── 测试 render 方法 ───────────────────────────────────────────────────────── @@ -464,8 +806,8 @@ class TestRender: with pytest.raises(ValueError, match="没有可渲染的片段"): svc.render() - def test_render_success(self): - """正常渲染流程。""" + def test_render_success_single_clip(self): + """单clip正常渲染（走直通路径）。""" clips = [_make_clip("c1", "main", order=0)] asset_paths = {"asset_c1.mp4": Path("/tmp/asset_c1.mp4")} svc = _make_service(clips, asset_paths) @@ -473,7 +815,7 @@ class TestRender: with ( _patch_path_exists(), patch("video_processing.unified_render_service.probe_duration", return_value=5.0), - patch.object(svc, "_execute_ffmpeg") as mock_exec, + patch.object(svc, "_render_pass_through") as mock_pass, patch.object(svc, "_probe_output", return_value=(5.0, 1024, 1280, 720)), ): result = svc.render() @@ -483,4 +825,31 @@ class TestRender: assert result.file_size == 1024 assert result.width == 1280 assert result.height == 720 + mock_pass.assert_called_once() + + def test_render_success_multi_clips(self): + """多clip正常渲染（走完整filter_complex路径）。""" + clips = [ + _make_clip("c1", "main", order=0, duration=3.0), + _make_clip("c2", "main", order=1, duration=3.0), + ] + asset_paths = { + "asset_c1.mp4": Path("/tmp/asset_c1.mp4"), + "asset_c2.mp4": Path("/tmp/asset_c2.mp4"), + } + svc = _make_service(clips, asset_paths) + + with ( + _patch_path_exists(), + patch("video_processing.unified_render_service.probe_duration", return_value=5.0), + patch.object(svc, "_execute_ffmpeg") as mock_exec, + patch.object(svc, "_probe_output", return_value=(5.5, 2048, 1280, 720)), + ): + result = svc.render() + + assert isinstance(result, RenderResult) + assert result.duration == 5.5 + assert result.file_size == 2048 + assert result.width == 1280 + assert result.height == 720 mock_exec.assert_called_once() diff --git a/tests/unit/test_video_compose_security.py b/tests/unit/test_video_compose_security.py deleted file mode 100755 index f3ace4fda..000000000 --- a/tests/unit/test_video_compose_security.py +++ /dev/null @@ -1,247 +0,0 @@ -""" -视频合成服务安全校验单元测试 -针对 PR #159 安全审计发现的问题进行测试 -""" - -import os -import tempfile -from unittest.mock import MagicMock, patch - -import pytest - -# 导入被测试的模块 -from apps.worker.video_processing.video_compose_service import ( - ALLOWED_INPUT_PREFIXES, - ALLOWED_OUTPUT_DIRS, - ALLOWED_TRANSITIONS, - Clip, - EditingMode, - EditingModeConfig, - VideoComposeService, -) - - -class TestOutputPathValidation: - """P0: 输出路径穿越校验测试""" - - def setup_method(self): - """测试前设置""" - self.config = EditingModeConfig(mode=EditingMode.ONE_TAKE) - self.service = VideoComposeService(self.config) - - def test_valid_output_path_in_allowed_dir(self): - """测试合法的输出路径""" - valid_path = "/tmp/video_output/test.mp4" - result = self.service._validate_output_path(valid_path) - assert result == os.path.abspath(valid_path) - - def test_valid_output_path_with_relative_components(self): - """测试带相对路径成分但最终在允许目录内的路径""" - valid_path = "/tmp/video_output/subdir/../test.mp4" - result = self.service._validate_output_path(valid_path) - assert result == os.path.abspath(valid_path) - - def test_path_traversal_attack_blocked(self): - """测试路径穿越攻击被阻止""" - # 尝试穿越到 /etc/passwd - malicious_path = "/tmp/video_output/../../../etc/passwd" - with pytest.raises(ValueError, match="输出路径不在允许范围内"): - self.service._validate_output_path(malicious_path) - - def test_path_traversal_attack_blocked_var_app(self): - """测试针对 /var/app 的路径穿越攻击被阻止""" - malicious_path = "/var/app/rendered/../../config/../../../etc/passwd" - with pytest.raises(ValueError, match="输出路径不在允许范围内"): - self.service._validate_output_path(malicious_path) - - def test_absolute_path_to_forbidden_location(self): - """测试直接访问禁止位置""" - forbidden_path = "/etc/shadow" - with pytest.raises(ValueError, match="输出路径不在允许范围内"): - self.service._validate_output_path(forbidden_path) - - def test_root_path_blocked(self): - """测试根目录被阻止""" - root_path = "/" - with pytest.raises(ValueError, match="输出路径不在允许范围内"): - self.service._validate_output_path(root_path) - - def test_absolute_path_to_tmp_not_allowed(self): - """测试 /tmp 不在白名单中时应被阻止""" - # /tmp 不在 ALLOWED_OUTPUT_DIRS 中 - tmp_path = "/tmp/test.mp4" - with pytest.raises(ValueError, match="输出路径不在允许范围内"): - self.service._validate_output_path(tmp_path) - - -class TestInputPathValidation: - """P1-1: 输入路径格式校验测试""" - - def setup_method(self): - """测试前设置""" - self.config = EditingModeConfig(mode=EditingMode.ONE_TAKE) - self.service = VideoComposeService(self.config) - - def test_valid_s3_path(self): - """测试 S3 路径""" - assert self.service._validate_input_path("s3://bucket/key.mp4") is True - - def test_valid_oss_path(self): - """测试 OSS 路径""" - assert self.service._validate_input_path("oss://bucket/key.mp4") is True - - def test_valid_local_path(self): - """测试 local:// 路径""" - assert self.service._validate_input_path("local://asset/123.mp4") is True - - def test_valid_var_storage_path(self): - """测试 /var/storage/ 路径""" - assert self.service._validate_input_path("/var/storage/assets/123.mp4") is True - - def test_path_traversal_in_input_rejected(self): - """测试输入路径中的路径穿越尝试被拒绝""" - malicious_path = "s3://bucket/../../etc/passwd" - # 这会通过前缀检查，但实际使用时文件系统访问会失败 - # 安全设计：只校验格式前缀 - assert self.service._validate_input_path(malicious_path) is True - - def test_malicious_input_path_blocked(self): - """测试恶意输入路径被阻止""" - assert self.service._validate_input_path("/etc/passwd") is False - assert self.service._validate_input_path("file:///etc/passwd") is False - assert self.service._validate_input_path("http://evil.com/shell.sh") is False - - def test_empty_path_rejected(self): - """测试空路径被拒绝""" - assert self.service._validate_input_path("") is False - - def test_random_string_rejected(self): - """测试随机字符串被拒绝""" - assert self.service._validate_input_path("random123") is False - assert self.service._validate_input_path("abc../../../etc") is False - - -class TestTransitionValidation: - """P1-2: 转场参数白名单校验测试""" - - def setup_method(self): - """测试前设置""" - self.config = EditingModeConfig(mode=EditingMode.ONE_TAKE) - self.service = VideoComposeService(self.config) - - @pytest.mark.parametrize("transition", list(ALLOWED_TRANSITIONS)) - def test_valid_transitions(self, transition): - """测试所有合法的转场效果""" - result = self.service._validate_transition(transition) - assert result == transition - - def test_invalid_transition_defaults_to_fade(self): - """测试非法转场效果默认为 fade""" - result = self.service._validate_transition("random_transition") - assert result == "fade" - - def test_sql_injection_in_transition_blocked(self): - """测试 SQL 注入尝试被阻止""" - result = self.service._validate_transition("fade; DROP TABLE videos;--") - assert result == "fade" - - def test_shell_injection_in_transition_blocked(self): - """测试 Shell 注入尝试被阻止""" - result = self.service._validate_transition("fade$(whoami)") - assert result == "fade" - - def test_empty_transition_handled(self): - """测试空转场名称""" - result = self.service._validate_transition("") - assert result == "fade" - - def test_none_transition_handled(self): - """测试 None 转场名称""" - result = self.service._validate_transition(None) - assert result == "fade" - - def test_get_validated_transition_returns_mapped(self): - """测试 _get_validated_transition 返回映射后的值""" - # "fade" 应该映射为 "fade" - result = self.service._get_validated_transition("fade") - assert result == "fade" - - -class TestComposeSecurityIntegration: - """安全集成测试""" - - def setup_method(self): - """测试前设置""" - self.config = EditingModeConfig(mode=EditingMode.ONE_TAKE) - self.service = VideoComposeService(self.config) - - def test_compose_rejects_malicious_output_path(self): - """测试 compose 方法拒绝恶意输出路径""" - clips = [ - Clip(asset_id="s3://bucket/video1.mp4"), - Clip(asset_id="s3://bucket/video2.mp4"), - ] - - with pytest.raises(ValueError, match="输出路径不在允许范围内"): - self.service.compose(clips, output_path="/etc/passwd") - - def test_compose_rejects_invalid_input_path(self): - """测试 compose 方法拒绝非法输入路径""" - clips = [ - Clip(asset_id="/etc/shadow"), # 非法路径 - ] - - with pytest.raises(ValueError, match="不合法的输入路径"): - self.service.compose(clips) - - def test_compose_with_valid_paths(self): - """测试合法路径可以正常处理""" - with tempfile.TemporaryDirectory() as tmpdir: - # 创建临时视频文件 - video_path = os.path.join(tmpdir, "input.mp4") - output_path = os.path.join("/tmp/video_output", "output.mp4") - - # 创建空的测试文件（实际测试需要真实视频） - with open(video_path, "wb") as f: - f.write(b"fake video data") - - clips = [ - Clip(asset_id=f"local://{video_path}"), - ] - - # 验证输入校验通过 - assert self.service._validate_input_path(f"local://{video_path}") is True - - def test_compose_empty_clips_rejected(self): - """测试空片段列表被拒绝""" - with pytest.raises(ValueError, match="clips 不能为空"): - self.service.compose([]) - - -class TestWhiteListConstants: - """白名单常量测试""" - - def test_allowed_output_dirs_not_empty(self): - """测试输出目录白名单不为空""" - assert len(ALLOWED_OUTPUT_DIRS) > 0 - assert "/tmp/video_output" in ALLOWED_OUTPUT_DIRS - assert "/var/app/rendered" in ALLOWED_OUTPUT_DIRS - - def test_allowed_input_prefixes_not_empty(self): - """测试输入路径前缀白名单不为空""" - assert len(ALLOWED_INPUT_PREFIXES) > 0 - assert "s3://" in ALLOWED_INPUT_PREFIXES - assert "oss://" in ALLOWED_INPUT_PREFIXES - assert "local://" in ALLOWED_INPUT_PREFIXES - assert "/var/storage/" in ALLOWED_INPUT_PREFIXES - - def test_allowed_transitions_not_empty(self): - """测试转场效果白名单不为空""" - assert len(ALLOWED_TRANSITIONS) > 0 - assert "fade" in ALLOWED_TRANSITIONS - assert "dissolve" in ALLOWED_TRANSITIONS - assert "slideleft" in ALLOWED_TRANSITIONS - - -if __name__ == "__main__": - pytest.main([__file__, "-v"])