test(wave105): extract render_layer_utils domain module + 65 unit tests

从 unified_render_service.py (1985→1947行) 抽出渲染图层纯逻辑工具: - resolve_layer_role / get_layer_z_index: 图层角色解析 - clip_effective_duration / clip_playback_speed / clip_adjusted_duration: 时长计算 - estimate_total_duration: 总时长估算 - can_pass_through: 直通路径判断 - LAYER_Z_INDEX / MAIN_LAYER_ROLES 等常量保留向后兼容: 模块级函数 + 类静态方法均委托到新模块验证: 88 passed (23原有 + 65新增)，全套render 373 passed
test(wave93): 43 unit tests for duplication domain models (#956 )
2026-07-26 22:19:12 +08:00 · 2026-07-26 18:17:04 +08:00 · 2026-07-26 18:15:59 +08:00 · 2026-07-26 18:15:48 +08:00 · 2026-07-26 18:15:38 +08:00 · 2026-07-26 18:15:22 +08:00
22 changed files with 5881 additions and 1258 deletions
@@ -26,6 +26,13 @@ from packages.domain.edit_plan import EditPlan
 from packages.domain.edit_plan_clip import EditPlanClip
 from packages.domain.edit_template import EditTemplate
 from packages.domain.editing_mode import EditingMode
+from packages.domain.plan_generator_utils import (
+    DEFAULT_CLIP_DURATION,
+    create_clips_from_configs,
+    distribute_assets,
+    generate_default_clips,
+    map_clip_types_for_mode,
+)
 from packages.domain.template_clip_config import ClipType, TemplateClipConfig

 logger = logging.getLogger(__name__)
@@ -163,83 +170,18 @@ class PlanGeneratorService:
        plan_id: str,
        clip_configs: List[TemplateClipConfig],
    ) -> List[EditPlanClip]:
-        """从 TemplateClipConfig 列表创建 EditPlanClip 列表（未持久化）"""
-        clips: List[EditPlanClip] = []
-        # 按 order 排序
-        sorted_configs = sorted(clip_configs, key=lambda c: c.order)
+        """从 TemplateClipConfig 列表创建 EditPlanClip 列表（未持久化）.

-        for cfg in sorted_configs:
-            # 计算时长：取 min_duration 和 max_duration 的中间值
-            if cfg.min_duration > 0 and cfg.max_duration > 0:
-                duration = (cfg.min_duration + cfg.max_duration) / 2
-            elif cfg.min_duration > 0:
-                duration = cfg.min_duration
-            elif cfg.max_duration > 0:
-                duration = cfg.max_duration
-            else:
-                duration = _DEFAULT_CLIP_DURATION
-
-            # clip_type 可能是枚举或字符串
-            clip_type = cfg.clip_type.value if hasattr(cfg.clip_type, "value") else cfg.clip_type
-
-            # transition_effect 可能是枚举或字符串
-            transition = (
-                cfg.transition_effect.value if hasattr(cfg.transition_effect, "value") else cfg.transition_effect
-            )
-
-            # 从 clip config 中解析 playback_speed（兼容 speed_ratio 字段名）
-            clip_cfg = cfg.config or {}
-            playback_speed = clip_cfg.get("playback_speed", clip_cfg.get("speed_ratio", 1.0)) or 1.0
-
-            clip = EditPlanClip.create(
-                plan_id=plan_id,
-                clip_type=clip_type,
-                order=cfg.order,
-                template_clip_config_id=cfg.id,
-                text_content=getattr(cfg, "text_template", "") or "",
-                duration=duration,
-                transition_effect=transition or "cut",
-                playback_speed=playback_speed,
-                config=clip_cfg,
-            )
-            clips.append(clip)
-
-        return clips
+        委托给 plan_generator_utils.create_clips_from_configs 纯函数。
+        """
+        return create_clips_from_configs(plan_id, clip_configs)

    def _map_clip_types_for_mode(self, clips: List[EditPlanClip], editing_mode: str) -> None:
-        """将模板 clip_config 生成的 MAIN 类型片段，按 editing_mode 映射为对应角色类型。
+        """将 MAIN 类型片段按 editing_mode 映射为对应角色类型.

-        模板的 clip_config 使用 ClipType 枚举（main/intro/outro 等），
-        但 PIP / VOICE_PIP 模式的素材分配和渲染分层依赖特定的 clip_type 命名
-        （overlay / background / corner_voice / b_roll）。
-
-        映射规则（仅修改 MAIN 类型片段，非 MAIN 片段保持原类型）：
-        - PIP: 第1个 MAIN → main（背景），其余 MAIN → overlay（画中画）
-        - VOICE_PIP: 第1个 → background，第2个 → corner_voice，第3+个 → b_roll
-        - ONE_TAKE / VOICE_OVER: 保持 main 不变
+        委托给 plan_generator_utils.map_clip_types_for_mode 纯函数。
        """
-        from packages.domain.template_clip_config import ClipType
-
-        main_clips = [c for c in clips if c.clip_type == ClipType.MAIN.value]
-        if not main_clips:
-            return
-
-        if editing_mode == EditingMode.PIP.value:
-            # 第1个 main 保持（背景层），其余改为 overlay（画中画层）
-            for i, clip in enumerate(main_clips):
-                if i > 0:
-                    clip.clip_type = "overlay"
-
-        elif editing_mode == EditingMode.VOICE_PIP.value:
-            for i, clip in enumerate(main_clips):
-                if i == 0:
-                    clip.clip_type = "background"
-                elif i == 1:
-                    clip.clip_type = "corner_voice"
-                else:
-                    clip.clip_type = "b_roll"
-
-        # ONE_TAKE / VOICE_OVER: 保持 main 不变，无需处理
+        map_clip_types_for_mode(clips, editing_mode)

    def _generate_default_clips(
        self,
@@ -247,101 +189,11 @@ class PlanGeneratorService:
        editing_mode: str,
        asset_count: int,
    ) -> List[EditPlanClip]:
-        """无 clip_configs 时，根据 editing_mode 生成默认 clip 结构
+        """无 clip_configs 时，根据 editing_mode 生成默认 clip 结构.

-        - ONE_TAKE: N 个 main clips（N = asset_count，至少1个）
-        - PIP: 1 个 main + (N-1) 个 overlay（N = asset_count）
-        - VOICE_OVER: N 个 main clips + 标记需要配音
-        - VOICE_PIP: 1 个 background + 1 个 corner_voice + (N-2) 个 b_roll
+        委托给 plan_generator_utils.generate_default_clips 纯函数。
        """
-        n = max(asset_count, 1)
-        clips: List[EditPlanClip] = []
-        order = 0
-
-        if editing_mode == EditingMode.PIP.value:
-            # 1 个 main（全屏背景）
-            clips.append(
-                EditPlanClip.create(
-                    plan_id=plan_id,
-                    clip_type=ClipType.MAIN.value,
-                    order=order,
-                    duration=_DEFAULT_CLIP_DURATION,
-                )
-            )
-            order += 1
-            # 剩余为 overlay
-            for _ in range(1, n):
-                clips.append(
-                    EditPlanClip.create(
-                        plan_id=plan_id,
-                        clip_type="overlay",
-                        order=order,
-                        duration=_DEFAULT_CLIP_DURATION,
-                    )
-                )
-                order += 1
-
-        elif editing_mode == EditingMode.VOICE_OVER.value:
-            # N 个 main clips（B-roll）
-            for _ in range(n):
-                clips.append(
-                    EditPlanClip.create(
-                        plan_id=plan_id,
-                        clip_type=ClipType.MAIN.value,
-                        order=order,
-                        duration=_DEFAULT_CLIP_DURATION,
-                        config={"role": "b_roll"},
-                    )
-                )
-                order += 1
-
-        elif editing_mode == EditingMode.VOICE_PIP.value:
-            # 1 个 background
-            clips.append(
-                EditPlanClip.create(
-                    plan_id=plan_id,
-                    clip_type="background",
-                    order=order,
-                    duration=_DEFAULT_CLIP_DURATION,
-                )
-            )
-            order += 1
-            # 1 个 corner_voice
-            clips.append(
-                EditPlanClip.create(
-                    plan_id=plan_id,
-                    clip_type="corner_voice",
-                    order=order,
-                    duration=_DEFAULT_CLIP_DURATION,
-                )
-            )
-            order += 1
-            # 剩余为 b_roll
-            for _ in range(2, n):
-                clips.append(
-                    EditPlanClip.create(
-                        plan_id=plan_id,
-                        clip_type="b_roll",
-                        order=order,
-                        duration=_DEFAULT_CLIP_DURATION,
-                    )
-                )
-                order += 1
-
-        else:
-            # ONE_TAKE: N 个 main clips
-            for _ in range(n):
-                clips.append(
-                    EditPlanClip.create(
-                        plan_id=plan_id,
-                        clip_type=ClipType.MAIN.value,
-                        order=order,
-                        duration=_DEFAULT_CLIP_DURATION,
-                    )
-                )
-                order += 1
-
-        return clips
+        return generate_default_clips(plan_id, editing_mode, asset_count)

    def _distribute_assets(
        self,
@@ -349,89 +201,9 @@ class PlanGeneratorService:
        asset_ids: List[str],
        editing_mode: str,
    ) -> None:
-        """按 editing_mode 将素材分配到 clips（就地修改，未持久化）
+        """按 editing_mode 将素材分配到 clips（就地修改，未持久化）.

-        分配策略:
-        - ONE_TAKE: 素材按顺序依次分配给 main 类型 clips
-        - PIP: 第1个素材→main(全屏背景)，其余→交替分配给 overlay clips
-        - VOICE_OVER: 素材→main clips (B-roll)
-        - VOICE_PIP: 第1个→background, 第2个→corner_voice, 其余→b_roll
+        委托给 plan_generator_utils.distribute_assets 纯函数。
        """
-        if not asset_ids or not clips:
-            return
+        distribute_assets(clips, asset_ids, editing_mode)

-        if editing_mode == EditingMode.ONE_TAKE.value:
-            self._distribute_one_take(clips, asset_ids)
-        elif editing_mode == EditingMode.PIP.value:
-            self._distribute_pip(clips, asset_ids)
-        elif editing_mode == EditingMode.VOICE_OVER.value:
-            self._distribute_voice_over(clips, asset_ids)
-        elif editing_mode == EditingMode.VOICE_PIP.value:
-            self._distribute_voice_pip(clips, asset_ids)
-        else:
-            # 未知模式，退化为 one_take
-            self._distribute_one_take(clips, asset_ids)
-
-    def _distribute_one_take(
-        self,
-        clips: List[EditPlanClip],
-        asset_ids: List[str],
-    ) -> None:
-        """ONE_TAKE: 素材按顺序依次分配给 main 类型 clips"""
-        main_clips = [c for c in clips if c.clip_type == ClipType.MAIN.value]
-        for i, clip in enumerate(main_clips):
-            if i < len(asset_ids):
-                clip.assign_asset(asset_ids[i])
-
-    def _distribute_pip(
-        self,
-        clips: List[EditPlanClip],
-        asset_ids: List[str],
-    ) -> None:
-        """PIP: 第1个素材→main(全屏背景)，其余→overlay clips"""
-        # 第1个素材 → main clip
-        main_clips = [c for c in clips if c.clip_type == ClipType.MAIN.value]
-        if main_clips and asset_ids:
-            main_clips[0].assign_asset(asset_ids[0])
-
-        # 其余素材 → overlay clips
-        overlay_clips = [c for c in clips if c.clip_type == "overlay"]
-        remaining = asset_ids[1:]
-        for i, clip in enumerate(overlay_clips):
-            if i < len(remaining):
-                clip.assign_asset(remaining[i])
-
-    def _distribute_voice_over(
-        self,
-        clips: List[EditPlanClip],
-        asset_ids: List[str],
-    ) -> None:
-        """VOICE_OVER: 素材→main clips (B-roll)"""
-        main_clips = [c for c in clips if c.clip_type == ClipType.MAIN.value]
-        for i, clip in enumerate(main_clips):
-            if i < len(asset_ids):
-                clip.assign_asset(asset_ids[i])
-
-    def _distribute_voice_pip(
-        self,
-        clips: List[EditPlanClip],
-        asset_ids: List[str],
-    ) -> None:
-        """VOICE_PIP: 第1个→background, 第2个→corner_voice, 其余→b_roll"""
-        bg_clips = [c for c in clips if c.clip_type == "background"]
-        corner_clips = [c for c in clips if c.clip_type == "corner_voice"]
-        broll_clips = [c for c in clips if c.clip_type == "b_roll"]
-
-        # 第1个素材 → background
-        if bg_clips and len(asset_ids) > 0:
-            bg_clips[0].assign_asset(asset_ids[0])
-
-        # 第2个素材 → corner_voice
-        if corner_clips and len(asset_ids) > 1:
-            corner_clips[0].assign_asset(asset_ids[1])
-
-        # 其余素材 → b_roll
-        remaining = asset_ids[2:]
-        for i, clip in enumerate(broll_clips):
-            if i < len(remaining):
-                clip.assign_asset(remaining[i])
@@ -13,57 +13,31 @@
  - 最低质量分门槛：自动过滤低质量素材
  - 时长多样性：保证选出的素材时长分布均匀（短/中/长各占一定比例）
  - 兼容全部模式：素材库模式和项目模式都可用
+
+纯逻辑部分已抽离到 packages.domain.asset_scoring。
 """

 from __future__ import annotations

 import logging
-from dataclasses import dataclass
+
+from packages.domain.asset_scoring import (
+    MEDIUM_BUCKET_MAX as _MEDIUM_BUCKET_MAX,
+    MIN_QUALITY_SCORE as _MIN_QUALITY_SCORE,
+    OPTIMAL_DURATION_MAX as _OPTIMAL_DURATION_MAX,
+    OPTIMAL_DURATION_MIN as _OPTIMAL_DURATION_MIN,
+    SHORT_BUCKET_MAX as _SHORT_BUCKET_MAX,
+    TARGET_HEIGHT as _TARGET_HEIGHT,
+    TARGET_WIDTH as _TARGET_WIDTH,
+    AssetScoreDetail,
+    SmartSelectResult,
+    diverse_selection,
+    filter_candidates,
+    score_asset_detail,
+)

 logger = logging.getLogger(__name__)

-# ── 评分权重 ──────────────────────────────────────────────────────────────────
-_WEIGHT_QUALITY = 0.5
-_WEIGHT_RESOLUTION = 0.2
-_WEIGHT_DURATION = 0.2
-_WEIGHT_BITRATE = 0.1
-
-# ── 评分参数 ──────────────────────────────────────────────────────────────────
-_TARGET_WIDTH = 1920  # 目标分辨率宽度基准
-_TARGET_HEIGHT = 1080  # 目标分辨率高度基准
-_MIN_QUALITY_SCORE = 30.0  # 最低质量分门槛（低于此值的素材直接排除）
-_OPTIMAL_DURATION_MIN = 3.0  # 最佳时长区间（秒）
-_OPTIMAL_DURATION_MAX = 30.0
-
-# ── 多样性分桶 ───────────────────────────────────────────────────────────────
-_SHORT_BUCKET_MAX = 5.0  # 短素材：< 5s
-_MEDIUM_BUCKET_MAX = 15.0  # 中素材：5-15s
-# 长素材：> 15s
-
-
-@dataclass
-class SmartSelectResult:
-    """智能选择结果."""
-
-    selected_ids: list[str]
-    total_candidates: int
-    filtered_out: int  # 被质量门槛过滤的数量
-    avg_score: float
-    details: list[AssetScoreDetail]
-
-
-@dataclass
-class AssetScoreDetail:
-    """单个素材的评分详情."""
-
-    asset_id: str
-    total_score: float
-    quality_score: float
-    resolution_score: float
-    duration_score: float
-    bitrate_score: float
-    duration: float | None
-

 class SmartAssetSelector:
    """智能素材选择器.
@@ -74,9 +48,9 @@ class SmartAssetSelector:

    def __init__(
        self,
-        min_quality_score: float = _MIN_QUALITY_SCORE,
-        target_width: int = _TARGET_WIDTH,
-        target_height: int = _TARGET_HEIGHT,
+        min_quality_score: float = 30.0,
+        target_width: int = 1920,
+        target_height: int = 1080,
    ):
        self.min_quality_score = min_quality_score
        self.target_width = target_width
@@ -102,21 +76,7 @@ class SmartAssetSelector:
            SmartSelectResult 选择结果
        """
        # 1. 过滤：只保留 ready 状态的视频素材 + 最低质量分门槛
-        candidates = []
-        filtered_out = 0
-        for asset in assets:
-            status = getattr(asset, "status", None)
-            status_val = status.value if hasattr(status, "value") else str(status)
-            if status_val != "ready":
-                continue
-            mime_type = getattr(asset, "mime_type", "") or ""
-            if not mime_type.startswith("video"):
-                continue
-            quality = getattr(asset, "quality_score", None)
-            if quality is not None and quality < self.min_quality_score:
-                filtered_out += 1
-                continue
-            candidates.append(asset)
+        candidates, filtered_out = filter_candidates(assets, self.min_quality_score)

        if not candidates:
            return SmartSelectResult(
@@ -130,7 +90,16 @@ class SmartAssetSelector:
        # 2. 对每个候选素材评分
        scored: list[AssetScoreDetail] = []
        for asset in candidates:
-            detail = self._score_asset(asset)
+            detail = score_asset_detail(
+                asset_id=asset.id,
+                quality=getattr(asset, "quality_score", None),
+                width=getattr(asset, "width", None),
+                height=getattr(asset, "height", None),
+                duration=getattr(asset, "duration", None),
+                file_size=getattr(asset, "file_size", 0) or 0,
+                target_width=self.target_width,
+                target_height=self.target_height,
+            )
            scored.append(detail)

        # 3. 按总分降序排列
@@ -138,7 +107,7 @@ class SmartAssetSelector:

        # 4. 多样性选择（如果需要且数量有限制）
        if ensure_diversity and count > 0 and len(scored) > count:
-            selected = self._diverse_selection(scored, count)
+            selected = diverse_selection(scored, count)
        else:
            # 无数量限制或不要求多样性，直接按排名取
            selected = scored if count <= 0 else scored[:count]
@@ -162,165 +131,39 @@ class SmartAssetSelector:
        )
        return result

-    # ── 内部方法 ──────────────────────────────────────────────────────────────
+    # ── 向后兼容：私有方法别名（委托给 asset_scoring 纯函数） ────────────────

    def _score_asset(self, asset) -> AssetScoreDetail:
-        """对单个素材进行多维度评分."""
-        # 质量分
-        quality = getattr(asset, "quality_score", None)
-        quality_score = (quality / 100.0) if quality is not None else 0.5
-
-        # 分辨率评分：越接近目标分辨率得分越高
-        width = getattr(asset, "width", None)
-        height = getattr(asset, "height", None)
-        resolution_score = self._score_resolution(width, height)
-
-        # 时长评分：在最佳区间内得分高，过短过长扣分
-        duration = getattr(asset, "duration", None)
-        duration_score = self._score_duration(duration)
-
-        # 码率评分：用 file_size/duration 估算，适中得分高
-        file_size = getattr(asset, "file_size", 0) or 0
-        bitrate_score = self._score_bitrate(file_size, duration)
-
-        # 加权总分
-        total = (
-            _WEIGHT_QUALITY * quality_score
-            + _WEIGHT_RESOLUTION * resolution_score
-            + _WEIGHT_DURATION * duration_score
-            + _WEIGHT_BITRATE * bitrate_score
-        )
-
-        return AssetScoreDetail(
+        """对单个素材进行多维度评分（向后兼容）."""
+        return score_asset_detail(
            asset_id=asset.id,
-            total_score=round(total, 4),
-            quality_score=round(quality_score, 4),
-            resolution_score=round(resolution_score, 4),
-            duration_score=round(duration_score, 4),
-            bitrate_score=round(bitrate_score, 4),
-            duration=duration,
+            quality=getattr(asset, "quality_score", None),
+            width=getattr(asset, "width", None),
+            height=getattr(asset, "height", None),
+            duration=getattr(asset, "duration", None),
+            file_size=getattr(asset, "file_size", 0) or 0,
+            target_width=self.target_width,
+            target_height=self.target_height,
        )

    def _score_resolution(self, width: int | None, height: int | None) -> float:
-        """分辨率评分：越接近目标分辨率得分越高，低于480p扣分严重."""
-        if width is None or height is None or width <= 0 or height <= 0:
-            return 0.5  # 未知分辨率给中评分
+        """分辨率评分（向后兼容）."""
+        from packages.domain.asset_scoring import score_resolution

-        target_pixels = self.target_width * self.target_height
-        actual_pixels = width * height
-
-        # 计算像素数比例
-        ratio = actual_pixels / target_pixels
-
-        if ratio >= 1.0:
-            # 高于或等于目标分辨率：满分，略高不扣分（4K也给满分）
-            return 1.0
-        else:
-            # 低于目标分辨率：线性衰减，但最低不低于 0.1
-            # 例如：720p (921600) / 1080p (2073600) = 0.44 → 得分 0.6
-            score = 0.3 + 0.7 * ratio
-            return max(0.1, min(1.0, score))
+        return score_resolution(width, height, self.target_width, self.target_height)

    def _score_duration(self, duration: float | None) -> float:
-        """时长评分：3-30秒最佳，过短或过长都扣分."""
-        if duration is None or duration <= 0:
-            return 0.5  # 未知时长给中评分
+        """时长评分（向后兼容）."""
+        from packages.domain.asset_scoring import score_duration

-        if _OPTIMAL_DURATION_MIN <= duration <= _OPTIMAL_DURATION_MAX:
-            # 最佳区间：满分
-            return 1.0
-
-        if duration < _OPTIMAL_DURATION_MIN:
-            # 太短：线性衰减，1秒以下给 0.3
-            ratio = duration / _OPTIMAL_DURATION_MIN
-            return 0.3 + 0.7 * ratio
-
-        # 太长：每超过最佳区间上限10秒扣 0.1 分，最低 0.2
-        excess = duration - _OPTIMAL_DURATION_MAX
-        penalty = min(0.8, excess / 10.0 * 0.1)
-        return max(0.2, 1.0 - penalty)
+        return score_duration(duration)

    def _score_bitrate(self, file_size: int, duration: float | None) -> float:
-        """码率评分：根据文件大小和时长估算码率，适中得分高."""
-        if not file_size or not duration or duration <= 0:
-            return 0.5  # 未知给中评分
+        """码率评分（向后兼容）."""
+        from packages.domain.asset_scoring import score_bitrate

-        # 估算码率（bps）
-        bitrate = (file_size * 8) / duration
-
-        # 最佳码率范围：2-8 Mbps
-        optimal_low = 2_000_000  # 2 Mbps
-        optimal_high = 8_000_000  # 8 Mbps
-
-        if optimal_low <= bitrate <= optimal_high:
-            return 1.0
-
-        if bitrate < optimal_low:
-            # 码率太低：线性衰减
-            ratio = bitrate / optimal_low
-            return 0.3 + 0.7 * ratio
-
-        # 码率太高（文件太大）：适度扣分
-        excess = bitrate / optimal_high - 1.0
-        penalty = min(0.5, excess * 0.2)
-        return max(0.5, 1.0 - penalty)
+        return score_bitrate(file_size, duration)

    def _diverse_selection(self, scored: list[AssetScoreDetail], count: int) -> list[AssetScoreDetail]:
-        """多样性选择：按时长分桶，保证每个桶都有素材.
-
-        策略：
-        1. 按时长分为三桶：短(<5s)、中(5-15s)、长(>15s)
-        2. 每个桶配额 = max(1, count / 3)
-        3. 先从每桶按配额取最高分的
-        4. 剩余名额从全局最高分中取（不重复）
-        """
-        # 分桶
-        short_bucket = [d for d in scored if d.duration is not None and d.duration < _SHORT_BUCKET_MAX]
-        medium_bucket = [
-            d for d in scored if d.duration is not None and _SHORT_BUCKET_MAX <= d.duration < _MEDIUM_BUCKET_MAX
-        ]
-        long_bucket = [d for d in scored if d.duration is not None and d.duration >= _MEDIUM_BUCKET_MAX]
-        unknown_bucket = [d for d in scored if d.duration is None]
-
-        buckets = [short_bucket, medium_bucket, long_bucket]
-        bucket_names = ["short", "medium", "long"]
-
-        # 每个桶基础配额（至少1个，如果桶非空且需要的话）
-        base_quota = max(1, count // 3)
-
-        selected: list[AssetScoreDetail] = []
-        selected_ids: set[str] = set()
-
-        # 先按配额从每个桶取
-        for bucket, _name in zip(buckets, bucket_names, strict=False):
-            quota = min(base_quota, len(bucket))
-            if quota <= 0:
-                continue
-            # 桶内已经按分数排好序了，直接取前 quota 个
-            for item in bucket[:quota]:
-                if item.asset_id not in selected_ids:
-                    selected.append(item)
-                    selected_ids.add(item.asset_id)
-                    if len(selected) >= count:
-                        return selected
-
-        # 剩余名额：从全局（未被选中的）中按分数高低取
-        remaining_needed = count - len(selected)
-        if remaining_needed > 0:
-            for item in scored:
-                if item.asset_id not in selected_ids:
-                    selected.append(item)
-                    selected_ids.add(item.asset_id)
-                    if len(selected) >= count:
-                        break
-
-        # 如果还不够（不应该发生），加上未知时长的
-        if len(selected) < count and unknown_bucket:
-            for item in unknown_bucket:
-                if item.asset_id not in selected_ids:
-                    selected.append(item)
-                    selected_ids.add(item.asset_id)
-                    if len(selected) >= count:
-                        break
-
-        return selected[:count]
+        """多样性选择（向后兼容）."""
+        return diverse_selection(scored, count)
@@ -74,7 +74,7 @@ const TtsPanel: React.FC<TtsPanelProps> = ({ open, onClose, config, onChange })
              className="tts-text-input"
              placeholder="请输入需要合成的文本内容..."
              value={config.text}
-              onChange={handleTextChange}
+              onChange={(e) => handleTextChange(e.target.value)}
              maxLength={5000}
              rows={5}
            />
@@ -2,7 +2,7 @@
 * 混剪单图层配置区
 */
 import React from "react"
-import type { PipLayer, PipAnimType, PipSlideDirection } from "@/pages/editing-planner/types"
+import type { PipLayer, PipAnimType, PipSlideDirection, PipGridPosition } from "@/pages/editing-planner/types"
 import {
  GRID_POSITIONS,
  ANIM_OPTIONS,
@@ -15,7 +15,7 @@ interface LayerConfigProps {
  layers: PipLayer[]
  totalDuration: number
  onUpdate: (id: string, partial: Partial<PipLayer>) => void
-  onGridClick: (pos: any) => void
+  onGridClick: (pos: PipGridPosition) => void
  onWidthChange: (val: number) => void
  onHeightChange: (val: number) => void
 }
@@ -46,7 +46,7 @@ const TtsSlider: React.FC<TtsSliderProps> = ({
        step={step}
        value={value}
        onChange={(v) => onChange(v as number)}
-        tooltip={tooltipFormatter ? { formatter: (v) => tooltipFormatter(v as number) } : false}
+        tooltip={tooltipFormatter ? { formatter: (v) => tooltipFormatter(v as number) } : undefined}
      />
      {marks && (
        <div className="tts-slider-marks">
@@ -1,7 +1,7 @@
 /**
 * ClipPropertiesPanel 相关类型定义
 */
-import type { ClipData, ClipType } from "@/pages/editing-planner/types"
+import type { ClipData } from "@/pages/editing-planner/types"
 import type { TemplateMode } from "@/api/editing-planner"
 import type { AssetItem } from "@/api/assets"

@@ -52,6 +52,15 @@ from video_processing.trim_engine import TrimConfig, TrimEngine, extract_trim_fr
 from video_processing.tts_engine import TtsEngine
 from video_processing.watermark_engine import WatermarkConfig, WatermarkEngine

+from packages.domain.render_layer_utils import (
+    LAYER_Z_INDEX as _IMPORTED_LAYER_Z_INDEX,
+    can_pass_through as _can_pass_through_pure,
+    clip_adjusted_duration as _clip_adjusted_duration_pure,
+    clip_effective_duration as _clip_effective_duration_pure,
+    clip_playback_speed as _clip_playback_speed_pure,
+    estimate_total_duration as _estimate_total_duration_pure,
+    resolve_layer_role as _resolve_layer_role_pure,
+)
 from packages.domain.tts_config import TtsConfig

 logger = logging.getLogger(__name__)
@@ -107,47 +116,16 @@ class RenderResult:


 def _resolve_layer_role(clip_type: str, config: dict[str, Any]) -> str:
-    """根据 clip_type 和 config.role 确定图层角色。
+    """根据 clip_type 和 config.role 确定图层角色（向后兼容别名）。

-    映射规则：
-      intro / outro         → "main"（按 order 排在首/尾）
-      overlay               → "overlay"（画中画叠加，z=1）
-      corner_voice          → "corner_voice"（右上角小窗，z=1）
-      background            → "background"（全屏底图，z=0）
-      b_roll                → "broll"（z=0）
-      main + config.role=b_roll → "broll"
-      main (default)        → "main"
+    实际实现移至 packages.domain.render_layer_utils.resolve_layer_role。
    """
-    role = config.get("role", "")
-
-    if clip_type in ("intro", "outro"):
-        return "main"
-    if clip_type == "overlay":
-        return "overlay"
-    if clip_type == "corner_voice":
-        return "corner_voice"
-    if clip_type == "background":
-        return "background"
-    if clip_type == "b_roll":
-        return "broll"
-    # main type
-    if role == "b_roll":
-        return "broll"
-    if role == "audio":
-        return "audio"
-    return "main"
+    return _resolve_layer_role_pure(clip_type, config)


 # ── 图层默认 z_index ─────────────────────────────────────────────────────────

-_LAYER_Z_INDEX: dict[str, int] = {
-    "background": -1,
-    "broll": 0,
-    "main": 0,
-    "overlay": 1,
-    "corner_voice": 1,
-    "audio": 2,
-}
+_LAYER_Z_INDEX: dict[str, int] = _IMPORTED_LAYER_Z_INDEX

 # 图层默认 PiP 位置（相对输出画布的偏移）
 _PIP_SCALE = 0.25  # PiP 占主画面的比例
@@ -489,29 +467,9 @@ class UnifiedRenderService:
    def _estimate_total_duration(self, layers: list[RenderLayer]) -> float:
        """估算视频总时长（用于字幕等需要）。

-        取主图层（main/broll/background）的总时长，转场重叠按 transition_duration 估算。
+        实际实现移至 packages.domain.render_layer_utils.estimate_total_duration。
        """
-        # 找主图层（第一个有视频内容的图层）
-        main_layer = None
-        for role in ("main", "broll", "background"):
-            for layer in layers:
-                if layer.role == role:
-                    main_layer = layer
-                    break
-            if main_layer:
-                break
-
-        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)
-
-        # 减去转场重叠时间（粗略估算）
-        n_clips = len(main_layer.clips)
-        if n_clips > 1:
-            total -= (n_clips - 1) * self.transition_duration
-
-        return max(0.1, total)
+        return _estimate_total_duration_pure(layers, self.transition_duration)

    def _maybe_generate_ass(self, video_duration: float) -> Path | None:
        """根据 plan.config 生成 ASS 字幕文件。
@@ -1869,10 +1827,11 @@ class UnifiedRenderService:

    @staticmethod
    def _clip_effective_duration(clip: ResolvedClip) -> float:
-        """计算 clip 的有效时长（原速 trim 后时长）."""
-        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
+        """计算 clip 的有效时长（原速 trim 后时长）。
+
+        实际实现移至 packages.domain.render_layer_utils.clip_effective_duration。
+        """
+        return _clip_effective_duration_pure(clip.duration, clip.actual_duration)

    # ── 画中画（PiP）相关方法 ──────────────────────────────────────────────────

@@ -1969,17 +1928,20 @@ class UnifiedRenderService:

    @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)
+        """获取 clip 的播放速度，无效值回退到 1.0。
+
+        实际实现移至 packages.domain.render_layer_utils.clip_playback_speed。
+        """
+        return _clip_playback_speed_pure(getattr(clip, "playback_speed", 1.0))

    @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
+        """计算调速后的 clip 实际时长（用于拼接计算）。
+
+        实际实现移至 packages.domain.render_layer_utils.clip_adjusted_duration。
+        """
+        return _clip_adjusted_duration_pure(
+            clip.duration,
+            clip.actual_duration,
+            getattr(clip, "playback_speed", 1.0),
+        )
@@ -19,74 +19,21 @@ from PIL import Image

 from packages.domain.classification import AssetClassification

+from .asset_quality_scoring import (
+    AudioAnalysis,
+    ClassificationResult,
+    ColorAnalysis,
+    MotionAnalysis,
+    QualityScore,
+    VideoInfo,
+    calculate_category_scores,
+    calculate_quality_score,
+    classify_from_analysis,
+)
+
 logger = logging.getLogger(__name__)


-@dataclass
-class VideoInfo:
-    """视频基本信息"""
-
-    width: int = 0
-    height: int = 0
-    fps: float = 0.0
-    duration: float = 0.0
-    bitrate: int = 0
-    codec: str = ""
-    has_audio: bool = False
-    file_size: int = 0
-
-
-@dataclass
-class ColorAnalysis:
-    """色彩分析结果"""
-
-    dominant_hue: float = 0.0  # 主色调 (0-360)
-    green_ratio: float = 0.0  # 绿色占比
-    warm_ratio: float = 0.0  # 暖色调占比
-    cool_ratio: float = 0.0  # 冷色调占比
-    avg_saturation: float = 0.0
-    avg_brightness: float = 0.0
-
-
-@dataclass
-class MotionAnalysis:
-    """运动分析结果"""
-
-    motion_score: float = 0.0  # 运动幅度 (0-1)
-    scene_changes: int = 0  # 场景切换次数
-
-
-@dataclass
-class AudioAnalysis:
-    """音频分析结果"""
-
-    has_audio: bool = False
-    speech_ratio: float = 0.0  # 人声比例
-    music_ratio: float = 0.0  # 音乐比例
-    ambient_ratio: float = 0.0  # 环境音比例
-
-
-@dataclass
-class ClassificationResult:
-    """分类结果"""
-
-    category: AssetClassification
-    confidence: float
-    scores: dict[str, float] = field(default_factory=dict)
-
-
-@dataclass
-class QualityScore:
-    """质量评分结果"""
-
-    total: float
-    resolution_score: float = 0.0
-    fps_score: float = 0.0
-    bitrate_score: float = 0.0
-    clarity_score: float = 0.0
-    stability_score: float = 0.0
-
-
 class AssetAnalyzer:
    """
    轻量级视频素材分析器
@@ -449,316 +396,37 @@ class AssetAnalyzer:
        """
        综合分析得出分类结果

+        评分逻辑在 asset_quality_scoring.calculate_category_scores / classify_from_analysis
+        纯函数中，此处只负责采集分析数据后委托计算。
+
        Returns:
            ClassificationResult 对象
        """
-        # 提取分析数据
        frames = self.extract_frames()
        color = self.analyze_color_distribution(frames)
        motion = self.analyze_motion(frames)
        audio = self.analyze_audio()

-        # 计算各类别得分
-        scores = self._calculate_category_scores(color, motion, audio)
-
-        # 找最高分
-        if not scores:
-            return ClassificationResult(
-                category=AssetClassification.OTHER,
-                confidence=0.3,
-                scores={},
-            )
-
-        best_category = max(scores.items(), key=lambda x: x[1])
-        category = AssetClassification(best_category[0])
-        confidence = min(0.95, max(0.3, best_category[1]))
-
-        return ClassificationResult(
-            category=category,
-            confidence=confidence,
-            scores=scores,
-        )
-
-    def _calculate_category_scores(
-        self,
-        color: ColorAnalysis,
-        motion: MotionAnalysis,
-        audio: AudioAnalysis,
-    ) -> dict[str, float]:
-        """
-        计算各类别的置信度得分
-
-        Args:
-            color: 色彩分析结果
-            motion: 运动分析结果
-            audio: 音频分析结果
-
-        Returns:
-            各类别得分字典
-        """
-        scores = {}
-
-        # 1. 风景 (scenic) - 绿色、户外、自然
-        scenic_score = 0.0
-        if color.green_ratio > 0.3:
-            scenic_score += 0.4 * color.green_ratio
-        if color.avg_saturation > 0.3:
-            scenic_score += 0.2 * color.avg_saturation
-        if color.avg_brightness > 0.4:
-            scenic_score += 0.2
-        if motion.motion_score > 0.1 and motion.motion_score < 0.5:
-            scenic_score += 0.2  # 适度运动（如云朵、树叶）
-        if not audio.has_audio or audio.ambient_ratio > 0.5:
-            scenic_score += 0.2  # 自然环境音
-        scores[AssetClassification.SCENIC.value] = min(1.0, scenic_score)
-
-        # 2. 产品 (product) - 中等亮度、均匀色彩、低运动
-        product_score = 0.0
-        if 0.3 < color.avg_brightness < 0.7:
-            product_score += 0.3
-        if color.avg_saturation < 0.5:
-            product_score += 0.2
-        if motion.motion_score < 0.15:
-            product_score += 0.4  # 低运动 = 产品展示
-        if color.cool_ratio > 0.3:
-            product_score += 0.2  # 冷色调 = 科技感
-        scores[AssetClassification.PRODUCT.value] = min(1.0, product_score)
-
-        # 3. 人物 (person) - 中等运动、有时有人声
-        person_score = 0.0
-        if 0.1 < motion.motion_score < 0.4:
-            person_score += 0.3  # 适度运动
-        if audio.has_audio and audio.speech_ratio > 0.3:
-            person_score += 0.5  # 有人声
-        if color.avg_brightness > 0.3:
-            person_score += 0.2
-        scores[AssetClassification.PERSON.value] = min(1.0, person_score)
-
-        # 4. 动物 (animal) - 高运动、有时自然音
-        animal_score = 0.0
-        if motion.motion_score > 0.3:
-            animal_score += 0.4  # 高运动
-        if motion.scene_changes > 2:
-            animal_score += 0.2
-        if audio.has_audio and (audio.ambient_ratio > 0.3 or audio.speech_ratio > 0.2):
-            animal_score += 0.3
-        scores[AssetClassification.ANIMAL.value] = min(1.0, animal_score)
-
-        # 5. 美食 (food) - 暖色调、高饱和度
-        food_score = 0.0
-        if color.warm_ratio > 0.4:
-            food_score += 0.5
-        if color.avg_saturation > 0.5:
-            food_score += 0.3
-        if 0.4 < color.avg_brightness < 0.8:
-            food_score += 0.2
-        scores[AssetClassification.FOOD.value] = min(1.0, food_score)
-
-        # 6. 科技 (tech) - 冷色调、低饱和度、低运动
-        tech_score = 0.0
-        if color.cool_ratio > 0.4:
-            tech_score += 0.4
-        if color.avg_saturation < 0.4:
-            tech_score += 0.3
-        if motion.motion_score < 0.2:
-            tech_score += 0.3
-        scores[AssetClassification.TECH.value] = min(1.0, tech_score)
-
-        # 7. 运动 (sport) - 高运动
-        sport_score = 0.0
-        if motion.motion_score > 0.4:
-            sport_score += 0.6
-        if motion.scene_changes > 3:
-            sport_score += 0.2
-        if color.avg_brightness > 0.4:
-            sport_score += 0.2
-        scores[AssetClassification.SPORT.value] = min(1.0, sport_score)
-
-        # 8. 音乐 (music) - 有节奏性音乐
-        music_score = 0.0
-        if audio.has_audio and audio.music_ratio > 0.4:
-            music_score += 0.6
-        # 纯视觉判断：色彩丰富但非自然
-        if color.avg_saturation > 0.5 and color.green_ratio < 0.2:
-            music_score += 0.3
-        scores[AssetClassification.MUSIC.value] = min(1.0, music_score)
-
-        # 9. 其他 (other) - 默认最低分
-        scores[AssetClassification.OTHER.value] = 0.1
-
-        return scores
+        return classify_from_analysis(color, motion, audio)

    def calculate_quality_score(self) -> QualityScore:
        """
        计算视频质量综合评分 (0-100)

+        评分逻辑在 asset_quality_scoring.calculate_quality_score 纯函数中，
+        此处只负责采集数据后委托计算。
+
        评分维度:
        1. 分辨率得分 (25分)
        2. 帧率得分 (20分)
        3. 码率得分 (20分)
-        4. 清晰度得分 (20分) - Laplacian 方差
-        5. 稳定性得分 (15分) - 帧间位移方差
+        4. 清晰度得分 (20分)
+        5. 稳定性得分 (15分)
        """
        info = self.get_video_info()
        frames = self.extract_frames()

-        # 1. 分辨率得分
-        resolution_score = self._score_resolution(info.width, info.height)
-
-        # 2. 帧率得分
-        fps_score = self._score_framerate(info.fps)
-
-        # 3. 码率得分
-        bitrate_score = self._score_bitrate(info.bitrate)
-
-        # 4. 清晰度得分
-        clarity_score = self._score_clarity(frames)
-
-        # 5. 稳定性得分
-        stability_score = self._score_stability(frames)
-
-        total = resolution_score + fps_score + bitrate_score + clarity_score + stability_score
-
-        return QualityScore(
-            total=round(min(100, max(0, total)), 1),
-            resolution_score=resolution_score,
-            fps_score=fps_score,
-            bitrate_score=bitrate_score,
-            clarity_score=clarity_score,
-            stability_score=stability_score,
-        )
-
-    def _score_resolution(self, width: int, height: int) -> float:
-        """分辨率评分 (满分 25)"""
-        pixels = width * height
-
-        if pixels >= 3840 * 2160:  # 4K
-            return 25.0
-        elif pixels >= 2560 * 1440:  # 2K
-            return 22.0
-        elif pixels >= 1920 * 1080:  # 1080p
-            return 20.0
-        elif pixels >= 1280 * 720:  # 720p
-            return 15.0
-        elif pixels >= 854 * 480:  # 480p
-            return 8.0
-        else:
-            return 3.0
-
-    def _score_framerate(self, fps: float) -> float:
-        """帧率评分 (满分 20)"""
-        if fps >= 60:
-            return 20.0
-        elif fps >= 30:
-            return 15.0
-        elif fps >= 24:
-            return 10.0
-        elif fps >= 15:
-            return 7.0
-        else:
-            return 5.0
-
-    def _score_bitrate(self, bitrate: int) -> float:
-        """码率评分 (满分 20)"""
-        bitrate_mbps = bitrate / 1_000_000
-
-        if bitrate_mbps > 10:
-            return 20.0
-        elif bitrate_mbps >= 5:
-            return 15.0
-        elif bitrate_mbps >= 2:
-            return 10.0
-        elif bitrate_mbps >= 0.5:
-            return 5.0
-        else:
-            return 3.0
-
-    def _score_clarity(self, frames: list[np.ndarray]) -> float:
-        """
-        清晰度评分 (满分 20)
-
-        使用 Laplacian 方差评估画面清晰度
-        高方差 = 细节丰富 = 高分
-        """
-        if not frames:
-            return 10.0  # 默认中等分
-
-        try:
-            variances = []
-
-            for frame in frames[:5]:  # 只分析前 5 帧
-                if len(frame.shape) == 3:
-                    # 转灰度
-                    gray = np.dot(frame[..., :3], [0.299, 0.587, 0.114]).astype(np.uint8)
-                else:
-                    gray = frame
-
-                # Laplacian 算子
-                laplacian = np.array([[0, 1, 0], [1, -4, 1], [0, 1, 0]], dtype=np.float32)
-
-                # 手动计算卷积
-                from scipy import signal
-
-                laplacian_img = signal.convolve2d(gray.astype(float), laplacian, mode="same")
-                variance = np.var(laplacian_img)
-                variances.append(variance)
-
-            # 归一化方差到 0-20 分
-            avg_variance = np.mean(variances)
-            # 根据经验值调整
-            score = min(20.0, avg_variance / 100)
-            return float(score)
-
-        except ImportError:
-            # 如果没有 scipy，使用简化方法
-            return 10.0
-        except Exception:
-            return 10.0
-
-    def _score_stability(self, frames: list[np.ndarray]) -> float:
-        """
-        稳定性评分 (满分 15)
-
-        分析帧间位移方差
-        画面稳定 = 高分
-        剧烈抖动 = 低分
-        """
-        if len(frames) < 2:
-            return 10.0  # 默认中等分
-
-        try:
-            displacements = []
-
-            for i in range(len(frames) - 1):
-                # 缩小帧以加速处理
-                scale = 0.25
-                new_h = int(frames[i].shape[0] * scale)
-                new_w = int(frames[i].shape[1] * scale)
-                frame1_small = np.array(Image.fromarray(frames[i]).resize((new_w, new_h)))
-                new_h2 = int(frames[i + 1].shape[0] * scale)
-                new_w2 = int(frames[i + 1].shape[1] * scale)
-                frame2_small = np.array(Image.fromarray(frames[i + 1]).resize((new_w2, new_h2)))
-
-                # 简单位移检测：灰度差
-                gray1 = np.mean(frame1_small, axis=2) if len(frame1_small.shape) == 3 else frame1_small
-                gray2 = np.mean(frame2_small, axis=2) if len(frame2_small.shape) == 3 else frame2_small
-
-                diff = np.abs(gray2.astype(float) - gray1.astype(float))
-                displacement = np.mean(diff) / 255.0
-                displacements.append(displacement)
-
-            # 高位移方差 = 不稳定
-            if displacements:
-                displacement_variance = np.var(displacements)
-                # 归一化
-                instability = min(1.0, displacement_variance * 10)
-                score = 15.0 * (1.0 - instability)
-                return float(max(0.0, score))
-
-            return 10.0
-
-        except Exception:
-            return 10.0
+        return calculate_quality_score(info, frames)


 def classify_asset_real(video_path: str) -> tuple[str, float]:
@@ -0,0 +1,459 @@
+"""素材质量与分类评分 — 纯逻辑模块.
+
+从 asset_analyzer.py 提取的评分计算逻辑，纯函数，无副作用。
+输入分析结果对象，输出评分/分类结果。
+
+拆分目的：
+1. 大文件瘦身（asset_analyzer.py 799行 → 拆出 200+ 行纯逻辑）
+2. 评分逻辑可独立单测，不依赖 FFmpeg/视频文件
+3. 评分策略调整时不需要触碰分析主流程
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from typing import Optional
+
+import numpy as np
+
+from packages.domain.classification import AssetClassification
+
+# ── 数据类 ────────────────────────────────────────────────────────────
+
+
+@dataclass
+class VideoInfo:
+    """视频基本信息"""
+
+    width: int = 0
+    height: int = 0
+    fps: float = 0.0
+    duration: float = 0.0
+    bitrate: int = 0
+    codec: str = ""
+    has_audio: bool = False
+    file_size: int = 0
+
+
+@dataclass
+class ColorAnalysis:
+    """色彩分析结果"""
+
+    dominant_hue: float = 0.0  # 主色调 (0-360)
+    green_ratio: float = 0.0  # 绿色占比
+    warm_ratio: float = 0.0  # 暖色调占比
+    cool_ratio: float = 0.0  # 冷色调占比
+    avg_saturation: float = 0.0
+    avg_brightness: float = 0.0
+
+
+@dataclass
+class MotionAnalysis:
+    """运动分析结果"""
+
+    motion_score: float = 0.0  # 运动幅度 (0-1)
+    scene_changes: int = 0  # 场景切换次数
+
+
+@dataclass
+class AudioAnalysis:
+    """音频分析结果"""
+
+    has_audio: bool = False
+    speech_ratio: float = 0.0  # 人声比例
+    music_ratio: float = 0.0  # 音乐比例
+    ambient_ratio: float = 0.0  # 环境音比例
+
+
+@dataclass
+class ClassificationResult:
+    """分类结果"""
+
+    category: AssetClassification
+    confidence: float
+    scores: dict[str, float] = field(default_factory=dict)
+
+
+@dataclass
+class QualityScore:
+    """质量评分结果"""
+
+    total: float
+    resolution_score: float = 0.0
+    fps_score: float = 0.0
+    bitrate_score: float = 0.0
+    clarity_score: float = 0.0
+    stability_score: float = 0.0
+
+
+# ── 质量评分纯函数 ────────────────────────────────────────────────────
+
+
+def score_resolution(width: int, height: int) -> float:
+    """分辨率评分 (满分 25).
+
+    按像素总数阶梯评分：4K > 2K > 1080p > 720p > 480p > 其他.
+
+    Args:
+        width: 视频宽度（像素）
+        height: 视频高度（像素）
+
+    Returns:
+        float: 0-25 分
+    """
+    pixels = width * height
+
+    if pixels >= 3840 * 2160:  # 4K
+        return 25.0
+    elif pixels >= 2560 * 1440:  # 2K
+        return 22.0
+    elif pixels >= 1920 * 1080:  # 1080p
+        return 20.0
+    elif pixels >= 1280 * 720:  # 720p
+        return 15.0
+    elif pixels >= 854 * 480:  # 480p
+        return 8.0
+    else:
+        return 3.0
+
+
+def score_framerate(fps: float) -> float:
+    """帧率评分 (满分 20).
+
+    60fps 满分，阶梯递减.
+
+    Args:
+        fps: 帧率（帧/秒）
+
+    Returns:
+        float: 0-20 分
+    """
+    if fps >= 60:
+        return 20.0
+    elif fps >= 30:
+        return 15.0
+    elif fps >= 24:
+        return 10.0
+    elif fps >= 15:
+        return 7.0
+    else:
+        return 5.0
+
+
+def score_bitrate(bitrate: int) -> float:
+    """码率评分 (满分 20).
+
+    按 Mbps 阶梯评分.
+
+    Args:
+        bitrate: 码率（bps）
+
+    Returns:
+        float: 0-20 分
+    """
+    bitrate_mbps = bitrate / 1_000_000
+
+    if bitrate_mbps > 10:
+        return 20.0
+    elif bitrate_mbps >= 5:
+        return 15.0
+    elif bitrate_mbps >= 2:
+        return 10.0
+    elif bitrate_mbps >= 0.5:
+        return 5.0
+    else:
+        return 3.0
+
+
+def score_clarity(frames: list[np.ndarray]) -> float:
+    """清晰度评分 (满分 20).
+
+    使用 Laplacian 方差评估画面清晰度。
+    高方差 = 细节丰富 = 高分.
+
+    Args:
+        frames: 视频帧列表（numpy 数组，RGB 或灰度）
+
+    Returns:
+        float: 0-20 分
+    """
+    if not frames:
+        return 10.0  # 默认中等分
+
+    try:
+        variances = []
+
+        for frame in frames[:5]:  # 只分析前 5 帧
+            if len(frame.shape) == 3:
+                # 转灰度
+                gray = np.dot(frame[..., :3], [0.299, 0.587, 0.114]).astype(np.uint8)
+            else:
+                gray = frame
+
+            # Laplacian 算子
+            laplacian = np.array([[0, 1, 0], [1, -4, 1], [0, 1, 0]], dtype=np.float32)
+
+            # 手动计算卷积
+            from scipy import signal
+
+            laplacian_img = signal.convolve2d(gray.astype(float), laplacian, mode="same")
+            variance = np.var(laplacian_img)
+            variances.append(variance)
+
+        # 归一化方差到 0-20 分
+        avg_variance = np.mean(variances)
+        score = min(20.0, avg_variance / 100)
+        return float(score)
+
+    except ImportError:
+        # 如果没有 scipy，使用简化方法
+        return 10.0
+    except Exception:
+        return 10.0
+
+
+def score_stability(frames: list[np.ndarray]) -> float:
+    """稳定性评分 (满分 15).
+
+    分析帧间位移方差。
+    画面稳定 = 高分；剧烈抖动 = 低分.
+
+    Args:
+        frames: 视频帧列表（numpy 数组，RGB 或灰度）
+
+    Returns:
+        float: 0-15 分
+    """
+    if len(frames) < 2:
+        return 10.0  # 默认中等分
+
+    try:
+        from PIL import Image
+
+        displacements = []
+
+        for i in range(len(frames) - 1):
+            # 缩小帧以加速处理
+            scale = 0.25
+            new_h = int(frames[i].shape[0] * scale)
+            new_w = int(frames[i].shape[1] * scale)
+            frame1_small = np.array(Image.fromarray(frames[i]).resize((new_w, new_h)))
+            new_h2 = int(frames[i + 1].shape[0] * scale)
+            new_w2 = int(frames[i + 1].shape[1] * scale)
+            frame2_small = np.array(Image.fromarray(frames[i + 1]).resize((new_w2, new_h2)))
+
+            # 简单位移检测：灰度差
+            gray1 = np.mean(frame1_small, axis=2) if len(frame1_small.shape) == 3 else frame1_small
+            gray2 = np.mean(frame2_small, axis=2) if len(frame2_small.shape) == 3 else frame2_small
+
+            diff = np.abs(gray2.astype(float) - gray1.astype(float))
+            displacement = np.mean(diff) / 255.0
+            displacements.append(displacement)
+
+        # 高位移方差 = 不稳定
+        if displacements:
+            displacement_variance = np.var(displacements)
+            # 归一化
+            instability = min(1.0, displacement_variance * 10)
+            score = 15.0 * (1.0 - instability)
+            return float(max(0.0, score))
+
+        return 10.0
+
+    except Exception:
+        return 10.0
+
+
+def calculate_quality_score(
+    info: VideoInfo,
+    frames: Optional[list[np.ndarray]] = None,
+) -> QualityScore:
+    """计算视频质量综合评分 (0-100).
+
+    评分维度:
+    1. 分辨率得分 (25分)
+    2. 帧率得分 (20分)
+    3. 码率得分 (20分)
+    4. 清晰度得分 (20分) - 无帧时默认10分
+    5. 稳定性得分 (15分) - 帧不足时默认10分
+
+    Args:
+        info: 视频基本信息
+        frames: 采样帧列表（可选，无则清晰度/稳定性给默认分）
+
+    Returns:
+        QualityScore: 各维度得分 + 总分
+    """
+    # 1. 分辨率得分
+    resolution_score = score_resolution(info.width, info.height)
+
+    # 2. 帧率得分
+    fps_score = score_framerate(info.fps)
+
+    # 3. 码率得分
+    bitrate_score = score_bitrate(info.bitrate)
+
+    # 4. 清晰度得分
+    clarity_score = score_clarity(frames) if frames else 10.0
+
+    # 5. 稳定性得分
+    stability_score = score_stability(frames) if frames and len(frames) >= 2 else 10.0
+
+    total = resolution_score + fps_score + bitrate_score + clarity_score + stability_score
+
+    return QualityScore(
+        total=round(min(100.0, max(0.0, total)), 1),
+        resolution_score=resolution_score,
+        fps_score=fps_score,
+        bitrate_score=bitrate_score,
+        clarity_score=clarity_score,
+        stability_score=stability_score,
+    )
+
+
+# ── 分类评分纯函数 ────────────────────────────────────────────────────
+
+
+def calculate_category_scores(
+    color: ColorAnalysis,
+    motion: MotionAnalysis,
+    audio: AudioAnalysis,
+) -> dict[str, float]:
+    """计算各类别的置信度得分.
+
+    9 个分类：风景、产品、人物、动物、美食、科技、运动、音乐、其他.
+
+    Args:
+        color: 色彩分析结果
+        motion: 运动分析结果
+        audio: 音频分析结果
+
+    Returns:
+        dict[str, float]: 各分类名称 -> 得分 (0-1)
+    """
+    scores: dict[str, float] = {}
+
+    # 1. 风景 (scenic) - 绿色、户外、自然
+    scenic_score = 0.0
+    if color.green_ratio > 0.3:
+        scenic_score += 0.4 * color.green_ratio
+    if color.avg_saturation > 0.3:
+        scenic_score += 0.2 * color.avg_saturation
+    if color.avg_brightness > 0.4:
+        scenic_score += 0.2
+    if 0.1 < motion.motion_score < 0.5:
+        scenic_score += 0.2  # 适度运动（如云朵、树叶）
+    if not audio.has_audio or audio.ambient_ratio > 0.5:
+        scenic_score += 0.2  # 自然环境音
+    scores[AssetClassification.SCENIC.value] = min(1.0, scenic_score)
+
+    # 2. 产品 (product) - 中等亮度、均匀色彩、低运动
+    product_score = 0.0
+    if 0.3 < color.avg_brightness < 0.7:
+        product_score += 0.3
+    if color.avg_saturation < 0.5:
+        product_score += 0.2
+    if motion.motion_score < 0.15:
+        product_score += 0.4  # 低运动 = 产品展示
+    if color.cool_ratio > 0.3:
+        product_score += 0.2  # 冷色调 = 科技感
+    scores[AssetClassification.PRODUCT.value] = min(1.0, product_score)
+
+    # 3. 人物 (person) - 中等运动、有时有人声
+    person_score = 0.0
+    if 0.1 < motion.motion_score < 0.4:
+        person_score += 0.3  # 适度运动
+    if audio.has_audio and audio.speech_ratio > 0.3:
+        person_score += 0.5  # 有人声
+    if color.avg_brightness > 0.3:
+        person_score += 0.2
+    scores[AssetClassification.PERSON.value] = min(1.0, person_score)
+
+    # 4. 动物 (animal) - 高运动、有时自然音
+    animal_score = 0.0
+    if motion.motion_score > 0.3:
+        animal_score += 0.4  # 高运动
+    if motion.scene_changes > 2:
+        animal_score += 0.2
+    if audio.has_audio and (audio.ambient_ratio > 0.3 or audio.speech_ratio > 0.2):
+        animal_score += 0.3
+    scores[AssetClassification.ANIMAL.value] = min(1.0, animal_score)
+
+    # 5. 美食 (food) - 暖色调、高饱和度
+    food_score = 0.0
+    if color.warm_ratio > 0.4:
+        food_score += 0.5
+    if color.avg_saturation > 0.5:
+        food_score += 0.3
+    if 0.4 < color.avg_brightness < 0.8:
+        food_score += 0.2
+    scores[AssetClassification.FOOD.value] = min(1.0, food_score)
+
+    # 6. 科技 (tech) - 冷色调、低饱和度、低运动
+    tech_score = 0.0
+    if color.cool_ratio > 0.4:
+        tech_score += 0.4
+    if color.avg_saturation < 0.4:
+        tech_score += 0.3
+    if motion.motion_score < 0.2:
+        tech_score += 0.3
+    scores[AssetClassification.TECH.value] = min(1.0, tech_score)
+
+    # 7. 运动 (sport) - 高运动
+    sport_score = 0.0
+    if motion.motion_score > 0.4:
+        sport_score += 0.6
+    if motion.scene_changes > 3:
+        sport_score += 0.2
+    if color.avg_brightness > 0.4:
+        sport_score += 0.2
+    scores[AssetClassification.SPORT.value] = min(1.0, sport_score)
+
+    # 8. 音乐 (music) - 有节奏性音乐
+    music_score = 0.0
+    if audio.has_audio and audio.music_ratio > 0.4:
+        music_score += 0.6
+    # 纯视觉判断：色彩丰富但非自然
+    if color.avg_saturation > 0.5 and color.green_ratio < 0.2:
+        music_score += 0.3
+    scores[AssetClassification.MUSIC.value] = min(1.0, music_score)
+
+    # 9. 其他 (other) - 默认最低分
+    scores[AssetClassification.OTHER.value] = 0.1
+
+    return scores
+
+
+def classify_from_analysis(
+    color: ColorAnalysis,
+    motion: MotionAnalysis,
+    audio: AudioAnalysis,
+) -> ClassificationResult:
+    """综合分析得出分类结果.
+
+    Args:
+        color: 色彩分析结果
+        motion: 运动分析结果
+        audio: 音频分析结果
+
+    Returns:
+        ClassificationResult: 分类结果（最高分类别 + 置信度 + 全部分数）
+    """
+    scores = calculate_category_scores(color, motion, audio)
+
+    if not scores:
+        return ClassificationResult(
+            category=AssetClassification.OTHER,
+            confidence=0.3,
+            scores={},
+        )
+
+    best_category = max(scores.items(), key=lambda x: x[1])
+    category = AssetClassification(best_category[0])
+    confidence = min(0.95, max(0.3, best_category[1]))
+
+    return ClassificationResult(
+        category=category,
+        confidence=confidence,
+        scores=scores,
+    )
@@ -17,7 +17,6 @@ import logging
 import os
 import tempfile
 import time
-from dataclasses import dataclass, field
 from pathlib import Path
 from typing import Any

@@ -25,6 +24,15 @@ from worker_app.celery_app import celery_app
 from worker_app.db import SessionLocal

 from packages.domain.bgm_utils import merge_bgm_config
+from video_processing.ffmpeg_utils import probe_duration
+from worker_app.tasks.generation_plan_builder import (
+    VirtualPlan as _VirtualPlan,
+    VirtualClip as _VirtualClip,
+    build_error_info as _build_error_info,
+    extract_intro_outro_from_clip_configs as _extract_intro_outro_from_clip_configs,
+    apply_template_clip_effects as _apply_template_clip_effects,
+    build_clips_by_mode,
+)

 OUTPUT_WIDTH = 1280
 OUTPUT_HEIGHT = 720
@@ -88,36 +96,6 @@ def _update_task_status(task_id: str, status_action: str, **kwargs) -> bool:
        return False


-def _build_error_info(error: Exception, stage: str = "render") -> dict:
-    """构建结构化错误信息。
-
-    Args:
-        error: 异常对象
-        stage: 发生错误的阶段（download/render/merge/upload等）
-
-    Returns:
-        包含 error_type, message, stack_trace, stage, failed_at 的字典
-    """
-    import traceback
-    from datetime import datetime, timezone
-
-    tb_str = traceback.format_exc()
-    # 截取堆栈前20行，避免字段过大
-    tb_lines = tb_str.strip().splitlines()
-    if len(tb_lines) > 20:
-        tb_summary = "\n".join(tb_lines[:20]) + f"\n... (truncated, total {len(tb_lines)} lines)"
-    else:
-        tb_summary = tb_str
-
-    return {
-        "error_type": type(error).__name__,
-        "message": str(error),
-        "stack_trace": tb_summary,
-        "stage": stage,
-        "failed_at": datetime.now(timezone.utc).isoformat(),
-    }
-
-
 # ── 日志持久化辅助 ────────────────────────────────────────────────────────────


@@ -148,36 +126,6 @@ from video_processing.oss_helpers import (
    upload_to_oss,
 )

-# ── 虚拟 Plan / Clip（内存中构建，不写数据库） ────────────────────────────────
-
-
-@dataclass
-class _VirtualPlan:
-    """内存中的虚拟剪辑计划，供 UnifiedRenderService 使用。"""
-
-    id: str
-    name: str = ""
-    config: dict[str, Any] = field(default_factory=dict)
-
-
-@dataclass
-class _VirtualClip:
-    """内存中的虚拟剪辑片段，供 UnifiedRenderService 使用。"""
-
-    id: str
-    plan_id: str = ""
-    clip_type: str = "main"
-    order: int = 0
-    asset_id: str = ""
-    text_content: str = ""
-    start_time: float = 0.0
-    duration: float = 0.0
-    transition_effect: str = "cut"
-    transition_duration: float = 0.0  # 0 表示使用全局默认值
-    playback_speed: float = 1.0
-    status: str = "ready"
-    config: dict[str, Any] = field(default_factory=dict)
-

 def _load_template_clip_configs(template_id: str) -> list:
    """从数据库读取模板的片段配置列表。
@@ -206,124 +154,6 @@ def _load_template_clip_configs(template_id: str) -> list:
        return []


-def _extract_intro_outro_from_clip_configs(clip_configs: list) -> dict[str, Any]:
-    """从模板的 intro/outro 类型 clip_config 中提取 plan 级 intro_outro 配置。
-
-    UnifiedRenderService 已支持 plan.config.intro_outro 路径，
-    这里把 intro/outro 片段配置转为统一格式注入。
-    """
-    intro_configs = [
-        c for c in clip_configs if (c.clip_type.value if hasattr(c.clip_type, "value") else c.clip_type) == "intro"
-    ]
-    outro_configs = [
-        c for c in clip_configs if (c.clip_type.value if hasattr(c.clip_type, "value") else c.clip_type) == "outro"
-    ]
-
-    result: dict[str, Any] = {}
-
-    if intro_configs:
-        intro = intro_configs[0]
-        intro_cfg = intro.config or {}
-        result["has_intro"] = True
-        result["intro_type"] = intro_cfg.get("intro_type", "text")
-        result["intro_duration"] = intro.default_duration or 3.0
-        if intro.text_template:
-            result["intro_text"] = intro.text_template
-        # 透传额外配置
-        for key in ("intro_text_color", "intro_bg_color", "intro_font_size", "intro_video_url", "intro_video_path"):
-            if key in intro_cfg:
-                result[key] = intro_cfg[key]
-
-    if outro_configs:
-        outro = outro_configs[0]
-        outro_cfg = outro.config or {}
-        result["has_outro"] = True
-        result["outro_type"] = outro_cfg.get("outro_type", "text")
-        result["outro_duration"] = outro.default_duration or 3.0
-        if outro.text_template:
-            result["outro_text"] = outro.text_template
-        for key in ("outro_text_color", "outro_bg_color", "outro_font_size", "outro_follow_text"):
-            if key in outro_cfg:
-                result[key] = outro_cfg[key]
-
-    return result
-
-
-def _apply_template_clip_effects(
-    clips: list[_VirtualClip],
-    clip_configs: list,
-    mode: str,
-) -> None:
-    """将模板的 clip 级效果层映射到素材 clips 上（就地修改）。
-
-    映射规则：
-    - 只对素材主体 clips 做映射（ONE_TAKE: main, PIP: main+overlay, VOICE_OVER: main, VOICE_PIP: background+b_roll）
-    - 从模板中筛选 main 类型的 clip_config 作为效果模板
-    - 素材 clips 按顺序循环匹配模板 clip_config（素材多的话重复使用最后一个模板配置）
-    - 映射字段：transition_effect, config.color_grade, config.speed
-    """
-    if not clip_configs or not clips:
-        return
-
-    # 筛选 main 类型的模板配置（作为效果模板池）
-    main_configs = [
-        c
-        for c in clip_configs
-        if (c.clip_type.value if hasattr(c.clip_type, "value") else c.clip_type) in ("main", "showcase", "b_roll")
-    ]
-    if not main_configs:
-        return
-
-    # 确定需要映射的素材 clips（排除 corner_voice 等特殊层）
-    target_clips = [c for c in clips if c.clip_type not in ("corner_voice",)]
-
-    for i, clip in enumerate(target_clips):
-        # 循环匹配：素材多了用最后一个模板配置
-        cfg_idx = min(i, len(main_configs) - 1)
-        template_cfg = main_configs[cfg_idx]
-
-        # 1. 转场效果 + 时长
-        transition = (
-            template_cfg.transition_effect.value
-            if hasattr(template_cfg.transition_effect, "value")
-            else template_cfg.transition_effect
-        )
-        if transition and transition != "cut":
-            clip.transition_effect = transition
-            # 同步转场时长（模板 clip_config 里的 transition_duration）
-            tpl_cfg = template_cfg.config or {}
-            tpl_duration = tpl_cfg.get("transition_duration")
-            if tpl_duration:
-                try:
-                    dur_val = float(tpl_duration)
-                    if dur_val > 0:
-                        clip.transition_duration = dur_val
-                except (ValueError, TypeError):
-                    pass
-
-        # 2. clip 级效果配置（滤镜、调速等）
-        template_clip_config = template_cfg.config or {}
-        if template_clip_config:
-            # 合并到 clip.config（保留已有配置如 role 等）
-            existing_config = clip.config or {}
-            # 需要从模板复制的效果层 key
-            effect_keys = ("color_grade", "speed", "playback_speed", "reverse", "chroma_key", "filter")
-            for key in effect_keys:
-                if key in template_clip_config:
-                    existing_config[key] = template_clip_config[key]
-            clip.config = existing_config
-
-            # 3. 调速：同步到 clip.playback_speed 顶级字段（渲染引擎读此字段）
-            template_speed = template_clip_config.get("playback_speed") or template_clip_config.get("speed")
-            if template_speed:
-                try:
-                    speed_val = float(template_speed)
-                    if speed_val > 0:
-                        clip.playback_speed = speed_val
-                except (ValueError, TypeError):
-                    pass
-
-
 def _build_plan_and_clips_from_task(
    task_id: str,
    downloaded_paths: list[Path],
@@ -0,0 +1,327 @@
+"""Generation task pure logic utilities — template mapping + plan/clip building.
+
+从 generation.py 抽出来的纯逻辑模块：
+- VirtualPlan / VirtualClip: 内存中的虚拟计划/片段数据类
+- extract_intro_outro_from_clip_configs: 从模板 clip_config 提取片头片尾配置
+- apply_template_clip_effects: 将模板效果层映射到素材 clips
+- build_clips_by_mode: 根据模式和素材列表构建虚拟 clips
+- build_error_info: 构建结构化错误信息
+"""
+
+from __future__ import annotations
+
+import traceback
+from dataclasses import dataclass, field
+from datetime import datetime, timezone
+from pathlib import Path
+from typing import Any, Optional
+
+# ── 数据类 ───────────────────────────────────────────────────────────────────
+
+
+@dataclass
+class VirtualPlan:
+    """内存中的虚拟剪辑计划，供 UnifiedRenderService 使用。"""
+
+    id: str
+    name: str = ""
+    config: dict[str, Any] = field(default_factory=dict)
+
+
+@dataclass
+class VirtualClip:
+    """内存中的虚拟剪辑片段，供 UnifiedRenderService 使用。"""
+
+    id: str
+    plan_id: str = ""
+    clip_type: str = "main"
+    order: int = 0
+    asset_id: str = ""
+    text_content: str = ""
+    start_time: float = 0.0
+    duration: float = 0.0
+    transition_effect: str = "cut"
+    transition_duration: float = 0.0  # 0 表示使用全局默认值
+    playback_speed: float = 1.0
+    status: str = "ready"
+    config: dict[str, Any] = field(default_factory=dict)
+
+
+# ── 模板片头片尾提取 ────────────────────────────────────────────────────────
+
+
+def _clip_type_value(c: Any) -> str:
+    """获取 clip_config 的 clip_type 字符串值（兼容 Enum 和 str）。"""
+    if hasattr(c, "value"):
+        return str(c.value)
+    return str(c)
+
+
+def _transition_value(t: Any) -> str:
+    """获取 transition_effect 字符串值（兼容 Enum 和 str）。"""
+    if hasattr(t, "value"):
+        return str(t.value)
+    return str(t) if t else ""
+
+
+def extract_intro_outro_from_clip_configs(clip_configs: list) -> dict[str, Any]:
+    """从模板的 intro/outro 类型 clip_config 中提取 plan 级 intro_outro 配置。
+
+    UnifiedRenderService 已支持 plan.config.intro_outro 路径，
+    这里把 intro/outro 片段配置转为统一格式注入。
+    """
+    intro_configs = [c for c in clip_configs if _clip_type_value(c.clip_type) == "intro"]
+    outro_configs = [c for c in clip_configs if _clip_type_value(c.clip_type) == "outro"]
+
+    result: dict[str, Any] = {}
+
+    if intro_configs:
+        intro = intro_configs[0]
+        intro_cfg = intro.config or {}
+        result["has_intro"] = True
+        result["intro_type"] = intro_cfg.get("intro_type", "text")
+        result["intro_duration"] = getattr(intro, "default_duration", 3.0) or 3.0
+        intro_text = getattr(intro, "text_template", "")
+        if intro_text:
+            result["intro_text"] = intro_text
+        # 透传额外配置
+        for key in (
+            "intro_text_color",
+            "intro_bg_color",
+            "intro_font_size",
+            "intro_video_url",
+            "intro_video_path",
+        ):
+            if key in intro_cfg:
+                result[key] = intro_cfg[key]
+
+    if outro_configs:
+        outro = outro_configs[0]
+        outro_cfg = outro.config or {}
+        result["has_outro"] = True
+        result["outro_type"] = outro_cfg.get("outro_type", "text")
+        result["outro_duration"] = getattr(outro, "default_duration", 3.0) or 3.0
+        outro_text = getattr(outro, "text_template", "")
+        if outro_text:
+            result["outro_text"] = outro_text
+        for key in (
+            "outro_text_color",
+            "outro_bg_color",
+            "outro_font_size",
+            "outro_follow_text",
+        ):
+            if key in outro_cfg:
+                result[key] = outro_cfg[key]
+
+    return result
+
+
+# ── 模板效果层映射 ──────────────────────────────────────────────────────────
+
+
+# 需要从模板复制的效果层 key
+_TEMPLATE_EFFECT_KEYS = (
+    "color_grade",
+    "speed",
+    "playback_speed",
+    "reverse",
+    "chroma_key",
+    "filter",
+)
+
+# 各模式下需要应用效果的 clip_type
+_EFFECT_TARGET_TYPES = {
+    "one_take": {"main"},
+    "pip": {"main", "overlay"},
+    "voice_over": {"main"},
+    "voice_pip": {"background", "b_roll"},
+}
+
+# 作为效果模板池的 clip_type
+_TEMPLATE_SOURCE_TYPES = {"main", "showcase", "b_roll"}
+
+# 不应用效果的 clip_type
+_SKIP_TYPES = {"corner_voice"}
+
+
+def apply_template_clip_effects(
+    clips: list[VirtualClip],
+    clip_configs: list,
+    mode: str,
+) -> None:
+    """将模板的 clip 级效果层映射到素材 clips 上（就地修改）。
+
+    映射规则：
+    - 只对素材主体 clips 做映射
+    - 从模板中筛选 main/showcase/b_roll 类型的 clip_config 作为效果模板
+    - 素材 clips 按顺序循环匹配模板 clip_config（素材多的话重复使用最后一个模板配置）
+    - 映射字段：transition_effect, transition_duration, config 中的效果层
+    """
+    if not clip_configs or not clips:
+        return
+
+    # 筛选 main 类型的模板配置（作为效果模板池）
+    main_configs = [c for c in clip_configs if _clip_type_value(c.clip_type) in _TEMPLATE_SOURCE_TYPES]
+    if not main_configs:
+        return
+
+    # 确定需要映射的素材 clips（排除特殊层）
+    target_clips = [c for c in clips if c.clip_type not in _SKIP_TYPES]
+
+    for i, clip in enumerate(target_clips):
+        # 循环匹配：素材多了用最后一个模板配置
+        cfg_idx = min(i, len(main_configs) - 1)
+        template_cfg = main_configs[cfg_idx]
+
+        # 1. 转场效果 + 时长
+        transition = _transition_value(template_cfg.transition_effect)
+        if transition and transition != "cut":
+            clip.transition_effect = transition
+            # 同步转场时长
+            tpl_cfg = template_cfg.config or {}
+            tpl_duration = tpl_cfg.get("transition_duration")
+            if tpl_duration:
+                try:
+                    dur_val = float(tpl_duration)
+                    if dur_val > 0:
+                        clip.transition_duration = dur_val
+                except (ValueError, TypeError):
+                    pass
+
+        # 2. clip 级效果配置（滤镜、调速等）
+        template_clip_config = template_cfg.config or {}
+        if template_clip_config:
+            existing_config = clip.config or {}
+            for key in _TEMPLATE_EFFECT_KEYS:
+                if key in template_clip_config:
+                    existing_config[key] = template_clip_config[key]
+            clip.config = existing_config
+
+            # 3. 调速：同步到 clip.playback_speed 顶级字段
+            template_speed = template_clip_config.get("playback_speed") or template_clip_config.get("speed")
+            if template_speed:
+                try:
+                    speed_val = float(template_speed)
+                    if speed_val > 0:
+                        clip.playback_speed = speed_val
+                except (ValueError, TypeError):
+                    pass
+
+
+# ── 按模式构建 clips ────────────────────────────────────────────────────────
+
+
+def build_clips_by_mode(
+    plan_id: str,
+    asset_infos: list[dict[str, Any]],
+    mode: str,
+) -> list[VirtualClip]:
+    """根据生成模式和素材信息，构建 VirtualClip 列表。
+
+    纯逻辑版本：不依赖 ffmpeg probe 或 DB，完全由输入数据驱动。
+
+    Args:
+        plan_id: 计划 ID
+        asset_infos: 素材信息列表，每项包含 asset_id / duration / path 等
+        mode: 生成模式 (one_take / pip / voice_over / voice_pip)
+
+    Returns:
+        VirtualClip 列表，按 order 排序
+
+    模式 → clip_type 映射：
+      one_take:     N 个 main clips
+      pip:          1 main + N-1 overlay
+      voice_over:   N 个 main (config.role=b_roll)
+      voice_pip:    1 background + 1 corner_voice + N-2 b_roll
+    """
+    clips: list[VirtualClip] = []
+
+    for i, info in enumerate(asset_infos):
+        asset_id = info.get("asset_id", f"asset_{i:03d}")
+        duration = float(info.get("duration", 0.0))
+
+        if mode == "pip":
+            clip_type = "main" if i == 0 else "overlay"
+            clips.append(
+                VirtualClip(
+                    id=f"vc_{i:03d}",
+                    plan_id=plan_id,
+                    clip_type=clip_type,
+                    order=i,
+                    asset_id=asset_id,
+                    duration=duration,
+                )
+            )
+        elif mode == "voice_over":
+            clips.append(
+                VirtualClip(
+                    id=f"vc_{i:03d}",
+                    plan_id=plan_id,
+                    clip_type="main",
+                    order=i,
+                    asset_id=asset_id,
+                    duration=duration,
+                    config={"role": "b_roll"},
+                )
+            )
+        elif mode == "voice_pip":
+            if i == 0:
+                clip_type = "background"
+            elif i == 1:
+                clip_type = "corner_voice"
+            else:
+                clip_type = "b_roll"
+            clips.append(
+                VirtualClip(
+                    id=f"vc_{i:03d}",
+                    plan_id=plan_id,
+                    clip_type=clip_type,
+                    order=i,
+                    asset_id=asset_id,
+                    duration=duration,
+                )
+            )
+        else:
+            # one_take (default): N 个 main clips
+            clips.append(
+                VirtualClip(
+                    id=f"vc_{i:03d}",
+                    plan_id=plan_id,
+                    clip_type="main",
+                    order=i,
+                    asset_id=asset_id,
+                    duration=duration,
+                )
+            )
+
+    return clips
+
+
+# ── 错误信息构建 ────────────────────────────────────────────────────────────
+
+
+def build_error_info(error: Exception, stage: str = "render") -> dict[str, Any]:
+    """构建结构化错误信息。
+
+    Args:
+        error: 异常对象
+        stage: 发生错误的阶段
+
+    Returns:
+        包含 error_type, message, stack_trace, stage, failed_at 的字典
+    """
+    tb_str = traceback.format_exc()
+    # 截取堆栈前20行，避免字段过大
+    tb_lines = tb_str.strip().splitlines()
+    if len(tb_lines) > 20:
+        tb_summary = "\n".join(tb_lines[:20]) + f"\n... (truncated, total {len(tb_lines)} lines)"
+    else:
+        tb_summary = tb_str
+
+    return {
+        "error_type": type(error).__name__,
+        "message": str(error),
+        "stack_trace": tb_summary,
+        "stage": stage,
+        "failed_at": datetime.now(timezone.utc).isoformat(),
+    }
@@ -0,0 +1,372 @@
+"""Asset scoring pure logic — multi-dimensional scoring + diverse selection.
+
+从 smart_asset_selector.py 抽出来的纯逻辑模块：
+- 评分维度：质量分、分辨率、时长、码率（加权求和，总分 0-1）
+- 多样性选择：按时长分桶（短/中/长）保证分布均匀
+- 数据类：AssetScoreDetail, SmartSelectResult
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from typing import Any
+
+# ── 评分权重（总和 = 1.0） ────────────────────────────────────────────────────
+
+WEIGHT_QUALITY = 0.5
+WEIGHT_RESOLUTION = 0.2
+WEIGHT_DURATION = 0.2
+WEIGHT_BITRATE = 0.1
+
+# ── 评分参数 ──────────────────────────────────────────────────────────────────
+
+TARGET_WIDTH = 1920  # 目标分辨率宽度基准
+TARGET_HEIGHT = 1080  # 目标分辨率高度基准
+MIN_QUALITY_SCORE = 30.0  # 最低质量分门槛（低于此值的素材直接排除）
+OPTIMAL_DURATION_MIN = 3.0  # 最佳时长区间（秒）
+OPTIMAL_DURATION_MAX = 30.0
+
+# ── 多样性分桶阈值 ───────────────────────────────────────────────────────────
+
+SHORT_BUCKET_MAX = 5.0  # 短素材：< 5s
+MEDIUM_BUCKET_MAX = 15.0  # 中素材：5-15s
+# 长素材：>= 15s
+
+
+# ── 数据类 ───────────────────────────────────────────────────────────────────
+
+
+@dataclass
+class AssetScoreDetail:
+    """单个素材的评分详情."""
+
+    asset_id: str
+    total_score: float
+    quality_score: float
+    resolution_score: float
+    duration_score: float
+    bitrate_score: float
+    duration: float | None
+
+
+@dataclass
+class SmartSelectResult:
+    """智能选择结果."""
+
+    selected_ids: list[str]
+    total_candidates: int
+    filtered_out: int  # 被质量门槛过滤的数量
+    avg_score: float
+    details: list[AssetScoreDetail] = field(default_factory=list)
+
+
+# ── 评分函数 ─────────────────────────────────────────────────────────────────
+
+
+def score_resolution(
+    width: int | None,
+    height: int | None,
+    target_width: int = TARGET_WIDTH,
+    target_height: int = TARGET_HEIGHT,
+) -> float:
+    """分辨率评分：越接近目标分辨率得分越高，低于480p扣分严重.
+
+    Args:
+        width: 素材宽度（像素）
+        height: 素材高度（像素）
+        target_width: 目标宽度
+        target_height: 目标高度
+
+    Returns:
+        0.0 - 1.0 之间的评分
+    """
+    if width is None or height is None or width <= 0 or height <= 0:
+        return 0.5  # 未知分辨率给中评分
+
+    target_pixels = target_width * target_height
+    actual_pixels = width * height
+
+    # 计算像素数比例
+    ratio = actual_pixels / target_pixels
+
+    if ratio >= 1.0:
+        # 高于或等于目标分辨率：满分，略高不扣分（4K也给满分）
+        return 1.0
+    else:
+        # 低于目标分辨率：线性衰减，但最低不低于 0.1
+        score = 0.3 + 0.7 * ratio
+        return max(0.1, min(1.0, score))
+
+
+def score_duration(duration: float | None) -> float:
+    """时长评分：3-30秒最佳，过短或过长都扣分.
+
+    Args:
+        duration: 时长（秒）
+
+    Returns:
+        0.0 - 1.0 之间的评分
+    """
+    if duration is None or duration <= 0:
+        return 0.5  # 未知时长给中评分
+
+    if OPTIMAL_DURATION_MIN <= duration <= OPTIMAL_DURATION_MAX:
+        # 最佳区间：满分
+        return 1.0
+
+    if duration < OPTIMAL_DURATION_MIN:
+        # 太短：线性衰减，趋近于 0.3
+        ratio = duration / OPTIMAL_DURATION_MIN
+        return 0.3 + 0.7 * ratio
+
+    # 太长：每超过最佳区间上限10秒扣 0.1 分，最低 0.2
+    excess = duration - OPTIMAL_DURATION_MAX
+    penalty = min(0.8, excess / 10.0 * 0.1)
+    return max(0.2, 1.0 - penalty)
+
+
+def score_bitrate(file_size: int, duration: float | None) -> float:
+    """码率评分：根据文件大小和时长估算码率，适中得分高.
+
+    Args:
+        file_size: 文件大小（字节）
+        duration: 时长（秒）
+
+    Returns:
+        0.0 - 1.0 之间的评分
+    """
+    if not file_size or not duration or duration <= 0:
+        return 0.5  # 未知给中评分
+
+    # 估算码率（bps）
+    bitrate = (file_size * 8) / duration
+
+    # 最佳码率范围：2-8 Mbps
+    optimal_low = 2_000_000  # 2 Mbps
+    optimal_high = 8_000_000  # 8 Mbps
+
+    if optimal_low <= bitrate <= optimal_high:
+        return 1.0
+
+    if bitrate < optimal_low:
+        # 码率太低：线性衰减
+        ratio = bitrate / optimal_low
+        return 0.3 + 0.7 * ratio
+
+    # 码率太高（文件太大）：适度扣分，最低 0.5
+    excess = bitrate / optimal_high - 1.0
+    penalty = min(0.5, excess * 0.2)
+    return max(0.5, 1.0 - penalty)
+
+
+def calculate_total_score(
+    quality_score: float,
+    resolution_score: float,
+    duration_score: float,
+    bitrate_score: float,
+) -> float:
+    """计算加权总分.
+
+    Args:
+        quality_score: 质量分（0-1）
+        resolution_score: 分辨率分（0-1）
+        duration_score: 时长分（0-1）
+        bitrate_score: 码率分（0-1）
+
+    Returns:
+        加权总分（0-1）
+    """
+    total = (
+        WEIGHT_QUALITY * quality_score
+        + WEIGHT_RESOLUTION * resolution_score
+        + WEIGHT_DURATION * duration_score
+        + WEIGHT_BITRATE * bitrate_score
+    )
+    return round(total, 4)
+
+
+def score_asset_detail(
+    asset_id: str,
+    quality: float | None,
+    width: int | None,
+    height: int | None,
+    duration: float | None,
+    file_size: int,
+    target_width: int = TARGET_WIDTH,
+    target_height: int = TARGET_HEIGHT,
+) -> AssetScoreDetail:
+    """对单个素材进行多维度评分，返回详细评分结果.
+
+    Args:
+        asset_id: 素材ID
+        quality: 质量分（0-100，None表示未知）
+        width: 宽度
+        height: 高度
+        duration: 时长
+        file_size: 文件大小
+        target_width: 目标宽度
+        target_height: 目标高度
+
+    Returns:
+        AssetScoreDetail 评分详情
+    """
+    # 质量分归一化到 0-1
+    quality_score = (quality / 100.0) if quality is not None else 0.5
+
+    resolution_score = score_resolution(width, height, target_width, target_height)
+    duration_score = score_duration(duration)
+    bitrate_score = score_bitrate(file_size, duration)
+
+    total_score = calculate_total_score(
+        quality_score,
+        resolution_score,
+        duration_score,
+        bitrate_score,
+    )
+
+    return AssetScoreDetail(
+        asset_id=asset_id,
+        total_score=total_score,
+        quality_score=round(quality_score, 4),
+        resolution_score=round(resolution_score, 4),
+        duration_score=round(duration_score, 4),
+        bitrate_score=round(bitrate_score, 4),
+        duration=duration,
+    )
+
+
+# ── 多样性选择 ───────────────────────────────────────────────────────────────
+
+
+def _bucket_by_duration(item: AssetScoreDetail) -> str:
+    """根据时长判断所属桶.
+
+    Returns:
+        'short' / 'medium' / 'long' / 'unknown'
+    """
+    if item.duration is None:
+        return "unknown"
+    if item.duration < SHORT_BUCKET_MAX:
+        return "short"
+    if item.duration < MEDIUM_BUCKET_MAX:
+        return "medium"
+    return "long"
+
+
+def diverse_selection(
+    scored: list[AssetScoreDetail],
+    count: int,
+) -> list[AssetScoreDetail]:
+    """多样性选择：按时长分桶，保证每个桶都有素材.
+
+    策略：
+    1. 按时长分为三桶：短(<5s)、中(5-15s)、长(>=15s)
+    2. 每个桶配额 = max(1, count // 3)
+    3. 先从每桶按配额取最高分的
+    4. 剩余名额从全局最高分中取（不重复）
+    5. 如果还不够，加上未知时长的
+
+    Args:
+        scored: 已按总分降序排列的评分列表
+        count: 需要选取的数量
+
+    Returns:
+        选中的评分列表（不超过 count 个）
+    """
+    if count <= 0 or not scored:
+        return []
+
+    # 分桶
+    short_bucket = [d for d in scored if _bucket_by_duration(d) == "short"]
+    medium_bucket = [d for d in scored if _bucket_by_duration(d) == "medium"]
+    long_bucket = [d for d in scored if _bucket_by_duration(d) == "long"]
+    unknown_bucket = [d for d in scored if _bucket_by_duration(d) == "unknown"]
+
+    buckets = [short_bucket, medium_bucket, long_bucket]
+
+    # 每个桶基础配额（至少1个，如果桶非空且需要的话）
+    base_quota = max(1, count // 3)
+
+    selected: list[AssetScoreDetail] = []
+    selected_ids: set[str] = set()
+
+    # 先按配额从每个桶取
+    for bucket in buckets:
+        quota = min(base_quota, len(bucket))
+        if quota <= 0:
+            continue
+        # 桶内已经按分数排好序了，直接取前 quota 个
+        for item in bucket[:quota]:
+            if item.asset_id not in selected_ids:
+                selected.append(item)
+                selected_ids.add(item.asset_id)
+                if len(selected) >= count:
+                    return selected
+
+    # 剩余名额：从全局（未被选中的）中按分数高低取
+    remaining_needed = count - len(selected)
+    if remaining_needed > 0:
+        for item in scored:
+            if item.asset_id not in selected_ids:
+                selected.append(item)
+                selected_ids.add(item.asset_id)
+                if len(selected) >= count:
+                    break
+
+    # 如果还不够（不应该发生），加上未知时长的
+    if len(selected) < count and unknown_bucket:
+        for item in unknown_bucket:
+            if item.asset_id not in selected_ids:
+                selected.append(item)
+                selected_ids.add(item.asset_id)
+                if len(selected) >= count:
+                    break
+
+    return selected[:count]
+
+
+# ── 候选过滤 ─────────────────────────────────────────────────────────────────
+
+
+def filter_candidates(
+    assets: list[Any],
+    min_quality_score: float = MIN_QUALITY_SCORE,
+) -> tuple[list[Any], int]:
+    """从素材列表中筛选出合格的候选素材.
+
+    筛选条件：
+    - status == 'ready'
+    - mime_type 以 'video' 开头
+    - quality_score >= min_quality_score（如果quality不为None）
+
+    Args:
+        assets: 素材列表
+        min_quality_score: 最低质量分门槛
+
+    Returns:
+        (合格素材列表, 被质量门槛过滤的数量)
+    """
+    candidates = []
+    filtered_out = 0
+
+    for asset in assets:
+        # 状态检查
+        status = getattr(asset, "status", None)
+        status_val = status.value if hasattr(status, "value") else str(status)
+        if status_val != "ready":
+            continue
+
+        # 类型检查
+        mime_type = getattr(asset, "mime_type", "") or ""
+        if not mime_type.startswith("video"):
+            continue
+
+        # 质量分门槛
+        quality = getattr(asset, "quality_score", None)
+        if quality is not None and quality < min_quality_score:
+            filtered_out += 1
+            continue
+
+        candidates.append(asset)
+
+    return candidates, filtered_out
@@ -0,0 +1,347 @@
+"""剪辑计划生成 — 纯逻辑工具函数.
+
+从 PlanGeneratorService 提取的纯业务逻辑：
+- 素材分配策略（4 种 editing_mode）
+- 默认 clip 结构生成
+- clip_type 按模式映射
+- 从 TemplateClipConfig 创建 EditPlanClip
+
+纯函数，无副作用，不依赖 DB/外部服务。
+"""
+
+from __future__ import annotations
+
+from typing import List
+
+from packages.domain.edit_plan_clip import EditPlanClip
+from packages.domain.editing_mode import EditingMode
+from packages.domain.template_clip_config import ClipType, TemplateClipConfig
+
+# ── 默认片段时长（秒） ────────────────────────────────────────────────────────
+DEFAULT_CLIP_DURATION = 5.0
+DEFAULT_INTRO_DURATION = 3.0
+DEFAULT_OUTRO_DURATION = 3.0
+
+
+# ── 素材分配 ────────────────────────────────────────────────────────────────
+
+
+def distribute_assets(
+    clips: List[EditPlanClip],
+    asset_ids: List[str],
+    editing_mode: str,
+) -> None:
+    """按 editing_mode 将素材分配到 clips（就地修改）.
+
+    分配策略:
+    - ONE_TAKE: 素材按顺序依次分配给 main 类型 clips
+    - PIP: 第1个素材→main(全屏背景)，其余→overlay clips
+    - VOICE_OVER: 素材→main clips (B-roll)
+    - VOICE_PIP: 第1个→background, 第2个→corner_voice, 其余→b_roll
+
+    Args:
+        clips: 剪辑片段列表（就地修改 asset_id）
+        asset_ids: 素材 ID 列表
+        editing_mode: 剪辑模式字符串
+    """
+    if not asset_ids or not clips:
+        return
+
+    if editing_mode == EditingMode.ONE_TAKE.value:
+        _distribute_one_take(clips, asset_ids)
+    elif editing_mode == EditingMode.PIP.value:
+        _distribute_pip(clips, asset_ids)
+    elif editing_mode == EditingMode.VOICE_OVER.value:
+        _distribute_voice_over(clips, asset_ids)
+    elif editing_mode == EditingMode.VOICE_PIP.value:
+        _distribute_voice_pip(clips, asset_ids)
+    else:
+        # 未知模式，退化为 one_take
+        _distribute_one_take(clips, asset_ids)
+
+
+def _distribute_one_take(
+    clips: List[EditPlanClip],
+    asset_ids: List[str],
+) -> None:
+    """ONE_TAKE: 素材按顺序依次分配给 main 类型 clips."""
+    main_clips = [c for c in clips if c.clip_type == ClipType.MAIN.value]
+    for i, clip in enumerate(main_clips):
+        if i < len(asset_ids):
+            clip.assign_asset(asset_ids[i])
+
+
+def _distribute_pip(
+    clips: List[EditPlanClip],
+    asset_ids: List[str],
+) -> None:
+    """PIP: 第1个素材→main(全屏背景)，其余→overlay clips."""
+    # 第1个素材 → main clip
+    main_clips = [c for c in clips if c.clip_type == ClipType.MAIN.value]
+    if main_clips and asset_ids:
+        main_clips[0].assign_asset(asset_ids[0])
+
+    # 其余素材 → overlay clips
+    overlay_clips = [c for c in clips if c.clip_type == "overlay"]
+    remaining = asset_ids[1:]
+    for i, clip in enumerate(overlay_clips):
+        if i < len(remaining):
+            clip.assign_asset(remaining[i])
+
+
+def _distribute_voice_over(
+    clips: List[EditPlanClip],
+    asset_ids: List[str],
+) -> None:
+    """VOICE_OVER: 素材→main clips (B-roll)."""
+    main_clips = [c for c in clips if c.clip_type == ClipType.MAIN.value]
+    for i, clip in enumerate(main_clips):
+        if i < len(asset_ids):
+            clip.assign_asset(asset_ids[i])
+
+
+def _distribute_voice_pip(
+    clips: List[EditPlanClip],
+    asset_ids: List[str],
+) -> None:
+    """VOICE_PIP: 第1个→background, 第2个→corner_voice, 其余→b_roll."""
+    bg_clips = [c for c in clips if c.clip_type == "background"]
+    voice_clips = [c for c in clips if c.clip_type == "corner_voice"]
+    broll_clips = [c for c in clips if c.clip_type == "b_roll"]
+
+    idx = 0
+
+    # 第1个 → background
+    if idx < len(asset_ids) and bg_clips:
+        bg_clips[0].assign_asset(asset_ids[idx])
+        idx += 1
+
+    # 第2个 → corner_voice
+    if idx < len(asset_ids) and voice_clips:
+        voice_clips[0].assign_asset(asset_ids[idx])
+        idx += 1
+
+    # 剩余 → b_roll clips
+    remaining = asset_ids[idx:]
+    for i, clip in enumerate(broll_clips):
+        if i < len(remaining):
+            clip.assign_asset(remaining[i])
+
+
+# ── clip_type 映射 ────────────────────────────────────────────────────────
+
+
+def map_clip_types_for_mode(
+    clips: List[EditPlanClip],
+    editing_mode: str,
+) -> None:
+    """将 MAIN 类型片段按 editing_mode 映射为对应角色类型.
+
+    模板的 clip_config 使用 ClipType 枚举（main/intro/outro 等），
+    但 PIP / VOICE_PIP 模式的素材分配和渲染分层依赖特定的 clip_type 命名
+    （overlay / background / corner_voice / b_roll）。
+
+    映射规则（仅修改 MAIN 类型片段，非 MAIN 片段保持原类型）：
+    - PIP: 第1个 MAIN → main（背景），其余 MAIN → overlay（画中画）
+    - VOICE_PIP: 第1个 → background，第2个 → corner_voice，第3+个 → b_roll
+    - ONE_TAKE / VOICE_OVER: 保持 main 不变
+
+    Args:
+        clips: 剪辑片段列表（就地修改 clip_type）
+        editing_mode: 剪辑模式字符串
+    """
+    main_clips = [c for c in clips if c.clip_type == ClipType.MAIN.value]
+    if not main_clips:
+        return
+
+    if editing_mode == EditingMode.PIP.value:
+        # 第1个 main 保持（背景层），其余改为 overlay（画中画层）
+        for i, clip in enumerate(main_clips):
+            if i > 0:
+                clip.clip_type = "overlay"
+
+    elif editing_mode == EditingMode.VOICE_PIP.value:
+        for i, clip in enumerate(main_clips):
+            if i == 0:
+                clip.clip_type = "background"
+            elif i == 1:
+                clip.clip_type = "corner_voice"
+            else:
+                clip.clip_type = "b_roll"
+
+    # ONE_TAKE / VOICE_OVER: 保持 main 不变，无需处理
+
+
+# ── 默认 clip 生成 ────────────────────────────────────────────────────────
+
+
+def generate_default_clips(
+    plan_id: str,
+    editing_mode: str,
+    asset_count: int,
+) -> List[EditPlanClip]:
+    """无 clip_configs 时，根据 editing_mode 生成默认 clip 结构.
+
+    - ONE_TAKE: N 个 main clips（N = asset_count，至少1个）
+    - PIP: 1 个 main + (N-1) 个 overlay（N = asset_count）
+    - VOICE_OVER: N 个 main clips + 标记需要配音
+    - VOICE_PIP: 1 个 background + 1 个 corner_voice + (N-2) 个 b_roll
+
+    Args:
+        plan_id: 剪辑计划 ID
+        editing_mode: 剪辑模式字符串
+        asset_count: 素材数量
+
+    Returns:
+        List[EditPlanClip]: 生成的默认剪辑片段列表
+    """
+    n = max(asset_count, 1)
+    clips: List[EditPlanClip] = []
+    order = 0
+
+    if editing_mode == EditingMode.PIP.value:
+        # 1 个 main（全屏背景）
+        clips.append(
+            EditPlanClip.create(
+                plan_id=plan_id,
+                clip_type=ClipType.MAIN.value,
+                order=order,
+                duration=DEFAULT_CLIP_DURATION,
+            )
+        )
+        order += 1
+        # 剩余为 overlay
+        for _ in range(1, n):
+            clips.append(
+                EditPlanClip.create(
+                    plan_id=plan_id,
+                    clip_type="overlay",
+                    order=order,
+                    duration=DEFAULT_CLIP_DURATION,
+                )
+            )
+            order += 1
+
+    elif editing_mode == EditingMode.VOICE_OVER.value:
+        # N 个 main clips（B-roll）
+        for _ in range(n):
+            clips.append(
+                EditPlanClip.create(
+                    plan_id=plan_id,
+                    clip_type=ClipType.MAIN.value,
+                    order=order,
+                    duration=DEFAULT_CLIP_DURATION,
+                    config={"role": "b_roll"},
+                )
+            )
+            order += 1
+
+    elif editing_mode == EditingMode.VOICE_PIP.value:
+        # 1 个 background
+        clips.append(
+            EditPlanClip.create(
+                plan_id=plan_id,
+                clip_type="background",
+                order=order,
+                duration=DEFAULT_CLIP_DURATION,
+            )
+        )
+        order += 1
+        # 1 个 corner_voice（至少有1个素材就有）
+        if n >= 2:
+            clips.append(
+                EditPlanClip.create(
+                    plan_id=plan_id,
+                    clip_type="corner_voice",
+                    order=order,
+                    duration=DEFAULT_CLIP_DURATION,
+                )
+            )
+            order += 1
+        # 剩余为 b_roll
+        for _ in range(2, n):
+            clips.append(
+                EditPlanClip.create(
+                    plan_id=plan_id,
+                    clip_type="b_roll",
+                    order=order,
+                    duration=DEFAULT_CLIP_DURATION,
+                )
+            )
+            order += 1
+
+    else:
+        # ONE_TAKE / 未知模式：N 个 main clips
+        for _ in range(n):
+            clips.append(
+                EditPlanClip.create(
+                    plan_id=plan_id,
+                    clip_type=ClipType.MAIN.value,
+                    order=order,
+                    duration=DEFAULT_CLIP_DURATION,
+                )
+            )
+            order += 1
+
+    return clips
+
+
+# ── 从配置创建 clips ────────────────────────────────────────────────────────
+
+
+def create_clips_from_configs(
+    plan_id: str,
+    clip_configs: List[TemplateClipConfig],
+) -> List[EditPlanClip]:
+    """从 TemplateClipConfig 列表创建 EditPlanClip 列表.
+
+    Args:
+        plan_id: 剪辑计划 ID
+        clip_configs: 模板片段配置列表
+
+    Returns:
+        List[EditPlanClip]: 创建的剪辑片段列表（按 order 排序）
+    """
+    clips: List[EditPlanClip] = []
+    # 按 order 排序
+    sorted_configs = sorted(clip_configs, key=lambda c: c.order)
+
+    for cfg in sorted_configs:
+        # 计算时长：取 min_duration 和 max_duration 的中间值
+        if cfg.min_duration > 0 and cfg.max_duration > 0:
+            duration = (cfg.min_duration + cfg.max_duration) / 2
+        elif cfg.min_duration > 0:
+            duration = cfg.min_duration
+        elif cfg.max_duration > 0:
+            duration = cfg.max_duration
+        else:
+            duration = DEFAULT_CLIP_DURATION
+
+        # clip_type 可能是枚举或字符串
+        clip_type = cfg.clip_type.value if hasattr(cfg.clip_type, "value") else cfg.clip_type
+
+        # transition_effect 可能是枚举或字符串
+        transition = (
+            cfg.transition_effect.value
+            if hasattr(cfg.transition_effect, "value")
+            else cfg.transition_effect
+        )
+
+        # 从 clip config 中解析 playback_speed（兼容 speed_ratio 字段名）
+        clip_cfg = cfg.config or {}
+        playback_speed = clip_cfg.get("playback_speed", clip_cfg.get("speed_ratio", 1.0)) or 1.0
+
+        clip = EditPlanClip.create(
+            plan_id=plan_id,
+            clip_type=clip_type,
+            order=cfg.order,
+            template_clip_config_id=cfg.id,
+            text_content=getattr(cfg, "text_template", "") or "",
+            duration=duration,
+            transition_effect=transition or "cut",
+            playback_speed=playback_speed,
+            config=clip_cfg,
+        )
+        clips.append(clip)
+
+    return clips
@@ -0,0 +1,241 @@
+"""渲染图层工具函数 — 纯函数集合.
+
+从 unified_render_service.py 抽离的纯逻辑，负责：
+- clip 时长计算（有效时长、调速后时长）
+- clip_type → layer_role 映射
+- 总时长估算
+- 图层默认属性（z_index 等）
+
+所有函数均为纯函数，不依赖 FFmpeg、数据库或外部 IO。
+"""
+
+from __future__ import annotations
+
+from typing import Any
+
+# ── 图层角色定义 ─────────────────────────────────────────────────────────────
+
+# 图层默认 z_index 映射
+LAYER_Z_INDEX: dict[str, int] = {
+    "background": -1,
+    "broll": 0,
+    "main": 0,
+    "overlay": 1,
+    "corner_voice": 1,
+    "audio": 2,
+}
+
+# 图层默认 PiP 缩放比例（相对于主画面）
+PIP_DEFAULT_SCALE = 0.25
+
+# 主视频图层角色（用于总时长计算、直通判断等）
+MAIN_LAYER_ROLES = frozenset({"main", "broll", "background"})
+
+
+# ── clip_type → layer_role 映射 ──────────────────────────────────────────────
+
+
+def resolve_layer_role(clip_type: str, config: dict[str, Any] | None = None) -> str:
+    """根据 clip_type 和 config.role 确定图层角色。
+
+    映射规则：
+      intro / outro         → "main"（按 order 排在首/尾）
+      overlay                → "overlay"（画中画叠加，z=1）
+      corner_voice           → "corner_voice"（右上角小窗，z=1）
+      background             → "background"（全屏底图，z=0）
+      b_roll                 → "broll"（z=0）
+      main + config.role=b_roll → "broll"
+      main + config.role=audio  → "audio"
+      main (default)         → "main"
+
+    Args:
+        clip_type: 片段类型字符串
+        config: 片段配置字典（可选）
+
+    Returns:
+        图层角色字符串
+    """
+    role = (config or {}).get("role", "") if config else ""
+
+    if clip_type in ("intro", "outro"):
+        return "main"
+    if clip_type == "overlay":
+        return "overlay"
+    if clip_type == "corner_voice":
+        return "corner_voice"
+    if clip_type == "background":
+        return "background"
+    if clip_type == "b_roll":
+        return "broll"
+    # main type
+    if role == "b_roll":
+        return "broll"
+    if role == "audio":
+        return "audio"
+    return "main"
+
+
+def get_layer_z_index(role: str) -> int:
+    """获取图层角色的默认 z_index。
+
+    Args:
+        role: 图层角色
+
+    Returns:
+        z_index 值，未知角色返回 0
+    """
+    return LAYER_Z_INDEX.get(role, 0)
+
+
+# ── clip 时长计算 ──────────────────────────────────────────────────────────
+
+
+def clip_effective_duration(
+    duration: float,
+    actual_duration: float = 0.0,
+) -> float:
+    """计算 clip 的有效时长（原速 trim 后时长）。
+
+    规则：
+    - duration > 0: min(duration, actual_duration)，actual=0 时用 duration
+    - duration <= 0: actual_duration，actual=0 时返回 0
+
+    Args:
+        duration: 配置的时长（0 表示使用完整素材）
+        actual_duration: 素材实际时长（probe 后的结果）
+
+    Returns:
+        有效时长（秒）
+    """
+    if duration > 0:
+        return min(duration, actual_duration) if actual_duration > 0 else duration
+    return actual_duration if actual_duration > 0 else 0.0
+
+
+def clip_playback_speed(playback_speed: Any) -> float:
+    """获取 clip 的播放速度，无效值回退到 1.0。
+
+    Args:
+        playback_speed: 播放速度（可为任意类型
+
+    Returns:
+        有效的播放速度（正数）
+    """
+    if not isinstance(playback_speed, (int, float)):
+        return 1.0
+    if playback_speed <= 0:
+        return 1.0
+    return float(playback_speed)
+
+
+def clip_adjusted_duration(
+    duration: float,
+    actual_duration: float = 0.0,
+    playback_speed: Any = 1.0,
+) -> float:
+    """计算调速后的 clip 实际时长（用于拼接计算）。
+
+    Args:
+        duration: 配置的时长
+        actual_duration: 素材实际时长
+        playback_speed: 播放速度
+
+    Returns:
+        调速后的时长
+    """
+    base = clip_effective_duration(duration, actual_duration)
+    speed = clip_playback_speed(playback_speed)
+    if abs(speed - 1.0) < 1e-6:
+        return base
+    return base / speed
+
+
+# ── 总时长估算 ────────────────────────────────────────────────────────────
+
+
+def estimate_total_duration(
+    layers: list[Any],
+    transition_duration: float = 0.0,
+) -> float:
+    """估算视频总时长。
+
+    取主图层（main/broll/background）的总调整后时长，减去转场重叠时间。
+
+    Args:
+        layers: 图层列表（每个元素需有 role 和 clips 属性，
+                clips 中元素需有 duration/actual_duration/playback_speed 属性）
+        transition_duration: 转场时长（秒），用于估算重叠时间
+
+    Returns:
+        估算的总时长（秒），最小 0.1
+    """
+    # 找主图层（第一个有视频内容的图层）
+    main_layer = None
+    for role in ("main", "broll", "background"):
+        for layer in layers:
+            if getattr(layer, "role", None) == role and getattr(layer, "clips", None):
+                main_layer = layer
+                break
+        if main_layer:
+            break
+
+    if not main_layer or not getattr(main_layer, "clips", None):
+        return 0.0
+
+    clips = getattr(main_layer, "clips", [])
+    total = sum(
+        clip_adjusted_duration(
+            duration=getattr(c, "duration", 0),
+            actual_duration=getattr(c, "actual_duration", 0.0),
+            playback_speed=getattr(c, "playback_speed", 1.0),
+        )
+        for c in clips
+    )
+
+    # 减去转场重叠时间（粗略估算）
+    n_clips = len(clips)
+    if n_clips > 1 and transition_duration > 0:
+        total -= (n_clips - 1) * transition_duration
+
+    return max(0.1, total)
+
+
+# ── 直通 / Stream Copy 判断辅助 ──────────────────────────────────────────────
+
+
+def can_pass_through(
+    layers: list[Any],
+    has_stickers: bool = False,
+    has_watermark: bool = False,
+) -> bool:
+    """判断是否可以走直通优化路径（单 clip 简单场景）。
+
+    条件：
+    1. 只有 1 个图层
+    2. 该图层是视频图层（main/broll/background）
+    3. 该图层只有 1 个 clip（无转场需求）
+    4. 没有贴纸
+    5. 没有水印
+
+    Args:
+        layers: 图层列表
+        has_stickers: 是否有贴纸
+        has_watermark: 是否有水印
+
+    Returns:
+        是否可以走直通
+    """
+    if len(layers) != 1:
+        return False
+    layer = layers[0]
+    role = getattr(layer, "role", "")
+    if role not in MAIN_LAYER_ROLES:
+        return False
+    clips = getattr(layer, "clips", [])
+    if len(clips) != 1:
+        return False
+    if has_stickers:
+        return False
+    if has_watermark:
+        return False
+    return True
@@ -0,0 +1,667 @@
+"""asset_quality_scoring 纯逻辑单测 — 第89波.
+
+测试评分纯函数，不依赖 FFmpeg/视频文件。
+覆盖：分辨率评分、帧率评分、码率评分、清晰度评分、稳定性评分、
+     质量总评分、9分类评分、分类结果计算。
+"""
+
+import numpy as np
+import pytest
+
+from apps.worker.worker_app.tasks.asset_quality_scoring import (
+    AudioAnalysis,
+    ClassificationResult,
+    ColorAnalysis,
+    MotionAnalysis,
+    QualityScore,
+    VideoInfo,
+    calculate_category_scores,
+    calculate_quality_score,
+    classify_from_analysis,
+    score_bitrate,
+    score_clarity,
+    score_framerate,
+    score_resolution,
+    score_stability,
+)
+from packages.domain.classification import AssetClassification
+
+# ── 分辨率评分 ────────────────────────────────────────────────────────
+
+
+class TestScoreResolution:
+    """分辨率评分边界测试."""
+
+    def test_4k_full_score(self):
+        """4K 及以上满分 25."""
+        assert score_resolution(3840, 2160) == 25.0
+        assert score_resolution(4096, 2160) == 25.0
+        assert score_resolution(7680, 4320) == 25.0  # 8K
+
+    def test_2k_score(self):
+        """2K 档 22 分."""
+        assert score_resolution(2560, 1440) == 22.0
+        assert score_resolution(3000, 1600) == 22.0
+        # 刚好低于 4K
+        assert score_resolution(3839, 2159) == 22.0
+
+    def test_1080p_score(self):
+        """1080p 档 20 分."""
+        assert score_resolution(1920, 1080) == 20.0
+        assert score_resolution(2000, 1080) == 20.0
+        # 刚好低于 2K
+        assert score_resolution(2559, 1439) == 20.0
+
+    def test_720p_score(self):
+        """720p 档 15 分."""
+        assert score_resolution(1280, 720) == 15.0
+        assert score_resolution(1280, 720) == 15.0
+        # 刚好低于 1080p
+        assert score_resolution(1919, 1079) == 15.0
+        # 1080x720 像素数 < 1280x720，掉到 480p 档
+        assert score_resolution(1080, 720) == 8.0
+
+    def test_480p_score(self):
+        """480p 档 8 分."""
+        assert score_resolution(854, 480) == 8.0
+        assert score_resolution(854, 480) == 8.0
+        # 刚好低于 720p
+        assert score_resolution(1279, 719) == 8.0
+        # 720x480 像素数 < 854x480，掉到最低档
+        assert score_resolution(720, 480) == 3.0
+
+    def test_low_resolution_score(self):
+        """低于 480p 给 3 分."""
+        assert score_resolution(640, 360) == 3.0
+        assert score_resolution(320, 240) == 3.0
+        assert score_resolution(0, 0) == 3.0
+
+    def test_non_standard_aspect_ratio(self):
+        """非标准宽高比按像素总数计算."""
+        # 竖屏 1080x1920 像素数 = 1080p
+        assert score_resolution(1080, 1920) == 20.0
+        # 超宽屏
+        assert score_resolution(2560, 1080) == 20.0  # 像素≈2.7M < 2K(3.6M)
+        # 1x1 极低分辨率
+        assert score_resolution(1, 1) == 3.0
+
+    def test_negative_values(self):
+        """负尺寸：负负得正按像素数算，一正一负 = 负数像素 = 最低档."""
+        # 一正一负 → 负像素总数 → < 480p → 3分
+        assert score_resolution(1920, -1080) == 3.0
+        assert score_resolution(-1920, 1080) == 3.0
+        # 都是 0 → 3分
+        assert score_resolution(0, 0) == 3.0
+
+
+# ── 帧率评分 ──────────────────────────────────────────────────────────
+
+
+class TestScoreFramerate:
+    """帧率评分边界测试."""
+
+    def test_60fps_full_score(self):
+        """60fps 及以上满分 20."""
+        assert score_framerate(60) == 20.0
+        assert score_framerate(120) == 20.0
+        assert score_framerate(240) == 20.0
+
+    def test_30fps_score(self):
+        """30-59fps 给 15 分."""
+        assert score_framerate(30) == 15.0
+        assert score_framerate(59.9) == 15.0
+        assert score_framerate(59) == 15.0
+
+    def test_24fps_score(self):
+        """24-29fps 给 10 分."""
+        assert score_framerate(24) == 10.0
+        assert score_framerate(29.97) == 10.0
+        assert score_framerate(25) == 10.0
+
+    def test_15fps_score(self):
+        """15-23fps 给 7 分."""
+        assert score_framerate(15) == 7.0
+        assert score_framerate(23.9) == 7.0
+        assert score_framerate(20) == 7.0
+
+    def test_low_fps_score(self):
+        """低于 15fps 给 5 分."""
+        assert score_framerate(10) == 5.0
+        assert score_framerate(1) == 5.0
+        assert score_framerate(0) == 5.0
+
+    def test_negative_fps(self):
+        """负帧率按最低档."""
+        assert score_framerate(-30) == 5.0
+
+    def test_float_fps(self):
+        """浮点帧率正确判断边界."""
+        # 29.97 (NTSC) < 30 → 24fps 档
+        assert score_framerate(29.97) == 10.0
+        assert score_framerate(23.976) == 7.0
+        assert score_framerate(59.94) == 15.0  # 59.94 < 60 → 30fps 档
+        assert score_framerate(30.0) == 15.0
+
+
+# ── 码率评分 ──────────────────────────────────────────────────────────
+
+
+class TestScoreBitrate:
+    """码率评分边界测试."""
+
+    def test_high_bitrate_full_score(self):
+        """10Mbps 以上满分 20."""
+        assert score_bitrate(10_000_001) == 20.0
+        assert score_bitrate(50_000_000) == 20.0
+        assert score_bitrate(100_000_000) == 20.0
+
+    def test_5mbps_score(self):
+        """5-10Mbps 给 15 分."""
+        assert score_bitrate(5_000_000) == 15.0
+        assert score_bitrate(8_000_000) == 15.0
+        assert score_bitrate(10_000_000) == 15.0  # 刚好 10Mbps = 不 > 10
+
+    def test_2mbps_score(self):
+        """2-5Mbps 给 10 分."""
+        assert score_bitrate(2_000_000) == 10.0
+        assert score_bitrate(3_000_000) == 10.0
+        assert score_bitrate(4_999_999) == 10.0
+
+    def test_05mbps_score(self):
+        """0.5-2Mbps 给 5 分."""
+        assert score_bitrate(500_000) == 5.0
+        assert score_bitrate(1_000_000) == 5.0
+        assert score_bitrate(1_999_999) == 5.0
+
+    def test_low_bitrate_score(self):
+        """低于 0.5Mbps 给 3 分."""
+        assert score_bitrate(499_999) == 3.0
+        assert score_bitrate(100_000) == 3.0
+        assert score_bitrate(0) == 3.0
+
+    def test_negative_bitrate(self):
+        """负码率按最低档."""
+        assert score_bitrate(-5_000_000) == 3.0
+
+    def test_zero_bitrate(self):
+        """0 码率 = 最低档."""
+        assert score_bitrate(0) == 3.0
+
+
+# ── 清晰度评分 ────────────────────────────────────────────────────────
+
+
+class TestScoreClarity:
+    """清晰度评分测试."""
+
+    def test_empty_frames_default_score(self):
+        """空帧列表给默认 10 分."""
+        assert score_clarity([]) == 10.0
+
+    def test_constant_image_low_clarity(self):
+        """纯色图像比高细节图像清晰度低很多."""
+        # 纯灰色图像
+        gray_frame = np.full((100, 100), 128, dtype=np.uint8)
+        score_constant = score_clarity([gray_frame])
+
+        # 高细节随机图像
+        detail_frame = np.random.randint(0, 256, (100, 100), dtype=np.uint8)
+        score_detail = score_clarity([detail_frame])
+
+        # 纯色图应该显著低于高细节图
+        assert score_constant < score_detail
+        assert 0.0 <= score_constant <= 20.0
+
+    def test_edge_rich_image_high_clarity(self):
+        """高频边缘图像有较高清晰度得分."""
+        # 棋盘格图案，边缘丰富
+        frame = np.zeros((100, 100), dtype=np.uint8)
+        for i in range(0, 100, 10):
+            for j in range(0, 100, 10):
+                if (i // 10 + j // 10) % 2 == 0:
+                    frame[i : i + 10, j : j + 10] = 255
+        score = score_clarity([frame])
+        assert score > 1.0  # 应有一定清晰度
+        assert 0.0 <= score <= 20.0
+
+    def test_rgb_frame_converts_to_gray(self):
+        """RGB 帧会被转灰度后计算."""
+        rgb_frame = np.random.randint(0, 256, (50, 50, 3), dtype=np.uint8)
+        score_rgb = score_clarity([rgb_frame])
+        # 对应灰度图
+        gray = np.dot(rgb_frame[..., :3], [0.299, 0.587, 0.114]).astype(np.uint8)
+        score_gray = score_clarity([gray])
+        # 两者应近似相等
+        assert abs(score_rgb - score_gray) < 0.01
+
+    def test_only_first_five_frames_analyzed(self):
+        """只分析前 5 帧."""
+        # 10 帧：前 5 帧纯色，后 5 帧高对比度
+        frames = []
+        for _ in range(5):
+            frames.append(np.full((50, 50), 128, dtype=np.uint8))
+        for _ in range(5):
+            high_freq = np.random.randint(0, 256, (50, 50), dtype=np.uint8)
+            frames.append(high_freq)
+        score_10 = score_clarity(frames)
+        score_5 = score_clarity(frames[:5])
+        # 前 5 帧相同，得分应相同
+        assert abs(score_10 - score_5) < 0.01
+
+    def test_score_within_bounds(self):
+        """得分始终在 0-20 范围内."""
+        for _ in range(10):
+            frame = np.random.randint(0, 256, (30, 30, 3), dtype=np.uint8)
+            score = score_clarity([frame])
+            assert 0.0 <= score <= 20.0
+
+    def test_multiple_frames_averaged(self):
+        """多帧取平均方差."""
+        # 第 1 帧低细节，第 2 帧高细节
+        low_detail = np.full((50, 50), 100, dtype=np.uint8)
+        high_detail = np.random.randint(0, 256, (50, 50), dtype=np.uint8)
+
+        score_low = score_clarity([low_detail])
+        score_high = score_clarity([high_detail])
+        score_both = score_clarity([low_detail, high_detail])
+
+        # 混合得分应在两者之间
+        assert score_low <= score_both <= score_high
+
+
+# ── 稳定性评分 ────────────────────────────────────────────────────────
+
+
+class TestScoreStability:
+    """稳定性评分测试."""
+
+    def test_single_frame_default_score(self):
+        """不足 2 帧给默认 10 分."""
+        assert score_stability([]) == 10.0
+        assert score_stability([np.zeros((10, 10, 3), dtype=np.uint8)]) == 10.0
+
+    def test_identical_frames_max_stability(self):
+        """完全相同的帧 = 高稳定性."""
+        frame = np.random.randint(50, 200, (100, 100, 3), dtype=np.uint8)
+        score = score_stability([frame, frame.copy()])
+        assert score > 10.0  # 应该接近满分 15
+
+    def test_very_different_frames_low_stability(self):
+        """位移方差大的多帧序列 = 低稳定性."""
+        # 构造 4 帧：帧间位移差异大（有的帧相似、有的帧完全不同）
+        # 位移方差大 → 不稳定 → 低分
+        base = np.random.randint(100, 150, (100, 100, 3), dtype=np.uint8)
+        frames = [
+            base,  # 帧0
+            base,  # 帧1 (完全相同 → 位移0)
+            np.full_like(base, 255),  # 帧2 (纯白 → 位移大)
+            base,  # 帧3 (回到基准 → 位移又大)
+        ]
+        score = score_stability(frames)
+        # 位移差异大 → 方差大 → 稳定性低
+        assert score < 10.0
+        assert 0.0 <= score <= 15.0
+
+    def test_score_within_bounds(self):
+        """得分始终在 0-15 范围内."""
+        for _ in range(10):
+            f1 = np.random.randint(0, 256, (40, 40, 3), dtype=np.uint8)
+            f2 = np.random.randint(0, 256, (40, 40, 3), dtype=np.uint8)
+            score = score_stability([f1, f2])
+            assert 0.0 <= score <= 15.0
+
+    def test_gray_frames_also_work(self):
+        """灰度帧也能计算."""
+        f1 = np.random.randint(0, 256, (50, 50), dtype=np.uint8)
+        f2 = np.random.randint(0, 256, (50, 50), dtype=np.uint8)
+        score = score_stability([f1, f2])
+        assert 0.0 <= score <= 15.0
+
+    def test_multiple_frame_pairs(self):
+        """多对帧取方差."""
+        base = np.random.randint(100, 150, (60, 60, 3), dtype=np.uint8)
+        # 5 帧相似的
+        frames = []
+        for i in range(5):
+            f = base.copy()
+            # 轻微变化
+            f = np.clip(f.astype(int) + np.random.randint(-5, 6, f.shape), 0, 255).astype(np.uint8)
+            frames.append(f)
+        score = score_stability(frames)
+        assert score > 5.0  # 相似帧应该有一定稳定性
+
+
+# ── 质量总评分 ────────────────────────────────────────────────────────
+
+
+class TestCalculateQualityScore:
+    """质量综合评分测试."""
+
+    def test_perfect_video_near_100(self):
+        """完美参数的视频接近 100 分."""
+        info = VideoInfo(
+            width=3840,
+            height=2160,
+            fps=60,
+            bitrate=20_000_000,
+        )
+        # 用高细节帧提升清晰度分
+        frame = np.random.randint(0, 256, (100, 100, 3), dtype=np.uint8)
+        result = calculate_quality_score(info, [frame])
+        assert isinstance(result, QualityScore)
+        assert result.resolution_score == 25.0
+        assert result.fps_score == 20.0
+        assert result.bitrate_score == 20.0
+        assert 50.0 <= result.total <= 100.0
+
+    def test_low_quality_video(self):
+        """低质量视频得分低."""
+        info = VideoInfo(
+            width=320,
+            height=240,
+            fps=10,
+            bitrate=100_000,
+        )
+        result = calculate_quality_score(info, [])
+        assert isinstance(result, QualityScore)
+        assert result.resolution_score == 3.0
+        assert result.fps_score == 5.0
+        assert result.bitrate_score == 3.0
+        # 无帧时清晰度和稳定性各给 10 分默认
+        assert result.clarity_score == 10.0
+        assert result.stability_score == 10.0
+        assert result.total == 31.0  # 3+5+3+10+10
+
+    def test_no_frames_uses_defaults(self):
+        """不传 frames 时清晰度/稳定性给默认分."""
+        info = VideoInfo(width=1920, height=1080, fps=30, bitrate=5_000_000)
+        result = calculate_quality_score(info)
+        assert result.clarity_score == 10.0
+        assert result.stability_score == 10.0
+        assert result.resolution_score == 20.0
+        assert result.fps_score == 15.0
+        assert result.bitrate_score == 15.0
+        assert result.total == 70.0
+
+    def test_total_capped_at_100(self):
+        """总分不超过 100."""
+        info = VideoInfo(
+            width=7680,
+            height=4320,
+            fps=240,
+            bitrate=100_000_000,
+        )
+        # 即使所有维度都满，总分不超 100
+        result = calculate_quality_score(info, [])
+        assert result.total <= 100.0
+
+    def test_total_minimum_zero(self):
+        """总分不低于 0."""
+        info = VideoInfo(width=0, height=0, fps=0, bitrate=0)
+        result = calculate_quality_score(info, [])
+        assert result.total >= 0.0
+
+    def test_total_is_rounded(self):
+        """总分保留 1 位小数."""
+        info = VideoInfo(width=1920, height=1080, fps=30, bitrate=5_000_000)
+        result = calculate_quality_score(info, [])
+        # 检查是 1 位小数
+        assert round(result.total, 1) == result.total
+
+
+# ── 分类评分 ──────────────────────────────────────────────────────────
+
+
+class TestCalculateCategoryScores:
+    """分类评分计算测试."""
+
+    def test_scenic_high_green_and_motion(self):
+        """绿色+适度运动+自然音 → 风景高分."""
+        color = ColorAnalysis(
+            green_ratio=0.5,
+            avg_saturation=0.5,
+            avg_brightness=0.6,
+            warm_ratio=0.2,
+            cool_ratio=0.3,
+        )
+        motion = MotionAnalysis(motion_score=0.3, scene_changes=1)
+        audio = AudioAnalysis(has_audio=True, ambient_ratio=0.7)
+        scores = calculate_category_scores(color, motion, audio)
+        assert scores[AssetClassification.SCENIC.value] > 0.5
+        assert scores[AssetClassification.SCENIC.value] <= 1.0
+
+    def test_product_low_motion_cool_tone(self):
+        """低运动+冷色调 → 产品高分."""
+        color = ColorAnalysis(
+            avg_brightness=0.5,
+            avg_saturation=0.3,
+            cool_ratio=0.5,
+            green_ratio=0.1,
+            warm_ratio=0.2,
+        )
+        motion = MotionAnalysis(motion_score=0.1, scene_changes=0)
+        audio = AudioAnalysis(has_audio=False)
+        scores = calculate_category_scores(color, motion, audio)
+        assert scores[AssetClassification.PRODUCT.value] > 0.5
+
+    def test_person_with_speech(self):
+        """有人声+适度运动 → 人物高分."""
+        color = ColorAnalysis(avg_brightness=0.5)
+        motion = MotionAnalysis(motion_score=0.25, scene_changes=1)
+        audio = AudioAnalysis(has_audio=True, speech_ratio=0.6)
+        scores = calculate_category_scores(color, motion, audio)
+        assert scores[AssetClassification.PERSON.value] > 0.5
+
+    def test_animal_high_motion(self):
+        """高运动+多场景切换 → 动物高分."""
+        color = ColorAnalysis()
+        motion = MotionAnalysis(motion_score=0.6, scene_changes=5)
+        audio = AudioAnalysis(has_audio=True, ambient_ratio=0.5)
+        scores = calculate_category_scores(color, motion, audio)
+        assert scores[AssetClassification.ANIMAL.value] > 0.5
+
+    def test_food_warm_saturated(self):
+        """暖色调+高饱和 → 美食高分."""
+        color = ColorAnalysis(
+            warm_ratio=0.6,
+            avg_saturation=0.7,
+            avg_brightness=0.6,
+            green_ratio=0.1,
+            cool_ratio=0.2,
+        )
+        motion = MotionAnalysis(motion_score=0.1, scene_changes=0)
+        audio = AudioAnalysis(has_audio=False)
+        scores = calculate_category_scores(color, motion, audio)
+        assert scores[AssetClassification.FOOD.value] > 0.5
+
+    def test_tech_cool_low_saturation(self):
+        """冷色调+低饱和+低运动 → 科技高分."""
+        color = ColorAnalysis(
+            cool_ratio=0.6,
+            avg_saturation=0.3,
+            avg_brightness=0.5,
+            green_ratio=0.1,
+            warm_ratio=0.2,
+        )
+        motion = MotionAnalysis(motion_score=0.1, scene_changes=0)
+        audio = AudioAnalysis(has_audio=False)
+        scores = calculate_category_scores(color, motion, audio)
+        assert scores[AssetClassification.TECH.value] > 0.5
+
+    def test_sport_high_motion(self):
+        """高运动+多场景 → 运动高分."""
+        color = ColorAnalysis(avg_brightness=0.6)
+        motion = MotionAnalysis(motion_score=0.7, scene_changes=5)
+        audio = AudioAnalysis(has_audio=False)
+        scores = calculate_category_scores(color, motion, audio)
+        assert scores[AssetClassification.SPORT.value] > 0.5
+
+    def test_music_high_music_ratio(self):
+        """高音乐比例 → 音乐高分."""
+        color = ColorAnalysis(avg_saturation=0.6, green_ratio=0.1)
+        motion = MotionAnalysis(motion_score=0.2)
+        audio = AudioAnalysis(has_audio=True, music_ratio=0.7)
+        scores = calculate_category_scores(color, motion, audio)
+        assert scores[AssetClassification.MUSIC.value] > 0.5
+
+    def test_other_has_base_score(self):
+        """其他分类有基础分 0.1."""
+        color = ColorAnalysis()
+        motion = MotionAnalysis()
+        audio = AudioAnalysis()
+        scores = calculate_category_scores(color, motion, audio)
+        assert scores[AssetClassification.OTHER.value] == 0.1
+
+    def test_all_scores_within_bounds(self):
+        """所有分类得分都在 0-1 范围内."""
+        color = ColorAnalysis(
+            green_ratio=0.9,
+            warm_ratio=0.9,
+            cool_ratio=0.9,
+            avg_saturation=0.99,
+            avg_brightness=0.99,
+        )
+        motion = MotionAnalysis(motion_score=0.9, scene_changes=100)
+        audio = AudioAnalysis(
+            has_audio=True,
+            speech_ratio=0.99,
+            music_ratio=0.99,
+            ambient_ratio=0.99,
+        )
+        scores = calculate_category_scores(color, motion, audio)
+        for cat, score in scores.items():
+            assert 0.0 <= score <= 1.0, f"{cat} score {score} out of bounds"
+
+    def test_all_nine_categories_present(self):
+        """返回 9 个分类的得分."""
+        color = ColorAnalysis()
+        motion = MotionAnalysis()
+        audio = AudioAnalysis()
+        scores = calculate_category_scores(color, motion, audio)
+        assert len(scores) == 9
+
+
+# ── 分类结果计算 ──────────────────────────────────────────────────────
+
+
+class TestClassifyFromAnalysis:
+    """分类结果计算测试."""
+
+    def test_returns_classification_result(self):
+        """返回 ClassificationResult 对象."""
+        color = ColorAnalysis()
+        motion = MotionAnalysis()
+        audio = AudioAnalysis()
+        result = classify_from_analysis(color, motion, audio)
+        assert isinstance(result, ClassificationResult)
+        assert isinstance(result.category, AssetClassification)
+        assert isinstance(result.confidence, float)
+        assert isinstance(result.scores, dict)
+
+    def test_highest_score_wins(self):
+        """得分最高的分类胜出."""
+        # 构造明显偏向风景的特征
+        color = ColorAnalysis(
+            green_ratio=0.8,
+            avg_saturation=0.6,
+            avg_brightness=0.7,
+        )
+        motion = MotionAnalysis(motion_score=0.3)
+        audio = AudioAnalysis(has_audio=True, ambient_ratio=0.8)
+        result = classify_from_analysis(color, motion, audio)
+        assert result.category == AssetClassification.SCENIC
+
+    def test_confidence_within_bounds(self):
+        """置信度在 0.3-0.95 范围内."""
+        color = ColorAnalysis()
+        motion = MotionAnalysis()
+        audio = AudioAnalysis()
+        result = classify_from_analysis(color, motion, audio)
+        assert 0.3 <= result.confidence <= 0.95
+
+    def test_confidence_capped_at_095(self):
+        """极高得分也被限制在 0.95."""
+        color = ColorAnalysis(
+            green_ratio=0.9,
+            avg_saturation=0.9,
+            avg_brightness=0.9,
+            warm_ratio=0.9,
+        )
+        motion = MotionAnalysis(motion_score=0.9, scene_changes=10)
+        audio = AudioAnalysis(
+            has_audio=True,
+            speech_ratio=0.9,
+            music_ratio=0.9,
+            ambient_ratio=0.9,
+        )
+        result = classify_from_analysis(color, motion, audio)
+        assert result.confidence <= 0.95
+
+    def test_confidence_floored_at_03(self):
+        """极低得分也有 0.3 最低置信度."""
+        color = ColorAnalysis()
+        motion = MotionAnalysis()
+        audio = AudioAnalysis()
+        result = classify_from_analysis(color, motion, audio)
+        assert result.confidence >= 0.3
+
+    def test_scores_dict_included(self):
+        """结果中包含完整分数字典."""
+        color = ColorAnalysis()
+        motion = MotionAnalysis()
+        audio = AudioAnalysis()
+        result = classify_from_analysis(color, motion, audio)
+        assert len(result.scores) == 9
+        assert AssetClassification.OTHER.value in result.scores
+
+    def test_food_category_wins_on_warm_colors(self):
+        """暖色调+高饱和 → 美食分类胜出."""
+        color = ColorAnalysis(
+            warm_ratio=0.7,
+            avg_saturation=0.8,
+            avg_brightness=0.6,
+            green_ratio=0.05,
+        )
+        motion = MotionAnalysis(motion_score=0.05)
+        audio = AudioAnalysis(has_audio=False)
+        result = classify_from_analysis(color, motion, audio)
+        assert result.category == AssetClassification.FOOD
+
+
+# ── 数据类默认值 ──────────────────────────────────────────────────────
+
+
+class TestDataclassDefaults:
+    """数据类默认值测试."""
+
+    def test_video_info_defaults(self):
+        info = VideoInfo()
+        assert info.width == 0
+        assert info.height == 0
+        assert info.fps == 0.0
+        assert info.bitrate == 0
+        assert info.has_audio is False
+
+    def test_color_analysis_defaults(self):
+        color = ColorAnalysis()
+        assert color.green_ratio == 0.0
+        assert color.avg_brightness == 0.0
+        assert color.dominant_hue == 0.0
+
+    def test_motion_analysis_defaults(self):
+        motion = MotionAnalysis()
+        assert motion.motion_score == 0.0
+        assert motion.scene_changes == 0
+
+    def test_audio_analysis_defaults(self):
+        audio = AudioAnalysis()
+        assert audio.has_audio is False
+        assert audio.speech_ratio == 0.0
+        assert audio.music_ratio == 0.0
+
+    def test_quality_score_requires_total(self):
+        with pytest.raises(TypeError):
+            QualityScore()
+        qs = QualityScore(total=50.0)
+        assert qs.total == 50.0
+        assert qs.resolution_score == 0.0
@@ -0,0 +1,755 @@
+"""Deep unit tests for asset_scoring.py — multi-dimensional scoring + diverse selection.
+
+深度覆盖：
+- score_resolution: 10+ 边界情况
+- score_duration: 10+ 边界情况
+- score_bitrate: 10+ 边界情况
+- calculate_total_score: 加权验证
+- score_asset_detail: 完整评分流程
+- diverse_selection: 各种分桶场景
+- filter_candidates: 各种过滤条件
+- 数据类 + 常量
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import Optional
+
+import pytest
+
+from packages.domain.asset_scoring import (
+    AssetScoreDetail,
+    MEDIUM_BUCKET_MAX,
+    MIN_QUALITY_SCORE,
+    OPTIMAL_DURATION_MAX,
+    OPTIMAL_DURATION_MIN,
+    SHORT_BUCKET_MAX,
+    SmartSelectResult,
+    TARGET_HEIGHT,
+    TARGET_WIDTH,
+    WEIGHT_BITRATE,
+    WEIGHT_DURATION,
+    WEIGHT_QUALITY,
+    WEIGHT_RESOLUTION,
+    _bucket_by_duration,
+    calculate_total_score,
+    diverse_selection,
+    filter_candidates,
+    score_asset_detail,
+    score_bitrate,
+    score_duration,
+    score_resolution,
+)
+
+# ── 辅助：模拟 asset 对象 ────────────────────────────────────────────────────
+
+
+class MockStatus:
+    def __init__(self, value: str):
+        self.value = value
+
+
+@dataclass
+class MockAsset:
+    id: str = "asset_001"
+    status: Any = None
+    mime_type: str = "video/mp4"
+    quality_score: Optional[float] = None
+    width: Optional[int] = None
+    height: Optional[int] = None
+    duration: Optional[float] = None
+    file_size: int = 0
+
+    def __post_init__(self):
+        if self.status is None:
+            self.status = MockStatus("ready")
+
+
+# ── 常量测试 ────────────────────────────────────────────────────────────────
+
+
+class TestConstants:
+    def test_weights_sum_to_one(self):
+        total = WEIGHT_QUALITY + WEIGHT_RESOLUTION + WEIGHT_DURATION + WEIGHT_BITRATE
+        assert abs(total - 1.0) < 0.001
+
+    def test_target_resolution_1080p(self):
+        assert TARGET_WIDTH == 1920
+        assert TARGET_HEIGHT == 1080
+
+    def test_bucket_thresholds(self):
+        assert SHORT_BUCKET_MAX == 5.0
+        assert MEDIUM_BUCKET_MAX == 15.0
+        assert SHORT_BUCKET_MAX < MEDIUM_BUCKET_MAX
+
+    def test_optimal_duration_range(self):
+        assert OPTIMAL_DURATION_MIN == 3.0
+        assert OPTIMAL_DURATION_MAX == 30.0
+        assert OPTIMAL_DURATION_MIN < OPTIMAL_DURATION_MAX
+
+    def test_min_quality_score(self):
+        assert MIN_QUALITY_SCORE == 30.0
+
+
+# ── 数据类测试 ──────────────────────────────────────────────────────────────
+
+
+class TestDataClasses:
+    def test_asset_score_detail_defaults(self):
+        detail = AssetScoreDetail(
+            asset_id="a1",
+            total_score=0.8,
+            quality_score=0.7,
+            resolution_score=0.9,
+            duration_score=0.85,
+            bitrate_score=0.75,
+            duration=10.0,
+        )
+        assert detail.asset_id == "a1"
+        assert detail.total_score == 0.8
+        assert detail.duration == 10.0
+
+    def test_smart_select_result_defaults(self):
+        result = SmartSelectResult(
+            selected_ids=["a1", "a2"],
+            total_candidates=10,
+            filtered_out=3,
+            avg_score=0.75,
+        )
+        assert result.selected_ids == ["a1", "a2"]
+        assert result.details == []
+        assert result.total_candidates == 10
+
+    def test_smart_select_result_with_details(self):
+        detail = AssetScoreDetail(
+            asset_id="a1",
+            total_score=0.9,
+            quality_score=0.8,
+            resolution_score=0.95,
+            duration_score=0.9,
+            bitrate_score=0.85,
+            duration=5.0,
+        )
+        result = SmartSelectResult(
+            selected_ids=["a1"],
+            total_candidates=5,
+            filtered_out=0,
+            avg_score=0.9,
+            details=[detail],
+        )
+        assert len(result.details) == 1
+        assert result.details[0].asset_id == "a1"
+
+
+# ── score_resolution ────────────────────────────────────────────────────────
+
+
+class TestScoreResolution:
+    def test_exact_target_1080p(self):
+        score = score_resolution(1920, 1080)
+        assert score == 1.0
+
+    def test_4k_full_score(self):
+        score = score_resolution(3840, 2160)
+        assert score == 1.0
+
+    def test_higher_than_target_full_score(self):
+        score = score_resolution(2560, 1440)
+        assert score == 1.0
+
+    def test_720p_lower(self):
+        score = score_resolution(1280, 720)
+        # 720p 像素 = 921600, 1080p = 2073600
+        # ratio = 0.444, score = 0.3 + 0.7 * 0.444 = 0.611
+        assert 0.5 < score < 0.75
+
+    def test_480p_much_lower(self):
+        score = score_resolution(854, 480)
+        # 480p = 409,920 pixels, ratio = 0.197
+        # score = 0.3 + 0.7 * 0.197 = 0.438
+        assert 0.3 < score < 0.5
+
+    def test_none_width(self):
+        score = score_resolution(None, 1080)
+        assert score == 0.5
+
+    def test_none_height(self):
+        score = score_resolution(1920, None)
+        assert score == 0.5
+
+    def test_both_none(self):
+        score = score_resolution(None, None)
+        assert score == 0.5
+
+    def test_zero_width(self):
+        score = score_resolution(0, 1080)
+        assert score == 0.5
+
+    def test_zero_height(self):
+        score = score_resolution(1920, 0)
+        assert score == 0.5
+
+    def test_negative_width(self):
+        score = score_resolution(-100, 1080)
+        assert score == 0.5
+
+    def test_very_low_res_floor(self):
+        score = score_resolution(100, 100)
+        # 10000 pixels, ratio = 0.0048, score = 0.3 + 0.7*0.0048 = 0.303
+        # 但最低不低于 0.1
+        assert score >= 0.1
+        assert score < 0.5
+
+    def test_custom_target_resolution(self):
+        score = score_resolution(1280, 720, target_width=1280, target_height=720)
+        assert score == 1.0
+
+    def test_sd_resolution(self):
+        score = score_resolution(640, 480)
+        # VGA = 307,200, ratio = 0.148
+        assert score > 0.1
+
+
+# ── score_duration ──────────────────────────────────────────────────────────
+
+
+class TestScoreDuration:
+    def test_none_duration(self):
+        assert score_duration(None) == 0.5
+
+    def test_zero_duration(self):
+        assert score_duration(0.0) == 0.5
+
+    def test_negative_duration(self):
+        assert score_duration(-5.0) == 0.5
+
+    def test_optimal_lower_bound(self):
+        assert score_duration(OPTIMAL_DURATION_MIN) == 1.0
+
+    def test_optimal_upper_bound(self):
+        assert score_duration(OPTIMAL_DURATION_MAX) == 1.0
+
+    def test_optimal_middle(self):
+        assert score_duration(10.0) == 1.0
+
+    def test_below_optimal_short(self):
+        score = score_duration(1.5)
+        # ratio = 1.5/3 = 0.5, score = 0.3 + 0.7*0.5 = 0.65
+        assert score == pytest.approx(0.65, rel=1e-3)
+
+    def test_very_short_approaches_03(self):
+        score = score_duration(0.1)
+        # ratio = 0.1/3 = 0.033, score = 0.3 + 0.7*0.033 = 0.323
+        assert 0.3 < score < 0.4
+
+    def test_just_below_optimal(self):
+        score = score_duration(2.9)
+        assert score < 1.0
+        assert score > 0.9
+
+    def test_above_optimal_slightly(self):
+        score = score_duration(35.0)
+        # excess = 5, penalty = 5/10 * 0.1 = 0.05, score = 0.95
+        assert score == pytest.approx(0.95, rel=1e-3)
+
+    def test_above_optimal_moderate(self):
+        score = score_duration(60.0)
+        # excess = 30, penalty = 30/10 * 0.1 = 0.3, score = 0.7
+        assert score == pytest.approx(0.7, rel=1e-3)
+
+    def test_very_long_floor(self):
+        score = score_duration(1000.0)
+        # excess = 970, penalty = 970/10 * 0.1 = 9.7, capped at 0.8
+        # score = max(0.2, 1.0 - 0.8) = 0.2
+        assert score == 0.2
+
+    def test_1_second(self):
+        score = score_duration(1.0)
+        # ratio = 1/3 = 0.333, score = 0.3 + 0.7*0.333 = 0.533
+        assert score == pytest.approx(0.3 + 0.7 * (1.0 / 3.0), rel=1e-3)
+
+
+# ── score_bitrate ───────────────────────────────────────────────────────────
+
+
+class TestScoreBitrate:
+    def test_no_file_size(self):
+        assert score_bitrate(0, 10.0) == 0.5
+
+    def test_no_duration(self):
+        assert score_bitrate(1_000_000, None) == 0.5
+
+    def test_zero_duration(self):
+        assert score_bitrate(1_000_000, 0.0) == 0.5
+
+    def test_negative_duration(self):
+        assert score_bitrate(1_000_000, -5.0) == 0.5
+
+    def test_optimal_low_end(self):
+        # 2 Mbps for 10s = 2.5 MB
+        file_size = int(2_000_000 * 10 / 8)
+        score = score_bitrate(file_size, 10.0)
+        assert score == 1.0
+
+    def test_optimal_high_end(self):
+        # 8 Mbps for 10s = 10 MB
+        file_size = int(8_000_000 * 10 / 8)
+        score = score_bitrate(file_size, 10.0)
+        assert score == 1.0
+
+    def test_optimal_middle(self):
+        # 5 Mbps for 10s = 6.25 MB
+        file_size = int(5_000_000 * 10 / 8)
+        score = score_bitrate(file_size, 10.0)
+        assert score == 1.0
+
+    def test_low_bitrate(self):
+        # 1 Mbps for 10s = 1.25 MB
+        file_size = int(1_000_000 * 10 / 8)
+        score = score_bitrate(file_size, 10.0)
+        # ratio = 1/2 = 0.5, score = 0.3 + 0.7*0.5 = 0.65
+        assert score == pytest.approx(0.65, rel=1e-2)
+
+    def test_very_low_bitrate(self):
+        # 100 kbps for 10s = 125 KB
+        file_size = int(100_000 * 10 / 8)
+        score = score_bitrate(file_size, 10.0)
+        # ratio = 0.05, score = 0.3 + 0.7*0.05 = 0.335
+        assert 0.3 < score < 0.5
+
+    def test_high_bitrate_slightly(self):
+        # 10 Mbps (just above 8Mbps)
+        file_size = int(10_000_000 * 10 / 8)
+        score = score_bitrate(file_size, 10.0)
+        # excess ratio = 10/8 - 1 = 0.25, penalty = min(0.5, 0.25*0.2) = 0.05
+        # score = max(0.5, 1.0 - 0.05) = 0.95
+        assert score == pytest.approx(0.95, rel=1e-2)
+
+    def test_very_high_bitrate_floor(self):
+        # 100 Mbps
+        file_size = int(100_000_000 * 10 / 8)
+        score = score_bitrate(file_size, 10.0)
+        # excess ratio = 100/8 - 1 = 11.5, penalty = min(0.5, 11.5*0.2) = 0.5
+        # score = max(0.5, 1.0 - 0.5) = 0.5
+        assert score == 0.5
+
+    def test_1mbps_file_10s(self):
+        file_size = 1_000_000  # 1 MB
+        score = score_bitrate(file_size, 10.0)
+        # bitrate = 8*1M/10 = 0.8 Mbps
+        assert 0.3 < score < 0.7
+
+
+# ── calculate_total_score ───────────────────────────────────────────────────
+
+
+class TestCalculateTotalScore:
+    def test_perfect_score(self):
+        total = calculate_total_score(1.0, 1.0, 1.0, 1.0)
+        assert total == 1.0
+
+    def test_zero_score(self):
+        total = calculate_total_score(0.0, 0.0, 0.0, 0.0)
+        assert total == 0.0
+
+    def test_weighted_sum(self):
+        # 各维度不同分数
+        q, r, d, b = 0.8, 0.6, 0.9, 0.7
+        expected = WEIGHT_QUALITY * q + WEIGHT_RESOLUTION * r + WEIGHT_DURATION * d + WEIGHT_BITRATE * b
+        total = calculate_total_score(q, r, d, b)
+        assert total == pytest.approx(expected, rel=1e-4)
+
+    def test_quality_dominates(self):
+        # 质量分权重最高(0.5)，变化影响最大
+        base = calculate_total_score(0.5, 0.5, 0.5, 0.5)
+        quality_up = calculate_total_score(1.0, 0.5, 0.5, 0.5)
+        resolution_up = calculate_total_score(0.5, 1.0, 0.5, 0.5)
+        # 质量分变化带来的差异最大
+        assert (quality_up - base) > (resolution_up - base)
+
+    def test_rounded_to_4_decimals(self):
+        # 1/3 这样的无限小数应该被截断
+        total = calculate_total_score(1 / 3, 1 / 3, 1 / 3, 1 / 3)
+        assert len(str(total).split(".")[-1]) <= 4
+
+
+# ── score_asset_detail ──────────────────────────────────────────────────────
+
+
+class TestScoreAssetDetail:
+    def test_full_asset(self):
+        detail = score_asset_detail(
+            asset_id="test_001",
+            quality=80.0,
+            width=1920,
+            height=1080,
+            duration=10.0,
+            file_size=5_000_000,
+        )
+        assert detail.asset_id == "test_001"
+        assert detail.quality_score == pytest.approx(0.8, rel=1e-3)
+        assert detail.resolution_score == 1.0
+        assert detail.duration_score == 1.0
+        assert 0.0 < detail.total_score <= 1.0
+        assert detail.duration == 10.0
+
+    def test_no_quality_default_05(self):
+        detail = score_asset_detail(
+            asset_id="a1",
+            quality=None,
+            width=1920,
+            height=1080,
+            duration=10.0,
+            file_size=5_000_000,
+        )
+        assert detail.quality_score == 0.5
+
+    def test_quality_100_is_1_0(self):
+        detail = score_asset_detail(
+            asset_id="a1",
+            quality=100.0,
+            width=1920,
+            height=1080,
+            duration=10.0,
+            file_size=5_000_000,
+        )
+        assert detail.quality_score == 1.0
+
+    def test_quality_zero_is_zero(self):
+        detail = score_asset_detail(
+            asset_id="a1",
+            quality=0.0,
+            width=1920,
+            height=1080,
+            duration=10.0,
+            file_size=5_000_000,
+        )
+        assert detail.quality_score == 0.0
+
+    def test_all_unknown_medium_score(self):
+        detail = score_asset_detail(
+            asset_id="a1",
+            quality=None,
+            width=None,
+            height=None,
+            duration=None,
+            file_size=0,
+        )
+        # 全部未知：质量0.5，分辨率0.5，时长0.5，码率0.5
+        assert detail.total_score == pytest.approx(0.5, rel=1e-3)
+
+    def test_custom_target_resolution(self):
+        detail = score_asset_detail(
+            asset_id="a1",
+            quality=100.0,
+            width=1280,
+            height=720,
+            duration=10.0,
+            file_size=5_000_000,
+            target_width=1280,
+            target_height=720,
+        )
+        assert detail.resolution_score == 1.0
+
+    def test_scores_are_rounded(self):
+        detail = score_asset_detail(
+            asset_id="a1",
+            quality=33.3,
+            width=854,
+            height=480,
+            duration=1.5,
+            file_size=1_000_000,
+        )
+        # 所有分数字符串长度不超过 0.xxxx 格式
+        for attr in ["quality_score", "resolution_score", "duration_score", "bitrate_score", "total_score"]:
+            val = getattr(detail, attr)
+            assert isinstance(val, float)
+
+
+# ── _bucket_by_duration ─────────────────────────────────────────────────────
+
+
+class TestBucketByDuration:
+    def test_short_bucket(self):
+        d = AssetScoreDetail("a", 0.5, 0.5, 0.5, 0.5, 0.5, 3.0)
+        assert _bucket_by_duration(d) == "short"
+
+    def test_short_bucket_boundary(self):
+        d = AssetScoreDetail("a", 0.5, 0.5, 0.5, 0.5, 0.5, 4.9)
+        assert _bucket_by_duration(d) == "short"
+
+    def test_medium_bucket(self):
+        d = AssetScoreDetail("a", 0.5, 0.5, 0.5, 0.5, 0.5, 10.0)
+        assert _bucket_by_duration(d) == "medium"
+
+    def test_medium_lower_boundary(self):
+        d = AssetScoreDetail("a", 0.5, 0.5, 0.5, 0.5, 0.5, SHORT_BUCKET_MAX)
+        assert _bucket_by_duration(d) == "medium"
+
+    def test_medium_upper_boundary(self):
+        d = AssetScoreDetail("a", 0.5, 0.5, 0.5, 0.5, 0.5, 14.9)
+        assert _bucket_by_duration(d) == "medium"
+
+    def test_long_bucket(self):
+        d = AssetScoreDetail("a", 0.5, 0.5, 0.5, 0.5, 0.5, 20.0)
+        assert _bucket_by_duration(d) == "long"
+
+    def test_long_lower_boundary(self):
+        d = AssetScoreDetail("a", 0.5, 0.5, 0.5, 0.5, 0.5, MEDIUM_BUCKET_MAX)
+        assert _bucket_by_duration(d) == "long"
+
+    def test_none_duration(self):
+        d = AssetScoreDetail("a", 0.5, 0.5, 0.5, 0.5, 0.5, None)
+        assert _bucket_by_duration(d) == "unknown"
+
+
+# ── diverse_selection ───────────────────────────────────────────────────────
+
+
+def _make_scored(items: list[tuple[str, float, float]]) -> list[AssetScoreDetail]:
+    """构造评分列表: (asset_id, total_score, duration)"""
+    return [
+        AssetScoreDetail(
+            asset_id=aid,
+            total_score=score,
+            quality_score=score,
+            resolution_score=score,
+            duration_score=score,
+            bitrate_score=score,
+            duration=dur,
+        )
+        for aid, score, dur in items
+    ]
+
+
+class TestDiverseSelection:
+    def test_empty_input(self):
+        result = diverse_selection([], 5)
+        assert result == []
+
+    def test_zero_count(self):
+        scored = _make_scored([("a1", 0.9, 10.0)])
+        result = diverse_selection(scored, 0)
+        assert result == []
+
+    def test_negative_count(self):
+        scored = _make_scored([("a1", 0.9, 10.0)])
+        result = diverse_selection(scored, -1)
+        assert result == []
+
+    def test_fewer_than_count(self):
+        scored = _make_scored([("a1", 0.9, 10.0)])
+        result = diverse_selection(scored, 5)
+        assert len(result) == 1
+
+    def test_mixed_buckets_diversity(self):
+        # 3短 + 3中 + 3长，取6个
+        items = [
+            ("s1", 0.9, 2.0),
+            ("s2", 0.8, 3.0),
+            ("s3", 0.7, 4.0),
+            ("m1", 0.95, 8.0),
+            ("m2", 0.85, 10.0),
+            ("m3", 0.75, 12.0),
+            ("l1", 0.92, 20.0),
+            ("l2", 0.82, 25.0),
+            ("l3", 0.72, 30.0),
+        ]
+        scored = _make_scored(items)
+        scored.sort(key=lambda d: d.total_score, reverse=True)
+        result = diverse_selection(scored, 6)
+        assert len(result) == 6
+        # 每个桶至少1个（base_quota = max(1, 6//3) = 2）
+        ids = [r.asset_id for r in result]
+        short_count = sum(1 for r in result if r.duration and r.duration < SHORT_BUCKET_MAX)
+        medium_count = sum(1 for r in result if r.duration and SHORT_BUCKET_MAX <= r.duration < MEDIUM_BUCKET_MAX)
+        long_count = sum(1 for r in result if r.duration and r.duration >= MEDIUM_BUCKET_MAX)
+        assert short_count >= 1
+        assert medium_count >= 1
+        assert long_count >= 1
+
+    def test_all_short_fallback_to_global(self):
+        items = [("s1", 0.9, 2.0), ("s2", 0.8, 3.0), ("s3", 0.7, 4.0)]
+        scored = _make_scored(items)
+        result = diverse_selection(scored, 3)
+        assert len(result) == 3
+        # 都是短素材，只能取短的
+        assert all(r.duration and r.duration < SHORT_BUCKET_MAX for r in result)
+
+    def test_sorted_by_score_descending(self):
+        items = [("a1", 0.5, 10.0), ("a2", 0.9, 10.0), ("a3", 0.7, 10.0)]
+        scored = _make_scored(items)
+        scored.sort(key=lambda d: d.total_score, reverse=True)
+        result = diverse_selection(scored, 3)
+        assert len(result) == 3
+        assert result[0].total_score >= result[1].total_score >= result[2].total_score
+
+    def test_count_one_each_bucket(self):
+        # count=3, base_quota=max(1,1)=1，每桶1个共3个
+        items = [
+            ("s1", 0.9, 2.0),
+            ("m1", 0.95, 8.0),
+            ("l1", 0.92, 20.0),
+        ]
+        scored = _make_scored(items)
+        scored.sort(key=lambda d: d.total_score, reverse=True)
+        result = diverse_selection(scored, 3)
+        assert len(result) == 3
+        # 每桶1个
+        assert any(r.duration and r.duration < SHORT_BUCKET_MAX for r in result)
+        assert any(r.duration and SHORT_BUCKET_MAX <= r.duration < MEDIUM_BUCKET_MAX for r in result)
+        assert any(r.duration and r.duration >= MEDIUM_BUCKET_MAX for r in result)
+
+    def test_unknown_duration_used_last(self):
+        items = [
+            ("u1", 0.99, None),  # 分最高但未知
+            ("s1", 0.9, 2.0),
+            ("m1", 0.8, 10.0),
+            ("l1", 0.7, 20.0),
+        ]
+        scored = _make_scored(items)
+        scored.sort(key=lambda d: d.total_score, reverse=True)
+        result = diverse_selection(scored, 3)
+        # 前3个应该是三个已知桶各一个
+        ids = [r.asset_id for r in result]
+        # u1 不应该在前3（因为 unknown 桶最后才用）
+        assert "s1" in ids
+        assert "m1" in ids
+        assert "l1" in ids
+
+    def test_no_duplicates(self):
+        items = [("s1", 0.9, 2.0), ("s2", 0.8, 3.0)]
+        scored = _make_scored(items)
+        result = diverse_selection(scored, 5)
+        ids = [r.asset_id for r in result]
+        assert len(ids) == len(set(ids))
+
+    def test_many_more_than_count(self):
+        # 30个素材，取6个
+        items = []
+        for i in range(10):
+            items.append((f"s{i}", 0.9 - i * 0.05, 2.0 + i * 0.2))
+            items.append((f"m{i}", 0.9 - i * 0.03, 6.0 + i * 0.8))
+            items.append((f"l{i}", 0.9 - i * 0.04, 16.0 + i * 1.5))
+        scored = _make_scored(items)
+        scored.sort(key=lambda d: d.total_score, reverse=True)
+        result = diverse_selection(scored, 6)
+        assert len(result) == 6
+        # 有多样性
+        durations = [r.duration for r in result]
+        short = sum(1 for d in durations if d and d < SHORT_BUCKET_MAX)
+        medium = sum(1 for d in durations if d and SHORT_BUCKET_MAX <= d < MEDIUM_BUCKET_MAX)
+        long_ = sum(1 for d in durations if d and d >= MEDIUM_BUCKET_MAX)
+        assert short >= 1
+        assert medium >= 1
+        assert long_ >= 1
+
+
+# ── filter_candidates ───────────────────────────────────────────────────────
+
+
+class TestFilterCandidates:
+    def test_empty_list(self):
+        candidates, filtered = filter_candidates([])
+        assert candidates == []
+        assert filtered == 0
+
+    def test_ready_video_passes(self):
+        assets = [MockAsset(id="a1", status=MockStatus("ready"), mime_type="video/mp4")]
+        candidates, filtered = filter_candidates(assets)
+        assert len(candidates) == 1
+        assert filtered == 0
+
+    def test_non_ready_filtered(self):
+        assets = [
+            MockAsset(id="a1", status=MockStatus("processing"), mime_type="video/mp4"),
+            MockAsset(id="a2", status=MockStatus("ready"), mime_type="video/mp4"),
+        ]
+        candidates, filtered = filter_candidates(assets)
+        assert len(candidates) == 1
+        assert candidates[0].id == "a2"
+        assert filtered == 0  # 非ready不算filtered_out（filtered_out只算质量分过滤的）
+
+    def test_non_video_filtered(self):
+        assets = [
+            MockAsset(id="a1", mime_type="image/jpeg"),
+            MockAsset(id="a2", mime_type="video/mp4"),
+        ]
+        candidates, filtered = filter_candidates(assets)
+        assert len(candidates) == 1
+        assert candidates[0].id == "a2"
+
+    def test_low_quality_filtered(self):
+        assets = [
+            MockAsset(id="low", quality_score=20.0),
+            MockAsset(id="high", quality_score=80.0),
+        ]
+        candidates, filtered = filter_candidates(assets, min_quality_score=30.0)
+        assert len(candidates) == 1
+        assert candidates[0].id == "high"
+        assert filtered == 1
+
+    def test_quality_none_passes(self):
+        assets = [MockAsset(id="a1", quality_score=None)]
+        candidates, filtered = filter_candidates(assets)
+        assert len(candidates) == 1
+        assert filtered == 0
+
+    def test_quality_exact_min_passes(self):
+        assets = [MockAsset(id="a1", quality_score=30.0)]
+        candidates, filtered = filter_candidates(assets, min_quality_score=30.0)
+        assert len(candidates) == 1
+        assert filtered == 0
+
+    def test_string_status(self):
+        # status 是字符串不是 Enum
+        @dataclass
+        class StrAsset:
+            id: str = "a1"
+            status: str = "ready"
+            mime_type: str = "video/mp4"
+            quality_score: float = 80.0
+            width: int = 1920
+            height: int = 1080
+            duration: float = 10.0
+            file_size: int = 5_000_000
+
+        assets = [StrAsset()]
+        candidates, filtered = filter_candidates(assets)
+        assert len(candidates) == 1
+
+    def test_empty_mime_type(self):
+        assets = [MockAsset(id="a1", mime_type="")]
+        candidates, filtered = filter_candidates(assets)
+        assert len(candidates) == 0
+
+    def test_none_mime_type(self):
+        # mime_type 是 None
+        @dataclass
+        class NoneMimeAsset:
+            id: str = "a1"
+            status: Any = None
+            mime_type: str | None = None
+            quality_score: float = 80.0
+            width: int = 1920
+            height: int = 1080
+            duration: float = 10.0
+            file_size: int = 5_000_000
+
+            def __post_init__(self):
+                if self.status is None:
+                    self.status = MockStatus("ready")
+
+        assets = [NoneMimeAsset()]
+        candidates, filtered = filter_candidates(assets)
+        assert len(candidates) == 0
+
+    def test_custom_min_quality(self):
+        assets = [
+            MockAsset(id="low", quality_score=40.0),
+            MockAsset(id="high", quality_score=60.0),
+        ]
+        candidates, filtered = filter_candidates(assets, min_quality_score=50.0)
+        assert len(candidates) == 1
+        assert filtered == 1
@@ -1,6 +1,6 @@
-"""
-Duplication 查重记录领域模型单元测试
-"""
+"""Duplication 领域模型单元测试。"""
+
+from datetime import datetime, timezone

 import pytest

@@ -8,255 +8,281 @@ from packages.domain.duplication import DuplicateSegment, DuplicationRecord


 class TestDuplicateSegmentCreate:
-    """DuplicateSegment.create 测试"""
-
    def test_create_success(self):
        seg = DuplicateSegment.create(
            source_start=10.0,
            source_end=20.0,
-            matched_video_id="vid_123",
-            matched_video_name="测试视频",
+            matched_video_id="vid123",
+            matched_video_name="test.mp4",
            matched_start=5.0,
            matched_end=15.0,
            similarity=85.5,
        )
        assert seg.id is not None
-        assert len(seg.id) == 32
+        assert len(seg.id) == 32  # uuid4 hex
        assert seg.source_start == 10.0
        assert seg.source_end == 20.0
-        assert seg.matched_video_id == "vid_123"
-        assert seg.matched_video_name == "测试视频"
+        assert seg.matched_video_id == "vid123"
+        assert seg.matched_video_name == "test.mp4"
        assert seg.matched_start == 5.0
        assert seg.matched_end == 15.0
        assert seg.similarity == 85.5

-    def test_invalid_source_negative_start(self):
+    def test_create_negative_source_start(self):
        with pytest.raises(ValueError, match="invalid source segment range"):
            DuplicateSegment.create(
                source_start=-1.0,
                source_end=10.0,
-                matched_video_id="v1",
-                matched_video_name="n1",
-                matched_start=0,
-                matched_end=10,
-                similarity=50,
+                matched_video_id="v",
+                matched_video_name="n",
+                matched_start=0.0,
+                matched_end=5.0,
+                similarity=50.0,
            )

-    def test_invalid_source_end_before_start(self):
-        with pytest.raises(ValueError, match="invalid source segment range"):
-            DuplicateSegment.create(
-                source_start=20.0,
-                source_end=10.0,
-                matched_video_id="v1",
-                matched_video_name="n1",
-                matched_start=0,
-                matched_end=10,
-                similarity=50,
-            )
-
-    def test_invalid_source_end_equals_start(self):
+    def test_create_source_end_equals_start(self):
        with pytest.raises(ValueError, match="invalid source segment range"):
            DuplicateSegment.create(
                source_start=10.0,
                source_end=10.0,
-                matched_video_id="v1",
-                matched_video_name="n1",
-                matched_start=0,
-                matched_end=10,
-                similarity=50,
+                matched_video_id="v",
+                matched_video_name="n",
+                matched_start=0.0,
+                matched_end=5.0,
+                similarity=50.0,
            )

-    def test_invalid_matched_negative_start(self):
+    def test_create_source_end_less_than_start(self):
+        with pytest.raises(ValueError, match="invalid source segment range"):
+            DuplicateSegment.create(
+                source_start=20.0,
+                source_end=10.0,
+                matched_video_id="v",
+                matched_video_name="n",
+                matched_start=0.0,
+                matched_end=5.0,
+                similarity=50.0,
+            )
+
+    def test_create_negative_matched_start(self):
        with pytest.raises(ValueError, match="invalid matched segment range"):
            DuplicateSegment.create(
-                source_start=0,
-                source_end=10,
-                matched_video_id="v1",
-                matched_video_name="n1",
-                matched_start=-5,
-                matched_end=10,
-                similarity=50,
+                source_start=0.0,
+                source_end=10.0,
+                matched_video_id="v",
+                matched_video_name="n",
+                matched_start=-1.0,
+                matched_end=5.0,
+                similarity=50.0,
            )

-    def test_invalid_matched_end_before_start(self):
+    def test_create_matched_end_equals_start(self):
        with pytest.raises(ValueError, match="invalid matched segment range"):
            DuplicateSegment.create(
-                source_start=0,
-                source_end=10,
-                matched_video_id="v1",
-                matched_video_name="n1",
-                matched_start=15,
-                matched_end=10,
-                similarity=50,
+                source_start=0.0,
+                source_end=10.0,
+                matched_video_id="v",
+                matched_video_name="n",
+                matched_start=5.0,
+                matched_end=5.0,
+                similarity=50.0,
            )

-    def test_invalid_similarity_negative(self):
+    def test_create_similarity_negative(self):
        with pytest.raises(ValueError, match="similarity must be between 0 and 100"):
            DuplicateSegment.create(
-                source_start=0,
-                source_end=10,
-                matched_video_id="v1",
-                matched_video_name="n1",
-                matched_start=0,
-                matched_end=10,
-                similarity=-1,
+                source_start=0.0,
+                source_end=10.0,
+                matched_video_id="v",
+                matched_video_name="n",
+                matched_start=0.0,
+                matched_end=5.0,
+                similarity=-1.0,
            )

-    def test_invalid_similarity_over_100(self):
+    def test_create_similarity_over_100(self):
        with pytest.raises(ValueError, match="similarity must be between 0 and 100"):
            DuplicateSegment.create(
-                source_start=0,
-                source_end=10,
-                matched_video_id="v1",
-                matched_video_name="n1",
-                matched_start=0,
-                matched_end=10,
-                similarity=101,
+                source_start=0.0,
+                source_end=10.0,
+                matched_video_id="v",
+                matched_video_name="n",
+                matched_start=0.0,
+                matched_end=5.0,
+                similarity=101.0,
            )

-    def test_similarity_boundary_zero(self):
+    def test_create_similarity_zero(self):
        seg = DuplicateSegment.create(
-            source_start=0,
-            source_end=10,
-            matched_video_id="v1",
-            matched_video_name="n1",
-            matched_start=0,
-            matched_end=10,
-            similarity=0,
+            source_start=0.0,
+            source_end=5.0,
+            matched_video_id="v",
+            matched_video_name="n",
+            matched_start=0.0,
+            matched_end=5.0,
+            similarity=0.0,
        )
-        assert seg.similarity == 0
+        assert seg.similarity == 0.0

-    def test_similarity_boundary_100(self):
+    def test_create_similarity_100(self):
        seg = DuplicateSegment.create(
-            source_start=0,
-            source_end=10,
-            matched_video_id="v1",
-            matched_video_name="n1",
-            matched_start=0,
-            matched_end=10,
-            similarity=100,
+            source_start=0.0,
+            source_end=5.0,
+            matched_video_id="v",
+            matched_video_name="n",
+            matched_start=0.0,
+            matched_end=5.0,
+            similarity=100.0,
        )
-        assert seg.similarity == 100
+        assert seg.similarity == 100.0
+
+    def test_create_unique_ids(self):
+        seg1 = DuplicateSegment.create(0, 1, "v", "n", 0, 1, 50.0)
+        seg2 = DuplicateSegment.create(0, 1, "v", "n", 0, 1, 50.0)
+        assert seg1.id != seg2.id


 class TestDuplicationRecordCreate:
-    """DuplicationRecord.create 测试"""
-
-    def test_create_minimal(self):
+    def test_create_success_defaults(self):
        record = DuplicationRecord.create(
            user_id="user123",
-            filename="test.mp4",
+            filename="my_video.mp4",
            file_size=1024000,
-            storage_key="oss://bucket/test.mp4",
+            storage_key="videos/vid1.mp4",
        )
        assert record.id is not None
        assert len(record.id) == 32
        assert record.user_id == "user123"
-        assert record.filename == "test.mp4"
+        assert record.filename == "my_video.mp4"
        assert record.file_size == 1024000
-        assert record.storage_key == "oss://bucket/test.mp4"
+        assert record.storage_key == "videos/vid1.mp4"
+        assert record.duration_seconds == 0.0
        assert record.status == "pending"
        assert record.duplicate_rate is None
        assert record.duplicate_count == 0
        assert record.segments == []
-        assert record.duration_seconds == 0.0
-        assert record.created_at is not None
-        assert record.updated_at is not None
+        assert record.error_message == ""
+        assert isinstance(record.created_at, datetime)
+        assert isinstance(record.updated_at, datetime)

    def test_create_with_duration(self):
        record = DuplicationRecord.create(
            user_id="u1",
-            filename="video.mp4",
-            file_size=5000,
-            storage_key="key",
+            filename="v.mp4",
+            file_size=100,
+            storage_key="k",
            duration_seconds=120.5,
        )
        assert record.duration_seconds == 120.5

-    def test_create_strips_whitespace(self):
+    def test_create_strips_user_id(self):
        record = DuplicationRecord.create(
-            user_id="  user456  ",
-            filename="  my video.mp4  ",
+            user_id="  user_trimmed  ",
+            filename="v.mp4",
            file_size=100,
-            storage_key="key",
+            storage_key="k",
        )
-        assert record.user_id == "user456"
-        assert record.filename == "my video.mp4"
+        assert record.user_id == "user_trimmed"

-    def test_empty_user_id_raises(self):
+    def test_create_strips_filename(self):
+        record = DuplicationRecord.create(
+            user_id="u1",
+            filename="  trimmed.mp4  ",
+            file_size=100,
+            storage_key="k",
+        )
+        assert record.filename == "trimmed.mp4"
+
+    def test_create_empty_user_id(self):
+        with pytest.raises(ValueError, match="user_id cannot be empty"):
+            DuplicationRecord.create(
+                user_id="",
+                filename="v.mp4",
+                file_size=100,
+                storage_key="k",
+            )
+
+    def test_create_whitespace_user_id(self):
        with pytest.raises(ValueError, match="user_id cannot be empty"):
            DuplicationRecord.create(
                user_id="   ",
-                filename="test.mp4",
+                filename="v.mp4",
                file_size=100,
-                storage_key="key",
+                storage_key="k",
            )

-    def test_empty_filename_raises(self):
+    def test_create_empty_filename(self):
        with pytest.raises(ValueError, match="filename cannot be empty"):
            DuplicationRecord.create(
                user_id="u1",
-                filename="  ",
+                filename="",
                file_size=100,
-                storage_key="key",
+                storage_key="k",
            )

-    def test_zero_file_size_raises(self):
+    def test_create_whitespace_filename(self):
+        with pytest.raises(ValueError, match="filename cannot be empty"):
+            DuplicationRecord.create(
+                user_id="u1",
+                filename="  \t  ",
+                file_size=100,
+                storage_key="k",
+            )
+
+    def test_create_zero_file_size(self):
        with pytest.raises(ValueError, match="file_size must be positive"):
            DuplicationRecord.create(
                user_id="u1",
-                filename="test.mp4",
+                filename="v.mp4",
                file_size=0,
-                storage_key="key",
+                storage_key="k",
            )

-    def test_negative_file_size_raises(self):
+    def test_create_negative_file_size(self):
        with pytest.raises(ValueError, match="file_size must be positive"):
            DuplicationRecord.create(
                user_id="u1",
-                filename="test.mp4",
+                filename="v.mp4",
                file_size=-100,
-                storage_key="key",
+                storage_key="k",
            )

+    def test_create_unique_ids(self):
+        r1 = DuplicationRecord.create("u", "f", 100, "k")
+        r2 = DuplicationRecord.create("u", "f", 100, "k")
+        assert r1.id != r2.id

-class TestDuplicationRecordLifecycle:
-    """生命周期状态转换测试"""

-    def test_mark_processing(self):
-        record = DuplicationRecord.create(user_id="u1", filename="t.mp4", file_size=100, storage_key="k")
-        old_updated = record.updated_at
+class TestMarkProcessing:
+    def test_mark_processing_from_pending(self):
+        record = DuplicationRecord.create("u", "f", 100, "k")
+        before = record.updated_at
        record.mark_processing()
        assert record.status == "processing"
-        assert record.updated_at >= old_updated
+        assert record.updated_at >= before

-    def test_mark_completed(self):
-        record = DuplicationRecord.create(user_id="u1", filename="t.mp4", file_size=100, storage_key="k")
+    def test_mark_processing_updates_timestamp(self):
+        record = DuplicationRecord.create("u", "f", 100, "k")
+        old_time = record.updated_at
+        # 确保时间戳会变（datetime.now 精度问题，直接赋值模拟）
        record.mark_processing()
-        segments = [
-            DuplicateSegment.create(
-                source_start=0,
-                source_end=10,
-                matched_video_id="v1",
-                matched_video_name="n1",
-                matched_start=0,
-                matched_end=10,
-                similarity=90,
-            )
-        ]
-        record.mark_completed(
-            duplicate_rate=25.5,
-            duplicate_count=1,
-            segments=segments,
-        )
+        assert record.status == "processing"
+        assert record.updated_at.tzinfo == timezone.utc
+
+
+class TestMarkCompleted:
+    def test_mark_completed_success(self):
+        record = DuplicationRecord.create("u", "f", 100, "k")
+        seg = DuplicateSegment.create(0, 5, "v", "n", 0, 5, 80.0)
+        record.mark_completed(duplicate_rate=75.5, duplicate_count=3, segments=[seg])
        assert record.status == "completed"
-        assert record.duplicate_rate == 25.5
-        assert record.duplicate_count == 1
+        assert record.duplicate_rate == 75.5
+        assert record.duplicate_count == 3
        assert len(record.segments) == 1
-        assert record.error_message == ""
+        assert record.segments[0].matched_video_id == "v"

    def test_mark_completed_zero_rate(self):
-        record = DuplicationRecord.create(user_id="u1", filename="t.mp4", file_size=100, storage_key="k")
+        record = DuplicationRecord.create("u", "f", 100, "k")
        record.mark_completed(duplicate_rate=0.0, duplicate_count=0, segments=[])
        assert record.status == "completed"
        assert record.duplicate_rate == 0.0
@@ -264,88 +290,133 @@ class TestDuplicationRecordLifecycle:
        assert record.segments == []

    def test_mark_completed_100_rate(self):
-        record = DuplicationRecord.create(user_id="u1", filename="t.mp4", file_size=100, storage_key="k")
+        record = DuplicationRecord.create("u", "f", 100, "k")
        record.mark_completed(duplicate_rate=100.0, duplicate_count=5, segments=[])
        assert record.duplicate_rate == 100.0

-    def test_mark_completed_invalid_rate_negative(self):
-        record = DuplicationRecord.create(user_id="u1", filename="t.mp4", file_size=100, storage_key="k")
+    def test_mark_completed_negative_rate(self):
+        record = DuplicationRecord.create("u", "f", 100, "k")
        with pytest.raises(ValueError, match="duplicate_rate must be between 0 and 100"):
-            record.mark_completed(duplicate_rate=-1, duplicate_count=0, segments=[])
+            record.mark_completed(duplicate_rate=-1.0, duplicate_count=0, segments=[])

-    def test_mark_completed_invalid_rate_over_100(self):
-        record = DuplicationRecord.create(user_id="u1", filename="t.mp4", file_size=100, storage_key="k")
+    def test_mark_completed_over_100_rate(self):
+        record = DuplicationRecord.create("u", "f", 100, "k")
        with pytest.raises(ValueError, match="duplicate_rate must be between 0 and 100"):
-            record.mark_completed(duplicate_rate=101, duplicate_count=0, segments=[])
+            record.mark_completed(duplicate_rate=101.0, duplicate_count=0, segments=[])

+    def test_mark_completed_updates_timestamp(self):
+        record = DuplicationRecord.create("u", "f", 100, "k")
+        record.mark_completed(50.0, 1, [])
+        assert record.updated_at.tzinfo == timezone.utc
+
+
+class TestMarkFailed:
    def test_mark_failed(self):
-        record = DuplicationRecord.create(user_id="u1", filename="t.mp4", file_size=100, storage_key="k")
+        record = DuplicationRecord.create("u", "f", 100, "k")
+        record.mark_failed("network timeout")
+        assert record.status == "failed"
+        assert record.error_message == "network timeout"
+
+    def test_mark_failed_empty_message(self):
+        record = DuplicationRecord.create("u", "f", 100, "k")
+        record.mark_failed("")
+        assert record.status == "failed"
+        assert record.error_message == ""
+
+    def test_mark_failed_from_processing(self):
+        record = DuplicationRecord.create("u", "f", 100, "k")
        record.mark_processing()
-        record.mark_failed("网络超时")
+        record.mark_failed("something went wrong")
        assert record.status == "failed"
-        assert record.error_message == "网络超时"
-        assert record.duplicate_rate is None
-
-    def test_mark_failed_from_pending(self):
-        record = DuplicationRecord.create(user_id="u1", filename="t.mp4", file_size=100, storage_key="k")
-        record.mark_failed("文件损坏")
-        assert record.status == "failed"
-        assert record.error_message == "文件损坏"
+        assert record.error_message == "something went wrong"


-class TestDuplicationRecordRetry:
-    """重试逻辑测试"""
-
+class TestCanRetry:
    def test_can_retry_failed(self):
-        record = DuplicationRecord.create(user_id="u1", filename="t.mp4", file_size=100, storage_key="k")
+        record = DuplicationRecord.create("u", "f", 100, "k")
        record.mark_failed("error")
        assert record.can_retry() is True

    def test_cannot_retry_pending(self):
-        record = DuplicationRecord.create(user_id="u1", filename="t.mp4", file_size=100, storage_key="k")
+        record = DuplicationRecord.create("u", "f", 100, "k")
        assert record.can_retry() is False

    def test_cannot_retry_processing(self):
-        record = DuplicationRecord.create(user_id="u1", filename="t.mp4", file_size=100, storage_key="k")
+        record = DuplicationRecord.create("u", "f", 100, "k")
        record.mark_processing()
        assert record.can_retry() is False

    def test_cannot_retry_completed(self):
-        record = DuplicationRecord.create(user_id="u1", filename="t.mp4", file_size=100, storage_key="k")
-        record.mark_completed(duplicate_rate=10, duplicate_count=1, segments=[])
+        record = DuplicationRecord.create("u", "f", 100, "k")
+        record.mark_completed(50.0, 1, [])
        assert record.can_retry() is False

-    def test_reset_for_retry(self):
-        record = DuplicationRecord.create(user_id="u1", filename="t.mp4", file_size=100, storage_key="k")
-        record.mark_processing()
-        segments = [
-            DuplicateSegment.create(
-                source_start=0,
-                source_end=5,
-                matched_video_id="v1",
-                matched_video_name="n1",
-                matched_start=0,
-                matched_end=5,
-                similarity=80,
-            )
-        ]
-        record.mark_completed(duplicate_rate=30, duplicate_count=1, segments=segments)
+
+class TestResetForRetry:
+    def test_reset_from_failed(self):
+        record = DuplicationRecord.create("u", "f", 100, "k")
+        seg = DuplicateSegment.create(0, 5, "v", "n", 0, 5, 80.0)
+        record.mark_completed(80.0, 2, [seg])
+        record.mark_failed("error")  # 模拟先完成再失败的场景不成立，直接从 failed 重置
+        # 直接设置到 failed 状态
        record.status = "failed"
-        record.error_message = "something wrong"
+        record.duplicate_rate = 50.0
+        record.duplicate_count = 3
+        record.error_message = "old error"
        record.video_fingerprint = {"hash": "abc"}

        record.reset_for_retry()
+
        assert record.status == "pending"
        assert record.duplicate_rate is None
        assert record.duplicate_count == 0
        assert record.error_message == ""
        assert record.segments == []
        assert record.video_fingerprint is None
-        assert record.updated_at is not None

-    def test_reset_for_retry_from_pending(self):
-        """即使从 pending 也能重置（调用方负责判断 can_retry）"""
-        record = DuplicationRecord.create(user_id="u1", filename="t.mp4", file_size=100, storage_key="k")
+    def test_reset_clears_segments(self):
+        record = DuplicationRecord.create("u", "f", 100, "k")
+        record.status = "failed"
+        record.segments = [
+            DuplicateSegment.create(0, 1, "v1", "n1", 0, 1, 50.0),
+            DuplicateSegment.create(2, 3, "v2", "n2", 0, 1, 60.0),
+        ]
        record.reset_for_retry()
-        assert record.status == "pending"
-        assert record.duplicate_count == 0
+        assert record.segments == []
+
+    def test_reset_preserves_identity(self):
+        record = DuplicationRecord.create("u", "f", 100, "k", duration_seconds=120.0)
+        record.status = "failed"
+        orig_id = record.id
+        orig_user = record.user_id
+        orig_filename = record.filename
+        orig_size = record.file_size
+        orig_storage = record.storage_key
+        orig_duration = record.duration_seconds
+
+        record.reset_for_retry()
+
+        assert record.id == orig_id
+        assert record.user_id == orig_user
+        assert record.filename == orig_filename
+        assert record.file_size == orig_size
+        assert record.storage_key == orig_storage
+        assert record.duration_seconds == orig_duration
+
+    def test_reset_updates_timestamp(self):
+        record = DuplicationRecord.create("u", "f", 100, "k")
+        record.status = "failed"
+        old_time = record.updated_at
+        record.reset_for_retry()
+        assert record.updated_at >= old_time
+
+
+class TestDataclassSlots:
+    def test_duplicate_segment_slots(self):
+        seg = DuplicateSegment.create(0, 1, "v", "n", 0, 1, 50.0)
+        # slots=True 时没有 __dict__
+        assert not hasattr(seg, "__dict__") or hasattr(seg, "__slots__")
+
+    def test_duplication_record_slots(self):
+        record = DuplicationRecord.create("u", "f", 100, "k")
+        assert hasattr(record, "__slots__") or hasattr(record, "__dict__")
@@ -0,0 +1,648 @@
+"""Unit tests for generation_plan_builder.py — pure logic utilities.
+
+覆盖:
+- VirtualPlan / VirtualClip 数据类
+- extract_intro_outro_from_clip_configs
+- apply_template_clip_effects
+- build_clips_by_mode (4种模式)
+- build_error_info
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import Any
+from unittest.mock import patch
+
+import pytest
+
+from worker_app.tasks.generation_plan_builder import (
+    VirtualClip,
+    VirtualPlan,
+    apply_template_clip_effects,
+    build_clips_by_mode,
+    build_error_info,
+    extract_intro_outro_from_clip_configs,
+)
+
+# ── 辅助：模拟 clip_config 对象 ──────────────────────────────────────────────
+
+
+@dataclass
+class MockClipType:
+    """模拟 Enum 类型的 clip_type。"""
+
+    value: str
+
+
+@dataclass
+class MockTransition:
+    """模拟 Enum 类型的 transition_effect。"""
+
+    value: str
+
+
+@dataclass
+class MockClipConfig:
+    """模拟 TemplateClipConfig 对象。"""
+
+    clip_type: Any
+    transition_effect: Any = "cut"
+    default_duration: float = 3.0
+    text_template: str = ""
+    config: dict | None = None
+
+
+def _make_config(
+    clip_type: str = "main",
+    transition: str = "cut",
+    duration: float = 3.0,
+    text: str = "",
+    config: dict | None = None,
+    use_enum: bool = True,
+) -> MockClipConfig:
+    """创建一个模拟 clip_config。"""
+    ct = MockClipType(clip_type) if use_enum else clip_type
+    tr = MockTransition(transition) if use_enum and transition != "cut" else transition
+    return MockClipConfig(
+        clip_type=ct,
+        transition_effect=tr,
+        default_duration=duration,
+        text_template=text,
+        config=config or {},
+    )
+
+
+# ── VirtualPlan / VirtualClip ───────────────────────────────────────────────
+
+
+class TestVirtualPlan:
+    def test_default_values(self):
+        plan = VirtualPlan(id="plan_001")
+        assert plan.id == "plan_001"
+        assert plan.name == ""
+        assert plan.config == {}
+
+    def test_full_init(self):
+        plan = VirtualPlan(id="p1", name="My Plan", config={"key": "value"})
+        assert plan.id == "p1"
+        assert plan.name == "My Plan"
+        assert plan.config == {"key": "value"}
+
+    def test_mutable_config(self):
+        plan = VirtualPlan(id="p1")
+        plan.config["new_key"] = "new_val"
+        assert plan.config == {"new_key": "new_val"}
+
+
+class TestVirtualClip:
+    def test_default_values(self):
+        clip = VirtualClip(id="c001")
+        assert clip.id == "c001"
+        assert clip.plan_id == ""
+        assert clip.clip_type == "main"
+        assert clip.order == 0
+        assert clip.asset_id == ""
+        assert clip.duration == 0.0
+        assert clip.transition_effect == "cut"
+        assert clip.transition_duration == 0.0
+        assert clip.playback_speed == 1.0
+        assert clip.status == "ready"
+        assert clip.config == {}
+
+    def test_full_init(self):
+        clip = VirtualClip(
+            id="c001",
+            plan_id="p1",
+            clip_type="overlay",
+            order=1,
+            asset_id="asset_001",
+            duration=5.5,
+            transition_effect="fade",
+            transition_duration=0.5,
+            playback_speed=1.5,
+            config={"role": "b_roll"},
+        )
+        assert clip.clip_type == "overlay"
+        assert clip.duration == 5.5
+        assert clip.playback_speed == 1.5
+
+    def test_mutable_config(self):
+        clip = VirtualClip(id="c001")
+        clip.config["effect"] = "vintage"
+        assert clip.config == {"effect": "vintage"}
+
+
+# ── extract_intro_outro_from_clip_configs ───────────────────────────────────
+
+
+class TestExtractIntroOutro:
+    def test_empty_configs(self):
+        result = extract_intro_outro_from_clip_configs([])
+        assert result == {}
+
+    def test_no_intro_no_outro(self):
+        configs = [_make_config("main"), _make_config("showcase")]
+        result = extract_intro_outro_from_clip_configs(configs)
+        assert result == {}
+
+    def test_intro_only_basic(self):
+        configs = [_make_config("intro", text="Hello", duration=2.5)]
+        result = extract_intro_outro_from_clip_configs(configs)
+        assert result["has_intro"] is True
+        assert result["intro_type"] == "text"
+        assert result["intro_duration"] == 2.5
+        assert result["intro_text"] == "Hello"
+        assert "has_outro" not in result
+
+    def test_outro_only_basic(self):
+        configs = [_make_config("outro", text="Bye", duration=3.0)]
+        result = extract_intro_outro_from_clip_configs(configs)
+        assert result["has_outro"] is True
+        assert result["outro_type"] == "text"
+        assert result["outro_duration"] == 3.0
+        assert result["outro_text"] == "Bye"
+
+    def test_both_intro_and_outro(self):
+        configs = [
+            _make_config("intro", text="Start"),
+            _make_config("main"),
+            _make_config("outro", text="End"),
+        ]
+        result = extract_intro_outro_from_clip_configs(configs)
+        assert result["has_intro"] is True
+        assert result["intro_text"] == "Start"
+        assert result["has_outro"] is True
+        assert result["outro_text"] == "End"
+
+    def test_intro_extra_config_pass_through(self):
+        configs = [
+            _make_config(
+                "intro",
+                config={
+                    "intro_text_color": "#ffffff",
+                    "intro_bg_color": "#000000",
+                    "intro_font_size": 32,
+                    "intro_video_url": "https://example.com/intro.mp4",
+                    "intro_video_path": "/tmp/intro.mp4",
+                    "random_key": "should_not_appear",
+                },
+            )
+        ]
+        result = extract_intro_outro_from_clip_configs(configs)
+        assert result["intro_text_color"] == "#ffffff"
+        assert result["intro_bg_color"] == "#000000"
+        assert result["intro_font_size"] == 32
+        assert result["intro_video_url"] == "https://example.com/intro.mp4"
+        assert "random_key" not in result
+
+    def test_outro_extra_config_pass_through(self):
+        configs = [
+            _make_config(
+                "outro",
+                config={
+                    "outro_text_color": "#ff0000",
+                    "outro_bg_color": "#00ff00",
+                    "outro_font_size": 24,
+                    "outro_follow_text": "关注我们",
+                },
+            )
+        ]
+        result = extract_intro_outro_from_clip_configs(configs)
+        assert result["outro_text_color"] == "#ff0000"
+        assert result["outro_follow_text"] == "关注我们"
+
+    def test_intro_type_from_config(self):
+        configs = [_make_config("intro", config={"intro_type": "video"})]
+        result = extract_intro_outro_from_clip_configs(configs)
+        assert result["intro_type"] == "video"
+
+    def test_default_duration_when_zero(self):
+        configs = [_make_config("intro", duration=0.0)]
+        result = extract_intro_outro_from_clip_configs(configs)
+        assert result["intro_duration"] == 3.0
+
+    def test_empty_text_not_included(self):
+        configs = [_make_config("intro", text="")]
+        result = extract_intro_outro_from_clip_configs(configs)
+        assert "intro_text" not in result
+
+    def test_with_string_clip_type_no_enum(self):
+        configs = [_make_config("intro", text="Hi", use_enum=False)]
+        result = extract_intro_outro_from_clip_configs(configs)
+        assert result["has_intro"] is True
+        assert result["intro_text"] == "Hi"
+
+    def test_first_intro_used_when_multiple(self):
+        configs = [
+            _make_config("intro", text="First"),
+            _make_config("intro", text="Second"),
+        ]
+        result = extract_intro_outro_from_clip_configs(configs)
+        assert result["intro_text"] == "First"
+
+    def test_none_config_handled(self):
+        cfg = _make_config("intro")
+        cfg.config = None
+        result = extract_intro_outro_from_clip_configs([cfg])
+        assert result["has_intro"] is True
+
+
+# ── apply_template_clip_effects ─────────────────────────────────────────────
+
+
+class TestApplyTemplateClipEffects:
+    def test_empty_clips(self):
+        clips: list[VirtualClip] = []
+        configs = [_make_config("main", transition="fade")]
+        apply_template_clip_effects(clips, configs, "one_take")
+        assert clips == []
+
+    def test_empty_configs(self):
+        clips = [VirtualClip(id="c1", clip_type="main")]
+        apply_template_clip_effects(clips, [], "one_take")
+        assert clips[0].transition_effect == "cut"
+
+    def test_no_main_configs(self):
+        clips = [VirtualClip(id="c1", clip_type="main")]
+        configs = [_make_config("intro"), _make_config("outro")]
+        apply_template_clip_effects(clips, configs, "one_take")
+        assert clips[0].transition_effect == "cut"
+
+    def test_transition_effect_applied(self):
+        clips = [VirtualClip(id="c1", clip_type="main")]
+        configs = [_make_config("main", transition="fade")]
+        apply_template_clip_effects(clips, configs, "one_take")
+        assert clips[0].transition_effect == "fade"
+
+    def test_transition_duration_applied(self):
+        clips = [VirtualClip(id="c1", clip_type="main")]
+        configs = [_make_config("main", transition="fade", config={"transition_duration": 0.8})]
+        apply_template_clip_effects(clips, configs, "one_take")
+        assert clips[0].transition_duration == 0.8
+
+    def test_transition_duration_invalid_ignored(self):
+        clips = [VirtualClip(id="c1", clip_type="main")]
+        configs = [_make_config("main", transition="fade", config={"transition_duration": "abc"})]
+        apply_template_clip_effects(clips, configs, "one_take")
+        assert clips[0].transition_duration == 0.0
+
+    def test_cut_transition_not_applied(self):
+        clips = [VirtualClip(id="c1", clip_type="main", transition_effect="dissolve")]
+        configs = [_make_config("main", transition="cut")]
+        apply_template_clip_effects(clips, configs, "one_take")
+        assert clips[0].transition_effect == "dissolve"  # 保留原值
+
+    def test_color_grade_applied(self):
+        clips = [VirtualClip(id="c1", clip_type="main")]
+        configs = [_make_config("main", config={"color_grade": "vintage"})]
+        apply_template_clip_effects(clips, configs, "one_take")
+        assert clips[0].config["color_grade"] == "vintage"
+
+    def test_multiple_effect_keys_applied(self):
+        clips = [VirtualClip(id="c1", clip_type="main")]
+        configs = [
+            _make_config(
+                "main",
+                config={
+                    "color_grade": "warm",
+                    "playback_speed": 1.5,
+                    "reverse": True,
+                    "chroma_key": {"color": "green"},
+                    "filter": "黑白",
+                },
+            )
+        ]
+        apply_template_clip_effects(clips, configs, "one_take")
+        assert clips[0].config["color_grade"] == "warm"
+        assert clips[0].config["playback_speed"] == 1.5
+        assert clips[0].config["reverse"] is True
+        assert clips[0].config["chroma_key"] == {"color": "green"}
+
+    def test_playback_speed_top_level_updated(self):
+        clips = [VirtualClip(id="c1", clip_type="main")]
+        configs = [_make_config("main", config={"playback_speed": 2.0})]
+        apply_template_clip_effects(clips, configs, "one_take")
+        assert clips[0].playback_speed == 2.0
+        assert clips[0].config["playback_speed"] == 2.0
+
+    def test_speed_fallback_sets_playback_speed(self):
+        clips = [VirtualClip(id="c1", clip_type="main")]
+        configs = [_make_config("main", config={"speed": 0.8})]
+        apply_template_clip_effects(clips, configs, "one_take")
+        assert clips[0].playback_speed == 0.8
+
+    def test_playback_speed_takes_priority_over_speed(self):
+        clips = [VirtualClip(id="c1", clip_type="main")]
+        configs = [_make_config("main", config={"speed": 0.5, "playback_speed": 2.0})]
+        apply_template_clip_effects(clips, configs, "one_take")
+        assert clips[0].playback_speed == 2.0
+
+    def test_existing_config_preserved(self):
+        clips = [VirtualClip(id="c1", clip_type="main", config={"role": "b_roll", "original": "value"})]
+        configs = [_make_config("main", config={"color_grade": "cool"})]
+        apply_template_clip_effects(clips, configs, "one_take")
+        assert clips[0].config["role"] == "b_roll"
+        assert clips[0].config["original"] == "value"
+        assert clips[0].config["color_grade"] == "cool"
+
+    def test_corner_voice_skipped(self):
+        clips = [
+            VirtualClip(id="c1", clip_type="main"),
+            VirtualClip(id="c2", clip_type="corner_voice"),
+        ]
+        configs = [_make_config("main", transition="fade")]
+        apply_template_clip_effects(clips, configs, "voice_pip")
+        assert clips[0].transition_effect == "fade"
+        assert clips[1].transition_effect == "cut"  # 不应用效果
+
+    def test_cyclic_matching_more_clips_than_configs(self):
+        clips = [
+            VirtualClip(id="c1", clip_type="main"),
+            VirtualClip(id="c2", clip_type="main"),
+            VirtualClip(id="c3", clip_type="main"),
+        ]
+        configs = [
+            _make_config("main", transition="fade"),
+            _make_config("main", transition="dissolve"),
+        ]
+        apply_template_clip_effects(clips, configs, "one_take")
+        assert clips[0].transition_effect == "fade"
+        assert clips[1].transition_effect == "dissolve"
+        assert clips[2].transition_effect == "dissolve"  # 循环用最后一个
+
+    def test_pip_mode_main_and_overlay_both_effected(self):
+        clips = [
+            VirtualClip(id="c1", clip_type="main"),
+            VirtualClip(id="c2", clip_type="overlay"),
+        ]
+        configs = [_make_config("main", transition="fade")]
+        apply_template_clip_effects(clips, configs, "pip")
+        assert clips[0].transition_effect == "fade"
+        assert clips[1].transition_effect == "fade"
+
+    def test_showcase_and_b_roll_count_as_template_source(self):
+        clips = [VirtualClip(id="c1", clip_type="main")]
+        configs = [
+            _make_config("showcase", transition="zoom_in"),
+            _make_config("b_roll", transition="slide_left"),
+        ]
+        apply_template_clip_effects(clips, configs, "one_take")
+        assert clips[0].transition_effect == "zoom_in"  # 用第一个匹配的
+
+    def test_string_transition_no_enum(self):
+        clips = [VirtualClip(id="c1", clip_type="main")]
+        cfg = _make_config("main")
+        cfg.transition_effect = "wipe"  # 直接字符串
+        apply_template_clip_effects(clips, [cfg], "one_take")
+        assert clips[0].transition_effect == "wipe"
+
+    def test_zero_speed_ignored(self):
+        clips = [VirtualClip(id="c1", clip_type="main")]
+        configs = [_make_config("main", config={"playback_speed": 0})]
+        apply_template_clip_effects(clips, configs, "one_take")
+        assert clips[0].playback_speed == 1.0  # 保持默认
+
+    def test_negative_speed_ignored(self):
+        clips = [VirtualClip(id="c1", clip_type="main")]
+        configs = [_make_config("main", config={"playback_speed": -1.0})]
+        apply_template_clip_effects(clips, configs, "one_take")
+        assert clips[0].playback_speed == 1.0
+
+
+# ── build_clips_by_mode ────────────────────────────────────────────────────
+
+
+class TestBuildClipsByMode:
+    def test_one_take_single_asset(self):
+        clips = build_clips_by_mode(
+            plan_id="p1",
+            asset_infos=[{"asset_id": "a1", "duration": 10.0}],
+            mode="one_take",
+        )
+        assert len(clips) == 1
+        assert clips[0].clip_type == "main"
+        assert clips[0].asset_id == "a1"
+        assert clips[0].duration == 10.0
+        assert clips[0].order == 0
+        assert clips[0].plan_id == "p1"
+
+    def test_one_take_multiple_assets(self):
+        clips = build_clips_by_mode(
+            plan_id="p1",
+            asset_infos=[
+                {"asset_id": "a1", "duration": 5.0},
+                {"asset_id": "a2", "duration": 10.0},
+                {"asset_id": "a3", "duration": 7.0},
+            ],
+            mode="one_take",
+        )
+        assert len(clips) == 3
+        assert all(c.clip_type == "main" for c in clips)
+        assert [c.order for c in clips] == [0, 1, 2]
+        assert [c.duration for c in clips] == [5.0, 10.0, 7.0]
+
+    def test_pip_mode_main_and_overlay(self):
+        clips = build_clips_by_mode(
+            plan_id="p1",
+            asset_infos=[
+                {"asset_id": "a1", "duration": 10.0},
+                {"asset_id": "a2", "duration": 5.0},
+                {"asset_id": "a3", "duration": 3.0},
+            ],
+            mode="pip",
+        )
+        assert len(clips) == 3
+        assert clips[0].clip_type == "main"
+        assert clips[1].clip_type == "overlay"
+        assert clips[2].clip_type == "overlay"
+
+    def test_pip_single_asset_is_main(self):
+        clips = build_clips_by_mode(
+            plan_id="p1",
+            asset_infos=[{"asset_id": "a1", "duration": 10.0}],
+            mode="pip",
+        )
+        assert len(clips) == 1
+        assert clips[0].clip_type == "main"
+
+    def test_voice_over_all_main_with_role(self):
+        clips = build_clips_by_mode(
+            plan_id="p1",
+            asset_infos=[
+                {"asset_id": "a1", "duration": 5.0},
+                {"asset_id": "a2", "duration": 8.0},
+            ],
+            mode="voice_over",
+        )
+        assert len(clips) == 2
+        assert all(c.clip_type == "main" for c in clips)
+        assert all(c.config.get("role") == "b_roll" for c in clips)
+
+    def test_voice_pip_three_layers(self):
+        clips = build_clips_by_mode(
+            plan_id="p1",
+            asset_infos=[
+                {"asset_id": "a1", "duration": 10.0},
+                {"asset_id": "a2", "duration": 5.0},
+                {"asset_id": "a3", "duration": 3.0},
+                {"asset_id": "a4", "duration": 4.0},
+            ],
+            mode="voice_pip",
+        )
+        assert len(clips) == 4
+        assert clips[0].clip_type == "background"
+        assert clips[1].clip_type == "corner_voice"
+        assert clips[2].clip_type == "b_roll"
+        assert clips[3].clip_type == "b_roll"
+
+    def test_voice_pip_two_assets(self):
+        clips = build_clips_by_mode(
+            plan_id="p1",
+            asset_infos=[
+                {"asset_id": "a1", "duration": 10.0},
+                {"asset_id": "a2", "duration": 5.0},
+            ],
+            mode="voice_pip",
+        )
+        assert len(clips) == 2
+        assert clips[0].clip_type == "background"
+        assert clips[1].clip_type == "corner_voice"
+
+    def test_voice_pip_single_asset(self):
+        clips = build_clips_by_mode(
+            plan_id="p1",
+            asset_infos=[{"asset_id": "a1", "duration": 10.0}],
+            mode="voice_pip",
+        )
+        assert len(clips) == 1
+        assert clips[0].clip_type == "background"
+
+    def test_default_mode_is_one_take(self):
+        clips = build_clips_by_mode(
+            plan_id="p1",
+            asset_infos=[{"asset_id": "a1", "duration": 10.0}],
+            mode="unknown_mode",
+        )
+        assert len(clips) == 1
+        assert clips[0].clip_type == "main"
+
+    def test_empty_assets(self):
+        clips = build_clips_by_mode(
+            plan_id="p1",
+            asset_infos=[],
+            mode="one_take",
+        )
+        assert clips == []
+
+    def test_default_asset_id_when_missing(self):
+        clips = build_clips_by_mode(
+            plan_id="p1",
+            asset_infos=[{"duration": 5.0}],
+            mode="one_take",
+        )
+        assert clips[0].asset_id == "asset_000"
+
+    def test_default_duration_when_missing(self):
+        clips = build_clips_by_mode(
+            plan_id="p1",
+            asset_infos=[{"asset_id": "a1"}],
+            mode="one_take",
+        )
+        assert clips[0].duration == 0.0
+
+    def test_clip_ids_sequential(self):
+        clips = build_clips_by_mode(
+            plan_id="p1",
+            asset_infos=[{"asset_id": f"a{i}"} for i in range(5)],
+            mode="one_take",
+        )
+        assert [c.id for c in clips] == ["vc_000", "vc_001", "vc_002", "vc_003", "vc_004"]
+
+    def test_plan_id_propagated(self):
+        clips = build_clips_by_mode(
+            plan_id="my_plan_123",
+            asset_infos=[{"asset_id": "a1"}, {"asset_id": "a2"}],
+            mode="pip",
+        )
+        assert all(c.plan_id == "my_plan_123" for c in clips)
+
+
+# ── build_error_info ───────────────────────────────────────────────────────
+
+
+class TestBuildErrorInfo:
+    def test_basic_structure(self):
+        try:
+            raise ValueError("test error")
+        except ValueError as e:
+            info = build_error_info(e, stage="render")
+
+        assert info["error_type"] == "ValueError"
+        assert info["message"] == "test error"
+        assert info["stage"] == "render"
+        assert "stack_trace" in info
+        assert "failed_at" in info
+        assert "ValueError" in info["stack_trace"]
+        assert "test error" in info["stack_trace"]
+
+    def test_default_stage(self):
+        try:
+            raise RuntimeError("oops")
+        except RuntimeError as e:
+            info = build_error_info(e)
+
+        assert info["stage"] == "render"
+
+    def test_custom_stage(self):
+        try:
+            raise TypeError("bad type")
+        except TypeError as e:
+            info = build_error_info(e, stage="download")
+
+        assert info["stage"] == "download"
+
+    def test_failed_at_is_iso_format(self):
+        try:
+            raise ValueError("x")
+        except ValueError as e:
+            info = build_error_info(e)
+
+        # ISO 格式检查：包含 T 和 +
+        assert "T" in info["failed_at"]
+
+    @patch("worker_app.tasks.generation_plan_builder.traceback.format_exc")
+    def test_long_stack_trace_truncated(self, mock_format):
+        # 构造30行堆栈
+        lines = [f'  File "file{i}.py", line {i}, in func{i}' for i in range(28)]
+        lines.append("ValueError: deep error")
+        mock_format.return_value = "\n".join(lines)
+
+        try:
+            raise ValueError("deep")
+        except ValueError as e:
+            info = build_error_info(e)
+
+        assert "truncated" in info["stack_trace"]
+        assert "total 29 lines" in info["stack_trace"]
+        # 确认只保留了前20行
+        assert "func19" in info["stack_trace"]
+        assert "func20" not in info["stack_trace"]
+
+    @patch("worker_app.tasks.generation_plan_builder.traceback.format_exc")
+    def test_short_stack_trace_not_truncated(self, mock_format):
+        # 构造5行堆栈（小于20）
+        mock_format.return_value = (
+            "Traceback (most recent call last):\n"
+            '  File "test.py", line 10, in foo\n'
+            '    raise ValueError("simple")\n'
+            "ValueError: simple\n"
+        )
+
+        try:
+            raise ValueError("simple")
+        except ValueError as e:
+            info = build_error_info(e)
+
+        assert "truncated" not in info["stack_trace"]
+        assert "ValueError: simple" in info["stack_trace"]
@@ -0,0 +1,670 @@
+"""plan_generator_utils 纯逻辑单测 — 第90波.
+
+测试素材分配、clip_type映射、默认clip生成、配置转clip等纯函数。
+不依赖 DB，使用领域对象直接构造。
+"""
+
+import pytest
+
+from packages.domain.edit_plan_clip import EditPlanClip
+from packages.domain.editing_mode import EditingMode
+from packages.domain.plan_generator_utils import (
+    DEFAULT_CLIP_DURATION,
+    create_clips_from_configs,
+    distribute_assets,
+    generate_default_clips,
+    map_clip_types_for_mode,
+)
+from packages.domain.template_clip_config import ClipType, TemplateClipConfig
+
+
+# ── 辅助函数 ──────────────────────────────────────────────────────────
+
+
+def _make_main_clip(plan_id: str = "plan1", order: int = 0) -> EditPlanClip:
+    """创建一个 MAIN 类型的 clip."""
+    return EditPlanClip.create(
+        plan_id=plan_id,
+        clip_type=ClipType.MAIN.value,
+        order=order,
+        duration=5.0,
+    )
+
+
+def _make_clips(n: int, clip_type: str = "main") -> list[EditPlanClip]:
+    """创建 n 个指定类型的 clip."""
+    return [
+        EditPlanClip.create(
+            plan_id="plan1",
+            clip_type=clip_type,
+            order=i,
+            duration=5.0,
+        )
+        for i in range(n)
+    ]
+
+
+def _collect_asset_ids(clips: list[EditPlanClip]) -> list[str]:
+    """按顺序收集 clips 的 asset_id（空的跳过）."""
+    return [c.asset_id for c in clips if c.asset_id]
+
+
+# ── distribute_assets: ONE_TAKE ─────────────────────────────────────
+
+
+class TestDistributeOneTake:
+    """ONE_TAKE 模式素材分配."""
+
+    def test_equal_count(self):
+        """素材数 == clip 数：一一对应."""
+        clips = _make_clips(3)
+        distribute_assets(clips, ["a1", "a2", "a3"], EditingMode.ONE_TAKE.value)
+        assert clips[0].asset_id == "a1"
+        assert clips[1].asset_id == "a2"
+        assert clips[2].asset_id == "a3"
+
+    def test_more_assets_than_clips(self):
+        """素材多于 clip：多余的不用."""
+        clips = _make_clips(2)
+        distribute_assets(clips, ["a1", "a2", "a3"], EditingMode.ONE_TAKE.value)
+        assert clips[0].asset_id == "a1"
+        assert clips[1].asset_id == "a2"
+
+    def test_fewer_assets_than_clips(self):
+        """素材少于 clip：后面的 clip 没素材."""
+        clips = _make_clips(5)
+        distribute_assets(clips, ["a1", "a2"], EditingMode.ONE_TAKE.value)
+        assert clips[0].asset_id == "a1"
+        assert clips[1].asset_id == "a2"
+        assert clips[2].asset_id == ""
+        assert clips[3].asset_id == ""
+        assert clips[4].asset_id == ""
+
+    def test_empty_assets(self):
+        """空素材列表：无分配."""
+        clips = _make_clips(3)
+        distribute_assets(clips, [], EditingMode.ONE_TAKE.value)
+        for c in clips:
+            assert c.asset_id == ""
+
+    def test_empty_clips(self):
+        """空 clip 列表：不报错."""
+        distribute_assets([], ["a1"], EditingMode.ONE_TAKE.value)
+
+    def test_only_main_clips_get_assigned(self):
+        """只分配给 MAIN 类型 clip，其他类型不受影响."""
+        clips = _make_clips(2) + _make_clips(2, "intro") + _make_clips(2, "outro")
+        distribute_assets(clips, ["a1", "a2", "a3", "a4"], EditingMode.ONE_TAKE.value)
+        mains = [c for c in clips if c.clip_type == "main"]
+        others = [c for c in clips if c.clip_type != "main"]
+        assert mains[0].asset_id == "a1"
+        assert mains[1].asset_id == "a2"
+        for c in others:
+            assert c.asset_id == ""
+
+
+# ── distribute_assets: PIP ──────────────────────────────────────────
+
+
+class TestDistributePip:
+    """PIP 模式素材分配."""
+
+    def test_basic_pip_distribution(self):
+        """第1个素材给 main，其余给 overlay."""
+        clips = _make_clips(1) + _make_clips(3, "overlay")
+        distribute_assets(clips, ["a1", "a2", "a3", "a4"], EditingMode.PIP.value)
+        mains = [c for c in clips if c.clip_type == "main"]
+        overlays = [c for c in clips if c.clip_type == "overlay"]
+        assert mains[0].asset_id == "a1"
+        assert overlays[0].asset_id == "a2"
+        assert overlays[1].asset_id == "a3"
+        assert overlays[2].asset_id == "a4"
+
+    def test_single_asset_only_main(self):
+        """只有1个素材：只分配给 main，overlay 没素材."""
+        clips = _make_clips(1) + _make_clips(2, "overlay")
+        distribute_assets(clips, ["a1"], EditingMode.PIP.value)
+        mains = [c for c in clips if c.clip_type == "main"]
+        overlays = [c for c in clips if c.clip_type == "overlay"]
+        assert mains[0].asset_id == "a1"
+        assert overlays[0].asset_id == ""
+        assert overlays[1].asset_id == ""
+
+    def test_more_overlays_than_assets(self):
+        """overlay 多于剩余素材：后面的 overlay 没素材."""
+        clips = _make_clips(1) + _make_clips(5, "overlay")
+        distribute_assets(clips, ["a1", "a2", "a3"], EditingMode.PIP.value)
+        overlays = [c for c in clips if c.clip_type == "overlay"]
+        assert overlays[0].asset_id == "a2"
+        assert overlays[1].asset_id == "a3"
+        assert overlays[2].asset_id == ""
+        assert overlays[3].asset_id == ""
+        assert overlays[4].asset_id == ""
+
+    def test_no_main_clip(self):
+        """没有 main clip：第1个素材没人拿，overlay 从第2个素材开始."""
+        clips = _make_clips(3, "overlay")
+        distribute_assets(clips, ["a1", "a2", "a3"], EditingMode.PIP.value)
+        overlays = [c for c in clips if c.clip_type == "overlay"]
+        # PIP 逻辑：先给 main 分配第1个素材（没有 main 则跳过），
+        # 剩余从第2个开始分配给 overlay
+        assert overlays[0].asset_id == "a2"
+        assert overlays[1].asset_id == "a3"
+        assert overlays[2].asset_id == ""
+
+
+# ── distribute_assets: VOICE_OVER ───────────────────────────────────
+
+
+class TestDistributeVoiceOver:
+    """VOICE_OVER 模式素材分配."""
+
+    def test_voice_over_same_as_one_take(self):
+        """VOICE_OVER 和 ONE_TAKE 分配策略相同：按顺序给 main."""
+        clips = _make_clips(3)
+        assets = ["a1", "a2", "a3"]
+        distribute_assets(clips, assets, EditingMode.VOICE_OVER.value)
+        assert clips[0].asset_id == "a1"
+        assert clips[1].asset_id == "a2"
+        assert clips[2].asset_id == "a3"
+
+    def test_voice_over_fewer_assets(self):
+        """素材不足时，后面的 main clip 没素材."""
+        clips = _make_clips(5)
+        distribute_assets(clips, ["a1"], EditingMode.VOICE_OVER.value)
+        assert clips[0].asset_id == "a1"
+        assert clips[1].asset_id == ""
+
+
+# ── distribute_assets: VOICE_PIP ────────────────────────────────────
+
+
+class TestDistributeVoicePip:
+    """VOICE_PIP 模式素材分配."""
+
+    def test_three_assets_full_distribution(self):
+        """3个素材：background + corner_voice + b_roll 各一个."""
+        clips = (
+            _make_clips(1, "background")
+            + _make_clips(1, "corner_voice")
+            + _make_clips(1, "b_roll")
+        )
+        distribute_assets(clips, ["a1", "a2", "a3"], EditingMode.VOICE_PIP.value)
+        bgs = [c for c in clips if c.clip_type == "background"]
+        voices = [c for c in clips if c.clip_type == "corner_voice"]
+        brolls = [c for c in clips if c.clip_type == "b_roll"]
+        assert bgs[0].asset_id == "a1"
+        assert voices[0].asset_id == "a2"
+        assert brolls[0].asset_id == "a3"
+
+    def test_single_asset_only_background(self):
+        """1个素材：只分配给 background."""
+        clips = (
+            _make_clips(1, "background")
+            + _make_clips(1, "corner_voice")
+            + _make_clips(2, "b_roll")
+        )
+        distribute_assets(clips, ["a1"], EditingMode.VOICE_PIP.value)
+        assert clips[0].asset_id == "a1"
+        assert clips[1].asset_id == ""
+        assert clips[2].asset_id == ""
+        assert clips[3].asset_id == ""
+
+    def test_two_assets_bg_and_voice(self):
+        """2个素材：background + corner_voice."""
+        clips = (
+            _make_clips(1, "background")
+            + _make_clips(1, "corner_voice")
+            + _make_clips(2, "b_roll")
+        )
+        distribute_assets(clips, ["a1", "a2"], EditingMode.VOICE_PIP.value)
+        bgs = [c for c in clips if c.clip_type == "background"]
+        voices = [c for c in clips if c.clip_type == "corner_voice"]
+        brolls = [c for c in clips if c.clip_type == "b_roll"]
+        assert bgs[0].asset_id == "a1"
+        assert voices[0].asset_id == "a2"
+        assert brolls[0].asset_id == ""
+
+    def test_many_broll_clips(self):
+        """多个 b_roll clip：按顺序分配剩余素材."""
+        clips = (
+            _make_clips(1, "background")
+            + _make_clips(1, "corner_voice")
+            + _make_clips(5, "b_roll")
+        )
+        distribute_assets(
+            clips,
+            ["a1", "a2", "a3", "a4", "a5"],
+            EditingMode.VOICE_PIP.value,
+        )
+        brolls = [c for c in clips if c.clip_type == "b_roll"]
+        assert brolls[0].asset_id == "a3"
+        assert brolls[1].asset_id == "a4"
+        assert brolls[2].asset_id == "a5"
+        assert brolls[3].asset_id == ""
+        assert brolls[4].asset_id == ""
+
+    def test_missing_some_layer_clips(self):
+        """缺少某些层的 clip 不影响其他层."""
+        # 没有 corner_voice，素材应该按顺序：bg 拿 a1，b_roll 从 a2 开始
+        clips = _make_clips(1, "background") + _make_clips(3, "b_roll")
+        distribute_assets(clips, ["a1", "a2", "a3"], EditingMode.VOICE_PIP.value)
+        bgs = [c for c in clips if c.clip_type == "background"]
+        brolls = [c for c in clips if c.clip_type == "b_roll"]
+        assert bgs[0].asset_id == "a1"
+        # 没有 corner_voice，b_roll 从第2个素材开始
+        assert brolls[0].asset_id == "a2"
+        assert brolls[1].asset_id == "a3"
+
+
+# ── distribute_assets: 边缘情况 ─────────────────────────────────────
+
+
+class TestDistributeEdgeCases:
+    """素材分配边缘情况."""
+
+    def test_unknown_mode_falls_back_to_one_take(self):
+        """未知模式退化为 ONE_TAKE."""
+        clips = _make_clips(3)
+        distribute_assets(clips, ["a1", "a2", "a3"], "unknown_mode")
+        assert clips[0].asset_id == "a1"
+        assert clips[1].asset_id == "a2"
+        assert clips[2].asset_id == "a3"
+
+    def test_both_empty(self):
+        """两边都空：不报错."""
+        distribute_assets([], [], EditingMode.ONE_TAKE.value)
+
+
+# ── map_clip_types_for_mode ─────────────────────────────────────────
+
+
+class TestMapClipTypesForMode:
+    """clip_type 按模式映射."""
+
+    def test_one_take_unchanged(self):
+        """ONE_TAKE 模式：main 保持 main."""
+        clips = _make_clips(5)
+        map_clip_types_for_mode(clips, EditingMode.ONE_TAKE.value)
+        for c in clips:
+            assert c.clip_type == "main"
+
+    def test_voice_over_unchanged(self):
+        """VOICE_OVER 模式：main 保持 main."""
+        clips = _make_clips(5)
+        map_clip_types_for_mode(clips, EditingMode.VOICE_OVER.value)
+        for c in clips:
+            assert c.clip_type == "main"
+
+    def test_pip_first_main_stays_rest_become_overlay(self):
+        """PIP 模式：第1个 main 保持，其余变 overlay."""
+        clips = _make_clips(5)
+        map_clip_types_for_mode(clips, EditingMode.PIP.value)
+        assert clips[0].clip_type == "main"
+        assert clips[1].clip_type == "overlay"
+        assert clips[2].clip_type == "overlay"
+        assert clips[3].clip_type == "overlay"
+        assert clips[4].clip_type == "overlay"
+
+    def test_pip_single_main_unchanged(self):
+        """PIP 模式只有1个 main：保持 main."""
+        clips = _make_clips(1)
+        map_clip_types_for_mode(clips, EditingMode.PIP.value)
+        assert clips[0].clip_type == "main"
+
+    def test_voice_pip_three_types(self):
+        """VOICE_PIP 模式：background + corner_voice + b_roll."""
+        clips = _make_clips(5)
+        map_clip_types_for_mode(clips, EditingMode.VOICE_PIP.value)
+        assert clips[0].clip_type == "background"
+        assert clips[1].clip_type == "corner_voice"
+        assert clips[2].clip_type == "b_roll"
+        assert clips[3].clip_type == "b_roll"
+        assert clips[4].clip_type == "b_roll"
+
+    def test_voice_pip_one_main(self):
+        """VOICE_PIP 只有1个 main：变成 background."""
+        clips = _make_clips(1)
+        map_clip_types_for_mode(clips, EditingMode.VOICE_PIP.value)
+        assert clips[0].clip_type == "background"
+
+    def test_voice_pip_two_mains(self):
+        """VOICE_PIP 2个 main：background + corner_voice."""
+        clips = _make_clips(2)
+        map_clip_types_for_mode(clips, EditingMode.VOICE_PIP.value)
+        assert clips[0].clip_type == "background"
+        assert clips[1].clip_type == "corner_voice"
+
+    def test_non_main_clips_unchanged(self):
+        """非 MAIN 类型 clip 不受影响."""
+        clips = (
+            _make_clips(1, "intro")
+            + _make_clips(3)  # main
+            + _make_clips(1, "outro")
+        )
+        map_clip_types_for_mode(clips, EditingMode.PIP.value)
+        assert clips[0].clip_type == "intro"
+        assert clips[1].clip_type == "main"  # 第1个 main
+        assert clips[2].clip_type == "overlay"  # 第2个 main → overlay
+        assert clips[3].clip_type == "overlay"  # 第3个 main → overlay
+        assert clips[4].clip_type == "outro"
+
+    def test_no_main_clips_noop(self):
+        """没有 main clip：什么都不做."""
+        clips = _make_clips(3, "intro")
+        original_types = [c.clip_type for c in clips]
+        map_clip_types_for_mode(clips, EditingMode.PIP.value)
+        assert [c.clip_type for c in clips] == original_types
+
+    def test_empty_clips_noop(self):
+        """空列表：不报错."""
+        map_clip_types_for_mode([], EditingMode.PIP.value)
+
+
+# ── generate_default_clips ──────────────────────────────────────────
+
+
+class TestGenerateDefaultClips:
+    """默认 clip 生成."""
+
+    def test_one_take_normal(self):
+        """ONE_TAKE：N 个 main clip."""
+        clips = generate_default_clips("plan1", EditingMode.ONE_TAKE.value, 5)
+        assert len(clips) == 5
+        for c in clips:
+            assert c.clip_type == "main"
+            assert c.plan_id == "plan1"
+            assert c.duration == DEFAULT_CLIP_DURATION
+        # order 递增
+        for i in range(5):
+            assert clips[i].order == i
+
+    def test_pip_structure(self):
+        """PIP：1个 main + (N-1)个 overlay."""
+        clips = generate_default_clips("plan1", EditingMode.PIP.value, 4)
+        assert len(clips) == 4
+        assert clips[0].clip_type == "main"
+        assert clips[1].clip_type == "overlay"
+        assert clips[2].clip_type == "overlay"
+        assert clips[3].clip_type == "overlay"
+        assert clips[0].order == 0
+        assert clips[3].order == 3
+
+    def test_pip_single_asset(self):
+        """PIP 只有1个素材：1个 main，没有 overlay."""
+        clips = generate_default_clips("plan1", EditingMode.PIP.value, 1)
+        assert len(clips) == 1
+        assert clips[0].clip_type == "main"
+
+    def test_voice_over_structure(self):
+        """VOICE_OVER：N 个 main clip，带 b_roll 标记."""
+        clips = generate_default_clips("plan1", EditingMode.VOICE_OVER.value, 3)
+        assert len(clips) == 3
+        for c in clips:
+            assert c.clip_type == "main"
+            assert c.config.get("role") == "b_roll"
+
+    def test_voice_pip_three_layers(self):
+        """VOICE_PIP：background + corner_voice + b_roll."""
+        clips = generate_default_clips("plan1", EditingMode.VOICE_PIP.value, 5)
+        assert len(clips) == 5
+        assert clips[0].clip_type == "background"
+        assert clips[1].clip_type == "corner_voice"
+        assert clips[2].clip_type == "b_roll"
+        assert clips[3].clip_type == "b_roll"
+        assert clips[4].clip_type == "b_roll"
+
+    def test_voice_pip_single_asset(self):
+        """VOICE_PIP 1个素材：只有 background."""
+        clips = generate_default_clips("plan1", EditingMode.VOICE_PIP.value, 1)
+        assert len(clips) == 1
+        assert clips[0].clip_type == "background"
+
+    def test_voice_pip_two_assets(self):
+        """VOICE_PIP 2个素材：background + corner_voice."""
+        clips = generate_default_clips("plan1", EditingMode.VOICE_PIP.value, 2)
+        assert len(clips) == 2
+        assert clips[0].clip_type == "background"
+        assert clips[1].clip_type == "corner_voice"
+
+    def test_zero_assets_at_least_one(self):
+        """0 个素材：至少生成 1 个 clip."""
+        for mode in [
+            EditingMode.ONE_TAKE.value,
+            EditingMode.PIP.value,
+            EditingMode.VOICE_OVER.value,
+            EditingMode.VOICE_PIP.value,
+        ]:
+            clips = generate_default_clips("plan1", mode, 0)
+            assert len(clips) >= 1
+
+    def test_unknown_mode_falls_back(self):
+        """未知模式退化为 ONE_TAKE 风格."""
+        clips = generate_default_clips("plan1", "unknown", 3)
+        assert len(clips) == 3
+        for c in clips:
+            assert c.clip_type == "main"
+
+    def test_order_is_sequential(self):
+        """所有模式下 order 都是从 0 开始连续递增."""
+        for mode in [
+            EditingMode.ONE_TAKE.value,
+            EditingMode.PIP.value,
+            EditingMode.VOICE_OVER.value,
+            EditingMode.VOICE_PIP.value,
+        ]:
+            clips = generate_default_clips("plan1", mode, 5)
+            for i, c in enumerate(clips):
+                assert c.order == i
+
+
+# ── create_clips_from_configs ───────────────────────────────────────
+
+
+class TestCreateClipsFromConfigs:
+    """从模板配置创建 clips."""
+
+    def test_basic_creation(self):
+        """基本创建：按 order 排序，属性正确传递."""
+        configs = [
+            TemplateClipConfig(
+                id="cfg1",
+                template_id="tpl1",
+                clip_type=ClipType.MAIN,
+                order=1,
+                min_duration=3.0,
+                max_duration=7.0,
+                transition_effect="fade",
+            ),
+            TemplateClipConfig(
+                id="cfg2",
+                template_id="tpl1",
+                clip_type=ClipType.INTRO,
+                order=0,
+                min_duration=2.0,
+                max_duration=4.0,
+                transition_effect="cut",
+            ),
+        ]
+        clips = create_clips_from_configs("plan1", configs)
+        assert len(clips) == 2
+        # 按 order 排序：intro(order=0) 在前，main(order=1) 在后
+        assert clips[0].clip_type == "intro"
+        assert clips[1].clip_type == "main"
+        assert clips[0].order == 0
+        assert clips[1].order == 1
+        assert clips[0].template_clip_config_id == "cfg2"
+        assert clips[1].template_clip_config_id == "cfg1"
+
+    def test_duration_average_of_min_max(self):
+        """min_duration 和 max_duration 都有时，取平均值."""
+        configs = [
+            TemplateClipConfig(
+                id="cfg1",
+                template_id="tpl1",
+                clip_type=ClipType.MAIN,
+                order=0,
+                min_duration=4.0,
+                max_duration=6.0,
+            ),
+        ]
+        clips = create_clips_from_configs("plan1", configs)
+        assert clips[0].duration == 5.0  # (4+6)/2
+
+    def test_duration_only_min(self):
+        """只有 min_duration 时，用 min_duration."""
+        configs = [
+            TemplateClipConfig(
+                id="cfg1",
+                template_id="tpl1",
+                clip_type=ClipType.MAIN,
+                order=0,
+                min_duration=3.5,
+                max_duration=0,
+            ),
+        ]
+        clips = create_clips_from_configs("plan1", configs)
+        assert clips[0].duration == 3.5
+
+    def test_duration_only_max(self):
+        """只有 max_duration 时，用 max_duration."""
+        configs = [
+            TemplateClipConfig(
+                id="cfg1",
+                template_id="tpl1",
+                clip_type=ClipType.MAIN,
+                order=0,
+                min_duration=0,
+                max_duration=8.0,
+            ),
+        ]
+        clips = create_clips_from_configs("plan1", configs)
+        assert clips[0].duration == 8.0
+
+    def test_duration_default_when_both_zero(self):
+        """都为 0 时用默认时长."""
+        configs = [
+            TemplateClipConfig(
+                id="cfg1",
+                template_id="tpl1",
+                clip_type=ClipType.MAIN,
+                order=0,
+                min_duration=0,
+                max_duration=0,
+            ),
+        ]
+        clips = create_clips_from_configs("plan1", configs)
+        assert clips[0].duration == DEFAULT_CLIP_DURATION
+
+    def test_empty_configs_returns_empty(self):
+        """空配置列表返回空列表."""
+        clips = create_clips_from_configs("plan1", [])
+        assert clips == []
+
+    def test_plan_id_passed_through(self):
+        """plan_id 正确传递给所有 clip."""
+        configs = [
+            TemplateClipConfig(
+                id=f"cfg{i}",
+                template_id="tpl1",
+                clip_type=ClipType.MAIN,
+                order=i,
+                min_duration=3,
+                max_duration=5,
+            )
+            for i in range(3)
+        ]
+        clips = create_clips_from_configs("my_plan", configs)
+        for c in clips:
+            assert c.plan_id == "my_plan"
+
+    def test_playback_speed_from_config(self):
+        """playback_speed 从 config.playback_speed 读取."""
+        configs = [
+            TemplateClipConfig(
+                id="cfg1",
+                template_id="tpl1",
+                clip_type=ClipType.MAIN,
+                order=0,
+                min_duration=3,
+                max_duration=5,
+                config={"playback_speed": 1.5},
+            ),
+        ]
+        clips = create_clips_from_configs("plan1", configs)
+        assert clips[0].playback_speed == 1.5
+
+    def test_playback_speed_fallback_to_speed_ratio(self):
+        """playback_speed 不存在时回退到 speed_ratio."""
+        configs = [
+            TemplateClipConfig(
+                id="cfg1",
+                template_id="tpl1",
+                clip_type=ClipType.MAIN,
+                order=0,
+                min_duration=3,
+                max_duration=5,
+                config={"speed_ratio": 0.8},
+            ),
+        ]
+        clips = create_clips_from_configs("plan1", configs)
+        assert clips[0].playback_speed == 0.8
+
+    def test_playback_speed_default_1(self):
+        """没有 speed 配置时默认为 1.0."""
+        configs = [
+            TemplateClipConfig(
+                id="cfg1",
+                template_id="tpl1",
+                clip_type=ClipType.MAIN,
+                order=0,
+                min_duration=3,
+                max_duration=5,
+                config={},
+            ),
+        ]
+        clips = create_clips_from_configs("plan1", configs)
+        assert clips[0].playback_speed == 1.0
+
+    def test_playback_speed_none_falls_back(self):
+        """playback_speed 为 None 时回退到 1.0."""
+        configs = [
+            TemplateClipConfig(
+                id="cfg1",
+                template_id="tpl1",
+                clip_type=ClipType.MAIN,
+                order=0,
+                min_duration=3,
+                max_duration=5,
+                config={"playback_speed": None},
+            ),
+        ]
+        clips = create_clips_from_configs("plan1", configs)
+        assert clips[0].playback_speed == 1.0
+
+    def test_transition_default_cut(self):
+        """transition_effect 为空时默认为 cut."""
+        configs = [
+            TemplateClipConfig(
+                id="cfg1",
+                template_id="tpl1",
+                clip_type=ClipType.MAIN,
+                order=0,
+                min_duration=3,
+                max_duration=5,
+                transition_effect=None,
+            ),
+        ]
+        clips = create_clips_from_configs("plan1", configs)
+        assert clips[0].transition_effect == "cut"
+
+
+# ── 常量导出 ────────────────────────────────────────────────────────
+
+
+class TestConstants:
+    """常量导出验证."""
+
+    def test_default_duration_value(self):
+        """默认片段时长应为 5 秒."""
+        assert DEFAULT_CLIP_DURATION == 5.0
@@ -0,0 +1,361 @@
+"""render_layer_utils 模块单元测试."""
+
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from typing import Any
+
+import pytest
+
+from packages.domain.render_layer_utils import (
+    LAYER_Z_INDEX,
+    MAIN_LAYER_ROLES,
+    PIP_DEFAULT_SCALE,
+    can_pass_through,
+    clip_adjusted_duration,
+    clip_effective_duration,
+    clip_playback_speed,
+    estimate_total_duration,
+    get_layer_z_index,
+    resolve_layer_role,
+)
+
+# ── 辅助数据类 ──────────────────────────────────────────────────────────────
+
+
+@dataclass
+class FakeClip:
+    duration: float = 0.0
+    actual_duration: float = 0.0
+    playback_speed: Any = 1.0
+
+
+@dataclass
+class FakeLayer:
+    role: str = "main"
+    clips: list[FakeClip] = field(default_factory=list)
+
+
+# ── 常量验证 ────────────────────────────────────────────────────────────────
+
+
+class TestConstants:
+    def test_layer_z_index_has_expected_keys(self):
+        assert set(LAYER_Z_INDEX.keys()) == {
+            "background",
+            "broll",
+            "main",
+            "overlay",
+            "corner_voice",
+            "audio",
+        }
+
+    def test_layer_z_index_ordering(self):
+        assert LAYER_Z_INDEX["background"] < LAYER_Z_INDEX["main"]
+        assert LAYER_Z_INDEX["main"] == LAYER_Z_INDEX["broll"]
+        assert LAYER_Z_INDEX["overlay"] > LAYER_Z_INDEX["main"]
+        assert LAYER_Z_INDEX["corner_voice"] > LAYER_Z_INDEX["main"]
+        assert LAYER_Z_INDEX["audio"] > LAYER_Z_INDEX["overlay"]
+
+    def test_pip_default_scale_positive(self):
+        assert 0 < PIP_DEFAULT_SCALE < 1
+
+    def test_main_layer_roles(self):
+        assert "main" in MAIN_LAYER_ROLES
+        assert "broll" in MAIN_LAYER_ROLES
+        assert "background" in MAIN_LAYER_ROLES
+        assert "overlay" not in MAIN_LAYER_ROLES
+
+
+# ── resolve_layer_role ──────────────────────────────────────────────────────
+
+
+class TestResolveLayerRole:
+    def test_intro_maps_to_main(self):
+        assert resolve_layer_role("intro") == "main"
+
+    def test_outro_maps_to_main(self):
+        assert resolve_layer_role("outro") == "main"
+
+    def test_overlay_maps_to_overlay(self):
+        assert resolve_layer_role("overlay") == "overlay"
+
+    def test_corner_voice_maps_to_corner_voice(self):
+        assert resolve_layer_role("corner_voice") == "corner_voice"
+
+    def test_background_maps_to_background(self):
+        assert resolve_layer_role("background") == "background"
+
+    def test_b_roll_maps_to_broll(self):
+        assert resolve_layer_role("b_roll") == "broll"
+
+    def test_main_defaults_to_main(self):
+        assert resolve_layer_role("main") == "main"
+
+    def test_main_with_b_roll_role(self):
+        assert resolve_layer_role("main", {"role": "b_roll"}) == "broll"
+
+    def test_main_with_audio_role(self):
+        assert resolve_layer_role("main", {"role": "audio"}) == "audio"
+
+    def test_main_with_other_role_stays_main(self):
+        assert resolve_layer_role("main", {"role": "overlay"}) == "main"
+
+    def test_none_config(self):
+        assert resolve_layer_role("main", None) == "main"
+
+    def test_empty_config(self):
+        assert resolve_layer_role("main", {}) == "main"
+
+    def test_unknown_type_defaults_to_main(self):
+        assert resolve_layer_role("unknown_type") == "main"
+
+
+# ── get_layer_z_index ──────────────────────────────────────────────────────
+
+
+class TestGetLayerZIndex:
+    def test_known_roles(self):
+        for role, expected in LAYER_Z_INDEX.items():
+            assert get_layer_z_index(role) == expected
+
+    def test_unknown_role_returns_zero(self):
+        assert get_layer_z_index("nonexistent") == 0
+
+    def test_empty_string_returns_zero(self):
+        assert get_layer_z_index("") == 0
+
+
+# ── clip_effective_duration ────────────────────────────────────────────────
+
+
+class TestClipEffectiveDuration:
+    def test_explicit_duration_no_actual(self):
+        assert clip_effective_duration(5.0) == 5.0
+
+    def test_explicit_duration_with_shorter_actual(self):
+        assert clip_effective_duration(5.0, 3.0) == 3.0
+
+    def test_explicit_duration_with_longer_actual(self):
+        assert clip_effective_duration(5.0, 10.0) == 5.0
+
+    def test_zero_duration_uses_actual(self):
+        assert clip_effective_duration(0, 8.0) == 8.0
+
+    def test_negative_duration_uses_actual(self):
+        assert clip_effective_duration(-1.0, 8.0) == 8.0
+
+    def test_zero_duration_zero_actual(self):
+        assert clip_effective_duration(0, 0) == 0.0
+
+    def test_no_args_returns_zero(self):
+        assert clip_effective_duration(0) == 0.0
+
+    def test_equal_duration_and_actual(self):
+        assert clip_effective_duration(5.0, 5.0) == 5.0
+
+
+# ── clip_playback_speed ────────────────────────────────────────────────────
+
+
+class TestClipPlaybackSpeed:
+    def test_normal_speed(self):
+        assert clip_playback_speed(1.0) == 1.0
+
+    def test_fast_speed(self):
+        assert clip_playback_speed(2.0) == 2.0
+
+    def test_slow_speed(self):
+        assert clip_playback_speed(0.5) == 0.5
+
+    def test_zero_speed_defaults_to_one(self):
+        assert clip_playback_speed(0) == 1.0
+
+    def test_negative_speed_defaults_to_one(self):
+        assert clip_playback_speed(-1.0) == 1.0
+
+    def test_none_defaults_to_one(self):
+        assert clip_playback_speed(None) == 1.0
+
+    def test_string_defaults_to_one(self):
+        assert clip_playback_speed("fast") == 1.0
+
+    def test_int_speed(self):
+        assert clip_playback_speed(2) == 2.0
+
+
+# ── clip_adjusted_duration ─────────────────────────────────────────────────
+
+
+class TestClipAdjustedDuration:
+    def test_normal_speed_same_as_effective(self):
+        assert clip_adjusted_duration(5.0, 10.0, 1.0) == 5.0
+
+    def test_double_speed_half_duration(self):
+        assert clip_adjusted_duration(10.0, 10.0, 2.0) == pytest.approx(5.0)
+
+    def test_half_speed_double_duration(self):
+        assert clip_adjusted_duration(5.0, 10.0, 0.5) == pytest.approx(10.0)
+
+    def test_invalid_speed_uses_default(self):
+        assert clip_adjusted_duration(5.0, 10.0, 0) == 5.0
+
+    def test_zero_duration(self):
+        assert clip_adjusted_duration(0, 0, 1.0) == 0.0
+
+    def test_actual_duration_only(self):
+        assert clip_adjusted_duration(0, 8.0, 1.0) == 8.0
+
+    def test_actual_duration_only_with_speed(self):
+        assert clip_adjusted_duration(0, 8.0, 2.0) == pytest.approx(4.0)
+
+    def test_very_close_to_normal_speed(self):
+        # 1.0000001 应该被认为接近 1.0，不做除法
+        result = clip_adjusted_duration(5.0, 10.0, 1.0 + 1e-10)
+        assert result == 5.0
+
+
+# ── estimate_total_duration ────────────────────────────────────────────────
+
+
+class TestEstimateTotalDuration:
+    def test_empty_layers(self):
+        assert estimate_total_duration([]) == 0.0
+
+    def test_no_main_layer(self):
+        layers = [FakeLayer(role="overlay", clips=[FakeClip(duration=5.0)])]
+        assert estimate_total_duration(layers) == 0.0
+
+    def test_single_clip_main_layer(self):
+        layers = [FakeLayer(role="main", clips=[FakeClip(duration=5.0)])]
+        assert estimate_total_duration(layers) == pytest.approx(5.0)
+
+    def test_multiple_clips_no_transition(self):
+        layers = [
+            FakeLayer(
+                role="main",
+                clips=[
+                    FakeClip(duration=3.0),
+                    FakeClip(duration=2.0),
+                    FakeClip(duration=5.0),
+                ],
+            )
+        ]
+        assert estimate_total_duration(layers) == pytest.approx(10.0)
+
+    def test_multiple_clips_with_transition(self):
+        layers = [
+            FakeLayer(
+                role="main",
+                clips=[
+                    FakeClip(duration=3.0),
+                    FakeClip(duration=2.0),
+                    FakeClip(duration=5.0),
+                ],
+            )
+        ]
+        # 3 + 2 + 5 - 2 * 0.5 = 9.0
+        assert estimate_total_duration(layers, transition_duration=0.5) == pytest.approx(9.0)
+
+    def test_prefers_main_over_broll(self):
+        layers = [
+            FakeLayer(role="broll", clips=[FakeClip(duration=10.0)]),
+            FakeLayer(role="main", clips=[FakeClip(duration=5.0)]),
+        ]
+        assert estimate_total_duration(layers) == pytest.approx(5.0)
+
+    def test_prefers_broll_over_background(self):
+        layers = [
+            FakeLayer(role="background", clips=[FakeClip(duration=10.0)]),
+            FakeLayer(role="broll", clips=[FakeClip(duration=5.0)]),
+        ]
+        assert estimate_total_duration(layers) == pytest.approx(5.0)
+
+    def test_main_layer_empty_clips(self):
+        layers = [FakeLayer(role="main", clips=[])]
+        assert estimate_total_duration(layers) == 0.0
+
+    def test_minimum_duration(self):
+        layers = [
+            FakeLayer(
+                role="main",
+                clips=[
+                    FakeClip(duration=0.01),
+                    FakeClip(duration=0.01),
+                ],
+            )
+        ]
+        result = estimate_total_duration(layers, transition_duration=0.5)
+        assert result >= 0.1
+
+    def test_with_playback_speed(self):
+        layers = [
+            FakeLayer(
+                role="main",
+                clips=[
+                    FakeClip(duration=10.0, playback_speed=2.0),
+                    FakeClip(duration=10.0, playback_speed=0.5),
+                ],
+            )
+        ]
+        # 5 + 20 = 25
+        assert estimate_total_duration(layers) == pytest.approx(25.0)
+
+
+# ── can_pass_through ──────────────────────────────────────────────────────
+
+
+class TestCanPassThrough:
+    def test_single_main_clip_no_effects(self):
+        layers = [FakeLayer(role="main", clips=[FakeClip(duration=5.0)])]
+        assert can_pass_through(layers) is True
+
+    def test_single_broll_clip(self):
+        layers = [FakeLayer(role="broll", clips=[FakeClip(duration=5.0)])]
+        assert can_pass_through(layers) is True
+
+    def test_single_background_clip(self):
+        layers = [FakeLayer(role="background", clips=[FakeClip(duration=5.0)])]
+        assert can_pass_through(layers) is True
+
+    def test_multiple_layers(self):
+        layers = [
+            FakeLayer(role="main", clips=[FakeClip(duration=5.0)]),
+            FakeLayer(role="overlay", clips=[FakeClip(duration=3.0)]),
+        ]
+        assert can_pass_through(layers) is False
+
+    def test_overlay_layer(self):
+        layers = [FakeLayer(role="overlay", clips=[FakeClip(duration=5.0)])]
+        assert can_pass_through(layers) is False
+
+    def test_multiple_clips_in_layer(self):
+        layers = [
+            FakeLayer(
+                role="main",
+                clips=[
+                    FakeClip(duration=3.0),
+                    FakeClip(duration=2.0),
+                ],
+            )
+        ]
+        assert can_pass_through(layers) is False
+
+    def test_with_stickers(self):
+        layers = [FakeLayer(role="main", clips=[FakeClip(duration=5.0)])]
+        assert can_pass_through(layers, has_stickers=True) is False
+
+    def test_with_watermark(self):
+        layers = [FakeLayer(role="main", clips=[FakeClip(duration=5.0)])]
+        assert can_pass_through(layers, has_watermark=True) is False
+
+    def test_with_stickers_and_watermark(self):
+        layers = [FakeLayer(role="main", clips=[FakeClip(duration=5.0)])]
+        assert can_pass_through(layers, has_stickers=True, has_watermark=True) is False
+
+    def test_empty_layer_list(self):
+        assert can_pass_through([]) is False
+
+    def test_audio_layer_only(self):
+        layers = [FakeLayer(role="audio", clips=[FakeClip(duration=5.0)])]
+        assert can_pass_through(layers) is False
@@ -0,0 +1,227 @@
+"""ReverseEngine 纯逻辑单测 — 配置解析 + 滤镜构建 + 安全限制.
+
+全纯函数测试，不依赖 FFmpeg。
+"""
+
+from __future__ import annotations
+
+from unittest.mock import patch
+
+import pytest
+from video_processing.reverse_engine import ReverseConfig, ReverseEngine
+
+# ── ReverseConfig 解析 ────────────────────────────────────────────────────
+
+
+class TestReverseConfigFromDict:
+    """ReverseConfig.from_dict 配置解析测试."""
+
+    def test_none_returns_disabled(self):
+        """None 输入返回 disabled 默认配置."""
+        config = ReverseConfig.from_dict(None)
+        assert config.enabled is False
+        assert config.reverse_video is True
+        assert config.reverse_audio is True
+
+    def test_empty_dict_returns_disabled(self):
+        """空 dict 返回 disabled."""
+        config = ReverseConfig.from_dict({})
+        assert config.enabled is False
+
+    def test_enabled_false_returns_disabled(self):
+        """显式 enabled=False."""
+        config = ReverseConfig.from_dict({"enabled": False})
+        assert config.enabled is False
+
+    def test_enabled_default_flags_default_video_audio(self):
+        """只启用时默认视频音频都倒放."""
+        config = ReverseConfig.from_dict({"enabled": True})
+        assert config.enabled is True
+        assert config.reverse_video is True
+        assert config.reverse_audio is True
+
+    def test_video_only(self):
+        """只倒放视频."""
+        config = ReverseConfig.from_dict(
+            {
+                "enabled": True,
+                "reverse_video": True,
+                "reverse_audio": False,
+            }
+        )
+        assert config.enabled is True
+        assert config.reverse_video is True
+        assert config.reverse_audio is False
+
+    def test_audio_only(self):
+        """只倒放音频."""
+        config = ReverseConfig.from_dict(
+            {
+                "enabled": True,
+                "reverse_video": False,
+                "reverse_audio": True,
+            }
+        )
+        assert config.enabled is True
+        assert config.reverse_video is False
+        assert config.reverse_audio is True
+
+    def test_both_disabled_but_enabled_flag_true(self):
+        """enabled=True 但两个子选项都关了（边缘情况）."""
+        config = ReverseConfig.from_dict(
+            {
+                "enabled": True,
+                "reverse_video": False,
+                "reverse_audio": False,
+            }
+        )
+        assert config.enabled is True
+        assert config.reverse_video is False
+        assert config.reverse_audio is False
+
+    def test_invalid_type_falls_back(self):
+        """非 dict 类型回退到默认 disabled."""
+        config = ReverseConfig.from_dict("reverse=true")
+        assert config.enabled is False
+
+    def test_attribute_error_falls_back(self):
+        """属性错误时回退到默认."""
+
+        class WeirdObj:
+            def get(self, key, default=None):
+                raise AttributeError("nope")
+
+        config = ReverseConfig.from_dict(WeirdObj())
+        assert config.enabled is False
+
+    def test_type_error_falls_back(self):
+        """类型错误时回退到默认."""
+        config = ReverseConfig.from_dict([1, 2, 3])
+        assert config.enabled is False
+
+
+# ── ReverseEngine 视频滤镜 ──────────────────────────────────────────────────
+
+
+class TestReverseEngineBuildVideoFilter:
+    """ReverseEngine.build_video_filter 视频滤镜测试."""
+
+    def test_disabled_returns_empty(self):
+        """disabled 返回空字符串."""
+        config = ReverseConfig(enabled=False)
+        assert ReverseEngine.build_video_filter(config) == ""
+
+    def test_enabled_returns_reverse(self):
+        """启用返回 reverse 滤镜."""
+        config = ReverseConfig(enabled=True, reverse_video=True)
+        assert ReverseEngine.build_video_filter(config) == "reverse"
+
+    def test_video_disabled_returns_empty(self):
+        """reverse_video=False 返回空."""
+        config = ReverseConfig(enabled=True, reverse_video=False, reverse_audio=True)
+        assert ReverseEngine.build_video_filter(config) == ""
+
+    def test_short_duration_ok(self):
+        """短时长正常返回 reverse."""
+        config = ReverseConfig(enabled=True)
+        result = ReverseEngine.build_video_filter(config, duration=10.0)
+        assert result == "reverse"
+
+    def test_exactly_max_duration_ok(self):
+        """恰好等于安全上限正常."""
+        config = ReverseConfig(enabled=True)
+        result = ReverseEngine.build_video_filter(config, duration=ReverseEngine.MAX_SAFE_DURATION)
+        assert result == "reverse"
+
+    def test_over_max_duration_skipped(self):
+        """超过安全时长跳过倒放."""
+        config = ReverseConfig(enabled=True)
+        result = ReverseEngine.build_video_filter(config, duration=200.0)
+        assert result == ""
+
+    def test_zero_duration_ok(self):
+        """零时长正常倒放."""
+        config = ReverseConfig(enabled=True)
+        result = ReverseEngine.build_video_filter(config, duration=0.0)
+        assert result == "reverse"
+
+
+# ── ReverseEngine 音频滤镜 ──────────────────────────────────────────────
+
+
+class TestReverseEngineBuildAudioFilter:
+    """ReverseEngine.build_audio_filter 音频滤镜测试."""
+
+    def test_disabled_returns_empty(self):
+        """disabled 返回空字符串."""
+        config = ReverseConfig(enabled=False)
+        assert ReverseEngine.build_audio_filter(config) == ""
+
+    def test_enabled_returns_areverse(self):
+        """启用返回 areverse 滤镜."""
+        config = ReverseConfig(enabled=True, reverse_audio=True)
+        assert ReverseEngine.build_audio_filter(config) == "areverse"
+
+    def test_audio_disabled_returns_empty(self):
+        """reverse_audio=False 返回空."""
+        config = ReverseConfig(enabled=True, reverse_video=True, reverse_audio=False)
+        assert ReverseEngine.build_audio_filter(config) == ""
+
+    def test_short_duration_ok(self):
+        """短时长正常返回 areverse."""
+        config = ReverseConfig(enabled=True)
+        result = ReverseEngine.build_audio_filter(config, duration=30.0)
+        assert result == "areverse"
+
+    def test_over_max_duration_skipped(self):
+        """超过安全时长跳过音频倒放."""
+        config = ReverseConfig(enabled=True)
+        result = ReverseEngine.build_audio_filter(config, duration=150.0)
+        assert result == ""
+
+    def test_exactly_max_duration_ok(self):
+        """恰好等于安全上限正常."""
+        config = ReverseConfig(enabled=True)
+        result = ReverseEngine.build_audio_filter(config, duration=ReverseEngine.MAX_SAFE_DURATION)
+        assert result == "areverse"
+
+
+# ── 组合场景 ────────────────────────────────────────────────────────────
+
+
+class TestReverseEngineCombined:
+    """组合场景测试."""
+
+    def test_both_video_audio_reverse(self):
+        """视频音频都倒放."""
+        config = ReverseConfig(enabled=True)
+        vf = ReverseEngine.build_video_filter(config)
+        af = ReverseEngine.build_audio_filter(config)
+        assert vf == "reverse"
+        assert af == "areverse"
+
+    def test_neither_video_nor_audio(self):
+        """都不倒放."""
+        config = ReverseConfig(enabled=True, reverse_video=False, reverse_audio=False)
+        assert ReverseEngine.build_video_filter(config) == ""
+        assert ReverseEngine.build_audio_filter(config) == ""
+
+    def test_long_video_both_skipped(self):
+        """超长视频两个都跳过."""
+        config = ReverseConfig(enabled=True)
+        duration = ReverseEngine.MAX_SAFE_DURATION + 1
+        assert ReverseEngine.build_video_filter(config, duration=duration) == ""
+        assert ReverseEngine.build_audio_filter(config, duration=duration) == ""
+
+    def test_from_dict_full_config_flow(self):
+        """从 dict 解析到滤镜构建的完整流程."""
+        data = {"enabled": True, "reverse_video": True, "reverse_audio": False}
+        config = ReverseConfig.from_dict(data)
+        assert ReverseEngine.build_video_filter(config, duration=10) == "reverse"
+        assert ReverseEngine.build_audio_filter(config, duration=10) == ""
+
+    def test_from_dict_disabled_flow(self):
+        """disabled 配置完整流程."""
+        config = ReverseConfig.from_dict({"enabled": False})
+        assert ReverseEngine.build_video_filter(config) == ""
+        assert ReverseEngine.build_audio_filter(config) == ""
@@ -0,0 +1,403 @@
+"""SpeedEngine 纯逻辑单测 — 配置解析 + 调速滤镜 + 时长计算.
+
+全纯函数测试，不依赖 FFmpeg 或外部服务。
+"""
+
+from __future__ import annotations
+
+import pytest
+from video_processing.speed_engine import (
+    MAX_SPEED,
+    MIN_SPEED,
+    SpeedConfig,
+    SpeedEngine,
+)
+
+# ── SpeedConfig 解析 ──────────────────────────────────────────────────────
+
+
+class TestSpeedConfigParse:
+    """SpeedConfig.from_dict / parse 解析测试."""
+
+    def test_none_data_returns_default(self):
+        """None 输入返回默认配置."""
+        config = SpeedConfig.parse(None)
+        assert config.speed == 1.0
+        assert config.pitch_correct is True
+
+    def test_empty_dict_returns_default(self):
+        """空 dict 返回默认配置."""
+        config = SpeedConfig.parse({})
+        assert config.speed == 1.0
+        assert config.pitch_correct is True
+
+    def test_valid_speed_and_pitch(self):
+        """正常速度和音调配置."""
+        config = SpeedConfig.parse({"speed": 2.0, "pitch_correct": False})
+        assert config.speed == 2.0
+        assert config.pitch_correct is False
+
+    def test_speed_as_int(self):
+        """整数 speed 自动转 float."""
+        config = SpeedConfig.parse({"speed": 2})
+        assert config.speed == 2.0
+        assert isinstance(config.speed, float)
+
+    def test_invalid_speed_type_falls_back(self):
+        """speed 类型错误回退到默认."""
+        config = SpeedConfig.parse({"speed": "fast"})
+        assert config.speed == 1.0
+
+    def test_invalid_pitch_correct_type_falls_back(self):
+        """pitch_correct 非 bool 回退到 True."""
+        config = SpeedConfig.parse({"pitch_correct": "yes"})
+        assert config.pitch_correct is True
+
+    def test_non_dict_input_falls_back(self):
+        """非 dict 输入回退到默认."""
+        config = SpeedConfig.parse("speed=2x")
+        assert config.speed == 1.0
+        assert config.pitch_correct is True
+
+
+class TestSpeedConfigClamp:
+    """SpeedConfig.clamp 钳制测试."""
+
+    def test_zero_speed_clamps_to_default(self):
+        """speed=0 钳制到默认 1.0."""
+        config = SpeedConfig(speed=0.0)
+        config.clamp()
+        assert config.speed == 1.0
+
+    def test_negative_speed_clamps_to_default(self):
+        """负速度钳制到默认 1.0."""
+        config = SpeedConfig(speed=-1.0)
+        config.clamp()
+        assert config.speed == 1.0
+
+    def test_below_min_clamps_to_min(self):
+        """低于最小速度钳制到 MIN_SPEED."""
+        config = SpeedConfig(speed=0.1)
+        config.clamp()
+        assert config.speed == MIN_SPEED
+
+    def test_at_min_stays(self):
+        """恰好在最小值保持不变."""
+        config = SpeedConfig(speed=MIN_SPEED)
+        config.clamp()
+        assert config.speed == MIN_SPEED
+
+    def test_above_max_clamps_to_max(self):
+        """超过最大速度钳制到 MAX_SPEED."""
+        config = SpeedConfig(speed=5.0)
+        config.clamp()
+        assert config.speed == MAX_SPEED
+
+    def test_at_max_stays(self):
+        """恰好在最大值保持不变."""
+        config = SpeedConfig(speed=MAX_SPEED)
+        config.clamp()
+        assert config.speed == MAX_SPEED
+
+    def test_normal_speed_stays(self):
+        """正常范围内速度保持不变."""
+        config = SpeedConfig(speed=1.5)
+        config.clamp()
+        assert config.speed == 1.5
+
+    def test_parse_auto_clamps(self):
+        """parse 自动执行 clamp."""
+        config = SpeedConfig.parse({"speed": 10.0})
+        assert config.speed == MAX_SPEED
+
+
+class TestSpeedConfigIsOriginal:
+    """SpeedConfig.is_original 属性测试."""
+
+    def test_default_is_original(self):
+        """默认配置为原速."""
+        assert SpeedConfig().is_original is True
+
+    def test_exactly_one_is_original(self):
+        """speed=1.0 为原速."""
+        assert SpeedConfig(speed=1.0).is_original is True
+
+    def test_very_close_is_original(self):
+        """浮点精度接近 1.0 视为原速."""
+        assert SpeedConfig(speed=1.0000001).is_original is True
+
+    def test_different_speed_not_original(self):
+        """非 1.0 速度不是原速."""
+        assert SpeedConfig(speed=2.0).is_original is False
+        assert SpeedConfig(speed=0.5).is_original is False
+
+
+# ── SpeedEngine 滤镜构建 ──────────────────────────────────────────────────
+
+
+class TestSpeedEngineBuildVideoFilter:
+    """SpeedEngine.build_video_filter 视频滤镜测试."""
+
+    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):
+        """2倍速 setpts=PTS/2."""
+        config = SpeedConfig(speed=2.0)
+        result = self.engine.build_video_filter(config)
+        assert result == "setpts=PTS/2.0000"
+
+    def test_half_speed(self):
+        """0.5倍速 setpts=PTS/0.5."""
+        config = SpeedConfig(speed=0.5)
+        result = self.engine.build_video_filter(config)
+        assert result == "setpts=PTS/0.5000"
+
+    def test_quarter_speed(self):
+        """0.25倍速."""
+        config = SpeedConfig(speed=0.25)
+        result = self.engine.build_video_filter(config)
+        assert "setpts=PTS/0.25" in result
+
+    def test_quad_speed(self):
+        """4倍速."""
+        config = SpeedConfig(speed=4.0)
+        result = self.engine.build_video_filter(config)
+        assert "setpts=PTS/4.0" in result
+
+    def test_custom_speed_precision(self):
+        """自定义速度保留4位小数."""
+        config = SpeedConfig(speed=1.333)
+        result = self.engine.build_video_filter(config)
+        assert result == "setpts=PTS/1.3330"
+
+
+class TestSpeedEngineBuildAudioFilter:
+    """SpeedEngine.build_audio_filter 音频滤镜测试."""
+
+    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_within_single_stage_range(self):
+        """单级 atempo 范围内返回一级."""
+        config = SpeedConfig(speed=1.5)
+        result = self.engine.build_audio_filter(config)
+        assert result == "atempo=1.5000"
+
+    def test_at_max_single_stage(self):
+        """恰好 2.0 单级."""
+        config = SpeedConfig(speed=2.0)
+        result = self.engine.build_audio_filter(config)
+        assert result == "atempo=2.0000"
+
+    def test_at_min_single_stage(self):
+        """恰好 0.5 单级."""
+        config = SpeedConfig(speed=0.5)
+        result = self.engine.build_audio_filter(config)
+        assert result == "atempo=0.5000"
+
+    def test_quad_speed_two_stages(self):
+        """4倍速 = atempo=2.0,atempo=2.0."""
+        config = SpeedConfig(speed=4.0)
+        result = self.engine.build_audio_filter(config)
+        stages = result.split(",")
+        assert len(stages) == 2
+        assert stages[0] == "atempo=2.0000"
+        assert stages[1] == "atempo=2.0000"
+
+    def test_quarter_speed_two_stages(self):
+        """0.25倍速 = atempo=0.5,atempo=0.5."""
+        config = SpeedConfig(speed=0.25)
+        result = self.engine.build_audio_filter(config)
+        stages = result.split(",")
+        assert len(stages) == 2
+        assert stages[0] == "atempo=0.5000"
+        assert stages[1] == "atempo=0.5000"
+
+    def test_triple_speed_two_stages(self):
+        """3倍速 = atempo=2.0,atempo=1.5 (2.0 * 1.5 = 3.0)."""
+        config = SpeedConfig(speed=3.0)
+        result = self.engine.build_audio_filter(config)
+        stages = result.split(",")
+        assert len(stages) == 2
+        # 乘积应为 3.0
+        values = [float(s.split("=")[1]) for s in stages]
+        product = 1.0
+        for v in values:
+            product *= v
+        assert abs(product - 3.0) < 0.01
+
+    def test_low_speed_two_stages(self):
+        """0.3倍速多级串联，乘积为 0.3."""
+        config = SpeedConfig(speed=0.3)
+        result = self.engine.build_audio_filter(config)
+        stages = result.split(",")
+        assert len(stages) >= 2
+        values = [float(s.split("=")[1]) for s in stages]
+        product = 1.0
+        for v in values:
+            product *= v
+        assert abs(product - 0.3) < 0.01
+
+    def test_all_stages_within_valid_range(self):
+        """所有 atempo 级都在 [0.5, 2.0] 范围内."""
+        for speed in [0.25, 0.3, 0.5, 0.8, 1.5, 2.0, 3.0, 4.0]:
+            config = SpeedConfig(speed=speed)
+            result = self.engine.build_audio_filter(config)
+            if not result:
+                continue
+            stages = result.split(",")
+            for stage in stages:
+                val = float(stage.split("=")[1])
+                assert 0.5 <= val <= 2.0, f"speed={speed}, stage={val} out of range"
+
+
+class TestSpeedEngineSplitAtempoStages:
+    """SpeedEngine._split_atempo_stages 拆分算法测试."""
+
+    def test_single_stage_within_range(self):
+        """范围内单级."""
+        stages = SpeedEngine._split_atempo_stages(1.5)
+        assert stages == [1.5]
+
+    def test_exactly_max_single_stage(self):
+        """恰好 2.0 单级."""
+        stages = SpeedEngine._split_atempo_stages(2.0)
+        assert stages == [2.0]
+
+    def test_exactly_min_single_stage(self):
+        """恰好 0.5 单级."""
+        stages = SpeedEngine._split_atempo_stages(0.5)
+        assert stages == [0.5]
+
+    def test_four_x_two_stages(self):
+        """4.0 拆为两级 2.0."""
+        stages = SpeedEngine._split_atempo_stages(4.0)
+        assert stages == [2.0, 2.0]
+
+    def test_quarter_x_two_stages(self):
+        """0.25 拆为两级 0.5."""
+        stages = SpeedEngine._split_atempo_stages(0.25)
+        assert stages == [0.5, 0.5]
+
+    def test_product_matches_original_speed(self):
+        """拆分后乘积应等于原速度."""
+        for speed in [0.25, 0.3, 0.5, 0.8, 1.0, 1.5, 2.0, 2.5, 3.0, 4.0]:
+            stages = SpeedEngine._split_atempo_stages(speed)
+            product = 1.0
+            for s in stages:
+                product *= s
+            assert abs(product - speed) < 0.001, f"speed={speed}, product={product}"
+
+
+class TestSpeedEngineAdjustDuration:
+    """SpeedEngine.adjust_duration 时长计算测试."""
+
+    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):
+        """2倍速时长减半."""
+        config = SpeedConfig(speed=2.0)
+        assert self.engine.adjust_duration(10.0, config) == 5.0
+
+    def test_half_speed_double_duration(self):
+        """0.5倍速时长加倍."""
+        config = SpeedConfig(speed=0.5)
+        assert self.engine.adjust_duration(10.0, config) == 20.0
+
+    def test_zero_duration_stays_zero(self):
+        """零时长保持零."""
+        config = SpeedConfig(speed=2.0)
+        assert self.engine.adjust_duration(0.0, config) == 0.0
+
+    def test_negative_duration_stays(self):
+        """负时长直接返回（不做调速）."""
+        config = SpeedConfig(speed=2.0)
+        assert self.engine.adjust_duration(-5.0, config) == -5.0
+
+    def test_quad_speed_quarter_duration(self):
+        """4倍速时长为1/4."""
+        config = SpeedConfig(speed=4.0)
+        assert self.engine.adjust_duration(20.0, config) == 5.0
+
+
+class TestSpeedEngineBuildClipSpeedFilter:
+    """SpeedEngine.build_clip_speed_filter 便捷方法测试."""
+
+    def setup_method(self):
+        self.engine = SpeedEngine()
+
+    def test_default_speed_returns_empty_filters(self):
+        """默认速度返回空滤镜."""
+        vf, af, config = self.engine.build_clip_speed_filter(1.0)
+        assert vf == ""
+        assert af == ""
+        assert config.is_original is True
+
+    def test_double_speed_filters(self):
+        """2倍速返回对应滤镜."""
+        vf, af, config = self.engine.build_clip_speed_filter(2.0)
+        assert vf == "setpts=PTS/2.0000"
+        assert af == "atempo=2.0000"
+        assert config.speed == 2.0
+
+    def test_speed_gets_clamped(self):
+        """超范围速度自动钳制."""
+        vf, af, config = self.engine.build_clip_speed_filter(10.0)
+        assert config.speed == MAX_SPEED
+        assert "setpts" in vf
+
+    def test_pitch_correct_false_still_has_audio_filter(self):
+        """pitch_correct=False 也返回音频滤镜（只是方式不同，当前实现仍用atempo）."""
+        vf, af, config = self.engine.build_clip_speed_filter(2.0, pitch_correct=False)
+        assert config.pitch_correct is False
+        # 当前实现 pitch_correct 不影响滤镜输出（atempo 本身保持音调）
+        assert "atempo" in af
+
+
+class TestSpeedEngineResolveClipSpeed:
+    """SpeedEngine.resolve_clip_speed 速度解析测试."""
+
+    def test_no_clip_config_uses_global(self):
+        """无 clip config 使用全局速度."""
+        assert SpeedEngine.resolve_clip_speed(None, 2.0) == 2.0
+
+    def test_empty_config_uses_global(self):
+        """空 config 使用全局速度."""
+        assert SpeedEngine.resolve_clip_speed({}, 1.5) == 1.5
+
+    def test_zero_speed_uses_global(self):
+        """playback_speed=0 使用全局."""
+        assert SpeedEngine.resolve_clip_speed({"playback_speed": 0}, 2.0) == 2.0
+
+    def test_valid_clip_speed(self):
+        """有效 clip 速度优先."""
+        assert SpeedEngine.resolve_clip_speed({"playback_speed": 1.5}, 1.0) == 1.5
+
+    def test_invalid_speed_type_uses_global(self):
+        """速度类型错误回退到全局."""
+        assert SpeedEngine.resolve_clip_speed({"playback_speed": "fast"}, 2.0) == 2.0
+
+    def test_negative_speed_uses_global(self):
+        """负速度回退到全局."""
+        assert SpeedEngine.resolve_clip_speed({"playback_speed": -1}, 1.0) == 1.0
+
+    def test_default_global_is_one(self):
+        """默认全局速度为 1.0."""
+        assert SpeedEngine.resolve_clip_speed({}) == 1.0
Author	SHA1	Message	Date
CI Bot	8f99774620	test(wave105): extract render_layer_utils domain module + 65 unit tests CI/CD Pipeline / Check if frontend-only change (pull_request) Successful in 22s Details CI/CD Pipeline / Frontend Lint (pull_request) Failing after 37s Details CI/CD Pipeline / Validate - Type Check (mypy) (pull_request) Successful in 1m10s Details CI/CD Pipeline / Validate - Migration (alembic) (pull_request) Successful in 54s Details CI/CD Pipeline / PR Build Worker Image (pull_request) Successful in 24s Details CI/CD Pipeline / Build Staging API Image (pull_request) Has been skipped Details CI/CD Pipeline / Build Staging Web Image (pull_request) Has been skipped Details CI/CD Pipeline / Build Staging Worker Image (pull_request) Has been skipped Details CI/CD Pipeline / PR Build Web Image (pull_request) Successful in 30s Details CI/CD Pipeline / Validate - Code Quality (pull_request) Failing after 1m51s Details PR Automation / Auto Merge on CI Green + Approved (pull_request) Successful in 49s Details Preview Deploy / Deploy Preview Environment (pull_request) Successful in 46s Details CI/CD Pipeline / PR Build API Image (pull_request) Successful in 2m13s Details CI/CD Pipeline / Frontend Unit Tests (pull_request) Has been skipped Details CI/CD Pipeline / Build Production API Image (pull_request) Has been skipped Details CI/CD Pipeline / Build Production Web Image (pull_request) Has been skipped Details CI/CD Pipeline / Build Production Worker Image (pull_request) Has been skipped Details CI/CD Pipeline / Deploy Staging (Watchtower auto-deploy) (pull_request) Has been skipped Details PR Automation / Auto Approve on CI Green (pull_request) Successful in 3m1s Details CI/CD Pipeline / Deploy Production (pull_request) Has been skipped Details CI/CD Pipeline / Staging E2E Tests (pull_request) Has been skipped Details CI/CD Pipeline / Staging API Integration Tests (pull_request) Has been skipped Details CI/CD Pipeline / ACR Image Cleanup (pull_request) Has been skipped Details CI/CD Pipeline / Production Browser E2E (pull_request) Has been skipped Details CI/CD Pipeline / Canary Release to Production (pull_request) Has been skipped Details AI Code Review / AI Code Review (pull_request) Successful in 3m36s Details CI/CD Pipeline / Integration Tests (pull_request) Successful in 1m56s Details CI/CD Pipeline / Unit Tests (pull_request) Failing after 2m20s Details ACR Cleanup / ACR Image Cleanup (pull_request_target) Has been cancelled Details Preview Cleanup / Cleanup Preview Environment (pull_request) Successful in 24s Details 从 unified_render_service.py (1985→1947行) 抽出渲染图层纯逻辑工具: - resolve_layer_role / get_layer_z_index: 图层角色解析 - clip_effective_duration / clip_playback_speed / clip_adjusted_duration: 时长计算 - estimate_total_duration: 总时长估算 - can_pass_through: 直通路径判断 - LAYER_Z_INDEX / MAIN_LAYER_ROLES 等常量保留向后兼容: 模块级函数 + 类静态方法均委托到新模块验证: 88 passed (23原有 + 65新增)，全套render 373 passed	2026-07-26 22:19:12 +08:00
xiaoxia	3ff041440b	test(wave93): 43 unit tests for duplication domain models (#956 ) CI/CD Pipeline / Build Production API Image (push) Has been skipped Details CI/CD Pipeline / Check if frontend-only change (push) Has been skipped Details CI/CD Pipeline / Validate - Code Quality (push) Failing after 2m37s Details CI/CD Pipeline / Validate - Type Check (mypy) (push) Successful in 1m20s Details CI/CD Pipeline / Validate - Migration (alembic) (push) Successful in 1m14s Details CI/CD Pipeline / Frontend Lint (push) Failing after 47s Details CI/CD Pipeline / PR Build API Image (push) Has been skipped Details CI/CD Pipeline / PR Build Web Image (push) Has been skipped Details CI/CD Pipeline / PR Build Worker Image (push) Has been skipped Details CI/CD Pipeline / Build Staging API Image (push) Successful in 19m58s Details CI/CD Pipeline / Build Staging Web Image (push) Successful in 2m19s Details CI/CD Pipeline / Build Staging Worker Image (push) Successful in 8m46s Details CI/CD Pipeline / Build Production Web Image (push) Has been skipped Details CI/CD Pipeline / Build Production Worker Image (push) Has been skipped Details CI/CD Pipeline / Frontend Unit Tests (push) Successful in 1m15s Details CI/CD Pipeline / Integration Tests (push) Successful in 2m34s Details CI/CD Pipeline / Deploy Production (push) Has been cancelled Details CI/CD Pipeline / Deploy Staging (Watchtower auto-deploy) (push) Successful in 2m4s Details CI/CD Pipeline / Production Browser E2E (push) Has been skipped Details CI/CD Pipeline / ACR Image Cleanup (push) Successful in 31s Details CI/CD Pipeline / Staging E2E Tests (push) Successful in 2m15s Details CI/CD Pipeline / Staging API Integration Tests (push) Successful in 4m55s Details CI/CD Pipeline / Canary Release to Production (push) Has been skipped Details CI/CD Pipeline / Unit Tests (push) Failing after 43m35s Details	2026-07-26 18:17:04 +08:00
xiaoxia	b896873ece	test(wave92): extract asset_scoring domain module + 89 unit tests (#955 ) CI/CD Pipeline / Check if frontend-only change (push) Has been cancelled Details CI/CD Pipeline / Validate - Code Quality (push) Has been cancelled Details CI/CD Pipeline / Validate - Type Check (mypy) (push) Has been cancelled Details CI/CD Pipeline / Validate - Migration (alembic) (push) Has been cancelled Details CI/CD Pipeline / Unit Tests (push) Has been cancelled Details CI/CD Pipeline / Integration Tests (push) Has been cancelled Details CI/CD Pipeline / Frontend Lint (push) Has been cancelled Details CI/CD Pipeline / Frontend Unit Tests (push) Has been cancelled Details CI/CD Pipeline / PR Build API Image (push) Has been cancelled Details CI/CD Pipeline / PR Build Web Image (push) Has been cancelled Details CI/CD Pipeline / PR Build Worker Image (push) Has been cancelled Details CI/CD Pipeline / Build Staging API Image (push) Has been cancelled Details CI/CD Pipeline / Build Staging Web Image (push) Has been cancelled Details CI/CD Pipeline / Build Staging Worker Image (push) Has been cancelled Details CI/CD Pipeline / Deploy Staging (Watchtower auto-deploy) (push) Has been cancelled Details CI/CD Pipeline / Staging E2E Tests (push) Has been cancelled Details CI/CD Pipeline / Staging API Integration Tests (push) Has been cancelled Details CI/CD Pipeline / Build Production API Image (push) Has been cancelled Details CI/CD Pipeline / Build Production Web Image (push) Has been cancelled Details CI/CD Pipeline / Build Production Worker Image (push) Has been cancelled Details CI/CD Pipeline / Deploy Production (push) Has been cancelled Details CI/CD Pipeline / Production Browser E2E (push) Has been cancelled Details CI/CD Pipeline / ACR Image Cleanup (push) Has been cancelled Details CI/CD Pipeline / Canary Release to Production (push) Has been cancelled Details	2026-07-26 18:15:59 +08:00
xiaoxia	4df4a937e4	test(wave91): add 59 unit tests for generation_plan_builder (#945 ) CI/CD Pipeline / Check if frontend-only change (push) Has been cancelled Details CI/CD Pipeline / Validate - Code Quality (push) Has been cancelled Details CI/CD Pipeline / Validate - Type Check (mypy) (push) Has been cancelled Details CI/CD Pipeline / Validate - Migration (alembic) (push) Has been cancelled Details CI/CD Pipeline / Unit Tests (push) Has been cancelled Details CI/CD Pipeline / Integration Tests (push) Has been cancelled Details CI/CD Pipeline / Frontend Lint (push) Has been cancelled Details CI/CD Pipeline / Frontend Unit Tests (push) Has been cancelled Details CI/CD Pipeline / PR Build API Image (push) Has been cancelled Details CI/CD Pipeline / PR Build Web Image (push) Has been cancelled Details CI/CD Pipeline / PR Build Worker Image (push) Has been cancelled Details CI/CD Pipeline / Build Staging API Image (push) Has been cancelled Details CI/CD Pipeline / Build Staging Web Image (push) Has been cancelled Details CI/CD Pipeline / Build Staging Worker Image (push) Has been cancelled Details CI/CD Pipeline / Deploy Staging (Watchtower auto-deploy) (push) Has been cancelled Details CI/CD Pipeline / Staging E2E Tests (push) Has been cancelled Details CI/CD Pipeline / Staging API Integration Tests (push) Has been cancelled Details CI/CD Pipeline / Build Production API Image (push) Has been cancelled Details CI/CD Pipeline / Build Production Web Image (push) Has been cancelled Details CI/CD Pipeline / Build Production Worker Image (push) Has been cancelled Details CI/CD Pipeline / Deploy Production (push) Has been cancelled Details CI/CD Pipeline / Production Browser E2E (push) Has been cancelled Details CI/CD Pipeline / ACR Image Cleanup (push) Has been cancelled Details CI/CD Pipeline / Canary Release to Production (push) Has been cancelled Details	2026-07-26 18:15:48 +08:00
xiaoxia	673d18aa83	test(wave90): add 52 unit tests for plan_generator_utils (#940 ) CI/CD Pipeline / Check if frontend-only change (push) Has been cancelled Details CI/CD Pipeline / Validate - Code Quality (push) Has been cancelled Details CI/CD Pipeline / Validate - Type Check (mypy) (push) Has been cancelled Details CI/CD Pipeline / Validate - Migration (alembic) (push) Has been cancelled Details CI/CD Pipeline / Unit Tests (push) Has been cancelled Details CI/CD Pipeline / Integration Tests (push) Has been cancelled Details CI/CD Pipeline / Frontend Lint (push) Has been cancelled Details CI/CD Pipeline / Frontend Unit Tests (push) Has been cancelled Details CI/CD Pipeline / PR Build API Image (push) Has been cancelled Details CI/CD Pipeline / PR Build Web Image (push) Has been cancelled Details CI/CD Pipeline / PR Build Worker Image (push) Has been cancelled Details CI/CD Pipeline / Build Staging API Image (push) Has been cancelled Details CI/CD Pipeline / Build Staging Web Image (push) Has been cancelled Details CI/CD Pipeline / Build Staging Worker Image (push) Has been cancelled Details CI/CD Pipeline / Deploy Staging (Watchtower auto-deploy) (push) Has been cancelled Details CI/CD Pipeline / Staging E2E Tests (push) Has been cancelled Details CI/CD Pipeline / Staging API Integration Tests (push) Has been cancelled Details CI/CD Pipeline / Build Production API Image (push) Has been cancelled Details CI/CD Pipeline / Build Production Web Image (push) Has been cancelled Details CI/CD Pipeline / Build Production Worker Image (push) Has been cancelled Details CI/CD Pipeline / Deploy Production (push) Has been cancelled Details CI/CD Pipeline / Production Browser E2E (push) Has been cancelled Details CI/CD Pipeline / ACR Image Cleanup (push) Has been cancelled Details CI/CD Pipeline / Canary Release to Production (push) Has been cancelled Details	2026-07-26 18:15:38 +08:00
xiaoxia	3b949a464f	第89波 - 拆分asset_analyzer评分纯逻辑 + 64单测 (#938 ) CI/CD Pipeline / Check if frontend-only change (push) Has been cancelled Details CI/CD Pipeline / Validate - Code Quality (push) Has been cancelled Details CI/CD Pipeline / Validate - Type Check (mypy) (push) Has been cancelled Details CI/CD Pipeline / Validate - Migration (alembic) (push) Has been cancelled Details CI/CD Pipeline / Unit Tests (push) Has been cancelled Details CI/CD Pipeline / Integration Tests (push) Has been cancelled Details CI/CD Pipeline / Frontend Lint (push) Has been cancelled Details CI/CD Pipeline / Frontend Unit Tests (push) Has been cancelled Details CI/CD Pipeline / PR Build API Image (push) Has been cancelled Details CI/CD Pipeline / PR Build Web Image (push) Has been cancelled Details CI/CD Pipeline / PR Build Worker Image (push) Has been cancelled Details CI/CD Pipeline / Build Staging API Image (push) Has been cancelled Details CI/CD Pipeline / Build Staging Web Image (push) Has been cancelled Details CI/CD Pipeline / Build Staging Worker Image (push) Has been cancelled Details CI/CD Pipeline / Deploy Staging (Watchtower auto-deploy) (push) Has been cancelled Details CI/CD Pipeline / Staging E2E Tests (push) Has been cancelled Details CI/CD Pipeline / Staging API Integration Tests (push) Has been cancelled Details CI/CD Pipeline / Build Production API Image (push) Has been cancelled Details CI/CD Pipeline / Build Production Web Image (push) Has been cancelled Details CI/CD Pipeline / Build Production Worker Image (push) Has been cancelled Details CI/CD Pipeline / Deploy Production (push) Has been cancelled Details CI/CD Pipeline / Production Browser E2E (push) Has been cancelled Details CI/CD Pipeline / ACR Image Cleanup (push) Has been cancelled Details CI/CD Pipeline / Canary Release to Production (push) Has been cancelled Details	2026-07-26 18:15:22 +08:00
xiaoxia	daa0e1f7b5	fix(lint): 修复EditingPlanner Phase2引入的TS类型错误 (#936 ) CI/CD Pipeline / Check if frontend-only change (push) Has been cancelled Details CI/CD Pipeline / Validate - Code Quality (push) Has been cancelled Details CI/CD Pipeline / Validate - Type Check (mypy) (push) Has been cancelled Details CI/CD Pipeline / Validate - Migration (alembic) (push) Has been cancelled Details CI/CD Pipeline / Unit Tests (push) Has been cancelled Details CI/CD Pipeline / Integration Tests (push) Has been cancelled Details CI/CD Pipeline / Frontend Lint (push) Has been cancelled Details CI/CD Pipeline / Frontend Unit Tests (push) Has been cancelled Details CI/CD Pipeline / PR Build API Image (push) Has been cancelled Details CI/CD Pipeline / PR Build Web Image (push) Has been cancelled Details CI/CD Pipeline / PR Build Worker Image (push) Has been cancelled Details CI/CD Pipeline / Build Staging API Image (push) Has been cancelled Details CI/CD Pipeline / Build Staging Web Image (push) Has been cancelled Details CI/CD Pipeline / Build Staging Worker Image (push) Has been cancelled Details CI/CD Pipeline / Deploy Staging (Watchtower auto-deploy) (push) Has been cancelled Details CI/CD Pipeline / Staging E2E Tests (push) Has been cancelled Details CI/CD Pipeline / Staging API Integration Tests (push) Has been cancelled Details CI/CD Pipeline / Build Production API Image (push) Has been cancelled Details CI/CD Pipeline / Build Production Web Image (push) Has been cancelled Details CI/CD Pipeline / Build Production Worker Image (push) Has been cancelled Details CI/CD Pipeline / Deploy Production (push) Has been cancelled Details CI/CD Pipeline / Production Browser E2E (push) Has been cancelled Details CI/CD Pipeline / ACR Image Cleanup (push) Has been cancelled Details CI/CD Pipeline / Canary Release to Production (push) Has been cancelled Details	2026-07-26 17:07:30 +08:00
CI Bot	5fa915cdad	style: auto-format with black + isort + prettier CI/CD Pipeline / Check if frontend-only change (push) Has been cancelled Details CI/CD Pipeline / Validate - Code Quality (push) Has been cancelled Details CI/CD Pipeline / Validate - Type Check (mypy) (push) Has been cancelled Details CI/CD Pipeline / Validate - Migration (alembic) (push) Has been cancelled Details CI/CD Pipeline / Unit Tests (push) Has been cancelled Details CI/CD Pipeline / Integration Tests (push) Has been cancelled Details CI/CD Pipeline / Frontend Lint (push) Has been cancelled Details CI/CD Pipeline / Frontend Unit Tests (push) Has been cancelled Details CI/CD Pipeline / PR Build API Image (push) Has been cancelled Details CI/CD Pipeline / PR Build Web Image (push) Has been cancelled Details CI/CD Pipeline / PR Build Worker Image (push) Has been cancelled Details CI/CD Pipeline / Build Staging API Image (push) Has been cancelled Details CI/CD Pipeline / Build Staging Web Image (push) Has been cancelled Details CI/CD Pipeline / Build Staging Worker Image (push) Has been cancelled Details CI/CD Pipeline / Deploy Staging (Watchtower auto-deploy) (push) Has been cancelled Details CI/CD Pipeline / Staging E2E Tests (push) Has been cancelled Details CI/CD Pipeline / Staging API Integration Tests (push) Has been cancelled Details CI/CD Pipeline / Build Production API Image (push) Has been cancelled Details CI/CD Pipeline / Build Production Web Image (push) Has been cancelled Details CI/CD Pipeline / Build Production Worker Image (push) Has been cancelled Details CI/CD Pipeline / Deploy Production (push) Has been cancelled Details CI/CD Pipeline / Production Browser E2E (push) Has been cancelled Details CI/CD Pipeline / ACR Image Cleanup (push) Has been cancelled Details CI/CD Pipeline / Canary Release to Production (push) Has been cancelled Details	2026-07-26 08:29:38 +00:00
xiaoxia	1242b62165	test(wave87): worker层speed_engine+reverse_engine纯逻辑单测 (+85) (#930 ) CI/CD Pipeline / Check if frontend-only change (push) Waiting to run Details CI/CD Pipeline / Validate - Code Quality (push) Waiting to run Details CI/CD Pipeline / Validate - Type Check (mypy) (push) Waiting to run Details CI/CD Pipeline / Validate - Migration (alembic) (push) Waiting to run Details CI/CD Pipeline / Unit Tests (push) Blocked by required conditions Details CI/CD Pipeline / Integration Tests (push) Blocked by required conditions Details CI/CD Pipeline / Frontend Lint (push) Waiting to run Details CI/CD Pipeline / Frontend Unit Tests (push) Blocked by required conditions Details CI/CD Pipeline / PR Build API Image (push) Waiting to run Details CI/CD Pipeline / PR Build Web Image (push) Waiting to run Details CI/CD Pipeline / PR Build Worker Image (push) Waiting to run Details CI/CD Pipeline / Build Staging API Image (push) Waiting to run Details CI/CD Pipeline / Build Staging Web Image (push) Waiting to run Details CI/CD Pipeline / Build Staging Worker Image (push) Waiting to run Details CI/CD Pipeline / Deploy Staging (Watchtower auto-deploy) (push) Blocked by required conditions Details CI/CD Pipeline / Staging E2E Tests (push) Blocked by required conditions Details CI/CD Pipeline / Staging API Integration Tests (push) Blocked by required conditions Details CI/CD Pipeline / Build Production API Image (push) Waiting to run Details CI/CD Pipeline / Build Production Web Image (push) Waiting to run Details CI/CD Pipeline / Build Production Worker Image (push) Waiting to run Details CI/CD Pipeline / Deploy Production (push) Blocked by required conditions Details CI/CD Pipeline / Production Browser E2E (push) Blocked by required conditions Details CI/CD Pipeline / ACR Image Cleanup (push) Blocked by required conditions Details CI/CD Pipeline / Canary Release to Production (push) Blocked by required conditions Details	2026-07-26 16:22:38 +08:00