Files
xiaoxia-saas/apps/api/app/services/video_compose_service.py
T
CI Bot 48bf66c298
CI/CD Pipeline / Check if frontend-only change (push) Has been skipped
CI/CD Pipeline / Validate - Code Quality (push) Failing after 1m24s
CI/CD Pipeline / PR Build API Image (push) Has been skipped
CI/CD Pipeline / PR Build Web Image (push) Has been skipped
CI/CD Pipeline / PR Build Worker Image (push) Has been skipped
CI/CD Pipeline / Frontend Lint (push) Successful in 32s
CI/CD Pipeline / Validate - Type Check (mypy) (push) Successful in 55s
CI/CD Pipeline / Build Production API Image (push) Has been skipped
CI/CD Pipeline / Build Production Web Image (push) Has been skipped
CI/CD Pipeline / Build Production Worker Image (push) Has been skipped
CI/CD Pipeline / Validate - Migration (alembic) (push) Successful in 1m3s
CI/CD Pipeline / Build Staging Web Image (push) Successful in 1m41s
CI/CD Pipeline / Build Staging Worker Image (push) Successful in 4m38s
CI/CD Pipeline / Deploy Production (push) Has been skipped
CI/CD Pipeline / Production Browser E2E (push) Has been skipped
CI/CD Pipeline / Frontend Unit Tests (push) Successful in 1m12s
CI/CD Pipeline / Integration Tests (push) Successful in 2m16s
CI/CD Pipeline / Build Staging API Image (push) Successful in 13m2s
CI/CD Pipeline / Deploy Staging (Watchtower auto-deploy) (push) Successful in 2m3s
CI/CD Pipeline / ACR Image Cleanup (push) Successful in 23s
CI/CD Pipeline / Staging E2E Tests (push) Successful in 3m5s
CI/CD Pipeline / Staging API Integration Tests (push) Successful in 5m14s
CI/CD Pipeline / Canary Release to Production (push) Has been skipped
CI/CD Pipeline / Unit Tests (push) Failing after 42m36s
style: auto-format with black + isort + prettier
2026-07-26 14:08:03 +00:00

434 lines
15 KiB
Python
Executable File
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
"""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_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)