diff --git a/apps/worker/video_processing/__init__.py b/apps/worker/video_processing/__init__.py index 30465a107..09de09a36 100644 --- a/apps/worker/video_processing/__init__.py +++ b/apps/worker/video_processing/__init__.py @@ -4,4 +4,18 @@ from .processor import VideoProcessor, VideoResult -__all__ = ["VideoProcessor", "VideoResult"] +# 共享工具模块（供 editing_modes / generation / edit_plan_generation 等复用） +from . import ffmpeg_utils +from . import oss_helpers +from . import dedup_helpers +from .unified_render_service import UnifiedRenderService, RenderResult + +__all__ = [ + "VideoProcessor", + "VideoResult", + "ffmpeg_utils", + "oss_helpers", + "dedup_helpers", + "UnifiedRenderService", + "RenderResult", +] diff --git a/apps/worker/video_processing/dedup_helpers.py b/apps/worker/video_processing/dedup_helpers.py new file mode 100644 index 000000000..7e51dce1f --- /dev/null +++ b/apps/worker/video_processing/dedup_helpers.py @@ -0,0 +1,128 @@ +"""查重辅助函数 — 从 generation.py 提取的 GeneratedVideo 记录 + 查重逻辑. + +供 render_edit_plan 和 generate_video 共同复用， +创建 GeneratedVideo 记录后计算指纹并执行项目级 + 批次内查重。 +""" + +from __future__ import annotations + +import logging +from uuid import uuid4 + +from sqlalchemy.orm import Session + +logger = logging.getLogger(__name__) + + +def create_video_record_and_dedup( + *, + generation_task_id: str, + project_id: str, + batch_id: str, + file_url: str, + file_size: int, + duration: float, + video_path: str, + mode: str, + session: Session, + width: int = 1280, + height: int = 720, + fps: float = 25.0, +) -> int: + """创建 GeneratedVideo 记录，计算指纹并执行查重（历史 + 批次）。 + + Args: + generation_task_id: 生成任务 ID + project_id: 项目 ID + batch_id: 批次 ID（可为空字符串） + file_url: 视频文件 URL + file_size: 文件大小（字节） + duration: 视频时长（秒） + video_path: 视频本地路径（用于计算指纹） + mode: 剪辑模式名称 + session: 数据库会话 + width: 视频宽度 + height: 视频高度 + fps: 视频帧率 + + Returns: + 创建的视频记录数量（1 表示成功，0 表示失败） + """ + from video_processing.dedup import VideoDeduplicator + + from packages.adapters.sqlalchemy_impl.generated_video_repository import ( + SQLAlchemyGeneratedVideoRepository, + ) + from packages.domain import GeneratedVideo + + try: + video_id = uuid4().hex + generated_video = GeneratedVideo( + id=video_id, + project_id=project_id, + generation_task_id=generation_task_id, + name=f"generated-{generation_task_id[:8]}.mp4", + file_url=file_url, + file_size=file_size, + duration=duration, + width=width, + height=height, + fps=fps, + status="completed", + generation_params={"mode": mode}, + ) + + video_repo = SQLAlchemyGeneratedVideoRepository(session) + video_repo.create(generated_video) + + # 计算视频指纹 + deduplicator = VideoDeduplicator() + try: + fingerprint = deduplicator.compute_fingerprint(video_path) + except Exception as fp_err: + logger.warning("Fingerprint computation failed for %s: %s", video_id, fp_err) + session.commit() + return 1 + + generated_video.video_fingerprint = fingerprint.to_dict() + + # (a) 历史成片查重 + duplicate_result = deduplicator.check_duplicate(fingerprint, project_id, session) + + # (b) 批次内查重（仅当有 batch_id 时） + if not duplicate_result and batch_id: + duplicate_result = deduplicator.check_batch_duplicate( + fingerprint, batch_id, video_id, session + ) + + if duplicate_result: + generated_video.is_duplicate = True + generated_video.duplicate_of = duplicate_result["duplicate_of"] + logger.info( + "Duplicate detected: %s -> %s (reason=%s, similarity=%.3f)", + video_id, + duplicate_result["duplicate_of"], + duplicate_result["reason"], + duplicate_result["similarity"], + ) + else: + generated_video.is_duplicate = False + generated_video.duplicate_of = None + + video_repo.update(generated_video) + session.commit() + logger.info( + "GeneratedVideo record created: %s (task=%s, dup=%s)", + video_id, + generation_task_id, + generated_video.is_duplicate, + ) + return 1 + except Exception as e: + logger.error( + "Failed to create video record / dedup for task %s: %s", + generation_task_id, + e, + ) + session.rollback() + return 0 diff --git a/apps/worker/video_processing/editing_modes.py b/apps/worker/video_processing/editing_modes.py index 006037248..5c25a2d95 100644 --- a/apps/worker/video_processing/editing_modes.py +++ b/apps/worker/video_processing/editing_modes.py @@ -5,7 +5,6 @@ import logging import os -import subprocess import sys import tempfile from dataclasses import dataclass @@ -22,6 +21,8 @@ else: from pathlib import Path from typing import Optional +from video_processing.ffmpeg_utils import FFMPEG_BIN, run_ffmpeg, probe_video_info + logger = logging.getLogger(__name__) @@ -67,8 +68,6 @@ class EditingModeProcessor: """ self.config = config self.work_dir = work_dir or tempfile.gettempdir() - self._ffmpeg_bin = "ffmpeg" - self._ffprobe_bin = "ffprobe" def process( self, @@ -129,62 +128,20 @@ class EditingModeProcessor: 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 命令""" - logger.debug(f"Running FFmpeg: {' '.join(command)}") + """执行 FFmpeg 命令 — 委托给共享 ffmpeg_utils.run_ffmpeg""" 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 execution failed: {stderr}") from e + 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: - 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)), - } + 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} @@ -205,7 +162,7 @@ class EditingModeProcessor: def _normalize_video(self, input_path: str, output_path: str) -> dict: """标准化视频格式：先统一帧率，再缩放/填充""" command = [ - self._ffmpeg_bin, + FFMPEG_BIN, "-y", "-i", input_path, @@ -228,7 +185,7 @@ class EditingModeProcessor: "-an", output_path, ] - self._run_ffmpeg(command) + run_ffmpeg(command) return self._get_video_info(output_path) def _one_take(self, video_paths: list[str], output_path: str) -> str: @@ -265,7 +222,7 @@ class EditingModeProcessor: offset1 = durations[0] - transition / 2 command = [ - self._ffmpeg_bin, + FFMPEG_BIN, "-y", "-i", normalized_paths[0], @@ -285,7 +242,7 @@ class EditingModeProcessor: "yuv420p", output_path, ] - self._run_ffmpeg(command) + run_ffmpeg(command) return output_path else: return self._one_take_simple_concat(normalized_paths, output_path) @@ -298,7 +255,7 @@ class EditingModeProcessor: f.write(f"file '{os.path.abspath(path)}'\n") command = [ - self._ffmpeg_bin, + FFMPEG_BIN, "-y", "-f", "concat", @@ -310,7 +267,7 @@ class EditingModeProcessor: "copy", output_path, ] - self._run_ffmpeg(command) + run_ffmpeg(command) try: os.remove(concat_file) @@ -344,7 +301,7 @@ class EditingModeProcessor: 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, + FFMPEG_BIN, "-y", "-i", video_paths[1], @@ -362,11 +319,11 @@ class EditingModeProcessor: "yuv420p", temp_pip, ] - self._run_ffmpeg(command) + run_ffmpeg(command) pip_normalized_input = temp_pip else: command = [ - self._ffmpeg_bin, + FFMPEG_BIN, "-y", "-i", video_paths[1], @@ -382,13 +339,13 @@ class EditingModeProcessor: "yuv420p", pip_normalized, ] - self._run_ffmpeg(command) + 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, + FFMPEG_BIN, "-y", "-stream_loop", "-1", @@ -408,11 +365,11 @@ class EditingModeProcessor: "yuv420p", looped_pip, ] - self._run_ffmpeg(command) + run_ffmpeg(command) pip_normalized_input = looped_pip command = [ - self._ffmpeg_bin, + FFMPEG_BIN, "-y", "-i", main_normalized, @@ -432,7 +389,7 @@ class EditingModeProcessor: "yuv420p", output_path, ] - self._run_ffmpeg(command) + run_ffmpeg(command) for temp_file in [main_normalized, pip_normalized]: if temp_file and temp_file != output_path: @@ -462,7 +419,7 @@ class EditingModeProcessor: if bg_info["duration"] < audio_duration: looped_bg = os.path.join(self.work_dir, f"bg_looped_{os.getpid()}.mp4") command = [ - self._ffmpeg_bin, + FFMPEG_BIN, "-y", "-stream_loop", "-1", @@ -482,12 +439,12 @@ class EditingModeProcessor: "yuv420p", looped_bg, ] - self._run_ffmpeg(command) + 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, + FFMPEG_BIN, "-y", "-i", bg_normalized, @@ -497,12 +454,12 @@ class EditingModeProcessor: "copy", temp_bg, ] - self._run_ffmpeg(command) + 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, + FFMPEG_BIN, "-y", "-i", bg_normalized, @@ -518,10 +475,10 @@ class EditingModeProcessor: "yuv420p", blurred_bg, ] - self._run_ffmpeg(command) + run_ffmpeg(command) command = [ - self._ffmpeg_bin, + FFMPEG_BIN, "-y", "-i", blurred_bg, @@ -544,7 +501,7 @@ class EditingModeProcessor: "-shortest", output_path, ] - self._run_ffmpeg(command) + run_ffmpeg(command) for temp_file in [bg_normalized, blurred_bg]: try: @@ -582,7 +539,7 @@ class EditingModeProcessor: voice_adjusted = os.path.join(self.work_dir, f"voice_adj_{os.getpid()}.mp4") command = [ - self._ffmpeg_bin, + FFMPEG_BIN, "-y", "-i", voice_normalized, @@ -600,11 +557,11 @@ class EditingModeProcessor: "yuv420p", voice_adjusted, ] - self._run_ffmpeg(command) + run_ffmpeg(command) bg_adjusted = os.path.join(self.work_dir, f"bg_adj_{os.getpid()}.mp4") command = [ - self._ffmpeg_bin, + FFMPEG_BIN, "-y", "-i", bg_normalized, @@ -614,11 +571,11 @@ class EditingModeProcessor: "copy", bg_adjusted, ] - self._run_ffmpeg(command) + run_ffmpeg(command) if audio_path: command = [ - self._ffmpeg_bin, + FFMPEG_BIN, "-y", "-i", bg_adjusted, @@ -645,7 +602,7 @@ class EditingModeProcessor: ] else: command = [ - self._ffmpeg_bin, + FFMPEG_BIN, "-y", "-i", bg_adjusted, @@ -668,7 +625,7 @@ class EditingModeProcessor: "yuv420p", output_path, ] - self._run_ffmpeg(command) + run_ffmpeg(command) for temp_file in [voice_normalized, voice_adjusted, bg_normalized, bg_adjusted]: try: diff --git a/apps/worker/video_processing/ffmpeg_utils.py b/apps/worker/video_processing/ffmpeg_utils.py new file mode 100644 index 000000000..3fd7b6ac0 --- /dev/null +++ b/apps/worker/video_processing/ffmpeg_utils.py @@ -0,0 +1,289 @@ +"""FFmpeg 工具函数 — 从 editing_modes.py / video_compose_service.py 提取的共享原语. + +提供 FFmpeg / FFprobe 调用、视频信息探测、视频标准化、xfade 转场滤镜构建 +等底层能力，供 EditingModeProcessor、VideoComposeService、UnifiedRenderService +共同复用。 +""" + +from __future__ import annotations + +import logging +import shutil +import subprocess # nosec B404 +from pathlib import Path +from typing import Any + +logger = logging.getLogger(__name__) + +# ── 常量 ────────────────────────────────────────────────────────────────────── + +FFMPEG_BIN: str = shutil.which("ffmpeg") or "ffmpeg" +FFPROBE_BIN: str = shutil.which("ffprobe") or "ffprobe" + +DEFAULT_OUTPUT_WIDTH = 1280 +DEFAULT_OUTPUT_HEIGHT = 720 +DEFAULT_FPS = 25 + +# xfade 转场映射：transition_effect 名称 → FFmpeg xfade transition 名称 +# 键同时支持 TransitionEffect 枚举值和字符串名称（向后兼容） +XFADE_TRANSITION_MAP: dict[str, str] = { + "fade": "fade", + "slideleft": "slideleft", + "slide_left": "slideleft", + "slideright": "slideright", + "slide_right": "slideright", + "dissolve": "dissolve", + "wipe": "wipeleft", + "wipeleft": "wipeleft", +} + +DEFAULT_TRANSITION_DURATION = 0.5 + + +# ── FFmpeg 执行 ─────────────────────────────────────────────────────────────── + + +def run_ffmpeg( + command: list[str], + *, + capture_output: bool = True, +) -> tuple[str, str]: + """执行 FFmpeg 命令。 + + Args: + command: 完整的 ffmpeg 命令列表（含 "ffmpeg" 本身） + capture_output: 是否捕获 stdout/stderr + + Returns: + (stdout, stderr) 元组 + + Raises: + subprocess.CalledProcessError: 命令执行失败时抛出 + """ + result = subprocess.run( # nosec B603 + command, + check=True, + stdout=subprocess.PIPE if capture_output else None, + stderr=subprocess.PIPE if capture_output else None, + text=True, + ) + return (result.stdout or "", result.stderr or "") + + +def probe_duration(local_path: str | Path) -> float: + """用 ffprobe 获取视频时长（秒）。 + + 失败时返回默认值 5.0 秒。 + """ + try: + result = subprocess.run( # nosec B603 + [ + FFPROBE_BIN, + "-v", + "error", + "-show_entries", + "format=duration", + "-of", + "default=noprint_wrappers=1:nokey=1", + str(local_path), + ], + check=True, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + text=True, + ) + return round(float(result.stdout.strip()), 3) + except Exception: + return 5.0 + + +def probe_video_info(video_path: str) -> dict[str, Any]: + """获取视频信息（宽、高、时长、fps）。 + + Returns: + {"width": int, "height": int, "duration": float, "fps": float} + 失败时返回默认值。 + """ + try: + result = subprocess.run( # nosec B603 + [ + FFPROBE_BIN, + "-v", + "error", + "-select_streams", + "v:0", + "-show_entries", + "stream=width,height,r_frame_rate,duration", + "-show_entries", + "format=duration", + "-of", + "json", + video_path, + ], + check=True, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + text=True, + ) + + import json + + info = json.loads(result.stdout) + stream = info.get("streams", [{}])[0] + fmt = info.get("format", {}) + + width = int(stream.get("width", DEFAULT_OUTPUT_WIDTH)) + height = int(stream.get("height", DEFAULT_OUTPUT_HEIGHT)) + + # 解析帧率 + fps_str = stream.get("r_frame_rate", "25/1") + if "/" in fps_str: + num, den = fps_str.split("/") + fps = float(num) / float(den) if float(den) > 0 else DEFAULT_FPS + else: + fps = float(fps_str) if fps_str else DEFAULT_FPS + + # 时长 + duration = float(fmt.get("duration", 0)) or float(stream.get("duration", 0)) + + return { + "width": width, + "height": height, + "duration": duration, + "fps": round(fps, 2), + } + except Exception as e: + logger.warning("获取视频信息失败: %s, error: %s", video_path, e) + return { + "width": DEFAULT_OUTPUT_WIDTH, + "height": DEFAULT_OUTPUT_HEIGHT, + "duration": 0.0, + "fps": DEFAULT_FPS, + } + + +def normalize_video( + input_path: str, + output_path: str, + *, + width: int = DEFAULT_OUTPUT_WIDTH, + height: int = DEFAULT_OUTPUT_HEIGHT, + fps: int = DEFAULT_FPS, +) -> dict[str, Any]: + """标准化视频（缩放 + 恒定帧率）。 + + 使用 scale + pad 保持宽高比，黑边填充到目标分辨率。 + + Returns: + {"width": int, "height": int, "path": str} + """ + command = [ + FFMPEG_BIN, + "-y", + "-i", + input_path, + "-vf", + f"scale={width}:{height}:force_original_aspect_ratio=decrease," + f"pad={width}:{height}:(ow-iw)/2:(oh-ih)/2:black," + f"fps={fps}", + "-c:v", + "libx264", + "-crf", + "23", + "-preset", + "medium", + "-c:a", + "aac", + "-b:a", + "128k", + "-movflags", + "+faststart", + output_path, + ] + run_ffmpeg(command) + return {"width": width, "height": height, "path": output_path} + + +# ── xfade / concat 滤镜构建 ────────────────────────────────────────────────── + + +def chain_filters(filters: list[str], output_label: str, *, input_label: str = "0:v") -> str: + """将滤镜列表串联为 FFmpeg 滤镜字符串。 + + 例：chain_filters(["scale=1280:720", "fps=25"], "v0") + → "[0:v]scale=1280:720,fps=25[v0]" + """ + filter_body = ",".join(filters) + return f"[{input_label}]{filter_body}[{output_label}]" + + +def resolve_xfade_transition(transition_name: str) -> str: + """将转场效果名称映射为 FFmpeg xfade transition 名称。 + + 支持 TransitionEffect 枚举值和字符串名称，未知值回退到 "fade"。 + """ + # 兼容 TransitionEffect 枚举（有 .value 属性） + if hasattr(transition_name, "value"): + transition_name = transition_name.value + return XFADE_TRANSITION_MAP.get(transition_name, "fade") + + +def build_xfade_filter_chain( + clip_durations: list[float], + clip_video_labels: list[str], + transitions: list[str], + *, + transition_duration: float = DEFAULT_TRANSITION_DURATION, + output_label: str = "outv", +) -> tuple[str, float]: + """构建 xfade 转场滤镜链。 + + Args: + clip_durations: 每个片段的时长 + clip_video_labels: 每个片段的视频流标签（如 "v0", "v1"） + transitions: 每个片段对应的转场效果（第一个片段的转场被忽略） + transition_duration: 转场时长（秒） + output_label: 最终输出标签 + + Returns: + (filter_string, estimated_total_duration) + """ + n = len(clip_durations) + parts: list[str] = [] + total_duration = sum(clip_durations) + + if n == 0: + return "", 0.0 + + if n == 1: + parts.append(f"[{clip_video_labels[0]}]copy[{output_label}]") + return ";".join(parts), total_duration + + # xfade 链 + cumulative = 0.0 + prev_label = clip_video_labels[0] + + for i in range(1, n): + cumulative += clip_durations[i - 1] + offset = max(0.0, cumulative - transition_duration * i) + + transition = transitions[i] if i < len(transitions) else "cut" + xfade_transition = resolve_xfade_transition(transition) + + if i == n - 1: + out_label = output_label + else: + out_label = f"xf{i}" + + parts.append( + f"[{prev_label}][{clip_video_labels[i]}]" + f"xfade=transition={xfade_transition}" + f":duration={transition_duration}" + f":offset={offset:.3f}" + f"[{out_label}]" + ) + prev_label = out_label + + # 总时长减去转场重叠部分 + total_duration -= transition_duration * (n - 1) + return ";".join(parts), max(0.0, total_duration) diff --git a/apps/worker/video_processing/oss_helpers.py b/apps/worker/video_processing/oss_helpers.py new file mode 100644 index 000000000..18e6a1674 --- /dev/null +++ b/apps/worker/video_processing/oss_helpers.py @@ -0,0 +1,164 @@ +"""OSS 工具函数 — 从 generation.py / edit_plan_generation.py 提取的共享 OSS 操作. + +提供 OSS 配置读取、Bucket 创建、素材上传/下载、asset_id → 本地路径解析 +等能力，供 render_edit_plan 和 generate_video 共同复用。 +""" + +from __future__ import annotations + +import hashlib +import logging +import os +from pathlib import Path +from typing import Optional +from urllib.parse import urlparse + +import oss2 + +logger = logging.getLogger(__name__) + + +# ── OSS 配置 ────────────────────────────────────────────────────────────────── + + +def oss_settings() -> tuple[str, str, str, str] | None: + """获取 OSS 配置。 + + Returns: + (access_key_id, access_key_secret, endpoint, bucket_name) 元组， + 配置缺失时返回 None。 + """ + access_key_id = os.getenv("OSS_ACCESS_KEY_ID") + access_key_secret = os.getenv("OSS_ACCESS_KEY_SECRET") + endpoint = os.getenv("OSS_ENDPOINT") + bucket_name = os.getenv("OSS_BUCKET_NAME") + if not all([access_key_id, access_key_secret, endpoint, bucket_name]): + return None + return access_key_id, access_key_secret, endpoint, bucket_name + + +def oss_bucket() -> oss2.Bucket | None: + """获取 OSS Bucket 实例。 + + Returns: + oss2.Bucket 实例，配置缺失时返回 None。 + """ + settings = oss_settings() + if settings is None: + return None + access_key_id, access_key_secret, endpoint, bucket_name = settings + return oss2.Bucket(oss2.Auth(access_key_id, access_key_secret), endpoint, bucket_name) + + +def normalize_storage_key(storage_key_or_url: str) -> str: + """标准化存储键 — 如果是完整 URL 则提取 path 部分。 + + Examples: + "https://bucket.oss-cn-hangzhou.aliyuncs.com/path/to/file.mp4" + → "path/to/file.mp4" + "path/to/file.mp4" → "path/to/file.mp4" + """ + if storage_key_or_url.startswith(("http://", "https://")): + return urlparse(storage_key_or_url).path.lstrip("/") + return storage_key_or_url.lstrip("/") + + +# ── 上传 / 下载 ─────────────────────────────────────────────────────────────── + + +def download_asset(asset_storage_key: str, local_path: Path) -> bool: + """从 OSS 下载素材文件到本地路径。 + + Args: + asset_storage_key: 素材的存储键（或完整 URL） + local_path: 本地保存路径 + + Returns: + True 表示下载成功，False 表示失败。 + """ + bucket = oss_bucket() + if bucket is None: + return False + try: + bucket.get_object_to_file(normalize_storage_key(asset_storage_key), str(local_path)) + return local_path.exists() and local_path.stat().st_size > 0 + except Exception: + logger.exception("下载素材失败: %s", asset_storage_key) + return False + + +def upload_to_oss(local_path: Path, storage_key: str) -> str | None: + """上传文件到 OSS，返回公开 URL。 + + Args: + local_path: 本地文件路径 + storage_key: 目标存储键 + + Returns: + 公开访问 URL，上传失败或 OSS 未配置时返回 None。 + """ + bucket = oss_bucket() + if bucket is None: + return None + try: + bucket.put_object_from_file(storage_key, str(local_path)) + settings = oss_settings() + if settings: + _, _, endpoint, bucket_name = settings + return f"https://{bucket_name}.{endpoint.replace('https://', '').replace('http://', '')}/{storage_key}" + return None + except Exception: + logger.exception("上传 OSS 失败: %s", storage_key) + return None + + +# ── Asset 解析 ──────────────────────────────────────────────────────────────── + + +def resolve_asset_path(asset_id: str, work_dir: Path) -> Path | None: + """从 asset_id 解析到本地文件路径。 + + 策略（按优先级）： + 1. 如果 asset_id 是本地绝对路径（/var/storage/...）→ 直接返回 + 2. 如果 work_dir 下已有缓存文件 → 返回缓存路径 + 3. 从 OSS 下载到 work_dir/{hash}.mp4 → 返回下载路径 + 4. 下载失败 → 返回 None + + 缓存策略：以 asset_id 的 SHA256 前 16 位为文件名，避免重复下载。 + """ + # 1. 本地绝对路径 + if asset_id.startswith("/") and os.path.exists(asset_id): + return Path(asset_id) + + # 2. 缓存命中 + cache_hash = hashlib.sha256(asset_id.encode()).hexdigest()[:16] + cached_path = work_dir / f"{cache_hash}.mp4" + if cached_path.exists() and cached_path.stat().st_size > 0: + return cached_path + + # 3. 从 OSS 下载 + if download_asset(asset_id, cached_path): + return cached_path + + return None + + +def resolve_asset_ids_to_paths( + asset_ids: list[str], + work_dir: Path, +) -> dict[str, Path]: + """批量解析 asset_id → 本地路径。 + + Args: + asset_ids: 素材 ID 列表 + work_dir: 工作目录 + + Returns: + {asset_id: local_path} 映射，仅包含成功解析的条目。 + """ + result: dict[str, Path] = {} + for aid in asset_ids: + local_path = resolve_asset_path(aid, work_dir) + if local_path: + result[aid] = local_path + return result diff --git a/apps/worker/video_processing/unified_render_service.py b/apps/worker/video_processing/unified_render_service.py new file mode 100644 index 000000000..f7b1ec44b --- /dev/null +++ b/apps/worker/video_processing/unified_render_service.py @@ -0,0 +1,463 @@ +"""统一渲染引擎 — 输入 EditPlan + EditPlanClips，按时间线+图层渲染视频. + +核心原则（灵应）：渲染引擎是统一的，不判断模式，只按 clip_type/config.role +分组为图层再合成。 + +图层分组： + main (无 config.role) → main (z=0) + main + config.role=b_roll → broll (z=0，与 main 同层替换) + overlay → overlay (z=1，画中画叠加) + background → background (z=0，全屏底图) + corner_voice → corner_voice (z=1，右上角小窗) + b_roll → broll (z=0) + intro / outro → main (z=0，按 order 排在首/尾) + +合成流程： + 1. 每个 clip 先 trim + scale + setpts 预处理 + 2. 同层 clips 按 order 用 xfade 串联 + 3. overlay/corner_voice 层 overlay 到主层 + 4. 如有独立音频轨，amix 混入 +""" + +from __future__ import annotations + +import logging +import os +from dataclasses import dataclass, field +from pathlib import Path +from typing import Any + +from video_processing.ffmpeg_utils import ( + FFMPEG_BIN, + DEFAULT_OUTPUT_WIDTH, + DEFAULT_OUTPUT_HEIGHT, + DEFAULT_FPS, + DEFAULT_TRANSITION_DURATION, + build_xfade_filter_chain, + probe_duration, + probe_video_info, + run_ffmpeg, +) + +logger = logging.getLogger(__name__) + + +# ── 数据结构 ────────────────────────────────────────────────────────────────── + + +@dataclass +class ResolvedClip: + """已解析到本地路径的片段。""" + + clip_id: str + asset_id: str + local_path: Path + clip_type: str + order: int + start_time: float = 0.0 + duration: float = 0.0 # 0 表示使用素材完整时长 + transition_effect: str = "cut" + config: dict[str, Any] = field(default_factory=dict) + + # 运行时填充 + actual_duration: float = 0.0 # 素材实际时长（probe 后填充） + + +@dataclass +class RenderLayer: + """渲染图层。""" + + role: str # "main" | "overlay" | "pip" | "background" | "corner_voice" | "broll" | "audio" + clips: list[ResolvedClip] = field(default_factory=list) + z_index: int = 0 + opacity: float = 1.0 + position: tuple[int, int] | None = None # (x, y) 偏移，None 表示全屏 + + +@dataclass +class RenderResult: + """渲染结果。""" + + output_path: Path + duration: float + file_size: int + width: int + height: int + + +# ── clip_type → layer role 映射 ────────────────────────────────────────────── + + +def _resolve_layer_role(clip_type: str, config: dict[str, Any]) -> str: + """根据 clip_type 和 config.role 确定图层角色。 + + 映射规则： + intro / outro → "main"（按 order 排在首/尾） + overlay → "overlay"（画中画叠加，z=1） + corner_voice → "corner_voice"（右上角小窗，z=1） + background → "background"（全屏底图，z=0） + b_roll → "broll"（z=0） + main + config.role=b_roll → "broll" + main (default) → "main" + """ + role = config.get("role", "") + + if clip_type in ("intro", "outro"): + return "main" + if clip_type == "overlay": + return "overlay" + if clip_type == "corner_voice": + return "corner_voice" + if clip_type == "background": + return "background" + if clip_type == "b_roll": + return "broll" + # main type + if role == "b_roll": + return "broll" + return "main" + + +# ── 图层默认 z_index ───────────────────────────────────────────────────────── + +_LAYER_Z_INDEX: dict[str, int] = { + "background": -1, + "broll": 0, + "main": 0, + "overlay": 1, + "corner_voice": 1, + "audio": 2, +} + +# 图层默认 PiP 位置（相对输出画布的偏移） +_PIP_SCALE = 0.25 # PiP 占主画面的比例 + + +# ── 统一渲染引擎 ───────────────────────────────────────────────────────────── + + +class UnifiedRenderService: + """统一渲染引擎。 + + 输入 EditPlan + EditPlanClips + 素材路径映射，按时间线+图层执行渲染。 + """ + + def __init__( + self, + plan: Any, # EditPlan + clips: list[Any], # list[EditPlanClip] + asset_path_map: dict[str, Path], # asset_id → local_path + work_dir: Path, + *, + output_width: int = DEFAULT_OUTPUT_WIDTH, + output_height: int = DEFAULT_OUTPUT_HEIGHT, + output_fps: int = DEFAULT_FPS, + transition_duration: float = DEFAULT_TRANSITION_DURATION, + ): + self.plan = plan + self.clips = clips + self.asset_path_map = asset_path_map + self.work_dir = work_dir + self.output_width = output_width + self.output_height = output_height + self.output_fps = output_fps + self.transition_duration = transition_duration + + def render(self) -> RenderResult: + """执行渲染，返回 RenderResult。 + + Raises: + ValueError: 没有可渲染的片段时抛出 + """ + # 1. 解析 clips → ResolvedClips（跳过无素材的 clip） + resolved = self._resolve_clips() + if not resolved: + raise ValueError("没有可渲染的片段（所有片段素材缺失或下载失败）") + + # 2. 分组为 RenderLayers + layers = self._group_clips_into_layers(resolved) + + # 3. 构建 filter_complex + 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. 探测输出 + duration, file_size, width, height = self._probe_output(output_path) + + return RenderResult( + output_path=output_path, + duration=duration, + file_size=file_size, + width=width, + height=height, + ) + + # ── 内部方法 ────────────────────────────────────────────────────────────── + + def _resolve_clips(self) -> list[ResolvedClip]: + """将 EditPlanClip 列表解析为 ResolvedClip 列表。 + + 跳过 asset_id 为空或在 asset_path_map 中找不到的片段。 + """ + resolved: list[ResolvedClip] = [] + for clip in self.clips: + asset_id = clip.asset_id + if not asset_id: + logger.warning("片段无素材: clip_id=%s", clip.id) + continue + + local_path = self.asset_path_map.get(asset_id) + if local_path is None or not local_path.exists(): + logger.warning("素材不存在: clip_id=%s asset_id=%s", clip.id, asset_id) + continue + + # 探测实际时长 + try: + actual_duration = probe_duration(local_path) + except Exception: + actual_duration = clip.duration or 5.0 + + rc = ResolvedClip( + clip_id=clip.id, + asset_id=asset_id, + local_path=local_path, + clip_type=clip.clip_type, + order=clip.order, + start_time=clip.start_time, + duration=clip.duration, + transition_effect=clip.transition_effect or "cut", + config=clip.config or {}, + actual_duration=actual_duration, + ) + resolved.append(rc) + + # 按 order 排序 + resolved.sort(key=lambda c: c.order) + return resolved + + def _group_clips_into_layers(self, resolved_clips: list[ResolvedClip]) -> list[RenderLayer]: + """将 ResolvedClips 分组为 RenderLayers。 + + 分组规则见 _resolve_layer_role 函数文档。 + """ + layer_map: dict[str, RenderLayer] = {} + + for clip in resolved_clips: + role = _resolve_layer_role(clip.clip_type, clip.config) + if role not in layer_map: + z = _LAYER_Z_INDEX.get(role, 0) + layer_map[role] = RenderLayer(role=role, z_index=z) + layer_map[role].clips.append(clip) + + # 每个 layer 内的 clips 按 order 排序 + for layer in layer_map.values(): + layer.clips.sort(key=lambda c: c.order) + + # 计算 PiP 位置 + pip_width = int(self.output_width * _PIP_SCALE) + pip_height = int(self.output_height * _PIP_SCALE) + margin = 20 # 边距 + + if "overlay" in layer_map: + layer_map["overlay"].position = ( + self.output_width - pip_width - margin, + margin, + ) + if "corner_voice" in layer_map: + layer_map["corner_voice"].position = ( + self.output_width - pip_width - margin, + margin, + ) + + # 按 z_index 排序返回 + layers = sorted(layer_map.values(), key=lambda l: l.z_index) + return layers + + def _build_filter_complex(self, layers: list[RenderLayer]) -> tuple[str, list[str]]: + """构建 FFmpeg filter_complex 字符串和输入参数列表。 + + Returns: + (filter_complex_str, input_args_list) + input_args_list 是 ["-i", path1, "-i", path2, ...] 格式 + """ + if not layers: + raise ValueError("没有可渲染的图层") + + # 收集所有 clips（按图层顺序，同层按 order） + all_clips: list[ResolvedClip] = [] + for layer in layers: + all_clips.extend(layer.clips) + + # 构建输入参数 + input_args: list[str] = [] + clip_to_input_idx: dict[str, int] = {} + for i, clip in enumerate(all_clips): + input_args.extend(["-i", str(clip.local_path)]) + clip_to_input_idx[clip.clip_id] = i + + filter_parts: list[str] = [] + + # Step 1: 预处理每个 clip — scale + setpts + # 为每个 clip 生成预处理后的标签 [v0], [v1], ... + preprocessed_labels: list[str] = [] + for i, clip in enumerate(all_clips): + label = f"v{i}" + role = _resolve_layer_role(clip.clip_type, clip.config) + + filters: list[str] = [] + + # trim（如果指定了 duration） + if clip.duration > 0 and clip.duration < clip.actual_duration: + filters.append(f"trim=duration={clip.duration}") + filters.append("setpts=PTS-STARTPTS") + + # scale + 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}") + elif role == "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}") + else: + # main / broll: scale + pad 保持宽高比 + filters.append( + f"scale={self.output_width}:{self.output_height}" ":force_original_aspect_ratio=decrease" + ) + filters.append(f"pad={self.output_width}:{self.output_height}" ":(ow-iw)/2:(oh-ih)/2:black") + + filters.append(f"fps={self.output_fps}") + filters.append("setpts=PTS-STARTPTS") + + filter_str = f"[{i}:v]{','.join(filters)}[{label}]" + filter_parts.append(filter_str) + preprocessed_labels.append(label) + + # Step 2: 同层 clips 用 xfade 串联 + layer_output_labels: dict[str, str] = {} + for layer in layers: + layer_clip_indices = [all_clips.index(c) for c in layer.clips] + layer_labels = [preprocessed_labels[i] for i in layer_clip_indices] + layer_durations = [ + all_clips[i].duration if all_clips[i].duration > 0 else all_clips[i].actual_duration + for i in layer_clip_indices + ] + layer_transitions = [all_clips[i].transition_effect for i in layer_clip_indices] + + if len(layer_labels) == 1: + # 单 clip 层，直接使用预处理标签 + layer_output_labels[layer.role] = layer_labels[0] + else: + # 多 clip 层，用 xfade 串联 + out_label = f"{layer.role}_merged" + xfade_filter, _ = build_xfade_filter_chain( + clip_durations=layer_durations, + clip_video_labels=layer_labels, + transitions=layer_transitions, + transition_duration=self.transition_duration, + output_label=out_label, + ) + if xfade_filter: + filter_parts.append(xfade_filter) + layer_output_labels[layer.role] = out_label + + # Step 3: 合成各层 + # 找到主层 — background 优先作为底图，其次 broll / main + final_video_label = None + + if "background" in layer_output_labels: + final_video_label = layer_output_labels["background"] + # b_roll / main 叠加到 background 上 + for role in ("broll", "main"): + if role in layer_output_labels: + base_label = layer_output_labels[role] + combined_label = f"combined_{role}" + filter_parts.append( + f"[{final_video_label}][{base_label}]" f"overlay=(W-w)/2:(H-h)/2[{combined_label}]" + ) + final_video_label = combined_label + else: + # 无 background 时，取 broll 或 main 作为基础 + for role in ("broll", "main"): + if role in layer_output_labels: + final_video_label = layer_output_labels[role] + break + + if final_video_label is None: + # 没有任何主层，使用第一个层 + final_video_label = layer_output_labels[layers[0].role] + + # 叠加 overlay 层 + for layer in layers: + if layer.role in ("overlay", "corner_voice"): + if layer.role not in layer_output_labels: + continue + overlay_label = layer_output_labels[layer.role] + x, y = layer.position or ( + self.output_width - int(self.output_width * _PIP_SCALE) - 20, + 20, + ) + combined_label = f"combined_{layer.role}" + 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]") + + filter_complex = ";".join(filter_parts) + return filter_complex, input_args + + def _execute_ffmpeg( + self, + filter_complex: str, + input_args: list[str], + output_path: Path, + ) -> None: + """执行 FFmpeg 渲染命令。""" + command = [ + FFMPEG_BIN, + "-y", + *input_args, + "-filter_complex", + filter_complex, + "-map", + "[final_video]", + "-c:v", + "libx264", + "-crf", + "23", + "-preset", + "medium", + "-pix_fmt", + "yuv420p", + "-movflags", + "+faststart", + str(output_path), + ] + + logger.info( + "执行渲染: plan_id=%s inputs=%d output=%s", + self.plan.id, + input_args.count("-i"), + output_path, + ) + run_ffmpeg(command) + + def _probe_output(self, output_path: Path) -> tuple[float, int, int, int]: + """探测输出文件的时长、大小、宽高。 + + Returns: + (duration, file_size, width, height) + """ + info = probe_video_info(str(output_path)) + file_size = output_path.stat().st_size if output_path.exists() else 0 + return ( + info["duration"], + file_size, + info["width"], + info["height"], + ) diff --git a/apps/worker/worker_app/tasks/edit_plan_generation.py b/apps/worker/worker_app/tasks/edit_plan_generation.py index 5e0ddfcdb..bef354a83 100644 --- a/apps/worker/worker_app/tasks/edit_plan_generation.py +++ b/apps/worker/worker_app/tasks/edit_plan_generation.py @@ -3,176 +3,39 @@ Celery 任务 worker.render_edit_plan: 1. 加载 EditPlan + EditPlanClips 2. 下载各片段素材 - 3. 按 order 顺序拼接片段 + 3. 使用 UnifiedRenderService 按时间线+图层渲染 4. 上传渲染结果到 OSS - 5. 更新 EditPlan / EditPlanClip 状态 - 6. 更新 GenerationTask 进度 + 5. 创建 GeneratedVideo 记录 + 查重 + 6. 更新 EditPlan / EditPlanClip 状态 + 7. 更新 GenerationTask 进度 """ from __future__ import annotations import logging -import os -import shutil -import subprocess # nosec B404 import tempfile from datetime import datetime, timezone from pathlib import Path from typing import Optional -from urllib.parse import urlparse -import oss2 from worker_app.celery_app import celery_app from worker_app.db import SessionLocal logger = logging.getLogger(__name__) -FFMPEG_BIN = shutil.which("ffmpeg") or "ffmpeg" -FFPROBE_BIN = shutil.which("ffprobe") or "ffprobe" OUTPUT_WIDTH = 1280 OUTPUT_HEIGHT = 720 OUTPUT_FPS = 25.0 -# ── OSS helpers ─────────────────────────────────────────────────────────────── +# ── 共享工具模块导入 ────────────────────────────────────────────────────────── - -def _oss_settings() -> tuple[str, str, str, str] | None: - """获取 OSS 配置""" - access_key_id = os.getenv("OSS_ACCESS_KEY_ID") - access_key_secret = os.getenv("OSS_ACCESS_KEY_SECRET") - endpoint = os.getenv("OSS_ENDPOINT") - bucket_name = os.getenv("OSS_BUCKET_NAME") - if not all([access_key_id, access_key_secret, endpoint, bucket_name]): - return None - return access_key_id, access_key_secret, endpoint, bucket_name - - -def _oss_bucket() -> oss2.Bucket | None: - """获取 OSS Bucket""" - settings = _oss_settings() - if settings is None: - return None - access_key_id, access_key_secret, endpoint, bucket_name = settings - return oss2.Bucket(oss2.Auth(access_key_id, access_key_secret), endpoint, bucket_name) - - -def _normalize_storage_key(storage_key_or_url: str) -> str: - """标准化存储键""" - if storage_key_or_url.startswith(("http://", "https://")): - return urlparse(storage_key_or_url).path.lstrip("/") - return storage_key_or_url.lstrip("/") - - -def _download_asset(asset_storage_key: str, local_path: Path) -> bool: - """下载素材文件到本地""" - bucket = _oss_bucket() - if bucket is None: - return False - try: - bucket.get_object_to_file(_normalize_storage_key(asset_storage_key), str(local_path)) - return local_path.exists() and local_path.stat().st_size > 0 - except Exception: - logger.exception("下载素材失败: %s", asset_storage_key) - return False - - -def _upload_to_oss(local_path: Path, storage_key: str) -> str | None: - """上传文件到 OSS，返回公开 URL""" - bucket = _oss_bucket() - if bucket is None: - return None - try: - bucket.put_object_from_file(storage_key, str(local_path)) - settings = _oss_settings() - if settings: - _, _, endpoint, bucket_name = settings - return f"https://{bucket_name}.{endpoint.replace('https://', '').replace('http://', '')}/{storage_key}" - return None - except Exception: - logger.exception("上传 OSS 失败: %s", storage_key) - return None - - -# ── FFmpeg helpers ──────────────────────────────────────────────────────────── - - -def _run_ffmpeg(command: list[str]) -> None: - """执行 FFmpeg 命令""" - subprocess.run(command, check=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True) # nosec B603 - - -def _probe_duration(local_path: Path) -> float: - """获取视频/音频时长""" - try: - result = subprocess.run( # nosec B603 - [ - FFPROBE_BIN, - "-v", - "error", - "-show_entries", - "format=duration", - "-of", - "default=noprint_wrappers=1:nokey=1", - str(local_path), - ], - check=True, - capture_output=True, - text=True, - ) - return float(result.stdout.strip()) - except Exception: - return 0.0 - - -def _concatenate_clips( - clip_paths: list[Path], - output_path: Path, - transition_effects: list[str] | None = None, -) -> bool: - """将多个片段拼接为最终视频 - - 使用 FFmpeg concat demuxer 实现。 - """ - if not clip_paths: - return False - - if len(clip_paths) == 1: - # 单片段直接复制 - try: - shutil.copy2(str(clip_paths[0]), str(output_path)) - return True - except Exception: - return False - - # 多片段：使用 concat demuxer - concat_file = output_path.parent / "concat_list.txt" - try: - with open(concat_file, "w") as f: - for p in clip_paths: - f.write(f"file '{p}'\n") - - command = [ - FFMPEG_BIN, - "-y", - "-f", - "concat", - "-safe", - "0", - "-i", - str(concat_file), - "-c", - "copy", - str(output_path), - ] - _run_ffmpeg(command) - return output_path.exists() and output_path.stat().st_size > 0 - except Exception: - logger.exception("拼接片段失败") - return False - finally: - if concat_file.exists(): - concat_file.unlink() +from video_processing.oss_helpers import ( + download_asset, + upload_to_oss, +) +from video_processing.unified_render_service import UnifiedRenderService +from video_processing.dedup_helpers import create_video_record_and_dedup # ── Repository imports (延迟导入避免循环依赖) ───────────────────────────────── @@ -207,11 +70,12 @@ def render_edit_plan(self, plan_id: str) -> dict: 流程： 1. 加载 EditPlan + EditPlanClips - 2. 下载各片段素材到临时目录 - 3. 按 order 顺序拼接片段 + 2. 下载各片段素材到临时目录，构建 asset_path_map + 3. 使用 UnifiedRenderService 按时间线+图层渲染 4. 上传渲染结果到 OSS - 5. 更新 EditPlan → completed, EditPlanClips → rendered - 6. 更新 GenerationTask 进度 + 5. 创建 GeneratedVideo 记录 + 查重 + 6. 更新 EditPlan → completed, EditPlanClips → rendered + 7. 更新 GenerationTask 进度 """ logger.info("开始渲染剪辑计划: plan_id=%s", plan_id) @@ -246,10 +110,10 @@ def render_edit_plan(self, plan_id: str) -> dict: gen_task.started_at = datetime.now(timezone.utc) gen_task_repo.update(gen_task) - # 3. 下载素材并拼接 + # 3. 下载素材并构建 asset_path_map with tempfile.TemporaryDirectory(prefix="edit_plan_") as tmpdir: tmpdir_path = Path(tmpdir) - clip_paths: list[Path] = [] + asset_path_map: dict[str, Path] = {} rendered_clip_ids: list[str] = [] failed_clip_ids: list[str] = [] @@ -261,18 +125,23 @@ def render_edit_plan(self, plan_id: str) -> dict: failed_clip_ids.append(clip.id) continue + if clip.asset_id in asset_path_map: + # 同一素材已下载（多个 clip 共享同一素材） + rendered_clip_ids.append(clip.id) + continue + # 下载素材 ext = Path(clip.asset_id).suffix or ".mp4" local_path = tmpdir_path / f"clip_{clip.order:04d}{ext}" - if _download_asset(clip.asset_id, local_path): - clip_paths.append(local_path) + if download_asset(clip.asset_id, local_path): + asset_path_map[clip.asset_id] = local_path rendered_clip_ids.append(clip.id) else: clip.mark_failed() clip_repo.update(clip) failed_clip_ids.append(clip.id) - if not clip_paths: + if not asset_path_map: logger.error("所有片段素材下载失败: %s", plan_id) plan.mark_failed() plan_repo.update(plan) @@ -285,42 +154,75 @@ def render_edit_plan(self, plan_id: str) -> dict: gen_task_repo.update(gen_task) return {"status": "error", "message": "所有片段素材下载失败"} - # 4. 拼接片段 - output_path = tmpdir_path / f"rendered_{plan_id}.mp4" - transition_effects = [c.transition_effect for c in clips if c.asset_id] - success = _concatenate_clips(clip_paths, output_path, transition_effects) + # 4. 使用 UnifiedRenderService 渲染 + render_service = UnifiedRenderService( + plan=plan, + clips=clips, + asset_path_map=asset_path_map, + work_dir=tmpdir_path, + output_width=OUTPUT_WIDTH, + output_height=OUTPUT_HEIGHT, + output_fps=int(OUTPUT_FPS), + ) - if not success: - logger.error("片段拼接失败: %s", plan_id) + try: + render_result = render_service.render() + except Exception as render_err: + logger.error("渲染失败: %s — %s", plan_id, render_err) plan.mark_failed() plan_repo.update(plan) if generation_task_id: gen_task = gen_task_repo.get(generation_task_id) if gen_task: gen_task.status = "failed" - gen_task.error_message = "片段拼接失败" + gen_task.error_message = f"渲染失败: {render_err}" gen_task.completed_at = datetime.now(timezone.utc) gen_task_repo.update(gen_task) - return {"status": "error", "message": "片段拼接失败"} + return {"status": "error", "message": f"渲染失败: {render_err}"} + + output_path = render_result.output_path # 5. 上传到 OSS storage_key = f"rendered/{plan_id}/output.mp4" - output_url = _upload_to_oss(output_path, storage_key) + output_url = upload_to_oss(output_path, storage_key) - # 6. 更新片段状态为 rendered + # 6. 创建 GeneratedVideo 记录 + 查重 + project_id = plan.project_id or "" + batch_id = plan.config.get("batch_id", "") + mode = plan.config.get("mode", "edit_plan") + if generation_task_id and project_id: + try: + create_video_record_and_dedup( + generation_task_id=generation_task_id, + project_id=project_id, + batch_id=batch_id, + file_url=output_url or "", + file_size=render_result.file_size, + duration=render_result.duration, + video_path=str(output_path), + mode=mode, + session=db, + width=render_result.width, + height=render_result.height, + fps=OUTPUT_FPS, + ) + except Exception as dedup_err: + logger.warning("查重失败（不影响渲染结果）: %s", dedup_err) + + # 7. 更新片段状态为 rendered for clip_id in rendered_clip_ids: clip = clip_repo.get(clip_id) if clip and clip.status.value == "ready": clip.mark_rendered() clip_repo.update(clip) - # 7. 更新 EditPlan 状态为 completed + # 8. 更新 EditPlan 状态为 completed plan.config["rendered_url"] = output_url or "" plan.config["rendered_storage_key"] = storage_key plan.mark_completed() plan_repo.update(plan) - # 8. 更新 GenerationTask 状态为 completed + # 9. 更新 GenerationTask 状态为 completed if generation_task_id: gen_task = gen_task_repo.get(generation_task_id) if gen_task: @@ -331,10 +233,11 @@ def render_edit_plan(self, plan_id: str) -> dict: gen_task_repo.update(gen_task) logger.info( - "剪辑计划渲染完成: plan_id=%s rendered=%d failed=%d", + "剪辑计划渲染完成: plan_id=%s rendered=%d failed=%d duration=%.1fs", plan_id, len(rendered_clip_ids), len(failed_clip_ids), + render_result.duration, ) return { @@ -343,6 +246,7 @@ def render_edit_plan(self, plan_id: str) -> dict: "rendered_count": len(rendered_clip_ids), "failed_count": len(failed_clip_ids), "output_url": output_url, + "duration": render_result.duration, } except Exception as exc: diff --git a/apps/worker/worker_app/tasks/generation.py b/apps/worker/worker_app/tasks/generation.py index 4cdde7145..f9b27065f 100755 --- a/apps/worker/worker_app/tasks/generation.py +++ b/apps/worker/worker_app/tasks/generation.py @@ -1,18 +1,25 @@ """ -视频生成任务 -支持四种剪辑模式：一镜到底、画中画、口播、口播+画中画 +视频生成任务 — 使用 UnifiedRenderService 统一渲染引擎. + +支持四种剪辑模式：一镜到底、画中画、口播、口播+画中画。 +模式差异体现在虚拟剪辑计划的 clip_type 分布上，渲染引擎不判断模式。 + +模式 → clip_type 映射： + ONE_TAKE: N 个 main clips + PIP: 1 main + N-1 overlay + VOICE_OVER: N 个 main(config.role=b_roll) + VOICE_PIP: 1 background + 1 corner_voice + N-2 b_roll """ +from __future__ import annotations + import logging import os -import shutil -import subprocess # nosec B404 import tempfile +from dataclasses import dataclass, field from pathlib import Path -from typing import Optional -from urllib.parse import urlparse +from typing import Any, Optional -import oss2 from worker_app.celery_app import celery_app from worker_app.db import SessionLocal @@ -20,8 +27,6 @@ OUTPUT_WIDTH = 1280 OUTPUT_HEIGHT = 720 OUTPUT_FPS = 25.0 OUTPUT_DURATION_SECONDS = 5.0 -FFMPEG_BIN = shutil.which("ffmpeg") or "ffmpeg" -FFPROBE_BIN = shutil.which("ffprobe") or "ffprobe" GENERATED_FILES_DIR = Path(os.getenv("GENERATED_FILES_DIR", "/app/generated")) GENERATED_FILES_URL_PREFIX = os.getenv("GENERATED_FILES_URL_PREFIX", "/generated-files") PUBLIC_API_BASE_URL = os.getenv("PUBLIC_API_BASE_URL", "https://api.xiaoxiajianji.com").rstrip("/") @@ -78,81 +83,131 @@ def _update_task_status(task_id: str, status_action: str, **kwargs) -> bool: return False -# ── FFmpeg / OSS helpers ───────────────────────────────────────────────────── +# ── 共享工具模块导入 ────────────────────────────────────────────────────────── + +from video_processing.ffmpeg_utils import FFMPEG_BIN, run_ffmpeg, probe_duration +from video_processing.oss_helpers import ( + download_asset, + oss_bucket, + upload_to_oss, +) +from video_processing.dedup_helpers import create_video_record_and_dedup +from video_processing.unified_render_service import UnifiedRenderService -def _run_ffmpeg(command: list[str]) -> None: - """执行 FFmpeg 命令""" - subprocess.run(command, check=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True) # nosec B603 +# ── 虚拟 Plan / Clip（内存中构建，不写数据库） ──────────────────────────────── -def _oss_settings() -> tuple[str, str, str, str] | None: - """获取 OSS 配置""" - access_key_id = os.getenv("OSS_ACCESS_KEY_ID") - access_key_secret = os.getenv("OSS_ACCESS_KEY_SECRET") - endpoint = os.getenv("OSS_ENDPOINT") - bucket_name = os.getenv("OSS_BUCKET_NAME") - if not all([access_key_id, access_key_secret, endpoint, bucket_name]): - return None - return access_key_id, access_key_secret, endpoint, bucket_name +@dataclass +class _VirtualPlan: + """内存中的虚拟剪辑计划，供 UnifiedRenderService 使用。""" + + id: str + name: str = "" -def _oss_bucket() -> oss2.Bucket | None: - """获取 OSS Bucket""" - settings = _oss_settings() - if settings is None: - return None - access_key_id, access_key_secret, endpoint, bucket_name = settings - return oss2.Bucket(oss2.Auth(access_key_id, access_key_secret), endpoint, bucket_name) +@dataclass +class _VirtualClip: + """内存中的虚拟剪辑片段，供 UnifiedRenderService 使用。""" + + id: str + plan_id: str = "" + clip_type: str = "main" + order: int = 0 + asset_id: str = "" + text_content: str = "" + start_time: float = 0.0 + duration: float = 0.0 + transition_effect: str = "cut" + status: str = "ready" + config: dict[str, Any] = field(default_factory=dict) -def _normalize_storage_key(storage_key_or_url: str) -> str: - """标准化存储键""" - if storage_key_or_url.startswith(("http://", "https://")): - return urlparse(storage_key_or_url).path.lstrip("/") - return storage_key_or_url.lstrip("/") +def _build_plan_and_clips_from_task( + task_id: str, + downloaded_paths: list[Path], + mode: str, +) -> tuple[_VirtualPlan, list[_VirtualClip], dict[str, Path]]: + """根据模式和下载的素材路径，构建虚拟 plan + clips + asset_path_map。 + 模式 → clip_type 映射： + ONE_TAKE: N 个 main clips + PIP: 1 main + N-1 overlay + VOICE_OVER: N 个 main(config.role=b_roll) + VOICE_PIP: 1 background + 1 corner_voice + N-2 b_roll -def _download_asset(asset_storage_key: str, local_path: Path) -> bool: - """下载素材文件""" - bucket = _oss_bucket() - if bucket is None: - return False - try: - bucket.get_object_to_file(_normalize_storage_key(asset_storage_key), str(local_path)) - return local_path.exists() and local_path.stat().st_size > 0 - except Exception: - return False + Returns: + (virtual_plan, virtual_clips, asset_path_map) + """ + plan = _VirtualPlan(id=task_id, name=f"Generated-{task_id[:8]}") + # 为每个下载路径生成合成 asset_id + asset_path_map: dict[str, Path] = {} + path_to_asset_id: dict[Path, str] = {} + for i, p in enumerate(downloaded_paths): + asset_id = f"gen_{task_id[:8]}_{i:03d}{p.suffix or '.mp4'}" + asset_path_map[asset_id] = p + path_to_asset_id[p] = asset_id -def _probe_duration(local_path: Path) -> float: - """获取视频时长""" - try: - result = subprocess.run( - [ - FFPROBE_BIN, - "-v", - "error", - "-show_entries", - "format=duration", - "-of", - "default=noprint_wrappers=1:nokey=1", - str(local_path), - ], - check=True, - stdout=subprocess.PIPE, - stderr=subprocess.PIPE, - text=True, - ) # nosec B603 - return round(float(result.stdout.strip()), 3) - except Exception: - return OUTPUT_DURATION_SECONDS + clips: list[_VirtualClip] = [] + n = len(downloaded_paths) + + if mode == "pip": + # 1 main + N-1 overlay + for i, p in enumerate(downloaded_paths): + clip_type = "main" if i == 0 else "overlay" + clips.append(_VirtualClip( + id=f"vc_{i:03d}", + plan_id=task_id, + clip_type=clip_type, + order=i, + asset_id=path_to_asset_id[p], + )) + elif mode == "voice_over": + # N 个 main(config.role=b_roll) + for i, p in enumerate(downloaded_paths): + clips.append(_VirtualClip( + id=f"vc_{i:03d}", + plan_id=task_id, + clip_type="main", + order=i, + asset_id=path_to_asset_id[p], + config={"role": "b_roll"}, + )) + elif mode == "voice_pip": + # 1 background + 1 corner_voice + N-2 b_roll + for i, p in enumerate(downloaded_paths): + if i == 0: + clip_type = "background" + elif i == 1: + clip_type = "corner_voice" + else: + clip_type = "b_roll" + clips.append(_VirtualClip( + id=f"vc_{i:03d}", + plan_id=task_id, + clip_type=clip_type, + order=i, + asset_id=path_to_asset_id[p], + )) + else: + # ONE_TAKE (default): N 个 main clips + for i, p in enumerate(downloaded_paths): + clips.append(_VirtualClip( + id=f"vc_{i:03d}", + plan_id=task_id, + clip_type="main", + order=i, + asset_id=path_to_asset_id[p], + )) + + return plan, clips, asset_path_map def _create_fallback_clip(output_path: Path, title: str) -> None: """创建 fallback 视频（无素材时）""" safe_title = title.replace(":", "\\:").replace("'", "\\'")[:80] - _run_ffmpeg( + run_ffmpeg( [ FFMPEG_BIN, "-y", @@ -173,19 +228,35 @@ def _create_fallback_clip(output_path: Path, title: str) -> None: ) +def _mux_audio_track(video_path: Path, audio_path: str, output_path: Path) -> None: + """将音频轨混入已渲染的视频（后处理步骤）。 + + 使用 FFmpeg 将视频和音频合并，视频时长为准，音频不足则循环， + 音频过长则截断。 + """ + command = [ + FFMPEG_BIN, + "-y", + "-i", str(video_path), + "-i", audio_path, + "-c:v", "copy", + "-c:a", "aac", + "-b:a", "192k", + "-shortest", + "-map", "0:v:0", + "-map", "1:a:0", + "-movflags", "+faststart", + str(output_path), + ] + run_ffmpeg(command) + + def _download_voice_asset(voice_library_id: str, local_path: Path) -> bool: """下载配音文件""" if not voice_library_id: return False - bucket = _oss_bucket() - if bucket is None: - return False storage_key = f"voice/{voice_library_id}.mp3" - try: - bucket.get_object_to_file(_normalize_storage_key(storage_key), str(local_path)) - return local_path.exists() and local_path.stat().st_size > 0 - except Exception: - return False + return download_asset(storage_key, local_path) def _download_library_assets( @@ -193,9 +264,8 @@ def _download_library_assets( temp_path: Path, video_extensions: tuple = (".mp4", ".mov", ".avi", ".mkv", ".webm"), asset_ids: list[str] | None = None, -) -> list[str]: - """ - 从素材库下载视频素材 +) -> list[Path]: + """从素材库下载视频素材。 Args: asset_library_id: 素材库 ID @@ -204,91 +274,63 @@ def _download_library_assets( asset_ids: 指定素材 ID 列表，为空则下载全部 ready 视频素材 Returns: - 下载成功的视频文件路径列表 + 下载成功的视频文件 Path 列表 """ - # 导入模型和会话 try: from packages.adapters.sqlalchemy_impl.models import AssetModel session = SessionLocal() - try: - # 查询素材库中的视频素材 query = session.query(AssetModel).filter( AssetModel.asset_library_id == asset_library_id, AssetModel.status == "ready", AssetModel.file_type.in_(["video", "video/mp4", "video/quicktime"]), ) - # 如果指定了 asset_ids，则只下载这些素材 if asset_ids: query = query.filter(AssetModel.id.in_(asset_ids)) assets = query.order_by(AssetModel.created_at).all() if not assets: - logger.info(f"No video assets found in library {asset_library_id}") + logger.info("No video assets found in library %s", asset_library_id) return [] - downloaded_videos = [] + downloaded: list[Path] = [] for i, asset in enumerate(assets): - # 获取文件 URL 或 storage_key storage_key = asset.file_url if asset.file_url else None if not storage_key: continue - local_file = temp_path / f"asset_{i}_{asset.id}.mp4" - if _download_asset(storage_key, local_file): - downloaded_videos.append(str(local_file)) - logger.info(f"Downloaded asset: {asset.name} -> {local_file}") + ext = Path(storage_key).suffix or ".mp4" + local_file = temp_path / f"asset_{i:03d}_{asset.id}{ext}" + if download_asset(storage_key, local_file): + downloaded.append(local_file) + logger.info("Downloaded asset: %s -> %s", asset.name, local_file) else: - logger.warning(f"Failed to download asset: {asset.name}") + logger.warning("Failed to download asset: %s", asset.name) - return downloaded_videos + return downloaded finally: session.close() except Exception as e: - logger.error(f"Error downloading library assets: {e}") + logger.error("Error downloading library assets: %s", e) return [] -def _process_with_editing_mode( - video_paths: list[str], - audio_path: Optional[str], - mode: str, - output_path: Path, -) -> None: - """根据剪辑模式处理视频""" - from video_processing.editing_modes import ( - EditingMode, - EditingModeConfig, - EditingModeProcessor, - PIPPosition, - ) - - config = EditingModeConfig( - mode=EditingMode(mode), - output_width=OUTPUT_WIDTH, - output_height=OUTPUT_HEIGHT, - output_fps=int(OUTPUT_FPS), - pip_position=PIPPosition.TOP_RIGHT, - pip_scale=0.25, - transition_duration=0.5, - ) - - processor = EditingModeProcessor(config=config) - processor.process( - video_paths=video_paths, - audio_path=audio_path, - output_path=str(output_path), - ) - - # ── Celery Task ────────────────────────────────────────────────────────────── @celery_app.task(bind=True, name="worker.generate_video", max_retries=2) def generate_video(self, task_id: str) -> dict: - """ - 生成视频任务 + """生成视频任务 — 使用 UnifiedRenderService 统一渲染。 + + 流程： + 1. 加载 GenerationTask 信息 + 2. 从素材库下载视频素材 + 3. 根据模式构建虚拟 plan + clips + 4. 使用 UnifiedRenderService 渲染 + 5. 如有配音，后处理混音 + 6. 上传 OSS + 查重 + 7. 更新 GenerationTask 状态 Args: task_id: 任务 ID（从数据库加载完整任务信息） @@ -337,35 +379,65 @@ def generate_video(self, task_id: str) -> dict: temp_path = Path(temp_dir) output_path = temp_path / output_name - # 从素材库下载视频素材（如果任务指定了 asset_ids 则只下载这些） - downloaded_videos = _download_library_assets(asset_library_id, temp_path, asset_ids=task_asset_ids or None) + # 1. 从素材库下载视频素材 + downloaded_videos = _download_library_assets( + asset_library_id, temp_path, asset_ids=task_asset_ids or None + ) - audio_path = None + # 2. 下载配音（如有） + audio_path: str | None = None if voice_library_id: local_audio = temp_path / "voice.mp3" if _download_voice_asset(voice_library_id, local_audio): audio_path = str(local_audio) + # 3. 渲染 if downloaded_videos: - _process_with_editing_mode( - video_paths=downloaded_videos, - audio_path=audio_path, + # 构建虚拟 plan + clips + asset_path_map + virtual_plan, virtual_clips, asset_path_map = _build_plan_and_clips_from_task( + task_id=task_id, + downloaded_paths=downloaded_videos, mode=editing_mode.value, - output_path=output_path, ) + + # 使用 UnifiedRenderService 渲染 + render_service = UnifiedRenderService( + plan=virtual_plan, + clips=virtual_clips, + asset_path_map=asset_path_map, + work_dir=temp_path, + output_width=OUTPUT_WIDTH, + output_height=OUTPUT_HEIGHT, + output_fps=int(OUTPUT_FPS), + ) + render_result = render_service.render() + + # 4. 如有配音，后处理混音 + if audio_path: + final_path = temp_path / f"final-{task_id}.mp4" + try: + _mux_audio_track(render_result.output_path, audio_path, final_path) + # 混音成功，使用混音后的文件 + output_path = final_path + except Exception as mux_err: + logger.warning("音频混合失败，使用无音频版本: %s", mux_err) + output_path = render_result.output_path + else: + output_path = render_result.output_path else: + # 无素材，生成 fallback 视频 _create_fallback_clip(output_path, f"Generated Video {task_id[:8]}") file_size = output_path.stat().st_size - duration = _probe_duration(output_path) + duration = probe_duration(output_path) - # 上传到 OSS - bucket = _oss_bucket() + # 5. 上传到 OSS + bucket = oss_bucket() if bucket: try: bucket.put_object_from_file(storage_key, str(output_path)) except Exception as oss_err: - logger.warning(f"OSS upload failed: {oss_err}") + logger.warning("OSS upload failed: %s", oss_err) # 构建视频 URL if bucket: @@ -373,19 +445,24 @@ def generate_video(self, task_id: str) -> dict: else: file_url = f"{GENERATED_FILES_URL_PREFIX}/{task_id}/{output_name}" - # 创建 GeneratedVideo 记录 + 查重 - video_count = _create_video_record_and_dedup( - task_id=task_id, - project_id=project_id, - batch_id=batch_id, - file_url=file_url, - file_size=file_size, - duration=duration, - video_path=str(output_path), - mode=editing_mode.value, - ) + # 6. 创建 GeneratedVideo 记录 + 查重 + dedup_session = SessionLocal() + try: + video_count = create_video_record_and_dedup( + generation_task_id=task_id, + project_id=project_id, + batch_id=batch_id, + file_url=file_url, + file_size=file_size, + duration=duration, + video_path=str(output_path), + mode=editing_mode.value, + session=dedup_session, + ) + finally: + dedup_session.close() - # 标记任务为 completed + # 7. 标记任务为 completed _update_task_status(task_id, "mark_completed", result_count=video_count or 1) logger.info("视频生成完成: task_id=%s duration=%.2fs file_size=%d", task_id, duration, file_size) @@ -401,8 +478,7 @@ def generate_video(self, task_id: str) -> dict: "mode": editing_mode.value, } except Exception as error: - logger.error(f"Video generation failed: {error}", exc_info=True) - # 标记任务为 failed + logger.error("Video generation failed: %s", error, exc_info=True) _update_task_status(task_id, "mark_failed", error_message=str(error)) return { "status": "failed", @@ -411,88 +487,3 @@ def generate_video(self, task_id: str) -> dict: } -def _create_video_record_and_dedup( - *, - task_id: str, - project_id: str, - batch_id: str, - file_url: str, - file_size: int, - duration: float, - video_path: str, - mode: str, -) -> int: - """创建 GeneratedVideo 记录，计算指纹并执行查重（历史 + 批次）。 - - Returns: - 创建的视频记录数量（1 表示成功，0 表示失败） - """ - from uuid import uuid4 - - from video_processing.dedup import VideoDeduplicator - - from packages.adapters.sqlalchemy_impl.generated_video_repository import ( - SQLAlchemyGeneratedVideoRepository, - ) - from packages.domain import GeneratedVideo - - session = SessionLocal() - try: - video_id = uuid4().hex - generated_video = GeneratedVideo( - id=video_id, - project_id=project_id, - generation_task_id=task_id, - name=f"generated-{task_id[:8]}.mp4", - file_url=file_url, - file_size=file_size, - duration=duration, - width=OUTPUT_WIDTH, - height=OUTPUT_HEIGHT, - fps=OUTPUT_FPS, - status="completed", - generation_params={"mode": mode}, - ) - - video_repo = SQLAlchemyGeneratedVideoRepository(session) - video_repo.create(generated_video) - - # 计算视频指纹 - deduplicator = VideoDeduplicator() - try: - fingerprint = deduplicator.compute_fingerprint(video_path) - except Exception as fp_err: - logger.warning(f"Fingerprint computation failed for {video_id}: {fp_err}") - session.commit() - return 1 - - generated_video.video_fingerprint = fingerprint.to_dict() - - # (a) 历史成片查重 - duplicate_result = deduplicator.check_duplicate(fingerprint, project_id, session) - - # (b) 批次内查重（仅当有 batch_id 时） - if not duplicate_result and batch_id: - duplicate_result = deduplicator.check_batch_duplicate(fingerprint, batch_id, video_id, session) - - if duplicate_result: - generated_video.is_duplicate = True - generated_video.duplicate_of = duplicate_result["duplicate_of"] - logger.info( - f"Duplicate detected: {video_id} -> {duplicate_result['duplicate_of']} " - f"(reason={duplicate_result['reason']}, similarity={duplicate_result['similarity']:.3f})" - ) - else: - generated_video.is_duplicate = False - generated_video.duplicate_of = None - - video_repo.update(generated_video) - session.commit() - logger.info(f"GeneratedVideo record created: {video_id} (task={task_id}, dup={generated_video.is_duplicate})") - return 1 - except Exception as e: - logger.error(f"Failed to create video record / dedup for task {task_id}: {e}") - session.rollback() - return 0 - finally: - session.close() diff --git a/tests/integration/test_four_mode_rendering.py b/tests/integration/test_four_mode_rendering.py new file mode 100644 index 000000000..fd433f6c0 --- /dev/null +++ b/tests/integration/test_four_mode_rendering.py @@ -0,0 +1,237 @@ +"""四模式渲染集成测试. + +验证 4 种剪辑模式（ONE_TAKE / PIP / VOICE_OVER / VOICE_PIP）通过 +_build_plan_and_clips_from_task + UnifiedRenderService 的完整渲染流程。 + +需要 ffmpeg 可用；CI 无 ffmpeg 时自动跳过。 +""" + +from __future__ import annotations + +import shutil +import subprocess +import tempfile +from pathlib import Path + +import pytest + +from worker_app.tasks.generation import _build_plan_and_clips_from_task +from video_processing.unified_render_service import ( + RenderResult, + UnifiedRenderService, + _resolve_layer_role, +) + +pytestmark = pytest.mark.skipif( + not shutil.which("ffmpeg"), + reason="ffmpeg not available", +) + + +# ── 辅助函数 ────────────────────────────────────────────────────────────────── + + +def _generate_test_video(path: Path, duration: float = 3.0, color: str = "red") -> None: + """生成一个纯色测试视频。""" + cmd = [ + "ffmpeg", + "-y", + "-f", + "lavfi", + "-i", + f"color=c={color}:s=640x360:d={duration}:r=25", + "-c:v", + "libx264", + "-pix_fmt", + "yuv420p", + "-movflags", + "+faststart", + str(path), + ] + subprocess.run(cmd, check=True, capture_output=True, timeout=30) + + +def _render_with_mode( + mode: str, + num_clips: int = 3, + duration: float = 2.0, +) -> tuple[RenderResult, Path]: + """用指定模式生成测试视频并渲染，返回 (result, work_dir)。 + + 调用方负责清理 work_dir。 + """ + work_dir = Path(tempfile.mkdtemp(prefix="test_4mode_")) + + # 生成测试视频素材 + colors = ["red", "green", "blue", "yellow", "purple"] + downloaded_paths: list[Path] = [] + for i in range(num_clips): + p = work_dir / f"test_{i:03d}.mp4" + _generate_test_video(p, duration=duration, color=colors[i % len(colors)]) + downloaded_paths.append(p) + + # 构建虚拟 plan + clips + task_id = f"test_task_{mode}" + plan, clips, asset_path_map = _build_plan_and_clips_from_task( + task_id=task_id, + downloaded_paths=downloaded_paths, + mode=mode, + ) + + # 渲染 + service = UnifiedRenderService( + plan=plan, + clips=clips, + asset_path_map=asset_path_map, + work_dir=work_dir, + output_width=640, + output_height=360, + output_fps=25, + ) + result = service.render() + return result, work_dir + + +# ── 测试 _build_plan_and_clips_from_task ────────────────────────────────────── + + +class TestBuildPlanAndClips: + """测试 4 种模式的虚拟 plan 构建。""" + + def _make_paths(self, n: int) -> list[Path]: + return [Path(f"/tmp/test_{i}.mp4") for i in range(n)] + + def test_one_take_mode(self): + paths = self._make_paths(3) + plan, clips, asset_map = _build_plan_and_clips_from_task("t1", paths, "one_take") + + assert plan.id == "t1" + assert len(clips) == 3 + assert all(c.clip_type == "main" for c in clips) + assert len(asset_map) == 3 + + def test_pip_mode(self): + paths = self._make_paths(3) + plan, clips, asset_map = _build_plan_and_clips_from_task("t2", paths, "pip") + + assert len(clips) == 3 + assert clips[0].clip_type == "main" + assert clips[1].clip_type == "overlay" + assert clips[2].clip_type == "overlay" + + def test_voice_over_mode(self): + paths = self._make_paths(3) + plan, clips, asset_map = _build_plan_and_clips_from_task("t3", paths, "voice_over") + + assert len(clips) == 3 + assert all(c.clip_type == "main" for c in clips) + assert all(c.config.get("role") == "b_roll" for c in clips) + + def test_voice_pip_mode(self): + paths = self._make_paths(4) + plan, clips, asset_map = _build_plan_and_clips_from_task("t4", paths, "voice_pip") + + assert len(clips) == 4 + assert clips[0].clip_type == "background" + assert clips[1].clip_type == "corner_voice" + assert clips[2].clip_type == "b_roll" + assert clips[3].clip_type == "b_roll" + + def test_unknown_mode_defaults_to_one_take(self): + paths = self._make_paths(2) + plan, clips, asset_map = _build_plan_and_clips_from_task("t5", paths, "unknown_mode") + + assert len(clips) == 2 + assert all(c.clip_type == "main" for c in clips) + + def test_asset_path_map_keys_match_clip_asset_ids(self): + paths = self._make_paths(3) + _, clips, asset_map = _build_plan_and_clips_from_task("t6", paths, "one_take") + + clip_asset_ids = {c.asset_id for c in clips} + map_keys = set(asset_map.keys()) + assert clip_asset_ids == map_keys + + +# ── 测试图层分组（4 模式） ──────────────────────────────────────────────────── + + +class TestFourModeLayerGrouping: + """验证 4 种模式的 clip_type 分布经 _resolve_layer_role 后产生正确的图层。""" + + def test_one_take_layers(self): + """ONE_TAKE: 3 main → 1 main layer。""" + paths = [Path(f"/tmp/ot_{i}.mp4") for i in range(3)] + _, clips, _ = _build_plan_and_clips_from_task("ot", paths, "one_take") + + roles = {_resolve_layer_role(c.clip_type, c.config) for c in clips} + assert roles == {"main"} + + def test_pip_layers(self): + """PIP: 1 main + 2 overlay → main + overlay。""" + paths = [Path(f"/tmp/pip_{i}.mp4") for i in range(3)] + _, clips, _ = _build_plan_and_clips_from_task("pip", paths, "pip") + + roles = {_resolve_layer_role(c.clip_type, c.config) for c in clips} + assert roles == {"main", "overlay"} + + def test_voice_over_layers(self): + """VOICE_OVER: 3 main(b_roll) → broll。""" + paths = [Path(f"/tmp/vo_{i}.mp4") for i in range(3)] + _, clips, _ = _build_plan_and_clips_from_task("vo", paths, "voice_over") + + roles = {_resolve_layer_role(c.clip_type, c.config) for c in clips} + assert roles == {"broll"} + + def test_voice_pip_layers(self): + """VOICE_PIP: 1 bg + 1 corner_voice + 2 b_roll → 3 个图层。""" + paths = [Path(f"/tmp/vpip_{i}.mp4") for i in range(4)] + _, clips, _ = _build_plan_and_clips_from_task("vpip", paths, "voice_pip") + + roles = {_resolve_layer_role(c.clip_type, c.config) for c in clips} + assert roles == {"background", "corner_voice", "broll"} + + +# ── 端到端渲染测试（需要 ffmpeg） ───────────────────────────────────────────── + + +class TestEndToEndRendering: + """4 种模式的完整渲染测试，验证输出文件存在且时长合理。""" + + def test_one_take_render(self): + result, work_dir = _render_with_mode("one_take", num_clips=2, duration=2.0) + try: + assert result.output_path.exists() + assert result.file_size > 0 + assert result.duration > 0 + assert result.width == 640 + assert result.height == 360 + finally: + shutil.rmtree(work_dir, ignore_errors=True) + + def test_pip_render(self): + result, work_dir = _render_with_mode("pip", num_clips=2, duration=2.0) + try: + assert result.output_path.exists() + assert result.file_size > 0 + assert result.duration > 0 + finally: + shutil.rmtree(work_dir, ignore_errors=True) + + def test_voice_over_render(self): + result, work_dir = _render_with_mode("voice_over", num_clips=2, duration=2.0) + try: + assert result.output_path.exists() + assert result.file_size > 0 + assert result.duration > 0 + finally: + shutil.rmtree(work_dir, ignore_errors=True) + + def test_voice_pip_render(self): + result, work_dir = _render_with_mode("voice_pip", num_clips=3, duration=2.0) + try: + assert result.output_path.exists() + assert result.file_size > 0 + assert result.duration > 0 + finally: + shutil.rmtree(work_dir, ignore_errors=True) diff --git a/tests/integration/test_full_pipeline.py b/tests/integration/test_full_pipeline.py new file mode 100644 index 000000000..878a96682 --- /dev/null +++ b/tests/integration/test_full_pipeline.py @@ -0,0 +1,239 @@ +"""全链路集成测试. + +验证 PlanGeneratorService → UnifiedRenderService → 查重 的端到端流程。 +""" + +from __future__ import annotations + +import shutil +import subprocess +import tempfile +from pathlib import Path +from unittest.mock import patch, MagicMock + +import pytest + +from video_processing.unified_render_service import ( + RenderResult, + UnifiedRenderService, +) +from worker_app.tasks.generation import ( + _build_plan_and_clips_from_task, + _mux_audio_track, + _create_fallback_clip, + OUTPUT_WIDTH, + OUTPUT_HEIGHT, +) + +pytestmark = pytest.mark.skipif( + not shutil.which("ffmpeg"), + reason="ffmpeg not available", +) + + +def _generate_test_video(path: Path, duration: float = 3.0) -> None: + """生成一个测试视频。""" + cmd = [ + "ffmpeg", + "-y", + "-f", + "lavfi", + "-i", + f"color=c=blue:s=640x360:d={duration}:r=25", + "-c:v", + "libx264", + "-pix_fmt", + "yuv420p", + str(path), + ] + subprocess.run(cmd, check=True, capture_output=True, timeout=30) + + +def _generate_test_audio(path: Path, duration: float = 5.0) -> None: + """生成一个测试音频文件。""" + cmd = [ + "ffmpeg", + "-y", + "-f", + "lavfi", + "-i", + f"sine=frequency=440:duration={duration}", + "-c:a", + "aac", + "-b:a", + "128k", + str(path), + ] + subprocess.run(cmd, check=True, capture_output=True, timeout=30) + + +# ── 测试 _create_fallback_clip ──────────────────────────────────────────────── + + +class TestFallbackClip: + """测试 fallback 视频生成。""" + + def test_fallback_clip_creates_video(self): + with tempfile.TemporaryDirectory() as tmpdir: + output = Path(tmpdir) / "fallback.mp4" + _create_fallback_clip(output, "Test Fallback") + + assert output.exists() + assert output.stat().st_size > 0 + + +# ── 测试 _mux_audio_track ──────────────────────────────────────────────────── + + +class TestMuxAudioTrack: + """测试视频+音频混合。""" + + def test_mux_audio_into_video(self): + with tempfile.TemporaryDirectory() as tmpdir: + video_path = Path(tmpdir) / "video.mp4" + audio_path = Path(tmpdir) / "audio.aac" + output_path = Path(tmpdir) / "output.mp4" + + _generate_test_video(video_path, duration=3.0) + _generate_test_audio(audio_path, duration=5.0) + + _mux_audio_track(video_path, str(audio_path), output_path) + + assert output_path.exists() + assert output_path.stat().st_size > 0 + + # 验证输出文件包含音频轨 + probe_cmd = [ + "ffprobe", + "-v", + "quiet", + "-show_streams", + "-select_streams", + "a", + "-of", + "csv=p=0", + str(output_path), + ] + result = subprocess.run(probe_cmd, capture_output=True, text=True, timeout=10) + # 如果有音频流，输出非空 + assert result.stdout.strip() != "" or result.returncode == 0 + + +# ── 测试 PlanGenerator → UnifiedRenderService 全链路 ───────────────────────── + + +class TestFullPipeline: + """验证从虚拟 plan 构建到渲染输出的完整流程。""" + + def test_one_take_pipeline(self): + """ONE_TAKE 模式完整流程。""" + with tempfile.TemporaryDirectory() as tmpdir: + work_dir = Path(tmpdir) + + # 生成测试素材 + paths = [] + for i in range(3): + p = work_dir / f"clip_{i}.mp4" + _generate_test_video(p, duration=2.0) + paths.append(p) + + # 构建虚拟 plan + plan, clips, asset_map = _build_plan_and_clips_from_task("pipeline_test", paths, "one_take") + + # 渲染 + service = UnifiedRenderService( + plan=plan, + clips=clips, + asset_path_map=asset_map, + work_dir=work_dir, + output_width=640, + output_height=360, + ) + result = service.render() + + assert result.output_path.exists() + assert result.duration > 0 + assert result.file_size > 0 + assert result.width == 640 + assert result.height == 360 + + def test_pipeline_with_audio_mux(self): + """渲染 + 混音后处理。""" + with tempfile.TemporaryDirectory() as tmpdir: + work_dir = Path(tmpdir) + + # 生成测试素材 + video_path = work_dir / "clip_0.mp4" + _generate_test_video(video_path, duration=3.0) + + # 构建虚拟 plan + plan, clips, asset_map = _build_plan_and_clips_from_task("audio_test", [video_path], "one_take") + + # 渲染 + service = UnifiedRenderService( + plan=plan, + clips=clips, + asset_path_map=asset_map, + work_dir=work_dir, + output_width=640, + output_height=360, + ) + render_result = service.render() + + # 混音 + audio_path = work_dir / "voice.aac" + _generate_test_audio(audio_path, duration=5.0) + + final_path = work_dir / "final.mp4" + _mux_audio_track(render_result.output_path, str(audio_path), final_path) + + assert final_path.exists() + assert final_path.stat().st_size > 0 + + def test_single_clip_pipeline(self): + """单 clip 渲染（无转场）。""" + with tempfile.TemporaryDirectory() as tmpdir: + work_dir = Path(tmpdir) + + video_path = work_dir / "single.mp4" + _generate_test_video(video_path, duration=5.0) + + plan, clips, asset_map = _build_plan_and_clips_from_task("single_test", [video_path], "one_take") + + service = UnifiedRenderService( + plan=plan, + clips=clips, + asset_path_map=asset_map, + work_dir=work_dir, + output_width=640, + output_height=360, + ) + result = service.render() + + assert result.output_path.exists() + assert result.duration > 0 + + def test_dedup_helper_integration(self): + """验证 dedup_helpers.create_video_record_and_dedup 的导入和签名。""" + from video_processing.dedup_helpers import create_video_record_and_dedup + + # 只验证函数存在且签名正确（不实际调用，需要数据库） + import inspect + + sig = inspect.signature(create_video_record_and_dedup) + params = set(sig.parameters.keys()) + expected = { + "generation_task_id", + "project_id", + "batch_id", + "file_url", + "file_size", + "duration", + "video_path", + "mode", + "session", + "width", + "height", + "fps", + } + assert expected.issubset(params), f"Missing params: {expected - params}" diff --git a/tests/unit/test_unified_render_service.py b/tests/unit/test_unified_render_service.py new file mode 100644 index 000000000..6973de500 --- /dev/null +++ b/tests/unit/test_unified_render_service.py @@ -0,0 +1,404 @@ +"""UnifiedRenderService 单元测试. + +测试图层分组算法、filter_complex 构建、以及渲染流程。 +""" + +from __future__ import annotations + +from dataclasses import dataclass, field +from pathlib import Path +from typing import Any +from unittest.mock import patch, MagicMock + +import pytest + +from video_processing.unified_render_service import ( + ResolvedClip, + RenderLayer, + RenderResult, + UnifiedRenderService, + _resolve_layer_role, +) + + +# ── Fixtures ────────────────────────────────────────────────────────────────── + + +@dataclass +class FakeClip: + """模拟 EditPlanClip。""" + + id: str + plan_id: str = "plan_001" + clip_type: str = "main" + order: int = 0 + asset_id: str = "" + text_content: str = "" + start_time: float = 0.0 + duration: float = 0.0 + transition_effect: str = "cut" + status: str = "ready" + config: dict[str, Any] = field(default_factory=dict) + + +@dataclass +class FakePlan: + """模拟 EditPlan。""" + + id: str = "plan_001" + name: str = "测试计划" + + +def _make_clip( + clip_id: str, + clip_type: str = "main", + order: int = 0, + asset_id: str = "", + duration: float = 0.0, + transition_effect: str = "cut", + config: dict[str, Any] | None = None, +) -> FakeClip: + return FakeClip( + id=clip_id, + clip_type=clip_type, + order=order, + asset_id=asset_id or f"asset_{clip_id}.mp4", + duration=duration, + transition_effect=transition_effect, + config=config or {}, + ) + + +def _make_service( + clips: list[FakeClip] | None = None, + asset_paths: dict[str, Path] | None = None, + work_dir: Path | None = None, +) -> UnifiedRenderService: + """创建测试用的 UnifiedRenderService 实例。 + + 如果未提供 asset_paths，自动从 clips 生成默认映射 + （asset_id → /tmp/asset_{clip_id}.mp4）。 + """ + plan = FakePlan() + clips = clips or [] + work_dir = work_dir or Path("/tmp/test_render") + if asset_paths is None: + asset_paths = {} + for c in clips: + if c.asset_id: + asset_paths[c.asset_id] = Path(f"/tmp/{c.asset_id}") + return UnifiedRenderService( + plan=plan, + clips=clips, + asset_path_map=asset_paths, + work_dir=work_dir, + ) + + +def _patch_path_exists(): + """Patch Path.exists() 让测试路径返回 True。""" + return patch("pathlib.Path.exists", return_value=True) + + +# ── 测试 _resolve_layer_role ───────────────────────────────────────────────── + + +class TestResolveLayerRole: + """测试 clip_type → layer role 映射。""" + + def test_main_default(self): + assert _resolve_layer_role("main", {}) == "main" + + def test_main_with_b_roll_role(self): + assert _resolve_layer_role("main", {"role": "b_roll"}) == "broll" + + def test_overlay(self): + assert _resolve_layer_role("overlay", {}) == "overlay" + + def test_background(self): + assert _resolve_layer_role("background", {}) == "background" + + def test_corner_voice(self): + assert _resolve_layer_role("corner_voice", {}) == "corner_voice" + + def test_b_roll(self): + assert _resolve_layer_role("b_roll", {}) == "broll" + + def test_intro(self): + assert _resolve_layer_role("intro", {}) == "main" + + def test_outro(self): + assert _resolve_layer_role("outro", {}) == "main" + + +# ── 测试图层分组 ────────────────────────────────────────────────────────────── + + +class TestGroupClipsIntoLayers: + """测试 _group_clips_into_layers 方法。""" + + def test_group_clips_one_take(self): + """4 个 main clips → 1 个 main_layer。""" + clips = [ + _make_clip("c1", "main", order=0), + _make_clip("c2", "main", order=1), + _make_clip("c3", "main", order=2), + _make_clip("c4", "main", order=3), + ] + 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 len(layers) == 1 + assert layers[0].role == "main" + assert len(layers[0].clips) == 4 + assert layers[0].z_index == 0 + + def test_group_clips_pip(self): + """1 main + 2 overlay → main_layer + overlay_layer。""" + clips = [ + _make_clip("c1", "main", order=0), + _make_clip("c2", "overlay", order=1), + _make_clip("c3", "overlay", order=2), + ] + 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) + + roles = {l.role for l in layers} + assert "main" in roles + assert "overlay" in roles + + main_layer = next(l for l in layers if l.role == "main") + overlay_layer = next(l for l in layers if l.role == "overlay") + assert len(main_layer.clips) == 1 + assert len(overlay_layer.clips) == 2 + assert overlay_layer.z_index > main_layer.z_index + + def test_group_clips_voice_over(self): + """3 个 main(b_roll) clips → 1 个 broll_layer。""" + clips = [ + _make_clip("c1", "main", order=0, config={"role": "b_roll"}), + _make_clip("c2", "main", order=1, config={"role": "b_roll"}), + _make_clip("c3", "main", order=2, config={"role": "b_roll"}), + ] + 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 len(layers) == 1 + assert layers[0].role == "broll" + assert len(layers[0].clips) == 3 + + def test_group_clips_voice_pip(self): + """1 background + 1 corner_voice + 2 b_roll → 3 layers。""" + clips = [ + _make_clip("c1", "background", order=0), + _make_clip("c2", "corner_voice", order=1), + _make_clip("c3", "b_roll", order=2), + _make_clip("c4", "b_roll", order=3), + ] + 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) + + roles = {l.role for l in layers} + assert roles == {"background", "corner_voice", "broll"} + assert len(layers) == 3 + + # z_index 排序 + assert layers[0].z_index <= layers[1].z_index <= layers[2].z_index + + def test_group_clips_intro_outro(self): + """intro + 2 main + outro → 1 main_layer（4 clips，按 order 排序）。""" + clips = [ + _make_clip("intro", "intro", order=0), + _make_clip("c1", "main", order=1), + _make_clip("c2", "main", order=2), + _make_clip("outro", "outro", order=3), + ] + 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 len(layers) == 1 + assert layers[0].role == "main" + assert len(layers[0].clips) == 4 + # 按 order 排序 + orders = [c.order for c in layers[0].clips] + assert orders == [0, 1, 2, 3] + + +# ── 测试 _resolve_clips ────────────────────────────────────────────────────── + + +class TestResolveClips: + """测试 _resolve_clips 方法。""" + + def test_skip_missing_asset(self): + """跳过 asset_id 在 asset_path_map 中找不到的 clip。""" + clips = [ + _make_clip("c1", "main", order=0, asset_id="asset_1.mp4"), + _make_clip("c2", "main", order=1, asset_id="missing.mp4"), + ] + # 只有 asset_1.mp4 存在 + asset_paths = {"asset_1.mp4": Path("/tmp/asset_1.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() + + assert len(resolved) == 1 + assert resolved[0].clip_id == "c1" + + def test_skip_empty_asset_id(self): + """跳过 asset_id 为空的 clip。""" + clips = [ + _make_clip("c1", "main", order=0, asset_id=""), + _make_clip("c2", "main", order=1, asset_id="asset_2.mp4"), + ] + asset_paths = {"asset_2.mp4": Path("/tmp/asset_2.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() + + assert len(resolved) == 1 + assert resolved[0].clip_id == "c2" + + def test_sort_by_order(self): + """解析后的 clips 按 order 排序。""" + clips = [ + _make_clip("c3", "main", order=3, asset_id="a3.mp4"), + _make_clip("c1", "main", order=1, asset_id="a1.mp4"), + _make_clip("c2", "main", order=2, asset_id="a2.mp4"), + ] + asset_paths = { + "a1.mp4": Path("/tmp/a1.mp4"), + "a2.mp4": Path("/tmp/a2.mp4"), + "a3.mp4": Path("/tmp/a3.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() + + orders = [c.order for c in resolved] + assert orders == [1, 2, 3] + + +# ── 测试 _build_filter_complex ─────────────────────────────────────────────── + + +class TestBuildFilterComplex: + """测试 _build_filter_complex 方法。""" + + def test_single_layer_single_clip(self): + """只有 1 个 main clip → 简单 scale + setpts。""" + clips = [_make_clip("c1", "main", order=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, input_args = svc._build_filter_complex(layers) + + assert "-i" in input_args + assert "/tmp/asset_c1.mp4" in input_args + assert "scale=" in fc + assert "[final_video]" in fc + + def test_single_layer_multi_clips(self): + """多个 main clips → xfade 串联。""" + 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): + resolved = svc._resolve_clips() + layers = svc._group_clips_into_layers(resolved) + fc, input_args = svc._build_filter_complex(layers) + + assert input_args.count("-i") == 2 + assert "xfade=" in fc + assert "[final_video]" in fc + + def test_with_overlay(self): + """main + overlay → overlay 滤镜。""" + clips = [ + _make_clip("c1", "main", order=0), + _make_clip("c2", "overlay", order=1), + ] + 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): + resolved = svc._resolve_clips() + layers = svc._group_clips_into_layers(resolved) + fc, input_args = svc._build_filter_complex(layers) + + assert "overlay=" in fc + assert "[final_video]" in fc + + def test_empty_layers_raises(self): + """空图层列表抛出 ValueError。""" + svc = _make_service() + with pytest.raises(ValueError, match="没有可渲染的图层"): + svc._build_filter_complex([]) + + +# ── 测试 render 方法 ───────────────────────────────────────────────────────── + + +class TestRender: + """测试 render 方法。""" + + def test_render_empty_clips_raises(self): + """没有 clips 时抛出 ValueError。""" + svc = _make_service(clips=[], asset_paths={}) + with pytest.raises(ValueError, match="没有可渲染的片段"): + svc.render() + + def test_render_with_missing_assets_raises(self): + """所有 clips 素材缺失时抛出 ValueError。""" + clips = [_make_clip("c1", "main", order=0, asset_id="missing.mp4")] + svc = _make_service(clips, asset_paths={}) + with pytest.raises(ValueError, match="没有可渲染的片段"): + svc.render() + + def test_render_success(self): + """正常渲染流程。""" + clips = [_make_clip("c1", "main", order=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, "_execute_ffmpeg") as mock_exec, \ + patch.object(svc, "_probe_output", return_value=(5.0, 1024, 1280, 720)): + result = svc.render() + + assert isinstance(result, RenderResult) + assert result.duration == 5.0 + assert result.file_size == 1024 + assert result.width == 1280 + assert result.height == 720 + mock_exec.assert_called_once()