From 93f43a5759b515692572aa125bcd38aa1be42a98 Mon Sep 17 00:00:00 2001 From: saas-backend-agent Date: Wed, 2 Sep 2026 00:58:22 +0800 Subject: [PATCH 1/2] ci: retrigger after ci-base image rebuild -- 2.54.0 From c4068c1d3f8b7a925fccdefbdcb081b90c3d277c Mon Sep 17 00:00:00 2001 From: saas-backend-agent Date: Wed, 2 Sep 2026 01:15:15 +0800 Subject: [PATCH 2/2] =?UTF-8?q?feat:=20SceneChange=20=E5=9C=BA=E6=99=AF?= =?UTF-8?q?=E6=A3=80=E6=B5=8B=E5=89=8D=E7=BD=AE=E5=88=B0=E6=B8=B2=E6=9F=93?= =?UTF-8?q?=E5=89=8D=EF=BC=8C=E5=9C=BA=E6=99=AF=E7=82=B9=E7=BC=93=E5=AD=98?= =?UTF-8?q?=E5=88=B0=E7=B4=A0=E6=9D=90=20metadata?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 现状问题： - 场景检测是接口返回后的后台任务，视频已用随机 start_time 渲染完， SceneChange 更新 DB 对当前成片查重率无作用 - 首次生成的素材没有缓存，渲染前仍走随机起点（后台任务异步补缓存） - 镜头段固定顺序分配（第N个clip固定用第N段），同样素材组合起始帧仍相同 改动： 1. 场景点持久化到素材 metadata（scene_change_points: list[float]） - 后台 _update_mediakit_recommendations_async：检测前查缓存命中跳过， detect_scene_changes 成功后写入 asset.metadata 并 repo.update 2. 渲染前同步读缓存选镜头段（clips.py from-assets 主流程） - 素材加载时 extract_scene_points_from_metadata 读缓存 - 有缓存：build_scene_segments → shuffle 随机镜头段顺序 → pick_scene_aware_start 段内取点 + 越界检查 + used_segments 冲突避让 （含 1.5s 边缘间隙） - 无缓存/全冲突：回退 _calc_random_start_time（受控复用逻辑不变） 3. 一键生成路径同步改造（plan_generator_service + plan_generator_utils） - distribute_assets 新增 asset_scene_points 参数，7 个分配点统一 走 _resolve_start_time（场景优先、随机兜底） - _fetch_asset_scene_points 读 metadata；预览随机模式不读 DB 4. 镜头段工具下沉 domain 层：build_scene_segments / pick_start_in_scene_segment / pick_scene_aware_start / extract_scene_points_from_metadata，clips.py 保留旧名兼容别名 - 镜头段必须 shuffle 后遍历，不按固定顺序分配 测试：新增 19 个测试覆盖缓存解析（合法/脏数据/自动补0）、 段内取点/冲突避让/shuffle 覆盖多段、两条路径集成、后台写缓存。 全量 14032 单测通过。 --- .../app/api/routes/templates_editor/clips.py | 135 ++++--- .../app/services/plan_generator_service.py | 27 +- packages/domain/plan_generator_utils.py | 213 +++++++++- tests/unit/test_scene_change_prerender.py | 368 ++++++++++++++++++ 4 files changed, 674 insertions(+), 69 deletions(-) create mode 100644 tests/unit/test_scene_change_prerender.py diff --git a/apps/api/app/api/routes/templates_editor/clips.py b/apps/api/app/api/routes/templates_editor/clips.py index b26f60368..98dcee836 100755 --- a/apps/api/app/api/routes/templates_editor/clips.py +++ b/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 # 镜头段太短，跳过 diff --git a/apps/api/app/services/plan_generator_service.py b/apps/api/app/services/plan_generator_service.py index 33e412b84..de0b9b938 100755 --- a/apps/api/app/services/plan_generator_service.py +++ b/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（注入随机噪声）。 diff --git a/packages/domain/plan_generator_utils.py b/packages/domain/plan_generator_utils.py index 3e57596ac..446f6da21 100755 --- a/packages/domain/plan_generator_utils.py +++ b/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: diff --git a/tests/unit/test_scene_change_prerender.py b/tests/unit/test_scene_change_prerender.py new file mode 100644 index 000000000..3f7d5f25a --- /dev/null +++ b/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() -- 2.54.0