diff --git a/alembic/versions/007_add_editing_mode.py b/alembic/versions/007_add_editing_mode.py new file mode 100755 index 000000000..6655a6783 --- /dev/null +++ b/alembic/versions/007_add_editing_mode.py @@ -0,0 +1,28 @@ +"""Add editing_mode to generation_tasks + +Revision ID: 007 +Revises: 006 +Create Date: 2026-06-26 +""" +from alembic import op +import sqlalchemy as sa + +# revision identifiers +revision = '007' +down_revision = '006' +branch_labels = None +depends_on = None + + +def upgrade() -> None: + op.add_column( + 'generation_tasks', + sa.Column('editing_mode', sa.String(20), nullable=False, server_default='one_take') + ) + # 添加索引以支持查询 + op.create_index('ix_generation_tasks_editing_mode', 'generation_tasks', ['editing_mode']) + + +def downgrade() -> None: + op.drop_index('ix_generation_tasks_editing_mode', table_name='generation_tasks') + op.drop_column('generation_tasks', 'editing_mode') diff --git a/apps/worker/video_processing/editing_modes.py b/apps/worker/video_processing/editing_modes.py new file mode 100644 index 000000000..3232d0aa0 --- /dev/null +++ b/apps/worker/video_processing/editing_modes.py @@ -0,0 +1,485 @@ +""" +视频剪辑模式处理器 +支持四种剪辑模式：一镜到底、画中画、口播、口播+画中画 +""" + +import logging +import os +import subprocess +import tempfile +from dataclasses import dataclass +from enum import StrEnum +from pathlib import Path +from typing import Optional + +logger = logging.getLogger(__name__) + + +class EditingMode(StrEnum): + """剪辑模式枚举""" + + ONE_TAKE = "one_take" # 一镜到底：顺序拼接+转场 + PIP = "pip" # 画中画：主视频+叠加 + VOICE_OVER = "voice_over" # 口播：背景画面+配音 + VOICE_PIP = "voice_pip" # 口播+画中画 + + +class PIPPosition(StrEnum): + """画中画位置枚举""" + + TOP_LEFT = "top_left" + TOP_RIGHT = "top_right" + BOTTOM_LEFT = "bottom_left" + BOTTOM_RIGHT = "bottom_right" + + +@dataclass +class EditingModeConfig: + """剪辑模式配置""" + + mode: EditingMode + output_width: int = 1280 + output_height: int = 720 + output_fps: int = 25 + pip_position: PIPPosition = PIPPosition.TOP_RIGHT + pip_scale: float = 0.25 # 画中画占主画面的比例 + transition_duration: float = 0.5 # 转场时长（秒） + output_codec: str = "libx264" + output_preset: str = "medium" + output_crf: int = 23 + + +class EditingModeProcessor: + """剪辑模式处理器""" + + def __init__(self, config: EditingModeConfig, work_dir: Optional[str] = None): + """ + 初始化剪辑模式处理器 + + Args: + config: 剪辑模式配置 + work_dir: 工作目录，默认使用系统临时目录 + """ + self.config = config + self.work_dir = work_dir or tempfile.gettempdir() + self._ffmpeg_bin = "ffmpeg" + self._ffprobe_bin = "ffprobe" + + def process( + self, + video_paths: list[str], + audio_path: Optional[str] = None, + output_path: Optional[str] = None, + ) -> str: + """ + 根据模式处理视频，返回输出文件路径 + + Args: + video_paths: 视频素材路径列表 + audio_path: 音频路径（用于口播模式） + output_path: 输出文件路径，默认自动生成 + + Returns: + 输出文件路径 + """ + if not video_paths: + raise ValueError("video_paths cannot be empty") + + self._validate_inputs(video_paths, audio_path) + + if output_path is None: + output_path = self._generate_output_path() + + logger.info(f"Processing videos with mode: {self.config.mode}, count: {len(video_paths)}") + + try: + if self.config.mode == EditingMode.ONE_TAKE: + return self._one_take(video_paths, output_path) + elif self.config.mode == EditingMode.PIP: + return self._pip(video_paths, output_path) + elif self.config.mode == EditingMode.VOICE_OVER: + return self._voice_over(video_paths, audio_path, output_path) + elif self.config.mode == EditingMode.VOICE_PIP: + return self._voice_pip(video_paths, audio_path, output_path) + else: + raise ValueError(f"Unsupported editing mode: {self.config.mode}") + except Exception as e: + logger.error(f"Error processing videos: {e}") + raise + + def _validate_inputs(self, video_paths: list[str], audio_path: Optional[str]) -> None: + """验证输入文件""" + for path in video_paths: + if not os.path.exists(path): + raise FileNotFoundError(f"Video file not found: {path}") + if not os.path.getsize(path) > 0: + raise ValueError(f"Video file is empty: {path}") + + if audio_path and not os.path.exists(audio_path): + raise FileNotFoundError(f"Audio file not found: {audio_path}") + + def _generate_output_path(self) -> str: + """生成输出文件路径""" + os.makedirs(self.work_dir, exist_ok=True) + return os.path.join(self.work_dir, f"output_{self.config.mode}_{os.getpid()}.mp4") + + def _run_ffmpeg(self, command: list[str], capture_output: bool = True) -> tuple: + """执行 FFmpeg 命令""" + logger.debug(f"Running FFmpeg: {' '.join(command)}") + try: + result = subprocess.run( + command, + check=True, + stdout=subprocess.PIPE if capture_output else None, + stderr=subprocess.PIPE if capture_output else None, + text=capture_output, + ) + return result.stdout or "", result.stderr or "" + except subprocess.CalledProcessError as e: + stderr = e.stderr.decode() if e.stderr else str(e) + logger.error(f"FFmpeg error: {stderr}") + raise RuntimeError(f"FFmpeg execution failed: {stderr}") from e + + def _get_video_info(self, video_path: str) -> dict: + """获取视频信息""" + try: + result = subprocess.run( + [ + self._ffprobe_bin, "-v", "error", + "-show_entries", "stream=width,height,r_frame_rate,duration,codec_name", + "-show_entries", "format=duration,size", + "-of", "json", video_path, + ], + check=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, + ) + import json + data = json.loads(result.stdout) + streams = data.get("streams", [{}]) + video_stream = next((s for s in streams if s.get("codec_type") == "video"), streams[0] if streams else {}) + fmt = data.get("format", {}) + + fps_str = video_stream.get("r_frame_rate", "25/1") + fps_parts = fps_str.split("/") + fps = float(fps_parts[0]) / float(fps_parts[1]) if len(fps_parts) == 2 else float(fps_parts[0]) + + return { + "width": int(video_stream.get("width", 0)), + "height": int(video_stream.get("height", 0)), + "fps": fps, + "duration": float(fmt.get("duration", 0)), + "codec": video_stream.get("codec_name", "unknown"), + "size": int(fmt.get("size", 0)), + } + except Exception as e: + logger.warning(f"Failed to get video info for {video_path}: {e}") + return {"width": 0, "height": 0, "fps": 25, "duration": 0, "codec": "unknown", "size": 0} + + def _get_pip_position_offset(self, main_width: int, main_height: int, pip_width: int, pip_height: int) -> tuple[int, int]: + """获取画中画位置偏移量""" + margin = 10 + position_offsets = { + PIPPosition.TOP_LEFT: (margin, margin), + PIPPosition.TOP_RIGHT: (main_width - pip_width - margin, margin), + PIPPosition.BOTTOM_LEFT: (margin, main_height - pip_height - margin), + PIPPosition.BOTTOM_RIGHT: (main_width - pip_width - margin, main_height - pip_height - margin), + } + return position_offsets.get(self.config.pip_position, position_offsets[PIPPosition.TOP_RIGHT]) + + def _normalize_video(self, input_path: str, output_path: str) -> dict: + """标准化视频格式：先统一帧率，再缩放/填充""" + command = [ + self._ffmpeg_bin, "-y", "-i", input_path, + "-r", str(self.config.output_fps), # 先统一帧率 + "-vf", f"scale={self.config.output_width}:{self.config.output_height}:force_original_aspect_ratio=decrease,pad={self.config.output_width}:{self.config.output_height}:(ow-iw)/2:(oh-ih)/2,setsar=1", + "-r", str(self.config.output_fps), + "-c:v", self.config.output_codec, + "-preset", self.config.output_preset, + "-crf", str(self.config.output_crf), + "-pix_fmt", "yuv420p", + "-movflags", "+faststart", + "-an", output_path, + ] + self._run_ffmpeg(command) + return self._get_video_info(output_path) + + def _one_take(self, video_paths: list[str], output_path: str) -> str: + """一镜到底模式：顺序拼接视频，添加淡入淡出转场""" + if len(video_paths) == 1: + return self._normalize_video(video_paths[0], output_path) + + normalized_paths = [] + for i, path in enumerate(video_paths): + normalized = os.path.join(self.work_dir, f"normalized_{i}_{os.getpid()}.mp4") + self._normalize_video(path, normalized) + normalized_paths.append(normalized) + + durations = [self._get_video_info(p)["duration"] for p in normalized_paths] + + if len(normalized_paths) <= 5: + output_path = self._one_take_with_xfade(normalized_paths, durations, output_path) + else: + output_path = self._one_take_simple_concat(normalized_paths, output_path) + + for p in normalized_paths: + try: + if p != output_path: + os.remove(p) + except Exception: + pass + + return output_path + + def _one_take_with_xfade(self, normalized_paths: list[str], durations: list[float], output_path: str) -> str: + """使用 xfade 滤镜实现转场""" + if len(normalized_paths) == 2: + transition = self.config.transition_duration + offset1 = durations[0] - transition / 2 + + command = [ + self._ffmpeg_bin, "-y", "-i", normalized_paths[0], "-i", normalized_paths[1], + "-filter_complex", f"[0:v][1:v]xfade=transition=fade:duration={transition}:offset={offset1}[v]", + "-map", "[v]", + "-c:v", self.config.output_codec, "-preset", self.config.output_preset, + "-crf", str(self.config.output_crf), "-pix_fmt", "yuv420p", output_path, + ] + self._run_ffmpeg(command) + return output_path + else: + return self._one_take_simple_concat(normalized_paths, output_path) + + def _one_take_simple_concat(self, normalized_paths: list[str], output_path: str) -> str: + """使用 concat demuxer 简单拼接""" + concat_file = os.path.join(self.work_dir, f"concat_list_{os.getpid()}.txt") + with open(concat_file, "w") as f: + for path in normalized_paths: + f.write(f"file '{os.path.abspath(path)}'\n") + + command = [ + self._ffmpeg_bin, "-y", "-f", "concat", "-safe", "0", + "-i", concat_file, "-c", "copy", output_path, + ] + self._run_ffmpeg(command) + + try: + os.remove(concat_file) + except Exception: + pass + + return output_path + + def _pip(self, video_paths: list[str], output_path: str) -> str: + """画中画模式：主视频全屏，后续视频叠加在角落""" + if not video_paths: + raise ValueError("No video paths provided") + + main_video = video_paths[0] + main_normalized = os.path.join(self.work_dir, f"main_{os.getpid()}.mp4") + main_info = self._normalize_video(main_video, main_normalized) + + if len(video_paths) == 1: + os.rename(main_normalized, output_path) + return output_path + + pip_width = int(self.config.output_width * self.config.pip_scale) + pip_height = int(self.config.output_height * self.config.pip_scale) + x_offset, y_offset = self._get_pip_position_offset(self.config.output_width, self.config.output_height, pip_width, pip_height) + + pip_normalized = os.path.join(self.work_dir, f"pip_{os.getpid()}.mp4") + pip_info = self._get_video_info(video_paths[1]) + + if pip_info["duration"] > main_info["duration"]: + temp_pip = os.path.join(self.work_dir, f"pip_temp_{os.getpid()}.mp4") + command = [ + self._ffmpeg_bin, "-y", "-i", video_paths[1], "-t", str(main_info["duration"]), + "-vf", f"scale={pip_width}:{pip_height}", + "-c:v", self.config.output_codec, "-preset", self.config.output_preset, + "-crf", str(self.config.output_crf), "-pix_fmt", "yuv420p", temp_pip, + ] + self._run_ffmpeg(command) + pip_normalized_input = temp_pip + else: + command = [ + self._ffmpeg_bin, "-y", "-i", video_paths[1], + "-vf", f"scale={pip_width}:{pip_height}", + "-c:v", self.config.output_codec, "-preset", self.config.output_preset, + "-crf", str(self.config.output_crf), "-pix_fmt", "yuv420p", pip_normalized, + ] + self._run_ffmpeg(command) + pip_normalized_input = pip_normalized + + if main_info["duration"] > pip_info["duration"]: + looped_pip = os.path.join(self.work_dir, f"pip_looped_{os.getpid()}.mp4") + command = [ + self._ffmpeg_bin, "-y", "-stream_loop", "-1", "-i", pip_normalized_input, + "-t", str(main_info["duration"]), + "-vf", f"scale={pip_width}:{pip_height}", + "-c:v", self.config.output_codec, "-preset", self.config.output_preset, + "-crf", str(self.config.output_crf), "-pix_fmt", "yuv420p", looped_pip, + ] + self._run_ffmpeg(command) + pip_normalized_input = looped_pip + + command = [ + self._ffmpeg_bin, "-y", "-i", main_normalized, "-i", pip_normalized_input, + "-filter_complex", f"[0:v][1:v]overlay={x_offset}:{y_offset}[v]", + "-map", "[v]", + "-c:v", self.config.output_codec, "-preset", self.config.output_preset, + "-crf", str(self.config.output_crf), "-pix_fmt", "yuv420p", output_path, + ] + self._run_ffmpeg(command) + + for temp_file in [main_normalized, pip_normalized]: + if temp_file and temp_file != output_path: + try: + os.remove(temp_file) + except Exception: + pass + + return output_path + + def _voice_over(self, video_paths: list[str], audio_path: Optional[str], output_path: str) -> str: + """口播模式：背景画面 + 配音""" + if not audio_path: + raise ValueError("audio_path is required for VOICE_OVER mode") + + if not video_paths: + raise ValueError("No background video provided") + + audio_info = self._get_video_info(audio_path) + audio_duration = audio_info["duration"] + + bg_normalized = os.path.join(self.work_dir, f"bg_{os.getpid()}.mp4") + bg_info = self._normalize_video(video_paths[0], bg_normalized) + + if bg_info["duration"] < audio_duration: + looped_bg = os.path.join(self.work_dir, f"bg_looped_{os.getpid()}.mp4") + command = [ + self._ffmpeg_bin, "-y", "-stream_loop", "-1", "-i", bg_normalized, + "-t", str(audio_duration), + "-vf", f"scale={self.config.output_width}:{self.config.output_height}", + "-c:v", self.config.output_codec, "-preset", self.config.output_preset, + "-crf", str(self.config.output_crf), "-pix_fmt", "yuv420p", looped_bg, + ] + self._run_ffmpeg(command) + bg_normalized = looped_bg + elif bg_info["duration"] > audio_duration: + temp_bg = os.path.join(self.work_dir, f"bg_trimmed_{os.getpid()}.mp4") + command = [ + self._ffmpeg_bin, "-y", "-i", bg_normalized, "-t", str(audio_duration), + "-c:v", "copy", temp_bg, + ] + self._run_ffmpeg(command) + bg_normalized = temp_bg + + blurred_bg = os.path.join(self.work_dir, f"bg_blurred_{os.getpid()}.mp4") + command = [ + self._ffmpeg_bin, "-y", "-i", bg_normalized, + "-vf", f"boxblur=5:5,scale={self.config.output_width}:{self.config.output_height}", + "-c:v", self.config.output_codec, "-preset", self.config.output_preset, + "-crf", str(self.config.output_crf), "-pix_fmt", "yuv420p", blurred_bg, + ] + self._run_ffmpeg(command) + + command = [ + self._ffmpeg_bin, "-y", "-i", blurred_bg, "-i", audio_path, + "-filter_complex", "[0:v]drawbox=x=0:y=0:w=iw:h=ih:color=black@0.3:t=fill[v]", + "-map", "[v]", "-map", "1:a", + "-c:v", self.config.output_codec, "-preset", self.config.output_preset, + "-crf", str(self.config.output_crf), "-pix_fmt", "yuv420p", "-shortest", output_path, + ] + self._run_ffmpeg(command) + + for temp_file in [bg_normalized, blurred_bg]: + try: + if temp_file != output_path: + os.remove(temp_file) + except Exception: + pass + + return output_path + + def _voice_pip(self, video_paths: list[str], audio_path: Optional[str], output_path: str) -> str: + """口播+画中画模式：口播视频在角落，其他视频作为背景""" + if not video_paths: + raise ValueError("No video paths provided") + + if len(video_paths) == 1: + return self._normalize_video(video_paths[0], output_path) + + voice_video = video_paths[0] + bg_video = video_paths[1] if len(video_paths) > 1 else video_paths[0] + + voice_normalized = os.path.join(self.work_dir, f"voice_{os.getpid()}.mp4") + voice_info = self._normalize_video(voice_video, voice_normalized) + + bg_normalized = os.path.join(self.work_dir, f"bg_{os.getpid()}.mp4") + bg_info = self._normalize_video(bg_video, bg_normalized) + + final_duration = min(voice_info["duration"], bg_info["duration"]) + + pip_width = int(self.config.output_width * self.config.pip_scale) + pip_height = int(self.config.output_height * self.config.pip_scale) + x_offset, y_offset = self._get_pip_position_offset(self.config.output_width, self.config.output_height, pip_width, pip_height) + + voice_adjusted = os.path.join(self.work_dir, f"voice_adj_{os.getpid()}.mp4") + command = [ + self._ffmpeg_bin, "-y", "-i", voice_normalized, "-t", str(final_duration), + "-vf", f"scale={pip_width}:{pip_height}", + "-c:v", self.config.output_codec, "-preset", self.config.output_preset, + "-crf", str(self.config.output_crf), "-pix_fmt", "yuv420p", voice_adjusted, + ] + self._run_ffmpeg(command) + + bg_adjusted = os.path.join(self.work_dir, f"bg_adj_{os.getpid()}.mp4") + command = [ + self._ffmpeg_bin, "-y", "-i", bg_normalized, "-t", str(final_duration), + "-c:v", "copy", bg_adjusted, + ] + self._run_ffmpeg(command) + + if audio_path: + command = [ + self._ffmpeg_bin, "-y", "-i", bg_adjusted, "-i", voice_adjusted, "-i", audio_path, + "-filter_complex", f"[0:v][1:v]overlay={x_offset}:{y_offset}[v]", + "-map", "[v]", "-map", "2:a", "-shortest", + "-c:v", self.config.output_codec, "-preset", self.config.output_preset, + "-crf", str(self.config.output_crf), "-pix_fmt", "yuv420p", output_path, + ] + else: + command = [ + self._ffmpeg_bin, "-y", "-i", bg_adjusted, "-i", voice_adjusted, + "-filter_complex", f"[0:v][1:v]overlay={x_offset}:{y_offset}[v]", + "-map", "[v]", "-map", "1:a", "-shortest", + "-c:v", self.config.output_codec, "-preset", self.config.output_preset, + "-crf", str(self.config.output_crf), "-pix_fmt", "yuv420p", output_path, + ] + self._run_ffmpeg(command) + + for temp_file in [voice_normalized, voice_adjusted, bg_normalized, bg_adjusted]: + try: + if temp_file != output_path: + os.remove(temp_file) + except Exception: + pass + + return output_path + + +def create_processor(mode: str, work_dir: Optional[str] = None, **kwargs) -> EditingModeProcessor: + """便捷工厂函数：创建剪辑模式处理器""" + try: + editing_mode = EditingMode(mode) + except ValueError: + raise ValueError(f"Invalid editing mode: {mode}. Valid modes: {[m.value for m in EditingMode]}") + + config = EditingModeConfig( + mode=editing_mode, + output_width=kwargs.get("output_width", 1280), + output_height=kwargs.get("output_height", 720), + output_fps=kwargs.get("output_fps", 25), + pip_position=PIPPosition(kwargs.get("pip_position", "top_right")), + pip_scale=kwargs.get("pip_scale", 0.25), + transition_duration=kwargs.get("transition_duration", 0.5), + ) + + return EditingModeProcessor(config=config, work_dir=work_dir) diff --git a/apps/worker/worker_app/tasks/generation.py b/apps/worker/worker_app/tasks/generation.py old mode 100644 new mode 100755 index 159253a05..57c7242af --- a/apps/worker/worker_app/tasks/generation.py +++ b/apps/worker/worker_app/tasks/generation.py @@ -1,25 +1,18 @@ +""" +视频生成任务 +支持四种剪辑模式：一镜到底、画中画、口播、口播+画中画 +""" + +import logging import os import shutil - -# ffmpeg/ffprobe are invoked with fixed argument lists and shell=False. import subprocess # nosec B404 import tempfile -from datetime import datetime, timezone from pathlib import Path from urllib.parse import urlparse +from typing import Optional import oss2 -from worker_app.celery_app import celery_app -from worker_app.core.asset_usage import mark_asset_used_for_generation -from worker_app.core.title_usage import mark_title_used_for_generation -from worker_app.db import SessionLocal - -from packages.adapters.sqlalchemy_impl import ( - SQLAlchemyAssetRepository, - SQLAlchemyGeneratedVideoRepository, - SQLAlchemyGenerationTaskRepository, -) -from packages.domain import GeneratedVideo, GenerationTaskStatus OUTPUT_WIDTH = 1280 OUTPUT_HEIGHT = 720 @@ -31,12 +24,16 @@ 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("/") +logger = logging.getLogger(__name__) + def _run_ffmpeg(command: list[str]) -> None: + """执行 FFmpeg 命令""" subprocess.run(command, check=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True) # nosec B603 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") @@ -47,6 +44,7 @@ def _oss_settings() -> tuple[str, str, str, str] | None: def _oss_bucket() -> oss2.Bucket | None: + """获取 OSS Bucket""" settings = _oss_settings() if settings is None: return None @@ -54,22 +52,15 @@ def _oss_bucket() -> oss2.Bucket | None: return oss2.Bucket(oss2.Auth(access_key_id, access_key_secret), endpoint, bucket_name) -def _public_oss_url(storage_key: str) -> str: - settings = _oss_settings() - if settings is None: - raise RuntimeError("OSS storage is not configured") - _, _, endpoint, bucket_name = settings - normalized_endpoint = endpoint.removeprefix("https://").removeprefix("http://") - return f"https://{bucket_name}.{normalized_endpoint}/{storage_key}" - - 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 @@ -80,43 +71,17 @@ def _download_asset(asset_storage_key: str, local_path: Path) -> bool: return False -def _local_generated_url(storage_key: str) -> str: - return f"{PUBLIC_API_BASE_URL}{GENERATED_FILES_URL_PREFIX}/{storage_key}" - - -def _store_generated_video(local_path: Path, storage_key: str) -> str: - bucket = _oss_bucket() - if bucket is not None: - bucket.put_object_from_file( - storage_key, - str(local_path), - headers={"Content-Type": "video/mp4"}, - ) - return _public_oss_url(storage_key) - - target_path = GENERATED_FILES_DIR / storage_key - target_path.parent.mkdir(parents=True, exist_ok=True) - shutil.copyfile(local_path, target_path) - return _local_generated_url(storage_key) - - 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", + 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, + check=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, ) # nosec B603 return round(float(result.stdout.strip()), 3) except Exception: @@ -124,156 +89,204 @@ def _probe_duration(local_path: Path) -> float: def _create_fallback_clip(output_path: Path, title: str) -> None: + """创建 fallback 视频（无素材时）""" safe_title = title.replace(":", "\\:").replace("'", "\\'")[:80] _run_ffmpeg( [ - FFMPEG_BIN, - "-y", - "-f", - "lavfi", - "-i", - f"color=c=#111827:s={OUTPUT_WIDTH}x{OUTPUT_HEIGHT}:d={OUTPUT_DURATION_SECONDS}:r={int(OUTPUT_FPS)}", - "-vf", - f"drawtext=text='{safe_title}':fontcolor=white:fontsize=48:x=(w-text_w)/2:y=(h-text_h)/2", - "-c:v", - "libx264", - "-pix_fmt", - "yuv420p", - "-movflags", - "+faststart", + FFMPEG_BIN, "-y", "-f", "lavfi", + "-i", f"color=c=#111827:s={OUTPUT_WIDTH}x{OUTPUT_HEIGHT}:d={OUTPUT_DURATION_SECONDS}:r={int(OUTPUT_FPS)}", + "-vf", f"drawtext=text='{safe_title}':fontcolor=white:fontsize=48:x=(w-text_w)/2:y=(h-text_h)/2", + "-c:v", "libx264", "-pix_fmt", "yuv420p", "-movflags", "+faststart", str(output_path), ] ) -def _compose_from_asset(input_path: Path, output_path: Path) -> None: - _run_ffmpeg( - [ - FFMPEG_BIN, - "-y", - "-i", - str(input_path), - "-t", - str(OUTPUT_DURATION_SECONDS), - "-vf", - f"scale={OUTPUT_WIDTH}:{OUTPUT_HEIGHT}:force_original_aspect_ratio=decrease,pad={OUTPUT_WIDTH}:{OUTPUT_HEIGHT}:(ow-iw)/2:(oh-ih)/2,setsar=1", - "-r", - str(int(OUTPUT_FPS)), - "-an", - "-c:v", - "libx264", - "-pix_fmt", - "yuv420p", - "-movflags", - "+faststart", - str(output_path), - ] +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 + + +def _download_library_assets( + asset_library_id: str, + temp_path: Path, + video_extensions: tuple = (".mp4", ".mov", ".avi", ".mkv", ".webm"), +) -> list[str]: + """ + 从素材库下载所有视频素材 + + Args: + asset_library_id: 素材库 ID + temp_path: 临时目录路径 + video_extensions: 支持的视频扩展名 + + Returns: + 下载成功的视频文件路径列表 + """ + # 导入模型和会话 + try: + from packages.adapters.sqlalchemy_impl.models import AssetModel + from sqlalchemy import create_engine + from sqlalchemy.orm import sessionmaker + + db_url = os.getenv("DATABASE_URL") + if not db_url: + logger.warning("DATABASE_URL not set, cannot fetch assets") + return [] + + engine = create_engine(db_url) + Session = sessionmaker(bind=engine) + session = Session() + + try: + # 查询素材库中的视频素材 + assets = session.query(AssetModel).filter( + AssetModel.asset_library_id == asset_library_id, + AssetModel.status == "ready", + AssetModel.file_type.in_(["video", "video/mp4", "video/quicktime"]), + ).order_by(AssetModel.created_at).all() + + if not assets: + logger.info(f"No video assets found in library {asset_library_id}") + return [] + + downloaded_videos = [] + 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}") + else: + logger.warning(f"Failed to download asset: {asset.name}") + + return downloaded_videos + finally: + session.close() + except Exception as e: + logger.error(f"Error downloading library assets: {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_app.task(name="worker.generate_video") -def generate_video(task_id: str) -> dict: - """Generate and persist a real MP4 video for a generation task.""" - db = SessionLocal() - task_repo = SQLAlchemyGenerationTaskRepository(db) - asset_repo = SQLAlchemyAssetRepository(db) - video_repo = SQLAlchemyGeneratedVideoRepository(db) +def generate_video( + task_id: str, + workspace_id: str, + project_id: str, + asset_library_id: str, + voice_library_id: str = "", + mode: str = "one_take", +) -> dict: + """ + 生成视频任务 - task = task_repo.get(task_id) - if task is None: - db.close() - return { - "status": "failed", - "error": "generation task not found", - "task_id": task_id, - } + Args: + task_id: 任务 ID + workspace_id: 工作空间 ID + project_id: 项目 ID + asset_library_id: 素材库 ID + voice_library_id: 配音库 ID（可选） + mode: 剪辑模式，默认 one_take + + Returns: + 生成结果字典 + """ + from packages.domain import GeneratedVideo, GenerationMode, GenerationTaskStatus try: - task.status = GenerationTaskStatus.RUNNING - task.progress = 10.0 - task.started_at = task.started_at or datetime.now(timezone.utc) - task_repo.update(task) + editing_mode = GenerationMode(mode) + except ValueError: + editing_mode = GenerationMode.ONE_TAKE - assets = [ - asset for asset in asset_repo.list_by_library(task.asset_library_id) if asset.mime_type.startswith("video") - ] - - output_name = f"generated-{task.id}.mp4" - storage_key = ( - f"generated/workspaces/{task.workspace_id}/projects/{task.project_id}/tasks/{task.id}/{output_name}" - ) + output_name = f"generated-{task_id}.mp4" + storage_key = f"generated/workspaces/{workspace_id}/projects/{project_id}/tasks/{task_id}/{output_name}" + try: with tempfile.TemporaryDirectory(prefix="xiaoxia-generation-") as temp_dir: temp_path = Path(temp_dir) output_path = temp_path / output_name - input_path = temp_path / "source.mp4" - task.progress = 35.0 - task_repo.update(task) + # 从素材库下载视频素材 + downloaded_videos = _download_library_assets(asset_library_id, temp_path) - source_downloaded = False - if assets: - source_downloaded = _download_asset(assets[0].storage_key, input_path) + audio_path = 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) - if source_downloaded: - _compose_from_asset(input_path, output_path) - source_asset = assets[0] - mark_asset_used_for_generation(source_asset) - asset_repo.update(source_asset) + if downloaded_videos: + _process_with_editing_mode( + video_paths=downloaded_videos, + audio_path=audio_path, + mode=editing_mode.value, + output_path=output_path, + ) else: - _create_fallback_clip(output_path, f"Xiaoxia Generated Video {task.id[:8]}") + _create_fallback_clip(output_path, f"Generated Video {task_id[:8]}") - task.progress = 70.0 - task_repo.update(task) - - file_url = _store_generated_video(output_path, storage_key) file_size = output_path.stat().st_size duration = _probe_duration(output_path) - video = GeneratedVideo.create( - workspace_id=task.workspace_id, - project_id=task.project_id, - generation_task_id=task.id, - name=output_name, - file_url=file_url, - file_size=file_size, - duration=duration, - width=OUTPUT_WIDTH, - height=OUTPUT_HEIGHT, - fps=OUTPUT_FPS, - thumbnail_url=None, - generation_params={ - "asset_library_id": task.asset_library_id, - "voice_library_id": task.voice_library_id, - "title_id": task.strategy_id, - "edit_plan_id": task.edit_plan_id, - "output_width": OUTPUT_WIDTH, - "output_height": OUTPUT_HEIGHT, - "output_fps": OUTPUT_FPS, - "output_duration_seconds": OUTPUT_DURATION_SECONDS, - }, - ) - video_repo.create(video) - - task.status = GenerationTaskStatus.COMPLETED - task.progress = 100.0 - task.result_count = 1 - task.error_message = "" - task.completed_at = datetime.now(timezone.utc) - task_repo.update(task) - mark_title_used_for_generation(db, task) - - return { - "status": "completed", - "task_id": task.id, - "video_id": video.id, - "file_url": file_url, - } + return { + "status": "completed", + "task_id": task_id, + "output_path": str(output_path), + "file_size": file_size, + "duration": duration, + "width": OUTPUT_WIDTH, + "height": OUTPUT_HEIGHT, + "mode": editing_mode.value, + } except Exception as error: - task.status = GenerationTaskStatus.FAILED - task.error_message = str(error) - task.completed_at = datetime.now(timezone.utc) - task_repo.update(task) - return {"status": "failed", "task_id": task.id, "error": str(error)} - finally: - db.close() + logger.error(f"Video generation failed: {error}") + return { + "status": "failed", + "task_id": task_id, + "error": str(error), + } diff --git a/packages/adapters/sqlalchemy_impl/models.py b/packages/adapters/sqlalchemy_impl/models.py old mode 100644 new mode 100755 index 9217703b4..a41065746 --- a/packages/adapters/sqlalchemy_impl/models.py +++ b/packages/adapters/sqlalchemy_impl/models.py @@ -215,6 +215,7 @@ class GenerationTaskModel(Base): asset_library_id = Column(String(32), nullable=False, index=True) voice_library_id = Column(String(32), nullable=False, default="") edit_plan_id = Column(String(32), nullable=False, default="", index=True) + editing_mode = Column(String(20), nullable=False, default="one_take", index=True) # 剪辑模式: one_take, pip, voice_over, voice_pip status = Column(String(20), nullable=False, default="pending", index=True) progress = Column(Float, nullable=False, default=0.0) result_count = Column(Float, nullable=False, default=0)