diff --git a/apps/worker/video_processing/trim_engine.py b/apps/worker/video_processing/trim_engine.py index 62bf6f88a..e9eb5d89c 100755 --- a/apps/worker/video_processing/trim_engine.py +++ b/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) diff --git a/packages/domain/trim_config.py b/packages/domain/trim_config.py new file mode 100755 index 000000000..2cf912390 --- /dev/null +++ b/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) diff --git a/tests/unit/test_trim_config.py b/tests/unit/test_trim_config.py new file mode 100755 index 000000000..b370b2934 --- /dev/null +++ b/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