xiaoxia-saas/packages/domain/video_concat.py

"""视频拼接领域模型 — 纯逻辑，无外部依赖.

抽离自 concat_engine.py 的数据类和配置解析逻辑，
方便单测覆盖，同时保持向后兼容。
"""

from __future__ import annotations

import logging
from dataclasses import dataclass, field
from typing import Any

logger = logging.getLogger(__name__)

# ── 常量 ──────────────────────────────────────────────────────────────────────

MAX_CONCAT_SEGMENTS = 50  # 最大拼接段数（安全上限，防止OOM）

ALLOWED_VIDEO_EXTENSIONS = {".mp4", ".mov", ".avi", ".mkv", ".webm", ".flv", ".wmv"}

# concat demuxer 要求一致的参数列表
CONCAT_DEMUXER_REQUIRED_PARAMS = [
    "codec_name",
    "width",
    "height",
    "r_frame_rate",
    "pix_fmt",
    "sample_rate",
    "channels",
    "audio_codec",
]


# ── 拼接片段配置 ──────────────────────────────────────────────────────────────


@dataclass
class ConcatSegment:
    """单个拼接片段."""

    video_path: str  # 视频文件路径
    start_time: float = 0.0  # 开始时间（秒）
    duration: float = 0.0  # 持续时长（秒），0表示取到末尾
    has_audio: bool = True  # 是否包含音频

    @classmethod
    def from_dict(cls, seg: dict[str, Any] | None) -> "ConcatSegment":
        """从字典创建拼接片段，带安全类型转换."""
        if not seg or not isinstance(seg, dict):
            return cls(video_path="")

        try:
            start_time = max(0.0, float(seg.get("start_time", 0.0)))
        except (TypeError, ValueError):
            start_time = 0.0

        try:
            duration = max(0.0, float(seg.get("duration", 0.0)))
        except (TypeError, ValueError):
            duration = 0.0

        return cls(
            video_path=str(seg.get("video_path", "")),
            start_time=start_time,
            duration=duration,
            has_audio=bool(seg.get("has_audio", True)),
        )

    @property
    def is_valid(self) -> bool:
        """是否为有效片段（有视频路径）."""
        return bool(self.video_path)

    @property
    def effective_duration(self) -> float:
        """有效时长（duration > 0 时取 duration，否则 0）."""
        return max(0.0, self.duration)


@dataclass
class ConcatConfig:
    """视频拼接配置."""

    segments: list[ConcatSegment] = field(default_factory=list)
    output_width: int = 0  # 输出宽度（0=自动取第一段）
    output_height: int = 0  # 输出高度（0=自动取第一段）
    output_fps: float = 0.0  # 输出帧率（0=自动取第一段）
    force_reencode: bool = False  # 强制重新编码
    transition: str = "none"  # 转场效果（none/crossfade）
    transition_duration: float = 0.3  # 转场时长

    @classmethod
    def from_config_dict(cls, config: dict[str, Any] | None) -> "ConcatConfig":
        """从配置字典创建 ConcatConfig."""
        if not config or not isinstance(config, dict):
            return cls()

        segments_raw = config.get("segments", [])
        segments: list[ConcatSegment] = []

        if isinstance(segments_raw, list):
            for s in segments_raw:
                if isinstance(s, dict) and s.get("video_path"):
                    try:
                        seg = ConcatSegment.from_dict(s)
                        if seg.is_valid:
                            segments.append(seg)
                    except Exception:
                        logger.warning("[concat] skip invalid segment: %s", s)
                        continue

        try:
            output_width = max(0, int(config.get("output_width", 0)))
        except (TypeError, ValueError):
            output_width = 0

        try:
            output_height = max(0, int(config.get("output_height", 0)))
        except (TypeError, ValueError):
            output_height = 0

        try:
            output_fps = max(0.0, float(config.get("output_fps", 0.0)))
        except (TypeError, ValueError):
            output_fps = 0.0

        try:
            transition_duration = max(0.1, float(config.get("transition_duration", 0.3)))
        except (TypeError, ValueError):
            transition_duration = 0.3

        return cls(
            segments=segments,
            output_width=output_width,
            output_height=output_height,
            output_fps=output_fps,
            force_reencode=bool(config.get("force_reencode", False)),
            transition=str(config.get("transition", "none")),
            transition_duration=transition_duration,
        )

    @property
    def has_effect(self) -> bool:
        """是否有有效片段需要拼接（至少2段）."""
        return self.valid_segment_count >= 2

    @property
    def valid_segment_count(self) -> int:
        """有效片段数量."""
        return sum(1 for s in self.segments if s.is_valid)

    @property
    def total_segments(self) -> int:
        """有效片段数量（向后兼容别名）."""
        return self.valid_segment_count

    @property
    def first_valid_segment(self) -> ConcatSegment | None:
        """第一个有效片段."""
        for s in self.segments:
            if s.is_valid:
                return s
        return None

    @property
    def estimated_total_duration(self) -> float:
        """估算总时长（只统计有明确duration的片段）."""
        total = 0.0
        for s in self.segments:
            if s.is_valid and s.duration > 0:
                total += s.duration
        return total

    def clamp_segments(self, max_segments: int = MAX_CONCAT_SEGMENTS) -> None:
        """截断片段数量，防止OOM."""
        if len(self.segments) > max_segments:
            self.segments = self.segments[:max_segments]