xiaoxia-saas/apps/api/app/services/video_compose_service.py

"""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)