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

@@ -47,7 +47,13 @@ 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.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
 
@@ -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 中持久化的历史已用区间（跨任务/跨调用去重），
@@ -771,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(
@@ -869,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],
@@ -1032,15 +1031,45 @@ def _update_mediakit_recommendations_async(  # pragma: no cover
 
             # 优先使用 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),
                     )
+                    # 检测结果写入素材 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:
@@ -1071,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

@@ -30,6 +30,7 @@ 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,
 )
@@ -223,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,
@@ -233,8 +240,22 @@ class PlanGeneratorService:
             editing_mode,
             random_selection=random_selection,
             asset_durations=asset_durations,
+            asset_scene_points=asset_scene_points,
         )
 
+    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（注入随机噪声）。
 

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:

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()