From 9fd3a469d7129efdc15f50bffd373f066c8d40b6 Mon Sep 17 00:00:00 2001 From: CI Bot Date: Tue, 14 Jul 2026 07:11:07 +0800 Subject: [PATCH 1/3] =?UTF-8?q?feat:=20=E8=A7=86=E9=A2=91=E8=A3=81?= =?UTF-8?q?=E5=89=AA/=E5=88=86=E5=89=B2=E8=83=BD=E5=8A=9B=EF=BC=88Trimming?= =?UTF-8?q?=20Engine=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 新增 TrimEngine 裁剪引擎，基于 FFmpeg trim/atrim 实现精确帧级裁剪： 核心能力： - 入点出点裁剪：start_time / end_time / duration 三选二 - 边界自动钳制：超出素材时长自动修正，不阻断渲染 - 多段裁剪：一个素材可裁剪出多段（trim_segments） - 音画同步：视频 + 音频同步裁剪 系统集成： - UnifiedRenderService Step 1 接入视频裁剪（修复原有 filter_complex 路径未使用 start_time 的问题） - render_audio 单 clip/多 clip/独立音轨 三处接入音频裁剪 - _resolve_clips 阶段解析裁剪配置 + 多段展开 129 个单测全绿（29 新增 + 100 现有） --- apps/worker/video_processing/render_audio.py | 16 +- apps/worker/video_processing/trim_engine.py | 343 ++++++++++++++++++ .../unified_render_service.py | 77 +++- tests/unit/test_trim_engine.py | 268 ++++++++++++++ 4 files changed, 695 insertions(+), 9 deletions(-) mode change 100644 => 100755 apps/worker/video_processing/render_audio.py create mode 100755 apps/worker/video_processing/trim_engine.py create mode 100755 tests/unit/test_trim_engine.py diff --git a/apps/worker/video_processing/render_audio.py b/apps/worker/video_processing/render_audio.py old mode 100644 new mode 100755 index 286b404fd..107076841 --- a/apps/worker/video_processing/render_audio.py +++ b/apps/worker/video_processing/render_audio.py @@ -145,6 +145,7 @@ def concat_main_audio( # 单 clip，直接提取音频，截断到 min(clip有效时长, 视频总时长) clip = clips[0] effective_duration = clip_effective_duration(clip) + trim_start = getattr(clip, "start_time", 0) or 0 # 最终时长：取 clip 有效时长和视频总时长的较小值 # （视频总时长由主图层决定，但单 clip 场景下两者应该一致，仍做保护） final_duration = effective_duration @@ -162,6 +163,8 @@ def concat_main_audio( "-b:a", "128k", ] + if trim_start > 0: + command.extend(["-ss", f"{trim_start:.3f}"]) if final_duration > 0: command.extend(["-t", f"{final_duration:.3f}"]) command.append(str(output_path)) @@ -175,8 +178,12 @@ def concat_main_audio( for i, clip in enumerate(clips): input_args.extend(["-i", str(clip.local_path)]) effective_duration = clip_effective_duration(clip) + trim_start = getattr(clip, "start_time", 0) or 0 if effective_duration > 0: - filter_parts.append(f"[{i}:a]atrim=0:{effective_duration:.3f},asetpts=PTS-STARTPTS[a{i}]") + filter_parts.append( + f"[{i}:a]atrim=start={trim_start:.3f}:duration={effective_duration:.3f}," + f"asetpts=PTS-STARTPTS[a{i}]" + ) else: filter_parts.append(f"[{i}:a]asetpts=PTS-STARTPTS[a{i}]") @@ -236,9 +243,11 @@ def mix_with_independent_audio( for clip in main_clips: input_args.extend(["-i", str(clip.local_path)]) effective_duration = clip_effective_duration(clip) + trim_start = getattr(clip, "start_time", 0) or 0 if effective_duration > 0: filter_parts.append( - f"[{input_idx}:a]atrim=0:{effective_duration:.3f},asetpts=PTS-STARTPTS[ma{input_idx}]" + f"[{input_idx}:a]atrim=start={trim_start:.3f}:duration={effective_duration:.3f}," + f"asetpts=PTS-STARTPTS[ma{input_idx}]" ) else: filter_parts.append(f"[{input_idx}:a]asetpts=PTS-STARTPTS[ma{input_idx}]") @@ -255,11 +264,12 @@ def mix_with_independent_audio( for j, clip in enumerate(audio_clips): input_args.extend(["-i", str(clip.local_path)]) effective_duration = clip_effective_duration(clip) + trim_start = getattr(clip, "start_time", 0) or 0 volume = clip.config.get("volume", 1.0) if clip.config else 1.0 label = f"ia{j}" filters = [] if effective_duration > 0: - filters.append(f"atrim=0:{effective_duration:.3f}") + filters.append(f"atrim=start={trim_start:.3f}:duration={effective_duration:.3f}") filters.append("asetpts=PTS-STARTPTS") if volume != 1.0: filters.append(f"volume={volume}") diff --git a/apps/worker/video_processing/trim_engine.py b/apps/worker/video_processing/trim_engine.py new file mode 100755 index 000000000..c17133d45 --- /dev/null +++ b/apps/worker/video_processing/trim_engine.py @@ -0,0 +1,343 @@ +"""裁剪引擎 — 基于 FFmpeg trim/atrim 的精确帧级裁剪. + +支持： +- 入点出点裁剪（start_time / end_time / duration 三选二） +- 边界自动钳制（超出素材时长自动修正，不阻断渲染） +- 多段裁剪（一个素材裁剪出多段） +- 音画同步（视频 + 音频同步裁剪） +""" + +from __future__ import annotations + +import logging +from dataclasses import dataclass +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)), + ) + + +class TrimEngine: + """裁剪引擎 — 生成 FFmpeg trim / atrim 滤镜.""" + + @staticmethod + 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}copy{output_label}" if False else 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 + + @staticmethod + 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 + + @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 + + @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) diff --git a/apps/worker/video_processing/unified_render_service.py b/apps/worker/video_processing/unified_render_service.py index a02c5a626..a0d88cb7d 100755 --- a/apps/worker/video_processing/unified_render_service.py +++ b/apps/worker/video_processing/unified_render_service.py @@ -47,6 +47,7 @@ from video_processing.subtitle_generator import generate_ass_from_timeline from video_processing.tts_engine import TtsEngine from packages.domain.tts_config import TtsConfig +from video_processing.trim_engine import TrimConfig, TrimEngine, extract_trim_from_clip_config logger = logging.getLogger(__name__) @@ -70,6 +71,7 @@ class ResolvedClip: # 运行时填充 actual_duration: float = 0.0 # 素材实际时长（probe 后填充） + trim_config: TrimConfig | None = None # 解析后的裁剪配置（运行时填充） @dataclass @@ -920,6 +922,7 @@ class UnifiedRenderService: """将 EditPlanClip 列表解析为 ResolvedClip 列表。 跳过 asset_id 为空或在 asset_path_map 中找不到的片段。 + 支持多段裁剪：一个 clip 配置了 trim_segments 时会展开为多个 ResolvedClip。 """ resolved: list[ResolvedClip] = [] for clip in self.clips: @@ -939,17 +942,75 @@ class UnifiedRenderService: except Exception: actual_duration = clip.duration or 5.0 + # 检查是否有多段裁剪配置 + clip_config = clip.config or {} + trim_segments = TrimEngine.parse_segments_from_config(clip_config) + + if trim_segments and len(trim_segments) > 1: + # 多段裁剪：展开为多个 clip + resolved_segments = TrimEngine.resolve_segments(trim_segments, actual_duration) + for i, seg in enumerate(resolved_segments): + # 每个段生成一个独立的 ResolvedClip + seg_clip_id = f"{clip.id}_seg_{seg.segment_id}" + seg_order = clip.order + seg.order * 0.001 + i * 0.0001 # 保持排序 + seg_start = seg.trim.start_time + seg_duration = seg.trim.duration + + rc = ResolvedClip( + clip_id=seg_clip_id, + asset_id=asset_id, + local_path=local_path, + clip_type=clip.clip_type, + order=seg_order, + start_time=seg_start, + duration=seg_duration, + transition_effect=clip.transition_effect or "cut", + config={**clip_config, "_segment_id": seg.segment_id}, + actual_duration=actual_duration, + trim_config=seg.trim, + ) + resolved.append(rc) + continue + + # 单段裁剪（或无裁剪） + # 解析裁剪配置：config 优先，否则用 clip.start_time + clip.duration + trim_config = extract_trim_from_clip_config(clip_config) + if trim_config is None and (clip.start_time > 0 or clip.duration > 0): + # 用旧字段构造 + trim_config = TrimConfig( + start_time=clip.start_time, + duration=clip.duration, + ) + + # 钳制到实际素材时长 + effective_trim: TrimConfig | None = None + final_start = clip.start_time + final_duration = clip.duration + + if trim_config is not None and actual_duration > 0: + effective_trim = trim_config.validate_and_resolve(actual_duration) + if effective_trim.is_valid: + final_start = effective_trim.start_time + final_duration = effective_trim.duration + else: + # 裁剪无效 → 使用完整素材 + logger.warning("裁剪配置无效，使用完整素材: clip_id=%s", clip.id) + effective_trim = None + final_start = 0.0 + final_duration = actual_duration + rc = ResolvedClip( clip_id=clip.id, asset_id=asset_id, local_path=local_path, clip_type=clip.clip_type, order=clip.order, - start_time=clip.start_time, - duration=clip.duration, + start_time=final_start, + duration=final_duration, transition_effect=clip.transition_effect or "cut", - config=clip.config or {}, + config=clip_config, actual_duration=actual_duration, + trim_config=effective_trim, ) resolved.append(rc) @@ -1024,7 +1085,7 @@ class UnifiedRenderService: filter_parts: list[str] = [] - # Step 1: 预处理每个 clip — scale + setpts + # Step 1: 预处理每个 clip — trim + scale + setpts # 为每个 clip 生成预处理后的标签 [v0], [v1], ... preprocessed_labels: list[str] = [] for i, clip in enumerate(all_clips): @@ -1033,11 +1094,15 @@ class UnifiedRenderService: filters: list[str] = [] - # trim — 始终将输出截断到有效时长，防止 xfade offset 与实际时长不匹配 + # trim — 裁剪到指定区间，精确到帧 effective_duration = UnifiedRenderService._clip_effective_duration(clip) + trim_start = getattr(clip, "start_time", 0) or 0 if effective_duration > 0: - filters.append(f"trim=duration={effective_duration}") + if trim_start > 0: + filters.append(f"trim=start={trim_start:.3f}:duration={effective_duration:.3f}") + else: + filters.append(f"trim=duration={effective_duration:.3f}") filters.append("setpts=PTS-STARTPTS") # scale diff --git a/tests/unit/test_trim_engine.py b/tests/unit/test_trim_engine.py new file mode 100755 index 000000000..d6b7a8eb8 --- /dev/null +++ b/tests/unit/test_trim_engine.py @@ -0,0 +1,268 @@ +"""裁剪引擎单元测试.""" + +import sys +import unittest +from pathlib import Path + +# 确保 apps/worker 在路径中 +sys.path.insert(0, str(Path(__file__).parent.parent.parent / "apps" / "worker")) + +from video_processing.trim_engine import ( + MIN_TRIM_DURATION, + TrimConfig, + TrimEngine, + TrimSegment, + extract_trim_from_clip_config, +) + + +class TestTrimConfig(unittest.TestCase): + """TrimConfig 单元测试.""" + + def test_from_dict_none(self): + """空字典返回 None（不裁剪）.""" + self.assertIsNone(TrimConfig.from_dict(None)) + self.assertIsNone(TrimConfig.from_dict({})) + + def test_from_dict_with_start(self): + """只有 start_time.""" + cfg = TrimConfig.from_dict({"start_time": 5.0}) + self.assertIsNotNone(cfg) + self.assertEqual(cfg.start_time, 5.0) + self.assertEqual(cfg.end_time, 0.0) + self.assertEqual(cfg.duration, 0.0) + + def test_from_dict_with_duration(self): + """只有 duration.""" + cfg = TrimConfig.from_dict({"duration": 10.0}) + self.assertIsNotNone(cfg) + self.assertEqual(cfg.start_time, 0.0) + self.assertEqual(cfg.duration, 10.0) + + def test_resolve_start_and_end(self): + """start + end 推导 duration.""" + cfg = TrimConfig(start_time=5.0, end_time=15.0) + resolved = cfg.validate_and_resolve(asset_duration=30.0) + self.assertEqual(resolved.start_time, 5.0) + self.assertEqual(resolved.end_time, 15.0) + self.assertAlmostEqual(resolved.duration, 10.0, places=3) + self.assertTrue(resolved.is_valid) + + def test_resolve_start_and_duration(self): + """start + duration 推导 end.""" + cfg = TrimConfig(start_time=5.0, duration=10.0) + resolved = cfg.validate_and_resolve(asset_duration=30.0) + self.assertEqual(resolved.start_time, 5.0) + self.assertAlmostEqual(resolved.end_time, 15.0, places=3) + self.assertEqual(resolved.duration, 10.0) + + def test_resolve_end_and_duration(self): + """end + duration 推导 start.""" + cfg = TrimConfig(end_time=20.0, duration=8.0) + resolved = cfg.validate_and_resolve(asset_duration=30.0) + self.assertAlmostEqual(resolved.start_time, 12.0, places=3) + self.assertEqual(resolved.end_time, 20.0) + self.assertEqual(resolved.duration, 8.0) + + def test_resolve_only_start(self): + """只有 start → 取到末尾.""" + cfg = TrimConfig(start_time=10.0) + resolved = cfg.validate_and_resolve(asset_duration=30.0) + self.assertEqual(resolved.start_time, 10.0) + self.assertEqual(resolved.end_time, 30.0) + self.assertAlmostEqual(resolved.duration, 20.0, places=3) + + def test_resolve_only_duration(self): + """只有 duration → 从开头取.""" + cfg = TrimConfig(duration=15.0) + resolved = cfg.validate_and_resolve(asset_duration=30.0) + self.assertEqual(resolved.start_time, 0.0) + self.assertAlmostEqual(resolved.end_time, 15.0, places=3) + self.assertEqual(resolved.duration, 15.0) + + def test_boundary_clamp_end(self): + """end 超出素材时长 → 钳制.""" + cfg = TrimConfig(start_time=5.0, duration=30.0) + resolved = cfg.validate_and_resolve(asset_duration=20.0) + self.assertEqual(resolved.start_time, 5.0) + self.assertEqual(resolved.end_time, 20.0) + self.assertAlmostEqual(resolved.duration, 15.0, places=3) + + def test_boundary_clamp_start_negative(self): + """start 为负 → 钳制到 0.""" + cfg = TrimConfig(start_time=-5.0, duration=10.0) + resolved = cfg.validate_and_resolve(asset_duration=30.0) + self.assertEqual(resolved.start_time, 0.0) + self.assertAlmostEqual(resolved.end_time, 10.0, places=3) + self.assertEqual(resolved.duration, 10.0) + + def test_boundary_start_past_end(self): + """start 超过素材总时长 → 钳制到末尾最小片段.""" + cfg = TrimConfig(start_time=50.0, duration=5.0) + resolved = cfg.validate_and_resolve(asset_duration=30.0) + self.assertTrue(resolved.start_time < 30.0) + self.assertEqual(resolved.end_time, 30.0) + self.assertTrue(resolved.duration >= MIN_TRIM_DURATION) + + def test_invalid_end_before_start(self): + """end <= start → 无效.""" + cfg = TrimConfig(start_time=15.0, end_time=10.0) + resolved = cfg.validate_and_resolve(asset_duration=30.0) + self.assertFalse(resolved.is_valid) + + def test_zero_duration_invalid(self): + """duration 为 0 → 无效.""" + cfg = TrimConfig(start_time=5.0, duration=0.0) + resolved = cfg.validate_and_resolve(asset_duration=30.0) + # 只有 start 没有 duration → 会被推导为取到末尾 + self.assertTrue(resolved.is_valid) + self.assertEqual(resolved.end_time, 30.0) + + def test_is_noop(self): + """is_noop 判断.""" + noop = TrimConfig(start_time=0.0, end_time=0.0, duration=0.0) + self.assertTrue(noop.is_noop) + + not_noop = TrimConfig(start_time=5.0, duration=10.0) + self.assertFalse(not_noop.is_noop) + + def test_zero_asset_duration(self): + """素材时长为 0 → 不裁剪.""" + cfg = TrimConfig(start_time=5.0, duration=10.0) + resolved = cfg.validate_and_resolve(asset_duration=0.0) + self.assertTrue(resolved.is_noop) + + def test_all_three_params_use_start_duration(self): + """三个参数都给了 → 以 start + duration 为准.""" + cfg = TrimConfig(start_time=5.0, end_time=20.0, duration=8.0) + resolved = cfg.validate_and_resolve(asset_duration=30.0) + # validate_and_resolve 中 start+end 优先于 start+duration + # 因为先检查的是 start>0 and end>0 + self.assertAlmostEqual(resolved.duration, 15.0, places=3) + + +class TestTrimEngine(unittest.TestCase): + """TrimEngine 单元测试.""" + + def test_build_video_trim_with_start_and_duration(self): + """视频裁剪：start + duration.""" + trim = TrimConfig(start_time=10.0, duration=5.0) + result = TrimEngine.build_video_trim_filter("[0:v]", trim, "[v0]") + self.assertIn("trim=start=10.000:duration=5.000", result) + self.assertIn("setpts=PTS-STARTPTS", result) + self.assertTrue(result.startswith("[0:v]")) + self.assertTrue(result.endswith("[v0]")) + + def test_build_video_trim_duration_only(self): + """视频裁剪：只有 duration.""" + trim = TrimConfig(start_time=0.0, duration=8.0) + result = TrimEngine.build_video_trim_filter("[0:v]", trim, "[v0]") + self.assertIn("trim=duration=8.000", result) + self.assertNotIn("start=", result.split("setpts")[0]) + + def test_build_audio_trim_with_start(self): + """音频裁剪：start + duration.""" + trim = TrimConfig(start_time=3.0, duration=7.0) + result = TrimEngine.build_audio_trim_filter("[0:a]", trim, "[a0]") + self.assertIn("atrim=start=3.000:duration=7.000", result) + self.assertIn("asetpts=PTS-STARTPTS", result) + + def test_build_audio_trim_noop(self): + """音频裁剪：noop.""" + trim = TrimConfig(start_time=0.0, end_time=0.0, duration=0.0) + result = TrimEngine.build_audio_trim_filter("[0:a]", trim, "[a0]") + self.assertIn("asetpts=PTS-STARTPTS", result) + self.assertNotIn("atrim=", result) + + def test_resolve_segments(self): + """多段裁剪解析.""" + segments = [ + TrimSegment(segment_id="s1", trim=TrimConfig(start_time=0.0, duration=5.0), order=0), + TrimSegment(segment_id="s2", trim=TrimConfig(start_time=10.0, duration=5.0), order=1), + TrimSegment(segment_id="s3", trim=TrimConfig(start_time=20.0, duration=5.0), order=2), + ] + resolved = TrimEngine.resolve_segments(segments, asset_duration=30.0) + self.assertEqual(len(resolved), 3) + self.assertEqual(resolved[0].segment_id, "s1") + self.assertEqual(resolved[0].trim.duration, 5.0) + self.assertEqual(resolved[1].segment_id, "s2") + self.assertEqual(resolved[1].trim.start_time, 10.0) + self.assertEqual(resolved[2].trim.start_time, 20.0) + + def test_resolve_segments_filter_invalid(self): + """多段裁剪：过滤无效段.""" + segments = [ + TrimSegment(segment_id="good", trim=TrimConfig(start_time=0.0, duration=5.0), order=0), + TrimSegment(segment_id="bad", trim=TrimConfig(start_time=10.0, end_time=5.0), order=1), # end < start + ] + resolved = TrimEngine.resolve_segments(segments, asset_duration=30.0) + self.assertEqual(len(resolved), 1) + self.assertEqual(resolved[0].segment_id, "good") + + def test_resolve_segments_boundary_clamp(self): + """多段裁剪：边界钳制.""" + segments = [ + TrimSegment(segment_id="s1", trim=TrimConfig(start_time=25.0, duration=10.0), order=0), + ] + resolved = TrimEngine.resolve_segments(segments, asset_duration=30.0) + self.assertEqual(len(resolved), 1) + self.assertEqual(resolved[0].trim.end_time, 30.0) + self.assertAlmostEqual(resolved[0].trim.duration, 5.0, places=3) + + def test_parse_segments_from_list(self): + """从 config 解析多段配置.""" + config = { + "trim_segments": [ + {"segment_id": "intro", "start_time": 0, "duration": 3, "order": 0}, + {"segment_id": "highlight", "start_time": 10, "duration": 5, "order": 1}, + {"segment_id": "outro", "start_time": 50, "duration": 3, "order": 2}, + ] + } + segments = TrimEngine.parse_segments_from_config(config) + self.assertEqual(len(segments), 3) + self.assertEqual(segments[0].segment_id, "intro") + self.assertEqual(segments[1].trim.start_time, 10.0) + self.assertEqual(segments[2].trim.duration, 3.0) + + def test_parse_segments_empty(self): + """无裁剪配置 → 空列表.""" + self.assertEqual(TrimEngine.parse_segments_from_config(None), []) + self.assertEqual(TrimEngine.parse_segments_from_config({}), []) + + def test_parse_single_trim_legacy(self): + """旧格式单段裁剪（trim_start/trim_duration）.""" + config = {"trim_start": 5.0, "trim_duration": 10.0} + segments = TrimEngine.parse_segments_from_config(config) + self.assertEqual(len(segments), 1) + self.assertEqual(segments[0].trim.start_time, 5.0) + self.assertEqual(segments[0].trim.duration, 10.0) + + +class TestExtractTrimFromClipConfig(unittest.TestCase): + """extract_trim_from_clip_config 单元测试.""" + + def test_trim_subdict(self): + """trim 子字典.""" + config = {"trim": {"start_time": 5.0, "duration": 10.0}} + result = extract_trim_from_clip_config(config) + self.assertIsNotNone(result) + self.assertEqual(result.start_time, 5.0) + self.assertEqual(result.duration, 10.0) + + def test_flat_fields(self): + """扁平字段（trim_start/trim_end/trim_duration）.""" + config = {"trim_start": 2.0, "trim_end": 8.0} + result = extract_trim_from_clip_config(config) + self.assertIsNotNone(result) + self.assertEqual(result.start_time, 2.0) + self.assertEqual(result.end_time, 8.0) + + def test_no_trim(self): + """无裁剪配置.""" + self.assertIsNone(extract_trim_from_clip_config(None)) + self.assertIsNone(extract_trim_from_clip_config({})) + self.assertIsNone(extract_trim_from_clip_config({"other": "value"})) + + +if __name__ == "__main__": + unittest.main() -- 2.54.0 From 70e514b3e4b05bfb81b8fb367e457f47231fb32b Mon Sep 17 00:00:00 2001 From: CI Bot Date: Tue, 14 Jul 2026 09:35:59 +0800 Subject: [PATCH 2/3] =?UTF-8?q?chore:=20rebase=E5=88=B0develop=20+=20?= =?UTF-8?q?=E6=A0=BC=E5=BC=8F=E5=8C=96=E4=BB=A3=E7=A0=81?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- apps/worker/video_processing/render_audio.py | 3 +-- apps/worker/video_processing/trim_engine.py | 8 ++------ 2 files changed, 3 insertions(+), 8 deletions(-) diff --git a/apps/worker/video_processing/render_audio.py b/apps/worker/video_processing/render_audio.py index 107076841..1c3082232 100755 --- a/apps/worker/video_processing/render_audio.py +++ b/apps/worker/video_processing/render_audio.py @@ -181,8 +181,7 @@ def concat_main_audio( trim_start = getattr(clip, "start_time", 0) or 0 if effective_duration > 0: filter_parts.append( - f"[{i}:a]atrim=start={trim_start:.3f}:duration={effective_duration:.3f}," - f"asetpts=PTS-STARTPTS[a{i}]" + f"[{i}:a]atrim=start={trim_start:.3f}:duration={effective_duration:.3f}," f"asetpts=PTS-STARTPTS[a{i}]" ) else: filter_parts.append(f"[{i}:a]asetpts=PTS-STARTPTS[a{i}]") diff --git a/apps/worker/video_processing/trim_engine.py b/apps/worker/video_processing/trim_engine.py index c17133d45..ada3f373a 100755 --- a/apps/worker/video_processing/trim_engine.py +++ b/apps/worker/video_processing/trim_engine.py @@ -293,9 +293,7 @@ class TrimEngine: return segments # 单段裁剪兼容：从 trim_start/trim_end/trim_duration 构造 - has_single = any( - k in config for k in ("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", @@ -329,9 +327,7 @@ def extract_trim_from_clip_config(config: dict[str, Any] | None) -> TrimConfig | return TrimConfig.from_dict(config["trim"]) # 扁平字段 - has_any = any( - k in config for k in ("trim_start", "trim_end", "trim_duration") - ) + has_any = any(k in config for k in ("trim_start", "trim_end", "trim_duration")) if not has_any: return None -- 2.54.0 From 9426425c60fb51dd804c0ad8f0c52382ab9d630b Mon Sep 17 00:00:00 2001 From: xiaoxia-bot Date: Tue, 14 Jul 2026 10:48:34 +0800 Subject: [PATCH 3/3] =?UTF-8?q?fix:=20isort=20import=E6=8E=92=E5=BA=8F?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- apps/worker/video_processing/unified_render_service.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/apps/worker/video_processing/unified_render_service.py b/apps/worker/video_processing/unified_render_service.py index a0d88cb7d..a59373fc5 100755 --- a/apps/worker/video_processing/unified_render_service.py +++ b/apps/worker/video_processing/unified_render_service.py @@ -44,10 +44,10 @@ from video_processing.pip_engine import PiPConfig, PiPEngine, PiPLayerConfig from video_processing.render_audio import RenderContext, merge_audio_video, mix_audio from video_processing.render_subtitles import generate_ass_subtitles from video_processing.subtitle_generator import generate_ass_from_timeline +from video_processing.trim_engine import TrimConfig, TrimEngine, extract_trim_from_clip_config from video_processing.tts_engine import TtsEngine from packages.domain.tts_config import TtsConfig -from video_processing.trim_engine import TrimConfig, TrimEngine, extract_trim_from_clip_config logger = logging.getLogger(__name__) -- 2.54.0