feat: SceneChange 场景检测前置到渲染前 + 场景点缓存到素材 metadata (#1622)

Co-authored-by: xiaoxia <dev@xiaoxiajianji.com>
Co-committed-by: xiaoxia <dev@xiaoxiajianji.com>

apps/api/app/api/routes/templates_editor/clips.py

@@ -47,8 +47,14 @@ from packages.adapters.sqlalchemy_impl.template_clip_config_repository import (
 from packages.adapters.sqlalchemy_impl.template_repository import (
     SQLAlchemyTemplateRepository,
 )
-from packages.domain.plan_generator_utils import _calc_random_start_time
-from packages.domain.smart_match import score_asset
+from packages.domain.plan_generator_utils import (
+    _calc_random_start_time,
+    build_scene_segments,
+    extract_scene_points_from_metadata,
+    pick_scene_aware_start,
+    pick_start_in_scene_segment,
+)
+from packages.domain.smart_match import SCORE_RANDOM_NOISE_MAX, score_asset
 from packages.shared.mediakit_client import get_mediakit_client
 
 from .dependencies import get_draft_plan_id, get_editor_services
@@ -475,6 +481,12 @@ def _recommended_time_conflicts(
     return False
 
 
+# 向后兼容别名：镜头段构建/段内取点逻辑已下沉到 packages.domain.plan_generator_utils，
+# 旧测试与历史代码仍按 clips._build_scene_segments / _pick_start_in_scene_segment 导入
+_build_scene_segments = build_scene_segments
+_pick_start_in_scene_segment = pick_start_in_scene_segment
+
+
 def _get_mediakit_recommendations(
     asset_ids: list[str],
     asset_repo,
@@ -673,6 +685,9 @@ def create_clips_from_assets_editor(
     unique_asset_ids = list(dict.fromkeys(asset_ids))
     asset_durations: dict[str, float] = {}
     asset_smart_scores: dict[str, float] = {}
+    # 素材 metadata 中缓存的场景切换点（由后台 MediaKit SceneChange 检测写入）：
+    # 有缓存时片段起点从随机镜头段中选取（不同片段来自不同镜头），无缓存回退随机起点
+    asset_scene_points: dict[str, list[float]] = {}
     for asset_id in unique_asset_ids:
         asset = asset_repo.get(asset_id)
         if asset and hasattr(asset, "duration"):
@@ -680,6 +695,15 @@ def create_clips_from_assets_editor(
             # 计算 smart_match 综合评分，用于候选排序
             smart_score, _ = score_asset(asset)
             asset_smart_scores[asset_id] = smart_score
+            # 读取场景切换点缓存（新素材未检测过时为 None，走随机起点兜底）
+            cached_points = extract_scene_points_from_metadata(getattr(asset, "metadata", None))
+            if cached_points:
+                asset_scene_points[asset_id] = cached_points
+    logger.info(
+        "from-assets 场景缓存命中: %d/%d 个素材有场景切换点",
+        len(asset_scene_points),
+        len(unique_asset_ids),
+    )
 
     # 3. 在内存中计算所有片段数据（使用随机起始时间，不调用MediaKit）
     # 读取素材 metadata 中持久化的历史已用区间（跨任务/跨调用去重），
@@ -744,14 +768,15 @@ def create_clips_from_assets_editor(
         clip_duration = 0.0
         start_time: float | None = None
         # 动态按使用次数排序：优先选使用最少的素材，同次数随机打散
-        asset_use_counts = {
-            aid: len(used_segments.get(aid, []))
-            for aid in asset_ids
-        }
+        asset_use_counts = {aid: len(used_segments.get(aid, [])) for aid in asset_ids}
+        # 排序键：smart_match 评分（注入随机噪声）→ 使用次数 → 纯随机。
+        # 噪声让得分接近的素材排名每次浮动，避免同一批素材反复选出相同组合，
+        # 从素材组合层面降低成片查重率；分差 > SCORE_RANDOM_NOISE_MAX 时排名稳定，
+        # 质量差距显著的素材仍保持优先级。
         sorted_candidates = sorted(
             asset_ids,
             key=lambda aid: (
-                -asset_smart_scores.get(aid, 0.0),
+                -(asset_smart_scores.get(aid, 0.0) + random.uniform(0.0, SCORE_RANDOM_NOISE_MAX)),
                 asset_use_counts.get(aid, 0),
                 random.random(),
             ),
@@ -770,16 +795,30 @@ def create_clips_from_assets_editor(
                     candidate,
                 )
                 continue
-            # 随机起始时间（不调用 MediaKit，保证接口快速返回）；100 次避不开
-            # 历史区间时走受控复用回调（复用片段累加 reused_durations，回调内部
-            # 预判复用后占比超 10% 则拒绝并返回 None）
-            candidate_start = _calc_random_start_time(
-                candidate,
-                candidate_duration,
-                asset_durations,
-                used_segments,
-                on_exhausted=reuse_cb,
-            )
+            # 起始时间选取（不调用 MediaKit，保证接口快速返回）：
+            # 1) 素材有场景切换点缓存时，优先从随机镜头段中选起点（不同片段来自不同镜头，
+            #    画面内容本质不同），与 used_segments 做冲突避让（含 1.5s 边缘间隙）
+            # 2) 无缓存 / 镜头段全冲突 → _calc_random_start_time 随机起点兜底；
+            #    100 次避不开历史区间时走受控复用回调（复用片段累加 reused_durations，
+            #    回调内部预判复用后占比超 10% 则拒绝并返回 None）
+            candidate_start = None
+            if candidate in asset_scene_points:
+                candidate_start = pick_scene_aware_start(
+                    candidate,
+                    candidate_duration,
+                    asset_durations,
+                    asset_scene_points,
+                    used_segments,
+                    edge_gap=SEGMENT_EDGE_GAP,
+                )
+            if candidate_start is None:
+                candidate_start = _calc_random_start_time(
+                    candidate,
+                    candidate_duration,
+                    asset_durations,
+                    used_segments,
+                    on_exhausted=reuse_cb,
+                )
             if candidate_start is None:
                 # 该素材可用区间耗尽且复用被闸门/use_count 上限拒绝 → 尝试下一素材
                 logger.info(
@@ -847,15 +886,15 @@ def create_clips_from_assets_editor(
         duplicate_warning = f"查重率 {dup_rate:.1f}% 超过50%，建议更换素材或模板"
         logger.warning(
             "from-assets 成片查重率超标: plan_id=%s dup_rate=%.1f%%",
-            plan_id, dup_rate,
+            plan_id,
+            dup_rate,
         )
 
     # 7. 素材耗尽提示
     exhaustion_warning = None
     if all_assets_exhausted and created_count < len(segments):
         exhaustion_warning = (
-            "素材可切区间不足，部分片段使用了复用素材。"
-            "建议：1) 补充更多素材到素材库 2) 使用不同的素材组合生成"
+            "素材可切区间不足，部分片段使用了复用素材。" "建议：1) 补充更多素材到素材库 2) 使用不同的素材组合生成"
         )
 
     # 8. 立即返回响应
@@ -868,45 +907,6 @@ def create_clips_from_assets_editor(
     )
 
 
-def _build_scene_segments(
-    scene_changes: list[float],
-    asset_duration: float,
-) -> list[tuple[float, float]]:
-    """根据场景切换点构建镜头段列表.
-
-    Args:
-        scene_changes: 场景切换点时间戳列表（已排序，首位为 0.0）
-        asset_duration: 素材总时长
-
-    Returns:
-        镜头段列表 [(start, end), ...]
-    """
-    segments: list[tuple[float, float]] = []
-    for i, ts in enumerate(scene_changes):
-        end = scene_changes[i + 1] if i + 1 < len(scene_changes) else asset_duration
-        # 只保留有效长度的镜头段（至少 0.5 秒）
-        if end - ts >= 0.5:
-            segments.append((ts, end))
-    return segments
-
-
-def _pick_start_in_scene_segment(
-    seg_start: float,
-    seg_end: float,
-    clip_duration: float,
-) -> float | None:
-    """在镜头段内随机选取一个起始时间点.
-
-    确保 start + clip_duration <= seg_end。
-    若镜头段长度不足以容纳片段，返回 None。
-    """
-    available = seg_end - seg_start - clip_duration
-    if available < 0:
-        return None
-    max_start = seg_start + available
-    return random.uniform(seg_start, max_start)
-
-
 def _update_mediakit_recommendations_async(  # pragma: no cover
     plan_id: str,
     asset_ids: list[str],
@@ -1022,21 +1022,54 @@ def _update_mediakit_recommendations_async(  # pragma: no cover
                     if cid != clip_id_inner and cid not in updated_clip_ids:
                         segs.append((c.start_time, c.start_time + c.duration))
                 segs.extend(updated_segments.get(asset_id_inner, []))
+
                 # 并入历史已用区间
                 def _norm(segs_in):
                     return {(round(float(a), 3), round(float(b), 3)) for a, b in segs_in}
+
                 return list(_norm(segs) | _norm(historical_segments.get(asset_id_inner, [])))
 
             # 优先使用 SceneChange 策略
             scene_segments: list[tuple[float, float]] = []
-            if client.is_available and video_url:
+            # 先查素材 metadata 中的场景点缓存：命中则直接复用，跳过 MediaKit 检测
+            # （缓存由本任务首次检测后写入，跨任务/跨 plan 复用）
+            cached_points = extract_scene_points_from_metadata(getattr(asset, "metadata", None))
+            if cached_points:
+                scene_segments = build_scene_segments(cached_points, asset_total)
+                logger.info(
+                    "后台任务: 命中场景点缓存: asset_id=%s scenes=%d",
+                    asset_id,
+                    len(scene_segments),
+                )
+
+            if not scene_segments and client.is_available and video_url:
                 scene_changes = client.detect_scene_changes(video_url)
                 if scene_changes is not None:
-                    scene_segments = _build_scene_segments(scene_changes, asset_total)
+                    scene_segments = build_scene_segments(scene_changes, asset_total)
                     logger.info(
                         "后台任务: 素材场景检测完成: asset_id=%s scenes=%d",
-                        asset_id, len(scene_segments),
+                        asset_id,
+                        len(scene_segments),
                     )
+                    # 检测结果写入素材 metadata 缓存：首次生成用随机起点，
+                    # 检测完成后后续生成的渲染前同步路径即可读缓存选镜头段
+                    try:
+                        existing_meta = dict(getattr(asset, "metadata", None) or {})
+                        existing_meta["scene_change_points"] = scene_changes
+                        asset.metadata = existing_meta
+                        asset_repo.update(asset)
+                        logger.info(
+                            "后台任务: 场景点已写入素材缓存: asset_id=%s points=%d",
+                            asset_id,
+                            len(scene_changes),
+                        )
+                    except Exception as cache_err:
+                        # 缓存写入失败不影响本次片段更新
+                        logger.warning(
+                            "后台任务: 场景点缓存写入失败: asset_id=%s error=%s",
+                            asset_id,
+                            cache_err,
+                        )
 
             # SceneChange 未获得有效结果 → 尝试 analyze_videos 作为 fallback
             if not scene_segments and video_url:
@@ -1047,7 +1080,8 @@ def _update_mediakit_recommendations_async(  # pragma: no cover
                     scene_segments = [(rec_start, asset_total)]
                     logger.info(
                         "后台任务: 使用 analyze_videos fallback: asset_id=%s start=%.2f",
-                        asset_id, rec_start,
+                        asset_id,
+                        rec_start,
                     )
 
             if not scene_segments:
@@ -1066,7 +1100,7 @@ def _update_mediakit_recommendations_async(  # pragma: no cover
 
                 # 从镜头段池中依次尝试，选一个不冲突的
                 for seg_idx, (seg_start, seg_end) in enumerate(scene_segments_pool):
-                    candidate_start = _pick_start_in_scene_segment(seg_start, seg_end, clip_duration)
+                    candidate_start = pick_start_in_scene_segment(seg_start, seg_end, clip_duration)
                     if candidate_start is None:
                         continue  # 镜头段太短，跳过
 

apps/api/app/services/plan_generator_service.py

@@ -13,6 +13,7 @@
 from __future__ import annotations
 
 import logging
+import random
 from typing import Any, List
 
 from sqlalchemy.orm import Session
@@ -29,10 +30,11 @@ from packages.domain.editing_mode import EditingMode
 from packages.domain.plan_generator_utils import (
     create_clips_from_configs,
     distribute_assets,
+    extract_scene_points_from_metadata,
     generate_default_clips,
     map_clip_types_for_mode,
 )
-from packages.domain.smart_match import score_asset
+from packages.domain.smart_match import SCORE_RANDOM_NOISE_MAX, score_asset
 from packages.domain.template_clip_config import TemplateClipConfig
 
 logger = logging.getLogger(__name__)
@@ -222,9 +224,15 @@ class PlanGeneratorService:
         先用 smart_match 评分对素材排序（高分优先），再委托给
         plan_generator_utils.distribute_assets 纯函数完成分配。
         """
-        # 用 smart_match 评分排序素材：高分（质量好/时长合适/新鲜/未使用）优先
-        if self._asset_repo and not random_selection:
-            asset_ids = self._sort_assets_by_smart_score(asset_ids)
+        # 预览随机模式：素材顺序已 shuffle，纯随机起点即可，不读 DB 评分/缓存
+        asset_scene_points: dict[str, list[float]] = {}
+        if not random_selection:
+            # 正式生成：smart_match 评分排序（高分优先）+ 场景切换点缓存
+            if self._asset_repo:
+                asset_ids = self._sort_assets_by_smart_score(asset_ids)
+                # 读取素材 metadata 中的场景切换点缓存（后台 SceneChange 检测写入）：
+                # 有缓存的素材片段起点从随机镜头段选取，无缓存走随机起点兜底
+                asset_scene_points = self._fetch_asset_scene_points(asset_ids)
 
         distribute_assets(
             clips,
@@ -232,12 +240,29 @@ class PlanGeneratorService:
             editing_mode,
             random_selection=random_selection,
             asset_durations=asset_durations,
+            asset_scene_points=asset_scene_points,
         )
 
-    def _sort_assets_by_smart_score(self, asset_ids: List[str]) -> List[str]:
-        """按 smart_match 综合评分降序排列素材 ID。
+    def _fetch_asset_scene_points(self, asset_ids: List[str]) -> dict[str, list[float]]:
+        """从素材 metadata 读取场景切换点缓存（无缓存的素材不包含在结果中）。"""
+        points_map: dict[str, list[float]] = {}
+        if not self._asset_repo:
+            return points_map
+        for asset_id in asset_ids:
+            asset = self._asset_repo.get(asset_id)
+            if asset:
+                points = extract_scene_points_from_metadata(getattr(asset, "metadata", None))
+                if points:
+                    points_map[asset_id] = points
+        return points_map
 
-        评分高的素材（质量好、时长合适、新鲜、使用次数少）排在前面。
+    def _sort_assets_by_smart_score(self, asset_ids: List[str]) -> List[str]:
+        """按 smart_match 综合评分降序排列素材 ID（注入随机噪声）。
+
+        评分高的素材（质量好、时长合适、新鲜、使用次数少）倾向排在前面；
+        排序时给每个素材的得分注入 0~SCORE_RANDOM_NOISE_MAX 的随机噪声，
+        使得分接近的素材排名每次浮动，避免一键生成反复选出相同素材组合，
+        从素材组合层面降低成片查重率。分差大于噪声上限时排名保持稳定。
         """
         scored: list[tuple[str, float]] = []
         for asset_id in asset_ids:
@@ -247,8 +272,11 @@ class PlanGeneratorService:
                 scored.append((asset_id, score))
             else:
                 scored.append((asset_id, 0.0))
-        # 按评分降序排列
-        scored.sort(key=lambda x: x[1], reverse=True)
+        # 评分 + 随机噪声后按降序排列
+        scored.sort(
+            key=lambda x: x[1] + random.uniform(0.0, SCORE_RANDOM_NOISE_MAX),
+            reverse=True,
+        )
         return [aid for aid, _ in scored]
 
     def _fetch_asset_durations(self, asset_ids: List[str]) -> dict[str, float]:

packages/domain/plan_generator_utils.py

@@ -27,6 +27,137 @@ DEFAULT_INTRO_DURATION = 3.0
 DEFAULT_OUTRO_DURATION = 3.0
 
 
+# ── SceneChange 镜头段工具 ────────────────────────────────────────────────────
+
+
+def build_scene_segments(
+    scene_changes: list[float],
+    asset_duration: float,
+) -> list[tuple[float, float]]:
+    """根据场景切换点构建镜头段列表.
+
+    Args:
+        scene_changes: 场景切换点时间戳列表（已排序，首位为 0.0）
+        asset_duration: 素材总时长
+
+    Returns:
+        镜头段列表 [(start, end), ...]，仅保留长度 >= 0.5s 的段
+    """
+    segments: list[tuple[float, float]] = []
+    for i, ts in enumerate(scene_changes):
+        end = scene_changes[i + 1] if i + 1 < len(scene_changes) else asset_duration
+        # 只保留有效长度的镜头段（至少 0.5 秒）
+        if end - ts >= 0.5:
+            segments.append((ts, end))
+    return segments
+
+
+def pick_start_in_scene_segment(
+    seg_start: float,
+    seg_end: float,
+    clip_duration: float,
+) -> float | None:
+    """在镜头段内随机选取一个起始时间点.
+
+    确保 start + clip_duration <= seg_end。
+    若镜头段长度不足以容纳片段，返回 None。
+    """
+    available = seg_end - seg_start - clip_duration
+    if available < 0:
+        return None
+    max_start = seg_start + available
+    return random.uniform(seg_start, max_start)
+
+
+def _segments_overlap(
+    start: float,
+    duration: float,
+    used: list[tuple[float, float]],
+    edge_gap: float = 0.0,
+) -> bool:
+    """候选区间 [start, start+duration] 是否与已用区间冲突（含边缘间隙扩边）。"""
+    end = start + duration
+    for used_start, used_end in used:
+        if start < used_end + edge_gap and end > used_start - edge_gap:
+            return True
+    return False
+
+
+def pick_scene_aware_start(
+    asset_id: str,
+    clip_duration: float,
+    asset_durations: dict[str, float],
+    asset_scene_points: dict[str, list[float]] | None,
+    used_segments: dict[str, list[tuple[float, float]]],
+    *,
+    edge_gap: float = 0.0,
+) -> float | None:
+    """基于缓存的场景切换点，从随机镜头段中选取不冲突的起始时间.
+
+    流程：
+    1. 读取 asset_scene_points 中该素材的场景切换点缓存 → 构建镜头段
+    2. random.shuffle 镜头段（保证同一素材多次生成选不同镜头，而非固定第N段）
+    3. 依次尝试：段内随机取点 → 越界检查 → 与 used_segments 冲突检查
+    4. 全部冲突/无缓存 → 返回 None，由调用方回退 _calc_random_start_time
+
+    Args:
+        asset_id: 素材 ID
+        clip_duration: 片段时长（秒）
+        asset_durations: 素材 ID -> 总时长
+        asset_scene_points: 素材 ID -> 场景切换点列表（metadata 缓存）
+        used_segments: 素材 ID -> 已用区间列表（冲突避让）
+        edge_gap: 冲突判定的边缘间隙（秒），已用区间按 [s-gap, e+gap] 扩边
+    """
+    asset_total = (asset_durations or {}).get(asset_id)
+    if not asset_total or asset_total <= 0:
+        return None
+    scene_points = (asset_scene_points or {}).get(asset_id)
+    if not scene_points:
+        return None
+    used = used_segments.get(asset_id, []) if used_segments else []
+
+    scene_segments = build_scene_segments(scene_points, asset_total)
+    if not scene_segments:
+        return None
+    random.shuffle(scene_segments)
+
+    for seg_start, seg_end in scene_segments:
+        candidate = pick_start_in_scene_segment(seg_start, seg_end, clip_duration)
+        if candidate is None:
+            continue
+        # 越界检查（防御：场景点末尾段理论上不越界，metadata 脏数据兜底）
+        if candidate + clip_duration > asset_total:
+            continue
+        # 与已用区间冲突检查
+        if _segments_overlap(candidate, clip_duration, used, edge_gap):
+            continue
+        return candidate
+
+    return None
+
+
+def extract_scene_points_from_metadata(metadata: object) -> list[float] | None:
+    """从素材 metadata 中提取并校验场景切换点缓存.
+
+    合法缓存：list 类型、至少 2 个数值点、单调非负；否则返回 None（按未缓存处理）。
+    """
+    if not isinstance(metadata, dict):
+        return None
+    points = metadata.get("scene_change_points")
+    if not isinstance(points, list) or len(points) < 2:
+        return None
+    try:
+        cleaned = [float(p) for p in points]
+    except (TypeError, ValueError):
+        return None
+    if any(p < 0 for p in cleaned):
+        return None
+    cleaned = sorted(cleaned)
+    if cleaned[0] != 0.0:
+        cleaned.insert(0, 0.0)
+    return cleaned
+
+
 # ── 素材分配 ────────────────────────────────────────────────────────────────
 
 
@@ -37,6 +168,7 @@ def distribute_assets(
     *,
     random_selection: bool = False,
     asset_durations: dict[str, float] | None = None,
+    asset_scene_points: dict[str, list[float]] | None = None,
 ) -> None:
     """按 editing_mode 将素材分配到 clips（就地修改）.
 
@@ -46,12 +178,16 @@ def distribute_assets(
     - VOICE_OVER: 素材→main clips (B-roll)
     - VOICE_PIP: 第1个→background, 第2个→corner_voice, 其余→b_roll
 
+    start_time 选取：素材 metadata 中有场景切换点缓存时，优先从随机镜头段
+    取起点（不同片段来自不同镜头）；无缓存或镜头段都冲突时回退随机起点。
+
     Args:
         clips: 剪辑片段列表（就地修改 asset_id）
         asset_ids: 素材 ID 列表
         editing_mode: 剪辑模式字符串
         random_selection: 是否随机选择素材（用于预览生成）
-        asset_durations: 素材 ID -> 时长（秒）映射，用于设置随机 start_time
+        asset_durations: 素材 ID -> 时长（秒）映射，用于设置 start_time
+        asset_scene_points: 素材 ID -> 场景切换点列表（metadata 缓存）
     """
     if not asset_ids or not clips:
         return
@@ -62,22 +198,56 @@ def distribute_assets(
         random.shuffle(asset_ids)
 
     if editing_mode == EditingMode.ONE_TAKE.value:
-        _distribute_one_take(clips, asset_ids, asset_durations)
+        _distribute_one_take(clips, asset_ids, asset_durations, asset_scene_points)
     elif editing_mode == EditingMode.PIP.value:
-        _distribute_pip(clips, asset_ids, asset_durations)
+        _distribute_pip(clips, asset_ids, asset_durations, asset_scene_points)
     elif editing_mode == EditingMode.VOICE_OVER.value:
-        _distribute_voice_over(clips, asset_ids, asset_durations)
+        _distribute_voice_over(clips, asset_ids, asset_durations, asset_scene_points)
     elif editing_mode == EditingMode.VOICE_PIP.value:
-        _distribute_voice_pip(clips, asset_ids, asset_durations)
+        _distribute_voice_pip(clips, asset_ids, asset_durations, asset_scene_points)
     else:
         # 未知模式，退化为 one_take
-        _distribute_one_take(clips, asset_ids, asset_durations)
+        _distribute_one_take(clips, asset_ids, asset_durations, asset_scene_points)
+
+
+def _resolve_start_time(
+    asset_id: str,
+    clip_duration: float,
+    asset_durations: dict[str, float] | None,
+    used_segments: dict[str, list[tuple[float, float]]],
+    asset_scene_points: dict[str, list[float]] | None = None,
+    on_exhausted: Callable[[str, float], tuple[float, float] | None] | None = None,
+) -> float | None:
+    """选取片段起点：场景缓存优先（随机镜头段），无缓存/全冲突回退随机起点.
+
+    场景路径与随机路径共享 used_segments 冲突避让；场景路径返回 None 时
+    （无缓存、镜头段全冲突）回退 _calc_random_start_time，其受控复用逻辑
+    （on_exhausted）不受影响。
+    """
+    if asset_scene_points and asset_scene_points.get(asset_id):
+        scene_start = pick_scene_aware_start(
+            asset_id,
+            clip_duration,
+            asset_durations or {},
+            asset_scene_points,
+            used_segments,
+        )
+        if scene_start is not None:
+            return scene_start
+    return _calc_random_start_time(
+        asset_id,
+        clip_duration,
+        asset_durations,
+        used_segments,
+        on_exhausted=on_exhausted,
+    )
 
 
 def _distribute_one_take(
     clips: List[EditPlanClip],
     asset_ids: List[str],
     asset_durations: dict[str, float] | None = None,
+    asset_scene_points: dict[str, list[float]] | None = None,
 ) -> None:
     """ONE_TAKE: 素材按顺序依次分配给 main 类型 clips."""
     used_segments: dict[str, list[tuple[float, float]]] = {}
@@ -85,7 +255,9 @@ def _distribute_one_take(
     for i, clip in enumerate(main_clips):
         if i < len(asset_ids):
             asset_id = asset_ids[i]
-            start_time = _calc_random_start_time(asset_id, clip.duration, asset_durations, used_segments)
+            start_time = _resolve_start_time(
+                asset_id, clip.duration, asset_durations, used_segments, asset_scene_points
+            )
             clip.assign_asset(asset_id, start_time=start_time)
             # Record used segment
             if start_time is not None and asset_durations is not None:
@@ -98,6 +270,7 @@ def _distribute_pip(
     clips: List[EditPlanClip],
     asset_ids: List[str],
     asset_durations: dict[str, float] | None = None,
+    asset_scene_points: dict[str, list[float]] | None = None,
 ) -> None:
     """PIP: 第1个素材→main(全屏背景)，其余→overlay clips."""
     used_segments: dict[str, list[tuple[float, float]]] = {}
@@ -105,7 +278,9 @@ def _distribute_pip(
     main_clips = [c for c in clips if c.clip_type == ClipType.MAIN.value]
     if main_clips and asset_ids:
         asset_id = asset_ids[0]
-        start_time = _calc_random_start_time(asset_id, main_clips[0].duration, asset_durations, used_segments)
+        start_time = _resolve_start_time(
+            asset_id, main_clips[0].duration, asset_durations, used_segments, asset_scene_points
+        )
         main_clips[0].assign_asset(asset_id, start_time=start_time)
         # Record used segment
         if start_time is not None and asset_durations is not None:
@@ -119,7 +294,9 @@ def _distribute_pip(
     for i, clip in enumerate(overlay_clips):
         if i < len(remaining):
             asset_id = remaining[i]
-            start_time = _calc_random_start_time(asset_id, clip.duration, asset_durations, used_segments)
+            start_time = _resolve_start_time(
+                asset_id, clip.duration, asset_durations, used_segments, asset_scene_points
+            )
             clip.assign_asset(asset_id, start_time=start_time)
             # Record used segment
             if start_time is not None and asset_durations is not None:
@@ -132,6 +309,7 @@ def _distribute_voice_over(
     clips: List[EditPlanClip],
     asset_ids: List[str],
     asset_durations: dict[str, float] | None = None,
+    asset_scene_points: dict[str, list[float]] | None = None,
 ) -> None:
     """VOICE_OVER: 素材→main clips (B-roll)."""
     used_segments: dict[str, list[tuple[float, float]]] = {}
@@ -139,7 +317,9 @@ def _distribute_voice_over(
     for i, clip in enumerate(main_clips):
         if i < len(asset_ids):
             asset_id = asset_ids[i]
-            start_time = _calc_random_start_time(asset_id, clip.duration, asset_durations, used_segments)
+            start_time = _resolve_start_time(
+                asset_id, clip.duration, asset_durations, used_segments, asset_scene_points
+            )
             clip.assign_asset(asset_id, start_time=start_time)
             # Record used segment
             if start_time is not None and asset_durations is not None:
@@ -152,6 +332,7 @@ def _distribute_voice_pip(
     clips: List[EditPlanClip],
     asset_ids: List[str],
     asset_durations: dict[str, float] | None = None,
+    asset_scene_points: dict[str, list[float]] | None = None,
 ) -> None:
     """VOICE_PIP: 第1个→background, 第2个→corner_voice, 其余→b_roll."""
     used_segments: dict[str, list[tuple[float, float]]] = {}
@@ -164,7 +345,9 @@ def _distribute_voice_pip(
     # 第1个 → background
     if idx < len(asset_ids) and bg_clips:
         asset_id = asset_ids[idx]
-        start_time = _calc_random_start_time(asset_id, bg_clips[0].duration, asset_durations, used_segments)
+        start_time = _resolve_start_time(
+            asset_id, bg_clips[0].duration, asset_durations, used_segments, asset_scene_points
+        )
         bg_clips[0].assign_asset(asset_id, start_time=start_time)
         # Record used segment
         if start_time is not None and asset_durations is not None:
@@ -176,7 +359,9 @@ def _distribute_voice_pip(
     # 第2个 → corner_voice
     if idx < len(asset_ids) and voice_clips:
         asset_id = asset_ids[idx]
-        start_time = _calc_random_start_time(asset_id, voice_clips[0].duration, asset_durations, used_segments)
+        start_time = _resolve_start_time(
+            asset_id, voice_clips[0].duration, asset_durations, used_segments, asset_scene_points
+        )
         voice_clips[0].assign_asset(asset_id, start_time=start_time)
         # Record used segment
         if start_time is not None and asset_durations is not None:
@@ -190,7 +375,9 @@ def _distribute_voice_pip(
     for i, clip in enumerate(broll_clips):
         if i < len(remaining):
             asset_id = remaining[i]
-            start_time = _calc_random_start_time(asset_id, clip.duration, asset_durations, used_segments)
+            start_time = _resolve_start_time(
+                asset_id, clip.duration, asset_durations, used_segments, asset_scene_points
+            )
             clip.assign_asset(asset_id, start_time=start_time)
             # Record used segment
             if start_time is not None and asset_durations is not None:

packages/domain/smart_match.py

@@ -14,6 +14,13 @@ from dataclasses import dataclass, field
 from datetime import datetime, timezone
 from typing import Any
 
+# 素材选取排序时注入的随机噪声上限（分）。
+# score_asset 综合得分范围为 0-100，噪声 0~20 意味着：
+# - 素材间得分差距 > 20 分时，排名不受影响（质量差异显著的素材保持稳定优先级）
+# - 得分接近（差距 <= 20 分）的素材排名会随机浮动，使每次生成选出的素材组合不同，
+#   从素材组合层面降低成片重复率；排名靠后的低分素材也有机会入选。
+SCORE_RANDOM_NOISE_MAX = 20.0
+
 
 @dataclass
 class SmartMatchResult:

tests/unit/test_editor_clips_random_start.py

@@ -38,6 +38,28 @@ def _segments(count: int, dur_min: float = 3.0, dur_max: float = 5.0):
     return [(i, dur_min, dur_max) for i in range(count)]
 
 
+def _patch_zero_noise():
+    """消除 clips.py 排序随机噪声，用于确定性断言（如均衡分配）。
+
+    排序噪声（random.uniform(0, SCORE_RANDOM_NOISE_MAX)）返回 0；
+    其他 uniform 调用（片段时长随机）委托给独立 Random 实例，行为不变。
+    """
+    import random as _stdlib_random
+
+    from app.api.routes.templates_editor import clips as clips_module
+
+    from packages.domain.smart_match import SCORE_RANDOM_NOISE_MAX
+
+    _fallback = _stdlib_random.Random()
+
+    def _fake_uniform(a, b):
+        if b == SCORE_RANDOM_NOISE_MAX:
+            return 0.0
+        return _fallback.uniform(a, b)
+
+    return patch.object(clips_module.random, "uniform", _fake_uniform)
+
+
 def _patch_segments(segments=None):
     return patch(
         "app.api.routes.templates_editor.clips._get_template_segments",
@@ -130,7 +152,8 @@ class TestEditorClipsBySegments:
 
         body = ClipsFromAssetsRequest(asset_ids=["a1", "a2"], required_clips_count=2)
 
-        with _patch_segments(DEFAULT_SEGMENTS):
+        # 均衡分配由 use_count 贪心保证，消除排序噪声后确定性断言
+        with _patch_zero_noise(), _patch_segments(DEFAULT_SEGMENTS):
             result = create_clips_from_assets_editor(
                 template_id="tpl-001",
                 body=body,
@@ -795,7 +818,8 @@ class TestClipsFromAssetsInvalidIds:
 
         body = ClipsFromAssetsRequest(asset_ids=["a1", None, "", "a2"])  # type: ignore[list-item]
 
-        with _patch_segments(_segments(2, dur_min=3.0, dur_max=5.0)):
+        # 消除排序噪声，确定性断言两条合法素材各被使用
+        with _patch_zero_noise(), _patch_segments(_segments(2, dur_min=3.0, dur_max=5.0)):
             result = create_clips_from_assets_editor(
                 template_id="tpl-001",
                 body=body,

tests/unit/test_mediakit_smart_clips.py

@@ -306,6 +306,27 @@ class TestGetTemplateSegments:
 # ── from-assets 端点集成测试 ────────────────────────────────────────────────
 
 
+def _patch_zero_noise():
+    """消除 clips.py 排序随机噪声（SCORE_RANDOM_NOISE_MAX 噪声返回 0）。
+
+    用于均衡分配等确定性断言；其他 uniform 调用（片段时长随机）行为不变。
+    """
+    import random as _stdlib_random
+
+    from app.api.routes.templates_editor import clips as clips_module
+
+    from packages.domain.smart_match import SCORE_RANDOM_NOISE_MAX
+
+    _fallback = _stdlib_random.Random()
+
+    def _fake_uniform(a, b):
+        if b == SCORE_RANDOM_NOISE_MAX:
+            return 0.0
+        return _fallback.uniform(a, b)
+
+    return patch.object(clips_module.random, "uniform", _fake_uniform)
+
+
 def _make_auth_user():
     auth = MagicMock()
     auth.user.id = "user-001"
@@ -470,16 +491,18 @@ class TestFromAssetsByTemplateSegments:
         mock_asset_repo.get.side_effect = get_asset
 
         body = ClipsFromAssetsRequest(asset_ids=["a1", "a2"])
-        create_clips_from_assets_editor(
-            template_id="tmpl-1",
-            body=body,
-            background_tasks=MagicMock(),
-            plan_id="plan-1",
-            services=(mock_tpl_svc, mock_plan_svc),
-            asset_repo=mock_asset_repo,
-            db=MagicMock(),
-            current_user=_make_auth_user(),
-        )
+        # 消除排序噪声，确定性断言贪心均衡分配
+        with _patch_zero_noise():
+            create_clips_from_assets_editor(
+                template_id="tmpl-1",
+                body=body,
+                background_tasks=MagicMock(),
+                plan_id="plan-1",
+                services=(mock_tpl_svc, mock_plan_svc),
+                asset_repo=mock_asset_repo,
+                db=MagicMock(),
+                current_user=_make_auth_user(),
+            )
 
         clips_data = _get_clips_data(mock_plan_svc)
         asset_ids = [c["asset_id"] for c in clips_data]

tests/unit/test_scene_change_prerender.py

@@ -0,0 +1,368 @@
+"""测试 SceneChange 场景检测前置到渲染前 + 场景点缓存读写。
+
+验证：
+- 场景点缓存读取（extract_scene_points_from_metadata）：合法/非法/脏数据
+- pick_scene_aware_start：随机镜头段选取、冲突避让、shuffle 随机化、无缓存回退 None
+- from-assets 路径：metadata 有 scene_change_points 时，start_time 落在镜头段内
+- 一键生成路径：distribute_assets 传入 asset_scene_points 时使用镜头段
+- 后台任务：检测结果写入素材 metadata（缓存）
+"""
+
+from __future__ import annotations
+
+import os
+import sys
+from pathlib import Path
+from unittest.mock import MagicMock, patch
+
+os.environ.setdefault("JWT_SECRET_KEY", "unit-test-secret-key-for-testing")
+os.environ.setdefault("DATABASE_URL", "sqlite:///test.db")
+
+sys.path.insert(0, str(Path(__file__).resolve().parents[2] / "apps" / "api"))
+
+import pytest
+
+from packages.domain.plan_generator_utils import (
+    _calc_random_start_time,
+    build_scene_segments,
+    distribute_assets,
+    extract_scene_points_from_metadata,
+    pick_scene_aware_start,
+)
+
+# ── metadata 缓存解析 ─────────────────────────────────────────────────────────
+
+
+class TestExtractScenePoints:
+    def test_valid_points(self):
+        md = {"scene_change_points": [0.0, 3.2, 7.8, 12.5]}
+        assert extract_scene_points_from_metadata(md) == [0.0, 3.2, 7.8, 12.5]
+
+    def test_missing_returns_none(self):
+        assert extract_scene_points_from_metadata({}) is None
+        assert extract_scene_points_from_metadata(None) is None
+        assert extract_scene_points_from_metadata("not-a-dict") is None
+
+    def test_empty_list_returns_none(self):
+        assert extract_scene_points_from_metadata({"scene_change_points": []}) is None
+        assert extract_scene_points_from_metadata({"scene_change_points": [0.0]}) is None
+
+    def test_dirty_data_returns_none(self):
+        assert extract_scene_points_from_metadata({"scene_change_points": ["a", 1.0]}) is None
+
+    def test_auto_prepends_zero(self):
+        result = extract_scene_points_from_metadata({"scene_change_points": [3.2, 7.8]})
+        assert result == [0.0, 3.2, 7.8]
+
+    def test_sorts_unsorted(self):
+        result = extract_scene_points_from_metadata({"scene_change_points": [0.0, 12.5, 3.2, 7.8]})
+        assert result == [0.0, 3.2, 7.8, 12.5]
+
+    def test_negative_rejected(self):
+        assert extract_scene_points_from_metadata({"scene_change_points": [-1.0, 3.2]}) is None
+
+
+# ── 镜头段选取 ────────────────────────────────────────────────────────────────
+
+
+class TestPickSceneAwareStart:
+    def test_start_within_some_scene_segment(self):
+        """有缓存时，起点落在某个镜头段内部。"""
+        points = [0.0, 5.0, 10.0, 15.0]
+        durations = {"a1": 18.0}
+        scene_points = {"a1": points}
+        used: dict = {}
+        start = pick_scene_aware_start("a1", 3.0, durations, scene_points, used)
+        assert start is not None
+        segments = build_scene_segments(points, 18.0)
+        assert any(seg_start <= start and start + 3.0 <= seg_end for seg_start, seg_end in segments)
+
+    def test_no_cache_returns_none(self):
+        """无缓存返回 None（调用方回退随机起点）。"""
+        result = pick_scene_aware_start("a1", 3.0, {"a1": 18.0}, {}, {})
+        assert result is None
+
+    def test_bounds_respected(self):
+        """起点 + 片段时长不超过素材总时长。"""
+        points = [0.0, 5.0, 10.0, 15.0]
+        for _ in range(30):
+            start = pick_scene_aware_start("a1", 4.0, {"a1": 18.0}, {"a1": points}, {})
+            assert start is not None
+            assert start + 4.0 <= 18.0 + 1e-6
+
+    def test_conflict_avoidance(self):
+        """所有镜头段都被占满时返回 None（回退随机路径）。"""
+        # 3 段各 6s，片段 5s；把所有段占满（[0,5.5] [5.5,11] 覆盖段1/2，段3太短放不下5s）
+        points = [0.0, 6.0, 12.0]
+        used = {"a1": [(0.0, 5.6), (6.0, 11.6)]}
+        # 段3 [12, 18] 可用 → 应返回其中起点
+        start = pick_scene_aware_start("a1", 5.0, {"a1": 18.0}, {"a1": points}, used)
+        assert start is not None
+        assert start >= 12.0
+
+    def test_all_segments_conflict_returns_none(self):
+        """全部镜头段都冲突时返回 None。"""
+        points = [0.0, 6.0, 12.0]
+        # 占满整个素材
+        used = {"a1": [(0.0, 18.0)]}
+        start = pick_scene_aware_start("a1", 5.0, {"a1": 18.0}, {"a1": points}, used)
+        assert start is None
+
+    def test_shuffle_produces_varied_segments(self):
+        """镜头段顺序被 shuffle：30 次选取，起点分布应覆盖多个镜头段。"""
+        points = [0.0, 5.0, 10.0, 15.0]
+        observed: set[int] = set()
+        for _ in range(40):
+            start = pick_scene_aware_start("a1", 2.0, {"a1": 18.0}, {"a1": points}, {})
+            assert start is not None
+            # 记录起点落在哪个段（段宽 5s）
+            observed.add(int(start // 5.0))
+        assert len(observed) >= 3, f"镜头段 shuffle 后应覆盖多个段，实际 {observed}"
+
+
+# ── 一键生成路径：distribute_assets 接入场景缓存 ──────────────────────────────
+
+
+class TestDistributeWithScenePoints:
+    def _make_clips(self, n):
+        from packages.domain.edit_plan_clip import EditPlanClip
+
+        return [EditPlanClip(id=f"c{i}", plan_id="p1", clip_type="main", duration=4.0, order=i) for i in range(n)]
+
+    def test_one_take_uses_scene_segments(self):
+        """ONE_TAKE 模式下，有场景缓存的素材起点落在镜头段内。"""
+        from packages.domain.editing_mode import EditingMode
+
+        clips = self._make_clips(2)
+        points = [0.0, 6.0, 12.0, 18.0]
+        distribute_assets(
+            clips,
+            ["a1"],
+            EditingMode.ONE_TAKE.value,
+            asset_durations={"a1": 24.0},
+            asset_scene_points={"a1": points},
+        )
+        segments = build_scene_segments(points, 24.0)
+        for clip in clips:
+            assert clip.start_time is not None
+            assert any(
+                s <= clip.start_time and clip.start_time + 4.0 <= e for s, e in segments
+            ), f"起点 {clip.start_time} 不在任何镜头段内"
+
+    def test_no_scene_points_falls_back_random(self):
+        """无场景缓存时正常分配（回退随机起点），不报错。"""
+        from packages.domain.editing_mode import EditingMode
+
+        clips = self._make_clips(1)
+        distribute_assets(
+            clips,
+            ["a1"],
+            EditingMode.ONE_TAKE.value,
+            asset_durations={"a1": 24.0},
+            asset_scene_points={},
+        )
+        for clip in clips:
+            assert clip.asset_id == "a1"
+            assert clip.start_time is not None
+            assert 0.0 <= clip.start_time <= 20.0
+
+    def test_random_preview_ignores_scene_points(self):
+        """random_selection 预览模式行为不变（不崩溃、正常分配）。"""
+        from packages.domain.editing_mode import EditingMode
+
+        clips = self._make_clips(2)
+        distribute_assets(
+            clips,
+            ["a1", "a2"],
+            EditingMode.ONE_TAKE.value,
+            random_selection=True,
+            asset_durations={"a1": 24.0, "a2": 24.0},
+            asset_scene_points={"a1": [0.0, 6.0]},
+        )
+        assert all(c.asset_id for c in clips)
+
+
+# ── from-assets 路径：渲染前读缓存选镜头段 ────────────────────────────────────
+
+
+def _make_auth_user():
+    auth = MagicMock()
+    auth.user.id = "user-001"
+    auth.user.email = "test@example.com"
+    auth.user.display_name = "测试用户"
+    auth.user_id = "user-001"
+    return auth
+
+
+def _make_asset_with_scenes(aid, duration, scene_points=None):
+    asset = MagicMock()
+    asset.id = aid
+    asset.duration = duration
+    asset.quality_score = None
+    asset.created_at = None
+    asset.metadata = {"scene_change_points": scene_points} if scene_points is not None else {}
+    return asset
+
+
+class TestFromAssetsSceneCache:
+    def test_cached_scene_points_used_for_start_time(self):
+        """素材 metadata 有场景点缓存时，片段起点落在镜头段内。"""
+        from app.api.routes.templates_editor.clips import create_clips_from_assets_editor
+        from app.api.routes.templates_editor.schemas import ClipsFromAssetsRequest
+
+        # 镜头段宽度 10s，片段最长 ~5.3s（含转场补偿），每段都能容纳
+        scene_points = [0.0, 10.0, 20.0, 30.0]
+        asset_duration = 40.0
+        mock_asset_repo = MagicMock()
+        mock_asset_repo.get = MagicMock(
+            side_effect=lambda aid: _make_asset_with_scenes(aid, asset_duration, scene_points)
+        )
+        mock_plan_svc = MagicMock()
+        mock_plan_svc.replace_all_clips_transactional = MagicMock(return_value=3)
+
+        segments = [(0, 3.0, 5.0), (1, 3.0, 5.0), (2, 3.0, 5.0)]
+
+        with (
+            patch(
+                "app.api.routes.templates_editor.clips._get_template_segments",
+                return_value=segments,
+            ),
+            patch("app.api.routes.templates_editor.clips.get_used_segments", return_value={}),
+            patch("app.api.routes.templates_editor.clips.record_used_segments", return_value=None),
+        ):
+            body = ClipsFromAssetsRequest(asset_ids=["a1"], required_clips_count=3)
+            create_clips_from_assets_editor(
+                template_id="tmpl-1",
+                body=body,
+                background_tasks=MagicMock(),
+                plan_id="plan-scene-1",
+                services=(MagicMock(), mock_plan_svc),
+                asset_repo=mock_asset_repo,
+                db=MagicMock(),
+                current_user=_make_auth_user(),
+            )
+
+        clips_data = mock_plan_svc.replace_all_clips_transactional.call_args.args[1]
+        scene_segments = build_scene_segments(scene_points, asset_duration)
+        for clip in clips_data:
+            start = clip["start_time"]
+            dur = clip["duration"]
+            in_segment = any(s <= start and start + dur <= e + 0.1 for s, e in scene_segments)
+            assert in_segment, f"起点 {start:.2f} 时长 {dur:.2f} 不在任何镜头段内"
+
+    def test_no_cache_falls_back_random_no_error(self):
+        """素材无场景缓存时正常走随机起点，流程不报错。"""
+        from app.api.routes.templates_editor.clips import create_clips_from_assets_editor
+        from app.api.routes.templates_editor.schemas import ClipsFromAssetsRequest
+
+        mock_asset_repo = MagicMock()
+        mock_asset_repo.get = MagicMock(side_effect=lambda aid: _make_asset_with_scenes(aid, 30.0, None))
+        mock_plan_svc = MagicMock()
+        mock_plan_svc.replace_all_clips_transactional = MagicMock(return_value=2)
+
+        segments = [(0, 3.0, 5.0), (1, 3.0, 5.0)]
+        with (
+            patch(
+                "app.api.routes.templates_editor.clips._get_template_segments",
+                return_value=segments,
+            ),
+            patch("app.api.routes.templates_editor.clips.get_used_segments", return_value={}),
+            patch("app.api.routes.templates_editor.clips.record_used_segments", return_value=None),
+        ):
+            body = ClipsFromAssetsRequest(asset_ids=["a1"], required_clips_count=2)
+            create_clips_from_assets_editor(
+                template_id="tmpl-1",
+                body=body,
+                background_tasks=MagicMock(),
+                plan_id="plan-scene-2",
+                services=(MagicMock(), mock_plan_svc),
+                asset_repo=mock_asset_repo,
+                db=MagicMock(),
+                current_user=_make_auth_user(),
+            )
+        clips_data = mock_plan_svc.replace_all_clips_transactional.call_args.args[1]
+        assert len(clips_data) == 2
+        assert all(c["start_time"] is not None for c in clips_data)
+
+
+# ── 后台任务：检测结果写缓存 ──────────────────────────────────────────────────
+
+
+class TestAsyncCacheWrite:
+    def test_scene_points_persisted_to_metadata(self):
+        """detect_scene_changes 返回结果后写入素材 metadata 并调用 repo.update。"""
+        from app.api.routes.templates_editor import clips as clips_module
+
+        detected_points = [0.0, 4.5, 9.0, 14.2]
+
+        mock_asset = MagicMock()
+        mock_asset.id = "a1"
+        mock_asset.duration = 20.0
+        mock_asset.storage_key = "v.mp4"
+        mock_asset.mime_type = "video/mp4"
+        mock_asset.metadata = {}
+
+        mock_asset_repo = MagicMock()
+        mock_asset_repo.find_by_ids = MagicMock(return_value=[mock_asset])
+        mock_asset_repo.update = MagicMock(side_effect=lambda a: a)
+
+        mock_clip = MagicMock()
+        mock_clip.id = "clip-1"
+        mock_clip.asset_id = "a1"
+        mock_clip.order = 0
+        mock_clip.start_time = 2.0
+        mock_clip.duration = 4.0
+
+        mock_plan_svc = MagicMock()
+        mock_plan_svc.list_clips = MagicMock(return_value=[mock_clip])
+        mock_plan_svc.update_clip = MagicMock()
+        plan_svc_factory = MagicMock(return_value=mock_plan_svc)
+
+        mock_client = MagicMock()
+        mock_client.is_available = True
+        mock_client.detect_scene_changes = MagicMock(return_value=detected_points)
+
+        mock_storage = MagicMock()
+        mock_storage.get_download_url = MagicMock(return_value="https://example.com/v.mp4")
+
+        mock_session = MagicMock()
+
+        with (
+            patch(
+                "packages.adapters.sqlalchemy_impl.asset_repository.SQLAlchemyAssetRepository",
+                return_value=mock_asset_repo,
+            ),
+            patch(
+                "app.api.routes.templates_editor.clips.EditPlanService",
+                plan_svc_factory,
+            ),
+            patch(
+                "packages.adapters.sqlalchemy_impl.session.SessionLocal",
+                MagicMock(return_value=mock_session),
+            ),
+            patch(
+                "app.api.routes.templates_editor.clips.get_storage_service",
+                return_value=mock_storage,
+            ),
+            patch(
+                "app.api.routes.templates_editor.clips.get_mediakit_client",
+                return_value=mock_client,
+            ),
+            patch(
+                "app.api.routes.templates_editor.clips.get_used_segments",
+                return_value={},
+            ),
+            patch(
+                "app.api.routes.templates_editor.clips.record_used_segments",
+                return_value=None,
+            ),
+            patch(
+                "app.api.routes.templates_editor.clips.remove_used_segment",
+                return_value=False,
+            ),
+        ):
+            clips_module._update_mediakit_recommendations_async("plan-1", ["a1"])
+
+        # 验证素材 metadata 被写入场景点并持久化
+        assert mock_asset.metadata.get("scene_change_points") == detected_points
+        mock_asset_repo.update.assert_called()

tests/unit/test_smart_match_integration.py

@@ -23,7 +23,7 @@ sys.path.insert(0, str(Path(__file__).resolve().parents[2] / "apps" / "api"))
 
 import pytest
 
-from packages.domain.smart_match import score_asset, smart_select_assets
+from packages.domain.smart_match import SCORE_RANDOM_NOISE_MAX, score_asset, smart_select_assets
 
 # ── 辅助工厂 ──────────────────────────────────────────────────────────────────
 
@@ -158,6 +158,38 @@ def _make_auth_user():
     return auth
 
 
+def _make_zero_noise_patcher(module):
+    """构造 patch(module.random.uniform)：噪声调用(上界=SCORE_RANDOM_NOISE_MAX)返回 0。
+
+    其他 uniform 调用（如片段时长随机）委托给一个独立的 Random 实例，
+    避免递归回已 patch 的全局函数。
+    """
+    import random as _stdlib_random
+
+    _fallback = _stdlib_random.Random()
+
+    def _fake_uniform(a, b):
+        if b == SCORE_RANDOM_NOISE_MAX:
+            return 0.0
+        return _fallback.uniform(a, b)
+
+    return patch.object(module.random, "uniform", _fake_uniform)
+
+
+def _patch_zero_noise_clips():
+    """消除 clips.py 排序噪声，其他 uniform 调用不受影响。"""
+    from app.api.routes.templates_editor import clips as clips_module
+
+    return _make_zero_noise_patcher(clips_module)
+
+
+def _patch_zero_noise_plan_service():
+    """消除 plan_generator_service.py 排序噪声，其他 uniform 调用不受影响。"""
+    from app.services import plan_generator_service as svc_module
+
+    return _make_zero_noise_patcher(svc_module)
+
+
 class TestFromAssetsSmartMatchIntegration:
     """验证 clips.py 中 sorted_candidates 使用 smart_match 评分。"""
 
@@ -180,6 +212,7 @@ class TestFromAssetsSmartMatchIntegration:
         segments = [(0, 3.0, 5.0), (1, 3.0, 5.0)]
 
         with (
+            _patch_zero_noise_clips(),
             patch(
                 "app.api.routes.templates_editor.clips._get_template_segments",
                 return_value=segments,
@@ -219,6 +252,63 @@ class TestFromAssetsSmartMatchIntegration:
             first_clip_asset == "a_fresh"
         ), f"第一个片段应分配给 smart_match 分更高的 a_fresh，实际是 {first_clip_asset}"
 
+    def test_score_noise_causes_varied_selection(self):
+        """得分接近（差距 < SCORE_RANDOM_NOISE_MAX）的素材，多次生成的素材组合应有变化。
+
+        两条同等质量/时长/新鲜度的素材（use_count 相同），smart_match 得分一致，
+        噪声让两者的相对排名随机浮动，多次调用首个片段的素材分布应两者都出现。
+        """
+        from app.api.routes.templates_editor.clips import create_clips_from_assets_editor
+        from app.api.routes.templates_editor.schemas import ClipsFromAssetsRequest
+
+        def _get_asset(aid):
+            return _make_mock_asset_for_clips(aid, 30.0, 0)
+
+        mock_asset_repo = MagicMock()
+        mock_asset_repo.get = MagicMock(side_effect=_get_asset)
+
+        segments = [(0, 3.0, 5.0), (1, 3.0, 5.0)]
+
+        first_assets: set[str] = set()
+        for _ in range(30):
+            mock_plan_svc = MagicMock()
+            mock_plan_svc.replace_all_clips_transactional = MagicMock(return_value=2)
+            with (
+                patch(
+                    "app.api.routes.templates_editor.clips._get_template_segments",
+                    return_value=segments,
+                ),
+                patch(
+                    "app.api.routes.templates_editor.clips.get_used_segments",
+                    return_value={},
+                ),
+                patch(
+                    "app.api.routes.templates_editor.clips.record_used_segments",
+                    return_value=None,
+                ),
+            ):
+                body = ClipsFromAssetsRequest(
+                    asset_ids=["a_x", "a_y"],
+                    required_clips_count=2,
+                )
+                create_clips_from_assets_editor(
+                    template_id="tmpl-1",
+                    body=body,
+                    background_tasks=MagicMock(),
+                    plan_id=f"plan-noise-{len(first_assets)}-{_}",
+                    services=(MagicMock(), mock_plan_svc),
+                    asset_repo=mock_asset_repo,
+                    db=MagicMock(),
+                    current_user=_make_auth_user(),
+                )
+            clips_data = mock_plan_svc.replace_all_clips_transactional.call_args.args[1]
+            first_assets.add(clips_data[0]["asset_id"])
+
+        assert first_assets == {
+            "a_x",
+            "a_y",
+        }, f"噪声应使两条等分素材的排名浮动，30 次调用首个片段应覆盖两者，实际 {first_assets}"
+
 
 # ── 一键生成路径集成测试 ─────────────────────────────────────────────────────
 
@@ -247,7 +337,8 @@ class TestPlanGeneratorSmartMatchIntegration:
         db = MagicMock()
         svc = PlanGeneratorService(db, asset_repo=mock_asset_repo)
 
-        sorted_ids = svc._sort_assets_by_smart_score(["high_use", "low_use", "mid_use"])
+        with _patch_zero_noise_plan_service():
+            sorted_ids = svc._sort_assets_by_smart_score(["high_use", "low_use", "mid_use"])
 
         # low_use (0次) 应排第一，high_use (10次) 应排最后
         assert sorted_ids[0] == "low_use"
@@ -284,7 +375,10 @@ class TestPlanGeneratorSmartMatchIntegration:
             EditPlanClip(id="c2", plan_id="p1", clip_type="main", duration=5.0, order=1),
         ]
 
-        with patch("app.services.plan_generator_service.distribute_assets") as mock_dist:
+        with (
+            _patch_zero_noise_plan_service(),
+            patch("app.services.plan_generator_service.distribute_assets") as mock_dist,
+        ):
             svc._distribute_assets(
                 clips,
                 ["old_asset", "new_asset"],
@@ -322,3 +416,60 @@ class TestPlanGeneratorSmartMatchIntegration:
             )
             # random_selection=True 时不应调用 asset_repo.get（不执行排序）
             mock_asset_repo.get.assert_not_called()
+
+
+class TestPlanGeneratorScoreNoise:
+    """验证一键生成路径的评分排序注入了随机噪声。"""
+
+    def test_equal_scores_produce_varied_order(self):
+        """两条 smart_match 得分相同的素材，多次排序的首位应覆盖两者。"""
+        from app.services.plan_generator_service import PlanGeneratorService
+
+        def _get_asset(aid):
+            asset = MagicMock()
+            asset.id = aid
+            asset.duration = 15.0
+            asset.quality_score = None
+            asset.created_at = None
+            asset.metadata = {"generation_use_count": 0}
+            return asset
+
+        mock_asset_repo = MagicMock()
+        mock_asset_repo.get = MagicMock(side_effect=_get_asset)
+        svc = PlanGeneratorService(MagicMock(), asset_repo=mock_asset_repo)
+
+        first_ids: set[str] = set()
+        for _ in range(30):
+            order = svc._sort_assets_by_smart_score(["equal_a", "equal_b"])
+            first_ids.add(order[0])
+
+        assert first_ids == {
+            "equal_a",
+            "equal_b",
+        }, f"噪声应使等分素材排名浮动，30 次排序首位应覆盖两者，实际 {first_ids}"
+
+    def test_large_score_gap_not_flipped(self):
+        """得分差距远大于噪声上限时，低分素材不会因噪声超过高分素材。
+
+        quality 100 vs 0 → quality 维度差距 40 分 > 噪声上限 20，
+        其余维度完全一致，50 次排序高质量素材必须始终排第一。
+        """
+        from app.services.plan_generator_service import PlanGeneratorService
+
+        def _get_asset(aid):
+            quality = {"top": 100.0, "bad": 0.0}[aid]
+            asset = MagicMock()
+            asset.id = aid
+            asset.duration = 15.0
+            asset.quality_score = quality
+            asset.created_at = None
+            asset.metadata = {"generation_use_count": 0}
+            return asset
+
+        mock_asset_repo = MagicMock()
+        mock_asset_repo.get = MagicMock(side_effect=_get_asset)
+        svc = PlanGeneratorService(MagicMock(), asset_repo=mock_asset_repo)
+
+        for _ in range(50):
+            order = svc._sort_assets_by_smart_score(["top", "bad"])
+            assert order[0] == "top", f"质量差距 40 分 > 噪声上限，top 应始终排第一，实际 {order}"