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')

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)

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),
+        }

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)