xiaoxia-saas/apps/worker/video_processing/trim_engine.py

"""裁剪引擎 — 基于 FFmpeg trim/atrim 的精确帧级裁剪.

支持：
- 入点出点裁剪（start_time / end_time / duration 三选二）
- 边界自动钳制（超出素材时长自动修正，不阻断渲染）
- 多段裁剪（一个素材裁剪出多段）
- 音画同步（视频 + 音频同步裁剪）

注：核心领域模型已抽离到 packages/domain/trim_config.py，
本模块保留薄包装层，确保向后兼容。
"""

from __future__ import annotations

import logging
from typing import Any

from packages.domain.trim_config import (  # noqa: F401 — 向后兼容
    MIN_TRIM_DURATION,
    TrimConfig,
    TrimSegment,
    build_audio_trim_filter as _build_audio_trim_filter,
    build_video_trim_filter as _build_video_trim_filter,
    extract_trim_from_clip_config,
    parse_segments_from_config as _parse_segments_from_config,
    resolve_segments as _resolve_segments,
)

logger = logging.getLogger(__name__)


class TrimEngine:
    """裁剪引擎 — 生成 FFmpeg trim / atrim 滤镜.

    薄包装层，实际逻辑委托给 packages.domain.trim_config。
    """

    @staticmethod
    def build_video_trim_filter(
        input_label: str,
        trim: TrimConfig,
        output_label: str,
    ) -> str:
        """构建视频裁剪滤镜链."""
        return _build_video_trim_filter(input_label, trim, output_label)

    @staticmethod
    def build_audio_trim_filter(
        input_label: str,
        trim: TrimConfig,
        output_label: str,
    ) -> str:
        """构建音频裁剪滤镜链."""
        return _build_audio_trim_filter(input_label, trim, output_label)

    @staticmethod
    def resolve_segments(
        segments: list[TrimSegment],
        asset_duration: float,
    ) -> list[TrimSegment]:
        """解析并钳制多段裁剪配置，过滤无效段."""
        return _resolve_segments(segments, asset_duration)

    @staticmethod
    def parse_segments_from_config(config: dict[str, Any] | None) -> list[TrimSegment]:
        """从 clip config 中解析多段裁剪配置."""
        return _parse_segments_from_config(config)