Compare commits
1 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| 2dbddf536e |
@@ -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",
|
||||
|
||||
Regular → Executable
+3
@@ -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",
|
||||
]
|
||||
|
||||
+289
@@ -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
|
||||
Regular → Executable
+3
@@ -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",
|
||||
|
||||
@@ -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)
|
||||
|
||||
Executable
+423
@@ -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
|
||||
Reference in New Issue
Block a user