"""VideoComposeService — Phase 8 任务 2.09. FFmpeg 视频合成编排服务： 1. 根据 EditPlan + EditPlanClips 生成 FFmpeg filter_complex 命令 2. 支持逐片段 scale / crop / trim / setpts 滤镜 3. 支持转场效果（fade / slide / dissolve / wipe） 4. 支持音频流合并 5. 提供合成前校验逻辑 设计原则： - 本服务只负责 **命令生成 + 校验**，不执行 FFmpeg - Worker 层（Celery task）调用本服务生成命令后执行 - API 层可调用 build_compose_command 做预览 / 调试 """ from __future__ import annotations import logging from dataclasses import dataclass, field from typing import Any 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 EditPlanStatus from packages.domain.edit_plan_clip import EditPlanClip, EditPlanClipStatus from packages.domain.video_filter_builder import ( DEFAULT_FPS, DEFAULT_OUTPUT_HEIGHT, DEFAULT_OUTPUT_WIDTH, DEFAULT_TRANSITION_DURATION, ClipFilterChain, build_clip_filter, ) from packages.domain.video_filter_builder import build_concat_filter as _build_concat_filter_func from packages.domain.video_filter_builder import build_filter_complex as _build_filter_complex from packages.domain.video_filter_builder import build_xfade_filter as _build_xfade_filter_func from packages.domain.video_filter_builder import chain_filters as _chain_filters_func from packages.domain.video_filter_builder import has_audio as _has_audio_func logger = logging.getLogger(__name__) # ── 常量（向后兼容别名） ────────────────────────────────────────────────────── # 实际定义已迁移至 packages/domain/video_filter_builder.py DEFAULT_CODEC = "libx264" DEFAULT_CRF = 23 DEFAULT_PRESET = "medium" # ── 数据结构 ────────────────────────────────────────────────────────────────── @dataclass(frozen=True) class ComposeCommand: """完整的 FFmpeg 合成命令描述。""" command: list[str] """可直接传给 subprocess.run 的命令列表。""" filter_complex: str """-filter_complex 参数值（方便调试 / 日志）。""" input_paths: list[str] """输入文件路径列表。""" output_path: str """输出文件路径。""" estimated_duration: float """预估输出时长（秒）。""" clip_chains: list[ClipFilterChain] """每个片段的滤镜链描述。""" @dataclass(frozen=True) class ComposeValidation: """合成前校验结果。""" valid: bool errors: list[str] = field(default_factory=list) warnings: list[str] = field(default_factory=list) ready_clip_count: int = 0 total_clip_count: int = 0 # ── 服务主体 ────────────────────────────────────────────────────────────────── class VideoComposeService: """FFmpeg 视频合成编排服务。 职责： - 根据 EditPlan 及其 Clips 生成 FFmpeg filter_complex 命令 - 校验合成前置条件 - 提供合成状态查询 用法:: svc = VideoComposeService(db) validation = svc.validate_compose(plan_id) if validation.valid: cmd = svc.build_compose_command(plan_id, output_path="/tmp/out.mp4") subprocess.run(cmd.command, check=True) """ def __init__(self, db: Session) -> None: self._db = db self._plan_repo = SQLAlchemyEditPlanRepository(db) self._clip_repo = SQLAlchemyEditPlanClipRepository(db) # ── 公开方法 ────────────────────────────────────────────────────────── def validate_compose(self, plan_id: str) -> ComposeValidation: """校验剪辑计划是否可以合成。 检查项： 1. 计划存在 2. 计划状态为 editing 或 rendering 3. 至少有一个 ready 状态的片段 4. 每个 ready 片段都有 asset_id 5. 每个 ready 片段都有 duration > 0 """ errors: list[str] = [] warnings: list[str] = [] plan = self._plan_repo.get(plan_id) if plan is None: return ComposeValidation( valid=False, errors=[f"剪辑计划不存在: {plan_id}"], ) # 状态检查 if plan.status not in (EditPlanStatus.EDITING, EditPlanStatus.RENDERING): errors.append(f"计划状态不正确，需要 editing 或 rendering，当前: {plan.status.value}") # 加载片段 clips = self._clip_repo.list_by_plan(plan_id, skip=0, limit=10000) if not clips: errors.append("计划没有任何片段") return ComposeValidation( valid=False, errors=errors, total_clip_count=0, ) # 按 order 排序 clips.sort(key=lambda c: c.order) ready_count = 0 pending_count = 0 no_asset_count = 0 no_duration_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 if clip.duration <= 0: warnings.append(f"片段 {clip.id} (order={clip.order}) 时长为 0，将使用默认时长") no_duration_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 ComposeValidation( valid=len(errors) == 0, errors=errors, warnings=warnings, ready_clip_count=ready_count, total_clip_count=len(clips), ) def build_compose_command( self, plan_id: str, output_path: str, *, output_width: int = DEFAULT_OUTPUT_WIDTH, output_height: int = DEFAULT_OUTPUT_HEIGHT, fps: int = DEFAULT_FPS, codec: str = DEFAULT_CODEC, crf: int = DEFAULT_CRF, preset: str = DEFAULT_PRESET, transition_duration: float = DEFAULT_TRANSITION_DURATION, ) -> ComposeCommand: """构建 FFmpeg 合成命令。 根据 EditPlan 的所有 ready 片段，生成完整的 filter_complex 命令。 滤镜链逻辑： - 每个片段：scale → crop → fps → setpts → trim → atrim - 多片段之间：concat 滤镜 或 xfade 转场 - 最终输出：-map '[outv]' -map '[outa]'（如有音频） """ plan = self._plan_repo.get(plan_id) if plan is None: raise ValueError(f"剪辑计划不存在: {plan_id}") clips = self._clip_repo.list_by_plan(plan_id, skip=0, limit=10000) if not clips: raise ValueError(f"剪辑计划没有片段: {plan_id}") # 只处理 ready 且有 asset_id 的片段 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: raise ValueError(f"剪辑计划没有可合成的片段: {plan_id}") # 构建每个片段的滤镜链 clip_chains: list[ClipFilterChain] = [] input_paths: list[str] = [] for idx, clip in enumerate(ready_clips): chain = self._build_clip_filter( clip=clip, input_index=idx, output_width=output_width, output_height=output_height, fps=fps, ) clip_chains.append(chain) input_paths.append(clip.asset_id) # asset_id 存储的是 storage_key / URL # 构建 filter_complex filter_complex, estimated_duration = self._build_filter_complex( clip_chains=clip_chains, output_width=output_width, output_height=output_height, transition_duration=transition_duration, transitions=[c.transition_effect for c in ready_clips], ) # 构建完整命令 command: list[str] = ["ffmpeg", "-y"] # 输入文件 for path in input_paths: command.extend(["-i", path]) # filter_complex command.extend(["-filter_complex", filter_complex]) # 映射输出流 command.extend(["-map", "[outv]"]) if self._has_audio(clip_chains): command.extend(["-map", "[outa]"]) # 编码参数 command.extend( [ "-c:v", codec, "-crf", str(crf), "-preset", preset, "-c:a", "aac", "-b:a", "192k", ] ) # 输出 command.append(output_path) return ComposeCommand( command=command, filter_complex=filter_complex, input_paths=input_paths, output_path=output_path, estimated_duration=estimated_duration, clip_chains=clip_chains, ) def build_single_clip_command( self, clip_id: str, output_path: str, *, output_width: int = DEFAULT_OUTPUT_WIDTH, output_height: int = DEFAULT_OUTPUT_HEIGHT, fps: int = DEFAULT_FPS, ) -> ComposeCommand: """为单个片段构建 FFmpeg 命令（预览 / 调试用）。""" clip = self._clip_repo.get(clip_id) if clip is None: raise ValueError(f"片段不存在: {clip_id}") if not clip.asset_id: raise ValueError(f"片段没有分配素材: {clip_id}") chain = self._build_clip_filter( clip=clip, input_index=0, output_width=output_width, output_height=output_height, fps=fps, ) # 简单命令：input → filter → output filter_str = ",".join(chain.filters) command = [ "ffmpeg", "-y", "-i", clip.asset_id, "-filter_complex", f"{filter_str}[outv]", "-map", "[outv]", "-c:v", DEFAULT_CODEC, "-crf", str(DEFAULT_CRF), "-preset", DEFAULT_PRESET, output_path, ] return ComposeCommand( command=command, filter_complex=filter_str, input_paths=[clip.asset_id], output_path=output_path, estimated_duration=clip.duration, clip_chains=[chain], ) def get_compose_status(self, plan_id: str) -> dict[str, Any]: """获取合成状态摘要。""" plan = self._plan_repo.get(plan_id) if plan is None: raise ValueError(f"剪辑计划不存在: {plan_id}") clips = self._clip_repo.list_by_plan(plan_id, skip=0, limit=10000) clips.sort(key=lambda c: c.order) total_duration = sum(c.duration for c in clips if c.duration > 0) ready_clips = [c for c in clips if c.status == EditPlanClipStatus.READY] pending_clips = [c for c in clips if c.status == EditPlanClipStatus.PENDING] rendered_clips = [c for c in clips if c.status == EditPlanClipStatus.RENDERED] failed_clips = [c for c in clips if c.status == EditPlanClipStatus.FAILED] return { "plan_id": plan_id, "plan_status": plan.status.value, "total_clips": len(clips), "ready_clips": len(ready_clips), "pending_clips": len(pending_clips), "rendered_clips": len(rendered_clips), "failed_clips": len(failed_clips), "total_duration": total_duration, "can_compose": len(ready_clips) > 0 and plan.status in ( EditPlanStatus.EDITING, EditPlanStatus.RENDERING, ), "rendered_url": plan.config.get("rendered_storage_key", "") or plan.config.get("rendered_url", ""), } # ── 内部方法 ────────────────────────────────────────────────────────── @staticmethod def _build_clip_filter( clip: EditPlanClip, input_index: int, output_width: int, output_height: int, fps: int, ) -> ClipFilterChain: """向后兼容：委托给 video_filter_builder.build_clip_filter。""" return build_clip_filter(clip, input_index, output_width, output_height, fps) @staticmethod def _build_filter_complex( clip_chains: list[ClipFilterChain], output_width: int, output_height: int, transition_duration: float, transitions: list[str], ) -> tuple[str, float]: """向后兼容：委托给 video_filter_builder.build_filter_complex。""" return _build_filter_complex(clip_chains, output_width, output_height, transition_duration, transitions) @staticmethod def _has_audio(clip_chains: list[ClipFilterChain]) -> bool: """向后兼容：委托给 video_filter_builder.has_audio。""" return _has_audio_func(clip_chains) # ── 模块级辅助函数（向后兼容别名） ────────────────────────────────────────── # 实际实现已迁移至 packages/domain/video_filter_builder.py # 保留此处别名以兼容现有测试与调用方 def _chain_filters(filters: list[str], output_label: str) -> str: """向后兼容：委托给 video_filter_builder.chain_filters。""" return _chain_filters_func(filters, output_label) def _build_concat_filter( clip_chains: list[ClipFilterChain], ) -> tuple[str, float]: """向后兼容：委托给 video_filter_builder.build_concat_filter。""" return _build_concat_filter_func(clip_chains) def _build_xfade_filter( clip_chains: list[ClipFilterChain], transition_duration: float, transitions: list[str], ) -> tuple[str, float]: """向后兼容：委托给 video_filter_builder.build_xfade_filter。""" return _build_xfade_filter_func(clip_chains, transition_duration, transitions)