diff --git a/alembic/versions/039_add_transition_duration.py b/alembic/versions/039_add_transition_duration.py new file mode 100644 index 000000000..6e9f2c36d --- /dev/null +++ b/alembic/versions/039_add_transition_duration.py @@ -0,0 +1,34 @@ +"""add transition_duration to edit_plan_clips + +Revision ID: 039_transition_duration +Revises: 038_error_retry +Create Date: 2026-07-14 09:00:00.000000 +""" + +from __future__ import annotations + +import sqlalchemy as sa + +from alembic import op + +# revision identifiers, used by Alembic. +revision = "039_transition_duration" +down_revision = "038_error_retry" +branch_labels = None +depends_on = None + + +def upgrade() -> None: + op.add_column( + "edit_plan_clips", + sa.Column( + "transition_duration", + sa.Float(), + nullable=False, + server_default="0.0", + ), + ) + + +def downgrade() -> None: + op.drop_column("edit_plan_clips", "transition_duration") diff --git a/apps/api/app/api/routes/edit_plans.py b/apps/api/app/api/routes/edit_plans.py index 85ba23b6a..1e124bcbc 100644 --- a/apps/api/app/api/routes/edit_plans.py +++ b/apps/api/app/api/routes/edit_plans.py @@ -146,6 +146,7 @@ class AIRecommendClipItem(BaseModel): text_content: str = Field(default="", description="文字内容") duration: float = Field(..., ge=0.0, description="片段时长（秒）") transition_effect: str = Field(default="cut", description="转场效果") + transition_duration: float = Field(default=0.0, ge=0.0, description="转场时长（秒），0 表示使用默认值") asset_id: str = Field(default="", description="关联素材 ID") start_time: float = Field(default=0.0, ge=0.0, description="素材截取起始时间（秒）") config: dict[str, Any] = Field(default_factory=dict, description="片段额外配置") @@ -209,6 +210,7 @@ class _PlanClipItem(BaseModel): start_time: float duration: float transition_effect: str + transition_duration: float status: str config: Optional[dict[str, Any]] = None created_at: datetime diff --git a/apps/api/app/api/routes/edit_plans_timeline.py b/apps/api/app/api/routes/edit_plans_timeline.py old mode 100644 new mode 100755 index e75a1e899..c1e6f1249 --- a/apps/api/app/api/routes/edit_plans_timeline.py +++ b/apps/api/app/api/routes/edit_plans_timeline.py @@ -210,6 +210,7 @@ def generate_from_template( start_time=c.start_time, duration=c.duration, transition_effect=c.transition_effect, + transition_duration=c.transition_duration, status=c.status.value if hasattr(c.status, "value") else c.status, config=c.config, created_at=c.created_at, diff --git a/apps/api/app/services/edit_plan_service.py b/apps/api/app/services/edit_plan_service.py old mode 100644 new mode 100755 index bf8122705..80280eb07 --- a/apps/api/app/services/edit_plan_service.py +++ b/apps/api/app/services/edit_plan_service.py @@ -281,6 +281,7 @@ class EditPlanService: start_time: float = 0.0, duration: float = 0.0, transition_effect: str = "cut", + transition_duration: float = 0.0, config: Optional[dict[str, Any]] = None, ) -> EditPlanClip: """创建片段 @@ -301,6 +302,7 @@ class EditPlanService: start_time=start_time, duration=duration, transition_effect=transition_effect, + transition_duration=transition_duration, config=config, ) created = self._clip_repo.create(clip) @@ -324,6 +326,7 @@ class EditPlanService: start_time: Optional[float] = None, duration: Optional[float] = None, transition_effect: Optional[str] = None, + transition_duration: Optional[float] = None, config: Optional[dict[str, Any]] = None, ) -> EditPlanClip: """更新片段 @@ -346,6 +349,9 @@ class EditPlanService: transition_effect=( transition_effect.strip() if transition_effect is not None else existing.transition_effect ), + transition_duration=( + transition_duration if transition_duration is not None else existing.transition_duration + ), status=existing.status, config=config if config is not None else existing.config, created_at=existing.created_at, diff --git a/apps/worker/video_processing/ffmpeg_utils.py b/apps/worker/video_processing/ffmpeg_utils.py index 02c1a099e..d40cc2d1f 100755 --- a/apps/worker/video_processing/ffmpeg_utils.py +++ b/apps/worker/video_processing/ffmpeg_utils.py @@ -25,8 +25,14 @@ DEFAULT_FPS = 25 # xfade 转场映射：transition_effect 名称 → FFmpeg xfade transition 名称 # 键同时支持 TransitionEffect 枚举值和字符串名称（向后兼容） +# "cut" 为特殊值：硬切，不使用 xfade（由调用方特殊处理） XFADE_TRANSITION_MAP: dict[str, str] = { + # 基础 "fade": "fade", + "dissolve": "dissolve", + "crossfade": "dissolve", + "crossdissolve": "dissolve", + # 滑入系列 "slideleft": "slideleft", "slide_left": "slideleft", "slideright": "slideright", @@ -35,9 +41,22 @@ XFADE_TRANSITION_MAP: dict[str, str] = { "slide_up": "slideup", "slidedown": "slidedown", "slide_down": "slidedown", - "dissolve": "dissolve", - "wipe": "wipeleft", + "slide": "slideleft", # 默认向左滑 + # 缩放 + "zoom": "zoomin", + "zoomin": "zoomin", + "zoomout": "zoomout", + # 擦除系列 + "wipe": "wipeleft", # 默认向左擦 "wipeleft": "wipeleft", + "wiperight": "wiperight", + "wipeup": "wipeup", + "wipedown": "wipedown", + # 特殊效果 + "circlecrop": "circlecrop", + "circle": "circlecrop", + "rectcrop": "rectcrop", + "rect": "rectcrop", } DEFAULT_TRANSITION_DURATION = 0.5 diff --git a/apps/worker/video_processing/transition_engine.py b/apps/worker/video_processing/transition_engine.py new file mode 100755 index 000000000..03b8409ff --- /dev/null +++ b/apps/worker/video_processing/transition_engine.py @@ -0,0 +1,381 @@ +"""转场特效引擎 — Phase 8 智能增强. + +基于 FFmpeg xfade 滤镜的统一转场抽象层，提供： + 1. 转场类型枚举与预设管理 + 2. 转场配置解析与边界校验 + 3. 降级策略（不支持的转场自动 fallback 到硬切） + 4. xfade 滤镜链构建（封装底层 ffmpeg_utils） + +新增转场只需在 TransitionType 中加一项 + 在 XFADE_TRANSITION_MAP 中映射。 +""" + +from __future__ import annotations + +import logging +import sys +from dataclasses import dataclass + +if sys.version_info >= (3, 11): + from enum import StrEnum +else: + from enum import Enum + + class StrEnum(str, Enum): + pass + + +from video_processing.ffmpeg_utils import build_xfade_filter_chain + +logger = logging.getLogger(__name__) + + +# ── 常量 ────────────────────────────────────────────────────────────────────── + +# 转场时长范围（秒） +MIN_TRANSITION_DURATION = 0.3 +MAX_TRANSITION_DURATION = 2.0 +DEFAULT_TRANSITION_DURATION = 0.5 + +# 硬切（无转场） +CUT_TRANSITION = "cut" + + +# ── 转场类型枚举 ────────────────────────────────────────────────────────────── + + +class TransitionType(StrEnum): + """支持的转场效果类型. + + 每种类型对应 FFmpeg xfade filter 的一个 transition 值。 + 新增转场只需在此添加一项，并在 _FFMPEG_XFADE_MAP 中映射。 + """ + + # 硬切（无转场效果，直接拼接） + CUT = "cut" + + # 淡入淡出（最常用，默认 fallback） + FADE = "fade" + + # 溶解（交叉溶解） + DISSOLVE = "dissolve" + + # 滑入系列 + SLIDE_LEFT = "slideleft" + SLIDE_RIGHT = "slideright" + SLIDE_UP = "slideup" + SLIDE_DOWN = "slidedown" + + # 缩放 + ZOOM = "zoom" + + # 擦除系列 + WIPE_LEFT = "wipeleft" + WIPE_RIGHT = "wiperight" + WIPE_UP = "wipeup" + WIPE_DOWN = "wipedown" + + # 圆形扩散 + CIRCLE_CROP = "circlecrop" + + # 矩形覆盖 + RECT_CROP = "rectcrop" + + @classmethod + def all_supported(cls) -> list[str]: + """返回所有支持的转场类型名称列表.""" + return [t.value for t in cls if t != cls.CUT] + + @classmethod + def is_supported(cls, name: str) -> bool: + """检查转场类型是否支持（不区分大小写和下划线）.""" + normalized = _normalize_transition_name(name) + return normalized in _NAME_TO_ENUM_MAP + + +# ── 名称 → 枚举 映射（支持多种别名）────────────────────────────────────────── + + +def _normalize_transition_name(name: str) -> str: + """标准化转场名称：小写 + 去下划线.""" + return name.lower().replace("_", "").replace("-", "") + + +# 构建别名映射 +_NAME_TO_ENUM_MAP: dict[str, TransitionType] = {} +for _t in TransitionType: + _NAME_TO_ENUM_MAP[_normalize_transition_name(_t.value)] = _t + +# 额外的别名 +_ALIASES: dict[str, TransitionType] = { + "dissolve": TransitionType.DISSOLVE, + "crossfade": TransitionType.DISSOLVE, + "crossdissolve": TransitionType.DISSOLVE, + "fadein": TransitionType.FADE, + "fadeout": TransitionType.FADE, + "fadeblack": TransitionType.FADE, + "slide": TransitionType.SLIDE_LEFT, # 默认向左滑 + "wipe": TransitionType.WIPE_LEFT, # 默认向左擦 + "zoomin": TransitionType.ZOOM, + "zoomout": TransitionType.ZOOM, + "circle": TransitionType.CIRCLE_CROP, + "rect": TransitionType.RECT_CROP, +} +for _alias, _type in _ALIASES.items(): + _key = _normalize_transition_name(_alias) + if _key not in _NAME_TO_ENUM_MAP: + _NAME_TO_ENUM_MAP[_key] = _type + + +# ── TransitionType → FFmpeg xfade transition 名称映射 ───────────────────────── + + +_FFMPEG_XFADE_MAP: dict[TransitionType, str] = { + TransitionType.FADE: "fade", + TransitionType.DISSOLVE: "dissolve", + TransitionType.SLIDE_LEFT: "slideleft", + TransitionType.SLIDE_RIGHT: "slideright", + TransitionType.SLIDE_UP: "slideup", + TransitionType.SLIDE_DOWN: "slidedown", + TransitionType.ZOOM: "zoomin", + TransitionType.WIPE_LEFT: "wipeleft", + TransitionType.WIPE_RIGHT: "wiperight", + TransitionType.WIPE_UP: "wipeup", + TransitionType.WIPE_DOWN: "wipedown", + TransitionType.CIRCLE_CROP: "circlecrop", + TransitionType.RECT_CROP: "rectcrop", +} + + +# ── 转场配置 ────────────────────────────────────────────────────────────────── + + +@dataclass(slots=True) +class TransitionConfig: + """转场效果配置. + + Attributes: + effect: 转场效果名称（见 TransitionType） + duration: 转场时长（秒），范围 0.3~2.0，默认 0.5 + """ + + effect: str = CUT_TRANSITION + duration: float = DEFAULT_TRANSITION_DURATION + + @classmethod + def parse(cls, effect: str | None = None, duration: float | None = None) -> "TransitionConfig": + """解析并验证转场配置，自动处理边界和降级. + + Args: + effect: 转场效果名称（None 或空则使用默认 cut） + duration: 转场时长（None 则使用默认值） + + Returns: + 验证后的 TransitionConfig + """ + # 处理 effect + final_effect = CUT_TRANSITION + if effect and effect.strip(): + effect_clean = effect.strip() + if TransitionType.is_supported(effect_clean): + final_effect = _resolve_transition_enum(effect_clean).value + elif effect_clean.lower() == CUT_TRANSITION: + final_effect = CUT_TRANSITION + else: + # 降级：不支持的转场 → 硬切，不阻断渲染 + logger.warning( + "不支持的转场效果 '%s'，已降级为硬切（cut）", + effect_clean, + ) + final_effect = CUT_TRANSITION + + # 处理 duration：边界钳制 + final_duration = DEFAULT_TRANSITION_DURATION + if duration is not None: + try: + d = float(duration) + if d < MIN_TRANSITION_DURATION: + logger.warning( + "转场时长 %.3fs 小于最小值 %.1fs，已钳制到最小值", + d, + MIN_TRANSITION_DURATION, + ) + final_duration = MIN_TRANSITION_DURATION + elif d > MAX_TRANSITION_DURATION: + logger.warning( + "转场时长 %.3fs 大于最大值 %.1fs，已钳制到最大值", + d, + MAX_TRANSITION_DURATION, + ) + final_duration = MAX_TRANSITION_DURATION + else: + final_duration = d + except (TypeError, ValueError): + logger.warning("无效的转场时长 '%s'，使用默认值 %.1fs", duration, DEFAULT_TRANSITION_DURATION) + final_duration = DEFAULT_TRANSITION_DURATION + + return cls(effect=final_effect, duration=final_duration) + + @property + def is_cut(self) -> bool: + """是否为硬切（无转场效果）.""" + return self.effect == CUT_TRANSITION + + @property + def ffmpeg_transition(self) -> str: + """获取对应的 FFmpeg xfade transition 名称.""" + if self.is_cut: + return "" + enum_type = _resolve_transition_enum(self.effect) + return _FFMPEG_XFADE_MAP.get(enum_type, "fade") + + +def _resolve_transition_enum(name: str) -> TransitionType: + """将名称解析为 TransitionType 枚举，必须先通过 is_supported 校验.""" + normalized = _normalize_transition_name(name) + return _NAME_TO_ENUM_MAP.get(normalized, TransitionType.FADE) + + +# ── 转场引擎 ────────────────────────────────────────────────────────────────── + + +class TransitionEngine: + """转场特效引擎. + + 封装转场配置验证、降级策略和 xfade 滤镜链构建， + 供 UnifiedRenderService 等上层调用。 + + 用法:: + + engine = TransitionEngine(default_duration=0.5) + config = engine.resolve_config("fade", 0.8) + filter_str, total_dur = engine.build_xfade_chain( + clip_durations=[3.0, 4.0, 5.0], + clip_video_labels=["v0", "v1", "v2"], + transitions=["cut", "fade", "dissolve"], + ) + """ + + def __init__(self, default_duration: float = DEFAULT_TRANSITION_DURATION) -> None: + """初始化转场引擎. + + Args: + default_duration: 默认转场时长（秒），用于未指定时长的 clip + """ + self._default_duration = default_duration + + def resolve_config( + self, + effect: str | None = None, + duration: float | None = None, + ) -> TransitionConfig: + """解析单个转场配置，应用验证和降级. + + Args: + effect: 转场效果名称 + duration: 转场时长 + + Returns: + 验证后的 TransitionConfig + """ + # 若未指定 duration，使用引擎默认值 + dur = duration if duration is not None else self._default_duration + return TransitionConfig.parse(effect=effect, duration=dur) + + def resolve_clip_transitions( + self, + clip_transitions: list[str], + clip_durations: list[float] | None = None, + ) -> list[TransitionConfig]: + """批量解析 clip 级别的转场配置. + + Args: + clip_transitions: 每个 clip 的转场效果名称列表 + clip_durations: 每个 clip 的时长列表（用于验证转场时长不超过片段时长） + + Returns: + TransitionConfig 列表 + """ + configs: list[TransitionConfig] = [] + for i, effect in enumerate(clip_transitions): + cfg = self.resolve_config(effect=effect) + # 额外校验：转场时长不能超过对应 clip 时长的一半（保守限制） + if clip_durations and i < len(clip_durations) and not cfg.is_cut: + max_safe_duration = max(MIN_TRANSITION_DURATION, clip_durations[i] * 0.5) + if cfg.duration > max_safe_duration: + cfg = TransitionConfig(effect=cfg.effect, duration=max_safe_duration) + configs.append(cfg) + return configs + + def build_xfade_chain( + self, + clip_durations: list[float], + clip_video_labels: list[str], + transitions: list[str], + *, + transition_duration: float | None = None, + output_label: str = "outv", + ) -> tuple[str, float]: + """构建 xfade 转场滤镜链. + + 对每步转场应用验证和降级，然后调用底层 ffmpeg_utils 构建。 + + Args: + clip_durations: 每个片段的时长 + clip_video_labels: 每个片段的视频流标签 + transitions: 每个片段对应的转场效果 + transition_duration: 统一转场时长，None 则使用引擎默认值 + output_label: 最终输出标签 + + Returns: + (filter_string, estimated_total_duration) + """ + if len(clip_durations) <= 1: + return build_xfade_filter_chain( + clip_durations=clip_durations, + clip_video_labels=clip_video_labels, + transitions=transitions, + transition_duration=transition_duration or self._default_duration, + output_label=output_label, + ) + + # 解析所有转场配置 + resolved = self.resolve_clip_transitions(transitions, clip_durations) + resolved_effects = [c.effect for c in resolved] + + # 使用统一的时长（取各转场中最大的时长作为基准，底层会做每步钳制） + dur = transition_duration or self._default_duration + if not dur: + dur = max(c.duration for c in resolved) if resolved else DEFAULT_TRANSITION_DURATION + + # 调用底层构建 + return build_xfade_filter_chain( + clip_durations=clip_durations, + clip_video_labels=clip_video_labels, + transitions=resolved_effects, + transition_duration=dur, + output_label=output_label, + ) + + @staticmethod + def supported_transitions() -> list[dict[str, str]]: + """获取所有支持的转场效果列表（用于 API 返回给前端）. + + Returns: + [{name, display_name, category}, ...] + """ + return [ + {"name": "cut", "display_name": "硬切", "category": "basic"}, + {"name": "fade", "display_name": "淡入淡出", "category": "basic"}, + {"name": "dissolve", "display_name": "溶解", "category": "basic"}, + {"name": "slideleft", "display_name": "左滑入", "category": "slide"}, + {"name": "slideright", "display_name": "右滑入", "category": "slide"}, + {"name": "slideup", "display_name": "上滑入", "category": "slide"}, + {"name": "slidedown", "display_name": "下滑入", "category": "slide"}, + {"name": "zoom", "display_name": "缩放", "category": "zoom"}, + {"name": "wipeleft", "display_name": "左擦除", "category": "wipe"}, + {"name": "wiperight", "display_name": "右擦除", "category": "wipe"}, + {"name": "wipeup", "display_name": "上擦除", "category": "wipe"}, + {"name": "wipedown", "display_name": "下擦除", "category": "wipe"}, + {"name": "circlecrop", "display_name": "圆形扩散", "category": "special"}, + {"name": "rectcrop", "display_name": "矩形扩散", "category": "special"}, + ] diff --git a/apps/worker/video_processing/unified_render_service.py b/apps/worker/video_processing/unified_render_service.py index 396fb210c..2b099bf0c 100755 --- a/apps/worker/video_processing/unified_render_service.py +++ b/apps/worker/video_processing/unified_render_service.py @@ -36,7 +36,6 @@ from video_processing.ffmpeg_utils import ( DEFAULT_OUTPUT_WIDTH, DEFAULT_TRANSITION_DURATION, FFMPEG_BIN, - build_xfade_filter_chain, probe_duration, probe_video_info, run_ffmpeg, @@ -46,6 +45,7 @@ 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.transition_engine import TransitionEngine from video_processing.trim_engine import TrimConfig, TrimEngine, extract_trim_from_clip_config from video_processing.tts_engine import TtsEngine from video_processing.watermark_engine import WatermarkConfig, WatermarkEngine @@ -70,6 +70,7 @@ class ResolvedClip: start_time: float = 0.0 duration: float = 0.0 # 0 表示使用素材完整时长 transition_effect: str = "cut" + transition_duration: float = 0.0 # 0 表示使用全局默认值 config: dict[str, Any] = field(default_factory=dict) # 运行时填充 @@ -182,6 +183,7 @@ class UnifiedRenderService: self.transition_duration = transition_duration self.asr_service = asr_service self.bgm_path = bgm_path + self._transition_engine = TransitionEngine(default_duration=transition_duration) def render(self) -> RenderResult: """执行渲染，返回 RenderResult. @@ -1173,6 +1175,7 @@ class UnifiedRenderService: start_time=final_start, duration=final_duration, transition_effect=clip.transition_effect or "cut", + transition_duration=getattr(clip, "transition_duration", 0.0) or 0.0, config=clip_config, actual_duration=actual_duration, trim_config=effective_trim, @@ -1323,18 +1326,25 @@ class UnifiedRenderService: # 使用 trim 后的有效时长，与 Step 1 的 trim=duration 保持一致 layer_durations = [UnifiedRenderService._clip_effective_duration(all_clips[i]) for i in layer_clip_indices] layer_transitions = [all_clips[i].transition_effect for i in layer_clip_indices] + layer_transition_durations = [all_clips[i].transition_duration for i in layer_clip_indices] if len(layer_labels) == 1: # 单 clip 层，直接使用预处理标签 layer_output_labels[layer.role] = layer_labels[0] else: - # 多 clip 层，用 xfade 串联 + # 多 clip 层，用 TransitionEngine 构建转场链 out_label = f"{layer.role}_merged" - xfade_filter, _ = build_xfade_filter_chain( + # 计算该层使用的转场时长（取首个非零值，否则用默认） + layer_dur = 0.0 + for d in layer_transition_durations: + if d > 0: + layer_dur = d + break + xfade_filter, _ = self._transition_engine.build_xfade_chain( clip_durations=layer_durations, clip_video_labels=layer_labels, transitions=layer_transitions, - transition_duration=self.transition_duration, + transition_duration=layer_dur if layer_dur > 0 else None, output_label=out_label, ) if xfade_filter: diff --git a/docs/schema-metadata-snapshot.json b/docs/schema-metadata-snapshot.json index 45a76d430..81b20bec0 100644 --- a/docs/schema-metadata-snapshot.json +++ b/docs/schema-metadata-snapshot.json @@ -858,6 +858,14 @@ "type": "VARCHAR(20)", "unique": false }, + { + "index": false, + "name": "transition_duration", + "nullable": false, + "primary_key": false, + "type": "FLOAT", + "unique": false + }, { "index": true, "name": "status", diff --git a/packages/adapters/sqlalchemy_impl/edit_plan_clip_repository.py b/packages/adapters/sqlalchemy_impl/edit_plan_clip_repository.py old mode 100644 new mode 100755 index 5e51b9dcc..8ae8a81ef --- a/packages/adapters/sqlalchemy_impl/edit_plan_clip_repository.py +++ b/packages/adapters/sqlalchemy_impl/edit_plan_clip_repository.py @@ -54,6 +54,7 @@ class SQLAlchemyEditPlanClipRepository: start_time=clip.start_time, duration=clip.duration, transition_effect=clip.transition_effect, + transition_duration=clip.transition_duration, status=clip.status, config=clip.config, ) @@ -76,6 +77,7 @@ class SQLAlchemyEditPlanClipRepository: model.start_time = clip.start_time model.duration = clip.duration model.transition_effect = clip.transition_effect + model.transition_duration = clip.transition_duration model.status = clip.status model.config = clip.config model.updated_at = clip.updated_at @@ -120,6 +122,7 @@ class SQLAlchemyEditPlanClipRepository: start_time=model.start_time or 0.0, duration=model.duration or 0.0, transition_effect=model.transition_effect or "cut", + transition_duration=getattr(model, "transition_duration", 0.0) or 0.0, status=EditPlanClipStatus(model.status) if model.status else EditPlanClipStatus.PENDING, config=model.config or {}, created_at=model.created_at, diff --git a/packages/adapters/sqlalchemy_impl/models.py b/packages/adapters/sqlalchemy_impl/models.py index fb955846f..00d4440b6 100755 --- a/packages/adapters/sqlalchemy_impl/models.py +++ b/packages/adapters/sqlalchemy_impl/models.py @@ -196,6 +196,7 @@ class EditPlanClipModel(Base): start_time = Column(Float, nullable=False, default=0.0) duration = Column(Float, nullable=False, default=0.0) transition_effect = Column(String(20), nullable=False, default="cut") + transition_duration = Column(Float, nullable=False, default=0.0) status = Column(String(20), nullable=False, default="pending", index=True) config = Column(JSON, nullable=False, default=dict) created_at = Column(DateTime, nullable=False, default=lambda: datetime.now(timezone.utc)) diff --git a/packages/domain/edit_plan_clip.py b/packages/domain/edit_plan_clip.py old mode 100644 new mode 100755 index 15171c23c..8fe9e9051 --- a/packages/domain/edit_plan_clip.py +++ b/packages/domain/edit_plan_clip.py @@ -50,6 +50,7 @@ class EditPlanClip: start_time: float = 0.0 duration: float = 0.0 transition_effect: str = "cut" + transition_duration: float = 0.0 # 0 表示使用全局默认值 status: EditPlanClipStatus = EditPlanClipStatus.PENDING config: dict[str, Any] = field(default_factory=dict) created_at: datetime = field(default_factory=lambda: datetime.now(timezone.utc)) @@ -68,6 +69,7 @@ class EditPlanClip: start_time: float = 0.0, duration: float = 0.0, transition_effect: str = "cut", + transition_duration: float = 0.0, config: dict[str, Any] | None = None, ) -> EditPlanClip: """创建剪辑计划片段""" @@ -91,6 +93,7 @@ class EditPlanClip: start_time=start_time, duration=duration, transition_effect=transition_effect.strip() or "cut", + transition_duration=max(0.0, transition_duration), status=EditPlanClipStatus.PENDING, config=config or {}, ) diff --git a/tests/unit/test_transition_engine.py b/tests/unit/test_transition_engine.py new file mode 100755 index 000000000..a41a9a92f --- /dev/null +++ b/tests/unit/test_transition_engine.py @@ -0,0 +1,484 @@ +"""转场特效引擎单测 — Phase 8 智能增强.""" + +from __future__ import annotations + +import pytest +from video_processing.transition_engine import ( + CUT_TRANSITION, + DEFAULT_TRANSITION_DURATION, + MAX_TRANSITION_DURATION, + MIN_TRANSITION_DURATION, + TransitionConfig, + TransitionEngine, + TransitionType, + _normalize_transition_name, +) + +# ── TransitionType 枚举测试 ────────────────────────────────────────────────── + + +class TestTransitionType: + """TransitionType 枚举测试.""" + + def test_all_supported_count(self): + """支持的转场类型数量（不含cut）.""" + supported = TransitionType.all_supported() + # 至少 8 种：fade, dissolve, slide*4, zoom, wipe*4, circlecrop, rectcrop + assert len(supported) >= 8 + assert "fade" in supported + assert "dissolve" in supported + assert "zoom" in supported + assert "circlecrop" in supported + assert "rectcrop" in supported + + def test_slide_directions(self): + """四个方向的滑入转场都支持.""" + assert TransitionType.is_supported("slideleft") + assert TransitionType.is_supported("slideright") + assert TransitionType.is_supported("slideup") + assert TransitionType.is_supported("slidedown") + + def test_wipe_directions(self): + """四个方向的擦除转场都支持.""" + assert TransitionType.is_supported("wipeleft") + assert TransitionType.is_supported("wiperight") + assert TransitionType.is_supported("wipeup") + assert TransitionType.is_supported("wipedown") + + def test_is_supported_case_insensitive(self): + """大小写不敏感.""" + assert TransitionType.is_supported("FADE") + assert TransitionType.is_supported("Fade") + assert TransitionType.is_supported("fade") + + def test_is_supported_with_underscores(self): + """下划线不影响判断.""" + assert TransitionType.is_supported("slide_left") + assert TransitionType.is_supported("slide-left") + + def test_is_supported_aliases(self): + """别名支持.""" + assert TransitionType.is_supported("crossfade") + assert TransitionType.is_supported("dissolve") + assert TransitionType.is_supported("zoomin") + assert TransitionType.is_supported("wipe") + + def test_unsupported_transition(self): + """不支持的转场返回 False.""" + assert not TransitionType.is_supported("nonexistent_effect") + assert not TransitionType.is_supported("random_stuff") + assert not TransitionType.is_supported("") + + def test_cut_not_in_supported(self): + """硬切不在"支持的转场效果"列表中（它不是特效）.""" + supported = TransitionType.all_supported() + assert "cut" not in supported + + +# ── 名称标准化测试 ──────────────────────────────────────────────────────────── + + +class TestNormalizeTransitionName: + """名称标准化函数测试.""" + + def test_lowercase(self): + """大写转小写.""" + assert _normalize_transition_name("FADE") == "fade" + assert _normalize_transition_name("Fade") == "fade" + + def test_remove_underscores(self): + """移除下划线.""" + assert _normalize_transition_name("slide_left") == "slideleft" + assert _normalize_transition_name("slide_up") == "slideup" + + def test_remove_hyphens(self): + """移除连字符.""" + assert _normalize_transition_name("slide-left") == "slideleft" + + def test_mixed(self): + """混合情况.""" + assert _normalize_transition_name("Slide_Left") == "slideleft" + assert _normalize_transition_name("FADE-IN") == "fadein" + + +# ── TransitionConfig 测试 ──────────────────────────────────────────────────── + + +class TestTransitionConfig: + """TransitionConfig 配置解析测试.""" + + # ── 默认值 ── + + def test_default_config(self): + """默认配置是硬切.""" + cfg = TransitionConfig.parse() + assert cfg.effect == CUT_TRANSITION + assert cfg.duration == DEFAULT_TRANSITION_DURATION + assert cfg.is_cut is True + + def test_none_effect(self): + """None effect 降级为 cut.""" + cfg = TransitionConfig.parse(effect=None) + assert cfg.effect == CUT_TRANSITION + assert cfg.is_cut is True + + def test_empty_effect(self): + """空字符串 effect 降级为 cut.""" + cfg = TransitionConfig.parse(effect="") + assert cfg.effect == CUT_TRANSITION + assert cfg.is_cut is True + + # ── 有效转场类型 ── + + def test_fade_effect(self): + """fade 转场.""" + cfg = TransitionConfig.parse(effect="fade") + assert cfg.effect == "fade" + assert cfg.is_cut is False + assert cfg.ffmpeg_transition == "fade" + + def test_dissolve_effect(self): + """dissolve 转场.""" + cfg = TransitionConfig.parse(effect="dissolve") + assert cfg.effect == "dissolve" + assert cfg.ffmpeg_transition == "dissolve" + + def test_zoom_effect(self): + """zoom 转场 → FFmpeg zoomin.""" + cfg = TransitionConfig.parse(effect="zoom") + assert cfg.effect == "zoom" + assert cfg.ffmpeg_transition == "zoomin" + + def test_slide_left_alias(self): + """slide_left 别名.""" + cfg = TransitionConfig.parse(effect="slide_left") + assert cfg.effect == "slideleft" + assert cfg.ffmpeg_transition == "slideleft" + + def test_wipe_alias(self): + """wipe 别名 → 默认向左擦.""" + cfg = TransitionConfig.parse(effect="wipe") + assert cfg.effect == "wipeleft" + assert cfg.ffmpeg_transition == "wipeleft" + + def test_circlecrop_effect(self): + """圆形扩散转场.""" + cfg = TransitionConfig.parse(effect="circlecrop") + assert cfg.effect == "circlecrop" + assert cfg.ffmpeg_transition == "circlecrop" + + def test_rectcrop_effect(self): + """矩形扩散转场.""" + cfg = TransitionConfig.parse(effect="rectcrop") + assert cfg.effect == "rectcrop" + assert cfg.ffmpeg_transition == "rectcrop" + + # ── 降级策略 ── + + def test_unsupported_fallback_to_cut(self): + """不支持的转场自动降级为硬切，不阻断渲染.""" + cfg = TransitionConfig.parse(effect="nonexistent_effect") + assert cfg.effect == CUT_TRANSITION + assert cfg.is_cut is True + + def test_unsupported_whitespace_fallback(self): + """带空格的不支持转场也降级.""" + cfg = TransitionConfig.parse(effect=" bad effect ") + assert cfg.effect == CUT_TRANSITION + + # ── 时长边界校验 ── + + def test_default_duration(self): + """默认时长 0.5s.""" + cfg = TransitionConfig.parse(effect="fade") + assert cfg.duration == 0.5 + + def test_duration_within_range(self): + """正常范围内的时长.""" + cfg = TransitionConfig.parse(effect="fade", duration=1.0) + assert cfg.duration == 1.0 + + def test_duration_min_boundary(self): + """最小值边界.""" + cfg = TransitionConfig.parse(effect="fade", duration=MIN_TRANSITION_DURATION) + assert cfg.duration == MIN_TRANSITION_DURATION + + def test_duration_max_boundary(self): + """最大值边界.""" + cfg = TransitionConfig.parse(effect="fade", duration=MAX_TRANSITION_DURATION) + assert cfg.duration == MAX_TRANSITION_DURATION + + def test_duration_below_min_clamped(self): + """低于最小值的时长被钳制.""" + cfg = TransitionConfig.parse(effect="fade", duration=0.1) + assert cfg.duration == MIN_TRANSITION_DURATION + assert cfg.duration >= MIN_TRANSITION_DURATION + + def test_duration_above_max_clamped(self): + """高于最大值的时长被钳制.""" + cfg = TransitionConfig.parse(effect="fade", duration=5.0) + assert cfg.duration == MAX_TRANSITION_DURATION + assert cfg.duration <= MAX_TRANSITION_DURATION + + def test_duration_zero_default_for_effect(self): + """有转场效果但 duration=0 时使用默认值.""" + # 0.0 会被当作小于最小值钳制到 0.3 + cfg = TransitionConfig.parse(effect="fade", duration=0.0) + assert cfg.duration == MIN_TRANSITION_DURATION + + def test_duration_negative_clamped(self): + """负时长被钳制到最小值.""" + cfg = TransitionConfig.parse(effect="fade", duration=-1.0) + assert cfg.duration == MIN_TRANSITION_DURATION + + def test_duration_none_uses_default(self): + """None duration 使用默认值.""" + cfg = TransitionConfig.parse(effect="fade", duration=None) + assert cfg.duration == DEFAULT_TRANSITION_DURATION + + def test_duration_invalid_type(self): + """无效类型的时长使用默认值.""" + cfg = TransitionConfig.parse(effect="fade", duration="abc") # type: ignore + assert cfg.duration == DEFAULT_TRANSITION_DURATION + + # ── cut 的 ffmpeg_transition ── + + def test_cut_ffmpeg_transition_empty(self): + """硬切没有对应的 FFmpeg xfade transition.""" + cfg = TransitionConfig.parse(effect="cut") + assert cfg.ffmpeg_transition == "" + + +# ── TransitionEngine 测试 ──────────────────────────────────────────────────── + + +class TestTransitionEngine: + """TransitionEngine 转场引擎测试.""" + + def test_default_engine(self): + """默认引擎初始化.""" + engine = TransitionEngine() + assert engine is not None + + def test_custom_default_duration(self): + """自定义默认时长.""" + engine = TransitionEngine(default_duration=1.0) + cfg = engine.resolve_config(effect="fade") + assert cfg.duration == 1.0 + + def test_resolve_config_fade(self): + """解析 fade 配置.""" + engine = TransitionEngine() + cfg = engine.resolve_config(effect="fade", duration=0.8) + assert cfg.effect == "fade" + assert cfg.duration == 0.8 + + def test_resolve_config_fallback(self): + """不支持的转场降级.""" + engine = TransitionEngine() + cfg = engine.resolve_config(effect="unknown_effect") + assert cfg.effect == CUT_TRANSITION + assert cfg.is_cut is True + + def test_resolve_config_duration_clamp(self): + """时长边界钳制.""" + engine = TransitionEngine() + cfg = engine.resolve_config(effect="fade", duration=3.0) + assert cfg.duration == MAX_TRANSITION_DURATION + + # ── 批量解析 ── + + def test_resolve_clip_transitions_all_valid(self): + """批量解析全部有效转场.""" + engine = TransitionEngine() + configs = engine.resolve_clip_transitions(["cut", "fade", "dissolve", "slideleft"]) + assert len(configs) == 4 + assert configs[0].effect == "cut" + assert configs[0].is_cut is True + assert configs[1].effect == "fade" + assert configs[2].effect == "dissolve" + assert configs[3].effect == "slideleft" + + def test_resolve_clip_transitions_with_fallback(self): + """批量解析包含不支持的转场，自动降级.""" + engine = TransitionEngine() + configs = engine.resolve_clip_transitions(["fade", "bad_effect", "dissolve", "worse_effect"]) + assert len(configs) == 4 + assert configs[0].effect == "fade" + assert configs[1].effect == "cut" # 降级 + assert configs[2].effect == "dissolve" + assert configs[3].effect == "cut" # 降级 + + def test_resolve_clip_transitions_with_durations(self): + """带时长校验的批量解析（转场时长不超过片段时长的一半）.""" + engine = TransitionEngine(default_duration=1.0) + # 片段只有 1.0s，转场时长被限制在 0.5s + configs = engine.resolve_clip_transitions( + ["fade", "dissolve"], + clip_durations=[1.0, 1.0], + ) + assert len(configs) == 2 + # 1.0s 默认值超过了片段时长的一半 (0.5s)，所以被钳制 + assert configs[0].duration <= 0.5 + assert configs[1].duration <= 0.5 + + def test_resolve_clip_transitions_short_clip_min_bound(self): + """超短片段的转场时长至少为最小值.""" + engine = TransitionEngine() + configs = engine.resolve_clip_transitions( + ["fade"], + clip_durations=[0.1], # 极短片段 + ) + assert len(configs) == 1 + # 0.1 * 0.5 = 0.05 < MIN_TRANSITION_DURATION，所以用最小值 + assert configs[0].duration == MIN_TRANSITION_DURATION + + # ── xfade 滤镜链构建 ── + + def test_build_xfade_single_clip(self): + """单 clip 直接 copy.""" + engine = TransitionEngine() + filter_str, total_dur = engine.build_xfade_chain( + clip_durations=[5.0], + clip_video_labels=["v0"], + transitions=["cut"], + output_label="outv", + ) + assert "copy" in filter_str + assert "[outv]" in filter_str + assert total_dur == pytest.approx(5.0, abs=0.01) + + def test_build_xfade_two_clips_fade(self): + """两个 clip 之间 fade 转场.""" + engine = TransitionEngine() + filter_str, total_dur = engine.build_xfade_chain( + clip_durations=[3.0, 4.0], + clip_video_labels=["v0", "v1"], + transitions=["cut", "fade"], + output_label="outv", + ) + assert "xfade" in filter_str + assert "transition=fade" in filter_str + # 总时长 = 3 + 4 - transition_duration (0.5) = 6.5 + assert total_dur == pytest.approx(6.5, abs=0.1) + + def test_build_xfade_three_clips_mixed(self): + """三个 clip 混合转场.""" + engine = TransitionEngine() + filter_str, total_dur = engine.build_xfade_chain( + clip_durations=[3.0, 4.0, 5.0], + clip_video_labels=["v0", "v1", "v2"], + transitions=["cut", "fade", "dissolve"], + output_label="outv", + ) + assert "xfade" in filter_str + assert "transition=fade" in filter_str + assert "transition=dissolve" in filter_str + # 总时长 ≈ 3 + 4 + 5 - 2 * 0.5 = 11.0 + assert total_dur == pytest.approx(11.0, abs=0.2) + + def test_build_xfade_with_custom_duration(self): + """自定义转场时长.""" + engine = TransitionEngine(default_duration=0.5) + filter_str, total_dur = engine.build_xfade_chain( + clip_durations=[3.0, 4.0], + clip_video_labels=["v0", "v1"], + transitions=["cut", "fade"], + transition_duration=1.0, + output_label="outv", + ) + assert "xfade" in filter_str + # 总时长 = 3 + 4 - 1.0 = 6.0 + assert total_dur == pytest.approx(6.0, abs=0.1) + + def test_build_xfade_zoom_transition(self): + """zoom 转场滤镜构建.""" + engine = TransitionEngine() + filter_str, _ = engine.build_xfade_chain( + clip_durations=[3.0, 4.0], + clip_video_labels=["v0", "v1"], + transitions=["cut", "zoom"], + ) + assert "xfade" in filter_str + assert "transition=zoomin" in filter_str # zoom → zoomin + + def test_build_xfade_slide_directions(self): + """四个方向的滑入转场.""" + engine = TransitionEngine() + for direction in ["slideleft", "slideright", "slideup", "slidedown"]: + filter_str, _ = engine.build_xfade_chain( + clip_durations=[3.0, 4.0], + clip_video_labels=["v0", "v1"], + transitions=["cut", direction], + ) + assert f"transition={direction}" in filter_str + + def test_build_xfade_fallback_transition(self): + """不支持的转场降级后构建（降级为cut，等效于极短fade）.""" + engine = TransitionEngine() + # bad_effect 降级为 cut，cut 使用极短转场 + filter_str, _ = engine.build_xfade_chain( + clip_durations=[3.0, 4.0], + clip_video_labels=["v0", "v1"], + transitions=["cut", "bad_effect"], + ) + # 降级后是 cut，cut 会被 xfade 层映射为 fade（因为 cut 不在 map 里） + # 但时长会很短，所以仍然有 xfade + assert "xfade" in filter_str + + # ── 支持的转场列表 ── + + def test_supported_transitions_list(self): + """获取支持的转场列表（给 API 用）.""" + transitions = TransitionEngine.supported_transitions() + assert len(transitions) >= 10 # cut + 至少 9 种特效 + # 检查结构 + for t in transitions: + assert "name" in t + assert "display_name" in t + assert "category" in t + # 检查分类 + names = [t["name"] for t in transitions] + assert "cut" in names + assert "fade" in names + assert "zoom" in names + assert "circlecrop" in names + + +# ── 集成测试：与 UnifiedRenderService 协作 ──────────────────────────────────── + + +class TestTransitionIntegration: + """转场引擎与统一渲染服务的集成测试.""" + + def test_unified_render_service_has_transition_engine(self): + """UnifiedRenderService 内部有 TransitionEngine 实例.""" + from pathlib import Path + + from video_processing.unified_render_service import UnifiedRenderService + + # 构造最小化的服务实例 + service = UnifiedRenderService( + plan=None, + clips=[], + asset_path_map={}, + work_dir=Path("/tmp"), + ) + assert hasattr(service, "_transition_engine") + assert isinstance(service._transition_engine, TransitionEngine) + + def test_resolved_clip_has_transition_duration(self): + """ResolvedClip 有 transition_duration 字段.""" + from video_processing.unified_render_service import ResolvedClip + + rc = ResolvedClip( + clip_id="test", + asset_id="asset1", + local_path=__file__, # 随便一个存在的路径 + clip_type="main", + order=0, + transition_effect="fade", + transition_duration=0.8, + ) + assert rc.transition_duration == 0.8 + assert rc.transition_effect == "fade"