fix: black格式化

alembic/versions/039_add_transition_duration.py

@@ -1,34 +0,0 @@
-"""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")

alembic/versions/040_add_playback_speed_to_edit_plan_clips.py

@@ -1,29 +0,0 @@
-"""add playback_speed to edit_plan_clips
-
-Revision ID: 040_playback_speed
-Revises: 039_transition_duration
-Create Date: 2026-07-14 10:00:00.000000
-"""
-
-from __future__ import annotations
-
-import sqlalchemy as sa
-
-from alembic import op
-
-# revision identifiers, used by Alembic.
-revision = "040_playback_speed"
-down_revision = "039_transition_duration"
-branch_labels = None
-depends_on = None
-
-
-def upgrade() -> None:
-    op.add_column(
-        "edit_plan_clips",
-        sa.Column("playback_speed", sa.Float(), nullable=False, server_default="1.0"),
-    )
-
-
-def downgrade() -> None:
-    op.drop_column("edit_plan_clips", "playback_speed")

apps/api/app/api/routes/edit_plans.py

@@ -146,7 +146,6 @@ 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="片段额外配置")
@@ -210,8 +209,6 @@ class _PlanClipItem(BaseModel):
     start_time: float
     duration: float
     transition_effect: str
-    transition_duration: float
-    playback_speed: float = 1.0
     status: str
     config: Optional[dict[str, Any]] = None
     created_at: datetime

apps/api/app/api/routes/edit_plans_timeline.py

@@ -210,7 +210,6 @@ 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,

apps/api/app/services/edit_plan_service.py

@@ -281,8 +281,6 @@ class EditPlanService:
         start_time: float = 0.0,
         duration: float = 0.0,
         transition_effect: str = "cut",
-        transition_duration: float = 0.0,
-        playback_speed: float = 1.0,
         config: Optional[dict[str, Any]] = None,
     ) -> EditPlanClip:
         """创建片段
@@ -303,8 +301,6 @@ class EditPlanService:
             start_time=start_time,
             duration=duration,
             transition_effect=transition_effect,
-            transition_duration=transition_duration,
-            playback_speed=playback_speed,
             config=config,
         )
         created = self._clip_repo.create(clip)
@@ -328,8 +324,6 @@ class EditPlanService:
         start_time: Optional[float] = None,
         duration: Optional[float] = None,
         transition_effect: Optional[str] = None,
-        transition_duration: Optional[float] = None,
-        playback_speed: Optional[float] = None,
         config: Optional[dict[str, Any]] = None,
     ) -> EditPlanClip:
         """更新片段
@@ -339,15 +333,6 @@ class EditPlanService:
         """
         existing = self.get_clip_or_raise(clip_id)
 
-        # 速度边界钳制
-        if playback_speed is not None:
-            if playback_speed <= 0:
-                playback_speed = 1.0
-            elif playback_speed < 0.25:
-                playback_speed = 0.25
-            elif playback_speed > 4.0:
-                playback_speed = 4.0
-
         updated = EditPlanClip(
             id=existing.id,
             plan_id=existing.plan_id,
@@ -361,10 +346,6 @@ 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
-            ),
-            playback_speed=playback_speed if playback_speed is not None else existing.playback_speed,
             status=existing.status,
             config=config if config is not None else existing.config,
             created_at=existing.created_at,

apps/worker/video_processing/ffmpeg_utils.py

@@ -25,14 +25,8 @@ 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",
@@ -41,22 +35,9 @@ XFADE_TRANSITION_MAP: dict[str, str] = {
     "slide_up": "slideup",
     "slidedown": "slidedown",
     "slide_down": "slidedown",
-    "slide": "slideleft",  # 默认向左滑
-    # 缩放
-    "zoom": "zoomin",
-    "zoomin": "zoomin",
-    "zoomout": "zoomout",
-    # 擦除系列
-    "wipe": "wipeleft",  # 默认向左擦
+    "dissolve": "dissolve",
+    "wipe": "wipeleft",
     "wipeleft": "wipeleft",
-    "wiperight": "wiperight",
-    "wipeup": "wipeup",
-    "wipedown": "wipedown",
-    # 特殊效果
-    "circlecrop": "circlecrop",
-    "circle": "circlecrop",
-    "rectcrop": "rectcrop",
-    "rect": "rectcrop",
 }
 
 DEFAULT_TRANSITION_DURATION = 0.5

apps/worker/video_processing/render_audio.py

@@ -23,7 +23,6 @@ from typing import TYPE_CHECKING
 
 from video_processing.ffmpeg_utils import FFMPEG_BIN, probe_has_audio, run_ffmpeg
 from video_processing.reverse_engine import ReverseConfig, ReverseEngine
-from video_processing.speed_engine import SpeedEngine
 
 if TYPE_CHECKING:
     from video_processing.unified_render_service import RenderLayer, ResolvedClip
@@ -226,118 +225,53 @@ def concat_main_audio(
         clip = clips[0]
         effective_duration = clip_effective_duration(clip)
         trim_start = getattr(clip, "start_time", 0) or 0
-        speed = getattr(clip, "playback_speed", 1.0) or 1.0
-        if not isinstance(speed, (int, float)) or speed <= 0:
-            speed = 1.0
-
-        # 调速后时长
-        adjusted_duration = effective_duration / speed if abs(speed - 1.0) >= 1e-6 else effective_duration
-        # 最终时长：取调速后时长和视频总时长的较小值
-        final_duration = adjusted_duration
+        # 最终时长：取 clip 有效时长和视频总时长的较小值
+        # （视频总时长由主图层决定，但单 clip 场景下两者应该一致，仍做保护）
+        final_duration = effective_duration
         if video_duration > 0 and (final_duration <= 0 or final_duration > video_duration):
             final_duration = video_duration
 
         # 音频倒放
         reverse_config = ReverseConfig.from_dict(clip.config.get("reverse"))
-        has_reverse = reverse_config.enabled and reverse_config.reverse_audio
-        has_speed = abs(speed - 1.0) >= 1e-6
+        af_filters = []
+        if reverse_config.enabled and reverse_config.reverse_audio:
+            reverse_filter = ReverseEngine.build_audio_filter(reverse_config, duration=effective_duration)
+            if reverse_filter:
+                af_filters.append(reverse_filter)
 
-        if not has_speed and not has_reverse:
-            # 无调速无倒放：简单命令行，-ss 裁剪更高效
-            command = [
-                FFMPEG_BIN,
-                "-y",
-                "-i",
-                str(clip.local_path),
-                "-vn",
-                "-acodec",
-                "aac",
-                "-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))
-            run_ffmpeg(command)
-        else:
-            # 有调速或倒放：用 filter_complex
-            speed_engine = SpeedEngine()
-            audio_filters = []
-            if effective_duration > 0:
-                audio_filters.append(f"atrim=start={trim_start:.3f}:duration={effective_duration:.3f}")
-                audio_filters.append("asetpts=PTS-STARTPTS")
-
-            # 音频调速
-            if has_speed:
-                from video_processing.speed_engine import SpeedConfig
-
-                config = SpeedConfig(speed=float(speed))
-                config.clamp()
-                atempo_filter = speed_engine.build_audio_filter(config)
-                if atempo_filter:
-                    audio_filters.append(atempo_filter)
-
-            # 音频倒放
-            if has_reverse:
-                reverse_filter = ReverseEngine.build_audio_filter(reverse_config, duration=effective_duration)
-                if reverse_filter:
-                    audio_filters.append(reverse_filter)
-
-            filter_parts: list[str] = [f"[0:a]{','.join(audio_filters)}[outa]"]
-            if video_duration > 0 and final_duration < adjusted_duration:
-                filter_parts.append(f"[outa]atrim=0:{final_duration:.3f}[final_audio]")
-                final_label = "final_audio"
-            else:
-                final_label = "outa"
-
-            filter_complex = ";".join(filter_parts)
-            command = [
-                FFMPEG_BIN,
-                "-y",
-                "-i",
-                str(clip.local_path),
-                "-filter_complex",
-                filter_complex,
-                "-map",
-                f"[{final_label}]",
-                "-acodec",
-                "aac",
-                "-b:a",
-                "128k",
-                str(output_path),
-            ]
-            run_ffmpeg(command)
+        command = [
+            FFMPEG_BIN,
+            "-y",
+            "-i",
+            str(clip.local_path),
+            "-vn",
+            "-acodec",
+            "aac",
+            "-b:a",
+            "128k",
+        ]
+        if trim_start > 0:
+            command.extend(["-ss", f"{trim_start:.3f}"])
+        if af_filters:
+            command.extend(["-af", ",".join(af_filters)])
+        if final_duration > 0:
+            command.extend(["-t", f"{final_duration:.3f}"])
+        command.append(str(output_path))
+        run_ffmpeg(command)
         return
 
     # 多 clip，用 filter_complex concat
     input_args: list[str] = []
     filter_parts: list[str] = []
-    speed_engine = SpeedEngine()
 
     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
-        speed = getattr(clip, "playback_speed", 1.0) or 1.0
-        if not isinstance(speed, (int, float)) or speed <= 0:
-            speed = 1.0
-
         audio_filters: list[str] = []
         if effective_duration > 0:
             audio_filters.append(f"atrim=start={trim_start:.3f}:duration={effective_duration:.3f}")
             audio_filters.append("asetpts=PTS-STARTPTS")
-
-        # 音频调速 — atempo 多级串联
-        if abs(speed - 1.0) >= 1e-6:
-            from video_processing.speed_engine import SpeedConfig
-
-            config = SpeedConfig(speed=float(speed))
-            config.clamp()
-            atempo_filter = speed_engine.build_audio_filter(config)
-            if atempo_filter:
-                audio_filters.append(atempo_filter)
         else:
             audio_filters.append("asetpts=PTS-STARTPTS")
 

apps/worker/video_processing/speed_engine.py

@@ -1,167 +0,0 @@
-"""视频调速引擎 — 基于 FFmpeg setpts + atempo 的速度调整能力。
-
-支持：
-- 0.25x ~ 4x 变速范围
-- 视频调速（setpts）
-- 音频调速（atempo，多级串联处理超范围值）
-- 音调修正（pitch_correct，默认开启）
-- 边界自动钳制，不阻断渲染
-"""
-
-from dataclasses import dataclass
-from typing import Optional
-
-# ─── 常量 ───────────────────────────────────────────────
-MIN_SPEED = 0.25
-MAX_SPEED = 4.0
-DEFAULT_SPEED = 1.0
-
-# atempo 单级有效范围
-_ATEMPO_MIN = 0.5
-_ATEMPO_MAX = 2.0
-
-
-@dataclass
-class SpeedConfig:
-    """调速配置。
-
-    Attributes:
-        speed: 播放速度，0.25~4.0，1.0 为原速
-        pitch_correct: 是否保持音调（默认 True，用 atempo 时间拉伸算法）
-    """
-
-    speed: float = DEFAULT_SPEED
-    pitch_correct: bool = True
-
-    @classmethod
-    def parse(cls, data: Optional[dict]) -> "SpeedConfig":
-        """从 dict 解析配置，无效值回退到默认。"""
-        if not data or not isinstance(data, dict):
-            return cls()
-
-        speed = data.get("speed", DEFAULT_SPEED)
-        if not isinstance(speed, (int, float)):
-            speed = DEFAULT_SPEED
-
-        pitch_correct = data.get("pitch_correct", True)
-        if not isinstance(pitch_correct, bool):
-            pitch_correct = True
-
-        config = cls(speed=float(speed), pitch_correct=pitch_correct)
-        config.clamp()
-        return config
-
-    def clamp(self) -> None:
-        """将速度钳制到合法范围。"""
-        if self.speed <= 0:
-            self.speed = DEFAULT_SPEED
-        elif self.speed < MIN_SPEED:
-            self.speed = MIN_SPEED
-        elif self.speed > MAX_SPEED:
-            self.speed = MAX_SPEED
-
-    @property
-    def is_original(self) -> bool:
-        """是否原速（无需调速）。"""
-        return abs(self.speed - 1.0) < 1e-6
-
-
-class SpeedEngine:
-    """调速引擎 — 生成 FFmpeg 调速滤镜链。
-
-    用法：
-        engine = SpeedEngine()
-        video_filter = engine.build_video_filter(config)
-        audio_filter = engine.build_audio_filter(config)
-        new_duration = engine.adjust_duration(duration, config)
-    """
-
-    def build_video_filter(self, config: SpeedConfig) -> str:
-        """生成视频调速滤镜字符串。
-
-        返回 setpts 滤镜表达式，原速时返回空字符串。
-        """
-        if config.is_original:
-            return ""
-        # setpts=PTS/speed — speed>1 加速，speed<1 减速
-        return f"setpts=PTS/{config.speed:.4f}"
-
-    def build_audio_filter(self, config: SpeedConfig) -> str:
-        """生成音频调速滤镜字符串。
-
-        atempo 单级范围 0.5~2.0，超出范围时自动多级串联：
-        - 0.25x → atempo=0.5,atempo=0.5
-        - 4x → atempo=2.0,atempo=2.0
-        - 0.3x → atempo=0.5,atempo=0.6
-        - 3x → atempo=2.0,atempo=1.5
-
-        原速时返回空字符串。
-        """
-        if config.is_original:
-            return ""
-
-        speed = config.speed
-        stages: list[float] = self._split_atempo_stages(speed)
-        return ",".join(f"atempo={s:.4f}" for s in stages)
-
-    @staticmethod
-    def _split_atempo_stages(speed: float) -> list[float]:
-        """将速度拆分为多级 atempo 串联，每级都在 [0.5, 2.0] 范围内。"""
-        if _ATEMPO_MIN <= speed <= _ATEMPO_MAX:
-            return [speed]
-
-        stages: list[float] = []
-        remaining = speed
-
-        # 加速场景（speed > 2.0）
-        if speed > _ATEMPO_MAX:
-            while remaining > _ATEMPO_MAX:
-                stages.append(_ATEMPO_MAX)
-                remaining /= _ATEMPO_MAX
-            stages.append(remaining)
-
-        # 减速场景（speed < 0.5）
-        else:
-            while remaining < _ATEMPO_MIN:
-                stages.append(_ATEMPO_MIN)
-                remaining /= _ATEMPO_MIN
-            stages.append(remaining)
-
-        return stages
-
-    def adjust_duration(self, original_duration: float, config: SpeedConfig) -> float:
-        """计算调速后的时长。
-
-        加速 → 时长变短；减速 → 时长变长。
-        """
-        if config.is_original or original_duration <= 0:
-            return original_duration
-        return original_duration / config.speed
-
-    def build_clip_speed_filter(
-        self,
-        speed: float,
-        pitch_correct: bool = True,
-    ) -> tuple[str, str, SpeedConfig]:
-        """便捷方法：从单一 speed 值生成视频+音频滤镜。
-
-        返回 (video_filter, audio_filter, config)。
-        """
-        config = SpeedConfig(speed=speed, pitch_correct=pitch_correct)
-        config.clamp()
-        return (
-            self.build_video_filter(config),
-            self.build_audio_filter(config),
-            config,
-        )
-
-    @staticmethod
-    def resolve_clip_speed(
-        clip_config: dict,
-        global_speed: float = DEFAULT_SPEED,
-    ) -> float:
-        """从 clip config 中解析 playback_speed，0 或缺失则使用全局速度。"""
-        speed = clip_config.get("playback_speed", 0) if clip_config else 0
-        if not isinstance(speed, (int, float)) or speed <= 0:
-            return global_speed
-        return float(speed)

apps/worker/video_processing/transition_engine.py

@@ -1,381 +0,0 @@
-"""转场特效引擎 — 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"},
-        ]

apps/worker/video_processing/unified_render_service.py

@@ -36,6 +36,7 @@ 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,
@@ -45,10 +46,8 @@ 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.reverse_engine import ReverseConfig, ReverseEngine
-from video_processing.speed_engine import SpeedConfig, SpeedEngine
 from video_processing.sticker_engine import StickerEngine, parse_stickers_from_config
 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
@@ -73,8 +72,6 @@ class ResolvedClip:
     start_time: float = 0.0
     duration: float = 0.0  # 0 表示使用素材完整时长
     transition_effect: str = "cut"
-    transition_duration: float = 0.0  # 0 表示使用全局默认值
-    playback_speed: float = 1.0  # 0 或 1.0 表示原速
     config: dict[str, Any] = field(default_factory=dict)
 
     # 运行时填充
@@ -187,8 +184,6 @@ 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)
-        self._speed_engine = SpeedEngine()
 
     def render(self) -> RenderResult:
         """执行渲染，返回 RenderResult.
@@ -496,7 +491,7 @@ class UnifiedRenderService:
         if not main_layer or not main_layer.clips:
             return 0.0
 
-        total = sum(UnifiedRenderService._clip_adjusted_duration(c) for c in main_layer.clips)
+        total = sum(UnifiedRenderService._clip_effective_duration(c) for c in main_layer.clips)
 
         # 减去转场重叠时间（粗略估算）
         n_clips = len(main_layer.clips)
@@ -1199,8 +1194,6 @@ 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,
-                playback_speed=getattr(clip, "playback_speed", 1.0) or 1.0,
                 config=clip_config,
                 actual_duration=actual_duration,
                 trim_config=effective_trim,
@@ -1298,11 +1291,6 @@ class UnifiedRenderService:
                     filters.append(f"trim=duration={effective_duration:.3f}")
                 filters.append("setpts=PTS-STARTPTS")
 
-            # 调速 — 基于 setpts 改变播放速度
-            speed = UnifiedRenderService._clip_speed(clip)
-            if abs(speed - 1.0) >= 1e-6:
-                filters.append(f"setpts=PTS/{speed:.4f}")
-
             # 倒放滤镜（在 trim 之后、scale 之前应用）
             reverse_config = ReverseConfig.from_dict(clip.config.get("reverse"))
             if reverse_config.enabled and reverse_config.reverse_video:
@@ -1360,28 +1348,21 @@ class UnifiedRenderService:
         for layer in layers:
             layer_clip_indices = [all_clips.index(c) for c in layer.clips]
             layer_labels = [preprocessed_labels[i] for i in layer_clip_indices]
-            # 使用调速后的实际时长，与 Step 1 的调速处理保持一致
-            layer_durations = [UnifiedRenderService._clip_adjusted_duration(all_clips[i]) for i in layer_clip_indices]
+            # 使用 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 层，用 TransitionEngine 构建转场链
+                # 多 clip 层，用 xfade 串联
                 out_label = f"{layer.role}_merged"
-                # 计算该层使用的转场时长（取首个非零值，否则用默认）
-                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(
+                xfade_filter, _ = build_xfade_filter_chain(
                     clip_durations=layer_durations,
                     clip_video_labels=layer_labels,
                     transitions=layer_transitions,
-                    transition_duration=layer_dur if layer_dur > 0 else None,
+                    transition_duration=self.transition_duration,
                     output_label=out_label,
                 )
                 if xfade_filter:
@@ -1606,7 +1587,7 @@ class UnifiedRenderService:
 
     @staticmethod
     def _clip_effective_duration(clip: ResolvedClip) -> float:
-        """计算 clip 的有效时长（原速 trim 后时长）."""
+        """计算 clip 的有效时长."""
         if clip.duration > 0:
             return min(clip.duration, clip.actual_duration) if clip.actual_duration > 0 else clip.duration
         return clip.actual_duration if clip.actual_duration > 0 else 0.0
@@ -1703,20 +1684,3 @@ class UnifiedRenderService:
         )
 
         return new_filter, new_input_args
-
-    @staticmethod
-    def _clip_speed(clip: ResolvedClip) -> float:
-        """获取 clip 的播放速度，无效值回退到 1.0."""
-        speed = getattr(clip, "playback_speed", 1.0)
-        if not isinstance(speed, (int, float)) or speed <= 0:
-            return 1.0
-        return float(speed)
-
-    @staticmethod
-    def _clip_adjusted_duration(clip: ResolvedClip) -> float:
-        """计算调速后的 clip 实际时长（用于拼接计算）."""
-        base = UnifiedRenderService._clip_effective_duration(clip)
-        speed = UnifiedRenderService._clip_speed(clip)
-        if abs(speed - 1.0) < 1e-6:
-            return base
-        return base / speed

docs/schema-metadata-snapshot.json

@@ -858,22 +858,6 @@
           "type": "VARCHAR(20)",
           "unique": false
         },
-        {
-          "index": false,
-          "name": "transition_duration",
-          "nullable": false,
-          "primary_key": false,
-          "type": "FLOAT",
-          "unique": false
-        },
-        {
-          "index": false,
-          "name": "playback_speed",
-          "nullable": false,
-          "primary_key": false,
-          "type": "FLOAT",
-          "unique": false
-        },
         {
           "index": true,
           "name": "status",

packages/adapters/sqlalchemy_impl/edit_plan_clip_repository.py

@@ -54,8 +54,6 @@ class SQLAlchemyEditPlanClipRepository:
             start_time=clip.start_time,
             duration=clip.duration,
             transition_effect=clip.transition_effect,
-            transition_duration=clip.transition_duration,
-            playback_speed=clip.playback_speed,
             status=clip.status,
             config=clip.config,
         )
@@ -78,8 +76,6 @@ 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.playback_speed = clip.playback_speed
         model.status = clip.status
         model.config = clip.config
         model.updated_at = clip.updated_at
@@ -124,8 +120,6 @@ 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,
-            playback_speed=model.playback_speed or 1.0,
             status=EditPlanClipStatus(model.status) if model.status else EditPlanClipStatus.PENDING,
             config=model.config or {},
             created_at=model.created_at,

packages/adapters/sqlalchemy_impl/models.py

@@ -196,8 +196,6 @@ 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)
-    playback_speed = Column(Float, nullable=False, default=1.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))

packages/domain/edit_plan_clip.py

@@ -50,8 +50,6 @@ class EditPlanClip:
     start_time: float = 0.0
     duration: float = 0.0
     transition_effect: str = "cut"
-    transition_duration: float = 0.0  # 0 表示使用全局默认值
-    playback_speed: float = 1.0  # 0 或 1.0 表示原速，范围 0.25~4.0
     status: EditPlanClipStatus = EditPlanClipStatus.PENDING
     config: dict[str, Any] = field(default_factory=dict)
     created_at: datetime = field(default_factory=lambda: datetime.now(timezone.utc))
@@ -70,8 +68,6 @@ class EditPlanClip:
         start_time: float = 0.0,
         duration: float = 0.0,
         transition_effect: str = "cut",
-        transition_duration: float = 0.0,
-        playback_speed: float = 1.0,
         config: dict[str, Any] | None = None,
     ) -> EditPlanClip:
         """创建剪辑计划片段"""
@@ -83,13 +79,6 @@ class EditPlanClip:
             raise ValueError("start_time 不能为负数")
         if duration < 0:
             raise ValueError("duration 不能为负数")
-        # 速度边界钳制
-        if playback_speed <= 0:
-            playback_speed = 1.0
-        elif playback_speed < 0.25:
-            playback_speed = 0.25
-        elif playback_speed > 4.0:
-            playback_speed = 4.0
 
         return cls(
             id=uuid4().hex,
@@ -102,8 +91,6 @@ class EditPlanClip:
             start_time=start_time,
             duration=duration,
             transition_effect=transition_effect.strip() or "cut",
-            transition_duration=max(0.0, transition_duration),
-            playback_speed=playback_speed,
             status=EditPlanClipStatus.PENDING,
             config=config or {},
         )

scripts/gray_deploy.sh

@@ -1,117 +0,0 @@
-#!/bin/bash
-# 灰度发布脚本：通过Nginx权重调整流量比例
-# 用法: ./scripts/gray_deploy.sh <版本号> <灰度百分比>
-#
-# 需要在目标服务器上执行，或通过SSH执行
-# 前提：服务器上运行两个版本的容器（stable + canary），Nginx做加权轮询
-
-set -euo pipefail
-
-VERSION="${1:-}"
-GRAY_PCT="${2:-10}"
-
-if [[ -z "$VERSION" ]]; then
-    echo "用法: $0 <版本号> [灰度百分比]"
-    echo "示例: $0 v0.1.129 5"
-    exit 1
-fi
-
-STABLE_VERSION="${STABLE_VERSION:-current}"
-REGISTRY="${REGISTRY:-git.xiaoxiajianji.com/xiaoxia/xiaoxia-saas}"
-
-echo "=========================================="
-echo "  灰度发布"
-echo "  新版本: $VERSION"
-echo "  灰度比例: ${GRAY_PCT}%"
-echo "  稳定版本: $STABLE_VERSION"
-echo "=========================================="
-
-# 1. 拉取新版本镜像
-echo ""
-echo ">>> 拉取新版本镜像..."
-for component in api worker web; do
-    echo "  拉取 $component:$VERSION ..."
-    docker pull "${REGISTRY}-${component}:${VERSION}" 2>&1 | tail -1
-done
-
-# 2. 启动灰度版本容器（canary）
-echo ""
-echo ">>> 启动灰度版本容器..."
-
-# API canary
-CANARY_API_NAME="saas-api-canary"
-if docker ps -a --format '{{.Names}}' | grep -q "^${CANARY_API_NAME}$"; then
-    echo "  停止旧 canary 容器..."
-    docker stop "$CANARY_API_NAME" 2>/dev/null || true
-    docker rm "$CANARY_API_NAME" 2>/dev/null || true
-fi
-
-echo "  启动 api canary..."
-docker run -d \
-    --name "$CANARY_API_NAME" \
-    --network saas-network \
-    -e DATABASE_URL="${DATABASE_URL}" \
-    -e REDIS_URL="${REDIS_URL}" \
-    -e FEATURE_FLAG_PROVIDER=redis \
-    --restart unless-stopped \
-    "${REGISTRY}-api:${VERSION}"
-
-# Worker canary
-CANARY_WORKER_NAME="saas-worker-canary"
-if docker ps -a --format '{{.Names}}' | grep -q "^${CANARY_WORKER_NAME}$"; then
-    echo "  停止旧 worker canary..."
-    docker stop "$CANARY_WORKER_NAME" 2>/dev/null || true
-    docker rm "$CANARY_WORKER_NAME" 2>/dev/null || true
-fi
-
-echo "  启动 worker canary..."
-docker run -d \
-    --name "$CANARY_WORKER_NAME" \
-    --network saas-network \
-    -e DATABASE_URL="${DATABASE_URL}" \
-    -e REDIS_URL="${REDIS_URL}" \
-    --restart unless-stopped \
-    "${REGISTRY}-worker:${VERSION}"
-
-# 3. 等待容器健康
-echo ""
-echo ">>> 等待容器健康..."
-sleep 10
-if ! docker ps --format '{{.Names}} {{.Status}}' | grep -q "$CANARY_API_NAME"; then
-    echo "错误: API canary 容器未运行"
-    docker logs "$CANARY_API_NAME" --tail 20
-    exit 1
-fi
-echo "  ✅ API canary 运行中"
-
-# 4. 更新Nginx权重
-echo ""
-echo ">>> 更新Nginx权重 (稳定: $((100-GRAY_PCT))% / 灰度: ${GRAY_PCT}%)..."
-
-NGINX_CONF="${NGINX_CONF:-/etc/nginx/conf.d/saas-api.conf}"
-if [[ -f "$NGINX_CONF" ]]; then
-    # 备份
-    cp "$NGINX_CONF" "${NGINX_CONF}.bak.$(date +%Y%m%d%H%M%S)"
-    
-    # 更新 upstream 权重（需要根据实际配置调整）
-    echo "  请手动更新 Nginx upstream 配置中的权重"
-    echo "  示例配置:"
-    cat <<EOF
-    upstream saas_api_backend {
-        server saas-api:8000 weight=$((100-GRAY_PCT));
-        server saas-api-canary:8000 weight=${GRAY_PCT};
-    }
-EOF
-    nginx -t && nginx -s reload
-    echo "  ✅ Nginx 已reload"
-else
-    echo "  警告: Nginx 配置文件不存在 ($NGINX_CONF)"
-    echo "  请手动配置灰度流量权重"
-fi
-
-echo ""
-echo "=========================================="
-echo "  ✅ 灰度发布完成"
-echo "  新版本: $VERSION (${GRAY_PCT}%流量)"
-echo "  监控: Grafana / 日志"
-echo "=========================================="

scripts/release.sh

@@ -1,126 +0,0 @@
-#!/bin/bash
-# 一键发布脚本：打tag → 触发生产镜像构建 → 部署到灰度
-# 用法: ./scripts/release.sh v0.1.129 [--gray 5]
-
-set -euo pipefail
-
-SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
-REPO_ROOT="$(cd "$SCRIPT_DIR/.." && pwd)"
-
-usage() {
-    echo "用法: $0 <版本号> [--gray 百分比] [--no-deploy]"
-    echo ""
-    echo "示例:"
-    echo "  $0 v0.1.129                    # 打tag + 全量发布"
-    echo "  $0 v0.1.129 --gray 5           # 打tag + 5%灰度发布"
-    echo "  $0 v0.1.129 --no-deploy        # 只打tag，不部署"
-    exit 1
-}
-
-# 参数解析
-VERSION=""
-GRAY_PCT=0
-DEPLOY=true
-
-while [[ $# -gt 0 ]]; do
-    case "$1" in
-        --gray)
-            GRAY_PCT="$2"
-            shift 2
-            ;;
-        --no-deploy)
-            DEPLOY=false
-            shift
-            ;;
-        -h|--help)
-            usage
-            ;;
-        v*)
-            VERSION="$1"
-            shift
-            ;;
-        *)
-            echo "未知参数: $1"
-            usage
-            ;;
-    esac
-done
-
-if [[ -z "$VERSION" ]]; then
-    echo "错误: 请指定版本号（如 v0.1.129）"
-    usage
-fi
-
-echo "=========================================="
-echo "  发布版本: $VERSION"
-echo "  灰度比例: ${GRAY_PCT}%"
-echo "  自动部署: $DEPLOY"
-echo "=========================================="
-
-cd "$REPO_ROOT"
-
-# 1. 确认在 develop 分支
-CURRENT_BRANCH=$(git rev-parse --abbrev-ref HEAD)
-if [[ "$CURRENT_BRANCH" != "develop" ]]; then
-    echo "错误: 请切换到 develop 分支后再发布"
-    exit 1
-fi
-
-# 2. 拉取最新代码
-echo ""
-echo ">>> 拉取最新代码..."
-git pull origin develop
-
-# 3. 生成 CHANGELOG
-echo ""
-echo ">>> 生成 CHANGELOG..."
-if [[ -f "scripts/generate_changelog.py" ]]; then
-    PREV_TAG=$(git describe --tags --abbrev=0 HEAD^ 2>/dev/null || echo "")
-    if [[ -n "$PREV_TAG" ]]; then
-        python3 scripts/generate_changelog.py \
-            --from-tag "$PREV_TAG" \
-            --to-tag HEAD \
-            --gitea-token "${GITEA_TOKEN:-}" \
-            --output /tmp/changelog_$$.md
-        echo "CHANGELOG 已生成到 /tmp/changelog_$$.md"
-    fi
-fi
-
-# 4. 打tag
-echo ""
-echo ">>> 打 tag $VERSION ..."
-if git rev-parse "$VERSION" >/dev/null 2>&1; then
-    echo "警告: tag $VERSION 已存在，跳过打tag"
-else
-    git tag -a "$VERSION" -m "Release $VERSION"
-    git push origin "$VERSION"
-    echo "Tag $VERSION 已推送，触发生产镜像构建..."
-fi
-
-# 5. 等待镜像构建
-if [[ "$DEPLOY" == "true" ]]; then
-    echo ""
-    echo ">>> 等待镜像构建完成（约10-15分钟）..."
-    echo "    镜像: git.xiaoxiajianji.com/xiaoxia/xiaoxia-saas-{api,worker,web}:$VERSION"
-    
-    # 这里可以加镜像存在性检查
-    echo "    （镜像构建由CI自动完成，请在Gitea Actions中确认）"
-fi
-
-# 6. 灰度部署
-if [[ "$DEPLOY" == "true" && "$GRAY_PCT" -gt 0 ]]; then
-    echo ""
-    echo ">>> 灰度部署: ${GRAY_PCT}% 流量到 $VERSION"
-    if [[ -f "scripts/gray_deploy.sh" ]]; then
-        ./scripts/gray_deploy.sh "$VERSION" "$GRAY_PCT"
-    else
-        echo "警告: gray_deploy.sh 不存在，跳过灰度部署"
-    fi
-fi
-
-echo ""
-echo "=========================================="
-echo "  ✅ 发布流程完成"
-echo "  版本: $VERSION"
-echo "  灰度: ${GRAY_PCT}%"
-echo "=========================================="

scripts/rollback.sh

@@ -1,42 +0,0 @@
-#!/bin/bash
-# 灰度回滚脚本：切回稳定版本流量
-# 用法: ./scripts/rollback.sh [稳定版本号]
-
-set -euo pipefail
-
-STABLE_VERSION="${1:-current}"
-
-echo "=========================================="
-echo "  灰度回滚"
-echo "  切回稳定版本: $STABLE_VERSION"
-echo "=========================================="
-
-# 1. 恢复Nginx全量到稳定版本
-echo ""
-echo ">>> 恢复Nginx全量流量到稳定版本..."
-
-NGINX_CONF="${NGINX_CONF:-/etc/nginx/conf.d/saas-api.conf}"
-if [[ -f "$NGINX_CONF" ]]; then
-    # 找最近的备份
-    LATEST_BAK=$(ls -t "${NGINX_CONF}".bak.* 2>/dev/null | head -1)
-    if [[ -n "$LATEST_BAK" ]]; then
-        cp "$LATEST_BAK" "$NGINX_CONF"
-        echo "  从备份恢复: $LATEST_BAK"
-    else
-        echo "  未找到备份，请手动移除 canary upstream"
-    fi
-    
-    nginx -t && nginx -s reload
-    echo "  ✅ Nginx 已回滚"
-fi
-
-# 2. 停止灰度版本容器（保留30分钟以便排查）
-echo ""
-echo ">>> 灰度版本容器将在30分钟后停止（便于排查问题）"
-echo "    立即停止请执行: docker stop saas-api-canary saas-worker-canary"
-
-echo ""
-echo "=========================================="
-echo "  ✅ 回滚完成"
-echo "  流量已全部切回稳定版本"
-echo "=========================================="

tests/integration/test_generation_api.py

@@ -138,70 +138,6 @@ class StubGenerationTaskRepository:
     def list_by_source_edit_plan(self, plan_id: str) -> list[GenerationTask]:
         return [t for t in self._tasks.values() if getattr(t, "source_edit_plan_id", "") == plan_id]
 
-    def list_by_user_filtered(
-        self,
-        user_id: str,
-        *,
-        status: str | None = None,
-        limit: int | None = None,
-        offset: int = 0,
-    ) -> list:
-        """按用户+状态筛选任务列表（stub实现）。"""
-        items = [t for t in self._tasks.values() if t.created_by_user_id == user_id]
-        if status:
-            items = [t for t in items if str(t.status) == status]
-        # 按创建时间倒序
-        items.sort(key=lambda t: t.created_at or "", reverse=True)
-        if offset:
-            items = items[offset:]
-        if limit is not None:
-            items = items[:limit]
-        return items
-
-    def count_by_user_filtered(
-        self,
-        user_id: str,
-        *,
-        status: str | None = None,
-    ) -> int:
-        """按用户+状态筛选计数（stub实现）。"""
-        items = [t for t in self._tasks.values() if t.created_by_user_id == user_id]
-        if status:
-            items = [t for t in items if str(t.status) == status]
-        return len(items)
-
-    def list_by_project_filtered(
-        self,
-        project_id: str,
-        *,
-        status: str | None = None,
-        limit: int | None = None,
-        offset: int = 0,
-    ) -> list:
-        """按项目+状态筛选任务列表（stub实现）。"""
-        items = [t for t in self._tasks.values() if t.project_id == project_id]
-        if status:
-            items = [t for t in items if str(t.status) == status]
-        # 按创建时间倒序
-        items.sort(key=lambda t: t.created_at or "", reverse=True)
-        if offset:
-            items = items[offset:]
-        if limit is not None:
-            items = items[:limit]
-        return items
-
-    def count_by_project_filtered(
-        self,
-        project_id: str,
-        *,
-        status: str | None = None,
-    ) -> int:
-        """按项目+状态筛选计数（stub实现）。"""
-        items = [t for t in self._tasks.values() if t.project_id == project_id]
-        if status:
-            items = [t for t in items if str(t.status) == status]
-        return len(items)
-
 
 class StubGeneratedVideoRepository:
     def __init__(self, videos: dict[str, GeneratedVideo] | None = None):

tests/integration/test_task_center_api.py

@@ -103,70 +103,6 @@ class StubGenerationTaskRepository:
     def list_by_source_edit_plan(self, plan_id: str) -> list[GenerationTask]:
         return [t for t in self._tasks.values() if getattr(t, "source_edit_plan_id", "") == plan_id]
 
-    def list_by_user_filtered(
-        self,
-        user_id: str,
-        *,
-        status: str | None = None,
-        limit: int | None = None,
-        offset: int = 0,
-    ) -> list:
-        """按用户+状态筛选任务列表（stub实现）。"""
-        items = [t for t in self._tasks.values() if t.created_by_user_id == user_id]
-        if status:
-            items = [t for t in items if str(t.status) == status]
-        # 按创建时间倒序
-        items.sort(key=lambda t: t.created_at or "", reverse=True)
-        if offset:
-            items = items[offset:]
-        if limit is not None:
-            items = items[:limit]
-        return items
-
-    def count_by_user_filtered(
-        self,
-        user_id: str,
-        *,
-        status: str | None = None,
-    ) -> int:
-        """按用户+状态筛选计数（stub实现）。"""
-        items = [t for t in self._tasks.values() if t.created_by_user_id == user_id]
-        if status:
-            items = [t for t in items if str(t.status) == status]
-        return len(items)
-
-    def list_by_project_filtered(
-        self,
-        project_id: str,
-        *,
-        status: str | None = None,
-        limit: int | None = None,
-        offset: int = 0,
-    ) -> list:
-        """按项目+状态筛选任务列表（stub实现）。"""
-        items = [t for t in self._tasks.values() if t.project_id == project_id]
-        if status:
-            items = [t for t in items if str(t.status) == status]
-        # 按创建时间倒序
-        items.sort(key=lambda t: t.created_at or "", reverse=True)
-        if offset:
-            items = items[offset:]
-        if limit is not None:
-            items = items[:limit]
-        return items
-
-    def count_by_project_filtered(
-        self,
-        project_id: str,
-        *,
-        status: str | None = None,
-    ) -> int:
-        """按项目+状态筛选计数（stub实现）。"""
-        items = [t for t in self._tasks.values() if t.project_id == project_id]
-        if status:
-            items = [t for t in items if str(t.status) == status]
-        return len(items)
-
 
 class StubIngestJobRepository:
     def __init__(self, jobs: dict[str, IngestJob] | None = None):
@@ -534,8 +470,8 @@ class TestRetryProjectTask:
         assert data["task_type"] == "generation"
         assert data["status"] == "pending"
         assert "current_step" in data
-        # 原地重试：source_id 保持不变（复用同一个任务）
-        assert data["source_id"] == "gen-failed-1"
+        # 验证新任务的 ID 不同于原任务
+        assert data["source_id"] != "gen-failed-1"
         # 验证 Celery 任务被发送
         assert mock_celery.send_task.called
 
@@ -694,13 +630,11 @@ class TestTaskCenterCrossEndpoint:
         retry_resp = tc.post("/tasks/gen-fail-cross/retry")
         assert retry_resp.status_code == 200
 
-        # 3. 再次列出：原地重试，任务数不变（仍是1个），但状态变为 pending
+        # 3. 再次列出，应有2个任务（旧的failed + 新的pending）
         list_resp2 = tc.get("/tasks")
         assert list_resp2.status_code == 200
         items2 = list_resp2.json()["items"]
-        assert len(items2) == 1
-        assert items2[0]["status"] == "pending"
-        assert items2[0]["source_id"] == "gen-fail-cross"
+        assert len(items2) == 2
 
         test_app.dependency_overrides.clear()
 

tests/unit/test_edit_plan_generation_api.py

@@ -193,70 +193,6 @@ class StubGenerationTaskRepository:
         items.sort(key=lambda t: t.created_at, reverse=True)
         return items[:limit]
 
-    def list_by_user_filtered(
-        self,
-        user_id: str,
-        *,
-        status: str | None = None,
-        limit: int | None = None,
-        offset: int = 0,
-    ) -> list:
-        """按用户+状态筛选任务列表（stub实现）。"""
-        items = [t for t in self._store.values() if t.created_by_user_id == user_id]
-        if status:
-            items = [t for t in items if str(t.status) == status]
-        # 按创建时间倒序
-        items.sort(key=lambda t: t.created_at or "", reverse=True)
-        if offset:
-            items = items[offset:]
-        if limit is not None:
-            items = items[:limit]
-        return items
-
-    def count_by_user_filtered(
-        self,
-        user_id: str,
-        *,
-        status: str | None = None,
-    ) -> int:
-        """按用户+状态筛选计数（stub实现）。"""
-        items = [t for t in self._store.values() if t.created_by_user_id == user_id]
-        if status:
-            items = [t for t in items if str(t.status) == status]
-        return len(items)
-
-    def list_by_project_filtered(
-        self,
-        project_id: str,
-        *,
-        status: str | None = None,
-        limit: int | None = None,
-        offset: int = 0,
-    ) -> list:
-        """按项目+状态筛选任务列表（stub实现）。"""
-        items = [t for t in self._store.values() if t.project_id == project_id]
-        if status:
-            items = [t for t in items if str(t.status) == status]
-        # 按创建时间倒序
-        items.sort(key=lambda t: t.created_at or "", reverse=True)
-        if offset:
-            items = items[offset:]
-        if limit is not None:
-            items = items[:limit]
-        return items
-
-    def count_by_project_filtered(
-        self,
-        project_id: str,
-        *,
-        status: str | None = None,
-    ) -> int:
-        """按项目+状态筛选计数（stub实现）。"""
-        items = [t for t in self._store.values() if t.project_id == project_id]
-        if status:
-            items = [t for t in items if str(t.status) == status]
-        return len(items)
-
 
 # ── Fixtures ──────────────────────────────────────────────────────────────────
 

tests/unit/test_edit_plan_service.py

@@ -200,70 +200,6 @@ class StubGenerationTaskRepository:
     def count_pending_total(self) -> int:
         return 0
 
-    def list_by_user_filtered(
-        self,
-        user_id: str,
-        *,
-        status: str | None = None,
-        limit: int | None = None,
-        offset: int = 0,
-    ) -> list:
-        """按用户+状态筛选任务列表（stub实现）。"""
-        items = [t for t in self._tasks.values() if t.created_by_user_id == user_id]
-        if status:
-            items = [t for t in items if str(t.status) == status]
-        # 按创建时间倒序
-        items.sort(key=lambda t: t.created_at or "", reverse=True)
-        if offset:
-            items = items[offset:]
-        if limit is not None:
-            items = items[:limit]
-        return items
-
-    def count_by_user_filtered(
-        self,
-        user_id: str,
-        *,
-        status: str | None = None,
-    ) -> int:
-        """按用户+状态筛选计数（stub实现）。"""
-        items = [t for t in self._tasks.values() if t.created_by_user_id == user_id]
-        if status:
-            items = [t for t in items if str(t.status) == status]
-        return len(items)
-
-    def list_by_project_filtered(
-        self,
-        project_id: str,
-        *,
-        status: str | None = None,
-        limit: int | None = None,
-        offset: int = 0,
-    ) -> list:
-        """按项目+状态筛选任务列表（stub实现）。"""
-        items = [t for t in self._tasks.values() if t.project_id == project_id]
-        if status:
-            items = [t for t in items if str(t.status) == status]
-        # 按创建时间倒序
-        items.sort(key=lambda t: t.created_at or "", reverse=True)
-        if offset:
-            items = items[offset:]
-        if limit is not None:
-            items = items[:limit]
-        return items
-
-    def count_by_project_filtered(
-        self,
-        project_id: str,
-        *,
-        status: str | None = None,
-    ) -> int:
-        """按项目+状态筛选计数（stub实现）。"""
-        items = [t for t in self._tasks.values() if t.project_id == project_id]
-        if status:
-            items = [t for t in items if str(t.status) == status]
-        return len(items)
-
 
 # ---------------------------------------------------------------------------
 # Service factory

tests/unit/test_generation_presigned_url.py

@@ -64,38 +64,6 @@ class StubGenerationTaskRepository:
     def count_pending_total(self):
         return 0
 
-    def list_by_user_filtered(self, user_id, *, status=None, limit=None, offset=0):
-        items = [t for t in self._tasks.values() if getattr(t, "created_by_user_id", None) == user_id]
-        if status:
-            items = [t for t in items if getattr(t, "status", None) == status]
-        if offset:
-            items = items[offset:]
-        if limit is not None:
-            items = items[:limit]
-        return items
-
-    def count_by_user_filtered(self, user_id, *, status=None):
-        items = [t for t in self._tasks.values() if getattr(t, "created_by_user_id", None) == user_id]
-        if status:
-            items = [t for t in items if getattr(t, "status", None) == status]
-        return len(items)
-
-    def list_by_project_filtered(self, project_id, *, status=None, limit=None, offset=0):
-        items = [t for t in self._tasks.values() if getattr(t, "project_id", None) == project_id]
-        if status:
-            items = [t for t in items if getattr(t, "status", None) == status]
-        if offset:
-            items = items[offset:]
-        if limit is not None:
-            items = items[:limit]
-        return items
-
-    def count_by_project_filtered(self, project_id, *, status=None):
-        items = [t for t in self._tasks.values() if getattr(t, "project_id", None) == project_id]
-        if status:
-            items = [t for t in items if getattr(t, "status", None) == status]
-        return len(items)
-
 
 class StubGeneratedVideoRepository:
     def __init__(self, videos=None):

tests/unit/test_speed_engine.py

@@ -1,269 +0,0 @@
-"""视频调速引擎单元测试."""
-
-import pytest
-from video_processing.speed_engine import (
-    MAX_SPEED,
-    MIN_SPEED,
-    SpeedConfig,
-    SpeedEngine,
-)
-
-# ─── SpeedConfig 解析与校验 ──────────────────────────────────
-
-
-class TestSpeedConfig:
-    def test_default_values(self):
-        config = SpeedConfig()
-        assert config.speed == 1.0
-        assert config.pitch_correct is True
-
-    def test_parse_none(self):
-        config = SpeedConfig.parse(None)
-        assert config.speed == 1.0
-        assert config.pitch_correct is True
-
-    def test_parse_empty_dict(self):
-        config = SpeedConfig.parse({})
-        assert config.speed == 1.0
-
-    def test_parse_valid_speed(self):
-        config = SpeedConfig.parse({"speed": 2.0})
-        assert config.speed == 2.0
-
-    def test_parse_pitch_correct_false(self):
-        config = SpeedConfig.parse({"pitch_correct": False})
-        assert config.pitch_correct is False
-
-    def test_parse_invalid_speed_type(self):
-        config = SpeedConfig.parse({"speed": "fast"})
-        assert config.speed == 1.0
-
-    def test_parse_invalid_pitch_type(self):
-        config = SpeedConfig.parse({"pitch_correct": "yes"})
-        assert config.pitch_correct is True
-
-    def test_clamp_below_min(self):
-        config = SpeedConfig(speed=0.1)
-        config.clamp()
-        assert config.speed == MIN_SPEED
-
-    def test_clamp_zero(self):
-        config = SpeedConfig(speed=0)
-        config.clamp()
-        assert config.speed == 1.0
-
-    def test_clamp_negative(self):
-        config = SpeedConfig(speed=-1.0)
-        config.clamp()
-        assert config.speed == 1.0
-
-    def test_clamp_above_max(self):
-        config = SpeedConfig(speed=10.0)
-        config.clamp()
-        assert config.speed == MAX_SPEED
-
-    def test_clamp_within_range(self):
-        config = SpeedConfig(speed=1.5)
-        config.clamp()
-        assert config.speed == 1.5
-
-    def test_is_original_true(self):
-        config = SpeedConfig(speed=1.0)
-        assert config.is_original is True
-
-    def test_is_original_false(self):
-        config = SpeedConfig(speed=1.5)
-        assert config.is_original is False
-
-    def test_parse_clamps_automatically(self):
-        """parse 方法应该自动调用 clamp."""
-        config = SpeedConfig.parse({"speed": 100.0})
-        assert config.speed == MAX_SPEED
-
-
-# ─── SpeedEngine 视频滤镜 ────────────────────────────────────
-
-
-class TestSpeedEngineVideoFilter:
-    def setup_method(self):
-        self.engine = SpeedEngine()
-
-    def test_original_speed_returns_empty(self):
-        config = SpeedConfig(speed=1.0)
-        assert self.engine.build_video_filter(config) == ""
-
-    def test_double_speed(self):
-        config = SpeedConfig(speed=2.0)
-        result = self.engine.build_video_filter(config)
-        assert "setpts=PTS/2.0" in result
-
-    def test_half_speed(self):
-        config = SpeedConfig(speed=0.5)
-        result = self.engine.build_video_filter(config)
-        assert "setpts=PTS/0.5" in result
-
-    def test_quarter_speed(self):
-        config = SpeedConfig(speed=0.25)
-        result = self.engine.build_video_filter(config)
-        assert "setpts=PTS/0.25" in result
-
-    def test_quad_speed(self):
-        config = SpeedConfig(speed=4.0)
-        result = self.engine.build_video_filter(config)
-        assert "setpts=PTS/4.0" in result
-
-
-# ─── SpeedEngine 音频滤镜（atempo 多级串联） ─────────────────
-
-
-class TestSpeedEngineAudioFilter:
-    def setup_method(self):
-        self.engine = SpeedEngine()
-
-    def test_original_speed_returns_empty(self):
-        config = SpeedConfig(speed=1.0)
-        assert self.engine.build_audio_filter(config) == ""
-
-    def test_double_speed_single_stage(self):
-        """2x 在 atempo 单级范围内，只需一个 atempo."""
-        config = SpeedConfig(speed=2.0)
-        result = self.engine.build_audio_filter(config)
-        assert result == "atempo=2.0000"
-
-    def test_half_speed_single_stage(self):
-        config = SpeedConfig(speed=0.5)
-        result = self.engine.build_audio_filter(config)
-        assert result == "atempo=0.5000"
-
-    def test_quad_speed_two_stages(self):
-        """4x 需要两级 atempo: 2.0 * 2.0."""
-        config = SpeedConfig(speed=4.0)
-        result = self.engine.build_audio_filter(config)
-        assert result == "atempo=2.0000,atempo=2.0000"
-
-    def test_quarter_speed_two_stages(self):
-        """0.25x 需要两级 atempo: 0.5 * 0.5."""
-        config = SpeedConfig(speed=0.25)
-        result = self.engine.build_audio_filter(config)
-        assert result == "atempo=0.5000,atempo=0.5000"
-
-    def test_triple_speed_two_stages(self):
-        """3x: 2.0 * 1.5."""
-        config = SpeedConfig(speed=3.0)
-        result = self.engine.build_audio_filter(config)
-        parts = result.split(",")
-        assert len(parts) == 2
-        assert "atempo=2.0000" in parts
-        assert "atempo=1.5000" in parts
-
-    def test_03_speed_two_stages(self):
-        """0.3x: 0.5 * 0.6."""
-        config = SpeedConfig(speed=0.3)
-        result = self.engine.build_audio_filter(config)
-        parts = result.split(",")
-        assert len(parts) == 2
-        assert "atempo=0.5000" in parts
-        assert "atempo=0.6000" in parts
-
-    def test_split_atempo_inside_range(self):
-        """0.5~2.0 范围内只返回一级."""
-        stages = SpeedEngine._split_atempo_stages(1.5)
-        assert len(stages) == 1
-        assert stages[0] == 1.5
-
-    def test_split_atempo_boundary_min(self):
-        stages = SpeedEngine._split_atempo_stages(0.5)
-        assert len(stages) == 1
-        assert stages[0] == 0.5
-
-    def test_split_atempo_boundary_max(self):
-        stages = SpeedEngine._split_atempo_stages(2.0)
-        assert len(stages) == 1
-        assert stages[0] == 2.0
-
-    def test_split_atempo_product_equals_speed(self):
-        """所有级联的乘积应该等于原速度."""
-        test_cases = [0.25, 0.3, 0.5, 0.75, 1.0, 1.5, 2.0, 3.0, 4.0]
-        for speed in test_cases:
-            stages = SpeedEngine._split_atempo_stages(speed)
-            product = 1.0
-            for s in stages:
-                product *= s
-            assert abs(product - speed) < 1e-6, f"speed={speed}, stages={stages}, product={product}"
-
-    def test_split_atempo_all_in_range(self):
-        """所有级都应该在 0.5~2.0 范围内."""
-        test_cases = [0.25, 0.3, 0.5, 0.75, 1.0, 1.5, 2.0, 3.0, 4.0]
-        for speed in test_cases:
-            stages = SpeedEngine._split_atempo_stages(speed)
-            for s in stages:
-                assert 0.5 <= s <= 2.0, f"speed={speed}, stage={s} out of range"
-
-
-# ─── SpeedEngine 时长计算 ────────────────────────────────────
-
-
-class TestSpeedEngineDuration:
-    def setup_method(self):
-        self.engine = SpeedEngine()
-
-    def test_original_speed_same_duration(self):
-        config = SpeedConfig(speed=1.0)
-        assert self.engine.adjust_duration(10.0, config) == 10.0
-
-    def test_double_speed_half_duration(self):
-        config = SpeedConfig(speed=2.0)
-        assert self.engine.adjust_duration(10.0, config) == 5.0
-
-    def test_half_speed_double_duration(self):
-        config = SpeedConfig(speed=0.5)
-        assert self.engine.adjust_duration(10.0, config) == 20.0
-
-    def test_quad_speed_quarter_duration(self):
-        config = SpeedConfig(speed=4.0)
-        assert self.engine.adjust_duration(10.0, config) == 2.5
-
-    def test_zero_duration(self):
-        config = SpeedConfig(speed=2.0)
-        assert self.engine.adjust_duration(0.0, config) == 0.0
-
-    def test_negative_duration(self):
-        config = SpeedConfig(speed=2.0)
-        assert self.engine.adjust_duration(-1.0, config) == -1.0
-
-
-# ─── SpeedEngine 便捷方法 ────────────────────────────────────
-
-
-class TestSpeedEngineHelper:
-    def setup_method(self):
-        self.engine = SpeedEngine()
-
-    def test_build_clip_speed_filter_original(self):
-        v_f, a_f, cfg = self.engine.build_clip_speed_filter(1.0)
-        assert v_f == ""
-        assert a_f == ""
-        assert cfg.speed == 1.0
-
-    def test_build_clip_speed_filter_2x(self):
-        v_f, a_f, cfg = self.engine.build_clip_speed_filter(2.0)
-        assert "setpts=PTS/2.0" in v_f
-        assert "atempo=2.0" in a_f
-        assert cfg.speed == 2.0
-
-    def test_build_clip_speed_clamped(self):
-        _, _, cfg = self.engine.build_clip_speed_filter(100.0)
-        assert cfg.speed == MAX_SPEED
-
-    def test_resolve_clip_speed_default(self):
-        assert SpeedEngine.resolve_clip_speed({}) == 1.0
-        assert SpeedEngine.resolve_clip_speed(None) == 1.0
-
-    def test_resolve_clip_speed_zero_uses_global(self):
-        assert SpeedEngine.resolve_clip_speed({"playback_speed": 0}, 1.5) == 1.5
-
-    def test_resolve_clip_speed_custom(self):
-        assert SpeedEngine.resolve_clip_speed({"playback_speed": 2.0}) == 2.0
-
-    def test_resolve_clip_speed_invalid_type(self):
-        assert SpeedEngine.resolve_clip_speed({"playback_speed": "fast"}) == 1.0

tests/unit/test_transition_engine.py

@@ -1,484 +0,0 @@
-"""转场特效引擎单测 — 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"