From 21d7a6f45a2cc099489a7a2cf924d8490f1c7cf0 Mon Sep 17 00:00:00 2001 From: saas-backend-agent Date: Sun, 6 Sep 2026 17:52:56 +0800 Subject: [PATCH 1/2] =?UTF-8?q?fix(#1743):=20=E6=89=B9=E9=87=8F=E5=8F=98?= =?UTF-8?q?=E4=BD=93=E7=8B=AC=E7=AB=8B=E9=80=89=E7=89=87=E2=80=94=E2=80=94?= =?UTF-8?q?=E5=AE=8C=E6=95=B4=E9=87=8D=E8=B7=91=E5=8D=95=E8=A7=86=E9=A2=91?= =?UTF-8?q?=E9=80=89=E7=89=87=E6=B5=81=E7=A8=8B+=E6=89=B9=E6=AC=A120%?= =?UTF-8?q?=E9=87=8D=E5=8F=A0=E9=81=BF=E8=AE=A9+=E6=9F=A5=E9=87=8D?= =?UTF-8?q?=E8=B6=85=E9=98=88=E9=87=8D=E6=B8=B2+=E5=B0=81=E9=9D=A2?= =?UTF-8?q?=E7=8B=AC=E7=AB=8B?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 根因：批量生成的变体 1..N 走 clone_plan_for_variant 只重算起点， 素材/顺序/速度与源 plan 完全相同，导致 N 条成片片段同源（批次查重率 60.1%/33.3%）。 修复（多视频 = 单视频逻辑 × N）： - 新增 packages/domain/variant_plan_selector.py 纯核心： 素材池 shuffle + main 片段顺序洗牌 + _resolve_start_time 起点重选 （镜头洗牌/随机起点/历史区间避让，与单视频同一入口）+ 批次内同素材 区间重叠 >20% 自动重选起点（最多 6 次） - EditPlanService.reselect_plan_for_variant：源 plan 骨架（clip_type/order/ duration/文案/转场/速度）保留，素材池=源素材∪批次素材，新片段区间 record_used_segments 写回（跨变体/跨任务避让），replace 事务落库 - 正式生成（generation_tasks）：创建任务前预生成 N-1 个独立 plan， 变体0保留用户预览 plan；count>1 无源 plan 直接 400（严禁 N 任务兜底 共用同一 plan）；任务循环兜底块加批量守卫（缺 plan 500 中断） - 批量预览（generation_preview）：count>1 变体0用源 plan、变体1..N-1 reselect 真实选片（所见即所得）；count==1 保持 clone 零回归 - worker 渲染：上传 OSS 与落库查重拆分（_upload_rendered_video/ _record_video_and_dedup）；批次任务 batch_similarity>20% 重选 plan 重渲一次（最多1次，重选失败保留首版，首版_v1后缀不覆盖）； 封面按 task_id md5 稳定哈希分散选候选帧（非批次仍取首帧） - dedup_helpers.create_video_record_and_dedup 返回 dict： {video_count,is_duplicate,batch_similarity,duplicate_of} 测试：新增 24 个单测（纯核心选片差异化/重叠避让/骨架保留、服务层 reselect、重渲阈值判定、封面哈希、批次查重 dict 契约），适配 6 个旧测试， 全量 14469 passed；diff coverage 63%。 --- apps/api/app/api/routes/generation_preview.py | 61 +++- apps/api/app/api/routes/generation_tasks.py | 95 ++++-- apps/api/app/services/edit_plan_service.py | 135 +++++++++ apps/worker/video_processing/dedup_helpers.py | 18 +- apps/worker/worker_app/tasks/generation.py | 245 ++++++++++++---- packages/domain/variant_plan_selector.py | 250 ++++++++++++++++ tests/unit/test_1677_batch_variants.py | 251 +++++++++++++--- tests/unit/test_batch_dedup_helpers_1743.py | 244 ++++++++++++++++ tests/unit/test_batch_no_plan_guard.py | 172 ++++++----- tests/unit/test_batch_render_rerender_1743.py | 80 ++++++ tests/unit/test_dedup_two_phase_commit.py | 18 +- .../test_generated_video_creation_logic.py | 8 +- tests/unit/test_reselect_plan_variant_1743.py | 206 +++++++++++++ tests/unit/test_temp_dir_cleanup_race.py | 5 +- tests/unit/test_variant_plan_selector_1743.py | 271 ++++++++++++++++++ 15 files changed, 1860 insertions(+), 199 deletions(-) create mode 100644 packages/domain/variant_plan_selector.py create mode 100644 tests/unit/test_batch_dedup_helpers_1743.py create mode 100644 tests/unit/test_batch_render_rerender_1743.py create mode 100644 tests/unit/test_reselect_plan_variant_1743.py create mode 100644 tests/unit/test_variant_plan_selector_1743.py diff --git a/apps/api/app/api/routes/generation_preview.py b/apps/api/app/api/routes/generation_preview.py index e8d0b8a53..50ca16034 100755 --- a/apps/api/app/api/routes/generation_preview.py +++ b/apps/api/app/api/routes/generation_preview.py @@ -420,42 +420,77 @@ def create_preview_generation_task( logger.error("[预览生成] 创建失败: %s", e, exc_info=True) raise HTTPException(status_code=500, detail="创建预览生成任务失败，请稍后再试") from e - # ── 克隆独立变体 plan：N 个预览全部克隆（预览不污染源 plan）── - # 源 plan 不存在（无编辑历史）时各任务走自身随机选片流程，不克隆。 + # ── 独立变体 plan（#1743）── + # count=1：克隆源 plan（预览不污染源 plan，仅起点重算），行为与旧版一致； + # count>1：变体 0 保留源 plan，变体 1..N-1 用 reselect_plan_for_variant 完整 + # 重跑单视频选片（素材洗牌+镜头洗牌+起点随机+跨变体避让+批次 20% 重叠重选）， + # 所见即所得——预览变体差异即正式成片差异。 source_plan_id = created_tasks[0].source_edit_plan_id if created_tasks else "" - if source_plan_id: + if source_plan_id and count == 1: + # 单预览：克隆一份（原逻辑） try: from app.services.edit_plan_service import EditPlanService _plan_svc = EditPlanService(db) - for variant_index in range(count): + variant_plan = _plan_svc.clone_plan_for_variant( + source_plan_id, + created_by_user_id=user_id, + name_suffix="预览变体", + ) + variant_plan_ids.append(variant_plan.id) + except Exception as e: + logger.error("[预览生成] 克隆预览 plan 异常: %s", e, exc_info=True) + for t in created_tasks: + _mark_task_failed(generation_task_repository, t, "预览计划创建失败") + raise HTTPException( + status_code=500, + detail="创建预览任务失败：无法生成独立剪辑计划，请重试", + ) from e + elif source_plan_id and count > 1: + try: + from app.services.edit_plan_service import EditPlanService + + _plan_svc = EditPlanService(db) + # 变体 0 直接用源 plan；变体 1..N-1 独立选片 + variant_plan_ids.append(source_plan_id) + batch_asset_pool = list(dict.fromkeys(request.asset_ids or [])) + for variant_index in range(1, count): last_err: Exception | None = None variant_plan = None for _attempt in range(2): # 1 次重试，抗 DB 瞬时抖动 try: - variant_plan = _plan_svc.clone_plan_for_variant( + variant_plan = _plan_svc.reselect_plan_for_variant( source_plan_id, + batch_asset_pool, created_by_user_id=user_id, - name_suffix=f"预览变体{variant_index + 1}" if count > 1 else "预览变体", + name_suffix=f"预览变体{variant_index + 1}", ) break - except Exception as clone_err: # noqa: PERF203 - last_err = clone_err + except ValueError as ve: + logger.warning("[预览生成] 变体独立选片失败(素材不足): %s", ve) + for t in created_tasks: + _mark_task_failed(generation_task_repository, t, "预览变体选片失败") + raise HTTPException( + status_code=400, + detail=f"批量预览第 {variant_index + 1} 个视频无法独立选片：{ve}。" + "请增加素材库中的视频素材后重试。", + ) from ve + except Exception as reselection_err: # noqa: PERF203 + last_err = reselection_err logger.warning( - "[预览生成] 克隆变体 plan 失败(尝试%d/2): variant=%d error=%s", + "[预览生成] 变体独立选片失败(尝试%d/2): variant=%d error=%s", _attempt + 1, variant_index, - clone_err, + reselection_err, exc_info=True, ) if variant_plan is None: logger.error( - "[预览生成] 克隆预览变体 plan 重试仍失败: variant=%d source=%s", + "[预览生成] 变体独立选片重试仍失败: variant=%d source=%s", variant_index, source_plan_id, exc_info=last_err, ) - # 标记已创建任务失败 for t in created_tasks: _mark_task_failed(generation_task_repository, t, "预览变体计划创建失败") raise HTTPException( @@ -466,7 +501,7 @@ def create_preview_generation_task( except HTTPException: raise except Exception as e: - logger.error("[预览生成] 克隆变体 plan 异常: %s", e, exc_info=True) + logger.error("[预览生成] 变体 plan 生成异常: %s", e, exc_info=True) for t in created_tasks: _mark_task_failed(generation_task_repository, t, "预览变体计划创建失败") raise HTTPException( diff --git a/apps/api/app/api/routes/generation_tasks.py b/apps/api/app/api/routes/generation_tasks.py index b0e01892f..f991713ea 100755 --- a/apps/api/app/api/routes/generation_tasks.py +++ b/apps/api/app/api/routes/generation_tasks.py @@ -432,39 +432,83 @@ def create_generation_task( logger.info("画中画已下线，strategy_id %s → one_take", effective_strategy_id) effective_strategy_id = "one_take" - # 批量生成时每个任务关联独立克隆 plan（片段起点重算）， - # 禁止 N 条任务共用同一 source_edit_plan_id 导致片段一模一样。 - # 在创建任何任务【之前】预克隆全部变体：克隆失败直接中断（此时无脏数据）， - # 绝不静默退回共用源 plan（否则批量视频内容重复，违反去重诉求）。 + # 批量生成（count>1）：每个变体必须走与单视频完全相同的独立选片流程（#1743）。 + # - 变体 0 保留源 plan（保留用户编辑结果）； + # - 变体 1..N-1 用 reselect_plan_for_variant 完整重跑选片（素材洗牌 + 镜头洗牌 + # + 起点随机 + 跨变体区间避让 + 批次 20% 重叠重选），而非"克隆只改起点"； + # - count>1 但没有源 plan（前端未传 source_edit_plan_id 且无模板 plan）时， + # 不允许 N 个任务兜底共用同一 plan，直接 4xx 中断（宁可不生成，也不出同源成片）。 + # 在创建任何任务【之前】预生成全部变体 plan：失败直接中断（此时无脏数据）。 variant_plan_ids: list[str] = [] - if count > 1 and request.source_edit_plan_id: + if count > 1: from app.services.edit_plan_service import EditPlanService _plan_svc = EditPlanService(db) + + # 解析批量源 plan：优先前端传入；否则按 template_id + user 查最新（与单任务兜底同源） + batch_source_plan_id = request.source_edit_plan_id + if not batch_source_plan_id and request.template_id: + try: + from packages.adapters.sqlalchemy_impl.models import EditPlanModel + + _latest = ( + db.query(EditPlanModel) + .filter( + EditPlanModel.template_id == request.template_id, + EditPlanModel.created_by_user_id == user_id, + ) + .order_by(EditPlanModel.created_at.desc()) + .first() + ) + if _latest: + batch_source_plan_id = _latest.id + except Exception: + logger.warning("[生成任务] 批量源 plan 解析失败", exc_info=True) + + if not batch_source_plan_id: + # 无任何可用源 plan：批量变体无从选片，明确报错，严禁静默共用/同源 + logger.error("[生成任务] 批量 count=%d 但无可编辑计划（无 source_edit_plan_id/template plan）", count) + raise HTTPException( + status_code=400, + detail="批量生成需要先完成预览生成（缺少剪辑计划）。请先生成预览后再批量创建。", + ) + + # 批次素材池：请求显式素材 + 库自动匹配素材（resolved_asset_ids） + batch_asset_pool = list(dict.fromkeys(resolved_asset_ids or [])) + for task_index in range(1, count): variant = None last_err: Exception | None = None for _attempt in range(2): # 1 次重试，抗 DB 瞬时抖动 try: - variant = _plan_svc.clone_plan_for_variant( - request.source_edit_plan_id, + variant = _plan_svc.reselect_plan_for_variant( + batch_source_plan_id, + batch_asset_pool, created_by_user_id=user_id, name_suffix=f"批量{task_index + 1}", ) break - except Exception as clone_err: # noqa: PERF203 - last_err = clone_err + except ValueError as ve: + # 素材不足等可预期错误：不重试，直接中断并给出明确提示 + logger.warning("[生成任务] 变体独立选片失败(素材不足): %s", ve) + raise HTTPException( + status_code=400, + detail=f"批量生成第 {task_index + 1} 个视频无法独立选片：{ve}。" + "请增加素材库中的视频素材后重试。", + ) from ve + except Exception as reselection_err: # noqa: PERF203 + last_err = reselection_err logger.warning( - "[生成任务] 克隆变体 plan 失败(尝试%d/2): source=%s error=%s", + "[生成任务] 变体独立选片失败(尝试%d/2): source=%s error=%s", _attempt + 1, - request.source_edit_plan_id, - clone_err, + batch_source_plan_id, + reselection_err, exc_info=True, ) if variant is None: logger.error( - "[生成任务] 克隆变体 plan 重试仍失败，中断批量创建: source=%s", - request.source_edit_plan_id, + "[生成任务] 变体独立选片重试仍失败，中断批量创建: source=%s", + batch_source_plan_id, exc_info=last_err, ) raise HTTPException( @@ -475,10 +519,11 @@ def create_generation_task( try: for task_index in range(count): - # 第 1 条复用源 plan（保留用户编辑结果）；其余使用预克隆的独立变体 plan。 - # 无源 plan（source_edit_plan_id 为空）时无可克隆对象，variant_plan_ids - # 为空列表：各任务走自身随机选片流程，不做索引访问（防 IndexError） - effective_plan_id = request.source_edit_plan_id + # 变体 0 复用源 plan（保留用户编辑结果）；变体 1..N-1 用预生成的独立选片 plan。 + # count>1 时上方已保证存在源 plan 且变体 plan 数量 == count-1。 + effective_plan_id = ( + request.source_edit_plan_id or batch_source_plan_id if count > 1 else request.source_edit_plan_id + ) if task_index > 0 and variant_plan_ids: effective_plan_id = variant_plan_ids[task_index - 1] @@ -523,7 +568,19 @@ def create_generation_task( try: # 兜底关联编辑计划：前端未传 source_edit_plan_id 时， # 通过 template_id + user_id 在 DB 层直接查找最新的 plan。 - # 必须在 enqueue 之前执行，避免 worker 读取时 source_edit_plan_id 为空（竞态条件） + # 必须在 enqueue 之前执行，避免 worker 读取时 source_edit_plan_id 为空（竞态条件）。 + # #1743：批量（count>1）场景严禁兜底共用——变体 plan 已在上方预生成， + # 走到这里还缺 plan 说明预生成漏配，直接报错中断，不允许 N 任务关联同一 plan。 + if not task.source_edit_plan_id and count > 1: + logger.error( + "[生成任务] 批量任务缺少独立 plan（禁止共用兜底）: task_index=%d task_id=%s", + task_index, + task.id, + ) + raise HTTPException( + status_code=500, + detail="创建批量任务失败：变体剪辑计划缺失，请重新预览后再批量生成。", + ) if not task.source_edit_plan_id and request.template_id: try: from packages.adapters.sqlalchemy_impl.models import EditPlanModel diff --git a/apps/api/app/services/edit_plan_service.py b/apps/api/app/services/edit_plan_service.py index 5ed03ae07..e6a25ba9c 100755 --- a/apps/api/app/services/edit_plan_service.py +++ b/apps/api/app/services/edit_plan_service.py @@ -409,8 +409,13 @@ class EditPlanService: clip_type=clip_item.get("clip_type", "main"), order=order, asset_id=clip_item.get("asset_id", ""), + text_content=clip_item.get("text_content", ""), start_time=clip_item.get("start_time", 0.0), duration=clip_item.get("duration", 0.0), + transition_effect=clip_item.get("transition_effect", "cut"), + transition_duration=clip_item.get("transition_duration", 0.0), + playback_speed=clip_item.get("playback_speed", 1.0), + config=clip_item.get("config") or None, ) model = EditPlanClipModel( id=clip.id, @@ -459,6 +464,136 @@ class EditPlanService: logger.exception("事务性替换片段失败: plan_id=%s", plan_id) raise + def reselect_plan_for_variant( + self, + source_plan_id: str, + candidate_asset_ids: list[str], + *, + created_by_user_id: str = "", + name_suffix: str = "变体", + rng=None, + ) -> EditPlan: + """为批量变体生成独立 plan：完整重跑单视频选片流程（#1743）。 + + 与 clone_plan_for_variant（只重算起点、素材/顺序不变）不同，本方法： + - 源 plan 片段骨架（clip_type/order/duration/文案/转场）保留； + - 素材池 shuffle 随机分配 + main 片段顺序洗牌； + - 起点走场景镜头洗牌/随机起点/历史区间避让（与单视频同一入口）； + - 批次内同素材区间重叠 >20% 自动重选起点； + - 新片段区间 record_used_segments 写回素材 metadata（跨变体/跨任务避让）。 + + Args: + source_plan_id: 源 plan（任务 0 / 预览源）。 + candidate_asset_ids: 素材池（源 plan 素材 ∪ 批次素材）。 + created_by_user_id: 新 plan 归属用户。 + name_suffix: plan 名后缀。 + rng: 可选随机数（测试注入种子）。 + + Raises: + ValueError: 源 plan 不存在/无片段、素材池为空或时长全未知。 + """ + from packages.adapters.sqlalchemy_impl.models import AssetModel + from packages.domain.plan_generator_utils import extract_scene_points_from_metadata + from packages.domain.variant_plan_selector import reselect_clips_for_variant + + source = self.get_plan_or_raise(source_plan_id) + + # 分页读取源 plan 全部片段 + clips: List[EditPlanClip] = [] + skip, page = 0, 500 + while True: + batch = self._clip_repo.list_by_plan(source_plan_id, skip=skip, limit=page) + if not batch: + break + clips.extend(batch) + if len(batch) < page: + break + skip += page + if not clips: + raise ValueError(f"源 plan 无片段，无法生成变体: {source_plan_id}") + + source_clips_data = [ + { + "order": c.order if c.order is not None else i, + "asset_id": c.asset_id, + "start_time": float(c.start_time or 0.0), + "duration": float(c.duration or 0.0), + "clip_type": c.clip_type, + "playback_speed": float(c.playback_speed or 1.0), + "transition_effect": c.transition_effect, + "transition_duration": float(c.transition_duration or 0.0), + "text_content": c.text_content or "", + "config": c.config or {}, + } + for i, c in enumerate(clips) + ] + + db = self._clip_repo.session + + # 素材池 = 源 plan 素材 ∪ 调用方传入素材（去重保序） + pool_ids: list[str] = [] + seen = set() + for aid in [c.asset_id for c in clips if c.asset_id] + list(candidate_asset_ids or []): + if aid and aid not in seen: + seen.add(aid) + pool_ids.append(aid) + + # 时长 + 场景点 + durations: dict[str, float] = {} + scene_points: dict[str, list[float]] = {} + if pool_ids: + for m in db.query(AssetModel).filter(AssetModel.id.in_(pool_ids)).all(): + durations[m.id] = float(getattr(m, "duration", 0.0) or 0.0) + pts = extract_scene_points_from_metadata(getattr(m, "metadata", None)) + if pts: + scene_points[m.id] = pts + + historical = get_used_segments(db, pool_ids) + + # 创建新 plan（复制模板归属与 config） + new_plan = self.create_plan( + template_id=source.template_id, + name=f"{source.name or '剪辑计划'} · {name_suffix}", + config=dict(source.config or {}), + total_duration=source.total_duration, + project_id=source.project_id or "", + created_by_user_id=created_by_user_id or (source.created_by_user_id or ""), + ) + + # 批次内区间：以源 plan（变体 0）片段为初始避让对象 + batch_segments: dict[str, list[tuple[float, float]]] = {} + for c in clips: + if c.asset_id and float(c.duration or 0) > 0: + st = float(c.start_time or 0.0) + batch_segments.setdefault(c.asset_id, []).append((st, st + float(c.duration))) + + clips_data = reselect_clips_for_variant( + source_clips_data, + pool_ids, + asset_durations=durations, + asset_scene_points=scene_points, + historical_used_segments=historical, + batch_segments=batch_segments, + rng=rng, + ) + + # 片段区间写回素材 metadata（与落库同事务；replace_all_clips_transactional 内 commit） + for item in clips_data: + aid = item.get("asset_id", "") + if aid: + st = float(item.get("start_time", 0.0)) + record_used_segments(db, aid, st, st + float(item.get("duration", 0.0)), new_plan.id) + + self.replace_all_clips_transactional(new_plan.id, clips_data) + logger.info( + "变体独立选片完成: source=%s new=%s clips=%d assets=%d", + source_plan_id, + new_plan.id, + len(clips_data), + len(pool_ids), + ) + return new_plan + def clone_plan_for_variant( self, source_plan_id: str, diff --git a/apps/worker/video_processing/dedup_helpers.py b/apps/worker/video_processing/dedup_helpers.py index d0ee22da8..48ab1e2c2 100755 --- a/apps/worker/video_processing/dedup_helpers.py +++ b/apps/worker/video_processing/dedup_helpers.py @@ -34,7 +34,9 @@ def create_video_record_and_dedup( fps: float = 25.0, name: str = "", thumbnail_url: str = "", -) -> int: +) -> dict: + """Returns: {"video_count": int, "is_duplicate": bool, "batch_similarity": float|None, + "duplicate_of": str|None} —— batch_similarity 为批次内最高相似度（无批次查重时 None）。""" """创建 GeneratedVideo 记录，计算指纹并执行查重（历史 + 批次）。 采用两阶段持久化：先计算所有指纹/查重数据（内存）， @@ -76,6 +78,7 @@ def create_video_record_and_dedup( # ── Phase 2: 计算指纹 & 查重（全部在内存） ──────────────── deduplicator = VideoDeduplicator() fingerprint = None + batch_similarity: float | None = None try: fingerprint = deduplicator.compute_fingerprint(video_path) @@ -105,9 +108,11 @@ def create_video_record_and_dedup( ) # (b) 批次内查重（仅当有 batch_id 时） + batch_similarity: float | None = None if not duplicate_result and batch_id: duplicate_result = deduplicator.check_batch_duplicate(fingerprint, batch_id, video_id, session) - + if duplicate_result: + batch_similarity = float(duplicate_result.get("similarity", 0.0)) if duplicate_result: generated_video.is_duplicate = True generated_video.duplicate_of = duplicate_result["duplicate_of"] @@ -161,7 +166,12 @@ def create_video_record_and_dedup( generated_video.is_duplicate, generated_video.duplicate_rate, ) - return 1 + return { + "video_count": 1, + "is_duplicate": bool(generated_video.is_duplicate), + "batch_similarity": batch_similarity, + "duplicate_of": generated_video.duplicate_of, + } except Exception as e: logger.error( "Failed to create video record / dedup for task %s: %s", @@ -169,4 +179,4 @@ def create_video_record_and_dedup( e, ) session.rollback() - return 0 + return {"video_count": 0, "is_duplicate": False, "batch_similarity": None, "duplicate_of": None} diff --git a/apps/worker/worker_app/tasks/generation.py b/apps/worker/worker_app/tasks/generation.py index 9a0a755e7..a0c7a79ac 100644 --- a/apps/worker/worker_app/tasks/generation.py +++ b/apps/worker/worker_app/tasks/generation.py @@ -383,35 +383,61 @@ def _load_task_info(task_id: str) -> dict | None: session.close() -def _upload_and_record( +# ── #1743 批量变体重渲/封面判定（纯函数，便于单测） ────────────────────── +BATCH_RENDER_SIMILARITY_LIMIT = 0.20 +"""批次内成片查重相似度阈值：超过则重选独立 plan 重渲一次（20%）。""" + + +def should_rerender_for_batch_dedup(*, batch_id: str, render_attempt: int, batch_similarity) -> bool: + """批次内查重后判定是否需要重选 plan 重渲。 + + 条件（全部满足才重渲）：批次任务、首版（attempt==0）、查重率已得出、相似度 > 20%。 + 非批次任务 / 已是重渲版 / 查重率缺失 / 相似度达标 → 不重渲。 + """ + if not batch_id: + return False + if render_attempt >= 1: + return False + if batch_similarity is None: + return False + return float(batch_similarity) > BATCH_RENDER_SIMILARITY_LIMIT + + +def pick_batch_cover_index(task_id: str, candidate_count: int) -> int: + """批次变体封面帧选取：按 task_id md5 稳定哈希分散到候选帧。 + + 同任务重试结果稳定；批次内不同 task_id 哈希后分散，避免 N 个变体都抽 frame_0 + 导致封面雷同。非批次调用方应直接取 0（主流程按 batch_id 区分）。 + """ + if candidate_count <= 1: + return 0 + import hashlib + + return int(hashlib.md5(task_id.encode()).hexdigest(), 16) % candidate_count + + +def _upload_rendered_video( task_id: str, output_path: Path, project_id: str, - batch_id: str, - editing_mode, - user_id: str = "", - video_name: str = "", - thumbnail_url: str = "", -) -> tuple[str, float, int, int]: - """上传 OSS、创建视频记录并查重。 + *, + attempt: int = 0, +) -> tuple[str, str]: + """上传成片到 OSS（不落库）。attempt>0 时文件名带轮次后缀，避免覆盖首版。 - Returns: - (file_url, duration, file_size, video_count) + Returns: (file_url, storage_key) """ - # project_id 可能为空（模板编辑器草稿不属于任何项目），过滤空段避免 OSS key 出现 // - path_parts = [p for p in ("generated", "projects", project_id, "tasks", task_id, output_path.name) if p] + suffix = f"_v{attempt}" if attempt > 0 else "" + stem = output_path.stem + name = f"{stem}{suffix}{output_path.suffix or '.mp4'}" + path_parts = [p for p in ("generated", "projects", project_id, "tasks", task_id, name) if p] storage_key = "/".join(path_parts) - file_size = output_path.stat().st_size - # 上传 OSS - logger.info("[task_id=%s] [OSS上传] 开始上传: size=%d", task_id, file_size) - upload_start = time.monotonic() + logger.info("[task_id=%s] [OSS上传] 开始上传(attempt=%d): size=%d", task_id, attempt, output_path.stat().st_size) file_url = upload_to_oss(output_path, storage_key) - upload_elapsed = time.monotonic() - upload_start if not file_url: raise RuntimeError(f"OSS 上传失败: task_id={task_id}, storage_key={storage_key}") - # 校验 URL 可达性（P0-2: 私有 bucket 用预签名 + object_exists 降级） verify_url = get_signed_download_url(file_url, expires_seconds=300) or file_url if not _verify_url_accessible(verify_url): from video_processing.oss_helpers import normalize_storage_key, oss_bucket @@ -420,25 +446,58 @@ def _upload_and_record( key = normalize_storage_key(file_url) if not (bucket and bucket.object_exists(key)): raise RuntimeError( - f"OSS 上传后 URL 不可访问且 object_exists 失败: file_url={file_url}, " f"storage_key={storage_key}" + f"OSS 上传后 URL 不可访问且 object_exists 失败: file_url={file_url}, storage_key={storage_key}" ) - logger.info( - "URL 校验失败但 object_exists 确认文件存在，视为上传成功: storage_key=%s", - key, - ) + logger.info("URL 校验失败但 object_exists 确认文件存在，视为上传成功: storage_key=%s", key) + return file_url, storage_key - logger.info( - "[task_id=%s] [OSS上传] 成功: 耗时=%.1fs, file_url=%s", - task_id, - upload_elapsed, - file_url, - ) - # 创建 GeneratedVideo 记录 + 查重 - duration = probe_duration(output_path) +def _reselect_plan_for_batch_retry(task_id: str, plan_id: str, task_info: dict) -> str | None: + """批次内查重超阈值后，为当前任务重新独立选片生成新 plan（#1743 自动重渲）。 + + 复用 API 侧同一套 EditPlanService.reselect_plan_for_variant（packages 层 + variant_plan_selector 纯核心），素材池来自任务 asset_ids + 源 plan 素材。 + 成功返回新 plan_id；失败返回 None（调用方放弃重渲，保留首版）。 + """ + try: + from app.services.edit_plan_service import EditPlanService + + db = SessionLocal() + try: + svc = EditPlanService(db) + asset_pool = list(task_info.get("task_asset_ids") or []) + new_plan = svc.reselect_plan_for_variant( + plan_id, + asset_pool, + created_by_user_id=task_info.get("user_id", ""), + name_suffix="重渲变体", + ) + return new_plan.id + finally: + db.close() + except Exception: + logger.warning("[task_id=%s] 批次重渲前重选 plan 失败，放弃重渲", task_id, exc_info=True) + return None + + +def _record_video_and_dedup( + *, + task_id: str, + project_id: str, + batch_id: str, + editing_mode, + user_id: str, + file_url: str, + file_size: int, + video_path: str, + video_name: str = "", + thumbnail_url: str = "", +) -> dict: + """成片落库 + 指纹查重（含批次内）。返回查重信息 dict。""" + duration = probe_duration(Path(video_path)) dedup_session = SessionLocal() try: - video_count = create_video_record_and_dedup( + result = create_video_record_and_dedup( generation_task_id=task_id, project_id=project_id, user_id=user_id, @@ -446,7 +505,7 @@ def _upload_and_record( file_url=file_url, file_size=file_size, duration=duration, - video_path=str(output_path), + video_path=video_path, mode=editing_mode.value, session=dedup_session, name=video_name, @@ -454,11 +513,12 @@ def _upload_and_record( ) finally: dedup_session.close() - - return file_url, duration, file_size, video_count or 1 + result["duration"] = duration + return result # ── Celery Task ────────────────────────────────────────────────────────────── +# ── Celery Task ────────────────────────────────────────────────────────────── def _sync_task_config_to_plan(source_edit_plan_id: str, task_info: dict, db) -> str | None: @@ -729,19 +789,29 @@ def generate_video(self, task_id: str) -> dict: gen_task.append_log("渲染模式", "从草稿数据渲染（与预览一致）") _flush_logs(task_id, gen_task) - output_path, render_duration, cover_candidates, voiceover_tmp_path, render_temp_dir, thumbnail_url = ( - _render_from_edit_plan( + # ── 渲染→上传→查重→（批次超阈值则重选 plan 重渲一次）循环（#1743）── + current_plan_id = source_edit_plan_id + file_url = "" + duration = 0.0 + file_size = 0 + video_count = 1 + file_size_final = 0 + for render_attempt in range(2): # 首版 + 最多 1 次重渲 + ( + output_path, + render_duration, + cover_candidates, + voiceover_tmp_path, + render_temp_dir, + thumbnail_url, + ) = _render_from_edit_plan( task_id=task_id, - source_edit_plan_id=source_edit_plan_id, + source_edit_plan_id=current_plan_id, task_info=task_info, ) - ) - # 从这里开始，render_temp_dir 已赋值，必须确保异常时也能清理 - try: if gen_task: - gen_task.append_log("渲染", f"渲染完成, 时长={render_duration:.1f}s") + gen_task.append_log("渲染", f"渲染完成(第{render_attempt + 1}版), 时长={render_duration:.1f}s") _flush_logs(task_id, gen_task) - _update_task_progress(task_id, 80, "渲染完成") # ── 3.5 随机边缘裁剪降重（#1664） ────────────────────────── @@ -751,7 +821,7 @@ def generate_video(self, task_id: str) -> dict: cropped_path = random_edge_crop(output_path) if cropped_path != output_path: output_path = cropped_path - if gen_task: + if gen_task and render_attempt == 0: gen_task.append_log("边缘裁剪", "已应用随机 2-5% 边缘裁剪降重") _flush_logs(task_id, gen_task) logger.info("[task_id=%s] 随机边缘裁剪完成: %s", task_id, output_path) @@ -762,45 +832,110 @@ def generate_video(self, task_id: str) -> dict: crop_err, exc_info=True, ) - if gen_task: - gen_task.append_log("边缘裁剪", f"裁剪失败，使用原始视频: {crop_err}") - _flush_logs(task_id, gen_task) - # ── 4. 上传 OSS + 查重记录 ─────────────────────────────── + # ── 4. 上传 OSS（不落库） ─────────────────────────────── _update_task_progress(task_id, 85, "开始上传") - file_url, duration, file_size, video_count = _upload_and_record( + file_url, _storage_key = _upload_rendered_video( task_id=task_id, output_path=output_path, project_id=project_id, + attempt=render_attempt, + ) + file_size = output_path.stat().st_size + + # ── 4.5 落库 + 查重（批次任务检查批次内相似度） ─────────── + dedup_info = _record_video_and_dedup( + task_id=task_id, + project_id=project_id, batch_id=batch_id, editing_mode=editing_mode, user_id=user_id, + file_url=file_url, + file_size=file_size, + video_path=str(output_path), video_name=task_info.get("video_title", ""), thumbnail_url=thumbnail_url, ) + duration = dedup_info.get("duration", render_duration) + video_count = dedup_info.get("video_count", 1) + batch_sim = dedup_info.get("batch_similarity") if gen_task: gen_task.append_log( "OSS上传", - f"上传成功, 大小={file_size}", + f"第{render_attempt + 1}版上传成功, 大小={file_size}" + + (f", 批次相似度={batch_sim:.0%}" if batch_sim is not None else ""), file_size=file_size, file_url=file_url, ) _flush_logs(task_id, gen_task) - _update_task_progress(task_id, 95, "上传完成") - finally: - # 清理渲染临时目录（无论后续步骤成功与否都清理） + # 非批次 / 相似度达标 / 已是最后一次 → 结束循环 + if not should_rerender_for_batch_dedup( + batch_id=batch_id, + render_attempt=render_attempt, + batch_similarity=batch_sim, + ): + file_size_final = file_size + break + + # 批次内相似度过高：重选独立 plan 后重渲一次 + logger.warning( + "[task_id=%s] 批次内查重相似度 %.2f 超阈值 %.2f，重选 plan 重渲", + task_id, + batch_sim, + BATCH_RENDER_SIMILARITY_LIMIT, + ) + if gen_task: + gen_task.append_log("批次查重", f"与批次内成片相似度过高({batch_sim:.0%})，重新选片渲染") + _flush_logs(task_id, gen_task) + new_plan_id = _reselect_plan_for_batch_retry(task_id, current_plan_id, task_info) + if not new_plan_id: + logger.warning("[task_id=%s] 重选 plan 失败，保留首版", task_id) + file_size_final = file_size + break + # 回写任务关联的 plan（重渲版以新 plan 渲染） + try: + _ps = SessionLocal() + try: + from packages.adapters.sqlalchemy_impl.generation_task_repository import ( + SQLAlchemyGenerationTaskRepository, + ) + + _pr = SQLAlchemyGenerationTaskRepository(_ps) + _gt = _pr.get(task_id) + if _gt: + _gt.source_edit_plan_id = new_plan_id + _pr.update(_gt) + finally: + _ps.close() + except Exception: + logger.warning("[task_id=%s] 回写重渲 plan_id 失败", task_id, exc_info=True) + current_plan_id = new_plan_id + # 清理本轮临时目录，下一轮重新渲染 if render_temp_dir: import shutil shutil.rmtree(render_temp_dir, ignore_errors=True) - logger.info("[task_id=%s] 渲染临时目录已清理: %s", task_id, render_temp_dir) + render_temp_dir = None + + file_size = file_size_final or file_size + _update_task_progress(task_id, 95, "上传完成") + + # 渲染结束后清理临时目录（重渲循环内每轮已清理，此处兜底最后一轮） + if render_temp_dir: + import shutil + + shutil.rmtree(render_temp_dir, ignore_errors=True) + logger.info("[task_id=%s] 渲染临时目录已清理: %s", task_id, render_temp_dir) # ── 4.5 封面帧持久化 ──────────────────────────────────────────── try: if cover_candidates: - first = cover_candidates[0] + # #1743：批量变体封面差异化——候选帧按 task_id 稳定哈希分散选取 + # （同任务重试稳定，批次内不同任务落在不同帧位），非批次取首帧。 + _cover_idx = pick_batch_cover_index(task_id, len(cover_candidates)) if batch_id else 0 + first = cover_candidates[_cover_idx] cover_frame_url = first.get("image_url") or first.get("url") or "" if cover_frame_url: _cover_session = SessionLocal() diff --git a/packages/domain/variant_plan_selector.py b/packages/domain/variant_plan_selector.py new file mode 100644 index 000000000..83ec700f4 --- /dev/null +++ b/packages/domain/variant_plan_selector.py @@ -0,0 +1,250 @@ +"""批量变体独立选片核心（#1743）。 + +总原则：多视频 = 单视频逻辑 × N。批量正式生成/批量预览时，变体 1..N-1 +不再"克隆源 plan 只重算起点"（那会导致同批素材、同顺序、同速度，成片同源）， +而是**完整重跑单视频的选片流程**： + +1. 源 plan 片段骨架（clip_type/order/duration/text/transition）保持不变 —— 保留 + 模板结构与用户编辑结果； +2. 素材池上做完整随机重选： + - 素材组合随机（shuffle 素材池 + smart_match 评分噪声由调用方排序决定）； + - main 片段之间随机洗牌顺序（片段顺序显著不同）； + - 起点走场景镜头洗牌 + 随机起点 + 历史已用区间避让 + （pick_scene_aware_start / _calc_random_start_time，与单视频同一入口）； + - 跨变体/跨任务避让：get_used_segments 读取素材 metadata 持久化的已用区间， + record_used_segments 随新片段写回（同事务），N 个变体串行选片时天然互相避让； +3. 批次内片段重叠检查：重选后与"本批次已选定片段"对比，同一 asset 时间区间 + 重叠占比 > 阈值（默认 20%）则该片段重选起点，最多重试若干次。 + +本模块只产出 clips_data（dict 列表，供 EditPlanService.replace_all_clips_transactional +落库），不碰 DB 事务边界；素材时长/场景点/已用区间由调用方注入，便于单测。 +""" + +from __future__ import annotations + +import logging +import random + +from packages.domain.plan_generator_utils import _resolve_start_time + +logger = logging.getLogger(__name__) + +# ── 阈值常量 ──────────────────────────────────────────────────────────────── +BATCH_CLIP_OVERLAP_LIMIT = 0.20 +"""批次内同一素材片段时间区间重叠占比上限（20%）。超过则重选起点。""" + +VARIANT_RESELECT_MAX_ATTEMPTS = 6 +"""单片段重叠避让/起点重选的最大尝试次数。""" + +MAIN_CLIP_TYPES = {"main"} +"""参与素材洗牌重选的片段类型（intro/outro/overlay 等固定角色片段保持源 plan）。""" + + +def _clip_overlap_ratio( + asset_id: str, + start: float, + duration: float, + batch_segments: dict[str, list[tuple[float, float]]], +) -> float: + """计算新区间 [start, start+duration) 与批次内同素材已选区间的重叠占比。 + + 返回重叠总时长 / 片段时长。 + """ + if not asset_id or duration <= 0: + return 0.0 + end = start + duration + overlap = 0.0 + for seg_start, seg_end in batch_segments.get(asset_id, []): + ov = max(0.0, min(end, seg_end) - max(start, seg_start)) + overlap += ov + return min(1.0, overlap / duration) + + +def reselect_clips_for_variant( + source_clips: list[dict], + candidate_asset_ids: list[str], + *, + asset_durations: dict[str, float], + asset_scene_points: dict[str, list[float]] | None = None, + historical_used_segments: dict[str, list[tuple[float, float]]] | None = None, + batch_segments: dict[str, list[tuple[float, float]]] | None = None, + rng: random.Random | None = None, +) -> list[dict]: + """为一个变体基于源片段骨架重新独立选片。 + + Args: + source_clips: 源 plan 片段（dict 列表，每项至少含 + order/asset_id/start_time/duration/clip_type，可含 + playback_speed/transition_effect/transition_duration/text_content）。 + candidate_asset_ids: 素材池（源 plan 素材 ∪ 批次任务素材），将被 shuffle + 后随机分配给 main 片段。 + asset_durations: {asset_id: 时长秒}，起点避让/区间计算必需。 + asset_scene_points: {asset_id: 场景切换点}，有则走镜头洗牌选起点。 + historical_used_segments: 素材 metadata 中持久化的历史已用区间 + （跨任务/跨变体避让），函数内会就地追加本变体选中的区间。 + batch_segments: 本批次已选片段区间（变体间避让 + 20% 重叠检查）， + 函数内会就地追加本变体选中的区间。 + rng: 可选随机数生成器（测试可注入固定种子）。 + + Returns: + clips_data: 与源片段等长、order 对齐的新片段 dict 列表。 + + Raises: + ValueError: 源片段为空 / 素材池为空 / 素材时长全为 0（无法差异化选片）。 + """ + rng = rng or random.Random() + if not source_clips: + raise ValueError("源 plan 无片段，无法为变体重新选片") + if not candidate_asset_ids: + raise ValueError("素材池为空，无法为变体独立选片（不允许退回同源成片）") + + # 仅保留时长可知（>0）的素材；时长未知无法做区间避让/重叠计算 + usable_assets = [a for a in dict.fromkeys(candidate_asset_ids) if asset_durations.get(a, 0.0) > 0] + if not usable_assets: + raise ValueError("素材池时长全部未知（0），无法为变体独立选片") + + # 历史已用区间：复制一份，本变体选中的区间就地追加（随 clip record 持久化由调用方负责） + used_segments: dict[str, list[tuple[float, float]]] = ( + {k: list(v) for k, v in (historical_used_segments or {}).items()} if historical_used_segments else {} + ) + batch_segments = batch_segments if batch_segments is not None else {} + + # 按 order 排序源片段，保持骨架顺序 + ordered = sorted(source_clips, key=lambda c: c.get("order", 0)) + + # ── 1. 素材洗牌：素材池 shuffle（组合随机） ───────────────────────────── + shuffled_pool = list(usable_assets) + rng.shuffle(shuffled_pool) + + # ── 2. main 片段之间洗牌顺序（顺序随机） ──────────────────────────────── + main_indexes = [i for i, c in enumerate(ordered) if c.get("clip_type", "main") in MAIN_CLIP_TYPES] + rng.shuffle(main_indexes) + + result: list[dict | None] = [None] * len(ordered) + pool_cursor = 0 + + for idx in main_indexes: + src = ordered[idx] + dur = float(src.get("duration", 0.0) or 0.0) + if dur <= 0: + # 异常片段：原样保留 + result[idx] = _base_clip_data( + src, asset_id=src.get("asset_id", ""), start=float(src.get("start_time", 0.0)) + ) + continue + + # 轮询取洗牌后素材（素材数 < 片段数时循环复用，但组合/顺序已随机） + asset_id = shuffled_pool[pool_cursor % len(shuffled_pool)] + pool_cursor += 1 + total = asset_durations.get(asset_id, 0.0) + eff_dur = min(dur, total) if total > 0 else dur + + # ── 3. 起点重选（镜头洗牌/随机起点/历史避让）+ 批次重叠避让 ────────── + start = _pick_start_with_overlap_avoid( + asset_id=asset_id, + clip_duration=eff_dur, + asset_durations=asset_durations, + used_segments=used_segments, + asset_scene_points=asset_scene_points, + batch_segments=batch_segments, + rng=rng, + ) + + interval = (start, start + eff_dur) + used_segments.setdefault(asset_id, []).append(interval) + batch_segments.setdefault(asset_id, []).append(interval) + + result[idx] = _base_clip_data(src, asset_id=asset_id, start=start) + + # ── 4. 非 main 片段（intro/outro/overlay 等固定角色）：保留源素材，仅重算起点 ── + for idx, c in enumerate(ordered): + if result[idx] is not None: + continue + src = c + aid = src.get("asset_id", "") + dur = float(src.get("duration", 0.0) or 0.0) + start = float(src.get("start_time", 0.0)) + total = asset_durations.get(aid, 0.0) + if aid and dur > 0 and total > 0: + eff_dur = min(dur, total) + new_start = _pick_start_with_overlap_avoid( + asset_id=aid, + clip_duration=eff_dur, + asset_durations=asset_durations, + used_segments=used_segments, + asset_scene_points=asset_scene_points, + batch_segments=batch_segments, + rng=rng, + ) + start = new_start + interval = (start, start + eff_dur) + used_segments.setdefault(aid, []).append(interval) + batch_segments.setdefault(aid, []).append(interval) + result[idx] = _base_clip_data(src, asset_id=aid, start=start) + + return [c for c in result if c is not None] + + +def _pick_start_with_overlap_avoid( + *, + asset_id: str, + clip_duration: float, + asset_durations: dict[str, float], + used_segments: dict[str, list[tuple[float, float]]], + asset_scene_points: dict[str, list[float]] | None, + batch_segments: dict[str, list[tuple[float, float]]], + rng: random.Random, +) -> float: + """选起点：优先单视频同一入口（镜头洗牌/随机/历史避让），再叠加批次 20% 重叠避让。 + + 批次内重叠超阈值时在素材可用范围内随机抖动重选，最多 VARIANT_RESELECT_MAX_ATTEMPTS 次； + 仍超阈值则返回最后一次结果（素材极少时的尽力而为，不阻塞生成）。 + """ + total = asset_durations.get(asset_id, 0.0) + max_start = max(0.0, total - clip_duration) + + candidate = _resolve_start_time( + asset_id, + clip_duration, + asset_durations, + used_segments, + asset_scene_points, + ) + if candidate is None: + candidate = rng.uniform(0.0, max_start) if max_start > 0 else 0.0 + + best_start = candidate + best_ratio = _clip_overlap_ratio(asset_id, candidate, clip_duration, batch_segments) + if best_ratio <= BATCH_CLIP_OVERLAP_LIMIT: + return candidate + + # 重叠超阈值：在可用范围内随机重试 + for _ in range(VARIANT_RESELECT_MAX_ATTEMPTS): + alt = rng.uniform(0.0, max_start) if max_start > 0 else 0.0 + ratio = _clip_overlap_ratio(asset_id, alt, clip_duration, batch_segments) + if ratio < best_ratio: + best_start, best_ratio = alt, ratio + if ratio <= BATCH_CLIP_OVERLAP_LIMIT: + return alt + logger.info( + "变体选片批次重叠避让达上限，采用最优起点: asset=%s overlap_ratio=%.2f", + asset_id, + best_ratio, + ) + return best_start + + +def _base_clip_data(src: dict, *, asset_id: str, start: float) -> dict: + """从源片段构造落库 dict（保留骨架/转场/文案/速度，替换素材与起点）。""" + return { + "order": src.get("order", 0), + "asset_id": asset_id, + "start_time": round(float(start), 3), + "duration": float(src.get("duration", 0.0) or 0.0), + "clip_type": src.get("clip_type", "main"), + "playback_speed": float(src.get("playback_speed", 1.0) or 1.0), + "transition_effect": src.get("transition_effect", "cut"), + "transition_duration": float(src.get("transition_duration", 0.0) or 0.0), + "text_content": src.get("text_content", ""), + "config": src.get("config") or {}, + } diff --git a/tests/unit/test_1677_batch_variants.py b/tests/unit/test_1677_batch_variants.py index bf67b5b62..c7c71f884 100644 --- a/tests/unit/test_1677_batch_variants.py +++ b/tests/unit/test_1677_batch_variants.py @@ -189,30 +189,32 @@ class TestBatchPreviewRoute: for i, item in enumerate(resp.items): assert item.variant_index == i - def test_preview_count_3_clones_three_variant_plans(self): - """有源 plan 时，N=3 克隆 3 个独立变体 plan（预览全部克隆，不用源 plan）""" + def test_preview_count_3_reselects_independent_variant_plans(self): + """#1743：有源 plan 时 N=3，变体0保留源 plan，变体1/2 各自独立选片（reselect）。""" from app.api.routes.generation_preview import create_preview_generation_task tasks = [_make_task(task_id=f"task_{i}", source_plan_id="source_plan") for i in range(3)] repo = _repo_mock() - cloned_plan_ids = ["clone_1", "clone_2", "clone_3"] + reselect_plan_ids = ["reselect_1", "reselect_2"] with patch("app.api.routes.generation_preview.CreateGenerationTaskUseCase") as MockUC: MockUC.return_value.execute.side_effect = tasks with patch("app.api.routes.generation_preview.safe_enqueue_generation_task", return_value=True): with patch("app.services.edit_plan_service.EditPlanService") as MockPlanSvc: - clone_results = [MagicMock(id=pid) for pid in cloned_plan_ids] - MockPlanSvc.return_value.clone_plan_for_variant.side_effect = clone_results + reselect_results = [MagicMock(id=pid) for pid in reselect_plan_ids] + MockPlanSvc.return_value.reselect_plan_for_variant.side_effect = reselect_results create_preview_generation_task( _make_preview_request(preview_count=3), authenticated_user=_make_user(), generation_task_repository=repo, db=MagicMock(), ) - # 克隆被调用 3 次 - assert MockPlanSvc.return_value.clone_plan_for_variant.call_count == 3 - # 每个任务关联到不同的克隆 plan - for i, task in enumerate(tasks): - assert task.source_edit_plan_id == cloned_plan_ids[i] + # 变体 1..N-1 各独立选片一次（共 2 次）；count>1 不再走 clone + assert MockPlanSvc.return_value.reselect_plan_for_variant.call_count == 2 + MockPlanSvc.return_value.clone_plan_for_variant.assert_not_called() + # 变体0保留源 plan；变体1/2 关联各自独立选出的 plan + assert tasks[0].source_edit_plan_id == "source_plan" + assert tasks[1].source_edit_plan_id == "reselect_1" + assert tasks[2].source_edit_plan_id == "reselect_2" def test_preview_variant_titles_injected_per_variant(self): """titles[] 按变体注入 title_config.text""" @@ -335,8 +337,8 @@ class TestBatchPreviewRoute: ) assert exc.value.status_code == 429 - def test_preview_clone_failure_marks_all_failed(self): - """克隆变体 plan 失败 → 已创建任务全部标记 failed 并 500""" + def test_preview_reselect_failure_marks_all_failed(self): + """#1743：变体独立选片（reselect）重试仍失败 → 已创建任务全部标记 failed 并 500""" from app.api.routes.generation_preview import create_preview_generation_task from fastapi import HTTPException @@ -345,7 +347,7 @@ class TestBatchPreviewRoute: with patch("app.api.routes.generation_preview.CreateGenerationTaskUseCase") as MockUC: MockUC.return_value.execute.side_effect = tasks with patch("app.services.edit_plan_service.EditPlanService") as MockPlanSvc: - MockPlanSvc.return_value.clone_plan_for_variant.side_effect = RuntimeError("db down") + MockPlanSvc.return_value.reselect_plan_for_variant.side_effect = RuntimeError("db down") with pytest.raises(HTTPException) as exc: create_preview_generation_task( _make_preview_request(preview_count=3), @@ -357,6 +359,73 @@ class TestBatchPreviewRoute: # 所有已创建任务都被标记 failed assert all(t.status == GenerationTaskStatus.FAILED for t in tasks) + def test_preview_reselect_value_error_returns_400(self): + """#1743：预览 count>1 reselect 素材不足（ValueError）→ 400，已建任务标 failed。""" + from app.api.routes.generation_preview import create_preview_generation_task + from fastapi import HTTPException + + tasks = [_make_task(task_id=f"task_{i}", source_plan_id="source_plan") for i in range(3)] + repo = _repo_mock() + with patch("app.api.routes.generation_preview.CreateGenerationTaskUseCase") as MockUC: + MockUC.return_value.execute.side_effect = tasks + with patch("app.api.routes.generation_preview.safe_enqueue_generation_task", return_value=True): + with patch("app.services.edit_plan_service.EditPlanService") as MockPlanSvc: + MockPlanSvc.return_value.reselect_plan_for_variant.side_effect = ValueError("素材池为空") + with pytest.raises(HTTPException) as exc: + create_preview_generation_task( + _make_preview_request(preview_count=3), + authenticated_user=_make_user(), + generation_task_repository=repo, + db=MagicMock(), + ) + assert exc.value.status_code == 400 + assert "无法独立选片" in exc.value.detail + assert all(t.status == GenerationTaskStatus.FAILED for t in tasks) + + def test_preview_reselect_retry_exhausted_returns_500(self): + """#1743：预览 count>1 reselect 连续失败（非 ValueError）→ 500，已建任务标 failed。""" + from app.api.routes.generation_preview import create_preview_generation_task + from fastapi import HTTPException + + tasks = [_make_task(task_id=f"task_{i}", source_plan_id="source_plan") for i in range(3)] + repo = _repo_mock() + with patch("app.api.routes.generation_preview.CreateGenerationTaskUseCase") as MockUC: + MockUC.return_value.execute.side_effect = tasks + with patch("app.api.routes.generation_preview.safe_enqueue_generation_task", return_value=True): + with patch("app.services.edit_plan_service.EditPlanService") as MockPlanSvc: + MockPlanSvc.return_value.reselect_plan_for_variant.side_effect = RuntimeError("db down") + with pytest.raises(HTTPException) as exc: + create_preview_generation_task( + _make_preview_request(preview_count=3), + authenticated_user=_make_user(), + generation_task_repository=repo, + db=MagicMock(), + ) + assert exc.value.status_code == 500 + assert all(t.status == GenerationTaskStatus.FAILED for t in tasks) + + def test_preview_count1_with_source_plan_clones(self): + """#1743 零回归：预览 count=1 且有源 plan 仍走 clone（不 reselect）。""" + from app.api.routes.generation_preview import create_preview_generation_task + + task = _make_task(task_id="task_1", source_plan_id="source_plan") + repo = _repo_mock() + with patch("app.api.routes.generation_preview.CreateGenerationTaskUseCase") as MockUC: + MockUC.return_value.execute.return_value = task + with patch("app.api.routes.generation_preview.safe_enqueue_generation_task", return_value=True): + with patch("app.services.edit_plan_service.EditPlanService") as MockPlanSvc: + MockPlanSvc.return_value.clone_plan_for_variant.return_value = MagicMock(id="clone_1") + resp = create_preview_generation_task( + _make_preview_request(preview_count=1), + authenticated_user=_make_user(), + generation_task_repository=repo, + db=MagicMock(), + ) + assert resp.total == 1 + MockPlanSvc.return_value.clone_plan_for_variant.assert_called_once() + MockPlanSvc.return_value.reselect_plan_for_variant.assert_not_called() + assert task.source_edit_plan_id == "clone_1" + # ════════════════════════════════════════════════════════════════════════════ # 批量正式生成：变体配置注入 @@ -366,7 +435,7 @@ class TestBatchPreviewRoute: class TestBatchGenerationVariantConfig: """POST /tasks count=N 时变体独立配置。""" - def _call_create_tasks(self, request, repo=None): + def _call_create_tasks(self, request, repo=None, db_latest_plan=None): from app.api.routes.generation_tasks import create_generation_task repo = repo or MagicMock() @@ -379,9 +448,9 @@ class TestBatchGenerationVariantConfig: asset_repo = MagicMock() asset_repo.find_by_id.return_value = None - # db.query().filter()...first() 返回 None：不走兜底关联编辑计划 + # db.query().filter()...first()：db_latest_plan 非空时模拟模板兜底查到最新 plan db = MagicMock() - db.query.return_value.filter.return_value.order_by.return_value.first.return_value = None + db.query.return_value.filter.return_value.order_by.return_value.first.return_value = db_latest_plan return create_generation_task( request, @@ -407,20 +476,29 @@ class TestBatchGenerationVariantConfig: t.title_config = cmd.title_config t.voice_library_id = cmd.voice_library_id t.cover_url = cmd.cover_url + t.source_edit_plan_id = cmd.source_edit_plan_id # #1743：usecase 落库关联 plan return t MockUC.return_value.execute.side_effect = _execute with patch.object(routes, "safe_enqueue_generation_task", return_value=True): - req = CreateGenerationTaskRequest( - template_id="tpl_1", - asset_ids=["a1"], - count=3, - title_config={"font": "宋体"}, - titles=["成片标题1", "成片标题2", "成片标题3"], - voice_library_ids=["v1", "v2", "v3"], - cover_urls=["http://c1", "http://c2", "http://c3"], - ) - resp = self._call_create_tasks(req) + # #1743：count>1 必须有源 plan，变体 1..N-1 走 reselect 独立选片 + with patch("app.services.edit_plan_service.EditPlanService") as MockPlanSvc: + MockPlanSvc.return_value.reselect_plan_for_variant.side_effect = [ + MagicMock(id="reselect_1"), + MagicMock(id="reselect_2"), + ] + req = CreateGenerationTaskRequest( + template_id="tpl_1", + asset_ids=["a1"], + count=3, + source_edit_plan_id="source_plan", + title_config={"font": "宋体"}, + titles=["成片标题1", "成片标题2", "成片标题3"], + voice_library_ids=["v1", "v2", "v3"], + cover_urls=["http://c1", "http://c2", "http://c3"], + ) + resp = self._call_create_tasks(req) + assert MockPlanSvc.return_value.reselect_plan_for_variant.call_count == 2 assert resp.total == 3 assert [c.title_config["text"] for c in captured] == ["成片标题1", "成片标题2", "成片标题3"] assert [c.voice_library_id for c in captured] == ["v1", "v2", "v3"] @@ -467,21 +545,124 @@ class TestBatchGenerationVariantConfig: def _execute(cmd): captured.append(cmd) - return tasks[len(captured) - 1] + t = tasks[len(captured) - 1] + t.source_edit_plan_id = cmd.source_edit_plan_id # #1743：usecase 落库关联 plan + return t MockUC.return_value.execute.side_effect = _execute with patch.object(routes, "safe_enqueue_generation_task", return_value=True): - req = CreateGenerationTaskRequest( - template_id="tpl_1", - asset_ids=["a1"], - count=3, - voice_library_ids=["shared_voice"], - cover_urls=["http://shared"], - ) - self._call_create_tasks(req) + # #1743：count>1 必须有源 plan，变体 1..N-1 走 reselect 独立选片 + with patch("app.services.edit_plan_service.EditPlanService") as MockPlanSvc: + MockPlanSvc.return_value.reselect_plan_for_variant.side_effect = [ + MagicMock(id="reselect_1"), + MagicMock(id="reselect_2"), + ] + req = CreateGenerationTaskRequest( + template_id="tpl_1", + asset_ids=["a1"], + count=3, + source_edit_plan_id="source_plan", + voice_library_ids=["shared_voice"], + cover_urls=["http://shared"], + ) + self._call_create_tasks(req) + assert MockPlanSvc.return_value.reselect_plan_for_variant.call_count == 2 assert all(c.voice_library_id == "shared_voice" for c in captured) assert all(c.cover_url == "http://shared" for c in captured) + def test_count3_template_fallback_plan_used(self): + """#1743：未传 source_edit_plan_id 时，模板兜底查到最新 plan 即作为批量源 plan。""" + from app.api.routes import generation_tasks as routes + + tasks = [_make_task(task_id=f"gen_fb_{i}") for i in range(3)] + + def _execute(cmd): + t = tasks[len([c for c in getattr(_execute, "caps", [])])] + t.source_edit_plan_id = cmd.source_edit_plan_id + _execute.caps.append(cmd) + return t + + _execute.caps = [] + + latest = MagicMock(id="fallback_plan_id") + req = CreateGenerationTaskRequest(template_id="tpl_1", asset_ids=["a1"], count=3) + + with patch.object(routes, "CreateGenerationTaskUseCase") as MockUC: + MockUC.return_value.execute.side_effect = _execute + with patch.object(routes, "safe_enqueue_generation_task", return_value=True): + with patch("app.services.edit_plan_service.EditPlanService") as MockPlanSvc: + MockPlanSvc.return_value.reselect_plan_for_variant.side_effect = [ + MagicMock(id="reselect_1"), + MagicMock(id="reselect_2"), + ] + self._call_create_tasks(req, db_latest_plan=latest) + + # 兜底 plan 被用作源；变体0关联兜底 plan，变体1/2关联 reselect plan + assert _execute.caps[0].source_edit_plan_id == "fallback_plan_id" + assert _execute.caps[1].source_edit_plan_id == "reselect_1" + assert _execute.caps[2].source_edit_plan_id == "reselect_2" + + def test_count3_reselect_value_error_returns_400(self): + """#1743：reselect 素材不足（ValueError）→ 400 明确报错，零任务入队。""" + from app.api.routes import generation_tasks as routes + from fastapi import HTTPException + + enqueue = MagicMock(return_value=True) + req = CreateGenerationTaskRequest( + template_id="tpl_1", asset_ids=["a1"], count=3, source_edit_plan_id="source_plan" + ) + with patch.object(routes, "CreateGenerationTaskUseCase") as MockUC: + MockUC.return_value.execute.side_effect = lambda cmd: _make_task(task_id="should_not_run") + with patch.object(routes, "safe_enqueue_generation_task", enqueue): + with patch("app.services.edit_plan_service.EditPlanService") as MockPlanSvc: + MockPlanSvc.return_value.reselect_plan_for_variant.side_effect = ValueError("素材池为空") + with pytest.raises(HTTPException) as exc: + self._call_create_tasks(req) + assert exc.value.status_code == 400 + assert "无法独立选片" in exc.value.detail + enqueue.assert_not_called() + + def test_count3_reselect_retry_exhausted_returns_500(self): + """#1743：reselect 连续 2 次都非 ValueError 失败 → 500，零任务入队。""" + from app.api.routes import generation_tasks as routes + from fastapi import HTTPException + + enqueue = MagicMock(return_value=True) + req = CreateGenerationTaskRequest( + template_id="tpl_1", asset_ids=["a1"], count=3, source_edit_plan_id="source_plan" + ) + with patch.object(routes, "CreateGenerationTaskUseCase") as MockUC: + MockUC.return_value.execute.side_effect = lambda cmd: _make_task(task_id="should_not_run") + with patch.object(routes, "safe_enqueue_generation_task", enqueue): + with patch("app.services.edit_plan_service.EditPlanService") as MockPlanSvc: + MockPlanSvc.return_value.reselect_plan_for_variant.side_effect = RuntimeError("db down") + with pytest.raises(HTTPException) as exc: + self._call_create_tasks(req) + assert exc.value.status_code == 500 + enqueue.assert_not_called() + + def test_count3_missing_plan_after_prebuild_raises_500(self): + """#1743 兜底守卫：任务落库时 plan 丢失（usecase 未透传）→ 500，严禁静默共用。""" + from app.api.routes import generation_tasks as routes + from fastapi import HTTPException + + req = CreateGenerationTaskRequest( + template_id="tpl_1", asset_ids=["a1"], count=3, source_edit_plan_id="source_plan" + ) + with patch.object(routes, "CreateGenerationTaskUseCase") as MockUC: + # usecase 返回的任务 source_edit_plan_id 为空（模拟落库丢 plan） + MockUC.return_value.execute.side_effect = lambda cmd: _make_task(task_id="lost_plan") + with patch.object(routes, "safe_enqueue_generation_task", return_value=True): + with patch("app.services.edit_plan_service.EditPlanService") as MockPlanSvc: + MockPlanSvc.return_value.reselect_plan_for_variant.side_effect = [ + MagicMock(id="reselect_1"), + MagicMock(id="reselect_2"), + ] + with pytest.raises(HTTPException) as exc: + self._call_create_tasks(req) + assert exc.value.status_code == 500 + assert "变体剪辑计划缺失" in exc.value.detail + class TestVariantValueHelper: """_variant_value 取值逻辑。""" diff --git a/tests/unit/test_batch_dedup_helpers_1743.py b/tests/unit/test_batch_dedup_helpers_1743.py new file mode 100644 index 000000000..e41f85e6c --- /dev/null +++ b/tests/unit/test_batch_dedup_helpers_1743.py @@ -0,0 +1,244 @@ +"""#1743 dedup_helpers 批次查重 + worker 重选 plan 重试函数测试。 + +覆盖： +- 批次任务且无历史重复时走 check_batch_duplicate，batch_similarity 透传到返回 dict +- 批次任务历史已重复 → 不再做批次查重，is_duplicate=True / duplicate_of 透传 +- 非批次任务 batch_similarity 恒为 None +- _reselect_plan_for_batch_retry：成功返回新 plan_id；异常返回 None（保留首版） +""" + +from __future__ import annotations + +import os +import sys +from pathlib import Path +from unittest.mock import MagicMock, patch + +import pytest + +REPO_ROOT = Path(__file__).resolve().parents[2] +for sub in ("apps/worker", "apps/api", "packages", ""): + p = str(REPO_ROOT / sub) if sub else str(REPO_ROOT) + if p not in sys.path: + sys.path.insert(0, p) + +from sqlalchemy import create_engine # noqa: E402 +from sqlalchemy.orm import sessionmaker # noqa: E402 + +os.environ.setdefault("DATABASE_URL", "sqlite:///:memory:") +from packages.adapters.sqlalchemy_impl.models import Base # noqa: E402 + + +@pytest.fixture +def session(): + engine = create_engine("sqlite:///:memory:") + Base.metadata.create_all(engine) + s = sessionmaker(bind=engine)() + try: + yield s + finally: + s.close() + + +def _dedup_kwargs(batch_id): + return dict( + generation_task_id="task-batch-1", + project_id="proj-1", + user_id="user-1", + batch_id=batch_id, + file_url="https://oss.example.com/v.mp4", + file_size=1024, + duration=10.0, + video_path="/tmp/fake.mp4", + mode="edit_plan", + width=1280, + height=720, + fps=25.0, + ) + + +class TestBatchDedupInHelpers: + """dedup_helpers 内部 lazy import video_processing.dedup（依赖 cv2）， + 复用 test_generated_video_creation_logic 的 mock 模块注册模式。""" + + @classmethod + def setup_class(cls): + import sys + + if "cv2" not in sys.modules: + sys.modules["cv2"] = MagicMock() + + # 保存 setup 前的真实模块引用，teardown 原样还原（沙箱无 cv2 时 reimport 会失败， + # 若直接 pop 掉 MagicMock 会让后续测试（如 test_dedup_1702）拿到残缺模块） + import video_processing + + cls._orig_dedup = sys.modules.get("video_processing.dedup") + cls._orig_dedup_attr = getattr(video_processing, "dedup", None) + cls._orig_thumb = sys.modules.get("video_processing.thumbnail_generator") + cls._orig_thumb_attr = getattr(video_processing, "thumbnail_generator", None) + + mock_dedup = MagicMock() + mock_dedup.VideoDeduplicator = MagicMock() + sys.modules["video_processing.dedup"] = mock_dedup + + mock_thumb = MagicMock() + mock_thumb.extract_first_frame = MagicMock() + sys.modules["video_processing.thumbnail_generator"] = mock_thumb + + video_processing.dedup = mock_dedup + video_processing.thumbnail_generator = mock_thumb + + @classmethod + def teardown_class(cls): + import sys + + import video_processing + + # 还原 setup 前状态：原本有真模块→放回；原本没有→移除 mock + if cls._orig_dedup is not None: + sys.modules["video_processing.dedup"] = cls._orig_dedup + else: + sys.modules.pop("video_processing.dedup", None) + if cls._orig_dedup_attr is not None: + video_processing.dedup = cls._orig_dedup_attr + elif hasattr(video_processing, "dedup"): + delattr(video_processing, "dedup") + + if cls._orig_thumb is not None: + sys.modules["video_processing.thumbnail_generator"] = cls._orig_thumb + else: + sys.modules.pop("video_processing.thumbnail_generator", None) + if cls._orig_thumb_attr is not None: + video_processing.thumbnail_generator = cls._orig_thumb_attr + elif hasattr(video_processing, "thumbnail_generator"): + delattr(video_processing, "thumbnail_generator") + + def test_batch_similarity_returned_when_batch_duplicate_found(self, session): + """批次任务 + 无历史重复 + 批次查重命中 → 返回 batch_similarity 与 is_duplicate。""" + from video_processing.dedup_helpers import create_video_record_and_dedup + + with patch("video_processing.dedup.VideoDeduplicator") as mock_cls: + dedup = mock_cls.return_value + dedup.compute_fingerprint.return_value = MagicMock(to_dict=lambda: {}) + dedup.check_duplicate.return_value = None # 历史无重复 + dedup.check_batch_duplicate.return_value = { + "duplicate_of": "video-existing", + "reason": "batch_similar", + "similarity": 0.601, + } + dedup.compute_duplicate_rate.return_value = { + "duplicate_rate": 0.0, + "visual_similarity": 0.0, + "match_count": 0, + } + + result = create_video_record_and_dedup(session=session, **_dedup_kwargs("batch-abc")) + + assert result["video_count"] == 1 + assert result["batch_similarity"] == pytest.approx(0.601) + assert result["is_duplicate"] is True + assert result["duplicate_of"] == "video-existing" + # 批次查重确实被调用（历史查重为 None 才走批次） + dedup.check_batch_duplicate.assert_called_once() + + def test_batch_check_skipped_when_historical_duplicate(self, session): + """历史查重已命中 → 不再批次查重，batch_similarity 为 None。""" + from video_processing.dedup_helpers import create_video_record_and_dedup + + with patch("video_processing.dedup.VideoDeduplicator") as mock_cls: + dedup = mock_cls.return_value + dedup.compute_fingerprint.return_value = MagicMock(to_dict=lambda: {}) + dedup.check_duplicate.return_value = { + "duplicate_of": "video-old", + "reason": "global", + "similarity": 0.85, + } + dedup.compute_duplicate_rate.return_value = { + "duplicate_rate": 85.0, + "visual_similarity": 0.85, + "match_count": 3, + } + + result = create_video_record_and_dedup(session=session, **_dedup_kwargs("batch-abc")) + + assert result["is_duplicate"] is True + assert result["duplicate_of"] == "video-old" + assert result["batch_similarity"] is None + dedup.check_batch_duplicate.assert_not_called() + + def test_non_batch_never_runs_batch_check(self, session): + """非批次任务（batch_id 为空）→ 不调用批次查重，batch_similarity None。""" + from video_processing.dedup_helpers import create_video_record_and_dedup + + with patch("video_processing.dedup.VideoDeduplicator") as mock_cls: + dedup = mock_cls.return_value + dedup.compute_fingerprint.return_value = MagicMock(to_dict=lambda: {}) + dedup.check_duplicate.return_value = None + dedup.check_batch_duplicate.return_value = {"similarity": 0.99} + dedup.compute_duplicate_rate.return_value = { + "duplicate_rate": 0.0, + "visual_similarity": 0.0, + "match_count": 0, + } + + result = create_video_record_and_dedup(session=session, **_dedup_kwargs("")) + + assert result["video_count"] == 1 + assert result["batch_similarity"] is None + assert result["is_duplicate"] is False + dedup.check_batch_duplicate.assert_not_called() + + def test_batch_no_duplicate_returns_none_similarity(self, session): + """批次任务但批次查重也未命中 → batch_similarity None（驱动不重渲）。""" + from video_processing.dedup_helpers import create_video_record_and_dedup + + with patch("video_processing.dedup.VideoDeduplicator") as mock_cls: + dedup = mock_cls.return_value + dedup.compute_fingerprint.return_value = MagicMock(to_dict=lambda: {}) + dedup.check_duplicate.return_value = None + dedup.check_batch_duplicate.return_value = None + dedup.compute_duplicate_rate.return_value = { + "duplicate_rate": 0.0, + "visual_similarity": 0.0, + "match_count": 0, + } + + result = create_video_record_and_dedup(session=session, **_dedup_kwargs("batch-xyz")) + + assert result["batch_similarity"] is None + assert result["is_duplicate"] is False + + +class TestReselectPlanForBatchRetry: + def test_success_returns_new_plan_id(self): + """重选成功 → 返回新 plan_id。""" + from worker_app.tasks.generation import _reselect_plan_for_batch_retry + + fake_svc = MagicMock() + fake_svc.reselect_plan_for_variant.return_value = MagicMock(id="new-plan-999") + with patch("app.services.edit_plan_service.EditPlanService", return_value=fake_svc): + with patch("worker_app.tasks.generation.SessionLocal") as mock_session_local: + mock_session_local.return_value = MagicMock() + new_id = _reselect_plan_for_batch_retry( + "task-1", + "old-plan", + {"task_asset_ids": ["a1", "a2"], "user_id": "u-1"}, + ) + assert new_id == "new-plan-999" + fake_svc.reselect_plan_for_variant.assert_called_once() + args, kwargs = fake_svc.reselect_plan_for_variant.call_args + assert args[0] == "old-plan" + assert args[1] == ["a1", "a2"] + assert kwargs["created_by_user_id"] == "u-1" + + def test_failure_returns_none_and_keeps_first_version(self): + """重选抛异常 → 返回 None（调用方放弃重渲、保留首版），不抛出。""" + from worker_app.tasks.generation import _reselect_plan_for_batch_retry + + fake_svc = MagicMock() + fake_svc.reselect_plan_for_variant.side_effect = RuntimeError("db down") + with patch("app.services.edit_plan_service.EditPlanService", return_value=fake_svc): + with patch("worker_app.tasks.generation.SessionLocal") as mock_session_local: + mock_session_local.return_value = MagicMock() + result = _reselect_plan_for_batch_retry("task-1", "old-plan", {"task_asset_ids": [], "user_id": "u-1"}) + assert result is None diff --git a/tests/unit/test_batch_no_plan_guard.py b/tests/unit/test_batch_no_plan_guard.py index ef29f0bcd..8381d7631 100755 --- a/tests/unit/test_batch_no_plan_guard.py +++ b/tests/unit/test_batch_no_plan_guard.py @@ -1,7 +1,9 @@ -"""AI Review 回归：批量生成 count>1 但 source_edit_plan_id 为空时不应 IndexError。 +"""#1743 批量生成无可用 plan 守卫。 -变体 plan 预克隆仅在 source_edit_plan_id 非空时执行；无源 plan 时 -variant_plan_ids 为空，循环中禁止索引访问，各任务走自身随机选片流程。 +新规则（P0 降重）：count>1 批量生成时必须存在源 plan（前端传入或按模板兜底 +解析到最新 plan），为每个变体独立选片；**无任何可用 plan 时直接 4xx 中断、 +不创建任务**，严禁 N 个任务兜底共用同一 plan 产出同源成片。 +N=1 单视频不受影响（无 plan 时走原有单任务流程）。 """ import sys @@ -9,6 +11,9 @@ from pathlib import Path from types import SimpleNamespace from unittest.mock import MagicMock, patch +import pytest +from fastapi import HTTPException + REPO_ROOT = Path(__file__).resolve().parents[2] if str(REPO_ROOT) not in sys.path: sys.path.insert(0, str(REPO_ROOT)) @@ -18,78 +23,117 @@ def _make_user(): return SimpleNamespace(user=SimpleNamespace(id="user-1")) -def _make_request(count): +def _make_request(count, **overrides): from app.schemas.generation_task import CreateGenerationTaskRequest - return CreateGenerationTaskRequest( + fields = dict( project_id="proj-1", asset_library_id="lib-1", strategy_id="one_take", asset_ids=["a1"], count=count, - source_edit_plan_id="", # 关键：无源 plan（空字符串为假值） + source_edit_plan_id="", ) + fields.update(overrides) + return CreateGenerationTaskRequest(**fields) -class TestBatchNoSourcePlanNoIndexError: - def test_count3_without_source_plan_creates_three_tasks(self): - """count=3 且无 source_edit_plan_id：不克隆、不 IndexError、创建 3 个任务。""" +def _common_patches(latest_plan=None): + """构造通用 patch 上下文（repo/usecase/enqueue 等）。latest_plan 为模板兜底 plan 或 None。""" + repo = MagicMock() + repo.count_pending_by_user.return_value = 0 + repo.count_pending_total.return_value = 0 + repo.create.side_effect = lambda t: t + repo.update.side_effect = lambda t: t + + created = [] + + def _fake_execute(cmd): + task = MagicMock() + task.id = f"task-{len(created) + 1}" + task.source_edit_plan_id = cmd.source_edit_plan_id + task.status = "pending" + task.progress = 0.0 + task.strategy_id = "one_take" + task.error_message = "" + task.cover_url = "" + task.title_config = {} + task.created_at = None + task.batch_id = "batch-1" + created.append(task) + return task + + db = MagicMock() + # 模板兜底查最新 plan：返回 latest_plan（None 表示查不到） + db.query.return_value.filter.return_value.order_by.return_value.first.return_value = latest_plan + + mock_uc = patch("app.api.routes.generation_tasks.CreateGenerationTaskUseCase") + other_patches = [ + patch("app.api.routes.generation_tasks.safe_enqueue_generation_task", return_value=True), + patch("app.api.routes.generation_tasks._writeback_edit_plan_config"), + patch( + "app.api.routes.generation_tasks._resolve_project_and_library", + return_value=("proj-1", ""), + ), + ] + return repo, db, created, mock_uc, other_patches, _fake_execute + + +class TestBatchNoSourcePlanGuard: + def test_count3_without_any_plan_rejects_4xx_and_creates_nothing(self): + """count=3 且无源 plan、模板兜底也查不到 → 400 中断，零任务创建（严禁同源成片）。""" from app.api.routes.generation_tasks import create_generation_task - repo = MagicMock() - repo.count_pending_by_user.return_value = 0 - repo.count_pending_total.return_value = 0 - repo.create.side_effect = lambda t: t - repo.update.side_effect = lambda t: t + repo, db, created, mock_uc, other_patches, fake_exec = _common_patches(latest_plan=None) + MockUC = mock_uc.start() + MockUC.return_value.execute.side_effect = fake_exec + for p in other_patches: + p.start() + all_patches = [mock_uc] + other_patches + try: + with pytest.raises(HTTPException) as exc_info: + create_generation_task( + _make_request(3), + authenticated_user=_make_user(), + generation_task_repository=repo, + project_repository=MagicMock(), + asset_repository=MagicMock(), + asset_library_repository=MagicMock(), + db=db, + ) + assert exc_info.value.status_code == 400 + finally: + for p in reversed(all_patches): + p.stop() + assert len(created) == 0, "无 plan 批量必须零任务创建" - created = [] + def test_count1_without_plan_still_works(self): + """N=1 单视频无 plan 不触发批量守卫（向后兼容，不 4xx）。""" + from app.api.routes.generation_tasks import create_generation_task - def _fake_execute(cmd): - task = MagicMock() - task.id = f"task-{len(created) + 1}" - task.source_edit_plan_id = cmd.source_edit_plan_id - task.status = "pending" - task.progress = 0.0 - task.strategy_id = "one_take" - task.error_message = "" - task.cover_url = None - task.title_config = {} - task.created_at = None - task.batch_id = "batch-1" - created.append(task) - return task - - with patch("app.api.routes.generation_tasks.CreateGenerationTaskUseCase") as MockUC: - MockUC.return_value.execute.side_effect = _fake_execute - with patch( - "app.api.routes.generation_tasks.safe_enqueue_generation_task", - return_value=True, - ): - with patch("app.api.routes.generation_tasks._writeback_edit_plan_config"): - with patch( - "app.api.routes.generation_tasks._resolve_project_and_library", - return_value=("proj-1", ""), - ): - # 核心断言：不得抛 IndexError（变体 plan 索引守卫）。 - # 响应序列化字段与本回归无关，ValidationError 可接受， - # 但 IndexError 必须不出现。 - try: - create_generation_task( - _make_request(3), - authenticated_user=_make_user(), - generation_task_repository=repo, - project_repository=MagicMock(), - asset_repository=MagicMock(), - asset_library_repository=MagicMock(), - db=MagicMock(), - ) - except IndexError as exc: # pragma: no cover - 不应发生 - pytest.fail(f"无源 plan 批量生成触发 IndexError: {exc}") - except Exception: - # 响应序列化等其他异常与本次守卫无关，忽略 - pass - - # 3 个任务全部创建（未因 IndexError 中断） - assert len(created) == 3 - # 无源 plan 时所有任务 source_edit_plan_id 均为空 - assert all(not t.source_edit_plan_id for t in created) + repo, db, created, mock_uc, other_patches, fake_exec = _common_patches(latest_plan=None) + MockUC = mock_uc.start() + MockUC.return_value.execute.side_effect = fake_exec + for p in other_patches: + p.start() + all_patches = [mock_uc] + other_patches + try: + create_generation_task( + _make_request(1), + authenticated_user=_make_user(), + generation_task_repository=repo, + project_repository=MagicMock(), + asset_repository=MagicMock(), + asset_library_repository=MagicMock(), + db=db, + ) + except HTTPException as e: + assert e.status_code != 400, f"N=1 不应被批量守卫拦截: {e.detail}" + except Exception: + # MagicMock 任务对象下游响应序列化可能抛 ValidationError 等，与批量守卫无关； + # 任务已在 usecase.execute 中创建，下方断言 created==1 即证明守卫未拦截。 + pass + finally: + for p in reversed(all_patches): + p.stop() + assert len(created) == 1 diff --git a/tests/unit/test_batch_render_rerender_1743.py b/tests/unit/test_batch_render_rerender_1743.py new file mode 100644 index 000000000..81646aee0 --- /dev/null +++ b/tests/unit/test_batch_render_rerender_1743.py @@ -0,0 +1,80 @@ +"""#1743 worker 批量重渲判定 + 封面哈希选帧纯函数测试。 + +- should_rerender_for_batch_dedup：批次首版查重率 >20% 才重渲（非批次/重渲版/无查重率不重渲） +- pick_batch_cover_index：task_id md5 稳定哈希分散候选帧（同任务稳定、跨任务分散） +""" + +from __future__ import annotations + +import hashlib +import sys +from pathlib import Path + +REPO_ROOT = Path(__file__).resolve().parents[2] +for sub in ("apps/worker", "apps/api", "packages", ""): + p = str(REPO_ROOT / sub) if sub else str(REPO_ROOT) + if p not in sys.path: + sys.path.insert(0, p) + +from worker_app.tasks.generation import ( # noqa: E402 + BATCH_RENDER_SIMILARITY_LIMIT, + pick_batch_cover_index, + should_rerender_for_batch_dedup, +) + + +class TestShouldRerenderForBatchDedup: + def test_non_batch_never_rerenders(self): + """非批次任务（batch_id 为空）任何查重率都不重渲。""" + assert should_rerender_for_batch_dedup(batch_id="", render_attempt=0, batch_similarity=0.99) is False + assert should_rerender_for_batch_dedup(batch_id="", render_attempt=0, batch_similarity=None) is False + + def test_batch_first_version_over_threshold_rerenders(self): + """批次首版（attempt=0）查重率 >20% → 重渲。""" + assert should_rerender_for_batch_dedup(batch_id="batch-1", render_attempt=0, batch_similarity=0.601) is True + assert should_rerender_for_batch_dedup(batch_id="batch-1", render_attempt=0, batch_similarity=0.21) is True + + def test_batch_first_version_under_threshold_no_rerender(self): + """批次首版查重率 ≤20% → 不重渲。""" + assert should_rerender_for_batch_dedup(batch_id="batch-1", render_attempt=0, batch_similarity=0.20) is False + assert should_rerender_for_batch_dedup(batch_id="batch-1", render_attempt=0, batch_similarity=0.0) is False + assert should_rerender_for_batch_dedup(batch_id="batch-1", render_attempt=0, batch_similarity=0.05) is False + + def test_rerendered_version_never_rerenders_again(self): + """重渲版（attempt=1）即使仍超阈值也不再重渲（最多重渲一次）。""" + assert should_rerender_for_batch_dedup(batch_id="batch-1", render_attempt=1, batch_similarity=0.99) is False + + def test_missing_similarity_no_rerender(self): + """查重率缺失（None，非批次查重路径）→ 不重渲。""" + assert should_rerender_for_batch_dedup(batch_id="batch-1", render_attempt=0, batch_similarity=None) is False + + def test_threshold_constant_is_20_percent(self): + assert BATCH_RENDER_SIMILARITY_LIMIT == 0.20 + + +class TestPickBatchCoverIndex: + def test_stable_for_same_task(self): + """同一 task_id 多次调用结果稳定（重试封面不变）。""" + first = pick_batch_cover_index("task-abc", 3) + for _ in range(5): + assert pick_batch_cover_index("task-abc", 3) == first + + def test_zero_or_one_candidate_returns_zero(self): + assert pick_batch_cover_index("task-x", 0) == 0 + assert pick_batch_cover_index("task-x", 1) == 0 + + def test_index_within_range(self): + for i in range(20): + idx = pick_batch_cover_index(f"task-{i}", 3) + assert 0 <= idx < 3 + + def test_matches_md5_formula(self): + """与主流程公式一致：md5(task_id) % 候选数。""" + task_id = "task-formula-check" + expected = int(hashlib.md5(task_id.encode()).hexdigest(), 16) % 3 + assert pick_batch_cover_index(task_id, 3) == expected + + def test_batch_tasks_spread_across_frames(self): + """30 个批次任务在 3 个候选帧上分散（不能全部落在 frame_0——旧 bug 回归守卫）。""" + indexes = {pick_batch_cover_index(f"batch-task-{i}", 3) for i in range(30)} + assert len(indexes) >= 2, f"封面帧应分散到多个帧位，实际全部落在: {indexes}" diff --git a/tests/unit/test_dedup_two_phase_commit.py b/tests/unit/test_dedup_two_phase_commit.py index 54e4e0c97..6f6a5b712 100644 --- a/tests/unit/test_dedup_two_phase_commit.py +++ b/tests/unit/test_dedup_two_phase_commit.py @@ -85,7 +85,10 @@ class TestTwoPhaseCommit: session=session, ) - assert result == 1 + # #1743：返回值由 int 改为 dict（含批次查重信息） + assert isinstance(result, dict) + assert result["video_count"] == 1 + assert result["batch_similarity"] is None # 非批次任务无批次相似度 # create() should be called exactly once with the complete video object mock_repo.create.assert_called_once() created_video = mock_repo.create.call_args[0][0] @@ -122,7 +125,10 @@ class TestTwoPhaseCommit: session=session, ) - assert result == 1 + # #1743：返回值由 int 改为 dict（含批次查重信息） + assert isinstance(result, dict) + assert result["video_count"] == 1 + assert result["batch_similarity"] is None # 非批次任务无批次相似度 mock_repo.create.assert_called_once() created_video = mock_repo.create.call_args[0][0] assert created_video.duplicate_rate is None @@ -161,7 +167,10 @@ class TestTwoPhaseCommit: session=session, ) - assert result == 1 + # #1743：返回值由 int 改为 dict（含批次查重信息） + assert isinstance(result, dict) + assert result["video_count"] == 1 + assert result["batch_similarity"] is None # 非批次任务无批次相似度 mock_repo.create.assert_called_once() created_video = mock_repo.create.call_args[0][0] # Fingerprint should be set @@ -229,6 +238,7 @@ class TestTwoPhaseCommit: session=session, ) - assert result == 0 + # #1743：失败路径返回 dict（video_count=0） + assert result["video_count"] == 0 session.commit.assert_not_called() session.rollback.assert_called_once() diff --git a/tests/unit/test_generated_video_creation_logic.py b/tests/unit/test_generated_video_creation_logic.py index c3e3a47ea..aca7fd112 100755 --- a/tests/unit/test_generated_video_creation_logic.py +++ b/tests/unit/test_generated_video_creation_logic.py @@ -381,7 +381,8 @@ class TestThumbnailInDedupHelpers: thumbnail_url=pre_thumb_url, ) - assert result == 1 + # #1743：返回 dict（非批次 batch_similarity=None） + assert result["video_count"] == 1 from packages.adapters.sqlalchemy_impl.models import GeneratedVideoModel @@ -427,7 +428,8 @@ class TestThumbnailInDedupHelpers: fps=25.0, ) - assert result == 1 + # #1743：返回 dict（非批次 batch_similarity=None） + assert result["video_count"] == 1 from packages.adapters.sqlalchemy_impl.models import GeneratedVideoModel @@ -474,7 +476,7 @@ class TestThumbnailInDedupHelpers: fps=25.0, ) - assert result == 1 # 不阻断 + assert result["video_count"] == 1 # 不阻断（#1743 dict 返回） from packages.adapters.sqlalchemy_impl.models import GeneratedVideoModel diff --git a/tests/unit/test_reselect_plan_variant_1743.py b/tests/unit/test_reselect_plan_variant_1743.py new file mode 100644 index 000000000..d0c997796 --- /dev/null +++ b/tests/unit/test_reselect_plan_variant_1743.py @@ -0,0 +1,206 @@ +"""#1743 EditPlanService.reselect_plan_for_variant 服务层测试。 + +与 clone_plan_for_variant（只重算起点、素材/顺序不变）不同，reselect 完整重跑 +单视频选片：素材池 shuffle + main 片段顺序洗牌 + 起点重选 + 批次 20% 重叠避让。 + +覆盖： +- 独立 plan：新 plan_id 与源不同、命名带后缀、模板/config 复制 +- 新片段经 replace_all_clips_transactional 落库，素材/起点与源 plan 存在差异 +- 源 plan 区间作为批次避让初始对象；record_used_segments 随新片段写回 +- 源 plan 无片段 → ValueError（不创建同源变体） +""" + +from __future__ import annotations + +import os +import sys +from pathlib import Path +from unittest.mock import MagicMock, patch + +import pytest + +os.environ.setdefault("JWT_SECRET_KEY", "unit-test-secret-key-for-testing") +os.environ.setdefault("DATABASE_URL", "sqlite:///test.db") + +REPO_ROOT = Path(__file__).resolve().parents[2] +sys.path.insert(0, str(REPO_ROOT / "apps" / "api")) +sys.path.insert(0, str(Path(__file__).resolve().parent)) # tests/unit + +from test_edit_plan_service import StubEditPlanClipRepository, StubEditPlanRepository, _make_service # noqa: E402 + +from packages.domain.edit_plan_clip import EditPlanClip # noqa: E402 + + +@pytest.fixture +def svc_with_source(): + """带源 plan（4 个 main 片段，素材 a1/a2/a3/a4）+ 2 个批次候选素材的 service。""" + svc = _make_service() + svc._clip_repo.session = MagicMock() + + source = svc.create_plan(template_id="tpl-001", name="9/6-草稿", total_duration=20.0, config={"title": "源配置"}) + for i, aid in enumerate(["a1", "a2", "a3", "a4"]): + clip = EditPlanClip.create( + plan_id=source.id, + clip_type="main", + order=i, + asset_id=aid, + start_time=float(i * 5), + duration=5.0, + text_content=f"文案{i}", + ) + svc._clip_repo.create(clip) + return svc, source + + +def _patch_deps(svc, durations): + """统一 patch reselect 的 DB/素材/历史区间依赖。返回 (patches, mock_replace)。""" + asset_models = [] + for aid, dur in durations.items(): + m = MagicMock(id=aid) + m.duration = dur + m.metadata = None # extract_scene_points_from_metadata(None) → 无场景点 + asset_models.append(m) + + mock_replace = patch.object(svc, "replace_all_clips_transactional", return_value=4) + patches = [ + patch("app.services.edit_plan_service.get_used_segments", return_value={}), + patch("app.services.edit_plan_service.record_used_segments", return_value=None), + patch("packages.domain.plan_generator_utils.extract_scene_points_from_metadata", return_value=[]), + patch("packages.adapters.sqlalchemy_impl.models.AssetModel", create=True), + mock_replace, + ] + started = [] + for p in patches: + started.append(p.start()) + # started[-1] 是 replace_all 的 MagicMock + mock_replace_obj = started[-1] + # db.query(AssetModel).filter(...).all() → 带 duration 的素材 mock + svc._clip_repo.session.query.return_value.filter.return_value.all.return_value = asset_models + return patches, mock_replace_obj + + +class TestReselectPlanForVariant: + def test_creates_independent_plan_with_different_clips(self, svc_with_source): + """reselect 产出新 plan（id/名称不同），片段素材或起点与源 plan 存在差异。""" + import random + + svc, source = svc_with_source + durations = {"a1": 300.0, "a2": 300.0, "a3": 300.0, "a4": 300.0, "a5": 300.0, "a6": 300.0} + patches, mock_replace = _patch_deps(svc, durations) + try: + new_plan = svc.reselect_plan_for_variant( + source.id, + ["a5", "a6"], # 批次素材并入素材池 + created_by_user_id="u-1", + name_suffix="批量2", + rng=random.Random(42), + ) + finally: + for p in reversed(patches): + p.stop() + + # 新 plan 独立、归属/模板/config 复制 + assert new_plan.id != source.id + assert "批量2" in new_plan.name + assert new_plan.template_id == "tpl-001" + assert new_plan.config == {"title": "源配置"} + assert new_plan.created_by_user_id == "u-1" + + # 落库片段数 == 源片段数，且 order 对齐 + clips_data = mock_replace.call_args.args[1] + assert len(clips_data) == 4 + assert [c["order"] for c in clips_data] == [0, 1, 2, 3] + + # 与源 plan 对比：素材序列或起点必须存在差异（降重核心——不是克隆） + source_pairs = [(c.asset_id, round(float(c.start_time), 2)) for c in svc._clip_repo.list_by_plan(source.id)] + new_pairs = [(c["asset_id"], round(float(c["start_time"]), 2)) for c in clips_data] + assert new_pairs != source_pairs, f"reselect 片段应与源 plan 不同，实际相同: {new_pairs}" + # 素材全部来自素材池（源 a1-a4 ∪ 批次 a5-a6） + for c in clips_data: + assert c["asset_id"] in durations + + def test_record_used_segments_called_per_clip(self, svc_with_source): + """每个新片段区间调用 record_used_segments 写回（跨变体/跨任务避让链路）。""" + import random + + svc, source = svc_with_source + durations = {"a1": 300.0, "a2": 300.0, "a3": 300.0, "a4": 300.0, "a5": 300.0} + + asset_models = [] + for aid in durations: + m = MagicMock(id=aid) + m.duration = durations[aid] + m.metadata = None + asset_models.append(m) + svc._clip_repo.session.query.return_value.filter.return_value.all.return_value = asset_models + + started = [ + patch("app.services.edit_plan_service.get_used_segments", return_value={}).start(), + patch("packages.domain.plan_generator_utils.extract_scene_points_from_metadata", return_value=[]).start(), + patch("packages.adapters.sqlalchemy_impl.models.AssetModel", create=True).start(), + patch.object(svc, "replace_all_clips_transactional", return_value=4).start(), + ] + mock_record = patch("app.services.edit_plan_service.record_used_segments", return_value=None).start() + try: + svc.reselect_plan_for_variant( + source.id, ["a5"], created_by_user_id="u-1", name_suffix="批量2", rng=random.Random(5) + ) + finally: + patch.stopall() + + assert mock_record.call_count == 4, "4 个片段应各写一次 used_segment" + for call in mock_record.call_args_list: + args = call.args + assert args[1] in durations, f"asset_id {args[1]} 不在素材池" # asset_id + assert args[3] > args[2], "区间 end 应大于 start" # end > start + assert args[4], "new plan_id 应非空" + + def test_source_clips_seed_batch_avoidance(self, svc_with_source): + """源 plan 片段区间进入批次避让集：与源完全同区间的起点重叠率应超限被避开。""" + import random + + svc, source = svc_with_source + # 素材池只有源素材（极端小池），时长充足 + durations = {"a1": 600.0, "a2": 600.0, "a3": 600.0, "a4": 600.0} + patches, mock_replace = _patch_deps(svc, durations) + try: + new_plan = svc.reselect_plan_for_variant( + source.id, [], created_by_user_id="u-1", name_suffix="批量2", rng=random.Random(99) + ) + finally: + for p in reversed(patches): + p.stop() + + clips_data = mock_replace.call_args.args[1] + source_clips = svc._clip_repo.list_by_plan(source.id) + source_by_asset = {} + for c in source_clips: + source_by_asset.setdefault(c.asset_id, []).append( + (float(c.start_time), float(c.start_time) + float(c.duration)) + ) + + # 同素材新片段与源区间的重叠占比均 ≤20% + from packages.domain.variant_plan_selector import _clip_overlap_ratio + + for c in clips_data: + ratio = _clip_overlap_ratio(c["asset_id"], float(c["start_time"]), float(c["duration"]), source_by_asset) + assert ( + ratio <= 0.20 + 1e-6 + ), f"变体片段与源 plan 同素材区间重叠超限: asset={c['asset_id']} ratio={ratio:.2%}" + assert new_plan.id != source.id + + def test_source_plan_without_clips_raises(self): + """源 plan 无片段 → ValueError（明确报错，不产出同源变体）。""" + svc = _make_service() + svc._clip_repo.session = MagicMock() + empty = svc.create_plan(template_id="tpl-x", name="空计划") + + with pytest.raises(ValueError, match="源 plan 无片段"): + svc.reselect_plan_for_variant(empty.id, ["a1"], created_by_user_id="u-1", name_suffix="变体") + + def test_missing_source_plan_raises(self): + """源 plan 不存在 → get_plan_or_raise 抛错。""" + svc = _make_service() + svc._clip_repo.session = MagicMock() + with pytest.raises((ValueError, KeyError, LookupError)): # get_plan_or_raise 抛错 + svc.reselect_plan_for_variant("not-exist-plan", ["a1"], created_by_user_id="u-1", name_suffix="变体") diff --git a/tests/unit/test_temp_dir_cleanup_race.py b/tests/unit/test_temp_dir_cleanup_race.py index 78cbbc3d9..7aae16a27 100644 --- a/tests/unit/test_temp_dir_cleanup_race.py +++ b/tests/unit/test_temp_dir_cleanup_race.py @@ -51,6 +51,7 @@ def test_generation_cleans_up_temp_dir(): assert "rmtree(render_temp_dir" in source, "generation.py 应清理 render_temp_dir" # 验证清理发生在上传之后（通过查找顺序） - upload_pos = source.find("_upload_and_record") + # #1743：_upload_and_record 拆分为 _upload_rendered_video（仅 OSS 上传） + upload_pos = source.find("_upload_rendered_video") cleanup_pos = source.find("rmtree(render_temp_dir") - assert upload_pos > 0 and cleanup_pos > upload_pos, "清理临时目录应在 _upload_and_record 之后执行" + assert upload_pos > 0 and cleanup_pos > upload_pos, "清理临时目录应在 _upload_rendered_video 之后执行" diff --git a/tests/unit/test_variant_plan_selector_1743.py b/tests/unit/test_variant_plan_selector_1743.py new file mode 100644 index 000000000..753b0a80a --- /dev/null +++ b/tests/unit/test_variant_plan_selector_1743.py @@ -0,0 +1,271 @@ +"""#1743 批量变体独立选片纯核心测试（packages/domain/variant_plan_selector.py）。 + +覆盖： +- 固定种子下 N 次独立选片：素材组合/片段顺序/起点显著不同（降重核心） +- 批次内同素材区间重叠 >20% 触发避让重选 +- 异常输入：空源片段/空素材池/时长全 0 → ValueError（严禁退回同源） +- 非 main 片段（intro/outro/overlay）保留源骨架素材，仅重算起点 +- 跨变体 batch_segments 就地累加（串行选片天然避让） +- 文案/转场/速度等骨架字段透传 +""" + +from __future__ import annotations + +import random +import sys +from pathlib import Path +from unittest.mock import patch + +import pytest + +REPO_ROOT = Path(__file__).resolve().parents[2] +for sub in ("packages", ""): + p = str(REPO_ROOT / sub) if sub else str(REPO_ROOT) + if p not in sys.path: + sys.path.insert(0, p) + +from packages.domain import variant_plan_selector as vps # noqa: E402 +from packages.domain.variant_plan_selector import ( # noqa: E402 + BATCH_CLIP_OVERLAP_LIMIT, + _clip_overlap_ratio, + reselect_clips_for_variant, +) + + +def _source_clips(assets=("a1", "a2", "a3"), dur=5.0, with_non_main=False): + """构造源片段骨架：main 片段若干，可选 intro/outro 固定角色片段。""" + clips = [] + order = 0 + if with_non_main: + clips.append( + { + "order": order, + "asset_id": "intro_asset", + "start_time": 0.0, + "duration": 3.0, + "clip_type": "intro", + "playback_speed": 1.0, + "transition_effect": "fade", + "transition_duration": 0.5, + "text_content": "片头", + "config": {"role": "intro"}, + } + ) + order += 1 + for i, aid in enumerate(assets): + clips.append( + { + "order": order, + "asset_id": aid, + "start_time": float(i * 10), + "duration": dur, + "clip_type": "main", + "playback_speed": 1.2, + "transition_effect": "cut", + "transition_duration": 0.0, + "text_content": f"文案{i}", + "config": {}, + } + ) + order += 1 + if with_non_main: + clips.append( + { + "order": order, + "asset_id": "outro_asset", + "start_time": 0.0, + "duration": 2.0, + "clip_type": "outro", + "playback_speed": 1.0, + "transition_effect": "fade", + "transition_duration": 0.5, + "text_content": "片尾", + "config": {"role": "outro"}, + } + ) + return clips + + +def _durations(asset_ids, total=120.0, extra=None): + d = {a: total for a in asset_ids} + if extra: + d.update(extra) + return d + + +class TestReselectClipsValidation: + def test_empty_source_clips_raises(self): + with pytest.raises(ValueError, match="源 plan 无片段"): + reselect_clips_for_variant( + [], + ["a1"], + asset_durations={"a1": 60.0}, + rng=random.Random(1), + ) + + def test_empty_asset_pool_raises(self): + with pytest.raises(ValueError, match="素材池为空"): + reselect_clips_for_variant( + _source_clips(), + [], + asset_durations={}, + rng=random.Random(1), + ) + + def test_all_zero_duration_raises(self): + with pytest.raises(ValueError, match="时长全部未知"): + reselect_clips_for_variant( + _source_clips(), + ["a1", "a2"], + asset_durations={"a1": 0.0, "a2": 0.0}, + rng=random.Random(1), + ) + + +class TestReselectClipsDifferentiation: + def test_three_variants_differ_in_assets_order_and_starts(self): + """核心验收：固定种子连续 3 次独立选片，素材组合/顺序/起点显著不同。""" + source = _source_clips(assets=("a1", "a2", "a3", "a4")) + pool = ["a1", "a2", "a3", "a4", "a5", "a6"] + durations = _durations(pool, total=300.0) + + batch_segments: dict = {} + variants = [] + for seed in range(3): + clips = reselect_clips_for_variant( + source, + pool, + asset_durations=durations, + batch_segments=batch_segments, # 串行调用：上一变体区间参与避让 + rng=random.Random(100 + seed), + ) + variants.append(clips) + + main_sequences = [] + for clips in variants: + main = [c for c in clips if c["clip_type"] == "main"] + main_sequences.append([(c["asset_id"], round(c["start_time"], 2)) for c in main]) + + # 1) 每个变体片段数与源骨架一致 + for clips in variants: + main_clips = [c for c in clips if c["clip_type"] == "main"] + assert len(main_clips) == 4 + + # 2) 三个变体的素材序列不全相同（素材洗牌 + main 顺序洗牌生效） + seq_sets = {tuple(a for a, _ in seq) for seq in main_sequences} + assert len(seq_sets) >= 2, f"变体素材序列应存在差异，实际全部相同: {seq_sets}" + + # 3) 起点组合不全相同（起点重选生效） + start_sets = {tuple(s for _, s in seq) for seq in main_sequences} + assert len(start_sets) >= 2, f"变体起点组合应存在差异，实际全部相同: {start_sets}" + + # 4) batch_segments 跨变体累加（串行避让链路存在） + total_segments = sum(len(v) for v in batch_segments.values()) + assert total_segments >= 12, f"3 变体 × 4 main 片段应累加 >=12 区间，实际 {total_segments}" + + def test_skeleton_fields_preserved(self): + """文案/转场/速度等骨架字段随片段透传（只换素材与起点）。""" + source = _source_clips(assets=("a1", "a2"), with_non_main=True) + pool = ["a1", "a2", "a3"] + durations = _durations(pool, total=120.0, extra={"intro_asset": 30.0, "outro_asset": 30.0}) + + clips = reselect_clips_for_variant( + source, + pool, + asset_durations=durations, + rng=random.Random(7), + ) + by_order = {c["order"]: c for c in clips} + + # intro/outro 骨架字段保留 + intro = next(c for c in clips if c["clip_type"] == "intro") + outro = next(c for c in clips if c["clip_type"] == "outro") + assert intro["asset_id"] == "intro_asset" + assert intro["text_content"] == "片头" + assert intro["transition_effect"] == "fade" + assert intro["transition_duration"] == 0.5 + assert outro["asset_id"] == "outro_asset" + assert outro["text_content"] == "片尾" + + # main 片段文案/速度随骨架 order 保留 + for c in clips: + if c["clip_type"] == "main": + assert c["playback_speed"] == 1.2 + assert c["text_content"].startswith("文案") + + +class TestBatchOverlapAvoidance: + def test_overlap_ratio_calculation(self): + segs = {"a1": [(10.0, 20.0)]} # 已占 10s 区间 + # 新区间 [10,20) 完全重叠 → 1.0 + assert _clip_overlap_ratio("a1", 10.0, 10.0, segs) == pytest.approx(1.0) + # 新区间 [20,30) 零重叠 → 0.0 + assert _clip_overlap_ratio("a1", 20.0, 10.0, segs) == pytest.approx(0.0) + # 新区间 [15,25) 重叠 5s / 10s → 0.5 + assert _clip_overlap_ratio("a1", 15.0, 10.0, segs) == pytest.approx(0.5) + # 空 asset / 零时长 → 0 + assert _clip_overlap_ratio("", 0.0, 10.0, segs) == 0.0 + assert _clip_overlap_ratio("a1", 10.0, 0.0, segs) == 0.0 + + def test_over_limit_triggers_reselect_to_non_overlapping(self): + """批次已占满素材前段时，避让重选应把起点挪到重叠 ≤20% 的位置。""" + source = _source_clips(assets=("a1",), dur=10.0) + durations = {"a1": 120.0} + # 批次已选区间：a1 [0, 100) 几乎占满前段 + batch_segments = {"a1": [(0.0, 100.0)]} + + # _resolve_start_time 第一次返回高重叠起点（2.0），之后 rng 抖动应找到低重叠位置 + call_count = {"n": 0} + + def fake_resolve(asset_id, clip_duration, asset_durations, used_segments, scene_points=None, on_exhausted=None): + call_count["n"] += 1 + return 2.0 if call_count["n"] == 1 else None # 后续回退 rng.uniform + + with patch.object(vps, "_resolve_start_time", side_effect=fake_resolve): + # rng.uniform 返回 105.0（与 [0,100) 零重叠）；rng 是 random.Random 实例， + # 需 patch 类方法 uniform 才能生效 + with patch.object(random.Random, "uniform", return_value=105.0): + clips = reselect_clips_for_variant( + source, + ["a1"], + asset_durations=durations, + batch_segments=batch_segments, + rng=random.Random(3), + ) + + main = [c for c in clips if c["clip_type"] == "main"] + assert len(main) == 1 + ratio = _clip_overlap_ratio("a1", main[0]["start_time"], 10.0, {"a1": [(0.0, 100.0)]}) + assert ratio <= BATCH_CLIP_OVERLAP_LIMIT, f"避让后重叠应 ≤20%，实际 {ratio:.2%}" + assert main[0]["start_time"] == pytest.approx(105.0, abs=0.01) + + def test_first_variant_segments_become_avoidance_target(self): + """变体 0 选定区间后，变体 1 选同素材时批次区间生效（不与源区间完全重合）。""" + source = _source_clips(assets=("a1", "a2"), dur=8.0) + pool = ["a1", "a2", "a3"] + durations = _durations(pool, total=600.0) + + batch_segments: dict = {} + v1 = reselect_clips_for_variant( + source, pool, asset_durations=durations, batch_segments=batch_segments, rng=random.Random(11) + ) + v1_main = [(c["asset_id"], c["start_time"], c["duration"]) for c in v1 if c["clip_type"] == "main"] + + # 快照 v1 之后的批次避让集（v2 调用会就地追加 v2 自身区间，断言必须用调用前快照） + import copy + + batch_snapshot = copy.deepcopy(batch_segments) + + v2 = reselect_clips_for_variant( + source, pool, asset_durations=durations, batch_segments=batch_segments, rng=random.Random(12) + ) + # v2 与 v1（快照）同素材片段的区间重叠占比均 ≤20% + for c in v2: + if c["clip_type"] != "main" or not c["asset_id"]: + continue + ratio = _clip_overlap_ratio(c["asset_id"], c["start_time"], c["duration"], batch_snapshot) + assert ( + ratio <= BATCH_CLIP_OVERLAP_LIMIT + 1e-6 + ), f"变体间片段重叠超限: asset={c['asset_id']} start={c['start_time']} ratio={ratio:.2%}" + # v1 片段确实进入了批次避让集 + assert any(a in batch_segments for a, _, _ in v1_main) -- 2.54.0 From c83febed61993cc9690a68042ac5cc9a1337608a Mon Sep 17 00:00:00 2001 From: saas-backend-agent Date: Sun, 6 Sep 2026 18:01:46 +0800 Subject: [PATCH 2/2] =?UTF-8?q?test(#1743):=20=E9=9B=86=E6=88=90=E6=B5=8B?= =?UTF-8?q?=E8=AF=95=E9=80=82=E9=85=8D=E6=89=B9=E9=87=8F=E6=97=A0plan?= =?UTF-8?q?=E5=AE=88=E5=8D=AB=E2=80=94=E2=80=94=E6=97=A0plan=20400?= =?UTF-8?q?=E4=B8=AD=E6=96=AD+=E6=9C=89=E6=BA=90plan=E7=8B=AC=E7=AB=8B?= =?UTF-8?q?=E5=8F=98=E4=BD=93plan=E6=88=90=E5=8A=9F=E9=93=BE=E8=B7=AF?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- tests/integration/test_generation_api.py | 38 ++++++++++++++++++++---- 1 file changed, 33 insertions(+), 5 deletions(-) diff --git a/tests/integration/test_generation_api.py b/tests/integration/test_generation_api.py index e516dd0d4..697214eff 100755 --- a/tests/integration/test_generation_api.py +++ b/tests/integration/test_generation_api.py @@ -347,8 +347,9 @@ class TestCreateGenerationTask: assert mock_celery.send_task.call_args[0][0] == "worker.generate_video" @patch("app.core.task_enqueue.celery_app") - def test_create_batch_tasks(self, mock_celery, client): - """批量创建多个生成任务。""" + def test_create_batch_tasks_without_plan_rejected(self, mock_celery, client): + """#1743：批量 count=3 但无剪辑计划（未传 source_edit_plan_id 且模板兜底无 plan） + → 400 中断，严禁 N 任务兜底共用同一 plan 产出同源成片。""" mock_celery.send_task = MagicMock() resp = client.post( @@ -361,17 +362,44 @@ class TestCreateGenerationTask: "count": 3, }, ) - assert resp.status_code == 200 + assert resp.status_code == 400 + assert "预览" in resp.json()["detail"] or "剪辑计划" in resp.json()["detail"] + mock_celery.send_task.assert_not_called() + + @patch("app.core.task_enqueue.celery_app") + def test_create_batch_tasks_with_independent_variant_plans(self, mock_celery, client): + """#1743：批量 count=3 且有源 plan → 变体 0 用源 plan，变体 1/2 各自 reselect + 独立选片，3 个任务关联 3 个不同 plan。""" + mock_celery.send_task = MagicMock() + + with patch("app.services.edit_plan_service.EditPlanService") as MockPlanSvc: + MockPlanSvc.return_value.reselect_plan_for_variant.side_effect = [ + MagicMock(id="variant-plan-1"), + MagicMock(id="variant-plan-2"), + ] + resp = client.post( + "/api/v1/generation/tasks", + json={ + "project_id": "proj-1", + "asset_library_id": "lib-1", + "strategy_id": "strategy-default", + "voice_library_id": "voice-lib-1", + "source_edit_plan_id": "source-plan-1", + "count": 3, + }, + ) + assert resp.status_code == 200, resp.text data = resp.json() assert len(data["items"]) == 3 assert data["total"] == 3 - # 验证所有任务都有不同的 ID task_ids = [t["id"] for t in data["items"]] assert len(set(task_ids)) == 3 - # 同一批次应有相同的 batch_id + # 同一批次 batch_id 相同 batch_ids = [t["batch_id"] for t in data["items"] if t["batch_id"]] assert len(batch_ids) == 3 assert len(set(batch_ids)) == 1 + # 变体 1/2 各自独立选片 + assert MockPlanSvc.return_value.reselect_plan_for_variant.call_count == 2 def test_create_task_project_not_found(self, client): """项目不存在返回 404。""" -- 2.54.0