refactor(#1341): 删除 EditingPlanner 遗留封面生成器 + 清理 compose_video 封面代码 #1353
@@ -1,431 +0,0 @@
|
||||
"""视频封面生成器 — 从视频中提取/生成封面图.
|
||||
|
||||
支持能力：
|
||||
- 指定时间点抽帧（默认第1秒）
|
||||
- 智能封面：抽取多帧选最清晰的一帧
|
||||
- 自定义上传封面图（直接返回路径）
|
||||
- 生成的封面图保存为 JPEG 格式，可复用
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
import subprocess
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
from video_processing.ffmpeg_utils import FFMPEG_BIN, probe_video_info, run_ffmpeg
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
# ── 配置常量 ──────────────────────────────────────────────────────────────────
|
||||
|
||||
# 智能封面抽帧数量
|
||||
SMART_COVER_FRAME_COUNT = 3
|
||||
|
||||
# 默认抽帧时间点（秒）
|
||||
DEFAULT_COVER_TIME = 1.0
|
||||
|
||||
# 封面输出尺寸（宽x高）
|
||||
DEFAULT_COVER_WIDTH = 1080
|
||||
DEFAULT_COVER_HEIGHT = 1920
|
||||
|
||||
# 封面质量（JPEG quality 1-31，越小质量越高）
|
||||
DEFAULT_COVER_QUALITY = 5
|
||||
|
||||
|
||||
# ── 数据模型 ──────────────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
class CoverGenerator:
|
||||
"""视频封面生成器.
|
||||
|
||||
三种模式：
|
||||
1. 指定时间点抽帧：从视频指定时间提取一帧
|
||||
2. 智能封面：抽取3帧，用 blur 检测选最清晰的
|
||||
3. 自定义上传：直接使用用户上传的图片
|
||||
"""
|
||||
|
||||
@staticmethod
|
||||
def extract_frame(
|
||||
video_path: str | Path,
|
||||
output_path: str | Path,
|
||||
*,
|
||||
time_sec: float = DEFAULT_COVER_TIME,
|
||||
width: int = DEFAULT_COVER_WIDTH,
|
||||
height: int = DEFAULT_COVER_HEIGHT,
|
||||
quality: int = DEFAULT_COVER_QUALITY,
|
||||
) -> Path:
|
||||
"""从视频指定时间点提取一帧作为封面.
|
||||
|
||||
Args:
|
||||
video_path: 视频文件路径
|
||||
output_path: 输出图片路径
|
||||
time_sec: 抽帧时间点（秒）
|
||||
width: 输出宽度
|
||||
height: 输出高度
|
||||
quality: JPEG 质量（1-31，越小越好）
|
||||
|
||||
Returns:
|
||||
封面图片路径
|
||||
|
||||
Raises:
|
||||
FileNotFoundError: 视频文件不存在
|
||||
subprocess.CalledProcessError: FFmpeg 执行失败
|
||||
"""
|
||||
video_path = Path(video_path)
|
||||
output_path = Path(output_path)
|
||||
|
||||
if not video_path.exists():
|
||||
raise FileNotFoundError(f"视频文件不存在: {video_path}")
|
||||
|
||||
# 确保输出目录存在
|
||||
output_path.parent.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
# 安全钳制时间
|
||||
info = probe_video_info(str(video_path))
|
||||
duration = info.get("duration", 0.0)
|
||||
if duration > 0 and time_sec >= duration:
|
||||
# 超过视频长度，取中间帧
|
||||
time_sec = max(0, duration / 2)
|
||||
if time_sec < 0:
|
||||
time_sec = 0
|
||||
|
||||
# scale + crop 实现 cover 裁剪（铺满输出尺寸）
|
||||
vf = f"scale={width}:{height}:force_original_aspect_ratio=increase," f"crop={width}:{height},format=yuvj420p"
|
||||
|
||||
command = [
|
||||
FFMPEG_BIN,
|
||||
"-y",
|
||||
"-ss",
|
||||
f"{time_sec:.3f}",
|
||||
"-i",
|
||||
str(video_path),
|
||||
"-vframes",
|
||||
"1",
|
||||
"-vf",
|
||||
vf,
|
||||
"-q:v",
|
||||
str(quality),
|
||||
"-f",
|
||||
"mjpeg",
|
||||
str(output_path),
|
||||
]
|
||||
|
||||
logger.info("抽取视频封面: video=%s time=%.2fs output=%s", video_path.name, time_sec, output_path.name)
|
||||
run_ffmpeg(command)
|
||||
|
||||
if not output_path.exists() or output_path.stat().st_size == 0:
|
||||
raise RuntimeError(f"封面生成失败: {output_path}")
|
||||
|
||||
return output_path
|
||||
|
||||
@staticmethod
|
||||
def extract_smart_cover(
|
||||
video_path: str | Path,
|
||||
output_path: str | Path,
|
||||
*,
|
||||
frame_count: int = SMART_COVER_FRAME_COUNT,
|
||||
width: int = DEFAULT_COVER_WIDTH,
|
||||
height: int = DEFAULT_COVER_HEIGHT,
|
||||
quality: int = DEFAULT_COVER_QUALITY,
|
||||
work_dir: str | Path | None = None,
|
||||
) -> Path:
|
||||
"""智能封面：抽取多帧，选最清晰的一帧.
|
||||
|
||||
清晰度判断：使用拉普拉斯方差（Variance of Laplacian），
|
||||
方差越大表示图像边缘越丰富，越清晰。
|
||||
|
||||
Args:
|
||||
video_path: 视频文件路径
|
||||
output_path: 最终输出封面路径
|
||||
frame_count: 抽帧数量（均匀分布在视频中）
|
||||
width: 输出宽度
|
||||
height: 输出高度
|
||||
quality: JPEG 质量
|
||||
work_dir: 临时工作目录（默认输出目录的父目录）
|
||||
|
||||
Returns:
|
||||
最佳封面图片路径
|
||||
"""
|
||||
video_path = Path(video_path)
|
||||
output_path = Path(output_path)
|
||||
|
||||
if not video_path.exists():
|
||||
raise FileNotFoundError(f"视频文件不存在: {video_path}")
|
||||
|
||||
# 获取视频时长
|
||||
info = probe_video_info(str(video_path))
|
||||
duration = info.get("duration", 0.0)
|
||||
|
||||
if duration <= 0 or frame_count <= 1:
|
||||
# 无法获取时长或只有1帧，退化为普通抽帧
|
||||
return CoverGenerator.extract_frame(
|
||||
video_path,
|
||||
output_path,
|
||||
time_sec=min(DEFAULT_COVER_TIME, max(0, duration / 2)),
|
||||
width=width,
|
||||
height=height,
|
||||
quality=quality,
|
||||
)
|
||||
|
||||
# 临时目录
|
||||
if work_dir is None:
|
||||
work_dir = output_path.parent
|
||||
work_dir = Path(work_dir)
|
||||
work_dir.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
# 均匀分布抽帧时间点（跳过首尾5%）
|
||||
start_pct = 0.05
|
||||
end_pct = 0.95
|
||||
if frame_count == 1:
|
||||
time_points = [duration * 0.5]
|
||||
else:
|
||||
step = (end_pct - start_pct) / (frame_count - 1)
|
||||
time_points = [duration * (start_pct + step * i) for i in range(frame_count)]
|
||||
|
||||
# 抽取候选帧
|
||||
candidate_frames: list[tuple[float, Path]] = []
|
||||
for i, t in enumerate(time_points):
|
||||
frame_path = work_dir / f"cover_candidate_{i}.jpg"
|
||||
try:
|
||||
CoverGenerator.extract_frame(
|
||||
video_path,
|
||||
frame_path,
|
||||
time_sec=t,
|
||||
width=width,
|
||||
height=height,
|
||||
quality=quality,
|
||||
)
|
||||
candidate_frames.append((t, frame_path))
|
||||
except Exception as e:
|
||||
logger.warning("智能封面抽帧失败（t=%.2fs）: %s", t, e)
|
||||
continue
|
||||
|
||||
if not candidate_frames:
|
||||
# 全部失败，退化到普通抽帧
|
||||
logger.warning("智能封面所有候选帧抽取失败，退化为普通抽帧")
|
||||
return CoverGenerator.extract_frame(
|
||||
video_path,
|
||||
output_path,
|
||||
time_sec=min(DEFAULT_COVER_TIME, duration / 2),
|
||||
width=width,
|
||||
height=height,
|
||||
quality=quality,
|
||||
)
|
||||
|
||||
if len(candidate_frames) == 1:
|
||||
# 只有一帧，直接用
|
||||
import shutil
|
||||
|
||||
shutil.copy2(candidate_frames[0][1], output_path)
|
||||
return output_path
|
||||
|
||||
# 计算每帧清晰度（用 FFmpeg 的 stats 滤镜或简化处理）
|
||||
# 简化方案：比较文件大小（同一尺寸下，JPEG文件越大通常细节越丰富、越清晰）
|
||||
# 更准确的方案是用拉普拉斯方差，但需要额外依赖
|
||||
# 这里用文件大小作为近似指标
|
||||
best_frame = max(candidate_frames, key=lambda x: x[1].stat().st_size)
|
||||
|
||||
# 复制最佳帧到输出路径
|
||||
import shutil
|
||||
|
||||
shutil.copy2(best_frame[1], output_path)
|
||||
|
||||
logger.info(
|
||||
"智能封面生成完成: 候选%d帧, 最佳t=%.2fs, 大小=%d字节",
|
||||
len(candidate_frames),
|
||||
best_frame[0],
|
||||
output_path.stat().st_size,
|
||||
)
|
||||
|
||||
# 清理临时文件
|
||||
for _, fp in candidate_frames:
|
||||
try:
|
||||
fp.unlink()
|
||||
except OSError:
|
||||
pass
|
||||
|
||||
return output_path
|
||||
|
||||
@staticmethod
|
||||
def process_custom_cover(
|
||||
image_path: str | Path,
|
||||
output_path: str | Path,
|
||||
*,
|
||||
width: int = DEFAULT_COVER_WIDTH,
|
||||
height: int = DEFAULT_COVER_HEIGHT,
|
||||
quality: int = DEFAULT_COVER_QUALITY,
|
||||
) -> Path:
|
||||
"""处理用户自定义上传的封面图.
|
||||
|
||||
调整尺寸、格式转换为标准封面格式。
|
||||
|
||||
Args:
|
||||
image_path: 用户上传的图片路径
|
||||
output_path: 输出封面路径
|
||||
width: 目标宽度
|
||||
height: 目标高度
|
||||
quality: JPEG 质量
|
||||
|
||||
Returns:
|
||||
处理后的封面图片路径
|
||||
"""
|
||||
image_path = Path(image_path)
|
||||
output_path = Path(output_path)
|
||||
|
||||
if not image_path.exists():
|
||||
raise FileNotFoundError(f"封面图片不存在: {image_path}")
|
||||
|
||||
output_path.parent.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
# scale + crop 实现 cover 裁剪
|
||||
vf = f"scale={width}:{height}:force_original_aspect_ratio=increase," f"crop={width}:{height},format=yuvj420p"
|
||||
|
||||
command = [
|
||||
FFMPEG_BIN,
|
||||
"-y",
|
||||
"-i",
|
||||
str(image_path),
|
||||
"-vf",
|
||||
vf,
|
||||
"-q:v",
|
||||
str(quality),
|
||||
"-f",
|
||||
"mjpeg",
|
||||
str(output_path),
|
||||
]
|
||||
|
||||
logger.info("处理自定义封面: input=%s output=%s", image_path.name, output_path.name)
|
||||
|
||||
try:
|
||||
run_ffmpeg(command)
|
||||
except subprocess.CalledProcessError:
|
||||
# 处理失败，直接复制原图
|
||||
logger.warning("自定义封面处理失败，使用原图")
|
||||
import shutil
|
||||
|
||||
shutil.copy2(image_path, output_path)
|
||||
|
||||
return output_path
|
||||
|
||||
@staticmethod
|
||||
def generate_cover(
|
||||
video_path: str | Path,
|
||||
output_path: str | Path,
|
||||
*,
|
||||
mode: str = "smart", # smart / time / custom
|
||||
time_sec: float = DEFAULT_COVER_TIME,
|
||||
custom_image: str | Path | None = None,
|
||||
width: int = DEFAULT_COVER_WIDTH,
|
||||
height: int = DEFAULT_COVER_HEIGHT,
|
||||
quality: int = DEFAULT_COVER_QUALITY,
|
||||
) -> Path:
|
||||
"""统一封面生成入口.
|
||||
|
||||
Args:
|
||||
video_path: 视频文件路径
|
||||
output_path: 输出封面路径
|
||||
mode: 模式 - smart（智能选帧）/ time（指定时间）/ custom（自定义图片）
|
||||
time_sec: time 模式下的抽帧时间点
|
||||
custom_image: custom 模式下的自定义图片路径
|
||||
width: 输出宽度
|
||||
height: 输出高度
|
||||
quality: JPEG 质量
|
||||
|
||||
Returns:
|
||||
封面图片路径
|
||||
"""
|
||||
if mode == "custom" and custom_image:
|
||||
return CoverGenerator.process_custom_cover(
|
||||
custom_image,
|
||||
output_path,
|
||||
width=width,
|
||||
height=height,
|
||||
quality=quality,
|
||||
)
|
||||
elif mode == "time":
|
||||
return CoverGenerator.extract_frame(
|
||||
video_path,
|
||||
output_path,
|
||||
time_sec=time_sec,
|
||||
width=width,
|
||||
height=height,
|
||||
quality=quality,
|
||||
)
|
||||
else:
|
||||
# 默认智能封面
|
||||
return CoverGenerator.extract_smart_cover(
|
||||
video_path,
|
||||
output_path,
|
||||
width=width,
|
||||
height=height,
|
||||
quality=quality,
|
||||
)
|
||||
|
||||
|
||||
# ── 便捷函数 ──────────────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
def generate_cover_from_plan(
|
||||
plan: Any,
|
||||
video_path: str | Path,
|
||||
output_dir: str | Path,
|
||||
) -> Path | None:
|
||||
"""从 EditPlan 配置生成封面图.
|
||||
|
||||
配置读取：plan.config.cover_config
|
||||
支持字段：
|
||||
- mode: smart / time / custom
|
||||
- time_sec: 抽帧时间（time模式）
|
||||
- custom_image_url: 自定义图片URL（需要先下载到本地）
|
||||
|
||||
Args:
|
||||
plan: EditPlan 对象
|
||||
video_path: 渲染后的视频路径
|
||||
output_dir: 封面输出目录
|
||||
|
||||
Returns:
|
||||
封面图片路径，或 None（不需要生成封面时）
|
||||
"""
|
||||
config = getattr(plan, "config", None) or {}
|
||||
cover_config = config.get("cover_config") if isinstance(config, dict) else None
|
||||
|
||||
if not cover_config:
|
||||
return None
|
||||
|
||||
mode = cover_config.get("mode", "smart")
|
||||
output_dir = Path(output_dir)
|
||||
output_dir.mkdir(parents=True, exist_ok=True)
|
||||
output_path = output_dir / f"cover_{plan.id}.jpg"
|
||||
|
||||
try:
|
||||
if mode == "custom":
|
||||
# 自定义封面：需要先有本地图片路径
|
||||
custom_path = cover_config.get("custom_image_path")
|
||||
if custom_path and Path(custom_path).exists():
|
||||
return CoverGenerator.process_custom_cover(
|
||||
custom_path,
|
||||
output_path,
|
||||
)
|
||||
else:
|
||||
logger.warning("自定义封面图片路径无效，退化为智能封面")
|
||||
mode = "smart"
|
||||
|
||||
if mode == "time":
|
||||
time_sec = float(cover_config.get("time_sec", DEFAULT_COVER_TIME))
|
||||
return CoverGenerator.extract_frame(
|
||||
video_path,
|
||||
output_path,
|
||||
time_sec=time_sec,
|
||||
)
|
||||
else:
|
||||
# smart
|
||||
return CoverGenerator.extract_smart_cover(
|
||||
video_path,
|
||||
output_path,
|
||||
)
|
||||
except Exception as e:
|
||||
logger.warning("封面生成失败: %s", e)
|
||||
return None
|
||||
@@ -163,36 +163,6 @@ def _compose_with_unified_engine(task, job_service, job, plan_id: str, db) -> di
|
||||
job_service.fail_job(job_id, error_msg[:500])
|
||||
raise RuntimeError(result.error_message)
|
||||
|
||||
# 生成封面（如果配置启用）
|
||||
cover_url = None
|
||||
try:
|
||||
from video_processing.cover_generator import generate_cover_from_plan
|
||||
|
||||
from packages.adapters.sqlalchemy_impl.edit_plan_repository import (
|
||||
SQLAlchemyEditPlanRepository as EditPlanRepository,
|
||||
)
|
||||
|
||||
# 获取 plan 对象
|
||||
plan_repo = EditPlanRepository(db)
|
||||
plan = plan_repo.get(plan_id)
|
||||
|
||||
if plan and result.output_path:
|
||||
# 检查 cover_config
|
||||
cover_config = (plan.config or {}).get("cover_config")
|
||||
if cover_config and cover_config.get("enabled", False):
|
||||
from pathlib import Path
|
||||
|
||||
output_dir = Path(result.output_path).parent
|
||||
cover_path = generate_cover_from_plan(plan, result.output_path, output_dir)
|
||||
if cover_path:
|
||||
# 生成 cover_url（相对路径或上传到存储）
|
||||
cover_url = f"/covers/{plan_id}.jpg"
|
||||
logger.info("封面生成成功: plan_id=%s cover_path=%s", plan_id, cover_path)
|
||||
else:
|
||||
logger.info("封面生成未启用: plan_id=%s", plan_id)
|
||||
except Exception as e:
|
||||
logger.warning("封面生成失败（不影响视频合成）: plan_id=%s error=%s", plan_id, e)
|
||||
|
||||
# 更新 Job 状态为完成
|
||||
result_data = {
|
||||
"plan_id": plan_id,
|
||||
@@ -205,7 +175,6 @@ def _compose_with_unified_engine(task, job_service, job, plan_id: str, db) -> di
|
||||
"width": result.width,
|
||||
"height": result.height,
|
||||
"file_size": result.file_size,
|
||||
"cover_url": cover_url,
|
||||
}
|
||||
job_service.complete_job(job_id, result=result_data)
|
||||
|
||||
|
||||
@@ -1,232 +0,0 @@
|
||||
"""测试 compose_video 任务中封面生成集成."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from pathlib import Path
|
||||
from unittest.mock import MagicMock, Mock, patch
|
||||
|
||||
import pytest
|
||||
|
||||
|
||||
class TestComposeVideoCoverIntegration:
|
||||
"""测试视频合成任务中的封面生成集成."""
|
||||
|
||||
@pytest.fixture
|
||||
def mock_job_service(self):
|
||||
"""模拟 JobService."""
|
||||
service = MagicMock()
|
||||
service.get_job.return_value = MagicMock(
|
||||
id="job_123",
|
||||
payload={"plan_id": "plan_456"},
|
||||
)
|
||||
return service
|
||||
|
||||
@pytest.fixture
|
||||
def mock_db(self):
|
||||
"""模拟数据库会话."""
|
||||
return MagicMock()
|
||||
|
||||
@pytest.fixture
|
||||
def mock_render_result(self):
|
||||
"""模拟渲染结果."""
|
||||
result = MagicMock()
|
||||
result.success = True
|
||||
result.output_path = Path("/tmp/output/video_123.mp4")
|
||||
result.output_url = "https://example.com/video_123.mp4"
|
||||
result.duration = 30.0
|
||||
result.clip_count = 5
|
||||
result.width = 1080
|
||||
result.height = 1920
|
||||
result.file_size = 1024000
|
||||
return result
|
||||
|
||||
@pytest.fixture
|
||||
def mock_plan_with_cover_enabled(self):
|
||||
"""模拟启用封面的 plan."""
|
||||
plan = MagicMock()
|
||||
plan.id = "plan_456"
|
||||
plan.config = {
|
||||
"cover_config": {
|
||||
"enabled": True,
|
||||
"mode": "smart",
|
||||
}
|
||||
}
|
||||
return plan
|
||||
|
||||
@pytest.fixture
|
||||
def mock_plan_with_cover_disabled(self):
|
||||
"""模拟禁用封面的 plan."""
|
||||
plan = MagicMock()
|
||||
plan.id = "plan_456"
|
||||
plan.config = {
|
||||
"cover_config": {
|
||||
"enabled": False,
|
||||
}
|
||||
}
|
||||
return plan
|
||||
|
||||
@pytest.fixture
|
||||
def mock_plan_without_cover_config(self):
|
||||
"""模拟没有 cover_config 的 plan."""
|
||||
plan = MagicMock()
|
||||
plan.id = "plan_456"
|
||||
plan.config = {}
|
||||
return plan
|
||||
|
||||
def test_cover_generation_called_when_enabled(
|
||||
self,
|
||||
mock_job_service,
|
||||
mock_db,
|
||||
mock_render_result,
|
||||
mock_plan_with_cover_enabled,
|
||||
):
|
||||
"""测试封面生成在启用时被调用."""
|
||||
from worker_app.tasks.compose_video import _compose_with_unified_engine
|
||||
|
||||
# 模拟 RenderAdapter
|
||||
with patch("video_processing.render_adapter.RenderAdapter") as MockAdapter:
|
||||
adapter_instance = MagicMock()
|
||||
adapter_instance.render_plan.return_value = mock_render_result
|
||||
adapter_instance.validate_plan.return_value = (True, [], [], 5, 5)
|
||||
MockAdapter.return_value = adapter_instance
|
||||
|
||||
# 模拟 EditPlanRepository
|
||||
with patch(
|
||||
"packages.adapters.sqlalchemy_impl.edit_plan_repository.SQLAlchemyEditPlanRepository"
|
||||
) as MockPlanRepo:
|
||||
plan_repo_instance = MagicMock()
|
||||
plan_repo_instance.get.return_value = mock_plan_with_cover_enabled
|
||||
MockPlanRepo.return_value = plan_repo_instance
|
||||
|
||||
# 模拟 generate_cover_from_plan
|
||||
with patch("video_processing.cover_generator.generate_cover_from_plan") as mock_gen_cover:
|
||||
mock_gen_cover.return_value = Path("/tmp/output/cover_plan_456.jpg")
|
||||
|
||||
# 执行
|
||||
task = MagicMock()
|
||||
result = _compose_with_unified_engine(
|
||||
task, mock_job_service, mock_job_service.get_job(), "plan_456", mock_db
|
||||
)
|
||||
|
||||
# 验证封面生成被调用
|
||||
mock_gen_cover.assert_called_once()
|
||||
call_args = mock_gen_cover.call_args
|
||||
assert call_args[0][0] == mock_plan_with_cover_enabled # plan
|
||||
assert call_args[0][1] == mock_render_result.output_path # video_path
|
||||
assert call_args[0][2] == mock_render_result.output_path.parent # output_dir
|
||||
|
||||
# 验证结果包含 cover_url
|
||||
assert "cover_url" in result["result"]
|
||||
assert result["result"]["cover_url"] == "/covers/plan_456.jpg"
|
||||
|
||||
def test_cover_generation_skipped_when_disabled(
|
||||
self,
|
||||
mock_job_service,
|
||||
mock_db,
|
||||
mock_render_result,
|
||||
mock_plan_with_cover_disabled,
|
||||
):
|
||||
"""测试封面生成在禁用时被跳过."""
|
||||
from worker_app.tasks.compose_video import _compose_with_unified_engine
|
||||
|
||||
with patch("video_processing.render_adapter.RenderAdapter") as MockAdapter:
|
||||
adapter_instance = MagicMock()
|
||||
adapter_instance.render_plan.return_value = mock_render_result
|
||||
adapter_instance.validate_plan.return_value = (True, [], [], 5, 5)
|
||||
MockAdapter.return_value = adapter_instance
|
||||
|
||||
with patch(
|
||||
"packages.adapters.sqlalchemy_impl.edit_plan_repository.SQLAlchemyEditPlanRepository"
|
||||
) as MockPlanRepo:
|
||||
plan_repo_instance = MagicMock()
|
||||
plan_repo_instance.get.return_value = mock_plan_with_cover_disabled
|
||||
MockPlanRepo.return_value = plan_repo_instance
|
||||
|
||||
with patch("video_processing.cover_generator.generate_cover_from_plan") as mock_gen_cover:
|
||||
task = MagicMock()
|
||||
result = _compose_with_unified_engine(
|
||||
task, mock_job_service, mock_job_service.get_job(), "plan_456", mock_db
|
||||
)
|
||||
|
||||
# 验证封面生成未被调用
|
||||
mock_gen_cover.assert_not_called()
|
||||
|
||||
# 验证结果中 cover_url 为 None
|
||||
assert "cover_url" in result["result"]
|
||||
assert result["result"]["cover_url"] is None
|
||||
|
||||
def test_cover_generation_skipped_when_no_config(
|
||||
self,
|
||||
mock_job_service,
|
||||
mock_db,
|
||||
mock_render_result,
|
||||
mock_plan_without_cover_config,
|
||||
):
|
||||
"""测试没有 cover_config 时封面生成被跳过."""
|
||||
from worker_app.tasks.compose_video import _compose_with_unified_engine
|
||||
|
||||
with patch("video_processing.render_adapter.RenderAdapter") as MockAdapter:
|
||||
adapter_instance = MagicMock()
|
||||
adapter_instance.render_plan.return_value = mock_render_result
|
||||
adapter_instance.validate_plan.return_value = (True, [], [], 5, 5)
|
||||
MockAdapter.return_value = adapter_instance
|
||||
|
||||
with patch(
|
||||
"packages.adapters.sqlalchemy_impl.edit_plan_repository.SQLAlchemyEditPlanRepository"
|
||||
) as MockPlanRepo:
|
||||
plan_repo_instance = MagicMock()
|
||||
plan_repo_instance.get.return_value = mock_plan_without_cover_config
|
||||
MockPlanRepo.return_value = plan_repo_instance
|
||||
|
||||
with patch("video_processing.cover_generator.generate_cover_from_plan") as mock_gen_cover:
|
||||
task = MagicMock()
|
||||
result = _compose_with_unified_engine(
|
||||
task, mock_job_service, mock_job_service.get_job(), "plan_456", mock_db
|
||||
)
|
||||
|
||||
# 验证封面生成未被调用
|
||||
mock_gen_cover.assert_not_called()
|
||||
|
||||
# 验证结果中 cover_url 为 None
|
||||
assert "cover_url" in result["result"]
|
||||
assert result["result"]["cover_url"] is None
|
||||
|
||||
def test_cover_generation_failure_does_not_break_video(
|
||||
self,
|
||||
mock_job_service,
|
||||
mock_db,
|
||||
mock_render_result,
|
||||
mock_plan_with_cover_enabled,
|
||||
):
|
||||
"""测试封面生成失败不影响视频合成."""
|
||||
from worker_app.tasks.compose_video import _compose_with_unified_engine
|
||||
|
||||
with patch("video_processing.render_adapter.RenderAdapter") as MockAdapter:
|
||||
adapter_instance = MagicMock()
|
||||
adapter_instance.render_plan.return_value = mock_render_result
|
||||
adapter_instance.validate_plan.return_value = (True, [], [], 5, 5)
|
||||
MockAdapter.return_value = adapter_instance
|
||||
|
||||
with patch(
|
||||
"packages.adapters.sqlalchemy_impl.edit_plan_repository.SQLAlchemyEditPlanRepository"
|
||||
) as MockPlanRepo:
|
||||
plan_repo_instance = MagicMock()
|
||||
plan_repo_instance.get.return_value = mock_plan_with_cover_enabled
|
||||
MockPlanRepo.return_value = plan_repo_instance
|
||||
|
||||
with patch("video_processing.cover_generator.generate_cover_from_plan") as mock_gen_cover:
|
||||
# 模拟封面生成抛出异常
|
||||
mock_gen_cover.side_effect = Exception("FFmpeg failed")
|
||||
|
||||
task = MagicMock()
|
||||
result = _compose_with_unified_engine(
|
||||
task, mock_job_service, mock_job_service.get_job(), "plan_456", mock_db
|
||||
)
|
||||
|
||||
# 验证视频合成仍然成功
|
||||
assert result["status"] == "completed"
|
||||
assert "result" in result
|
||||
assert result["result"]["output_url"] == mock_render_result.output_url
|
||||
|
||||
# 验证结果中 cover_url 为 None
|
||||
assert result["result"]["cover_url"] is None
|
||||
@@ -1,538 +0,0 @@
|
||||
"""CoverGenerator 纯逻辑单测 — 时间钳制 + 智能选帧算法.
|
||||
|
||||
通过 mock run_ffmpeg 和 probe_video_info 验证纯逻辑部分，
|
||||
不实际执行 FFmpeg，确保测试轻量快速。
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from pathlib import Path
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
import pytest
|
||||
from video_processing.cover_generator import (
|
||||
DEFAULT_COVER_HEIGHT,
|
||||
DEFAULT_COVER_QUALITY,
|
||||
DEFAULT_COVER_TIME,
|
||||
DEFAULT_COVER_WIDTH,
|
||||
SMART_COVER_FRAME_COUNT,
|
||||
CoverGenerator,
|
||||
)
|
||||
|
||||
|
||||
class TestCoverGeneratorConstants:
|
||||
"""常量默认值测试."""
|
||||
|
||||
def test_default_cover_time(self):
|
||||
"""默认抽帧时间为 1.0 秒."""
|
||||
assert DEFAULT_COVER_TIME == 1.0
|
||||
|
||||
def test_default_dimensions(self):
|
||||
"""默认封面尺寸 1080x1920 (竖屏)."""
|
||||
assert DEFAULT_COVER_WIDTH == 1080
|
||||
assert DEFAULT_COVER_HEIGHT == 1920
|
||||
|
||||
def test_default_quality(self):
|
||||
"""默认质量为 5 (JPEG q:v, 越小越好)."""
|
||||
assert DEFAULT_COVER_QUALITY == 5
|
||||
|
||||
def test_smart_cover_frame_count(self):
|
||||
"""智能封面默认抽 3 帧."""
|
||||
assert SMART_COVER_FRAME_COUNT == 3
|
||||
|
||||
|
||||
class TestExtractFrameCommand:
|
||||
"""extract_frame 命令构建测试."""
|
||||
|
||||
def _probe_video_info_mock(self, duration=10.0):
|
||||
"""创建 probe_video_info 的 mock."""
|
||||
return {"duration": duration, "width": 1920, "height": 1080, "fps": 25.0}
|
||||
|
||||
def test_default_params_command(self, tmp_path):
|
||||
"""默认参数下 FFmpeg 命令正确."""
|
||||
video_file = tmp_path / "test.mp4"
|
||||
video_file.write_bytes(b"fake")
|
||||
output_file = tmp_path / "cover.jpg"
|
||||
|
||||
with (
|
||||
patch(
|
||||
"video_processing.cover_generator.probe_video_info",
|
||||
return_value=self._probe_video_info_mock(),
|
||||
),
|
||||
patch("video_processing.cover_generator.run_ffmpeg") as mock_run,
|
||||
):
|
||||
# 让 output_path 在 run_ffmpeg 后存在
|
||||
def fake_run(cmd):
|
||||
output_file.write_bytes(b"fake jpg")
|
||||
|
||||
mock_run.side_effect = fake_run
|
||||
result = CoverGenerator.extract_frame(str(video_file), str(output_file))
|
||||
|
||||
assert result == Path(output_file)
|
||||
mock_run.assert_called_once()
|
||||
cmd = mock_run.call_args[0][0]
|
||||
|
||||
# 基本结构
|
||||
assert cmd[0].endswith("ffmpeg") or "ffmpeg" in cmd[0]
|
||||
assert "-y" in cmd
|
||||
assert "-vframes" in cmd
|
||||
assert cmd[cmd.index("-vframes") + 1] == "1"
|
||||
assert "-f" in cmd
|
||||
assert "mjpeg" in cmd[cmd.index("-f") + 1]
|
||||
|
||||
# 时间点
|
||||
ss_idx = cmd.index("-ss")
|
||||
assert float(cmd[ss_idx + 1]) == pytest.approx(DEFAULT_COVER_TIME, abs=0.001)
|
||||
|
||||
# 输入文件
|
||||
i_idx = cmd.index("-i")
|
||||
assert cmd[i_idx + 1] == str(video_file)
|
||||
|
||||
# 输出文件
|
||||
assert cmd[-1] == str(output_file)
|
||||
|
||||
# scale + crop 滤镜
|
||||
vf_idx = cmd.index("-vf")
|
||||
vf_value = cmd[vf_idx + 1]
|
||||
assert "scale=" in vf_value
|
||||
assert "crop=" in vf_value
|
||||
assert "force_original_aspect_ratio=increase" in vf_value
|
||||
|
||||
def test_custom_time(self, tmp_path):
|
||||
"""自定义抽帧时间点."""
|
||||
video_file = tmp_path / "test.mp4"
|
||||
video_file.write_bytes(b"fake")
|
||||
output_file = tmp_path / "cover.jpg"
|
||||
|
||||
with (
|
||||
patch(
|
||||
"video_processing.cover_generator.probe_video_info",
|
||||
return_value=self._probe_video_info_mock(duration=30.0),
|
||||
),
|
||||
patch("video_processing.cover_generator.run_ffmpeg") as mock_run,
|
||||
):
|
||||
|
||||
def fake_run(cmd):
|
||||
output_file.write_bytes(b"fake jpg")
|
||||
|
||||
mock_run.side_effect = fake_run
|
||||
CoverGenerator.extract_frame(str(video_file), str(output_file), time_sec=5.5)
|
||||
|
||||
cmd = mock_run.call_args[0][0]
|
||||
ss_idx = cmd.index("-ss")
|
||||
assert float(cmd[ss_idx + 1]) == pytest.approx(5.5, abs=0.001)
|
||||
|
||||
def test_custom_dimensions(self, tmp_path):
|
||||
"""自定义输出尺寸."""
|
||||
video_file = tmp_path / "test.mp4"
|
||||
video_file.write_bytes(b"fake")
|
||||
output_file = tmp_path / "cover.jpg"
|
||||
|
||||
with (
|
||||
patch(
|
||||
"video_processing.cover_generator.probe_video_info",
|
||||
return_value=self._probe_video_info_mock(),
|
||||
),
|
||||
patch("video_processing.cover_generator.run_ffmpeg") as mock_run,
|
||||
):
|
||||
|
||||
def fake_run(cmd):
|
||||
output_file.write_bytes(b"fake jpg")
|
||||
|
||||
mock_run.side_effect = fake_run
|
||||
CoverGenerator.extract_frame(str(video_file), str(output_file), width=1920, height=1080)
|
||||
|
||||
cmd = mock_run.call_args[0][0]
|
||||
vf_idx = cmd.index("-vf")
|
||||
vf_value = cmd[vf_idx + 1]
|
||||
assert "scale=1920:1080:" in vf_value
|
||||
assert "crop=1920:1080" in vf_value
|
||||
|
||||
def test_custom_quality(self, tmp_path):
|
||||
"""自定义 JPEG 质量."""
|
||||
video_file = tmp_path / "test.mp4"
|
||||
video_file.write_bytes(b"fake")
|
||||
output_file = tmp_path / "cover.jpg"
|
||||
|
||||
with (
|
||||
patch(
|
||||
"video_processing.cover_generator.probe_video_info",
|
||||
return_value=self._probe_video_info_mock(),
|
||||
),
|
||||
patch("video_processing.cover_generator.run_ffmpeg") as mock_run,
|
||||
):
|
||||
|
||||
def fake_run(cmd):
|
||||
output_file.write_bytes(b"fake jpg")
|
||||
|
||||
mock_run.side_effect = fake_run
|
||||
CoverGenerator.extract_frame(str(video_file), str(output_file), quality=2)
|
||||
|
||||
cmd = mock_run.call_args[0][0]
|
||||
q_idx = cmd.index("-q:v")
|
||||
assert cmd[q_idx + 1] == "2"
|
||||
|
||||
def test_time_exceeds_duration_clamps_to_midpoint(self, tmp_path):
|
||||
"""抽帧时间超过视频时长时，钳制到中间帧."""
|
||||
video_file = tmp_path / "test.mp4"
|
||||
video_file.write_bytes(b"fake")
|
||||
output_file = tmp_path / "cover.jpg"
|
||||
|
||||
with (
|
||||
patch(
|
||||
"video_processing.cover_generator.probe_video_info",
|
||||
return_value=self._probe_video_info_mock(duration=5.0),
|
||||
),
|
||||
patch("video_processing.cover_generator.run_ffmpeg") as mock_run,
|
||||
):
|
||||
|
||||
def fake_run(cmd):
|
||||
output_file.write_bytes(b"fake jpg")
|
||||
|
||||
mock_run.side_effect = fake_run
|
||||
CoverGenerator.extract_frame(str(video_file), str(output_file), time_sec=10.0)
|
||||
|
||||
cmd = mock_run.call_args[0][0]
|
||||
ss_idx = cmd.index("-ss")
|
||||
# 钳制到 duration/2 = 2.5
|
||||
assert float(cmd[ss_idx + 1]) == pytest.approx(2.5, abs=0.001)
|
||||
|
||||
def test_negative_time_clamps_to_zero(self, tmp_path):
|
||||
"""负时间钳制到 0."""
|
||||
video_file = tmp_path / "test.mp4"
|
||||
video_file.write_bytes(b"fake")
|
||||
output_file = tmp_path / "cover.jpg"
|
||||
|
||||
with (
|
||||
patch(
|
||||
"video_processing.cover_generator.probe_video_info",
|
||||
return_value=self._probe_video_info_mock(duration=10.0),
|
||||
),
|
||||
patch("video_processing.cover_generator.run_ffmpeg") as mock_run,
|
||||
):
|
||||
|
||||
def fake_run(cmd):
|
||||
output_file.write_bytes(b"fake jpg")
|
||||
|
||||
mock_run.side_effect = fake_run
|
||||
CoverGenerator.extract_frame(str(video_file), str(output_file), time_sec=-2.0)
|
||||
|
||||
cmd = mock_run.call_args[0][0]
|
||||
ss_idx = cmd.index("-ss")
|
||||
assert float(cmd[ss_idx + 1]) == pytest.approx(0.0, abs=0.001)
|
||||
|
||||
def test_time_equals_duration_clamps_to_midpoint(self, tmp_path):
|
||||
"""时间点等于时长时钳制到中间帧."""
|
||||
video_file = tmp_path / "test.mp4"
|
||||
video_file.write_bytes(b"fake")
|
||||
output_file = tmp_path / "cover.jpg"
|
||||
|
||||
with (
|
||||
patch(
|
||||
"video_processing.cover_generator.probe_video_info",
|
||||
return_value=self._probe_video_info_mock(duration=10.0),
|
||||
),
|
||||
patch("video_processing.cover_generator.run_ffmpeg") as mock_run,
|
||||
):
|
||||
|
||||
def fake_run(cmd):
|
||||
output_file.write_bytes(b"fake jpg")
|
||||
|
||||
mock_run.side_effect = fake_run
|
||||
CoverGenerator.extract_frame(str(video_file), str(output_file), time_sec=10.0)
|
||||
|
||||
cmd = mock_run.call_args[0][0]
|
||||
ss_idx = cmd.index("-ss")
|
||||
assert float(cmd[ss_idx + 1]) == pytest.approx(5.0, abs=0.001)
|
||||
|
||||
def test_zero_duration_video(self, tmp_path):
|
||||
"""视频时长为 0 时的行为（不钳制，用原始时间）."""
|
||||
video_file = tmp_path / "test.mp4"
|
||||
video_file.write_bytes(b"fake")
|
||||
output_file = tmp_path / "cover.jpg"
|
||||
|
||||
with (
|
||||
patch(
|
||||
"video_processing.cover_generator.probe_video_info",
|
||||
return_value=self._probe_video_info_mock(duration=0.0),
|
||||
),
|
||||
patch("video_processing.cover_generator.run_ffmpeg") as mock_run,
|
||||
):
|
||||
|
||||
def fake_run(cmd):
|
||||
output_file.write_bytes(b"fake jpg")
|
||||
|
||||
mock_run.side_effect = fake_run
|
||||
CoverGenerator.extract_frame(str(video_file), str(output_file), time_sec=0.5)
|
||||
|
||||
cmd = mock_run.call_args[0][0]
|
||||
ss_idx = cmd.index("-ss")
|
||||
assert float(cmd[ss_idx + 1]) == pytest.approx(0.5, abs=0.001)
|
||||
|
||||
def test_video_not_found_raises(self, tmp_path):
|
||||
"""视频文件不存在时抛出 FileNotFoundError."""
|
||||
output_file = tmp_path / "cover.jpg"
|
||||
|
||||
with pytest.raises(FileNotFoundError):
|
||||
CoverGenerator.extract_frame(str(tmp_path / "nonexistent.mp4"), str(output_file))
|
||||
|
||||
def test_output_creates_parent_dir(self, tmp_path):
|
||||
"""输出目录不存在时自动创建."""
|
||||
video_file = tmp_path / "test.mp4"
|
||||
video_file.write_bytes(b"fake")
|
||||
out_dir = tmp_path / "deep" / "nested"
|
||||
output_file = out_dir / "cover.jpg"
|
||||
|
||||
with (
|
||||
patch(
|
||||
"video_processing.cover_generator.probe_video_info",
|
||||
return_value=self._probe_video_info_mock(),
|
||||
),
|
||||
patch("video_processing.cover_generator.run_ffmpeg") as mock_run,
|
||||
):
|
||||
|
||||
def fake_run(cmd):
|
||||
output_file.write_bytes(b"fake jpg")
|
||||
|
||||
mock_run.side_effect = fake_run
|
||||
CoverGenerator.extract_frame(str(video_file), str(output_file))
|
||||
|
||||
assert out_dir.exists()
|
||||
assert out_dir.is_dir()
|
||||
|
||||
def test_ffmpeg_failure_propagates(self, tmp_path):
|
||||
"""FFmpeg 失败时异常向上传递."""
|
||||
video_file = tmp_path / "test.mp4"
|
||||
video_file.write_bytes(b"fake")
|
||||
output_file = tmp_path / "cover.jpg"
|
||||
|
||||
with (
|
||||
patch(
|
||||
"video_processing.cover_generator.probe_video_info",
|
||||
return_value=self._probe_video_info_mock(),
|
||||
),
|
||||
patch(
|
||||
"video_processing.cover_generator.run_ffmpeg",
|
||||
side_effect=RuntimeError("FFmpeg error"),
|
||||
),
|
||||
):
|
||||
with pytest.raises(RuntimeError, match="FFmpeg error"):
|
||||
CoverGenerator.extract_frame(str(video_file), str(output_file))
|
||||
|
||||
|
||||
class TestSmartCoverTimePoints:
|
||||
"""智能封面时间点计算测试."""
|
||||
|
||||
def test_single_frame_falls_back_to_default(self, tmp_path):
|
||||
"""只有 1 帧时退化为普通抽帧（取 DEFAULT_COVER_TIME 和 midpoint 中较小值）."""
|
||||
video_file = tmp_path / "test.mp4"
|
||||
video_file.write_bytes(b"fake")
|
||||
output_file = tmp_path / "cover.jpg"
|
||||
|
||||
with (
|
||||
patch(
|
||||
"video_processing.cover_generator.probe_video_info",
|
||||
return_value={"duration": 20.0},
|
||||
),
|
||||
patch("video_processing.cover_generator.run_ffmpeg") as mock_run,
|
||||
):
|
||||
|
||||
def fake_run(cmd):
|
||||
output_file.write_bytes(b"fake jpg")
|
||||
|
||||
mock_run.side_effect = fake_run
|
||||
# frame_count=1 时退化为普通抽帧
|
||||
CoverGenerator.extract_smart_cover(str(video_file), str(output_file), frame_count=1)
|
||||
|
||||
# 只调用一次（退化路径）
|
||||
assert mock_run.call_count == 1
|
||||
cmd = mock_run.call_args[0][0]
|
||||
ss_idx = cmd.index("-ss")
|
||||
# min(DEFAULT_COVER_TIME=1.0, duration/2=10.0) = 1.0
|
||||
assert float(cmd[ss_idx + 1]) == pytest.approx(1.0, abs=0.001)
|
||||
|
||||
def test_zero_duration_falls_back(self, tmp_path):
|
||||
"""视频时长为 0 时退化为普通抽帧."""
|
||||
video_file = tmp_path / "test.mp4"
|
||||
video_file.write_bytes(b"fake")
|
||||
output_file = tmp_path / "cover.jpg"
|
||||
|
||||
with (
|
||||
patch(
|
||||
"video_processing.cover_generator.probe_video_info",
|
||||
return_value={"duration": 0.0},
|
||||
),
|
||||
patch("video_processing.cover_generator.run_ffmpeg") as mock_run,
|
||||
):
|
||||
|
||||
def fake_run(cmd):
|
||||
output_file.write_bytes(b"fake jpg")
|
||||
|
||||
mock_run.side_effect = fake_run
|
||||
CoverGenerator.extract_smart_cover(str(video_file), str(output_file))
|
||||
|
||||
# 只调用一次（退化路径）
|
||||
assert mock_run.call_count == 1
|
||||
|
||||
def test_three_frames_uniform_distribution(self, tmp_path):
|
||||
"""3 帧均匀分布在 5%~95% 区间."""
|
||||
video_file = tmp_path / "test.mp4"
|
||||
video_file.write_bytes(b"fake")
|
||||
output_file = tmp_path / "cover.jpg"
|
||||
|
||||
call_times = []
|
||||
|
||||
with (
|
||||
patch(
|
||||
"video_processing.cover_generator.probe_video_info",
|
||||
return_value={"duration": 100.0},
|
||||
),
|
||||
patch("video_processing.cover_generator.run_ffmpeg") as mock_run,
|
||||
):
|
||||
|
||||
def fake_run(cmd):
|
||||
# 记录抽帧时间
|
||||
ss_idx = cmd.index("-ss")
|
||||
call_times.append(float(cmd[ss_idx + 1]))
|
||||
# 在输出路径写文件
|
||||
output_arg = cmd[-1]
|
||||
Path(output_arg).parent.mkdir(parents=True, exist_ok=True)
|
||||
# 不同文件大小，让第三帧"最清晰"
|
||||
idx = len(call_times) - 1
|
||||
size = 1000 * (idx + 1) # 递增的文件大小
|
||||
Path(output_arg).write_bytes(b"x" * size)
|
||||
|
||||
mock_run.side_effect = fake_run
|
||||
CoverGenerator.extract_smart_cover(str(video_file), str(output_file))
|
||||
|
||||
# 3 帧：5%、50%、95%
|
||||
assert len(call_times) == 3
|
||||
assert call_times[0] == pytest.approx(5.0, abs=0.1) # 5%
|
||||
assert call_times[1] == pytest.approx(50.0, abs=0.1) # 50%
|
||||
assert call_times[2] == pytest.approx(95.0, abs=0.1) # 95%
|
||||
|
||||
def test_five_frames_distribution(self, tmp_path):
|
||||
"""5 帧均匀分布."""
|
||||
video_file = tmp_path / "test.mp4"
|
||||
video_file.write_bytes(b"fake")
|
||||
output_file = tmp_path / "cover.jpg"
|
||||
|
||||
call_times = []
|
||||
|
||||
with (
|
||||
patch(
|
||||
"video_processing.cover_generator.probe_video_info",
|
||||
return_value={"duration": 100.0},
|
||||
),
|
||||
patch("video_processing.cover_generator.run_ffmpeg") as mock_run,
|
||||
):
|
||||
|
||||
def fake_run(cmd):
|
||||
ss_idx = cmd.index("-ss")
|
||||
call_times.append(float(cmd[ss_idx + 1]))
|
||||
output_arg = cmd[-1]
|
||||
Path(output_arg).parent.mkdir(parents=True, exist_ok=True)
|
||||
idx = len(call_times) - 1
|
||||
Path(output_arg).write_bytes(b"x" * (1000 * (idx + 1)))
|
||||
|
||||
mock_run.side_effect = fake_run
|
||||
CoverGenerator.extract_smart_cover(str(video_file), str(output_file), frame_count=5)
|
||||
|
||||
assert len(call_times) == 5
|
||||
# step = (95-5) / (5-1) = 22.5
|
||||
# times: 5, 27.5, 50, 72.5, 95
|
||||
assert call_times[0] == pytest.approx(5.0, abs=0.1)
|
||||
assert call_times[1] == pytest.approx(27.5, abs=0.1)
|
||||
assert call_times[2] == pytest.approx(50.0, abs=0.1)
|
||||
assert call_times[3] == pytest.approx(72.5, abs=0.1)
|
||||
assert call_times[4] == pytest.approx(95.0, abs=0.1)
|
||||
|
||||
def test_selects_largest_file_as_best(self, tmp_path):
|
||||
"""选择文件最大的帧作为最佳封面（清晰度近似）."""
|
||||
video_file = tmp_path / "test.mp4"
|
||||
video_file.write_bytes(b"fake")
|
||||
output_file = tmp_path / "cover.jpg"
|
||||
|
||||
sizes = [5000, 15000, 8000] # 第二帧最大
|
||||
|
||||
with (
|
||||
patch(
|
||||
"video_processing.cover_generator.probe_video_info",
|
||||
return_value={"duration": 100.0},
|
||||
),
|
||||
patch("video_processing.cover_generator.run_ffmpeg") as mock_run,
|
||||
):
|
||||
call_idx = [0]
|
||||
|
||||
def fake_run(cmd):
|
||||
output_arg = cmd[-1]
|
||||
Path(output_arg).parent.mkdir(parents=True, exist_ok=True)
|
||||
idx = call_idx[0]
|
||||
Path(output_arg).write_bytes(b"x" * sizes[idx])
|
||||
call_idx[0] += 1
|
||||
|
||||
mock_run.side_effect = fake_run
|
||||
result = CoverGenerator.extract_smart_cover(str(video_file), str(output_file))
|
||||
|
||||
# 第二帧（索引1）应该是最佳
|
||||
assert result == output_file
|
||||
# 输出文件大小应等于第二帧大小
|
||||
assert output_file.stat().st_size == 15000
|
||||
|
||||
|
||||
class TestProcessCustomCover:
|
||||
"""自定义封面处理测试."""
|
||||
|
||||
def test_custom_cover_resize_command(self, tmp_path):
|
||||
"""自定义封面调整尺寸命令正确."""
|
||||
input_file = tmp_path / "upload.jpg"
|
||||
input_file.write_bytes(b"fake")
|
||||
output_file = tmp_path / "cover.jpg"
|
||||
|
||||
with patch("video_processing.cover_generator.run_ffmpeg") as mock_run:
|
||||
|
||||
def fake_run(cmd):
|
||||
output_file.write_bytes(b"fake jpg")
|
||||
|
||||
mock_run.side_effect = fake_run
|
||||
CoverGenerator.process_custom_cover(str(input_file), str(output_file))
|
||||
|
||||
mock_run.assert_called_once()
|
||||
cmd = mock_run.call_args[0][0]
|
||||
|
||||
assert "-i" in cmd
|
||||
assert cmd[cmd.index("-i") + 1] == str(input_file)
|
||||
assert cmd[-1] == str(output_file)
|
||||
|
||||
# scale + crop
|
||||
vf_idx = cmd.index("-vf")
|
||||
vf_value = cmd[vf_idx + 1]
|
||||
assert "scale=" in vf_value
|
||||
assert "crop=" in vf_value
|
||||
|
||||
def test_custom_cover_not_found_raises(self, tmp_path):
|
||||
"""自定义封面文件不存在时抛出 FileNotFoundError."""
|
||||
output_file = tmp_path / "cover.jpg"
|
||||
|
||||
with pytest.raises(FileNotFoundError):
|
||||
CoverGenerator.process_custom_cover(str(tmp_path / "nonexistent.jpg"), str(output_file))
|
||||
|
||||
def test_custom_cover_custom_dimensions(self, tmp_path):
|
||||
"""自定义封面自定义输出尺寸."""
|
||||
input_file = tmp_path / "upload.jpg"
|
||||
input_file.write_bytes(b"fake")
|
||||
output_file = tmp_path / "cover.jpg"
|
||||
|
||||
with patch("video_processing.cover_generator.run_ffmpeg") as mock_run:
|
||||
|
||||
def fake_run(cmd):
|
||||
output_file.write_bytes(b"fake jpg")
|
||||
|
||||
mock_run.side_effect = fake_run
|
||||
CoverGenerator.process_custom_cover(str(input_file), str(output_file), width=800, height=600)
|
||||
|
||||
cmd = mock_run.call_args[0][0]
|
||||
vf_idx = cmd.index("-vf")
|
||||
vf_value = cmd[vf_idx + 1]
|
||||
assert "scale=800:600:" in vf_value
|
||||
assert "crop=800:600" in vf_value
|
||||
@@ -1,860 +0,0 @@
|
||||
"""封面生成 + 视频倒放 + 贴纸叠加 单元测试.
|
||||
|
||||
覆盖三个新渲染能力的核心场景和降级逻辑。
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from dataclasses import dataclass, field
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
import pytest
|
||||
from video_processing.cover_generator import (
|
||||
DEFAULT_COVER_HEIGHT,
|
||||
DEFAULT_COVER_WIDTH,
|
||||
CoverGenerator,
|
||||
generate_cover_from_plan,
|
||||
)
|
||||
from video_processing.reverse_engine import ReverseConfig, ReverseEngine
|
||||
from video_processing.sticker_engine import (
|
||||
POSITION_PRESETS,
|
||||
STICKER_CATEGORIES,
|
||||
ImageStickerConfig,
|
||||
StickerEngine,
|
||||
TextStickerConfig,
|
||||
get_sticker_categories,
|
||||
parse_stickers_from_config,
|
||||
)
|
||||
from video_processing.unified_render_service import (
|
||||
ResolvedClip,
|
||||
UnifiedRenderService,
|
||||
)
|
||||
|
||||
# ── Fixtures ──────────────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
@dataclass
|
||||
class FakePlan:
|
||||
"""模拟 EditPlan."""
|
||||
|
||||
id: str = "plan_001"
|
||||
name: str = "测试计划"
|
||||
config: dict[str, Any] = field(default_factory=dict)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def sample_video(tmp_path):
|
||||
"""创建一个测试视频文件（空文件，仅用于路径测试）."""
|
||||
video_path = tmp_path / "test_video.mp4"
|
||||
video_path.write_bytes(b"fake video data")
|
||||
return video_path
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def sample_image(tmp_path):
|
||||
"""创建一个测试图片文件."""
|
||||
img_path = tmp_path / "sticker.png"
|
||||
img_path.write_bytes(b"fake png data")
|
||||
return img_path
|
||||
|
||||
|
||||
# ═══════════════════════════════════════════════════════════════════════════════
|
||||
# 一、视频倒放引擎测试
|
||||
# ═══════════════════════════════════════════════════════════════════════════════
|
||||
|
||||
|
||||
class TestReverseConfig:
|
||||
"""ReverseConfig 配置解析测试."""
|
||||
|
||||
def test_default_disabled(self):
|
||||
"""默认配置为关闭."""
|
||||
config = ReverseConfig.from_dict(None)
|
||||
assert config.enabled is False
|
||||
assert config.reverse_video is True
|
||||
assert config.reverse_audio is True
|
||||
|
||||
def test_empty_dict(self):
|
||||
"""空字典视为关闭."""
|
||||
config = ReverseConfig.from_dict({})
|
||||
assert config.enabled is False
|
||||
|
||||
def test_enabled(self):
|
||||
"""启用倒放."""
|
||||
config = ReverseConfig.from_dict({"enabled": True})
|
||||
assert config.enabled is True
|
||||
assert config.reverse_video is True
|
||||
assert config.reverse_audio is True
|
||||
|
||||
def test_video_only(self):
|
||||
"""只倒放视频."""
|
||||
config = ReverseConfig.from_dict(
|
||||
{
|
||||
"enabled": True,
|
||||
"reverse_video": True,
|
||||
"reverse_audio": False,
|
||||
}
|
||||
)
|
||||
assert config.enabled is True
|
||||
assert config.reverse_video is True
|
||||
assert config.reverse_audio is False
|
||||
|
||||
def test_audio_only(self):
|
||||
"""只倒放音频."""
|
||||
config = ReverseConfig.from_dict(
|
||||
{
|
||||
"enabled": True,
|
||||
"reverse_video": False,
|
||||
"reverse_audio": True,
|
||||
}
|
||||
)
|
||||
assert config.reverse_video is False
|
||||
assert config.reverse_audio is True
|
||||
|
||||
def test_invalid_config_fallback(self):
|
||||
"""无效配置降级为默认."""
|
||||
config = ReverseConfig.from_dict("invalid") # type: ignore
|
||||
assert config.enabled is False
|
||||
|
||||
def test_none_config(self):
|
||||
"""None 配置."""
|
||||
config = ReverseConfig.from_dict(None)
|
||||
assert config.enabled is False
|
||||
|
||||
|
||||
class TestReverseEngine:
|
||||
"""ReverseEngine 滤镜生成测试."""
|
||||
|
||||
def test_video_reverse_filter(self):
|
||||
"""视频倒放滤镜生成."""
|
||||
config = ReverseConfig(enabled=True, reverse_video=True)
|
||||
f = ReverseEngine.build_video_filter(config, duration=10.0)
|
||||
assert f == "reverse"
|
||||
|
||||
def test_video_disabled(self):
|
||||
"""视频倒放关闭时返回空."""
|
||||
config = ReverseConfig(enabled=False)
|
||||
f = ReverseEngine.build_video_filter(config, duration=10.0)
|
||||
assert f == ""
|
||||
|
||||
def test_video_disabled_flag(self):
|
||||
"""启用但 reverse_video=False."""
|
||||
config = ReverseConfig(enabled=True, reverse_video=False)
|
||||
f = ReverseEngine.build_video_filter(config, duration=10.0)
|
||||
assert f == ""
|
||||
|
||||
def test_audio_reverse_filter(self):
|
||||
"""音频倒放滤镜生成."""
|
||||
config = ReverseConfig(enabled=True, reverse_audio=True)
|
||||
f = ReverseEngine.build_audio_filter(config, duration=10.0)
|
||||
assert f == "areverse"
|
||||
|
||||
def test_audio_disabled(self):
|
||||
"""音频倒放关闭."""
|
||||
config = ReverseConfig(enabled=False)
|
||||
f = ReverseEngine.build_audio_filter(config, duration=10.0)
|
||||
assert f == ""
|
||||
|
||||
def test_long_video_safety_limit(self):
|
||||
"""超长视频安全限制：跳过倒放."""
|
||||
config = ReverseConfig(enabled=True)
|
||||
f = ReverseEngine.build_video_filter(config, duration=200.0)
|
||||
assert f == "" # 超过 MAX_SAFE_DURATION
|
||||
|
||||
def test_long_audio_safety_limit(self):
|
||||
"""超长音频安全限制."""
|
||||
config = ReverseConfig(enabled=True)
|
||||
f = ReverseEngine.build_audio_filter(config, duration=200.0)
|
||||
assert f == ""
|
||||
|
||||
def test_duration_zero(self):
|
||||
"""时长为0时正常返回."""
|
||||
config = ReverseConfig(enabled=True)
|
||||
f = ReverseEngine.build_video_filter(config, duration=0.0)
|
||||
assert f == "reverse"
|
||||
|
||||
|
||||
# ═══════════════════════════════════════════════════════════════════════════════
|
||||
# 二、贴纸引擎测试
|
||||
# ═══════════════════════════════════════════════════════════════════════════════
|
||||
|
||||
|
||||
class TestStickerPosition:
|
||||
"""贴纸位置计算测试."""
|
||||
|
||||
def test_presets_exist(self):
|
||||
"""9宫格预设存在."""
|
||||
assert "top_left" in POSITION_PRESETS
|
||||
assert "center" in POSITION_PRESETS
|
||||
assert "bottom_right" in POSITION_PRESETS
|
||||
assert len(POSITION_PRESETS) == 9
|
||||
|
||||
def test_resolve_position_center(self):
|
||||
"""居中位置计算."""
|
||||
sticker = ImageStickerConfig(position="center")
|
||||
x, y = StickerEngine._resolve_position(sticker, 1000, 1000, 200, 200)
|
||||
assert abs(x - 400) < 1 # (1000-200)/2 = 400
|
||||
assert abs(y - 400) < 1
|
||||
|
||||
def test_resolve_position_top_left(self):
|
||||
"""左上角位置."""
|
||||
sticker = ImageStickerConfig(position="top_left")
|
||||
x, y = StickerEngine._resolve_position(sticker, 1000, 1000, 100, 100)
|
||||
assert x == 0 # 0.05*1000 - 50 = 0 (clamped)
|
||||
assert y == 0
|
||||
|
||||
def test_custom_position_percent(self):
|
||||
"""自定义百分比位置."""
|
||||
sticker = ImageStickerConfig(
|
||||
position="center",
|
||||
x=30.0,
|
||||
y=70.0,
|
||||
x_unit="percent",
|
||||
y_unit="percent",
|
||||
)
|
||||
x, y = StickerEngine._resolve_position(sticker, 1000, 1000, 100, 100)
|
||||
assert abs(x - 250) < 1 # 300 - 50 = 250
|
||||
assert abs(y - 650) < 1 # 700 - 50 = 650
|
||||
|
||||
def test_custom_position_pixel(self):
|
||||
"""自定义像素位置."""
|
||||
sticker = ImageStickerConfig(
|
||||
position="center",
|
||||
x=100.0,
|
||||
y=200.0,
|
||||
x_unit="pixel",
|
||||
y_unit="pixel",
|
||||
)
|
||||
x, y = StickerEngine._resolve_position(sticker, 1000, 1000, 50, 50)
|
||||
assert abs(x - 75) < 1 # 100 - 25 = 75
|
||||
assert abs(y - 175) < 1 # 200 - 25 = 175
|
||||
|
||||
def test_position_clamped(self):
|
||||
"""位置钳制在画布内."""
|
||||
sticker = ImageStickerConfig(
|
||||
position="center",
|
||||
x=-10.0,
|
||||
y=-10.0,
|
||||
x_unit="pixel",
|
||||
y_unit="pixel",
|
||||
)
|
||||
x, y = StickerEngine._resolve_position(sticker, 1000, 1000, 50, 50)
|
||||
assert x >= 0
|
||||
assert y >= 0
|
||||
|
||||
|
||||
class TestTextSticker:
|
||||
"""文字贴纸测试."""
|
||||
|
||||
def test_drawtext_filter_basic(self):
|
||||
"""基础文字贴纸滤镜生成."""
|
||||
sticker = TextStickerConfig(
|
||||
enabled=True,
|
||||
text="Hello World",
|
||||
font_size=36,
|
||||
font_color="#FFFFFF",
|
||||
position="center",
|
||||
)
|
||||
f = StickerEngine._build_drawtext_filter(sticker, "[in]", "[out]", 1080, 1920)
|
||||
assert "drawtext" in f
|
||||
assert "Hello World" in f
|
||||
assert "fontsize=36" in f
|
||||
assert "[in]" in f
|
||||
assert "[out]" in f
|
||||
|
||||
def test_drawtext_with_stroke(self):
|
||||
"""带描边的文字贴纸."""
|
||||
sticker = TextStickerConfig(
|
||||
enabled=True,
|
||||
text="Test",
|
||||
stroke_width=3,
|
||||
stroke_color="#FF0000",
|
||||
)
|
||||
f = StickerEngine._build_drawtext_filter(sticker, "[in]", "[out]", 1080, 1920)
|
||||
assert "borderw=3" in f
|
||||
assert "bordercolor=#FF0000" in f
|
||||
|
||||
def test_drawtext_with_shadow(self):
|
||||
"""带阴影的文字贴纸."""
|
||||
sticker = TextStickerConfig(
|
||||
enabled=True,
|
||||
text="Shadow",
|
||||
shadow_x=4,
|
||||
shadow_y=4,
|
||||
shadow_alpha=0.5,
|
||||
)
|
||||
f = StickerEngine._build_drawtext_filter(sticker, "[in]", "[out]", 1080, 1920)
|
||||
assert "shadowx=4" in f
|
||||
assert "shadowy=4" in f
|
||||
|
||||
def test_drawtext_time_range(self):
|
||||
"""带时间范围的文字贴纸."""
|
||||
sticker = TextStickerConfig(
|
||||
enabled=True,
|
||||
text="Timed",
|
||||
start_time=2.0,
|
||||
duration=3.0,
|
||||
)
|
||||
f = StickerEngine._build_drawtext_filter(sticker, "[in]", "[out]", 1080, 1920)
|
||||
assert "enable='between(t,2.0,5.0)'" in f
|
||||
|
||||
def test_drawtext_empty_text(self):
|
||||
"""空文字直通."""
|
||||
sticker = TextStickerConfig(enabled=True, text="")
|
||||
f = StickerEngine._build_drawtext_filter(sticker, "[in]", "[out]", 1080, 1920)
|
||||
assert "[in]copy[out]" in f
|
||||
|
||||
def test_drawtext_with_fade(self):
|
||||
"""带淡入淡出的文字贴纸."""
|
||||
sticker = TextStickerConfig(
|
||||
enabled=True,
|
||||
text="Fade",
|
||||
start_time=1.0,
|
||||
duration=5.0,
|
||||
fade_in=0.5,
|
||||
fade_out=0.5,
|
||||
)
|
||||
f = StickerEngine._build_drawtext_filter(sticker, "[in]", "[out]", 1080, 1920)
|
||||
assert "alpha=" in f
|
||||
|
||||
|
||||
class TestImageSticker:
|
||||
"""图片贴纸测试."""
|
||||
|
||||
def test_image_sticker_overlay(self, sample_image):
|
||||
"""图片贴纸 overlay 滤镜生成."""
|
||||
result = StickerEngine.build_sticker_chain(
|
||||
stickers=[
|
||||
{
|
||||
"type": "image",
|
||||
"image_path": str(sample_image),
|
||||
"position": "top_right",
|
||||
"scale": 0.5,
|
||||
"opacity": 0.8,
|
||||
"z_index": 10,
|
||||
}
|
||||
],
|
||||
input_label="[base]",
|
||||
output_label="[final]",
|
||||
canvas_w=1080,
|
||||
canvas_h=1920,
|
||||
)
|
||||
assert result.filter_str != ""
|
||||
assert "overlay" in result.filter_str
|
||||
assert len(result.extra_inputs) == 1
|
||||
assert result.extra_inputs[0] == str(sample_image)
|
||||
|
||||
def test_image_sticker_missing_file(self):
|
||||
"""图片贴纸素材不存在时跳过."""
|
||||
result = StickerEngine.build_sticker_chain(
|
||||
stickers=[
|
||||
{
|
||||
"type": "image",
|
||||
"image_path": "/nonexistent/image.png",
|
||||
"position": "center",
|
||||
}
|
||||
],
|
||||
input_label="[in]",
|
||||
output_label="[out]",
|
||||
canvas_w=1080,
|
||||
canvas_h=1920,
|
||||
)
|
||||
# 素材不存在，跳过，返回直通
|
||||
assert "[in]copy[out]" in result.filter_str
|
||||
assert len(result.extra_inputs) == 0
|
||||
|
||||
def test_mixed_stickers(self, sample_image):
|
||||
"""混合贴纸：图片 + 文字."""
|
||||
result = StickerEngine.build_sticker_chain(
|
||||
stickers=[
|
||||
{
|
||||
"type": "image",
|
||||
"image_path": str(sample_image),
|
||||
"position": "top_left",
|
||||
"z_index": 5,
|
||||
},
|
||||
{
|
||||
"type": "text",
|
||||
"text": "Hello",
|
||||
"position": "bottom_center",
|
||||
"z_index": 10,
|
||||
},
|
||||
],
|
||||
input_label="[in]",
|
||||
output_label="[out]",
|
||||
canvas_w=1080,
|
||||
canvas_h=1920,
|
||||
)
|
||||
assert "overlay" in result.filter_str
|
||||
assert "drawtext" in result.filter_str
|
||||
assert len(result.extra_inputs) == 1
|
||||
|
||||
def test_sticker_z_index_order(self, sample_image):
|
||||
"""贴纸按 z_index 排序."""
|
||||
result = StickerEngine.build_sticker_chain(
|
||||
stickers=[
|
||||
{"type": "text", "text": "Top", "z_index": 20, "position": "center"},
|
||||
{"type": "text", "text": "Bottom", "z_index": 5, "position": "center"},
|
||||
],
|
||||
input_label="[in]",
|
||||
output_label="[out]",
|
||||
canvas_w=1080,
|
||||
canvas_h=1920,
|
||||
)
|
||||
# z_index 小的先叠加，大的后叠加（在上面）
|
||||
assert result.filter_str.count("drawtext") == 2
|
||||
|
||||
def test_empty_stickers(self):
|
||||
"""空贴纸列表."""
|
||||
result = StickerEngine.build_sticker_chain(
|
||||
stickers=[],
|
||||
input_label="[in]",
|
||||
output_label="[out]",
|
||||
canvas_w=1080,
|
||||
canvas_h=1920,
|
||||
)
|
||||
assert "[in]copy[out]" in result.filter_str
|
||||
assert result.extra_inputs == []
|
||||
|
||||
def test_invalid_sticker_skipped(self):
|
||||
"""无效贴纸配置跳过."""
|
||||
result = StickerEngine.build_sticker_chain(
|
||||
stickers=[{"invalid": "data"}],
|
||||
input_label="[in]",
|
||||
output_label="[out]",
|
||||
canvas_w=1080,
|
||||
canvas_h=1920,
|
||||
)
|
||||
# 解析失败，跳过，直通
|
||||
assert "[in]copy[out]" in result.filter_str
|
||||
|
||||
|
||||
class TestStickerHelpers:
|
||||
"""贴纸辅助函数测试."""
|
||||
|
||||
def test_parse_stickers_empty(self):
|
||||
"""空配置解析."""
|
||||
assert parse_stickers_from_config(None) == []
|
||||
assert parse_stickers_from_config({}) == []
|
||||
|
||||
def test_parse_stickers_list(self):
|
||||
"""正常贴纸列表解析."""
|
||||
config = {"stickers": [{"type": "text", "text": "A"}, {"type": "text", "text": "B"}]}
|
||||
result = parse_stickers_from_config(config)
|
||||
assert len(result) == 2
|
||||
|
||||
def test_parse_stickers_not_list(self):
|
||||
"""非列表类型返回空."""
|
||||
config = {"stickers": "not a list"}
|
||||
assert parse_stickers_from_config(config) == []
|
||||
|
||||
def test_get_categories(self):
|
||||
"""贴纸分类列表."""
|
||||
cats = get_sticker_categories()
|
||||
assert len(cats) == len(STICKER_CATEGORIES)
|
||||
assert cats[0][0] == "emoji"
|
||||
|
||||
|
||||
# ═══════════════════════════════════════════════════════════════════════════════
|
||||
# 三、封面生成器测试
|
||||
# ═══════════════════════════════════════════════════════════════════════════════
|
||||
|
||||
|
||||
class TestCoverGenerator:
|
||||
"""CoverGenerator 测试."""
|
||||
|
||||
def test_default_dimensions(self):
|
||||
"""默认封面尺寸."""
|
||||
assert DEFAULT_COVER_WIDTH == 1080
|
||||
assert DEFAULT_COVER_HEIGHT == 1920
|
||||
|
||||
@patch("video_processing.cover_generator.run_ffmpeg")
|
||||
@patch("video_processing.cover_generator.probe_video_info")
|
||||
def test_extract_frame_basic(self, mock_probe, mock_run, sample_video, tmp_path):
|
||||
"""基础抽帧测试."""
|
||||
mock_probe.return_value = {"duration": 30.0}
|
||||
|
||||
# mock run_ffmpeg 实际创建输出文件
|
||||
def fake_run_ffmpeg(cmd):
|
||||
# 找到输出路径并创建文件
|
||||
output_path = Path(cmd[-1])
|
||||
output_path.parent.mkdir(parents=True, exist_ok=True)
|
||||
output_path.write_bytes(b"fake jpeg data")
|
||||
|
||||
mock_run.side_effect = fake_run_ffmpeg
|
||||
|
||||
output = tmp_path / "cover.jpg"
|
||||
result = CoverGenerator.extract_frame(
|
||||
sample_video,
|
||||
output,
|
||||
time_sec=2.0,
|
||||
)
|
||||
|
||||
assert result == output
|
||||
mock_run.assert_called_once()
|
||||
# 检查命令参数
|
||||
cmd = mock_run.call_args[0][0]
|
||||
assert "-ss" in cmd
|
||||
assert "2.000" in cmd
|
||||
assert "-vframes" in cmd
|
||||
assert "1" in cmd
|
||||
|
||||
@patch("video_processing.cover_generator.run_ffmpeg")
|
||||
@patch("video_processing.cover_generator.probe_video_info")
|
||||
def test_extract_frame_time_clamped(self, mock_probe, mock_run, sample_video, tmp_path):
|
||||
"""抽帧时间超过视频长度时钳制."""
|
||||
mock_probe.return_value = {"duration": 10.0}
|
||||
|
||||
def fake_run_ffmpeg(cmd):
|
||||
output_path = Path(cmd[-1])
|
||||
output_path.parent.mkdir(parents=True, exist_ok=True)
|
||||
output_path.write_bytes(b"fake jpeg data")
|
||||
|
||||
mock_run.side_effect = fake_run_ffmpeg
|
||||
|
||||
output = tmp_path / "cover.jpg"
|
||||
CoverGenerator.extract_frame(
|
||||
sample_video,
|
||||
output,
|
||||
time_sec=100.0, # 超过视频时长
|
||||
)
|
||||
|
||||
cmd = mock_run.call_args[0][0]
|
||||
ss_idx = cmd.index("-ss")
|
||||
time_val = float(cmd[ss_idx + 1])
|
||||
# 应该被钳制到中间帧（5秒左右）
|
||||
assert time_val <= 10.0
|
||||
|
||||
@patch("video_processing.cover_generator.run_ffmpeg")
|
||||
@patch("video_processing.cover_generator.probe_video_info")
|
||||
def test_extract_frame_negative_time(self, mock_probe, mock_run, sample_video, tmp_path):
|
||||
"""负时间钳制到0."""
|
||||
mock_probe.return_value = {"duration": 30.0}
|
||||
|
||||
def fake_run_ffmpeg(cmd):
|
||||
output_path = Path(cmd[-1])
|
||||
output_path.parent.mkdir(parents=True, exist_ok=True)
|
||||
output_path.write_bytes(b"fake jpeg data")
|
||||
|
||||
mock_run.side_effect = fake_run_ffmpeg
|
||||
|
||||
output = tmp_path / "cover.jpg"
|
||||
CoverGenerator.extract_frame(
|
||||
sample_video,
|
||||
output,
|
||||
time_sec=-5.0,
|
||||
)
|
||||
|
||||
cmd = mock_run.call_args[0][0]
|
||||
ss_idx = cmd.index("-ss")
|
||||
time_val = float(cmd[ss_idx + 1])
|
||||
assert time_val >= 0
|
||||
|
||||
def test_extract_frame_file_not_found(self, tmp_path):
|
||||
"""视频文件不存在抛异常."""
|
||||
with pytest.raises(FileNotFoundError):
|
||||
CoverGenerator.extract_frame(
|
||||
"/nonexistent/video.mp4",
|
||||
tmp_path / "cover.jpg",
|
||||
)
|
||||
|
||||
@patch("video_processing.cover_generator.CoverGenerator.extract_frame")
|
||||
@patch("video_processing.cover_generator.probe_video_info")
|
||||
def test_smart_cover_3_frames(self, mock_probe, mock_extract, sample_video, tmp_path):
|
||||
"""智能封面抽取3帧选最佳."""
|
||||
mock_probe.return_value = {"duration": 30.0}
|
||||
|
||||
# 创建三个大小不同的临时文件（模拟清晰度不同）
|
||||
def create_frame(video_path, output_path, **kwargs):
|
||||
# 第二帧最大（最清晰）
|
||||
p = Path(output_path)
|
||||
p.parent.mkdir(parents=True, exist_ok=True)
|
||||
if "candidate_1" in str(p):
|
||||
p.write_bytes(b"x" * 10000) # 最大 = 最清晰
|
||||
elif "candidate_0" in str(p):
|
||||
p.write_bytes(b"x" * 1000)
|
||||
else:
|
||||
p.write_bytes(b"x" * 5000)
|
||||
return p
|
||||
|
||||
mock_extract.side_effect = create_frame
|
||||
|
||||
output = tmp_path / "smart_cover.jpg"
|
||||
result = CoverGenerator.extract_smart_cover(
|
||||
sample_video,
|
||||
output,
|
||||
frame_count=3,
|
||||
)
|
||||
|
||||
assert result == output
|
||||
assert output.exists()
|
||||
# 应该选最大的那个文件（candidate_1）
|
||||
assert output.stat().st_size == 10000
|
||||
|
||||
@patch("video_processing.cover_generator.CoverGenerator.extract_frame")
|
||||
@patch("video_processing.cover_generator.probe_video_info")
|
||||
def test_smart_cover_fallback(self, mock_probe, mock_extract, sample_video, tmp_path):
|
||||
"""智能封面全部失败时降级."""
|
||||
mock_probe.return_value = {"duration": 0.0} # 时长为0
|
||||
|
||||
output = tmp_path / "cover.jpg"
|
||||
output.write_bytes(b"x" * 100)
|
||||
mock_extract.return_value = output
|
||||
|
||||
result = CoverGenerator.extract_smart_cover(sample_video, output, frame_count=3)
|
||||
assert result == output
|
||||
|
||||
@patch("video_processing.cover_generator.run_ffmpeg")
|
||||
def test_custom_cover(self, mock_run, sample_image, tmp_path):
|
||||
"""自定义封面处理."""
|
||||
output = tmp_path / "custom_cover.jpg"
|
||||
|
||||
result = CoverGenerator.process_custom_cover(
|
||||
sample_image,
|
||||
output,
|
||||
)
|
||||
|
||||
assert result == output
|
||||
mock_run.assert_called_once()
|
||||
cmd = mock_run.call_args[0][0]
|
||||
assert str(sample_image) in cmd
|
||||
|
||||
def test_custom_cover_not_found(self, tmp_path):
|
||||
"""自定义封面文件不存在."""
|
||||
with pytest.raises(FileNotFoundError):
|
||||
CoverGenerator.process_custom_cover(
|
||||
"/nonexistent/img.png",
|
||||
tmp_path / "cover.jpg",
|
||||
)
|
||||
|
||||
@patch("video_processing.cover_generator.CoverGenerator.extract_frame")
|
||||
def test_generate_cover_time_mode(self, mock_extract, sample_video, tmp_path):
|
||||
"""统一入口 - time 模式."""
|
||||
output = tmp_path / "cover.jpg"
|
||||
mock_extract.return_value = output
|
||||
|
||||
result = CoverGenerator.generate_cover(
|
||||
sample_video,
|
||||
output,
|
||||
mode="time",
|
||||
time_sec=3.0,
|
||||
)
|
||||
|
||||
assert result == output
|
||||
mock_extract.assert_called_once()
|
||||
|
||||
@patch("video_processing.cover_generator.CoverGenerator.extract_smart_cover")
|
||||
def test_generate_cover_smart_mode(self, mock_smart, sample_video, tmp_path):
|
||||
"""统一入口 - smart 模式."""
|
||||
output = tmp_path / "cover.jpg"
|
||||
mock_smart.return_value = output
|
||||
|
||||
result = CoverGenerator.generate_cover(
|
||||
sample_video,
|
||||
output,
|
||||
mode="smart",
|
||||
)
|
||||
|
||||
assert result == output
|
||||
mock_smart.assert_called_once()
|
||||
|
||||
@patch("video_processing.cover_generator.CoverGenerator.process_custom_cover")
|
||||
def test_generate_cover_custom_mode(self, mock_custom, sample_video, sample_image, tmp_path):
|
||||
"""统一入口 - custom 模式."""
|
||||
output = tmp_path / "cover.jpg"
|
||||
mock_custom.return_value = output
|
||||
|
||||
result = CoverGenerator.generate_cover(
|
||||
sample_video,
|
||||
output,
|
||||
mode="custom",
|
||||
custom_image=sample_image,
|
||||
)
|
||||
|
||||
assert result == output
|
||||
mock_custom.assert_called_once()
|
||||
|
||||
|
||||
class TestGenerateCoverFromPlan:
|
||||
"""从 plan 配置生成封面测试."""
|
||||
|
||||
@patch("video_processing.cover_generator.CoverGenerator.extract_smart_cover")
|
||||
def test_smart_mode_from_plan(self, mock_smart, sample_video, tmp_path):
|
||||
"""plan 配置 smart 模式."""
|
||||
plan = FakePlan(id="plan_001", config={"cover_config": {"mode": "smart"}})
|
||||
mock_smart.return_value = tmp_path / "cover.jpg"
|
||||
(tmp_path / "cover.jpg").write_bytes(b"test")
|
||||
|
||||
result = generate_cover_from_plan(plan, sample_video, tmp_path)
|
||||
assert result is not None
|
||||
|
||||
def test_no_cover_config(self, sample_video, tmp_path):
|
||||
"""没有封面配置时返回 None."""
|
||||
plan = FakePlan(id="plan_001", config={})
|
||||
result = generate_cover_from_plan(plan, sample_video, tmp_path)
|
||||
assert result is None
|
||||
|
||||
def test_none_config(self, sample_video, tmp_path):
|
||||
"""config 为 None."""
|
||||
plan = FakePlan(id="plan_001", config=None) # type: ignore
|
||||
result = generate_cover_from_plan(plan, sample_video, tmp_path)
|
||||
assert result is None
|
||||
|
||||
|
||||
# ═══════════════════════════════════════════════════════════════════════════════
|
||||
# 四、UnifiedRenderService 集成测试
|
||||
# ═══════════════════════════════════════════════════════════════════════════════
|
||||
|
||||
|
||||
def _make_clip(clip_id="c1", asset_id="a1", path=Path("/fake/video.mp4"), clip_type="main", config=None):
|
||||
"""创建测试用 ResolvedClip."""
|
||||
return ResolvedClip(
|
||||
clip_id=clip_id,
|
||||
asset_id=asset_id,
|
||||
local_path=path,
|
||||
clip_type=clip_type,
|
||||
order=0,
|
||||
start_time=0.0,
|
||||
duration=0.0,
|
||||
transition_effect="cut",
|
||||
config=config or {},
|
||||
actual_duration=10.0,
|
||||
)
|
||||
|
||||
|
||||
def _make_service(plan, clips, asset_path_map=None, work_dir=None, tmp_path=None):
|
||||
"""创建测试用 UnifiedRenderService."""
|
||||
from pathlib import Path as P
|
||||
|
||||
work_dir = work_dir or (tmp_path or P("/tmp")) / "render_test"
|
||||
work_dir.mkdir(exist_ok=True, parents=True)
|
||||
return UnifiedRenderService(
|
||||
plan=plan,
|
||||
clips=clips,
|
||||
asset_path_map=asset_path_map or {},
|
||||
work_dir=work_dir,
|
||||
output_width=1080,
|
||||
output_height=1920,
|
||||
output_fps=30,
|
||||
transition_duration=0.5,
|
||||
)
|
||||
|
||||
|
||||
class TestReverseIntegration:
|
||||
"""倒放功能集成测试."""
|
||||
|
||||
@patch("video_processing.unified_render_service.probe_video_info")
|
||||
@patch("video_processing.unified_render_service.run_ffmpeg")
|
||||
def test_reverse_in_filter_complex(self, mock_run, mock_probe, tmp_path):
|
||||
"""filter_complex 路径中包含倒放滤镜."""
|
||||
mock_probe.return_value = {"duration": 10.0, "has_audio": True, "width": 1920, "height": 1080}
|
||||
mock_run.return_value = None
|
||||
|
||||
plan = FakePlan(id="p1")
|
||||
clip = _make_clip(config={"reverse": {"enabled": True}})
|
||||
clip.actual_duration = 5.0
|
||||
# 两个 clip 触发 filter_complex 路径
|
||||
clip2 = _make_clip(clip_id="c2", config={})
|
||||
clip2.actual_duration = 5.0
|
||||
clip2.order = 1
|
||||
|
||||
service = _make_service(plan, [clip, clip2], tmp_path=tmp_path)
|
||||
|
||||
# 直接测 _build_filter_complex
|
||||
from video_processing.unified_render_service import RenderLayer
|
||||
|
||||
layer = RenderLayer(role="main", clips=[clip, clip2])
|
||||
filter_str, inputs = service._build_filter_complex([layer])
|
||||
|
||||
assert "reverse" in filter_str
|
||||
|
||||
def test_can_use_pass_through_with_reverse(self, tmp_path):
|
||||
"""倒放不影响直通模式判断（只有贴纸才禁用）."""
|
||||
plan = FakePlan(id="p1")
|
||||
clip = _make_clip(config={"reverse": {"enabled": True}})
|
||||
clip.actual_duration = 5.0
|
||||
|
||||
service = _make_service(plan, [clip], tmp_path=tmp_path)
|
||||
from video_processing.unified_render_service import RenderLayer
|
||||
|
||||
layer = RenderLayer(role="main", clips=[clip])
|
||||
layers = [layer]
|
||||
|
||||
assert service._can_use_pass_through(layers) is True
|
||||
|
||||
|
||||
class TestStickerIntegration:
|
||||
"""贴纸功能集成测试."""
|
||||
|
||||
def test_can_use_pass_through_with_stickers(self, tmp_path):
|
||||
"""有贴纸时禁用直通模式."""
|
||||
plan = FakePlan(id="p1", config={"stickers": [{"type": "text", "text": "Hello", "position": "center"}]})
|
||||
clip = _make_clip()
|
||||
clip.actual_duration = 5.0
|
||||
|
||||
service = _make_service(plan, [clip], tmp_path=tmp_path)
|
||||
from video_processing.unified_render_service import RenderLayer
|
||||
|
||||
layer = RenderLayer(role="main", clips=[clip])
|
||||
layers = [layer]
|
||||
|
||||
assert service._can_use_pass_through(layers) is False
|
||||
|
||||
def test_can_use_pass_through_no_stickers(self, tmp_path):
|
||||
"""无贴纸时直通模式正常."""
|
||||
plan = FakePlan(id="p1", config={})
|
||||
clip = _make_clip()
|
||||
clip.actual_duration = 5.0
|
||||
|
||||
service = _make_service(plan, [clip], tmp_path=tmp_path)
|
||||
from video_processing.unified_render_service import RenderLayer
|
||||
|
||||
layer = RenderLayer(role="main", clips=[clip])
|
||||
layers = [layer]
|
||||
|
||||
assert service._can_use_pass_through(layers) is True
|
||||
|
||||
def test_build_sticker_filters_text(self, tmp_path):
|
||||
"""文字贴纸滤镜构建."""
|
||||
plan = FakePlan(
|
||||
id="p1", config={"stickers": [{"type": "text", "text": "Hello", "position": "top_center", "z_index": 10}]}
|
||||
)
|
||||
service = _make_service(plan, [], tmp_path=tmp_path)
|
||||
|
||||
filter_str, extra_inputs = service._build_sticker_filters("in", "out")
|
||||
|
||||
assert "drawtext" in filter_str
|
||||
assert len(extra_inputs) == 0
|
||||
|
||||
def test_build_sticker_filters_empty(self, tmp_path):
|
||||
"""无贴纸返回空."""
|
||||
plan = FakePlan(id="p1", config={})
|
||||
service = _make_service(plan, [], tmp_path=tmp_path)
|
||||
|
||||
filter_str, extra_inputs = service._build_sticker_filters("in", "out")
|
||||
|
||||
assert filter_str == ""
|
||||
assert extra_inputs == []
|
||||
|
||||
def test_build_sticker_filters_image(self, sample_image, tmp_path):
|
||||
"""图片贴纸滤镜构建 + 额外输入."""
|
||||
plan = FakePlan(
|
||||
id="p1",
|
||||
config={
|
||||
"stickers": [
|
||||
{
|
||||
"type": "image",
|
||||
"image_path": str(sample_image),
|
||||
"position": "bottom_right",
|
||||
"z_index": 5,
|
||||
}
|
||||
]
|
||||
},
|
||||
)
|
||||
service = _make_service(plan, [], tmp_path=tmp_path)
|
||||
|
||||
filter_str, extra_inputs = service._build_sticker_filters("in", "out")
|
||||
|
||||
assert "overlay" in filter_str
|
||||
assert len(extra_inputs) == 1
|
||||
Reference in New Issue
Block a user