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