From b99146f353f24f34f4454b003e93b958819dfe23 Mon Sep 17 00:00:00 2001 From: xiaoxia Date: Thu, 13 Aug 2026 01:01:58 +0800 Subject: [PATCH] =?UTF-8?q?refactor(#1341):=20=E5=88=A0=E9=99=A4=20Editing?= =?UTF-8?q?Planner=20=E6=97=B6=E4=BB=A3=E9=81=97=E7=95=99=E7=9A=84?= =?UTF-8?q?=E5=B0=81=E9=9D=A2=E7=94=9F=E6=88=90=E5=99=A8?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 删除 cover_generator.py（431行），旧封面逻辑已不再使用 - compose_video.py 移除封面生成代码块，智能剪辑封面由 Step6 前端 通过 /generate-cover API → ai_service.py → MediaKit 独立处理 - 删除关联的 3 个测试文件（test_cover_generator_pure, test_cover_reverse_sticker, test_compose_video_cover_integration） - 保留唯一链路: Step6 → /generate-cover → MediaKit AI 抽帧 --- .../video_processing/cover_generator.py | 431 --------- apps/worker/worker_app/tasks/compose_video.py | 31 - .../test_compose_video_cover_integration.py | 232 ----- tests/unit/test_cover_generator_pure.py | 538 ----------- tests/unit/test_cover_reverse_sticker.py | 860 ------------------ 5 files changed, 2092 deletions(-) delete mode 100755 apps/worker/video_processing/cover_generator.py delete mode 100644 tests/unit/test_compose_video_cover_integration.py delete mode 100755 tests/unit/test_cover_generator_pure.py delete mode 100755 tests/unit/test_cover_reverse_sticker.py diff --git a/apps/worker/video_processing/cover_generator.py b/apps/worker/video_processing/cover_generator.py deleted file mode 100755 index 5d945c529..000000000 --- a/apps/worker/video_processing/cover_generator.py +++ /dev/null @@ -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 diff --git a/apps/worker/worker_app/tasks/compose_video.py b/apps/worker/worker_app/tasks/compose_video.py index 1b1f2be4b..5cb97b80e 100644 --- a/apps/worker/worker_app/tasks/compose_video.py +++ b/apps/worker/worker_app/tasks/compose_video.py @@ -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) diff --git a/tests/unit/test_compose_video_cover_integration.py b/tests/unit/test_compose_video_cover_integration.py deleted file mode 100644 index b0312e9d7..000000000 --- a/tests/unit/test_compose_video_cover_integration.py +++ /dev/null @@ -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 diff --git a/tests/unit/test_cover_generator_pure.py b/tests/unit/test_cover_generator_pure.py deleted file mode 100755 index ca97d7308..000000000 --- a/tests/unit/test_cover_generator_pure.py +++ /dev/null @@ -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 diff --git a/tests/unit/test_cover_reverse_sticker.py b/tests/unit/test_cover_reverse_sticker.py deleted file mode 100755 index b2460494b..000000000 --- a/tests/unit/test_cover_reverse_sticker.py +++ /dev/null @@ -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 -- 2.54.0