test(wave116): 抽离trim_config领域模型 + 64单测

- 抽离TrimConfig/TrimSegment数据类、三选二推导、边界钳制、多段裁剪、滤镜构建到packages/domain/trim_config.py
- trim_engine.py: 339→67行 (-80%)，保留TrimEngine薄包装类+模块级向后兼容导入
- 64个纯逻辑单测全绿，原有34个测试无回归
- 合计98 passed

apps/worker/video_processing/trim_engine.py

@@ -5,169 +5,35 @@
 - 边界自动钳制（超出素材时长自动修正，不阻断渲染）
 - 多段裁剪（一个素材裁剪出多段）
 - 音画同步（视频 + 音频同步裁剪）
+
+注：核心领域模型已抽离到 packages/domain/trim_config.py，
+本模块保留薄包装层，确保向后兼容。
 """
 
 from __future__ import annotations
 
 import logging
-from dataclasses import dataclass
 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__)
 
-# 最小裁剪时长（秒），低于此值视为无效
-MIN_TRIM_DURATION = 0.1
-
-
-@dataclass
-class TrimConfig:
-    """裁剪配置.
-
-    三选二规则：start_time / end_time / duration 中必须至少给出两个，
-    第三个会被自动推导。如果三个都给了，以 start_time + duration 为准。
-
-    边界保护：
-    - start_time < 0 → 钳制到 0
-    - end_time > 素材时长 → 钳制到素材时长
-    - 计算出的 duration < 最小阈值 → 标记为无效
-    """
-
-    start_time: float = 0.0  # 入点（素材内时间，秒）
-    end_time: float = 0.0  # 出点（素材内时间，秒），0 表示未指定
-    duration: float = 0.0  # 裁剪时长（秒），0 表示未指定
-
-    @classmethod
-    def from_dict(cls, data: dict[str, Any] | None) -> TrimConfig | None:
-        """从字典构造，无有效裁剪参数时返回 None（不裁剪）."""
-        if not data:
-            return None
-
-        start = float(data.get("start_time", 0) or 0)
-        end = float(data.get("end_time", 0) or 0)
-        dur = float(data.get("duration", 0) or 0)
-
-        # 三个参数都没有 → 不裁剪
-        if start <= 0 and end <= 0 and dur <= 0:
-            return None
-
-        # 至少有两个参数（或一个合理的 start/duration）
-        # 兼容：只传了 start_time → 从 start 开始取到末尾
-        # 兼容：只传了 duration → 从 0 开始取 duration
-        if start > 0 and end <= 0 and dur <= 0:
-            # 只有 start，取到末尾 → 这是"从某点开始"的语义，算有效
-            pass
-        elif dur > 0 and start <= 0 and end <= 0:
-            # 只有 duration → 从开头取 duration，算有效
-            pass
-        elif start <= 0 and end <= 0 and dur <= 0:
-            return None
-
-        return cls(start_time=start, end_time=end, duration=dur)
-
-    def validate_and_resolve(self, asset_duration: float) -> TrimConfig:
-        """根据素材实际时长，解析并钳制裁剪参数.
-
-        返回一个新的 TrimConfig，其中 start_time / end_time / duration 都已确定。
-        如果裁剪无效（时长为0或负数），仍返回但调用方应检查 is_valid。
-        """
-        start = self.start_time
-        end = self.end_time
-        dur = self.duration
-
-        # 边界：start 不能为负
-        if start < 0:
-            start = 0.0
-
-        # 边界：asset_duration 为 0 时保守处理（不裁剪，取全部）
-        if asset_duration <= 0:
-            return TrimConfig(start_time=0.0, end_time=0.0, duration=0.0)
-
-        # 三选二推导
-        # 判断顺序很重要：先判断需要两个显式值的组合，最后判断含默认值的
-        # 情况1：start + end 都有显式值
-        if start > 0 and end > 0:
-            if end <= start:
-                # 出点 <= 入点，无效 → 返回 start 处一个极短片段（调用方会判无效）
-                return TrimConfig(start_time=start, end_time=start, duration=0.0)
-            dur = end - start
-        # 情况2：end + duration 都有显式值
-        elif end > 0 and dur > 0:
-            start = end - dur
-            if start < 0:
-                start = 0.0
-                dur = end  # 重新计算
-        # 情况3：start + duration 都有值（start 可以是 0）
-        elif dur > 0:
-            end = start + dur
-        # 情况4：只有 start → 取到素材末尾
-        elif start > 0 and end <= 0 and dur <= 0:
-            end = asset_duration
-            dur = end - start
-        # 情况5：只有 end → 从开头取到 end
-        elif end > 0 and start <= 0 and dur <= 0:
-            start = 0.0
-            dur = end
-        else:
-            # 都没有 → 不裁剪
-            return TrimConfig(start_time=0.0, end_time=0.0, duration=0.0)
-
-        # 边界钳制：end 不能超过素材时长
-        if end > asset_duration:
-            end = asset_duration
-            dur = end - start
-
-        # 边界钳制：start 不能超过素材时长
-        if start >= asset_duration:
-            start = max(0.0, asset_duration - MIN_TRIM_DURATION)
-            dur = asset_duration - start
-            end = asset_duration
-
-        # 保证 duration 不为负
-        if dur < 0:
-            dur = 0.0
-
-        return TrimConfig(start_time=start, end_time=end, duration=dur)
-
-    @property
-    def is_valid(self) -> bool:
-        """裁剪是否有效（时长大于最小阈值）."""
-        return self.duration >= MIN_TRIM_DURATION
-
-    @property
-    def is_noop(self) -> bool:
-        """是否等价于不裁剪（从0开始取全部）."""
-        return self.start_time <= 0 and self.duration <= 0
-
-    @property
-    def trim_from_start(self) -> bool:
-        """是否从开头裁剪（start_time == 0）."""
-        return self.start_time <= 0
-
-
-@dataclass
-class TrimSegment:
-    """多段裁剪中的一段."""
-
-    segment_id: str  # 段 ID（用于生成唯一标签）
-    trim: TrimConfig  # 裁剪配置
-    order: int = 0  # 排序
-
-    @classmethod
-    def from_dict(cls, data: dict[str, Any], default_order: int = 0) -> TrimSegment:
-        """从字典构造."""
-        return cls(
-            segment_id=str(data.get("segment_id", "") or f"seg_{default_order}"),
-            trim=TrimConfig(
-                start_time=float(data.get("start_time", 0) or 0),
-                end_time=float(data.get("end_time", 0) or 0),
-                duration=float(data.get("duration", 0) or 0),
-            ),
-            order=int(data.get("order", default_order)),
-        )
-
 
 class TrimEngine:
-    """裁剪引擎 — 生成 FFmpeg trim / atrim 滤镜."""
+    """裁剪引擎 — 生成 FFmpeg trim / atrim 滤镜.
+
+    薄包装层，实际逻辑委托给 packages.domain.trim_config。
+    """
 
     @staticmethod
     def build_video_trim_filter(
@@ -175,38 +41,8 @@ class TrimEngine:
         trim: TrimConfig,
         output_label: str,
     ) -> str:
-        """构建视频裁剪滤镜链.
-
-        Args:
-            input_label: 输入视频标签，如 "[0:v]"
-            trim: 裁剪配置（已解析钳制）
-            output_label: 输出视频标签，如 "[v0_trimmed]"
-
-        Returns:
-            FFmpeg filter 字符串，如 "[0:v]trim=start=10:duration=5,setpts=PTS-STARTPTS[v0_trimmed]"
-        """
-        if trim.is_noop:
-            # 不裁剪，直接直通（仅重置时间戳）
-            return f"{input_label}setpts=PTS-STARTPTS{output_label}"
-
-        parts: list[str] = []
-
-        # trim 滤镜参数
-        trim_args: list[str] = []
-        if trim.start_time > 0:
-            trim_args.append(f"start={trim.start_time:.3f}")
-        if trim.duration > 0:
-            trim_args.append(f"duration={trim.duration:.3f}")
-        elif trim.end_time > 0:
-            # end 用 duration 表示（start 到 end 的时长）
-            # 但 validate_and_resolve 后应该已经有 duration 了
-            pass
-
-        parts.append(f"trim={':'.join(trim_args)}")
-        parts.append("setpts=PTS-STARTPTS")
-
-        filter_str = f"{input_label}{','.join(parts)}{output_label}"
-        return filter_str
+        """构建视频裁剪滤镜链."""
+        return _build_video_trim_filter(input_label, trim, output_label)
 
     @staticmethod
     def build_audio_trim_filter(
@@ -214,126 +50,18 @@ class TrimEngine:
         trim: TrimConfig,
         output_label: str,
     ) -> str:
-        """构建音频裁剪滤镜链.
-
-        Args:
-            input_label: 输入音频标签，如 "[0:a]"
-            trim: 裁剪配置（已解析钳制）
-            output_label: 输出音频标签，如 "[a0_trimmed]"
-
-        Returns:
-            FFmpeg filter 字符串，如 "[0:a]atrim=start=10:duration=5,asetpts=PTS-STARTPTS[a0_trimmed]"
-        """
-        if trim.is_noop:
-            return f"{input_label}asetpts=PTS-STARTPTS{output_label}"
-
-        parts: list[str] = []
-
-        trim_args: list[str] = []
-        if trim.start_time > 0:
-            trim_args.append(f"start={trim.start_time:.3f}")
-        if trim.duration > 0:
-            trim_args.append(f"duration={trim.duration:.3f}")
-
-        parts.append(f"atrim={':'.join(trim_args)}")
-        parts.append("asetpts=PTS-STARTPTS")
-
-        filter_str = f"{input_label}{','.join(parts)}{output_label}"
-        return filter_str
+        """构建音频裁剪滤镜链."""
+        return _build_audio_trim_filter(input_label, trim, output_label)
 
     @staticmethod
     def resolve_segments(
         segments: list[TrimSegment],
         asset_duration: float,
     ) -> list[TrimSegment]:
-        """解析并钳制多段裁剪配置，过滤无效段.
-
-        Args:
-            segments: 原始段列表
-            asset_duration: 素材实际时长
-
-        Returns:
-            解析后的有效段列表，按 order 排序
-        """
-        resolved: list[TrimSegment] = []
-        for i, seg in enumerate(segments):
-            resolved_trim = seg.trim.validate_and_resolve(asset_duration)
-            if not resolved_trim.is_valid:
-                logger.warning("裁剪段无效，跳过: segment_id=%s duration=%.3f", seg.segment_id, resolved_trim.duration)
-                continue
-            resolved.append(
-                TrimSegment(
-                    segment_id=seg.segment_id,
-                    trim=resolved_trim,
-                    order=seg.order if seg.order >= 0 else i,
-                )
-            )
-
-        resolved.sort(key=lambda s: s.order)
-        return resolved
+        """解析并钳制多段裁剪配置，过滤无效段."""
+        return _resolve_segments(segments, asset_duration)
 
     @staticmethod
     def parse_segments_from_config(config: dict[str, Any] | None) -> list[TrimSegment]:
-        """从 clip config 中解析多段裁剪配置.
-
-        config 中支持：
-        - trim_segments: [ {segment_id, start_time, end_time, duration, order}, ... ]
-        - trim_start / trim_end / trim_duration: 单段裁剪（兼容旧格式）
-        """
-        if not config:
-            return []
-
-        # 优先解析多段
-        raw_segments = config.get("trim_segments", [])
-        if raw_segments and isinstance(raw_segments, list):
-            segments = []
-            for i, raw in enumerate(raw_segments):
-                if isinstance(raw, dict):
-                    segments.append(TrimSegment.from_dict(raw, default_order=i))
-            return segments
-
-        # 单段裁剪兼容：从 trim_start/trim_end/trim_duration 构造
-        has_single = any(k in config for k in ("trim_start", "trim_end", "trim_duration"))
-        if has_single:
-            seg = TrimSegment(
-                segment_id="main",
-                trim=TrimConfig(
-                    start_time=float(config.get("trim_start", 0) or 0),
-                    end_time=float(config.get("trim_end", 0) or 0),
-                    duration=float(config.get("trim_duration", 0) or 0),
-                ),
-                order=0,
-            )
-            return [seg]
-
-        return []
-
-
-# ── 工具函数 ──────────────────────────────────────────────────────────────────
-
-
-def extract_trim_from_clip_config(config: dict[str, Any] | None) -> TrimConfig | None:
-    """从 clip config 中提取单段裁剪配置.
-
-    兼容以下字段名：
-    - trim_start / trim_end / trim_duration
-    - start_time / end_time / duration（在 trim 子字典里）
-    """
-    if not config:
-        return None
-
-    # trim 子字典
-    if "trim" in config and isinstance(config["trim"], dict):
-        return TrimConfig.from_dict(config["trim"])
-
-    # 扁平字段
-    has_any = any(k in config for k in ("trim_start", "trim_end", "trim_duration"))
-    if not has_any:
-        return None
-
-    data = {
-        "start_time": config.get("trim_start", 0),
-        "end_time": config.get("trim_end", 0),
-        "duration": config.get("trim_duration", 0),
-    }
-    return TrimConfig.from_dict(data)
+        """从 clip config 中解析多段裁剪配置."""
+        return _parse_segments_from_config(config)

packages/domain/trim_config.py

@@ -0,0 +1,352 @@
+"""裁剪配置领域模型 — 纯逻辑，无FFmpeg依赖.
+
+抽离自 trim_engine.py，包含：
+- TrimConfig 数据类（三选二推导 + 边界钳制 + 有效性判断）
+- TrimSegment 数据类（多段裁剪）
+- 滤镜字符串构建（build_video_trim_filter / build_audio_trim_filter）
+- 多段解析（resolve_segments / parse_segments_from_config）
+- 工具函数（extract_trim_from_clip_config）
+"""
+
+from __future__ import annotations
+
+import logging
+from dataclasses import dataclass, field
+from typing import Any
+
+logger = logging.getLogger(__name__)
+
+
+# ── 常量 ────────────────────────────────────────────────────────────────────
+
+# 最小裁剪时长（秒），低于此值视为无效
+MIN_TRIM_DURATION = 0.1
+
+
+# ── 数据类 ──────────────────────────────────────────────────────────────────
+
+
+@dataclass
+class TrimConfig:
+    """裁剪配置.
+
+    三选二规则：start_time / end_time / duration 中必须至少给出两个，
+    第三个会被自动推导。如果三个都给了，以 start_time + duration 为准。
+
+    边界保护：
+    - start_time < 0 → 钳制到 0
+    - end_time > 素材时长 → 钳制到素材时长
+    - 计算出的 duration < 最小阈值 → 标记为无效
+    """
+
+    start_time: float = 0.0  # 入点（素材内时间，秒）
+    end_time: float = 0.0  # 出点（素材内时间，秒），0 表示未指定
+    duration: float = 0.0  # 裁剪时长（秒），0 表示未指定
+
+    @classmethod
+    def from_dict(cls, data: dict[str, Any] | None) -> TrimConfig | None:
+        """从字典构造，无有效裁剪参数时返回 None（不裁剪）."""
+        if not data:
+            return None
+
+        start = float(data.get("start_time", 0) or 0)
+        end = float(data.get("end_time", 0) or 0)
+        dur = float(data.get("duration", 0) or 0)
+
+        # 三个参数都没有 → 不裁剪
+        if start <= 0 and end <= 0 and dur <= 0:
+            return None
+
+        # 至少有两个参数（或一个合理的 start/duration）
+        # 兼容：只传了 start_time → 从 start 开始取到末尾
+        # 兼容：只传了 duration → 从 0 开始取 duration
+        if start > 0 and end <= 0 and dur <= 0:
+            # 只有 start，取到末尾 → 这是"从某点开始"的语义，算有效
+            pass
+        elif dur > 0 and start <= 0 and end <= 0:
+            # 只有 duration → 从开头取 duration，算有效
+            pass
+        elif start <= 0 and end <= 0 and dur <= 0:
+            return None
+
+        return cls(start_time=start, end_time=end, duration=dur)
+
+    def validate_and_resolve(self, asset_duration: float) -> TrimConfig:
+        """根据素材实际时长，解析并钳制裁剪参数.
+
+        返回一个新的 TrimConfig，其中 start_time / end_time / duration 都已确定。
+        如果裁剪无效（时长为0或负数），仍返回但调用方应检查 is_valid。
+        """
+        start = self.start_time
+        end = self.end_time
+        dur = self.duration
+
+        # 边界：start 不能为负
+        if start < 0:
+            start = 0.0
+
+        # 边界：asset_duration 为 0 时保守处理（不裁剪，取全部）
+        if asset_duration <= 0:
+            return TrimConfig(start_time=0.0, end_time=0.0, duration=0.0)
+
+        # 三选二推导
+        # 判断顺序很重要：先判断需要两个显式值的组合，最后判断含默认值的
+        # 情况1：start + end 都有显式值
+        if start > 0 and end > 0:
+            if end <= start:
+                # 出点 <= 入点，无效 → 返回 start 处一个极短片段（调用方会判无效）
+                return TrimConfig(start_time=start, end_time=start, duration=0.0)
+            dur = end - start
+        # 情况2：end + duration 都有显式值
+        elif end > 0 and dur > 0:
+            start = end - dur
+            if start < 0:
+                start = 0.0
+                dur = end  # 重新计算
+        # 情况3：start + duration 都有值（start 可以是 0）
+        elif dur > 0:
+            end = start + dur
+        # 情况4：只有 start → 取到素材末尾
+        elif start > 0 and end <= 0 and dur <= 0:
+            end = asset_duration
+            dur = end - start
+        # 情况5：只有 end → 从开头取到 end
+        elif end > 0 and start <= 0 and dur <= 0:
+            start = 0.0
+            dur = end
+        else:
+            # 都没有 → 不裁剪
+            return TrimConfig(start_time=0.0, end_time=0.0, duration=0.0)
+
+        # 边界钳制：end 不能超过素材时长
+        if end > asset_duration:
+            end = asset_duration
+            dur = end - start
+
+        # 边界钳制：start 不能超过素材时长
+        if start >= asset_duration:
+            start = max(0.0, asset_duration - MIN_TRIM_DURATION)
+            dur = asset_duration - start
+            end = asset_duration
+
+        # 保证 duration 不为负
+        if dur < 0:
+            dur = 0.0
+
+        return TrimConfig(start_time=start, end_time=end, duration=dur)
+
+    @property
+    def is_valid(self) -> bool:
+        """裁剪是否有效（时长大于最小阈值）."""
+        return self.duration >= MIN_TRIM_DURATION
+
+    @property
+    def is_noop(self) -> bool:
+        """是否等价于不裁剪（从0开始取全部）."""
+        return self.start_time <= 0 and self.duration <= 0
+
+    @property
+    def trim_from_start(self) -> bool:
+        """是否从开头裁剪（start_time == 0）."""
+        return self.start_time <= 0
+
+
+@dataclass
+class TrimSegment:
+    """多段裁剪中的一段."""
+
+    segment_id: str  # 段 ID（用于生成唯一标签）
+    trim: TrimConfig  # 裁剪配置
+    order: int = 0  # 排序
+
+    @classmethod
+    def from_dict(cls, data: dict[str, Any], default_order: int = 0) -> TrimSegment:
+        """从字典构造."""
+        return cls(
+            segment_id=str(data.get("segment_id", "") or f"seg_{default_order}"),
+            trim=TrimConfig(
+                start_time=float(data.get("start_time", 0) or 0),
+                end_time=float(data.get("end_time", 0) or 0),
+                duration=float(data.get("duration", 0) or 0),
+            ),
+            order=int(data.get("order", default_order)),
+        )
+
+
+# ── 滤镜构建 ────────────────────────────────────────────────────────────────
+
+
+def build_video_trim_filter(
+    input_label: str,
+    trim: TrimConfig,
+    output_label: str,
+) -> str:
+    """构建视频裁剪滤镜链.
+
+    Args:
+        input_label: 输入视频标签，如 "[0:v]"
+        trim: 裁剪配置（已解析钳制）
+        output_label: 输出视频标签，如 "[v0_trimmed]"
+
+    Returns:
+        FFmpeg filter 字符串，如 "[0:v]trim=start=10:duration=5,setpts=PTS-STARTPTS[v0_trimmed]"
+    """
+    if trim.is_noop:
+        # 不裁剪，直接直通（仅重置时间戳）
+        return f"{input_label}setpts=PTS-STARTPTS{output_label}"
+
+    parts: list[str] = []
+
+    # trim 滤镜参数
+    trim_args: list[str] = []
+    if trim.start_time > 0:
+        trim_args.append(f"start={trim.start_time:.3f}")
+    if trim.duration > 0:
+        trim_args.append(f"duration={trim.duration:.3f}")
+    elif trim.end_time > 0:
+        # end 用 duration 表示（start 到 end 的时长）
+        # 但 validate_and_resolve 后应该已经有 duration 了
+        pass
+
+    parts.append(f"trim={':'.join(trim_args)}")
+    parts.append("setpts=PTS-STARTPTS")
+
+    filter_str = f"{input_label}{','.join(parts)}{output_label}"
+    return filter_str
+
+
+def build_audio_trim_filter(
+    input_label: str,
+    trim: TrimConfig,
+    output_label: str,
+) -> str:
+    """构建音频裁剪滤镜链.
+
+    Args:
+        input_label: 输入音频标签，如 "[0:a]"
+        trim: 裁剪配置（已解析钳制）
+        output_label: 输出音频标签，如 "[a0_trimmed]"
+
+    Returns:
+        FFmpeg filter 字符串，如 "[0:a]atrim=start=10:duration=5,asetpts=PTS-STARTPTS[a0_trimmed]"
+    """
+    if trim.is_noop:
+        return f"{input_label}asetpts=PTS-STARTPTS{output_label}"
+
+    parts: list[str] = []
+
+    trim_args: list[str] = []
+    if trim.start_time > 0:
+        trim_args.append(f"start={trim.start_time:.3f}")
+    if trim.duration > 0:
+        trim_args.append(f"duration={trim.duration:.3f}")
+
+    parts.append(f"atrim={':'.join(trim_args)}")
+    parts.append("asetpts=PTS-STARTPTS")
+
+    filter_str = f"{input_label}{','.join(parts)}{output_label}"
+    return filter_str
+
+
+# ── 多段裁剪 ────────────────────────────────────────────────────────────────
+
+
+def resolve_segments(
+    segments: list[TrimSegment],
+    asset_duration: float,
+) -> list[TrimSegment]:
+    """解析并钳制多段裁剪配置，过滤无效段.
+
+    Args:
+        segments: 原始段列表
+        asset_duration: 素材实际时长
+
+    Returns:
+        解析后的有效段列表，按 order 排序
+    """
+    resolved: list[TrimSegment] = []
+    for i, seg in enumerate(segments):
+        resolved_trim = seg.trim.validate_and_resolve(asset_duration)
+        if not resolved_trim.is_valid:
+            logger.warning(
+                "裁剪段无效，跳过: segment_id=%s duration=%.3f",
+                seg.segment_id,
+                resolved_trim.duration,
+            )
+            continue
+        resolved.append(
+            TrimSegment(
+                segment_id=seg.segment_id,
+                trim=resolved_trim,
+                order=seg.order if seg.order >= 0 else i,
+            )
+        )
+
+    resolved.sort(key=lambda s: s.order)
+    return resolved
+
+
+def parse_segments_from_config(config: dict[str, Any] | None) -> list[TrimSegment]:
+    """从 clip config 中解析多段裁剪配置.
+
+    config 中支持：
+    - trim_segments: [ {segment_id, start_time, end_time, duration, order}, ... ]
+    - trim_start / trim_end / trim_duration: 单段裁剪（兼容旧格式）
+    """
+    if not config:
+        return []
+
+    # 优先解析多段
+    raw_segments = config.get("trim_segments", [])
+    if raw_segments and isinstance(raw_segments, list):
+        segments = []
+        for i, raw in enumerate(raw_segments):
+            if isinstance(raw, dict):
+                segments.append(TrimSegment.from_dict(raw, default_order=i))
+        return segments
+
+    # 单段裁剪兼容：从 trim_start/trim_end/trim_duration 构造
+    has_single = any(k in config for k in ("trim_start", "trim_end", "trim_duration"))
+    if has_single:
+        seg = TrimSegment(
+            segment_id="main",
+            trim=TrimConfig(
+                start_time=float(config.get("trim_start", 0) or 0),
+                end_time=float(config.get("trim_end", 0) or 0),
+                duration=float(config.get("trim_duration", 0) or 0),
+            ),
+            order=0,
+        )
+        return [seg]
+
+    return []
+
+
+# ── 工具函数 ────────────────────────────────────────────────────────────────
+
+
+def extract_trim_from_clip_config(config: dict[str, Any] | None) -> TrimConfig | None:
+    """从 clip config 中提取单段裁剪配置.
+
+    兼容以下字段名：
+    - trim_start / trim_end / trim_duration
+    - start_time / end_time / duration（在 trim 子字典里）
+    """
+    if not config:
+        return None
+
+    # trim 子字典
+    if "trim" in config and isinstance(config["trim"], dict):
+        return TrimConfig.from_dict(config["trim"])
+
+    # 扁平字段
+    has_any = any(k in config for k in ("trim_start", "trim_end", "trim_duration"))
+    if not has_any:
+        return None
+
+    data = {
+        "start_time": config.get("trim_start", 0),
+        "end_time": config.get("trim_end", 0),
+        "duration": config.get("trim_duration", 0),
+    }
+    return TrimConfig.from_dict(data)

tests/unit/test_trim_config.py

@@ -0,0 +1,423 @@
+"""trim_config 领域模型单测."""
+
+from __future__ import annotations
+
+import pytest
+
+from packages.domain.trim_config import (
+    MIN_TRIM_DURATION,
+    TrimConfig,
+    TrimSegment,
+    build_audio_trim_filter,
+    build_video_trim_filter,
+    extract_trim_from_clip_config,
+    parse_segments_from_config,
+    resolve_segments,
+)
+
+# ── TrimConfig.from_dict 测试 ─────────────────────────────────────────────
+
+
+class TestTrimConfigFromDict:
+    def test_none_returns_none(self):
+        assert TrimConfig.from_dict(None) is None
+
+    def test_empty_dict_returns_none(self):
+        assert TrimConfig.from_dict({}) is None
+
+    def test_all_zero_returns_none(self):
+        assert TrimConfig.from_dict({"start_time": 0, "end_time": 0, "duration": 0}) is None
+
+    def test_start_only_valid(self):
+        cfg = TrimConfig.from_dict({"start_time": 5.0})
+        assert cfg is not None
+        assert cfg.start_time == 5.0
+        assert cfg.end_time == 0
+        assert cfg.duration == 0
+
+    def test_duration_only_valid(self):
+        cfg = TrimConfig.from_dict({"duration": 10.0})
+        assert cfg is not None
+        assert cfg.duration == 10.0
+        assert cfg.start_time == 0
+
+    def test_start_and_duration(self):
+        cfg = TrimConfig.from_dict({"start_time": 2.0, "duration": 5.0})
+        assert cfg is not None
+        assert cfg.start_time == 2.0
+        assert cfg.duration == 5.0
+
+    def test_start_and_end(self):
+        cfg = TrimConfig.from_dict({"start_time": 1.0, "end_time": 5.0})
+        assert cfg is not None
+        assert cfg.start_time == 1.0
+        assert cfg.end_time == 5.0
+
+    def test_end_only(self):
+        cfg = TrimConfig.from_dict({"end_time": 8.0})
+        assert cfg is not None
+        assert cfg.end_time == 8.0
+
+    def test_string_values_coerced(self):
+        cfg = TrimConfig.from_dict({"start_time": "3.5", "duration": "2.0"})
+        assert cfg is not None
+        assert cfg.start_time == 3.5
+        assert cfg.duration == 2.0
+
+    def test_falsy_values_treated_as_zero(self):
+        cfg = TrimConfig.from_dict({"start_time": None, "duration": None})
+        assert cfg is None
+
+    def test_default_values(self):
+        cfg = TrimConfig()
+        assert cfg.start_time == 0.0
+        assert cfg.end_time == 0.0
+        assert cfg.duration == 0.0
+
+
+# ── validate_and_resolve 测试 ─────────────────────────────────────────────
+
+
+class TestValidateAndResolve:
+    def test_start_and_end_resolves_duration(self):
+        cfg = TrimConfig(start_time=2.0, end_time=7.0)
+        resolved = cfg.validate_and_resolve(100.0)
+        assert resolved.start_time == 2.0
+        assert resolved.end_time == 7.0
+        assert resolved.duration == 5.0
+
+    def test_start_and_duration_resolves_end(self):
+        cfg = TrimConfig(start_time=3.0, duration=10.0)
+        resolved = cfg.validate_and_resolve(100.0)
+        assert resolved.start_time == 3.0
+        assert resolved.duration == 10.0
+        assert resolved.end_time == 13.0
+
+    def test_end_and_duration_resolves_start(self):
+        cfg = TrimConfig(end_time=15.0, duration=5.0)
+        resolved = cfg.validate_and_resolve(100.0)
+        assert resolved.end_time == 15.0
+        assert resolved.duration == 5.0
+        assert resolved.start_time == 10.0
+
+    def test_start_only_takes_to_end(self):
+        cfg = TrimConfig(start_time=5.0)
+        resolved = cfg.validate_and_resolve(30.0)
+        assert resolved.start_time == 5.0
+        assert resolved.end_time == 30.0
+        assert resolved.duration == 25.0
+
+    def test_end_only_takes_from_start(self):
+        cfg = TrimConfig(end_time=8.0)
+        resolved = cfg.validate_and_resolve(30.0)
+        assert resolved.start_time == 0.0
+        assert resolved.end_time == 8.0
+        assert resolved.duration == 8.0
+
+    def test_duration_only_from_zero(self):
+        cfg = TrimConfig(duration=10.0)
+        resolved = cfg.validate_and_resolve(30.0)
+        assert resolved.start_time == 0.0
+        assert resolved.duration == 10.0
+        assert resolved.end_time == 10.0
+
+    def test_negative_start_clamped(self):
+        cfg = TrimConfig(start_time=-5.0, duration=10.0)
+        resolved = cfg.validate_and_resolve(30.0)
+        assert resolved.start_time == 0.0
+
+    def test_end_exceeds_asset_clamped(self):
+        cfg = TrimConfig(start_time=5.0, duration=50.0)
+        resolved = cfg.validate_and_resolve(30.0)
+        assert resolved.end_time == 30.0
+        assert resolved.duration == 25.0
+
+    def test_start_exceeds_asset_clamped(self):
+        cfg = TrimConfig(start_time=50.0, duration=10.0)
+        resolved = cfg.validate_and_resolve(30.0)
+        assert resolved.start_time < 30.0
+        assert resolved.end_time == 30.0
+
+    def test_end_before_start_invalid(self):
+        cfg = TrimConfig(start_time=10.0, end_time=5.0)
+        resolved = cfg.validate_and_resolve(30.0)
+        assert resolved.duration == 0.0
+        assert resolved.is_valid is False
+
+    def test_zero_asset_duration(self):
+        cfg = TrimConfig(start_time=1.0, duration=5.0)
+        resolved = cfg.validate_and_resolve(0.0)
+        assert resolved.is_noop
+
+    def test_negative_asset_duration(self):
+        cfg = TrimConfig(start_time=1.0, duration=5.0)
+        resolved = cfg.validate_and_resolve(-1.0)
+        assert resolved.is_noop
+
+    def test_end_and_duration_with_negative_start(self):
+        cfg = TrimConfig(end_time=3.0, duration=10.0)
+        resolved = cfg.validate_and_resolve(30.0)
+        assert resolved.start_time == 0.0
+        assert resolved.end_time == 3.0
+        assert resolved.duration == 3.0
+
+    def test_all_three_params_uses_start_duration(self):
+        cfg = TrimConfig(start_time=2.0, end_time=8.0, duration=3.0)
+        resolved = cfg.validate_and_resolve(30.0)
+        # 有 start + end 时应该用 start+end 推导 duration
+        assert resolved.start_time == 2.0
+        assert resolved.end_time == 8.0
+        assert resolved.duration == 6.0
+
+    def test_empty_config_returns_noop(self):
+        cfg = TrimConfig()
+        resolved = cfg.validate_and_resolve(30.0)
+        assert resolved.is_noop
+
+
+# ── is_valid / is_noop / trim_from_start 测试 ─────────────────────────────
+
+
+class TestProperties:
+    def test_is_valid_true_for_normal(self):
+        cfg = TrimConfig(start_time=0, end_time=0, duration=5.0)
+        assert cfg.is_valid is True
+
+    def test_is_valid_false_for_zero(self):
+        cfg = TrimConfig(duration=0.0)
+        assert cfg.is_valid is False
+
+    def test_is_valid_false_for_very_small(self):
+        cfg = TrimConfig(duration=0.01)
+        assert cfg.is_valid is False
+
+    def test_is_valid_true_at_boundary(self):
+        cfg = TrimConfig(duration=MIN_TRIM_DURATION)
+        assert cfg.is_valid is True
+
+    def test_is_noop_true_for_default(self):
+        cfg = TrimConfig()
+        assert cfg.is_noop is True
+
+    def test_is_noop_false_with_start(self):
+        cfg = TrimConfig(start_time=1.0)
+        assert cfg.is_noop is False
+
+    def test_is_noop_false_with_duration(self):
+        cfg = TrimConfig(duration=1.0)
+        assert cfg.is_noop is False
+
+    def test_trim_from_start_true(self):
+        cfg = TrimConfig(start_time=0.0, duration=5.0)
+        assert cfg.trim_from_start is True
+
+    def test_trim_from_start_false(self):
+        cfg = TrimConfig(start_time=2.0, duration=5.0)
+        assert cfg.trim_from_start is False
+
+
+# ── TrimSegment 测试 ───────────────────────────────────────────────────────
+
+
+class TestTrimSegment:
+    def test_from_dict_basic(self):
+        seg = TrimSegment.from_dict({"segment_id": "s1", "start_time": 1.0, "duration": 3.0})
+        assert seg.segment_id == "s1"
+        assert seg.trim.start_time == 1.0
+        assert seg.trim.duration == 3.0
+        assert seg.order == 0
+
+    def test_from_dict_with_order(self):
+        seg = TrimSegment.from_dict({"segment_id": "s2", "start_time": 0, "end_time": 5.0, "order": 2})
+        assert seg.order == 2
+
+    def test_from_dict_default_order(self):
+        seg = TrimSegment.from_dict({"start_time": 1.0}, default_order=5)
+        assert seg.order == 5
+
+    def test_from_dict_default_segment_id(self):
+        seg = TrimSegment.from_dict({"start_time": 1.0}, default_order=3)
+        assert seg.segment_id == "seg_3"
+
+
+# ── build_video_trim_filter 测试 ───────────────────────────────────────────
+
+
+class TestBuildVideoTrimFilter:
+    def test_noop_returns_setpts(self):
+        cfg = TrimConfig()
+        result = build_video_trim_filter("[0:v]", cfg, "[v]")
+        assert "setpts=PTS-STARTPTS" in result
+        assert "trim=" not in result
+        assert "[0:v]" in result
+        assert "[v]" in result
+
+    def test_with_start_and_duration(self):
+        cfg = TrimConfig(start_time=5.0, end_time=10.0, duration=5.0)
+        result = build_video_trim_filter("[0:v]", cfg, "[out]")
+        assert "trim=" in result
+        assert "start=5.000" in result
+        assert "duration=5.000" in result
+        assert "setpts=PTS-STARTPTS" in result
+
+    def test_contains_input_and_output_labels(self):
+        cfg = TrimConfig(start_time=1.0, duration=2.0)
+        result = build_video_trim_filter("[in_v]", cfg, "[out_v]")
+        assert "[in_v]" in result
+        assert "[out_v]" in result
+
+    def test_duration_only(self):
+        cfg = TrimConfig(duration=3.5)
+        result = build_video_trim_filter("[0:v]", cfg, "[v]")
+        assert "duration=3.500" in result
+        assert "start=" not in result
+
+
+# ── build_audio_trim_filter 测试 ───────────────────────────────────────────
+
+
+class TestBuildAudioTrimFilter:
+    def test_noop_returns_asetpts(self):
+        cfg = TrimConfig()
+        result = build_audio_trim_filter("[0:a]", cfg, "[a]")
+        assert "asetpts=PTS-STARTPTS" in result
+        assert "atrim=" not in result
+
+    def test_with_start_and_duration(self):
+        cfg = TrimConfig(start_time=2.0, end_time=7.0, duration=5.0)
+        result = build_audio_trim_filter("[0:a]", cfg, "[out]")
+        assert "atrim=" in result
+        assert "start=2.000" in result
+        assert "duration=5.000" in result
+        assert "asetpts=PTS-STARTPTS" in result
+
+    def test_contains_input_and_output_labels(self):
+        cfg = TrimConfig(start_time=1.0, duration=2.0)
+        result = build_audio_trim_filter("[in_a]", cfg, "[out_a]")
+        assert "[in_a]" in result
+        assert "[out_a]" in result
+
+
+# ── resolve_segments 测试 ──────────────────────────────────────────────────
+
+
+class TestResolveSegments:
+    def test_empty_list_returns_empty(self):
+        result = resolve_segments([], 30.0)
+        assert result == []
+
+    def test_single_segment(self):
+        segs = [TrimSegment(segment_id="s1", trim=TrimConfig(start_time=1.0, duration=5.0), order=0)]
+        result = resolve_segments(segs, 30.0)
+        assert len(result) == 1
+        assert result[0].segment_id == "s1"
+        assert result[0].trim.duration == 5.0
+
+    def test_invalid_segment_filters_out(self):
+        segs = [
+            TrimSegment(segment_id="good", trim=TrimConfig(start_time=0, duration=5.0), order=0),
+            TrimSegment(
+                segment_id="bad",
+                trim=TrimConfig(start_time=5.0, end_time=5.0),  # end == start → duration 0
+                order=1,
+            ),
+        ]
+        result = resolve_segments(segs, 30.0)
+        assert len(result) == 1
+        assert result[0].segment_id == "good"
+
+    def test_sorted_by_order(self):
+        segs = [
+            TrimSegment(segment_id="s2", trim=TrimConfig(start_time=5.0, duration=3.0), order=2),
+            TrimSegment(segment_id="s1", trim=TrimConfig(start_time=0, duration=3.0), order=1),
+            TrimSegment(segment_id="s0", trim=TrimConfig(start_time=10.0, duration=3.0), order=0),
+        ]
+        result = resolve_segments(segs, 30.0)
+        assert [s.segment_id for s in result] == ["s0", "s1", "s2"]
+
+    def test_negative_order_uses_index(self):
+        segs = [
+            TrimSegment(segment_id="s0", trim=TrimConfig(duration=3.0), order=-1),
+        ]
+        result = resolve_segments(segs, 30.0)
+        assert len(result) == 1
+        assert result[0].order == 0
+
+
+# ── parse_segments_from_config 测试 ────────────────────────────────────────
+
+
+class TestParseSegmentsFromConfig:
+    def test_none_returns_empty(self):
+        assert parse_segments_from_config(None) == []
+
+    def test_empty_dict_returns_empty(self):
+        assert parse_segments_from_config({}) == []
+
+    def test_trim_segments_list(self):
+        config = {
+            "trim_segments": [
+                {"segment_id": "s1", "start_time": 0, "duration": 3.0, "order": 0},
+                {"segment_id": "s2", "start_time": 5.0, "duration": 2.0, "order": 1},
+            ]
+        }
+        result = parse_segments_from_config(config)
+        assert len(result) == 2
+        assert result[0].segment_id == "s1"
+        assert result[1].segment_id == "s2"
+
+    def test_trim_segments_skips_non_dict(self):
+        config = {"trim_segments": [{"segment_id": "s1", "duration": 3.0}, "invalid", None]}
+        result = parse_segments_from_config(config)
+        assert len(result) == 1
+
+    def test_single_trim_compat(self):
+        config = {"trim_start": 1.0, "trim_duration": 5.0}
+        result = parse_segments_from_config(config)
+        assert len(result) == 1
+        assert result[0].segment_id == "main"
+        assert result[0].trim.start_time == 1.0
+        assert result[0].trim.duration == 5.0
+
+    def test_no_trim_fields_returns_empty(self):
+        config = {"other_field": "value"}
+        assert parse_segments_from_config(config) == []
+
+
+# ── extract_trim_from_clip_config 测试 ────────────────────────────────────
+
+
+class TestExtractTrimFromClipConfig:
+    def test_none_returns_none(self):
+        assert extract_trim_from_clip_config(None) is None
+
+    def test_empty_dict_returns_none(self):
+        assert extract_trim_from_clip_config({}) is None
+
+    def test_trim_subdict(self):
+        config = {"trim": {"start_time": 2.0, "duration": 5.0}}
+        cfg = extract_trim_from_clip_config(config)
+        assert cfg is not None
+        assert cfg.start_time == 2.0
+        assert cfg.duration == 5.0
+
+    def test_flat_trim_fields(self):
+        config = {"trim_start": 1.0, "trim_end": 6.0}
+        cfg = extract_trim_from_clip_config(config)
+        assert cfg is not None
+        assert cfg.start_time == 1.0
+        assert cfg.end_time == 6.0
+
+    def test_trim_subdict_empty(self):
+        config = {"trim": {}}
+        assert extract_trim_from_clip_config(config) is None
+
+    def test_no_trim_fields(self):
+        config = {"foo": "bar"}
+        assert extract_trim_from_clip_config(config) is None
+
+    def test_flat_trim_duration_only(self):
+        config = {"trim_duration": 10.0}
+        cfg = extract_trim_from_clip_config(config)
+        assert cfg is not None
+        assert cfg.duration == 10.0