diff --git a/apps/worker/video_processing/concat_engine.py b/apps/worker/video_processing/concat_engine.py new file mode 100755 index 000000000..dc58a6723 --- /dev/null +++ b/apps/worker/video_processing/concat_engine.py @@ -0,0 +1,627 @@ +"""视频拼接/合并引擎 — 多段视频按顺序拼接成一个成片. + +基于 FFmpeg 实现两种拼接模式： +1. **concat demuxer（stream copy）**：最快，所有视频编码参数必须一致 +2. **concat filter（重新编码）**：更灵活，支持不同分辨率/编码/帧率的视频 + +使用场景： +- 多段素材按顺序合并成一个视频 +- 视频分割后重新拼接 +- 片头 + 正片 + 片尾拼接 + +降级策略： +- 优先尝试 stream copy（速度快、无质量损失） +- 参数不一致时自动降级到 concat filter +- 某段视频失败时跳过，不阻断整体拼接 +""" + +from __future__ import annotations + +import logging +import tempfile +from dataclasses import dataclass, field +from pathlib import Path +from typing import Any + +from video_processing.ffmpeg_utils import FFMPEG_BIN, probe_duration, probe_video_info, run_ffmpeg + +logger = logging.getLogger(__name__) + + +# ── 常量 ────────────────────────────────────────────────────────────────────── + +# concat demuxer 要求一致的参数列表 +CONCAT_DEMUXER_REQUIRED_PARAMS = [ + "codec_name", # 视频编码 + "width", # 宽度 + "height", # 高度 + "r_frame_rate", # 帧率 + "pix_fmt", # 像素格式 + "sample_rate", # 音频采样率 + "channels", # 音频声道数 + "audio_codec", # 音频编码 +] + + +# ── 拼接片段配置 ────────────────────────────────────────────────────────────── + + +@dataclass +class ConcatSegment: + """单个拼接片段.""" + + video_path: str # 视频文件路径 + start_time: float = 0.0 # 开始时间（秒），从视频的哪个位置开始取 + duration: float = 0.0 # 持续时长（秒），0表示取到末尾 + has_audio: bool = True # 是否包含音频 + + @classmethod + def from_dict(cls, seg: dict) -> "ConcatSegment": + """从字典创建拼接片段，带安全类型转换.""" + try: + start_time = max(0.0, float(seg.get("start_time", 0.0))) + except (TypeError, ValueError): + start_time = 0.0 + + try: + duration = max(0.0, float(seg.get("duration", 0.0))) + except (TypeError, ValueError): + duration = 0.0 + + return cls( + video_path=str(seg.get("video_path", "")), + start_time=start_time, + duration=duration, + has_audio=bool(seg.get("has_audio", True)), + ) + + +@dataclass +class ConcatConfig: + """视频拼接配置.""" + + segments: list[ConcatSegment] = field(default_factory=list) + output_width: int = 0 # 输出宽度（0=自动取第一段） + output_height: int = 0 # 输出高度（0=自动取第一段） + output_fps: float = 0.0 # 输出帧率（0=自动取第一段） + force_reencode: bool = False # 强制重新编码（不用 stream copy） + transition: str = "none" # 转场效果（none/crossfade）- 预留 + transition_duration: float = 0.3 # 转场时长 + + @classmethod + def from_config_dict(cls, config: dict | None) -> "ConcatConfig": + """从配置字典创建 ConcatConfig.""" + if not config or not isinstance(config, dict): + return cls() + + segments_raw = config.get("segments", []) + segments: list[ConcatSegment] = [] + + if isinstance(segments_raw, list): + for s in segments_raw: + if isinstance(s, dict) and s.get("video_path"): + try: + seg = ConcatSegment.from_dict(s) + if seg.video_path: + segments.append(seg) + except Exception: + logger.warning("[concat] skip invalid segment: %s", s) + continue + + try: + output_width = max(0, int(config.get("output_width", 0))) + except (TypeError, ValueError): + output_width = 0 + + try: + output_height = max(0, int(config.get("output_height", 0))) + except (TypeError, ValueError): + output_height = 0 + + try: + output_fps = max(0.0, float(config.get("output_fps", 0.0))) + except (TypeError, ValueError): + output_fps = 0.0 + + return cls( + segments=segments, + output_width=output_width, + output_height=output_height, + output_fps=output_fps, + force_reencode=bool(config.get("force_reencode", False)), + transition=str(config.get("transition", "none")), + transition_duration=max(0.1, float(config.get("transition_duration", 0.3))), + ) + + @property + def has_effect(self) -> bool: + """是否有有效片段需要拼接.""" + return len([s for s in self.segments if s.video_path]) >= 2 + + @property + def total_segments(self) -> int: + """有效片段数量.""" + return len([s for s in self.segments if s.video_path]) + + +# ── 视频拼接引擎 ────────────────────────────────────────────────────────────── + + +class ConcatEngine: + """视频拼接引擎 — 支持 stream copy 和重新编码两种模式.""" + + def __init__(self, work_dir: Path): + self.work_dir = work_dir + self.work_dir.mkdir(parents=True, exist_ok=True) + + # ── 主入口 ──────────────────────────────────────────────────────── + + def concat_videos( + self, + config: ConcatConfig, + output_path: Path, + ) -> Path: + """拼接多段视频. + + 自动选择最优拼接策略： + 1. 所有片段参数一致 → concat demuxer（stream copy，最快） + 2. 参数不一致或有裁剪 → concat filter（重新编码） + + Args: + config: 拼接配置 + output_path: 输出文件路径 + + Returns: + 输出文件路径 + """ + valid_segments = [s for s in config.segments if s.video_path] + + if not valid_segments: + raise ValueError("No valid video segments to concat") + + if len(valid_segments) == 1: + # 只有一段，直接复制 + import shutil + + logger.info("[concat] single segment, copy directly") + shutil.copy2(valid_segments[0].video_path, output_path) + return output_path + + # 判断能否用 stream copy + can_stream_copy = self._can_use_stream_copy(config) + + if can_stream_copy and not config.force_reencode: + logger.info("[concat] using concat demuxer (stream copy)") + try: + return self._concat_demuxer(config, output_path) + except Exception as e: + logger.warning("[concat] demuxer failed, fallback to filter: %s", e) + + # 降级到 concat filter + logger.info("[concat] using concat filter (re-encode)") + return self._concat_filter(config, output_path) + + # ── 模式判断 ────────────────────────────────────────────────────── + + def _can_use_stream_copy(self, config: ConcatConfig) -> bool: + """判断是否可以使用 concat demuxer（stream copy）. + + 条件： + 1. 所有视频编码参数一致（分辨率、帧率、编码、像素格式） + 2. 所有音频参数一致（采样率、声道、编码） + 3. 没有设置 start_time 裁剪（或可以通过 concat demuxer 的 inpoint/outpoint 实现） + 4. 没有强制重新编码 + """ + if config.force_reencode: + return False + + # 如果有转场效果，必须重新编码 + if config.transition != "none": + return False + + # 探测所有视频的参数 + video_infos = [] + for seg in config.segments: + if not seg.video_path: + continue + try: + info = probe_video_info(seg.video_path) + video_infos.append(info) + except Exception: + logger.warning("[concat] probe failed for %s", seg.video_path[-40:]) + return False + + if len(video_infos) < 2: + return False + + # 检查参数一致性 + base_info = video_infos[0] + for info in video_infos[1:]: + for param in CONCAT_DEMUXER_REQUIRED_PARAMS: + base_val = base_info.get(param) + curr_val = info.get(param) + if base_val != curr_val: + logger.debug( + "[concat] param mismatch: %s (%s vs %s)", + param, + base_val, + curr_val, + ) + return False + + # 检查是否有裁剪需求 + # concat demuxer 支持 inpoint/outpoint，所以有裁剪也可以用 + # 但为了简单和稳定性，有裁剪时也用 filter 模式 + # （inpoint/outpoint 不是所有格式都支持得好） + has_trimming = any(seg.start_time > 0 or seg.duration > 0 for seg in config.segments if seg.video_path) + if has_trimming: + return False + + return True + + # ── 模式1：concat demuxer（stream copy） ────────────────────────── + + def _concat_demuxer(self, config: ConcatConfig, output_path: Path) -> Path: + """使用 concat demuxer 拼接（stream copy）. + + 优点：速度极快，无质量损失 + 缺点：要求所有视频参数完全一致 + """ + # 生成 concat 文件列表 + list_file = self.work_dir / "concat_list.txt" + lines = [] + for seg in config.segments: + if not seg.video_path: + continue + # 路径转义：单引号替换为 '\'' + safe_path = str(seg.video_path).replace("'", "'\\''") + lines.append(f"file '{safe_path}'") + + list_file.write_text("\n".join(lines), encoding="utf-8") + + command = [ + FFMPEG_BIN, + "-y", + "-f", + "concat", + "-safe", + "0", + "-i", + str(list_file), + "-c", + "copy", + "-copyts", + str(output_path), + ] + + logger.info("[concat] demuxer: %d segments", config.total_segments) + run_ffmpeg(command) + return output_path + + # ── 模式2：concat filter（重新编码） ────────────────────────────── + + def _concat_filter(self, config: ConcatConfig, output_path: Path) -> Path: + """使用 concat filter 拼接（重新编码）. + + 优点：支持不同参数的视频，支持裁剪 + 缺点：需要重新编码，较慢 + """ + valid_segments = [s for s in config.segments if s.video_path] + num_segments = len(valid_segments) + + # 构建输入参数 + input_args: list[str] = [] + for seg in valid_segments: + input_args.extend(["-i", seg.video_path]) + + # 确定输出参数 + output_width, output_height, output_fps = self._get_output_params(config) + + # 构建 filter_complex + filter_parts: list[str] = [] + concat_inputs = "" + + for i, seg in enumerate(valid_segments): + vid_label = f"v{i}" + aud_label = f"a{i}" + + seg_filters: list[str] = [] + + # 1. 裁剪（start_time + duration） + if seg.start_time > 0 or seg.duration > 0: + start = seg.start_time + if seg.duration > 0: + end = start + seg.duration + seg_filters.append(f"trim=start={start:.3f}:end={end:.3f}") + else: + seg_filters.append(f"trim=start={start:.3f}") + seg_filters.append("setpts=PTS-STARTPTS") + + # 音频同步裁剪 + if seg.has_audio: + if seg.duration > 0: + filter_parts.append( + f"[{i}:a]atrim=start={start:.3f}:end={end:.3f}," f"asetpts=PTS-STARTPTS[{aud_label}]" + ) + else: + filter_parts.append(f"[{i}:a]atrim=start={start:.3f}," f"asetpts=PTS-STARTPTS[{aud_label}]") + else: + # 无音频时生成静音轨 + filter_parts.append( + f"[{i}:v]trim=start={start:.3f}," f"setpts=PTS-STARTPTS, " f"aevalsrc=0:d={0.1}[{aud_label}]" + ) + else: + # 无裁剪，直接用原始标签 + if not seg.has_audio: + # 无音频时需要生成静音 + try: + dur = probe_duration(seg.video_path) + except Exception: + dur = 10.0 + filter_parts.append(f"aevalsrc=0:d={dur:.3f}:s=44100[{aud_label}]") + + # 2. 缩放/帧率统一 + vf_parts = [] + if not seg_filters: + vf_parts.append(f"[{i}:v]") + else: + vf_parts.append("") + + # 分辨率统一 + if output_width and output_height: + vf_parts.append( + f"scale={output_width}:{output_height}:force_original_aspect_ratio=decrease," + f"pad={output_width}:{output_height}:(ow-iw)/2:(oh-ih)/2:black" + ) + + # 帧率统一 + if output_fps > 0: + vf_parts.append(f"fps={output_fps}") + + # 像素格式统一 + vf_parts.append("format=yuv420p") + + if len(vf_parts) > 1 or (seg_filters and vf_parts): + if seg_filters: + # 先裁剪后缩放 + crop_str = "".join(seg_filters) + scale_str = "".join(vf_parts[1:]) # 跳过空字符串 + if scale_str: + filter_parts.append(f"[{i}:v]{crop_str},{scale_str}[{vid_label}]") + else: + filter_parts.append(f"[{i}:v]{crop_str}[{vid_label}]") + else: + filter_parts.append(f"{vf_parts[0]}{''.join(vf_parts[1:])}[{vid_label}]") + else: + if seg_filters: + filter_parts.append(f"[{i}:v]{''.join(seg_filters)}[{vid_label}]") + else: + # 什么都不需要，直接用输入 + pass + + # 拼接 concat 的输入标签 + if seg_filters or (output_width and output_height) or output_fps > 0: + concat_inputs += f"[{vid_label}]" + else: + concat_inputs += f"[{i}:v]" + + # 音频标签 + if seg.start_time > 0 or seg.duration > 0: + # 已经生成了 aud_label + pass + elif not seg.has_audio: + # 已经生成了静音 aud_label + pass + else: + # 使用原始音频 + pass + + # 简化处理：用更直接的方式构建 filter + # 重新整理一下，确保所有输入都有对应的 v_i 和 a_i 标签 + filter_parts.clear() + concat_inputs = "" # 按段交织: [v0][a0][v1][a1]... + + for i, seg in enumerate(valid_segments): + v_label = f"v{i}_in" + a_label = f"a{i}_in" + + # 视频处理链 + v_steps: list[str] = [f"[{i}:v]"] + + # 裁剪 + if seg.start_time > 0 or seg.duration > 0: + start = seg.start_time + if seg.duration > 0: + end = start + seg.duration + v_steps.append(f"trim=start={start:.3f}:end={end:.3f},") + else: + v_steps.append(f"trim=start={start:.3f},") + v_steps.append("setpts=PTS-STARTPTS,") + + # 缩放 + if output_width and output_height: + v_steps.append( + f"scale={output_width}:{output_height}:force_original_aspect_ratio=decrease," + f"pad={output_width}:{output_height}:(ow-iw)/2:(oh-ih)/2:black," + ) + + # 帧率 + if output_fps > 0: + v_steps.append(f"fps={output_fps},") + + # 像素格式 + v_steps.append("format=yuv420p") + + v_filter = "".join(v_steps) + f"[{v_label}]" + filter_parts.append(v_filter) + + # 音频处理链 + a_steps: list[str] = [] + if seg.has_audio: + a_steps.append(f"[{i}:a]") + + if seg.start_time > 0 or seg.duration > 0: + start = seg.start_time + if seg.duration > 0: + end = start + seg.duration + a_steps.append(f"atrim=start={start:.3f}:end={end:.3f},") + else: + a_steps.append(f"atrim=start={start:.3f},") + a_steps.append("asetpts=PTS-STARTPTS,") + + a_steps.append("aformat=sample_fmts=fltp:sample_rates=44100:channel_layouts=stereo") + else: + # 生成静音音频 + try: + dur = probe_duration(seg.video_path) + except Exception: + dur = 10.0 + # 减去裁剪 + if seg.start_time > 0: + dur = max(0.1, dur - seg.start_time) + if seg.duration > 0 and seg.duration < dur: + dur = seg.duration + a_steps.append(f"aevalsrc=0:d={dur:.3f}:s=44100:c=stereo") + + a_filter = "".join(a_steps) + f"[{a_label}]" + filter_parts.append(a_filter) + + # 按段交织排列（v_i, a_i），这是 FFmpeg concat filter 要求的顺序 + concat_inputs += f"[{v_label}][{a_label}]" + + # concat filter: 输入按 [v0][a0][v1][a1]... 顺序 + filter_parts.append(f"{concat_inputs}" f"concat=n={num_segments}:v=1:a=1[vout][aout]") + + filter_complex = ";".join(filter_parts) + + command = [ + FFMPEG_BIN, + "-y", + *input_args, + "-filter_complex", + filter_complex, + "-map", + "[vout]", + "-map", + "[aout]", + "-c:v", + "libx264", + "-preset", + "fast", + "-crf", + "23", + "-c:a", + "aac", + "-b:a", + "128k", + "-movflags", + "+faststart", + str(output_path), + ] + + logger.info( + "[concat] filter: %d segments, %dx%d, %.2f fps", + num_segments, + output_width, + output_height, + output_fps, + ) + run_ffmpeg(command) + return output_path + + # ── 辅助方法 ────────────────────────────────────────────────────── + + def _get_output_params(self, config: ConcatConfig) -> tuple[int, int, float]: + """获取输出参数（宽、高、帧率）. + + 优先级： + 1. config 中显式指定的 + 2. 第一段视频的参数 + """ + valid_segments = [s for s in config.segments if s.video_path] + + width = config.output_width + height = config.output_height + fps = config.output_fps + + # 如果没有显式指定，用第一段的参数 + if (width == 0 or height == 0 or fps == 0) and valid_segments: + try: + info = probe_video_info(valid_segments[0].video_path) + if width == 0: + width = int(info.get("width", 1080)) + if height == 0: + height = int(info.get("height", 1920)) + if fps == 0: + fps_str = info.get("r_frame_rate", "30/1") + if "/" in str(fps_str): + num, den = str(fps_str).split("/") + try: + fps = float(num) / float(den) + except (ValueError, ZeroDivisionError): + fps = 30.0 + else: + fps = float(fps_str) if fps_str else 30.0 + except Exception: + # 探测失败，用默认值 + if width == 0: + width = 1080 + if height == 0: + height = 1920 + if fps == 0: + fps = 30.0 + + return width, height, fps + + +# ── 便捷函数 ────────────────────────────────────────────────────────────────── + + +def concat_video_files( + video_paths: list[str], + output_path: Path, + *, + work_dir: Path | None = None, + force_reencode: bool = False, +) -> Path: + """简单拼接多个视频文件. + + Args: + video_paths: 视频文件路径列表 + output_path: 输出路径 + work_dir: 工作目录（默认输出文件所在目录） + force_reencode: 是否强制重新编码 + + Returns: + 输出文件路径 + """ + if work_dir is None: + work_dir = output_path.parent + + segments = [ConcatSegment(video_path=p) for p in video_paths if p] + config = ConcatConfig(segments=segments, force_reencode=force_reencode) + + engine = ConcatEngine(work_dir) + return engine.concat_videos(config, output_path) + + +def concat_videos_from_config( + config_dict: dict | None, + output_path: Path, + *, + work_dir: Path, +) -> Path | None: + """从配置字典执行视频拼接. + + 降级策略：配置无效或拼接失败时返回 None. + """ + config = ConcatConfig.from_config_dict(config_dict) + if not config.has_effect: + return None + + try: + engine = ConcatEngine(work_dir) + return engine.concat_videos(config, output_path) + except Exception as e: + logger.error("[concat] concat failed: %s", e) + return None diff --git a/apps/worker/video_processing/multi_track_mixer.py b/apps/worker/video_processing/multi_track_mixer.py new file mode 100755 index 000000000..b626d0a73 --- /dev/null +++ b/apps/worker/video_processing/multi_track_mixer.py @@ -0,0 +1,392 @@ +"""多轨道混音引擎 — 支持多路音频独立音量调节与混合. + +基于 FFmpeg amix / amerge 实现： +- 支持任意数量音频轨道（原音、BGM、配音、音效等） +- 每轨独立音量调节 +- 每轨独立淡入淡出 +- 每轨独立时间偏移（delay） +- 总输出音量归一化补偿 + +作为 render_audio.py 的增强模块，在 mix_audio 后处理阶段被调用。 +与 bgm_mixer.py 的关系： +- bgm_mixer 专注 BGM 单轨道的复杂处理（循环、人声闪避） +- 本模块专注多路轨道的统一音量调节与混合 +""" + +from __future__ import annotations + +import logging +from dataclasses import dataclass, field +from pathlib import Path +from typing import TYPE_CHECKING + +from video_processing.ffmpeg_utils import FFMPEG_BIN, probe_duration, run_ffmpeg + +if TYPE_CHECKING: + from video_processing.render_audio import RenderContext + +logger = logging.getLogger(__name__) + + +# ── 常量 ────────────────────────────────────────────────────────────────────── + +TRACK_TYPE_MAIN = "main" # 原音（视频原声） +TRACK_TYPE_BGM = "bgm" # 背景音乐 +TRACK_TYPE_VOICEOVER = "voiceover" # 配音（TTS/人声） +TRACK_TYPE_SFX = "sfx" # 音效 +TRACK_TYPE_AMBIENT = "ambient" # 环境音 + +# 各轨道默认音量（相对主音频） +DEFAULT_VOLUMES = { + TRACK_TYPE_MAIN: 1.0, + TRACK_TYPE_BGM: 0.3, + TRACK_TYPE_VOICEOVER: 1.0, + TRACK_TYPE_SFX: 0.7, + TRACK_TYPE_AMBIENT: 0.2, +} + + +@dataclass +class AudioTrack: + """单条音频轨道配置.""" + + track_id: str # 轨道唯一标识 + track_type: str # 轨道类型（main/bgm/voiceover/sfx/ambient） + audio_path: str # 音频文件路径 + volume: float = 1.0 # 音量 0.0 ~ 2.0 + fade_in: float = 0.0 # 淡入时长（秒） + fade_out: float = 0.0 # 淡出时长（秒） + start_time: float = 0.0 # 开始时间（相对于视频起点，秒） + duration: float = 0.0 # 持续时长（0表示到文件末尾） + enabled: bool = True # 是否启用 + + @classmethod + def from_dict(cls, track: dict) -> "AudioTrack": + """从字典创建 AudioTrack，带安全类型转换.""" + track_type = str(track.get("track_type", TRACK_TYPE_SFX)) + default_vol = DEFAULT_VOLUMES.get(track_type, 1.0) + + try: + volume = float(track.get("volume", default_vol)) + except (TypeError, ValueError): + volume = default_vol + volume = max(0.0, min(2.0, volume)) + + try: + fade_in = max(0.0, float(track.get("fade_in", 0.0))) + except (TypeError, ValueError): + fade_in = 0.0 + + try: + fade_out = max(0.0, float(track.get("fade_out", 0.0))) + except (TypeError, ValueError): + fade_out = 0.0 + + try: + start_time = max(0.0, float(track.get("start_time", 0.0))) + except (TypeError, ValueError): + start_time = 0.0 + + try: + duration = max(0.0, float(track.get("duration", 0.0))) + except (TypeError, ValueError): + duration = 0.0 + + return cls( + track_id=str(track.get("track_id", "")), + track_type=track_type, + audio_path=str(track.get("audio_path", "")), + volume=volume, + fade_in=fade_in, + fade_out=fade_out, + start_time=start_time, + duration=duration, + enabled=bool(track.get("enabled", True)), + ) + + +@dataclass +class MultiTrackMixConfig: + """多轨道混音配置.""" + + tracks: list[AudioTrack] = field(default_factory=list) + master_volume: float = 1.0 # 主输出音量 + normalize: bool = True # 是否自动归一化补偿 + max_output_volume: float = 1.5 # 最大输出音量（防止爆音） + + @classmethod + def from_config_dict(cls, config: dict | None) -> "MultiTrackMixConfig": + """从 plan.config.audio_tracks 字典创建配置.""" + if not config or not isinstance(config, dict): + return cls() + + tracks_raw = config.get("tracks", []) + tracks: list[AudioTrack] = [] + + if isinstance(tracks_raw, list): + for t in tracks_raw: + if isinstance(t, dict) and t.get("audio_path"): + try: + track = AudioTrack.from_dict(t) + if track.enabled and track.audio_path: + tracks.append(track) + except Exception: + logger.warning("[multi-track] skip invalid track config: %s", t) + continue + + try: + master_volume = float(config.get("master_volume", 1.0)) + master_volume = max(0.0, min(2.0, master_volume)) + except (TypeError, ValueError): + master_volume = 1.0 + + return cls( + tracks=tracks, + master_volume=master_volume, + normalize=bool(config.get("normalize", True)), + max_output_volume=float(config.get("max_output_volume", 1.5)), + ) + + @property + def has_effect(self) -> bool: + """是否有有效轨道需要混音.""" + return len([t for t in self.tracks if t.enabled and t.audio_path]) > 0 + + +# ── 单轨道预处理 ──────────────────────────────────────────────────────────── + + +def _prepare_single_track( + ctx: "RenderContext", + track: AudioTrack, + target_duration: float, + output_path: Path, +) -> bool: + """预处理单条轨道：音量 + 淡入淡出 + 时间偏移 + 截断. + + 生成一个精确对齐时间轴的音频文件，后续统一 amix 混音。 + + Returns: + True 表示处理成功，False 表示失败（跳过） + """ + try: + audio_dur = probe_duration(track.audio_path) + except Exception: + logger.warning("[multi-track] probe failed, skip track: %s", track.track_id) + return False + + if audio_dur <= 0: + return False + + # 计算实际有效时长 + effective_start = track.start_time + if track.duration > 0: + effective_dur = min(track.duration, audio_dur) + else: + effective_dur = audio_dur + + # 如果轨道完全在视频时长之外，跳过 + if effective_start >= target_duration: + return False + if effective_start + effective_dur <= 0: + return False + + # 构建滤镜链 + filter_parts: list[str] = [] + + # 1. 先截断到有效范围 + trim_start = 0.0 # 从源文件的哪个位置开始取 + if effective_start < 0: + trim_start = -effective_start + effective_start = 0.0 + + # 实际需要的源时长 + need_dur = min(effective_dur, target_duration - effective_start) + if need_dur <= 0: + return False + + filter_parts.append(f"atrim={trim_start:.3f}:{trim_start + need_dur:.3f}") + filter_parts.append("asetpts=N/SR/TB") + + # 2. 音量调节 + if abs(track.volume - 1.0) > 0.001: + filter_parts.append(f"volume={track.volume:.3f}") + + # 3. 淡入 + if track.fade_in > 0 and track.fade_in < need_dur: + filter_parts.append(f"afade=t=in:st=0:d={track.fade_in:.3f}") + + # 4. 淡出 + if track.fade_out > 0 and track.fade_out < need_dur: + fade_start = need_dur - track.fade_out + if fade_start > 0: + filter_parts.append(f"afade=t=out:st={fade_start:.3f}:d={track.fade_out:.3f}") + + # 5. 时间偏移（用 adelay 实现开头静音填充） + if effective_start > 0.01: + delay_ms = int(effective_start * 1000) + filter_parts.append(f"adelay={delay_ms}|{delay_ms}") + + # 6. 最终截断到目标总时长 + filter_parts.append(f"atrim=0:{target_duration:.3f}") + filter_parts.append("asetpts=N/SR/TB") + + filter_str = ",".join(filter_parts) + + command = [ + FFMPEG_BIN, + "-y", + "-i", + track.audio_path, + "-filter:a", + filter_str, + "-c:a", + "aac", + "-b:a", + "128k", + str(output_path), + ] + + logger.info( + "[multi-track] prepare track: id=%s type=%s vol=%.2f start=%.2f dur=%.2f", + track.track_id, + track.track_type, + track.volume, + effective_start, + need_dur, + ) + + try: + run_ffmpeg(command) + return True + except Exception as e: + logger.warning("[multi-track] track prepare failed: %s, error=%s", track.track_id, e) + return False + + +# ── 多轨道混音主入口 ───────────────────────────────────────────────────────── + + +def mix_multi_track( + ctx: "RenderContext", + main_audio_path: Path, + config: MultiTrackMixConfig, + target_duration: float, +) -> Path: + """多轨道混音：主音频 + 多条附加轨道. + + Args: + ctx: 渲染上下文 + main_audio_path: 主音频文件路径（原音） + config: 多轨道混音配置 + target_duration: 目标总时长 + + Returns: + 混音后的音频文件路径 + """ + output_path = ctx.work_dir / f"multi_track_mix_{ctx.plan_id}.aac" + + if target_duration <= 0: + target_duration = 5.0 + + # 收集所有有效轨道（已预处理好的） + prepared_tracks: list[Path] = [] + + # 主音频作为第0轨 + prepared_tracks.append(main_audio_path) + + # 预处理每条附加轨道 + for i, track in enumerate(config.tracks): + if not track.enabled or not track.audio_path: + continue + + track_out = ctx.work_dir / f"track_{i}_{ctx.plan_id}.aac" + if _prepare_single_track(ctx, track, target_duration, track_out): + prepared_tracks.append(track_out) + + # 如果只有主音频，直接返回（无需混音） + if len(prepared_tracks) <= 1: + import shutil + + shutil.copy2(main_audio_path, output_path) + return output_path + + # 使用 amix 混音 + num_inputs = len(prepared_tracks) + + # 构建输入参数 + input_args: list[str] = [] + for tp in prepared_tracks: + input_args.extend(["-i", str(tp)]) + + # amix 的 duration=first 以第一个输入（主音频）时长为准 + # normalize 补偿：amix 会把每路音量除以 N，需要乘回来 + # 但如果所有轨道都同时有声，可能会爆音，所以用 master_volume 控制 + if config.normalize: + # 经验值：不是所有轨道都同时有声，补偿系数取 N * 0.7 + compensate = num_inputs * 0.7 + else: + compensate = 1.0 + + final_volume = compensate * config.master_volume + final_volume = min(final_volume, config.max_output_volume) + + # 构建 filter_complex + inputs_label = "".join(f"[{i}:a]" for i in range(num_inputs)) + filter_complex = ( + f"{inputs_label}amix=inputs={num_inputs}:duration=first:dropout_transition=0[outa];" + f"[outa]volume={final_volume:.3f}[final]" + ) + + command = [ + FFMPEG_BIN, + "-y", + *input_args, + "-filter_complex", + filter_complex, + "-map", + "[final]", + "-c:a", + "aac", + "-b:a", + "128k", + str(output_path), + ] + + logger.info( + "[multi-track] mix %d tracks, master_vol=%.2f compensate=%.2f final_vol=%.2f", + num_inputs, + config.master_volume, + compensate, + final_volume, + ) + + try: + run_ffmpeg(command) + except Exception as e: + logger.error("[multi-track] mix failed, fallback to main audio only: %s", e) + import shutil + + shutil.copy2(main_audio_path, output_path) + + return output_path + + +# ── 便捷函数：从 plan.config 快速混音 ─────────────────────────────────────── + + +def mix_audio_tracks_from_config( + ctx: "RenderContext", + main_audio_path: Path, + audio_tracks_config: dict | None, + target_duration: float, +) -> Path: + """从 plan.config.audio_tracks 配置执行多轨道混音. + + 降级策略：配置无效或混音失败时返回主音频。 + """ + config = MultiTrackMixConfig.from_config_dict(audio_tracks_config) + if not config.has_effect: + return main_audio_path + + return mix_multi_track(ctx, main_audio_path, config, target_duration) diff --git a/apps/worker/video_processing/render_audio.py b/apps/worker/video_processing/render_audio.py index 0aa7f4415..3cde25c39 100755 --- a/apps/worker/video_processing/render_audio.py +++ b/apps/worker/video_processing/render_audio.py @@ -76,6 +76,7 @@ def mix_audio( *, bgm_path: str | None = None, bgm_config: dict | None = None, + audio_tracks_config: dict | None = None, ) -> Path | None: """音频后处理混音. @@ -86,6 +87,8 @@ def mix_audio( 4. 输出时长截断到 video_duration 5. 无音频流的 clip 会被自动跳过，避免 FFmpeg 引用 [i:a] 失败 6. 如果提供了 bgm_path，则额外混入 BGM（支持淡入淡出、循环、人声闪避） + 7. 如果配置了 audio_tracks，则混入多轨道音频（配音、音效等） + 8. 如果配置了降噪，最后应用降噪 Args: ctx: 渲染上下文 @@ -93,6 +96,7 @@ def mix_audio( video_duration: 视频总时长（用于截断音频） bgm_path: BGM 音频本地路径，为 None 时不混入 BGM bgm_config: BGM 配置字典（volume/fade_in/fade_out/sidechain 等） + audio_tracks_config: 多轨道音频配置（tracks/master_volume 等） Returns: 混音后的音频文件路径，无音频时返回 None @@ -156,10 +160,21 @@ def mix_audio( try: # 这里 main_audio 就是 output_path，先有主音频再混 BGM final_path = mix_bgm_with_main(ctx, output_path, bgm_cfg, video_duration) - return _apply_noise_reduction_if_needed(ctx, final_path) + output_path = final_path except Exception: logger.exception("[bgm] BGM 混音失败，回退到无 BGM 音频: plan_id=%s", ctx.plan_id) - return _apply_noise_reduction_if_needed(ctx, output_path) + + # ── 多轨道混音（配音/音效等） ── + if audio_tracks_config and audio_tracks_config.get("enabled", False): + from video_processing.multi_track_mixer import mix_audio_tracks_from_config + + try: + tracks_config = audio_tracks_config.get("tracks_config") or audio_tracks_config + multi_output = mix_audio_tracks_from_config(ctx, output_path, tracks_config, video_duration) + if multi_output and multi_output != output_path: + output_path = multi_output + except Exception: + logger.exception("[multi-track] 多轨道混音失败，回退: plan_id=%s", ctx.plan_id) return _apply_noise_reduction_if_needed(ctx, output_path) diff --git a/apps/worker/video_processing/subtitle_render_engine.py b/apps/worker/video_processing/subtitle_render_engine.py new file mode 100755 index 000000000..931b2240a --- /dev/null +++ b/apps/worker/video_processing/subtitle_render_engine.py @@ -0,0 +1,633 @@ +"""字幕渲染引擎 — 统一管理字幕样式配置与视频烧录. + +与现有模块的关系： +- render_subtitles.py：生成静态整段标题/字幕的 ASS 文件 +- subtitle_generator.py：从 ASR 时间轴生成 ASS 文件 +- 本模块：统一的字幕样式配置 + 烧录滤镜生成 + 多源字幕合并 + +支持的字幕来源： +1. 静态标题/字幕（title_config / subtitle_config） +2. ASR 自动字幕（asr_subtitle_timeline） +3. 手动字幕（manual_subtitles 时间轴） + +支持的样式配置： +- 字体、字号、颜色 +- 描边（颜色、宽度） +- 阴影（偏移、模糊、颜色） +- 背景框（颜色、透明度、圆角、边距） +- 位置（9宫格 + 自定义坐标） +- 对齐方式 +- 动画（淡入淡出、滑入滑出、打字机） +- 多行/换行规则 +""" + +from __future__ import annotations + +import logging +from dataclasses import dataclass, field +from pathlib import Path +from typing import Any + +from video_processing.render_subtitles import generate_ass_subtitles +from video_processing.subtitle_generator import generate_ass_from_timeline + +logger = logging.getLogger(__name__) + + +# ── 常量 ────────────────────────────────────────────────────────────────────── + +# 9宫格位置映射（ASS alignment 编号） +POSITION_ALIGNMENT = { + "top_left": 7, + "top_center": 8, + "top_right": 9, + "middle_left": 4, + "center": 5, + "middle_right": 6, + "bottom_left": 1, + "bottom_center": 2, + "bottom_right": 3, +} + +# 位置简称兼容 +POSITION_ALIASES = { + "top": "top_center", + "bottom": "bottom_center", + "middle": "center", + "left": "middle_left", + "right": "middle_right", +} + +DEFAULT_FONT = "思源黑体" +DEFAULT_FONT_SIZE = 24 +DEFAULT_COLOR = "#FFFFFF" +DEFAULT_STROKE_COLOR = "#000000" +DEFAULT_STROKE_WIDTH = 1.5 +DEFAULT_POSITION = "bottom_center" +DEFAULT_MAX_CHARS_PER_LINE = 20 + + +# ── 字幕样式配置 ──────────────────────────────────────────────────────────── + + +@dataclass +class SubtitleStyle: + """字幕样式配置.""" + + font_name: str = DEFAULT_FONT + font_size: int = DEFAULT_FONT_SIZE + font_color: str = DEFAULT_COLOR + bold: bool = False + italic: bool = False + + # 描边 + stroke_enabled: bool = True + stroke_color: str = DEFAULT_STROKE_COLOR + stroke_width: float = DEFAULT_STROKE_WIDTH + + # 阴影 + shadow_enabled: bool = False + shadow_color: str = "#000000" + shadow_offset_x: int = 2 + shadow_offset_y: int = 2 + shadow_blur: float = 0.0 + + # 背景框 + background_enabled: bool = False + background_color: str = "#000000" + background_opacity: float = 0.5 # 0.0 ~ 1.0 + background_padding: int = 8 + background_radius: int = 4 + + # 位置 + position: str = DEFAULT_POSITION # 9宫格位置名 + margin_v: int = 60 # 垂直边距 + margin_l: int = 40 # 左边距 + margin_r: int = 40 # 右边距 + + # 多行 + max_chars_per_line: int = DEFAULT_MAX_CHARS_PER_LINE + line_spacing: int = 0 # 行间距 + + # 动画 + fade_in: float = 0.0 # 淡入时长（秒） + fade_out: float = 0.0 # 淡出时长（秒） + animation_type: str = "none" # none/fade/slide/typewriter + + @classmethod + def from_dict(cls, config: dict[str, Any] | None) -> "SubtitleStyle": + """从字典创建样式配置，带安全类型转换.""" + if not config or not isinstance(config, dict): + return cls() + + def safe_str(key: str, default: str) -> str: + val = config.get(key, default) + return str(val) if val is not None else default + + def safe_int(key: str, default: int) -> int: + try: + return int(config.get(key, default)) + except (TypeError, ValueError): + return default + + def safe_float(key: str, default: float) -> float: + try: + return float(config.get(key, default)) + except (TypeError, ValueError): + return default + + def safe_bool(key: str, default: bool) -> bool: + return bool(config.get(key, default)) + + position = safe_str("position", DEFAULT_POSITION) + position = POSITION_ALIASES.get(position, position) + if position not in POSITION_ALIGNMENT: + position = DEFAULT_POSITION + + return cls( + font_name=safe_str("font", DEFAULT_FONT), + font_size=safe_int("size", DEFAULT_FONT_SIZE), + font_color=safe_str("color", DEFAULT_COLOR), + bold=safe_bool("bold", False), + italic=safe_bool("italic", False), + stroke_enabled=safe_bool("stroke_enabled", True), + stroke_color=safe_str("stroke_color", DEFAULT_STROKE_COLOR), + stroke_width=safe_float("stroke_width", DEFAULT_STROKE_WIDTH), + shadow_enabled=safe_bool("shadow_enabled", False), + shadow_color=safe_str("shadow_color", "#000000"), + shadow_offset_x=safe_int("shadow_offset_x", 2), + shadow_offset_y=safe_int("shadow_offset_y", 2), + shadow_blur=safe_float("shadow_blur", 0.0), + background_enabled=safe_bool("background_enabled", False), + background_color=safe_str("background_color", "#000000"), + background_opacity=max(0.0, min(1.0, safe_float("background_opacity", 0.5))), + background_padding=safe_int("background_padding", 8), + background_radius=safe_int("background_radius", 4), + position=position, + margin_v=safe_int("margin_v", 60), + margin_l=safe_int("margin_l", 40), + margin_r=safe_int("margin_r", 40), + max_chars_per_line=safe_int("max_chars_per_line", DEFAULT_MAX_CHARS_PER_LINE), + line_spacing=safe_int("line_spacing", 0), + fade_in=max(0.0, safe_float("fade_in", 0.0)), + fade_out=max(0.0, safe_float("fade_out", 0.0)), + animation_type=safe_str("animation_type", "none"), + ) + + @property + def alignment(self) -> int: + """获取 ASS alignment 编号.""" + return POSITION_ALIGNMENT.get(self.position, 2) + + @property + def ass_font_color(self) -> str: + """ASS 格式颜色 &HAABBGGRR.""" + return _hex_to_ass_color(self.font_color) + + @property + def ass_stroke_color(self) -> str: + return _hex_to_ass_color(self.stroke_color) + + @property + def ass_shadow_color(self) -> str: + return _hex_to_ass_color(self.shadow_color) + + @property + def ass_background_color(self) -> str: + """背景框颜色（ASS BackColour），带透明度.""" + alpha_hex = _opacity_to_ass_alpha(self.background_opacity) + color_bgr = _hex_to_ass_bgr(self.background_color) + return f"&H{alpha_hex}{color_bgr}" + + +# ── 工具函数 ────────────────────────────────────────────────────────────────── + + +def _hex_to_ass_color(hex_color: str) -> str: + """HEX → ASS 颜色 &HAABBGGRR（默认不透明）.""" + hex_color = hex_color.lstrip("#") + if len(hex_color) != 6: + return "&H00FFFFFF" + r, g, b = hex_color[0:2], hex_color[2:4], hex_color[4:6] + return f"&H00{b.upper()}{g.upper()}{r.upper()}" + + +def _hex_to_ass_bgr(hex_color: str) -> str: + """HEX → ASS BGR 部分（不含 alpha）.""" + hex_color = hex_color.lstrip("#") + if len(hex_color) != 6: + return "FFFFFF" + r, g, b = hex_color[0:2], hex_color[2:4], hex_color[4:6] + return f"{b.upper()}{g.upper()}{r.upper()}" + + +def _opacity_to_ass_alpha(opacity: float) -> str: + """不透明度 → ASS alpha（00=不透明，FF=完全透明）.""" + alpha = 255 - int(opacity * 255) + return f"{alpha:02X}" + + +def _escape_ass_text(text: str) -> str: + """转义 ASS 文本特殊字符.""" + text = text.replace("\r\n", "\\N").replace("\n", "\\N").replace("\r", "\\N") + 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 _wrap_text(text: str, max_chars: int) -> list[str]: + """按字数换行，优先标点断开.""" + if len(text) <= max_chars: + return [text] + + lines: list[str] = [] + remaining = text + + while len(remaining) > max_chars: + break_point = max_chars + punctuations = "，。！？、；：,.;:!?" + + for i in range(max_chars, max_chars // 2, -1): + if i < len(remaining) and remaining[i] in punctuations: + break_point = i + 1 + break + + lines.append(remaining[:break_point]) + remaining = remaining[break_point:] + + if remaining: + lines.append(remaining) + + return lines + + +# ── 字幕片段 ────────────────────────────────────────────────────────────────── + + +@dataclass +class SubtitleSegment: + """单个字幕片段.""" + + start: float # 开始时间（秒） + end: float # 结束时间（秒） + text: str # 字幕文本 + style_name: str = "Default" # 使用的样式名 + + +# ── 字幕渲染引擎 ────────────────────────────────────────────────────────────── + + +class SubtitleRenderEngine: + """字幕渲染引擎 — 统一管理多源字幕的 ASS 文件生成. + + 支持合并多个字幕来源到同一个 ASS 文件： + - 标题（顶部，单独样式） + - 字幕（底部，单独样式） + - ASR 时间轴字幕 + - 手动字幕 + + 输出一个统一的 ASS 文件，供 FFmpeg subtitles filter 烧录。 + """ + + def __init__( + self, + video_width: int = 1080, + video_height: int = 1920, + video_duration: float = 0.0, + ): + self.video_width = video_width + self.video_height = video_height + self.video_duration = video_duration + self._styles: dict[str, SubtitleStyle] = {} + self._segments: list[SubtitleSegment] = [] + self._style_counter = 0 + + # ── 样式管理 ────────────────────────────────────────────────────── + + def add_style(self, name: str, style: SubtitleStyle) -> str: + """注册一个样式，返回样式名.""" + self._styles[name] = style + return name + + def get_or_create_style(self, base_name: str, style: SubtitleStyle) -> str: + """获取或创建样式（避免重复）.""" + if base_name in self._styles: + return base_name + self._styles[base_name] = style + return base_name + + # ── 字幕源添加 ──────────────────────────────────────────────────── + + def add_title(self, text: str, style: SubtitleStyle | None = None) -> None: + """添加整段标题（显示整个视频时长）.""" + if not text or not text.strip(): + return + + style = style or SubtitleStyle( + position="top_center", + font_size=48, + bold=True, + stroke_enabled=True, + stroke_width=2.0, + ) + style_name = self.get_or_create_style("TitleStyle", style) + + self._segments.append( + SubtitleSegment( + start=0.0, + end=self.video_duration if self.video_duration > 0 else 9999.0, + text=text.strip(), + style_name=style_name, + ) + ) + + def add_subtitle_text(self, text: str, style: SubtitleStyle | None = None) -> None: + """添加整段字幕（显示整个视频时长）.""" + if not text or not text.strip(): + return + + style = style or SubtitleStyle() + style_name = self.get_or_create_style("SubtitleStyle", style) + + self._segments.append( + SubtitleSegment( + start=0.0, + end=self.video_duration if self.video_duration > 0 else 9999.0, + text=text.strip(), + style_name=style_name, + ) + ) + + def add_timeline_segments( + self, + segments: list[dict] | list[SubtitleSegment], + style: SubtitleStyle | None = None, + ) -> None: + """添加时间轴字幕片段（ASR 或手动字幕）. + + segments 可以是： + - SubtitleSegment 列表 + - dict 列表，每个 dict 含 start/end/text 字段 + """ + if not segments: + return + + style = style or SubtitleStyle() + style_name = self.get_or_create_style("Default", style) + + for seg in segments: + if isinstance(seg, SubtitleSegment): + seg.style_name = style_name + self._segments.append(seg) + elif isinstance(seg, dict): + try: + start = float(seg.get("start", 0)) + end = float(seg.get("end", 0)) + text = str(seg.get("text", "")) + if end > start and text.strip(): + self._segments.append( + SubtitleSegment( + start=start, + end=end, + text=text.strip(), + style_name=style_name, + ) + ) + except (TypeError, ValueError): + continue + + def add_asr_timeline(self, timeline: Any, style: SubtitleStyle | None = None) -> None: + """从 SubtitleTimeline 对象添加 ASR 字幕.""" + if not timeline or not hasattr(timeline, "segments") or not timeline.segments: + return + + style = style or SubtitleStyle() + style_name = self.get_or_create_style("ASRStyle", style) + + for seg in timeline.segments: + if hasattr(seg, "start") and hasattr(seg, "end") and hasattr(seg, "text"): + if seg.end > seg.start and seg.text.strip(): + self._segments.append( + SubtitleSegment( + start=seg.start, + end=seg.end, + text=seg.text.strip(), + style_name=style_name, + ) + ) + + # ── ASS 文件生成 ────────────────────────────────────────────────── + + def generate_ass(self, output_path: Path) -> Path: + """生成 ASS 字幕文件. + + Returns: + 生成的文件路径；如果没有字幕内容，返回空文件。 + """ + if not self._segments: + output_path.parent.mkdir(parents=True, exist_ok=True) + output_path.write_text("", encoding="utf-8") + return output_path + + # 确保至少有 Default 样式 + if "Default" not in self._styles: + self._styles["Default"] = SubtitleStyle() + + # 生成样式行 + style_lines = [] + for name, style in self._styles.items(): + style_lines.append(self._build_ass_style_line(name, style)) + + # 生成事件行（按时间排序） + self._segments.sort(key=lambda s: s.start) + event_lines = [] + for seg in self._segments: + event_lines.append(self._build_ass_event_line(seg)) + + # 组装文件 + ass_content = f"""[Script Info] +ScriptType: v4.00+ +PlayResX: {self.video_width} +PlayResY: {self.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(style_lines)} + +[Events] +Format: Layer, Start, End, Style, Name, MarginL, MarginR, MarginV, Effect, Text +{chr(10).join(event_lines)} +""" + + output_path.parent.mkdir(parents=True, exist_ok=True) + output_path.write_text(ass_content, encoding="utf-8") + return output_path + + def _build_ass_style_line(self, name: str, style: SubtitleStyle) -> str: + """构建一条 ASS Style 行.""" + bold_val = -1 if style.bold else 0 + italic_val = -1 if style.italic else 0 + + # BorderStyle: 1=outline+shadow, 3=opaque box（背景框） + if style.background_enabled: + border_style = 3 + back_color = style.ass_background_color + else: + border_style = 1 + back_color = style.ass_shadow_color if style.shadow_enabled else style.ass_font_color + + outline_val = style.stroke_width if style.stroke_enabled else 0.0 + shadow_val = style.shadow_offset_y if style.shadow_enabled else 0 + + return ( + f"Style: {name},{style.font_name},{style.font_size},{style.ass_font_color}," + f"&H000000FF,{style.ass_stroke_color},{back_color}," + f"{bold_val},{italic_val},0,0,100,100,0,0," + f"{border_style},{outline_val},{shadow_val},{style.alignment}," + f"{style.margin_l},{style.margin_r},{style.margin_v},1" + ) + + def _build_ass_event_line(self, seg: SubtitleSegment) -> str: + """构建一条 ASS Dialogue 事件行.""" + style = self._styles.get(seg.style_name, SubtitleStyle()) + max_chars = style.max_chars_per_line + + # 自动换行 + lines = _wrap_text(seg.text, max_chars) + display_text = "\\N".join(lines) + + # 动画效果（淡入淡出） + effect_tags = "" + if style.fade_in > 0 or style.fade_out > 0: + fade_in_ms = int(style.fade_in * 1000) + fade_out_ms = int(style.fade_out * 1000) + effect_tags = f"{{\\fad({fade_in_ms},{fade_out_ms})}}" + + safe_text = _escape_ass_text(display_text) + start_time = _format_ass_time(max(0, seg.start)) + end_time = _format_ass_time(max(seg.start + 0.1, seg.end)) + + return f"Dialogue: 0,{start_time},{end_time},{seg.style_name},,0,0,0,," f"{effect_tags}{safe_text}" + + @property + def has_subtitles(self) -> bool: + """是否有字幕内容.""" + return len(self._segments) > 0 + + +# ── 便捷函数：从 plan.config 快速生成 ASS ──────────────────────────────────── + + +def build_subtitles_from_plan( + output_path: Path, + plan_config: dict, + *, + video_width: int, + video_height: int, + video_duration: float, + asr_timeline: Any = None, +) -> Path | None: + """从 plan.config 构建字幕 ASS 文件. + + 支持的配置项： + - title_config: 标题配置（含 text/style） + - subtitle_config: 字幕配置（含 text/style） + - asr_subtitles: ASR 字幕开关 + 样式 + - manual_subtitles: 手动字幕片段列表 + + Returns: + 生成的 ASS 文件路径；如果没有任何字幕，返回 None + """ + engine = SubtitleRenderEngine( + video_width=video_width, + video_height=video_height, + video_duration=video_duration, + ) + + has_any = False + + # 1. 标题 + title_cfg = plan_config.get("title_config") or {} + if isinstance(title_cfg, dict): + title_text = str(title_cfg.get("text", "")) + title_enabled = title_cfg.get("enabled", True) + if title_enabled and title_text.strip(): + style_dict = title_cfg.get("style") or {} + style = SubtitleStyle.from_dict(style_dict) + # 标题默认样式：顶部、大字号、粗体 + if style.position == DEFAULT_POSITION and style.font_size == DEFAULT_FONT_SIZE: + style.position = "top_center" + style.font_size = 48 + style.bold = True + engine.add_title(title_text, style) + has_any = True + + # 2. 静态字幕 + sub_cfg = plan_config.get("subtitle_config") or {} + if isinstance(sub_cfg, dict): + sub_text = str(sub_cfg.get("text", "")) + sub_enabled = sub_cfg.get("enabled", True) + if sub_enabled and sub_text.strip(): + style_dict = sub_cfg.get("style") or {} + style = SubtitleStyle.from_dict(style_dict) + engine.add_subtitle_text(sub_text, style) + has_any = True + + # 3. ASR 自动字幕 + asr_cfg = plan_config.get("asr_subtitles") or {} + if isinstance(asr_cfg, dict) and asr_cfg.get("enabled", False): + if asr_timeline is not None: + style_dict = asr_cfg.get("style") or {} + style = SubtitleStyle.from_dict(style_dict) + engine.add_asr_timeline(asr_timeline, style) + has_any = has_any or engine.has_subtitles + + # 4. 手动字幕 + manual_segs = plan_config.get("manual_subtitles") or [] + if isinstance(manual_segs, list) and manual_segs: + style_dict = (plan_config.get("manual_subtitle_style") or {}) or {} + style = SubtitleStyle.from_dict(style_dict) + engine.add_timeline_segments(manual_segs, style) + has_any = has_any or engine.has_subtitles + + if not has_any: + return None + + return engine.generate_ass(output_path) + + +# ── FFmpeg 烧录滤镜生成 ─────────────────────────────────────────────────────── + + +def build_subtitle_filter( + ass_path: Path | str, + *, + video_input_label: str = "0:v", + output_label: str = "subtitled", +) -> str: + """生成 FFmpeg subtitles 滤镜字符串. + + Args: + ass_path: ASS 字幕文件路径 + video_input_label: 视频输入标签（如 "0:v" 或 "[v_out]"） + output_label: 输出标签 + + Returns: + filter_complex 片段，如 "[0:v]subtitles=xxx.ass[subtitled]" + """ + # FFmpeg subtitles filter 的路径需要转义： + # - Windows 路径的 \ → / + # - 冒号 : → \: + # - 单引号 ' → '\'' + safe_path = str(ass_path).replace("\\", "/").replace(":", "\\:").replace("'", "'\\''") + return f"{video_input_label}subtitles='{safe_path}'[{output_label}]" diff --git a/apps/worker/video_processing/unified_render_service.py b/apps/worker/video_processing/unified_render_service.py index 42d6f960b..c385e0439 100755 --- a/apps/worker/video_processing/unified_render_service.py +++ b/apps/worker/video_processing/unified_render_service.py @@ -340,6 +340,7 @@ class UnifiedRenderService: else: config = self.plan.config or {} bgm_config = config.get("bgm", {}) or {} + audio_tracks_config = config.get("audio_tracks") or {} noise_reduction_config = config.get("audio_noise_reduction") ctx = RenderContext( work_dir=self.work_dir, @@ -352,6 +353,7 @@ class UnifiedRenderService: video_duration, bgm_path=self.bgm_path, bgm_config=bgm_config, + audio_tracks_config=audio_tracks_config, ) t_audio_end = time.time() audio_mix_ms = int((t_audio_end - t_audio_start) * 1000) diff --git a/tests/unit/test_multi_track_subtitle_concat.py b/tests/unit/test_multi_track_subtitle_concat.py new file mode 100755 index 000000000..3d80e565d --- /dev/null +++ b/tests/unit/test_multi_track_subtitle_concat.py @@ -0,0 +1,1005 @@ +"""多轨道混音 + 字幕渲染引擎 + 视频拼接 单元测试. + +测试： +1. 多轨道混音：配置解析、轨道预处理、多轨混音、降级 +2. 字幕渲染引擎：样式配置、ASS生成、多源字幕合并、滤镜构建 +3. 视频拼接：配置解析、stream copy、concat filter、降级 +""" + +import sys +import tempfile +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parents[2] / "apps" / "worker")) +sys.path.insert(0, str(Path(__file__).resolve().parents[2] / "apps" / "api")) + +import pytest + +# ── Fixtures ────────────────────────────────────────────────────────────────── + + +@pytest.fixture +def work_dir(tmp_path): + return tmp_path + + +@pytest.fixture +def main_audio_path(work_dir): + """生成 10 秒测试主音频.""" + import subprocess + + path = work_dir / "main.aac" + subprocess.run( + [ + "ffmpeg", + "-y", + "-f", + "lavfi", + "-i", + "sine=frequency=440:duration=10:sample_rate=44100", + "-c:a", + "aac", + "-b:a", + "128k", + str(path), + ], + capture_output=True, + check=True, + timeout=30, + ) + return path + + +@pytest.fixture +def sfx_audio_path(work_dir): + """生成 3 秒效音频.""" + import subprocess + + path = work_dir / "sfx.aac" + subprocess.run( + [ + "ffmpeg", + "-y", + "-f", + "lavfi", + "-i", + "sine=frequency=880:duration=3:sample_rate=44100", + "-c:a", + "aac", + "-b:a", + "128k", + str(path), + ], + capture_output=True, + check=True, + timeout=30, + ) + return path + + +@pytest.fixture +def voiceover_audio_path(work_dir): + """生成 5 秒配音音频.""" + import subprocess + + path = work_dir / "voiceover.aac" + subprocess.run( + [ + "ffmpeg", + "-y", + "-f", + "lavfi", + "-i", + "sine=frequency=220:duration=5:sample_rate=44100", + "-c:a", + "aac", + "-b:a", + "128k", + str(path), + ], + capture_output=True, + check=True, + timeout=30, + ) + return path + + +@pytest.fixture +def test_video_1(work_dir): + """生成 5 秒测试视频1（1080x1920, 30fps）.""" + import subprocess + + path = work_dir / "video1.mp4" + subprocess.run( + [ + "ffmpeg", + "-y", + "-f", + "lavfi", + "-i", + "color=c=red:s=1080x1920:d=5:r=30", + "-f", + "lavfi", + "-i", + "sine=frequency=440:duration=5:sample_rate=44100", + "-c:v", + "libx264", + "-preset", + "ultrafast", + "-c:a", + "aac", + "-b:a", + "128k", + "-shortest", + str(path), + ], + capture_output=True, + check=True, + timeout=60, + ) + return path + + +@pytest.fixture +def test_video_2(work_dir): + """生成 5 秒测试视频2（1080x1920, 30fps）.""" + import subprocess + + path = work_dir / "video2.mp4" + subprocess.run( + [ + "ffmpeg", + "-y", + "-f", + "lavfi", + "-i", + "color=c=blue:s=1080x1920:d=5:r=30", + "-f", + "lavfi", + "-i", + "sine=frequency=660:duration=5:sample_rate=44100", + "-c:v", + "libx264", + "-preset", + "ultrafast", + "-c:a", + "aac", + "-b:a", + "128k", + "-shortest", + str(path), + ], + capture_output=True, + check=True, + timeout=60, + ) + return path + + +# ============================================================================ +# 一、多轨道混音测试 +# ============================================================================ + + +class TestAudioTrack: + """AudioTrack 配置解析测试.""" + + def test_default_values(self): + from video_processing.multi_track_mixer import AudioTrack + + track = AudioTrack.from_dict({"track_id": "t1", "audio_path": "/tmp/test.aac"}) + assert track.track_id == "t1" + assert track.audio_path == "/tmp/test.aac" + assert track.volume == pytest.approx(0.7) # sfx 默认音量 + assert track.track_type == "sfx" + assert track.enabled is True + assert track.fade_in == 0.0 + assert track.start_time == 0.0 + + def test_volume_clamping(self): + from video_processing.multi_track_mixer import AudioTrack + + track = AudioTrack.from_dict( + { + "track_id": "t1", + "audio_path": "/tmp/test.aac", + "volume": 5.0, + } + ) + assert track.volume == pytest.approx(2.0) # 上限钳制 + + track2 = AudioTrack.from_dict( + { + "track_id": "t2", + "audio_path": "/tmp/test.aac", + "volume": -1.0, + } + ) + assert track2.volume == pytest.approx(0.0) # 下限钳制 + + def test_track_type_default_volume(self): + from video_processing.multi_track_mixer import DEFAULT_VOLUMES, AudioTrack + + for track_type, expected_vol in DEFAULT_VOLUMES.items(): + track = AudioTrack.from_dict( + { + "track_id": "t1", + "track_type": track_type, + "audio_path": "/tmp/test.aac", + } + ) + assert track.volume == pytest.approx(expected_vol) + + def test_invalid_config_safe(self): + from video_processing.multi_track_mixer import AudioTrack + + # 无效值应该安全降级到默认值 + track = AudioTrack.from_dict( + { + "track_id": "t1", + "audio_path": "/tmp/test.aac", + "volume": "invalid", + "fade_in": "abc", + "start_time": None, + } + ) + assert track.volume > 0 # 有默认值 + assert track.fade_in == 0.0 + assert track.start_time == 0.0 + + def test_disabled_track(self): + from video_processing.multi_track_mixer import AudioTrack + + track = AudioTrack.from_dict( + { + "track_id": "t1", + "audio_path": "/tmp/test.aac", + "enabled": False, + } + ) + assert track.enabled is False + + +class TestMultiTrackMixConfig: + """MultiTrackMixConfig 配置解析测试.""" + + def test_empty_config(self): + from video_processing.multi_track_mixer import MultiTrackMixConfig + + config = MultiTrackMixConfig.from_config_dict(None) + assert config.has_effect is False + assert len(config.tracks) == 0 + + def test_empty_dict(self): + from video_processing.multi_track_mixer import MultiTrackMixConfig + + config = MultiTrackMixConfig.from_config_dict({}) + assert config.has_effect is False + assert len(config.tracks) == 0 + + def test_valid_tracks(self): + from video_processing.multi_track_mixer import MultiTrackMixConfig + + config = MultiTrackMixConfig.from_config_dict( + { + "tracks": [ + {"track_id": "sfx1", "track_type": "sfx", "audio_path": "/tmp/sfx1.aac", "volume": 0.5}, + {"track_id": "vo1", "track_type": "voiceover", "audio_path": "/tmp/vo1.aac"}, + ], + "master_volume": 0.8, + } + ) + assert config.has_effect is True + assert len(config.tracks) == 2 + assert config.tracks[0].volume == pytest.approx(0.5) + assert config.master_volume == pytest.approx(0.8) + + def test_skip_invalid_tracks(self): + from video_processing.multi_track_mixer import MultiTrackMixConfig + + config = MultiTrackMixConfig.from_config_dict( + { + "tracks": [ + {"track_id": "valid", "audio_path": "/tmp/valid.aac"}, + {"track_id": "no_path"}, # 没有 audio_path，应该跳过 + "not_a_dict", # 不是字典，应该跳过 + {"track_id": "disabled", "audio_path": "/tmp/dis.aac", "enabled": False}, + ], + } + ) + # 只有 valid 一个有效（disabled 的也跳过） + assert len([t for t in config.tracks if t.enabled]) == 1 + + +class TestMultiTrackMix: + """多轨道混音集成测试.""" + + def test_mix_two_tracks(self, work_dir, main_audio_path, sfx_audio_path): + """主音频 + 音效轨混音.""" + from video_processing.multi_track_mixer import AudioTrack, MultiTrackMixConfig, mix_multi_track + from video_processing.render_audio import RenderContext + + ctx = RenderContext(work_dir=work_dir, plan_id="test") + + config = MultiTrackMixConfig( + tracks=[ + AudioTrack( + track_id="sfx1", + track_type="sfx", + audio_path=str(sfx_audio_path), + volume=0.5, + start_time=2.0, + ), + ], + master_volume=1.0, + ) + + output = mix_multi_track(ctx, main_audio_path, config, target_duration=10.0) + assert output.exists() + assert output.stat().st_size > 0 + + def test_mix_with_voiceover(self, work_dir, main_audio_path, voiceover_audio_path): + """主音频 + 配音轨混音.""" + from video_processing.multi_track_mixer import AudioTrack, MultiTrackMixConfig, mix_multi_track + from video_processing.render_audio import RenderContext + + ctx = RenderContext(work_dir=work_dir, plan_id="test2") + + config = MultiTrackMixConfig( + tracks=[ + AudioTrack( + track_id="vo1", + track_type="voiceover", + audio_path=str(voiceover_audio_path), + volume=1.0, + start_time=1.0, + fade_in=0.5, + fade_out=0.5, + ), + ], + ) + + output = mix_multi_track(ctx, main_audio_path, config, target_duration=10.0) + assert output.exists() + assert output.stat().st_size > 0 + + def test_no_tracks_returns_main(self, work_dir, main_audio_path): + """没有附加轨道时返回主音频副本.""" + from video_processing.multi_track_mixer import MultiTrackMixConfig, mix_multi_track + from video_processing.render_audio import RenderContext + + ctx = RenderContext(work_dir=work_dir, plan_id="test3") + config = MultiTrackMixConfig(tracks=[]) + + output = mix_multi_track(ctx, main_audio_path, config, target_duration=10.0) + assert output.exists() + assert output.stat().st_size > 0 + + def test_mix_with_fade(self, work_dir, main_audio_path, sfx_audio_path): + """带淡入淡出的混音.""" + from video_processing.multi_track_mixer import AudioTrack, MultiTrackMixConfig, mix_multi_track + from video_processing.render_audio import RenderContext + + ctx = RenderContext(work_dir=work_dir, plan_id="test_fade") + + config = MultiTrackMixConfig( + tracks=[ + AudioTrack( + track_id="sfx_fade", + track_type="sfx", + audio_path=str(sfx_audio_path), + fade_in=0.3, + fade_out=0.3, + start_time=1.0, + ), + ], + ) + + output = mix_multi_track(ctx, main_audio_path, config, target_duration=10.0) + assert output.exists() + assert output.stat().st_size > 0 + + def test_mix_audio_tracks_from_config(self, work_dir, main_audio_path, sfx_audio_path): + """从配置字典混音的便捷函数.""" + from video_processing.multi_track_mixer import mix_audio_tracks_from_config + from video_processing.render_audio import RenderContext + + ctx = RenderContext(work_dir=work_dir, plan_id="test_config") + + config_dict = { + "tracks": [ + { + "track_id": "sfx1", + "track_type": "sfx", + "audio_path": str(sfx_audio_path), + "volume": 0.6, + "start_time": 1.0, + }, + ], + "enabled": True, + } + + output = mix_audio_tracks_from_config(ctx, main_audio_path, config_dict, 10.0) + assert output.exists() + assert output.stat().st_size > 0 + + def test_disabled_config_returns_main(self, work_dir, main_audio_path): + """配置未启用时返回主音频.""" + from video_processing.multi_track_mixer import mix_audio_tracks_from_config + from video_processing.render_audio import RenderContext + + ctx = RenderContext(work_dir=work_dir, plan_id="test_disabled") + + output = mix_audio_tracks_from_config(ctx, main_audio_path, None, 10.0) + assert output == main_audio_path # 直接返回原文件 + + +# ============================================================================ +# 二、字幕渲染引擎测试 +# ============================================================================ + + +class TestSubtitleStyle: + """SubtitleStyle 配置解析测试.""" + + def test_default_style(self): + from video_processing.subtitle_render_engine import SubtitleStyle + + style = SubtitleStyle.from_dict({}) + assert style.font_size == 24 + assert style.position == "bottom_center" + assert style.stroke_enabled is True + assert style.background_enabled is False + assert style.alignment == 2 # bottom_center → ASS alignment 2 + + def test_position_aliases(self): + from video_processing.subtitle_render_engine import SubtitleStyle + + style = SubtitleStyle.from_dict({"position": "top"}) + assert style.position == "top_center" + assert style.alignment == 8 + + style2 = SubtitleStyle.from_dict({"position": "bottom"}) + assert style2.position == "bottom_center" + assert style2.alignment == 2 + + style3 = SubtitleStyle.from_dict({"position": "center"}) + assert style3.position == "center" + assert style3.alignment == 5 + + def test_9grid_positions(self): + from video_processing.subtitle_render_engine import POSITION_ALIGNMENT, SubtitleStyle + + for pos, align in POSITION_ALIGNMENT.items(): + style = SubtitleStyle.from_dict({"position": pos}) + assert style.position == pos + assert style.alignment == align + + def test_invalid_position_fallback(self): + from video_processing.subtitle_render_engine import SubtitleStyle + + style = SubtitleStyle.from_dict({"position": "invalid_position"}) + assert style.position == "bottom_center" # 降级到默认 + + def test_background_style(self): + from video_processing.subtitle_render_engine import SubtitleStyle + + style = SubtitleStyle.from_dict( + { + "background_enabled": True, + "background_color": "#000000", + "background_opacity": 0.7, + } + ) + assert style.background_enabled is True + assert style.background_opacity == pytest.approx(0.7) + + def test_color_conversion(self): + from video_processing.subtitle_render_engine import SubtitleStyle + + style = SubtitleStyle.from_dict({"color": "#FF0000"}) + # #FF0000 → &H000000FF (ASS 格式: &HAABBGGRR) + assert "FF" in style.ass_font_color + assert "0000" in style.ass_font_color # BB 和 GG 都是 00 + + def test_safe_type_conversion(self): + from video_processing.subtitle_render_engine import SubtitleStyle + + style = SubtitleStyle.from_dict( + { + "size": "invalid", + "margin_v": None, + "bold": "true", # 字符串真值 + } + ) + assert style.font_size == 24 # 降级到默认 + assert style.margin_v == 60 + # bool("true") = True，但这是 Python 行为，可以接受 + + +class TestSubtitleRenderEngine: + """字幕渲染引擎测试.""" + + def test_empty_engine(self, work_dir): + """空引擎生成空文件.""" + from video_processing.subtitle_render_engine import SubtitleRenderEngine + + engine = SubtitleRenderEngine(video_width=1080, video_height=1920, video_duration=10.0) + assert engine.has_subtitles is False + + output = work_dir / "empty.ass" + engine.generate_ass(output) + assert output.exists() + assert output.read_text(encoding="utf-8") == "" + + def test_add_title(self, work_dir): + """添加标题字幕.""" + from video_processing.subtitle_render_engine import SubtitleRenderEngine, SubtitleStyle + + engine = SubtitleRenderEngine(video_width=1080, video_height=1920, video_duration=10.0) + engine.add_title("测试标题") + assert engine.has_subtitles is True + + output = work_dir / "title.ass" + result = engine.generate_ass(output) + assert result.exists() + content = result.read_text(encoding="utf-8") + assert "测试标题" in content + assert "TitleStyle" in content + + def test_add_subtitle_text(self, work_dir): + """添加整段字幕.""" + from video_processing.subtitle_render_engine import SubtitleRenderEngine, SubtitleStyle + + engine = SubtitleRenderEngine(video_width=1080, video_height=1920, video_duration=10.0) + engine.add_subtitle_text("这是一段字幕") + assert engine.has_subtitles is True + + output = work_dir / "subtitle.ass" + engine.generate_ass(output) + content = output.read_text(encoding="utf-8") + assert "这是一段字幕" in content + assert "SubtitleStyle" in content + + def test_add_timeline_segments(self, work_dir): + """添加时间轴字幕片段.""" + from video_processing.subtitle_render_engine import SubtitleRenderEngine, SubtitleStyle + + engine = SubtitleRenderEngine(video_width=1080, video_height=1920, video_duration=10.0) + segments = [ + {"start": 0.0, "end": 2.0, "text": "第一段字幕"}, + {"start": 2.0, "end": 5.0, "text": "第二段字幕"}, + {"start": 5.0, "end": 10.0, "text": "第三段字幕"}, + ] + engine.add_timeline_segments(segments) + assert engine.has_subtitles is True + + output = work_dir / "timeline.ass" + engine.generate_ass(output) + content = output.read_text(encoding="utf-8") + assert "第一段字幕" in content + assert "第二段字幕" in content + assert "第三段字幕" in content + # 检查有 3 条 Dialogue 事件 + assert content.count("Dialogue:") == 3 + + def test_mixed_sources(self, work_dir): + """标题 + 时间轴字幕 混合.""" + from video_processing.subtitle_render_engine import SubtitleRenderEngine, SubtitleStyle + + engine = SubtitleRenderEngine(video_width=1080, video_height=1920, video_duration=10.0) + engine.add_title("视频标题") + engine.add_timeline_segments( + [ + {"start": 0.0, "end": 3.0, "text": "ASR 结果1"}, + {"start": 3.0, "end": 7.0, "text": "ASR 结果2"}, + ] + ) + + output = work_dir / "mixed.ass" + engine.generate_ass(output) + content = output.read_text(encoding="utf-8") + assert "视频标题" in content + assert "ASR 结果1" in content + assert "ASR 结果2" in content + assert content.count("Dialogue:") == 3 + + def test_fade_animation(self, work_dir): + """淡入淡出动画效果.""" + from video_processing.subtitle_render_engine import SubtitleRenderEngine, SubtitleStyle + + style = SubtitleStyle(fade_in=0.5, fade_out=0.5) + engine = SubtitleRenderEngine(video_width=1080, video_height=1920, video_duration=10.0) + engine.add_timeline_segments( + [{"start": 0.0, "end": 5.0, "text": "淡入淡出测试"}], + style=style, + ) + + output = work_dir / "fade.ass" + engine.generate_ass(output) + content = output.read_text(encoding="utf-8") + assert "\\fad" in content # ASS 淡入淡出标签 + + def test_text_wrapping(self, work_dir): + """长文本自动换行.""" + from video_processing.subtitle_render_engine import SubtitleRenderEngine, SubtitleStyle + + style = SubtitleStyle(max_chars_per_line=10) + engine = SubtitleRenderEngine(video_width=1080, video_height=1920, video_duration=10.0) + engine.add_subtitle_text("这是一段非常长的字幕文本，应该会自动换行显示", style=style) + + output = work_dir / "wrap.ass" + engine.generate_ass(output) + content = output.read_text(encoding="utf-8") + assert "\\N" in content # ASS 换行符 + + def test_escape_special_chars(self, work_dir): + """ASS 特殊字符转义.""" + from video_processing.subtitle_render_engine import SubtitleRenderEngine + + engine = SubtitleRenderEngine(video_width=1080, video_height=1920, video_duration=10.0) + engine.add_subtitle_text("测试{大括号}换行\n第二行") + + output = work_dir / "escape.ass" + engine.generate_ass(output) + content = output.read_text(encoding="utf-8") + # 大括号应该被转义 + assert "{" not in content.split("Dialogue:")[1].split("测试")[1][:10] or "(" in content + assert "\\N" in content # 换行转义 + + +class TestBuildSubtitlesFromPlan: + """从 plan.config 构建字幕测试.""" + + def test_empty_config(self, work_dir): + from video_processing.subtitle_render_engine import build_subtitles_from_plan + + output = work_dir / "empty_plan.ass" + result = build_subtitles_from_plan( + output, + {}, + video_width=1080, + video_height=1920, + video_duration=10.0, + ) + assert result is None + + def test_title_only(self, work_dir): + from video_processing.subtitle_render_engine import build_subtitles_from_plan + + output = work_dir / "title_plan.ass" + config = { + "title_config": { + "enabled": True, + "text": "我的视频标题", + "style": {"size": 48, "bold": True, "position": "top"}, + } + } + result = build_subtitles_from_plan(output, config, video_width=1080, video_height=1920, video_duration=10.0) + assert result is not None + assert result.exists() + content = result.read_text(encoding="utf-8") + assert "我的视频标题" in content + + def test_manual_subtitles(self, work_dir): + from video_processing.subtitle_render_engine import build_subtitles_from_plan + + output = work_dir / "manual.ass" + config = { + "manual_subtitles": [ + {"start": 0.0, "end": 2.0, "text": "手动字幕1"}, + {"start": 2.5, "end": 5.0, "text": "手动字幕2"}, + ], + "manual_subtitle_style": {"size": 28, "color": "#FFFF00"}, + } + result = build_subtitles_from_plan(output, config, video_width=1080, video_height=1920, video_duration=10.0) + assert result is not None + content = result.read_text(encoding="utf-8") + assert "手动字幕1" in content + assert "手动字幕2" in content + assert content.count("Dialogue:") == 2 + + def test_disabled_title_skipped(self, work_dir): + from video_processing.subtitle_render_engine import build_subtitles_from_plan + + output = work_dir / "disabled.ass" + config = { + "title_config": { + "enabled": False, + "text": "不显示的标题", + } + } + result = build_subtitles_from_plan(output, config, video_width=1080, video_height=1920, video_duration=10.0) + assert result is None + + +class TestSubtitleFilter: + """字幕滤镜构建测试.""" + + def test_build_subtitle_filter(self): + from video_processing.subtitle_render_engine import build_subtitle_filter + + result = build_subtitle_filter("/tmp/test.ass", video_input_label="[v_in]", output_label="out") + assert "subtitles=" in result + assert "[v_in]" in result + assert "[out]" in result + + def test_default_labels(self): + from video_processing.subtitle_render_engine import build_subtitle_filter + + result = build_subtitle_filter("/tmp/sub.ass") + assert "0:v" in result + assert "[subtitled]" in result + + +# ============================================================================ +# 三、视频拼接引擎测试 +# ============================================================================ + + +class TestConcatSegment: + """ConcatSegment 配置解析测试.""" + + def test_default_values(self): + from video_processing.concat_engine import ConcatSegment + + seg = ConcatSegment.from_dict({"video_path": "/tmp/test.mp4"}) + assert seg.video_path == "/tmp/test.mp4" + assert seg.start_time == 0.0 + assert seg.duration == 0.0 + assert seg.has_audio is True + + def test_trimming_config(self): + from video_processing.concat_engine import ConcatSegment + + seg = ConcatSegment.from_dict( + { + "video_path": "/tmp/test.mp4", + "start_time": 5.0, + "duration": 10.0, + } + ) + assert seg.start_time == 5.0 + assert seg.duration == 10.0 + + def test_invalid_values_safe(self): + from video_processing.concat_engine import ConcatSegment + + seg = ConcatSegment.from_dict( + { + "video_path": "/tmp/test.mp4", + "start_time": "invalid", + "duration": -5.0, + } + ) + assert seg.start_time == 0.0 + assert seg.duration == 0.0 + + +class TestConcatConfig: + """ConcatConfig 配置解析测试.""" + + def test_empty_config(self): + from video_processing.concat_engine import ConcatConfig + + config = ConcatConfig.from_config_dict(None) + assert config.has_effect is False + assert config.total_segments == 0 + + def test_single_segment_no_effect(self): + from video_processing.concat_engine import ConcatConfig + + config = ConcatConfig.from_config_dict( + { + "segments": [{"video_path": "/tmp/1.mp4"}], + } + ) + assert config.has_effect is False # 只有一段不需要拼接 + + def test_multiple_segments(self): + from video_processing.concat_engine import ConcatConfig + + config = ConcatConfig.from_config_dict( + { + "segments": [ + {"video_path": "/tmp/1.mp4"}, + {"video_path": "/tmp/2.mp4"}, + {"video_path": "/tmp/3.mp4"}, + ], + "output_width": 1080, + "output_height": 1920, + "output_fps": 30.0, + } + ) + assert config.has_effect is True + assert config.total_segments == 3 + assert config.output_width == 1080 + + def test_skip_invalid_segments(self): + from video_processing.concat_engine import ConcatConfig + + config = ConcatConfig.from_config_dict( + { + "segments": [ + {"video_path": "/tmp/1.mp4"}, + {}, # 没有 path + {"video_path": ""}, # 空 path + {"video_path": "/tmp/2.mp4"}, + ], + } + ) + assert config.total_segments == 2 + assert config.has_effect is True + + +class TestConcatEngine: + """视频拼接引擎集成测试.""" + + def test_concat_demuxer_stream_copy(self, work_dir, test_video_1, test_video_2): + """concat demuxer 模式（stream copy）.""" + from video_processing.concat_engine import ConcatConfig, ConcatEngine, ConcatSegment + + engine = ConcatEngine(work_dir=work_dir) + config = ConcatConfig( + segments=[ + ConcatSegment(video_path=str(test_video_1)), + ConcatSegment(video_path=str(test_video_2)), + ], + ) + + output = work_dir / "concat_demuxer.mp4" + result = engine.concat_videos(config, output) + assert result.exists() + assert result.stat().st_size > 0 + + # 验证时长大约是两段之和（5+5=10秒） + from video_processing.ffmpeg_utils import probe_duration + + dur = probe_duration(str(result)) + assert dur > 8.0 # 留一些误差余量 + assert dur < 12.0 + + def test_concat_filter_reencode(self, work_dir, test_video_1, test_video_2): + """concat filter 模式（强制重新编码）.""" + from video_processing.concat_engine import ConcatConfig, ConcatEngine, ConcatSegment + + engine = ConcatEngine(work_dir=work_dir) + config = ConcatConfig( + segments=[ + ConcatSegment(video_path=str(test_video_1)), + ConcatSegment(video_path=str(test_video_2)), + ], + force_reencode=True, + ) + + output = work_dir / "concat_filter.mp4" + result = engine.concat_videos(config, output) + assert result.exists() + assert result.stat().st_size > 0 + + from video_processing.ffmpeg_utils import probe_duration + + dur = probe_duration(str(result)) + assert dur > 8.0 + assert dur < 12.0 + + def test_concat_with_trimming(self, work_dir, test_video_1, test_video_2): + """带裁剪的拼接（自动用 filter 模式）.""" + from video_processing.concat_engine import ConcatConfig, ConcatEngine, ConcatSegment + + engine = ConcatEngine(work_dir=work_dir) + config = ConcatConfig( + segments=[ + ConcatSegment(video_path=str(test_video_1), start_time=1.0, duration=2.0), + ConcatSegment(video_path=str(test_video_2), start_time=0.0, duration=3.0), + ], + ) + + output = work_dir / "concat_trimmed.mp4" + result = engine.concat_videos(config, output) + assert result.exists() + assert result.stat().st_size > 0 + + from video_processing.ffmpeg_utils import probe_duration + + dur = probe_duration(str(result)) + assert dur > 3.0 # 2+3=5秒 + assert dur < 7.0 + + def test_single_segment_copy(self, work_dir, test_video_1): + """单片段直接复制.""" + from video_processing.concat_engine import ConcatConfig, ConcatEngine, ConcatSegment + + engine = ConcatEngine(work_dir=work_dir) + config = ConcatConfig( + segments=[ConcatSegment(video_path=str(test_video_1))], + ) + + output = work_dir / "single.mp4" + result = engine.concat_videos(config, output) + assert result.exists() + + def test_concat_video_files_helper(self, work_dir, test_video_1, test_video_2): + """便捷函数 concat_video_files.""" + from video_processing.concat_engine import concat_video_files + + output = work_dir / "concat_helper.mp4" + result = concat_video_files( + [str(test_video_1), str(test_video_2)], + output, + work_dir=work_dir, + ) + assert result.exists() + assert result.stat().st_size > 0 + + def test_concat_from_config(self, work_dir, test_video_1, test_video_2): + """从配置字典拼接的便捷函数.""" + from video_processing.concat_engine import concat_videos_from_config + + output = work_dir / "concat_config.mp4" + config_dict = { + "segments": [ + {"video_path": str(test_video_1)}, + {"video_path": str(test_video_2)}, + ], + } + result = concat_videos_from_config(config_dict, output, work_dir=work_dir) + assert result is not None + assert result.exists() + + def test_concat_empty_config_returns_none(self, work_dir): + """空配置返回 None.""" + from video_processing.concat_engine import concat_videos_from_config + + output = work_dir / "empty.mp4" + result = concat_videos_from_config(None, output, work_dir=work_dir) + assert result is None + + def test_three_videos_concat(self, work_dir, test_video_1, test_video_2): + """三段视频拼接.""" + from video_processing.concat_engine import ConcatConfig, ConcatEngine, ConcatSegment + + engine = ConcatEngine(work_dir=work_dir) + config = ConcatConfig( + segments=[ + ConcatSegment(video_path=str(test_video_1)), + ConcatSegment(video_path=str(test_video_2)), + ConcatSegment(video_path=str(test_video_1)), + ], + force_reencode=True, + ) + + output = work_dir / "three_videos.mp4" + result = engine.concat_videos(config, output) + assert result.exists() + + from video_processing.ffmpeg_utils import probe_duration + + dur = probe_duration(str(result)) + assert dur > 12.0 # 5+5+5=15秒 + assert dur < 18.0 + + def test_output_resolution_override(self, work_dir, test_video_1, test_video_2): + """指定输出分辨率.""" + from video_processing.concat_engine import ConcatConfig, ConcatEngine, ConcatSegment + + engine = ConcatEngine(work_dir=work_dir) + config = ConcatConfig( + segments=[ + ConcatSegment(video_path=str(test_video_1)), + ConcatSegment(video_path=str(test_video_2)), + ], + output_width=720, + output_height=1280, + force_reencode=True, + ) + + output = work_dir / "concat_720p.mp4" + result = engine.concat_videos(config, output) + assert result.exists() + + from video_processing.ffmpeg_utils import probe_video_info + + info = probe_video_info(str(result)) + assert int(info.get("width", 0)) == 720 + assert int(info.get("height", 0)) == 1280