diff --git a/apps/api/app/config.py b/apps/api/app/config.py index 81f35b11c..c38067339 100755 --- a/apps/api/app/config.py +++ b/apps/api/app/config.py @@ -109,6 +109,9 @@ class Settings(BaseSettings): LOG_LEVEL: str = "INFO" CORS_ORIGINS_RAW: str = "http://localhost:3000,http://localhost:5173,http://localhost:8000" + # 渲染引擎选择：legacy=旧VideoComposeService，unified=新UnifiedRenderService + RENDER_ENGINE: str = "legacy" + model_config = SettingsConfigDict( env_file=".env", env_file_encoding="utf-8", diff --git a/apps/worker/video_processing/__init__.py b/apps/worker/video_processing/__init__.py old mode 100644 new mode 100755 index 7cb64d4bc..7012f2e1c --- a/apps/worker/video_processing/__init__.py +++ b/apps/worker/video_processing/__init__.py @@ -6,6 +6,7 @@ from . import dedup_helpers, ffmpeg_utils, oss_helpers from .processor import VideoProcessor, VideoResult from .unified_render_service import RenderResult, UnifiedRenderService +from .render_adapter import RenderAdapter, RenderAdapterResult __all__ = [ "VideoProcessor", @@ -15,4 +16,6 @@ __all__ = [ "dedup_helpers", "UnifiedRenderService", "RenderResult", + "RenderAdapter", + "RenderAdapterResult", ] diff --git a/apps/worker/video_processing/render_adapter.py b/apps/worker/video_processing/render_adapter.py new file mode 100755 index 000000000..8e0635391 --- /dev/null +++ b/apps/worker/video_processing/render_adapter.py @@ -0,0 +1,289 @@ +"""统一渲染引擎适配层 — Phase 2. + +将 EditPlan + EditPlanClips（来自 DB）适配为 UnifiedRenderService 的输入格式， +封装素材下载、渲染执行、结果上传的完整流程。 + +职责： + 1. 从 DB 读取 EditPlan + EditPlanClips + 2. 下载素材到本地，构建 asset_path_map + 3. 调用 UnifiedRenderService 执行渲染 + 4. 上传渲染结果到 OSS + 5. 支持进度回调（对接 JobService） +""" + +from __future__ import annotations + +import logging +import tempfile +from dataclasses import dataclass +from pathlib import Path +from typing import Any, Callable + +from sqlalchemy.orm import Session + +from packages.adapters.sqlalchemy_impl.edit_plan_clip_repository import ( + SQLAlchemyEditPlanClipRepository, +) +from packages.adapters.sqlalchemy_impl.edit_plan_repository import ( + SQLAlchemyEditPlanRepository, +) +from packages.domain.edit_plan import EditPlan, EditPlanStatus +from packages.domain.edit_plan_clip import EditPlanClip, EditPlanClipStatus +from video_processing.oss_helpers import download_asset, upload_to_oss +from video_processing.unified_render_service import ( + RenderResult, + UnifiedRenderService, +) + +logger = logging.getLogger(__name__) + + +# ── 数据结构 ────────────────────────────────────────────────────────────────── + + +@dataclass +class RenderAdapterResult: + """渲染适配结果。""" + + success: bool + output_url: str = "" + output_path: Path | None = None + duration: float = 0.0 + file_size: int = 0 + width: int = 0 + height: int = 0 + clip_count: int = 0 + error_message: str = "" + + +ProgressCallback = Callable[[float, str], None] +"""进度回调：(progress_0_100, stage_description) → None""" + + +# ── 适配层主体 ──────────────────────────────────────────────────────────────── + + +class RenderAdapter: + """统一渲染引擎适配层。 + + 桥接 EditPlan 领域模型与 UnifiedRenderService 图层模型。 + + 用法:: + + adapter = RenderAdapter(db) + result = adapter.render_plan( + plan_id=plan_id, + job_id=job_id, + progress_cb=lambda p, s: job_service.update_progress(job_id, p, s), + ) + """ + + def __init__(self, db: Session) -> None: + self._db = db + self._plan_repo = SQLAlchemyEditPlanRepository(db) + self._clip_repo = SQLAlchemyEditPlanClipRepository(db) + + # ── 公开方法 ────────────────────────────────────────────────────────── + + def render_plan( + self, + plan_id: str, + *, + job_id: str = "", + work_dir: Path | None = None, + progress_cb: ProgressCallback | None = None, + ) -> RenderAdapterResult: + """渲染一个 EditPlan。 + + 完整流程： + 1. 加载计划与片段 + 2. 下载素材 + 3. 执行统一渲染 + 4. 上传结果 + + Args: + plan_id: EditPlan ID + job_id: 关联的 Job ID（用于结果存储路径） + work_dir: 工作目录，不传则使用临时目录 + progress_cb: 进度回调函数 + + Returns: + RenderAdapterResult + """ + temp_dir = None + try: + # 0. 准备工作目录 + if work_dir is None: + temp_dir = tempfile.mkdtemp(prefix="render_") + work_dir = Path(temp_dir) + work_dir.mkdir(parents=True, exist_ok=True) + + self._report_progress(progress_cb, 5.0, "加载剪辑计划") + + # 1. 加载计划与片段 + plan = self._plan_repo.get(plan_id) + if plan is None: + return RenderAdapterResult( + success=False, + error_message=f"剪辑计划不存在: {plan_id}", + ) + + clips = self._clip_repo.list_by_plan(plan_id, skip=0, limit=10000) + ready_clips = [c for c in clips if c.status == EditPlanClipStatus.READY and c.asset_id] + ready_clips.sort(key=lambda c: c.order) + + if not ready_clips: + return RenderAdapterResult( + success=False, + error_message="没有可渲染的就绪片段", + clip_count=0, + ) + + logger.info( + "开始渲染: plan_id=%s job_id=%s ready_clips=%d", + plan_id, + job_id, + len(ready_clips), + ) + + self._report_progress(progress_cb, 15.0, f"下载素材（{len(ready_clips)} 个）") + + # 2. 下载素材 + asset_path_map = self._download_assets(ready_clips, work_dir) + if not asset_path_map: + return RenderAdapterResult( + success=False, + error_message="所有素材下载失败", + clip_count=len(ready_clips), + ) + + self._report_progress(progress_cb, 40.0, "执行视频渲染") + + # 3. 执行统一渲染 + render_svc = UnifiedRenderService( + plan=plan, + clips=ready_clips, + asset_path_map=asset_path_map, + work_dir=work_dir, + ) + result = render_svc.render() + + self._report_progress(progress_cb, 80.0, "上传渲染结果") + + # 4. 上传结果 + storage_key = f"rendered/{plan_id}/{job_id or plan_id}.mp4" + output_url = upload_to_oss(result.output_path, storage_key) + + self._report_progress(progress_cb, 100.0, "渲染完成") + + return RenderAdapterResult( + success=True, + output_url=output_url or "", + output_path=result.output_path, + duration=result.duration, + file_size=result.file_size, + width=result.width, + height=result.height, + clip_count=len(ready_clips), + ) + + except Exception as exc: + logger.exception("渲染失败: plan_id=%s", plan_id) + return RenderAdapterResult( + success=False, + error_message=str(exc)[:500], + ) + finally: + # 清理临时目录 + if temp_dir: + import shutil + + try: + shutil.rmtree(temp_dir, ignore_errors=True) + except Exception: + pass + + def validate_plan(self, plan_id: str) -> tuple[bool, list[str], list[str], int, int]: + """校验计划是否可渲染（兼容 VideoComposeService.validate_compose 接口）。 + + Returns: + (valid, errors, warnings, ready_clip_count, total_clip_count) + """ + errors: list[str] = [] + warnings: list[str] = [] + + plan = self._plan_repo.get(plan_id) + if plan is None: + return False, [f"剪辑计划不存在: {plan_id}"], [], 0, 0 + + if plan.status not in (EditPlanStatus.EDITING, EditPlanStatus.RENDERING): + errors.append(f"计划状态不正确，需要 editing 或 rendering，当前: {plan.status}") + + clips = self._clip_repo.list_by_plan(plan_id, skip=0, limit=10000) + if not clips: + errors.append("计划没有任何片段") + return False, errors, warnings, 0, 0 + + clips.sort(key=lambda c: c.order) + + ready_count = 0 + pending_count = 0 + no_asset_count = 0 + + for clip in clips: + if clip.status == EditPlanClipStatus.READY: + ready_count += 1 + if not clip.asset_id: + errors.append(f"片段 {clip.id} (order={clip.order}) 没有分配素材") + no_asset_count += 1 + elif clip.status == EditPlanClipStatus.PENDING: + pending_count += 1 + elif clip.status == EditPlanClipStatus.FAILED: + warnings.append(f"片段 {clip.id} (order={clip.order}) 状态为 failed，已跳过") + + if ready_count == 0: + errors.append("没有就绪（ready）的片段可以合成") + + if pending_count > 0: + warnings.append(f"有 {pending_count} 个片段仍处于 pending 状态") + + return len(errors) == 0, errors, warnings, ready_count, len(clips) + + # ── 内部方法 ────────────────────────────────────────────────────────── + + @staticmethod + def _report_progress(progress_cb: ProgressCallback | None, progress: float, stage: str) -> None: + """上报进度。""" + if progress_cb is not None: + try: + progress_cb(progress, stage) + except Exception: + logger.exception("进度回调失败") + + @staticmethod + def _download_assets(clips: list[EditPlanClip], work_dir: Path) -> dict[str, Path]: + """下载片段素材到本地，返回 asset_id → local_path 映射。 + + 只保留下载成功的素材。 + """ + asset_dir = work_dir / "assets" + asset_dir.mkdir(exist_ok=True) + + asset_path_map: dict[str, Path] = {} + + for clip in clips: + asset_id = clip.asset_id + if not asset_id: + continue + + # 生成安全的本地文件名 + safe_name = f"clip_{clip.order:04d}_{abs(hash(asset_id)) % 100000:05d}.mp4" + local_path = asset_dir / safe_name + + if download_asset(asset_id, local_path): + asset_path_map[asset_id] = local_path + logger.debug("素材下载成功: clip_id=%s asset_id=%s", clip.id, asset_id[:60]) + else: + logger.warning("素材下载失败: clip_id=%s asset_id=%s", clip.id, asset_id[:60]) + + return asset_path_map diff --git a/apps/worker/worker_app/core/config.py b/apps/worker/worker_app/core/config.py old mode 100644 new mode 100755 index 849d3f7cb..977a25e0b --- a/apps/worker/worker_app/core/config.py +++ b/apps/worker/worker_app/core/config.py @@ -18,6 +18,9 @@ class WorkerSettings(BaseSettings): environment: str = "development" auto_create_schema: bool = False + # 渲染引擎选择：legacy=旧VideoComposeService，unified=新UnifiedRenderService + render_engine: str = "legacy" + model_config = SettingsConfigDict( env_file=".env", env_file_encoding="utf-8", diff --git a/apps/worker/worker_app/tasks/compose_video.py b/apps/worker/worker_app/tasks/compose_video.py index 44baf137a..f4fd89620 100755 --- a/apps/worker/worker_app/tasks/compose_video.py +++ b/apps/worker/worker_app/tasks/compose_video.py @@ -39,6 +39,10 @@ def _get_job_service(): def compose_video(self, job_id: str, **kwargs): """视频合成任务。 + 根据 RENDER_ENGINE 配置选择渲染引擎： + - legacy: 旧 VideoComposeService（filter_complex 模式） + - unified: 新 UnifiedRenderService（图层架构） + Args: job_id: JobService 中的任务 ID **kwargs: 来自 Job.payload 的额外参数（plan_id, output_path 等） @@ -56,66 +60,16 @@ def compose_video(self, job_id: str, **kwargs): job_service.fail_job(job_id, "Missing plan_id in job payload") return {"status": "error", "message": "Missing plan_id"} - # 标记为 running - job_service.update_progress(job_id, progress=10.0, current_stage="初始化合成环境") + # 判断使用哪个渲染引擎 + from worker_app.core.config import get_settings as get_worker_settings - # 延迟导入 VideoComposeService - from apps.api.app.services.video_compose_service import VideoComposeService + worker_settings = get_worker_settings() + engine = (worker_settings.render_engine or "legacy").lower() - compose_svc = VideoComposeService(db) - - # 校验合成条件 - job_service.update_progress(job_id, progress=20.0, current_stage="校验合成条件") - validation = compose_svc.validate_compose(plan_id) - if not validation.valid: - error_msg = "; ".join(validation.errors) - job_service.fail_job(job_id, f"合成校验失败: {error_msg}") - return {"status": "error", "message": error_msg} - - # 构建合成命令 - job_service.update_progress(job_id, progress=30.0, current_stage="构建 FFmpeg 命令") - _output_dir = os.environ.get("VIDEO_OUTPUT_DIR", os.path.join(tempfile.gettempdir(), "video_output")) - output_path = os.path.join(_output_dir, f"{job_id}.mp4") - compose_cmd = compose_svc.build_compose_command(plan_id, output_path) - - # 执行 FFmpeg - job_service.update_progress(job_id, progress=50.0, current_stage="正在执行视频合成") - logger.info("Executing FFmpeg for job %s, plan %s", job_id, plan_id) - - try: - subprocess.run( - compose_cmd.command, - check=True, - stdout=subprocess.PIPE, - stderr=subprocess.PIPE, - text=True, - timeout=3600, - ) - except subprocess.CalledProcessError as e: - job_service.fail_job(job_id, f"FFmpeg 执行失败: {e.stderr[:500]}") - raise - - # 上传结果 - job_service.update_progress(job_id, progress=80.0, current_stage="上传合成结果") - storage_key = f"rendered/{plan_id}/{job_id}.mp4" - - from worker_app.tasks.edit_plan_generation import _upload_to_oss - - output_url = _upload_to_oss(Path(output_path), storage_key) - - # 更新 Job 状态为完成 - result_data = { - "plan_id": plan_id, - "output_path": output_path, - "storage_key": storage_key, - "output_url": output_url or "", - "estimated_duration": compose_cmd.estimated_duration, - "clip_count": len(compose_cmd.clip_chains), - } - job_service.complete_job(job_id, result=result_data) - - logger.info("视频合成完成: job_id=%s, plan_id=%s", job_id, plan_id) - return {"status": "completed", "job_id": job_id, "result": result_data} + if engine == "unified": + return _compose_with_unified_engine(self, job_service, job, plan_id, db) + else: + return _compose_with_legacy_engine(self, job_service, job, plan_id, db) except self.retry_exc as exc: logger.warning("视频合成重试中: job_id=%s, exc=%s", job_id, exc) @@ -129,11 +83,140 @@ def compose_video(self, job_id: str, **kwargs): raise self.retry(exc=exc, countdown=60) finally: db.close() - # 清理临时文件 + + +def _compose_with_legacy_engine(task, job_service, job, plan_id: str, db) -> dict: + """旧引擎渲染路径（VideoComposeService）。""" + job_id = job.id + + # 标记为 running + job_service.update_progress(job_id, progress=10.0, current_stage="初始化合成环境") + + # 延迟导入 VideoComposeService + from apps.api.app.services.video_compose_service import VideoComposeService + + compose_svc = VideoComposeService(db) + + # 校验合成条件 + job_service.update_progress(job_id, progress=20.0, current_stage="校验合成条件") + validation = compose_svc.validate_compose(plan_id) + if not validation.valid: + error_msg = "; ".join(validation.errors) + job_service.fail_job(job_id, f"合成校验失败: {error_msg}") + return {"status": "error", "message": error_msg} + + # 构建合成命令 + job_service.update_progress(job_id, progress=30.0, current_stage="构建 FFmpeg 命令") + _output_dir = os.environ.get("VIDEO_OUTPUT_DIR", os.path.join(tempfile.gettempdir(), "video_output")) + output_path = os.path.join(_output_dir, f"{job_id}.mp4") + compose_cmd = compose_svc.build_compose_command(plan_id, output_path) + + # 执行 FFmpeg + job_service.update_progress(job_id, progress=50.0, current_stage="正在执行视频合成") + logger.info("Executing FFmpeg for job %s, plan %s", job_id, plan_id) + + try: + subprocess.run( + compose_cmd.command, + check=True, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + text=True, + timeout=3600, + ) + except subprocess.CalledProcessError as e: + job_service.fail_job(job_id, f"FFmpeg 执行失败: {e.stderr[:500]}") + raise + + # 上传结果 + job_service.update_progress(job_id, progress=80.0, current_stage="上传合成结果") + storage_key = f"rendered/{plan_id}/{job_id}.mp4" + + from worker_app.tasks.edit_plan_generation import _upload_to_oss + + output_url = _upload_to_oss(Path(output_path), storage_key) + + # 更新 Job 状态为完成 + result_data = { + "plan_id": plan_id, + "output_path": output_path, + "storage_key": storage_key, + "output_url": output_url or "", + "estimated_duration": compose_cmd.estimated_duration, + "clip_count": len(compose_cmd.clip_chains), + "engine": "legacy", + } + job_service.complete_job(job_id, result=result_data) + + logger.info("视频合成完成(legacy): job_id=%s, plan_id=%s", job_id, plan_id) + return {"status": "completed", "job_id": job_id, "result": result_data} + + +def _compose_with_unified_engine(task, job_service, job, plan_id: str, db) -> dict: + """新引擎渲染路径（UnifiedRenderService + RenderAdapter）。""" + job_id = job.id + + # 标记为 running + job_service.update_progress(job_id, progress=10.0, current_stage="初始化统一渲染引擎") + + from video_processing.render_adapter import RenderAdapter + + adapter = RenderAdapter(db) + + # 校验合成条件 + job_service.update_progress(job_id, progress=15.0, current_stage="校验合成条件") + valid, errors, warnings, ready_count, total_count = adapter.validate_plan(plan_id) + if not valid: + error_msg = "; ".join(errors) + job_service.fail_job(job_id, f"合成校验失败: {error_msg}") + return {"status": "error", "message": error_msg} + + # 进度回调 + def progress_cb(progress: float, stage: str) -> None: try: - _output_dir = os.environ.get("VIDEO_OUTPUT_DIR", os.path.join(tempfile.gettempdir(), "video_output")) - output_path = os.path.join(_output_dir, f"{job_id}.mp4") - if Path(output_path).exists(): - Path(output_path).unlink() - except Exception as e: - logger.warning(f"Operation failed in apps/worker/worker_app/tasks/compose_video.py: {e}", exc_info=True) + job_service.update_progress(job_id, progress=progress, current_stage=stage) + except Exception: + logger.exception("更新进度失败") + + # 执行渲染 + job_service.update_progress(job_id, progress=20.0, current_stage="开始渲染") + logger.info("统一渲染引擎开始: job_id=%s plan_id=%s", job_id, plan_id) + + result = adapter.render_plan( + plan_id=plan_id, + job_id=job_id, + progress_cb=progress_cb, + ) + + if not result.success: + job_service.fail_job(job_id, f"渲染失败: {result.error_message}") + raise RuntimeError(result.error_message) + + # 更新 Job 状态为完成 + result_data = { + "plan_id": plan_id, + "output_path": str(result.output_path) if result.output_path else "", + "storage_key": f"rendered/{plan_id}/{job_id}.mp4", + "output_url": result.output_url, + "estimated_duration": result.duration, + "clip_count": result.clip_count, + "engine": "unified", + "width": result.width, + "height": result.height, + "file_size": result.file_size, + } + job_service.complete_job(job_id, result=result_data) + + logger.info("视频合成完成(unified): job_id=%s plan_id=%s duration=%.2fs", job_id, plan_id, result.duration) + return {"status": "completed", "job_id": job_id, "result": result_data} + + +def _cleanup_output(job_id: str) -> None: + """清理临时输出文件。""" + try: + _output_dir = os.environ.get("VIDEO_OUTPUT_DIR", os.path.join(tempfile.gettempdir(), "video_output")) + output_path = os.path.join(_output_dir, f"{job_id}.mp4") + if Path(output_path).exists(): + Path(output_path).unlink() + except Exception as e: + logger.warning(f"清理输出文件失败: {e}", exc_info=True) diff --git a/tests/unit/test_render_adapter.py b/tests/unit/test_render_adapter.py new file mode 100755 index 000000000..706075a4e --- /dev/null +++ b/tests/unit/test_render_adapter.py @@ -0,0 +1,423 @@ +"""RenderAdapter 单元测试 — Phase 2. + +测试适配层的计划加载、素材下载、引擎调用、结果上传等逻辑。 +""" + +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.render_adapter import RenderAdapter, RenderAdapterResult + + +# ── Fixtures ────────────────────────────────────────────────────────────────── + + +@dataclass +class FakeClip: + """模拟 EditPlanClip。""" + + id: str + plan_id: str = "plan_001" + clip_type: str = "main" + order: int = 0 + asset_id: str = "" + text_content: str = "" + start_time: float = 0.0 + duration: float = 0.0 + transition_effect: str = "cut" + status: str = "ready" + config: dict[str, Any] = field(default_factory=dict) + + +@dataclass +class FakePlan: + """模拟 EditPlan。""" + + id: str = "plan_001" + name: str = "测试计划" + status: str = "editing" + config: dict[str, Any] = field(default_factory=dict) + + +def _make_clip( + clip_id: str, + clip_type: str = "main", + order: int = 0, + asset_id: str | None = None, + duration: float = 5.0, + status: str = "ready", + transition_effect: str = "cut", + config: dict[str, Any] | None = None, +) -> FakeClip: + # asset_id 为 None 时生成默认值，为空字符串时保留空串 + if asset_id is None: + asset_id = f"asset_{clip_id}.mp4" + return FakeClip( + id=clip_id, + clip_type=clip_type, + order=order, + asset_id=asset_id, + duration=duration, + status=status, + transition_effect=transition_effect, + config=config or {}, + ) + + +def _make_adapter( + plan: FakePlan | None = None, + clips: list[FakeClip] | None = None, +) -> tuple[RenderAdapter, MagicMock, MagicMock]: + """创建测试用的 RenderAdapter 及 mock repo。 + + Returns: + (adapter, mock_plan_repo, mock_clip_repo) + """ + mock_db = MagicMock() + adapter = RenderAdapter(mock_db) + + # 替换内部 repo + mock_plan_repo = MagicMock() + mock_clip_repo = MagicMock() + adapter._plan_repo = mock_plan_repo + adapter._clip_repo = mock_clip_repo + + # 设置默认返回 + if plan is not None: + mock_plan_repo.get.return_value = plan + if clips is not None: + mock_clip_repo.list_by_plan.return_value = clips + + return adapter, mock_plan_repo, mock_clip_repo + + +# ── validate_plan 测试 ─────────────────────────────────────────────────────── + + +class TestValidatePlan: + def test_plan_not_found(self): + """计划不存在时校验失败。""" + adapter, mock_plan_repo, _ = _make_adapter(plan=None) + mock_plan_repo.get.return_value = None + + valid, errors, warnings, ready_count, total_count = adapter.validate_plan("plan_001") + + assert not valid + assert len(errors) == 1 + assert "不存在" in errors[0] + assert ready_count == 0 + assert total_count == 0 + + def test_no_clips(self): + """没有任何片段时校验失败。""" + plan = FakePlan(id="plan_001", status="editing") + adapter, _, mock_clip_repo = _make_adapter(plan=plan, clips=[]) + + valid, errors, warnings, ready_count, total_count = adapter.validate_plan("plan_001") + + assert not valid + assert any("没有任何片段" in e for e in errors) + + def test_no_ready_clips(self): + """没有 ready 片段时校验失败。""" + plan = FakePlan(id="plan_001", status="editing") + clips = [ + _make_clip("c1", status="pending"), + _make_clip("c2", status="pending"), + ] + adapter, _, _ = _make_adapter(plan=plan, clips=clips) + + valid, errors, warnings, ready_count, total_count = adapter.validate_plan("plan_001") + + assert not valid + assert any("没有就绪" in e for e in errors) + assert ready_count == 0 + assert total_count == 2 + + def test_ready_clip_no_asset(self): + """ready 片段没有 asset_id 时报错。""" + plan = FakePlan(id="plan_001", status="editing") + clips = [ + _make_clip("c1", asset_id=""), + ] + adapter, _, _ = _make_adapter(plan=plan, clips=clips) + + valid, errors, warnings, ready_count, total_count = adapter.validate_plan("plan_001") + + assert not valid + assert any("没有分配素材" in e for e in errors) + + def test_valid_plan(self): + """正常计划校验通过。""" + plan = FakePlan(id="plan_001", status="editing") + clips = [ + _make_clip("c1", order=0, duration=3.0), + _make_clip("c2", order=1, duration=4.0), + ] + adapter, _, _ = _make_adapter(plan=plan, clips=clips) + + valid, errors, warnings, ready_count, total_count = adapter.validate_plan("plan_001") + + assert valid + assert len(errors) == 0 + assert ready_count == 2 + assert total_count == 2 + + def test_wrong_status(self): + """计划状态不正确时报错。""" + plan = FakePlan(id="plan_001", status="draft") + clips = [_make_clip("c1")] + adapter, _, _ = _make_adapter(plan=plan, clips=clips) + + valid, errors, _, _, _ = adapter.validate_plan("plan_001") + + assert not valid + assert any("状态不正确" in e for e in errors) + + def test_mixed_status_with_warnings(self): + """混合状态时有 pending/failed 警告。""" + plan = FakePlan(id="plan_001", status="editing") + clips = [ + _make_clip("c1", order=0, status="ready"), + _make_clip("c2", order=1, status="pending"), + _make_clip("c3", order=2, status="failed"), + ] + adapter, _, _ = _make_adapter(plan=plan, clips=clips) + + valid, errors, warnings, ready_count, total_count = adapter.validate_plan("plan_001") + + assert valid + assert any("pending" in w for w in warnings) + assert any("failed" in w for w in warnings) + assert ready_count == 1 + assert total_count == 3 + + +# ── render_plan 测试 ───────────────────────────────────────────────────────── + + +class TestRenderPlan: + def test_plan_not_found(self): + """计划不存在时返回失败。""" + adapter, mock_plan_repo, _ = _make_adapter(plan=None) + mock_plan_repo.get.return_value = None + + result = adapter.render_plan("plan_001") + + assert not result.success + assert "不存在" in result.error_message + + def test_no_ready_clips(self): + """没有 ready 片段时返回失败。""" + plan = FakePlan(id="plan_001", status="editing") + clips = [_make_clip("c1", status="pending")] + adapter, _, _ = _make_adapter(plan=plan, clips=clips) + + result = adapter.render_plan("plan_001") + + assert not result.success + assert "没有可渲染" in result.error_message + assert result.clip_count == 0 + + @patch("video_processing.render_adapter.download_asset") + def test_all_assets_download_fail(self, mock_download): + """所有素材下载失败时返回失败。""" + mock_download.return_value = False + + plan = FakePlan(id="plan_001", status="editing") + clips = [_make_clip("c1", order=0, duration=5.0)] + adapter, _, _ = _make_adapter(plan=plan, clips=clips) + + result = adapter.render_plan("plan_001") + + assert not result.success + assert "素材下载失败" in result.error_message + + @patch("video_processing.render_adapter.upload_to_oss") + @patch("video_processing.render_adapter.UnifiedRenderService") + @patch("video_processing.render_adapter.download_asset") + def test_successful_render(self, mock_download, mock_render_cls, mock_upload, tmp_path): + """完整渲染流程成功。""" + # 素材下载成功 + def _fake_download(asset_id, local_path): + local_path.parent.mkdir(parents=True, exist_ok=True) + local_path.write_bytes(b"fake video data") + return True + + mock_download.side_effect = _fake_download + + # 渲染成功 + mock_render = MagicMock() + mock_render.render.return_value = MagicMock( + output_path=tmp_path / "output.mp4", + duration=10.0, + file_size=102400, + width=1280, + height=720, + ) + mock_render_cls.return_value = mock_render + + # 上传成功 + mock_upload.return_value = "https://oss.example.com/rendered/plan_001/job_001.mp4" + + plan = FakePlan(id="plan_001", status="editing") + clips = [ + _make_clip("c1", order=0, duration=5.0), + _make_clip("c2", order=1, duration=5.0), + ] + adapter, _, _ = _make_adapter(plan=plan, clips=clips) + + result = adapter.render_plan( + "plan_001", + job_id="job_001", + work_dir=tmp_path / "work", + ) + + assert result.success + assert result.output_url.startswith("https://") + assert result.duration == 10.0 + assert result.width == 1280 + assert result.height == 720 + assert result.clip_count == 2 + + # 验证 UnifiedRenderService 被正确调用 + mock_render_cls.assert_called_once() + call_kwargs = mock_render_cls.call_args + assert call_kwargs.kwargs["plan"] is plan + assert len(call_kwargs.kwargs["clips"]) == 2 + assert len(call_kwargs.kwargs["asset_path_map"]) == 2 + + @patch("video_processing.render_adapter.download_asset") + def test_progress_callback(self, mock_download, tmp_path): + """进度回调被正确触发。""" + def _fake_download(asset_id, local_path): + local_path.parent.mkdir(parents=True, exist_ok=True) + local_path.write_bytes(b"fake data") + return True + + mock_download.side_effect = _fake_download + + # 模拟渲染异常，避免走到最后 + with patch("video_processing.render_adapter.UnifiedRenderService") as mock_render_cls: + mock_render = MagicMock() + mock_render.render.side_effect = RuntimeError("render error") + mock_render_cls.return_value = mock_render + + plan = FakePlan(id="plan_001", status="editing") + clips = [_make_clip("c1", order=0, duration=5.0)] + adapter, _, _ = _make_adapter(plan=plan, clips=clips) + + progress_values = [] + + def progress_cb(progress: float, stage: str) -> None: + progress_values.append((progress, stage)) + + result = adapter.render_plan( + "plan_001", + work_dir=tmp_path / "work", + progress_cb=progress_cb, + ) + + # 即使渲染失败，前期进度也应该上报了 + assert len(progress_values) > 0 + # 第一个进度应该是加载计划 + assert progress_values[0][1] == "加载剪辑计划" + + @patch("video_processing.render_adapter.download_asset") + def test_partial_asset_download(self, mock_download, tmp_path): + """部分素材下载失败时，只使用成功的素材。""" + download_results = [True, False, True] # 3个素材中2个成功 + + def _fake_download(asset_id, local_path): + idx = hash(asset_id) % 3 + if download_results[idx]: + local_path.parent.mkdir(parents=True, exist_ok=True) + local_path.write_bytes(b"fake data") + return True + return False + + mock_download.side_effect = _fake_download + + with patch("video_processing.render_adapter.UnifiedRenderService") as mock_render_cls: + mock_render = MagicMock() + mock_render.render.return_value = MagicMock( + output_path=tmp_path / "out.mp4", + duration=5.0, + file_size=1024, + width=1280, + height=720, + ) + mock_render_cls.return_value = mock_render + + with patch("video_processing.render_adapter.upload_to_oss", return_value="https://example.com/out.mp4"): + plan = FakePlan(id="plan_001", status="editing") + clips = [ + _make_clip("c1", order=0, duration=3.0, asset_id="asset_001.mp4"), + _make_clip("c2", order=1, duration=3.0, asset_id="asset_002.mp4"), + _make_clip("c3", order=2, duration=3.0, asset_id="asset_003.mp4"), + ] + adapter, _, _ = _make_adapter(plan=plan, clips=clips) + + result = adapter.render_plan( + "plan_001", + work_dir=tmp_path / "work", + ) + + # 至少有部分素材成功，渲染应该进行 + # （具体成功数量取决于 hash 结果，但至少1个成功就能渲染） + assert result.success or "素材下载失败" in result.error_message + + +# ── _download_assets 测试 ──────────────────────────────────────────────────── + + +class TestDownloadAssets: + @patch("video_processing.render_adapter.download_asset") + def test_all_download_success(self, mock_download, tmp_path): + """全部素材下载成功。""" + mock_download.return_value = True + + clips = [ + _make_clip("c1", order=0, asset_id="key1.mp4"), + _make_clip("c2", order=1, asset_id="key2.mp4"), + ] + + result = RenderAdapter._download_assets(clips, tmp_path) + + assert len(result) == 2 + assert "key1.mp4" in result + assert "key2.mp4" in result + assert mock_download.call_count == 2 + + @patch("video_processing.render_adapter.download_asset") + def test_empty_asset_id_skipped(self, mock_download, tmp_path): + """空 asset_id 的片段被跳过。""" + clips = [ + _make_clip("c1", order=0, asset_id=""), + _make_clip("c2", order=1, asset_id="key2.mp4"), + ] + mock_download.return_value = True + + result = RenderAdapter._download_assets(clips, tmp_path) + + assert len(result) == 1 + assert "key2.mp4" in result + assert mock_download.call_count == 1 # 只调用了一次下载 + + @patch("video_processing.render_adapter.download_asset") + def test_all_download_fail(self, mock_download, tmp_path): + """全部下载失败返回空字典。""" + mock_download.return_value = False + + clips = [ + _make_clip("c1", order=0, asset_id="key1.mp4"), + ] + + result = RenderAdapter._download_assets(clips, tmp_path) + + assert len(result) == 0