diff --git a/apps/api/app/api/routes/assets.py b/apps/api/app/api/routes/assets.py index 814b4a6c8..3a407e81b 100755 --- a/apps/api/app/api/routes/assets.py +++ b/apps/api/app/api/routes/assets.py @@ -26,6 +26,7 @@ from app.schemas.asset import ( UpdateAssetReviewRequest, ) from app.schemas.tag import TagAssetsRequest +from app.services.asset_segment_tracker import compute_asset_availability from fastapi import APIRouter, Depends, HTTPException, Query, Response from packages.domain.smart_match import smart_select_assets @@ -35,6 +36,23 @@ logger = logging.getLogger(__name__) router = APIRouter() +def _asset_availability_fields(item) -> dict: + """视频素材返回余量四字段；非视频/无时长/异常时返回 None + usable=True（零影响）。""" + try: + info = compute_asset_availability(item) + except Exception: + logger.warning("计算素材余量失败，按可用处理: asset_id=%s", getattr(item, "id", "?"), exc_info=True) + info = None + if info is None: + return { + "used_duration": None, + "available_duration": None, + "used_ratio": None, + "usable": True, + } + return info + + def _to_asset_response(item, storage_service=None) -> AssetResponse: # 生成签名文件 URL（用于视频播放 / 文件下载） file_url = None @@ -79,6 +97,7 @@ def _to_asset_response(item, storage_service=None) -> AssetResponse: created_at=format_utc_datetime(item.created_at), uploaded_by_user_id=item.uploaded_by_user_id, tag_ids=getattr(item, "tag_ids", []), + **_asset_availability_fields(item), ) @@ -569,13 +588,35 @@ def smart_match_assets( kind=None, ) + # 结果层过滤：usable=false（零重复可切区间耗尽且历史区间均达复用上限）的素材 + # 不返回给前端；不动 smart_select_assets 评分逻辑本身 + filtered_results = [] + for r in results: + try: + avail = compute_asset_availability(r.asset) + except Exception: + logger.warning( + "smart-match 余量计算失败，按可用处理: asset_id=%s", + getattr(r.asset, "id", "?"), + exc_info=True, + ) + avail = None + if avail is not None and not avail["usable"]: + logger.info( + "smart-match 排除已用尽素材: asset_id=%s name=%s", + getattr(r.asset, "id", "?"), + getattr(r.asset, "name", ""), + ) + continue + filtered_results.append(r) + items = [ SmartMatchItem( asset=_to_asset_response(r.asset), score=r.score, breakdown=r.breakdown, ) - for r in results + for r in filtered_results ] return SmartMatchResponse(items=items, total_candidates=total_candidates) diff --git a/apps/api/app/api/routes/generation_preview.py b/apps/api/app/api/routes/generation_preview.py index 03be31208..91f2458dc 100755 --- a/apps/api/app/api/routes/generation_preview.py +++ b/apps/api/app/api/routes/generation_preview.py @@ -358,6 +358,41 @@ def create_preview_generation_task( exc_info=True, ) + # 每条预览都关联独立克隆 plan：多预览前端为 N 次并发调用，若共用同一 plan + # 则 N 条预览片段完全相同；克隆时片段起点按持久化历史区间重算（含受控复用）， + # 保证各预览版本内容不同 + if task.source_edit_plan_id: + try: + from app.services.edit_plan_service import EditPlanService + + _plan_svc = EditPlanService(db) + _preview_plan = _plan_svc.clone_plan_for_variant( + task.source_edit_plan_id, + created_by_user_id=user_id, + name_suffix="预览变体", + ) + task.source_edit_plan_id = _preview_plan.id + generation_task_repository.update(task) + logger.info( + "[预览生成] 预览关联独立克隆 plan: task_id=%s clone_plan_id=%s", + task.id, + _preview_plan.id, + ) + except Exception as clone_err: + # 不退回共用原 plan（否则多条预览内容相同，违反去重诉求）： + # 标记任务失败并中断，前端可重新发起预览 + logger.error( + "[预览生成] 克隆预览变体 plan 失败，任务标记失败: task_id=%s error=%s", + task.id, + clone_err, + exc_info=True, + ) + _mark_task_failed(generation_task_repository, task, "预览变体计划创建失败") + raise HTTPException( + status_code=500, + detail="创建预览任务失败：无法生成独立剪辑计划，请重试", + ) from clone_err + # 入队执行；若入队失败则标记任务为 failed 避免僵尸数据 try: if not safe_enqueue_generation_task( diff --git a/apps/api/app/api/routes/generation_tasks.py b/apps/api/app/api/routes/generation_tasks.py index 13e361ba0..7b4a476c8 100755 --- a/apps/api/app/api/routes/generation_tasks.py +++ b/apps/api/app/api/routes/generation_tasks.py @@ -418,8 +418,56 @@ 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（否则批量视频内容重复，违反去重诉求）。 + variant_plan_ids: list[str] = [] + if count > 1 and request.source_edit_plan_id: + from app.services.edit_plan_service import EditPlanService + + _plan_svc = EditPlanService(db) + 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, + created_by_user_id=user_id, + name_suffix=f"批量{task_index + 1}", + ) + break + except Exception as clone_err: # noqa: PERF203 + last_err = clone_err + logger.warning( + "[生成任务] 克隆变体 plan 失败(尝试%d/2): source=%s error=%s", + _attempt + 1, + request.source_edit_plan_id, + clone_err, + exc_info=True, + ) + if variant is None: + logger.error( + "[生成任务] 克隆变体 plan 重试仍失败，中断批量创建: source=%s", + request.source_edit_plan_id, + exc_info=last_err, + ) + raise HTTPException( + status_code=500, + detail="创建批量任务失败：无法生成独立剪辑计划，请重试", + ) from last_err + variant_plan_ids.append(variant.id) + try: - for _ in range(count): + 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 + if task_index > 0 and variant_plan_ids: + effective_plan_id = variant_plan_ids[task_index - 1] + task = use_case.execute( CreateGenerationTaskCommand( project_id=project_id, @@ -431,7 +479,7 @@ def create_generation_task( title_ids=request.title_ids, voice_ids=request.voice_ids, created_by_user_id=user_id, - source_edit_plan_id=request.source_edit_plan_id, + source_edit_plan_id=effective_plan_id, asset_select_mode=request.asset_select_mode, batch_id=batch_id, video_title=request.video_title, diff --git a/apps/api/app/api/routes/templates_editor/clips.py b/apps/api/app/api/routes/templates_editor/clips.py index b1289970a..8f25ddcb0 100755 --- a/apps/api/app/api/routes/templates_editor/clips.py +++ b/apps/api/app/api/routes/templates_editor/clips.py @@ -24,8 +24,10 @@ from app.auth import AuthenticatedUser, get_current_user from app.core.storage import get_storage_service from app.dependencies import get_asset_repository, get_db_session from app.services.asset_segment_tracker import ( + REUSE_RATIO_LIMIT, + SEGMENT_EDGE_GAP, get_used_segments, - make_reset_callback, + make_reuse_callback, record_used_segments, remove_used_segment, ) @@ -436,11 +438,16 @@ def _recommended_time_conflicts( start: float, duration: float, used: list[tuple[float, float]], + edge_gap: float = SEGMENT_EDGE_GAP, ) -> bool: - """检查推荐起始时间是否与已使用时间段冲突.""" + """检查推荐起始时间是否与已使用时间段冲突. + + 冲突检测统一加 ``edge_gap`` 秒边缘间隙：已用区间按 [s-gap, e+gap] 扩边后判定， + 避免推荐片段与已用片段首尾紧贴导致画面观感重复。 + """ end = start + duration for used_start, used_end in used: - if start < used_end and end > used_start: + if start < used_end + edge_gap and end > used_start - edge_gap: return True return False @@ -611,55 +618,98 @@ def create_clips_from_assets_editor( used_segments: dict[str, list[tuple[float, float]]] = get_used_segments( db, unique_asset_ids ) - reset_cb = make_reset_callback(db, used_segments) + # 受控复用回调：可用区间耗尽时复用最久未用且未达复用上限（3次）的历史区间， + # 复用片段时长累加到 reused_durations 供 15% 占比控制 + reused_durations: dict[str, float] = {} + # 本条成片中每个素材被分配的片段总时长（复用占比分母） + asset_assigned_durations: dict[str, float] = {} + # 受控复用回调：区间耗尽时复用最久未用且 use_count<3 的历史区间； + # 回调内部预判复用后占比是否超 15%，超限拒绝复用（返回 None） + reuse_cb = make_reuse_callback( + db, + asset_durations, + reused_durations, + assigned_tracker=asset_assigned_durations, + ) clips_data: list[dict] = [] + def _reuse_ratio_exceeded(aid: str, extra: float = 0.0) -> bool: + """该素材在本条成片中「已复用片段时长 / 已分配片段总时长」是否已超 15%。 + + 在为下一片段选素材时调用：本片段尚未分配，复用状态只在分配后的回调里 + 更新，因此直接检查当前占比——一旦已超 15%，该素材不再参与后续分配。 + assigned=0（首个片段）放行；reused=0（尚未发生复用）时不误拦正常分配。 + """ + assigned = asset_assigned_durations.get(aid, 0.0) + if assigned <= 0: + return False + return reused_durations.get(aid, 0.0) / assigned > REUSE_RATIO_LIMIT + for i, (_seg_order, dur_min, dur_max) in enumerate(segments): - # 轮询分配素材 - asset_id = body.asset_ids[i % len(body.asset_ids)] - asset_total = asset_durations.get(asset_id, 0.0) - - # 素材时长为 0 或缺失时无法创建有效片段 - if asset_total <= 0: - raise HTTPException( - status_code=status.HTTP_400_BAD_REQUEST, - detail=f"素材 {asset_id} 时长信息缺失或为0，无法创建片段", - ) - # 在 segment 的 duration_min ~ duration_max 之间随机取值（保留一位小数） raw_duration = random.uniform(dur_min, dur_max) - clip_duration = round(raw_duration, 1) - # 素材时长不足时缩短 clip duration - clip_duration = min(clip_duration, asset_total) - - if clip_duration <= 0: - raise HTTPException( - status_code=status.HTTP_400_BAD_REQUEST, - detail=f"素材 {asset_id} 时长不足，无法创建有效片段", + # 轮询分配素材：跳过时长缺失、复用占比已超 15% 阈值的素材； + # 选中后计算起点，若该素材可用区间耗尽且复用被闸门拒绝（calc 返回 None）， + # 继续轮询下一个素材 + asset_id = "" + clip_duration = 0.0 + start_time: float | None = None + n_assets = len(body.asset_ids) + for offset in range(n_assets): + candidate = body.asset_ids[(i + offset) % n_assets] + candidate_total = asset_durations.get(candidate, 0.0) + if candidate_total <= 0: + continue + candidate_duration = min(round(raw_duration, 1), candidate_total) + if candidate_duration <= 0: + continue + if _reuse_ratio_exceeded(candidate, candidate_duration): + logger.info( + "from-assets 素材复用占比超 %.0f%% 阈值，跳过分配: asset_id=%s", + REUSE_RATIO_LIMIT * 100, + candidate, + ) + continue + # 随机起始时间（不调用 MediaKit，保证接口快速返回）；100 次避不开 + # 历史区间时走受控复用回调（复用片段累加 reused_durations，回调内部 + # 预判复用后占比超 15% 则拒绝并返回 None） + candidate_start = _calc_random_start_time( + candidate, + candidate_duration, + asset_durations, + used_segments, + on_exhausted=reuse_cb, ) + if candidate_start is None: + # 该素材可用区间耗尽且复用被闸门/use_count 上限拒绝 → 尝试下一素材 + logger.info( + "from-assets 素材无可用可切区间（复用被拒），轮询下一素材: asset_id=%s", + candidate, + ) + continue + asset_id = candidate + clip_duration = candidate_duration + start_time = candidate_start + break - # 使用随机起始时间（不调用MediaKit，保证接口快速返回） - start_time = _calc_random_start_time( - asset_id, - clip_duration, - asset_durations, - used_segments, - on_exhausted=reset_cb, - ) - - if start_time is None: + if not asset_id or start_time is None: + # 所有素材时长缺失、复用占比超阈值，或区间耗尽且复用被拒 → 素材可切区间不足 raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, - detail=f"素材 {asset_id} 时长信息缺失，无法计算起始时间", + detail="素材可切区间不足，请补充新素材", ) # 记录已使用时间段（内存，供本次后续片段避开） used_segments.setdefault(asset_id, []).append( (start_time, start_time + clip_duration) ) + asset_assigned_durations[asset_id] = ( + asset_assigned_durations.get(asset_id, 0.0) + clip_duration + ) # 同步写入素材 metadata（不 commit，与下方 replace_all_clips_transactional - # 处于同一事务，任一步失败整体回滚，不留脏数据） + # 处于同一事务，任一步失败整体回滚，不留脏数据）； + # 复用区间与历史记录高度重叠时 record 内部自动累加 use_count record_used_segments( db, asset_id, start_time, start_time + clip_duration, plan_id ) @@ -766,6 +816,10 @@ def _update_mediakit_recommendations_async( # pragma: no cover (clip.id, clip.start_time, clip.start_time + clip.duration) ) + # 读取素材全部历史已用区间（跨任务/跨 plan 持久化记录）： + # MediaKit 挪点必须与随机选片一样避让历史区间，否则会把片段挪回已用过的画面 + historical_segments = get_used_segments(db, unique_asset_ids) + # 已更新的片段ID（用于排除已移动的旧时间段） updated_clip_ids: set[str] = set() # 已更新的时间段 @@ -807,11 +861,23 @@ def _update_mediakit_recommendations_async( # pragma: no cover if cid != clip.id and cid not in updated_clip_ids ] other_segments.extend(updated_segments.get(asset_id, [])) + # 并入该素材全部历史已用区间（含其他 plan/其他任务），set 去重： + # 本 plan 片段创建时已写入历史记录 + # 并入该素材全部历史已用区间（含其他 plan/其他任务）。 + # set 去重前先归一化精度（round 3 位），避免浮点尾差导致逻辑相同的 + # 区间（如 1.0 与 1.0000000001）被误判为不同区间 + def _norm(segs): + return {(round(float(a), 3), round(float(b), 3)) for a, b in segs} - # 检查是否与同素材其他片段时间段冲突 + other_segments = list( + _norm(other_segments) | _norm(historical_segments.get(asset_id, [])) + ) + + # 检查推荐时间是否与同 plan 片段或历史已用区间冲突（含 0.3s 边缘间隙）： + # 冲突时放弃该推荐、保留原随机起点（不硬挪到已用过的画面） if _recommended_time_conflicts(recommended_start, clip_duration, other_segments): logger.info( - "后台任务: 推荐时间冲突，跳过: asset_id=%s recommended=%.2f", + "后台任务: 推荐时间与同片/历史区间冲突，保留原起点: asset_id=%s recommended=%.2f", asset_id, recommended_start, ) diff --git a/apps/api/app/schemas/asset.py b/apps/api/app/schemas/asset.py index 2532467d0..1f9d2927c 100755 --- a/apps/api/app/schemas/asset.py +++ b/apps/api/app/schemas/asset.py @@ -53,6 +53,14 @@ class AssetResponse(BaseModel): created_at: str uploaded_by_user_id: str tag_ids: list[str] = Field(default_factory=list) + # 片段级余量信息（仅视频素材返回，非视频/无时长记录为 None，前端按可用处理） + used_duration: float | None = Field(default=None, description="已使用片段时长（秒，历史区间合并去重后）") + available_duration: float | None = Field(default=None, description="剩余可用时长（秒）= 素材总时长 - 已用时长") + used_ratio: float | None = Field(default=None, description="已用时长占比（0~1）") + usable: bool = Field( + default=True, + description="是否仍可用于新片段：零重复可切区间耗尽且所有历史区间复用次数" "（use_count）均达上限时为 false", + ) MAX_BATCH_SIZE = 200 diff --git a/apps/api/app/services/asset_segment_tracker.py b/apps/api/app/services/asset_segment_tracker.py index bb13c1c67..56c76545f 100644 --- a/apps/api/app/services/asset_segment_tracker.py +++ b/apps/api/app/services/asset_segment_tracker.py @@ -1,17 +1,27 @@ -"""素材片段级使用记录追踪. +"""素材片段级使用记录追踪与受控复用. 在素材 metadata（assets.classification_result JSON）中持久化已使用的片段时间区间， -供 from-assets 创建片段时避开历史区间，实现跨任务/跨调用的片段去重。 +供 from-assets 创建片段时避开历史区间，实现跨任务/跨调用的片段去重； +素材可用区间耗尽后进入受控复用：允许有限次数（MAX_RANGE_USE_COUNT）复用最久未用 +的历史区间，配合调用方的成片复用占比控制（MAX_REUSE_RATIO = 15%），把任意两条 +成片的画面重复率控制在阈值内。 -metadata 中新增字段 ``used_time_ranges``:: +metadata 中的记录字段 ``used_time_ranges``:: "used_time_ranges": [ - {"start": 12.5, "end": 20.3, "plan_id": "plan-xxx", "created_at": "2026-08-29T12:00:00+00:00"}, + { + "start": 12.5, "end": 20.3, + "plan_id": "plan-xxx", + "created_at": "2026-08-29T12:00:00+00:00", + "use_count": 1, # 该区间累计被使用次数（复用一次 +1） + "last_used_at": "2026-08-29T12:00:00+00:00" # 最近一次使用时间 + }, ... ] 注意：本模块所有函数都不自行 commit，由调用方控制事务边界 （from-assets 与 replace_all_clips_transactional 同事务；异步任务各自 commit）。 +历史记录永不自动清空（自动轮回重置已下线，reset_used_segments 仅保留给运维/测试）。 """ from __future__ import annotations @@ -29,24 +39,51 @@ logger = logging.getLogger(__name__) USED_RANGES_KEY = "used_time_ranges" +# ── 受控复用配置常量 ───────────────────────────────────────────────────────── +MAX_RANGE_USE_COUNT = 3 +"""单条历史区间最多被使用次数（含首次），达到后不再参与复用。""" -def _read_ranges(model: AssetModel) -> list[dict]: +REUSE_RATIO_LIMIT = 0.15 +"""单条成片中，单个素材的复用片段累计时长 / 该素材在成片中的总时长上限（15%）。 +超过则该素材不再分配新片段（调用方在轮询分配时跳过）。""" + +SEGMENT_EDGE_GAP = 0.3 +"""冲突判定边缘间隙（秒）：历史区间按 [start-gap, end+gap] 扩边后参与冲突检测， +避免两条片段首尾紧贴导致画面观感重复；记录仍存实际值。""" + +# 判定"新片段与历史区间为同一次使用（复用）"的重叠率阈值： +# 重叠时长 / 新区间时长超过该比例视为复用该历史区间（累加 use_count）而非新增记录。 +_REUSE_OVERLAP_RATIO = 0.6 + + +def _now_iso() -> str: + return datetime.now(timezone.utc).isoformat() + + +def _read_meta(model: AssetModel) -> dict: """从 AssetModel 读取 metadata dict（classification_result 列承载的 JSON）.""" if not model.classification_result: return {} try: - return json.loads(model.classification_result) + data = json.loads(model.classification_result) + return data if isinstance(data, dict) else {} except Exception: return {} +def _get_model(db: Session, asset_id: str, for_update: bool = False) -> AssetModel | None: + query = db.query(AssetModel).filter(AssetModel.id == asset_id) + if for_update: + # 行级锁（PostgreSQL SELECT ... FOR UPDATE）：序列化同一素材的 + # classification_result 读-改-写，避免并发事务丢失使用记录。 + # SQLite 不支持时 SQLAlchemy 会忽略该子句（no-op）。 + query = query.with_for_update() + return query.first() + + def get_used_segments(db: Session, asset_ids: list[str]) -> dict[str, list[tuple[float, float]]]: """聚合多个素材的历史已用片段区间。 - Args: - db: SQLAlchemy session - asset_ids: 素材 ID 列表 - Returns: ``{asset_id: [(start, end), ...]}`` 格式，与 ``_calc_random_start_time`` 的 ``used_segments`` 参数格式一致，可直接传入。 @@ -56,7 +93,7 @@ def get_used_segments(db: Session, asset_ids: list[str]) -> dict[str, list[tuple result: dict[str, list[tuple[float, float]]] = {} models = db.query(AssetModel).filter(AssetModel.id.in_(list(set(asset_ids)))).all() for model in models: - meta = _read_ranges(model) + meta = _read_meta(model) ranges = meta.get(USED_RANGES_KEY) or [] segments: list[tuple[float, float]] = [] for r in ranges: @@ -76,19 +113,49 @@ def record_used_segments( end: float, plan_id: str, ) -> None: - """向素材 metadata 追加一条片段使用记录（不 commit）.""" - model = db.query(AssetModel).filter(AssetModel.id == asset_id).first() + """记录一次片段使用（不 commit）. + + 若新区间与某条历史区间高度重叠（复用场景，如受控复用回调返回的区间、 + MediaKit 挪到历史区间），则累加该记录的 ``use_count`` 并刷新 ``last_used_at``， + 不新增记录；否则追加一条新记录（use_count=1）。 + """ + # 行级锁读取：与并发生成任务互斥，保证区间记录读-改-写一致 + model = _get_model(db, asset_id, for_update=True) if model is None: logger.warning("[片段追踪] 素材不存在，跳过记录: asset_id=%s", asset_id) return - meta = _read_ranges(model) + meta = _read_meta(model) ranges = list(meta.get(USED_RANGES_KEY) or []) + + new_start = round(float(start), 3) + new_end = round(float(end), 3) + new_dur = max(new_end - new_start, 1e-6) + now = _now_iso() + + for r in ranges: + try: + rs, re_ = float(r["start"]), float(r["end"]) + except (KeyError, TypeError, ValueError): + continue + overlap = max(0.0, min(new_end, re_) - max(new_start, rs)) + if overlap / new_dur >= _REUSE_OVERLAP_RATIO: + # 复用同一条历史区间：累加次数、刷新时间 + r["use_count"] = int(r.get("use_count", 1)) + 1 + r["last_used_at"] = now + r["plan_id"] = plan_id + meta[USED_RANGES_KEY] = ranges + model.classification_result = json.dumps(meta, ensure_ascii=False) + model.updated_at = datetime.now(timezone.utc) + return + ranges.append( { - "start": round(float(start), 3), - "end": round(float(end), 3), + "start": new_start, + "end": new_end, "plan_id": plan_id, - "created_at": datetime.now(timezone.utc).isoformat(), + "created_at": now, + "use_count": 1, + "last_used_at": now, } ) meta[USED_RANGES_KEY] = ranges @@ -111,10 +178,10 @@ def remove_used_segment( Returns: 是否找到并删除了记录。 """ - model = db.query(AssetModel).filter(AssetModel.id == asset_id).first() + model = _get_model(db, asset_id) if model is None: return False - meta = _read_ranges(model) + meta = _read_meta(model) ranges = list(meta.get(USED_RANGES_KEY) or []) remaining: list[dict] = [] removed = False @@ -127,8 +194,7 @@ def remove_used_segment( remaining.append(r) continue # plan_id 校验：传入 plan_id 时，记录有 plan_id 则必须相等； - # 记录本身缺 plan_id（本功能上线前的旧数据）时退化为按时间匹配， - # 避免旧区间永远删不掉导致素材容量泄漏 + # 记录本身缺 plan_id（旧数据）时退化为按时间匹配，避免旧区间永远删不掉 if plan_id is not None and r.get("plan_id") is not None and r.get("plan_id") != plan_id: match = False if match and not removed: @@ -145,30 +211,219 @@ def remove_used_segment( def reset_used_segments(db: Session, asset_id: str) -> None: """清空单个素材的历史片段使用记录（不 commit）. - 单个素材的可用区间被全部占用（轮回一圈）后调用，使后续片段可重新使用整段素材。 + 仅供运维/测试使用；正常生成流程中历史记录永不自动清空（受控复用取代自动轮回）。 """ - model = db.query(AssetModel).filter(AssetModel.id == asset_id).first() + model = _get_model(db, asset_id) if model is None: return - meta = _read_ranges(model) + meta = _read_meta(model) if meta.get(USED_RANGES_KEY): meta[USED_RANGES_KEY] = [] model.classification_result = json.dumps(meta, ensure_ascii=False) model.updated_at = datetime.now(timezone.utc) - logger.info("[片段追踪] 素材区间轮回重置: asset_id=%s", asset_id) + logger.info("[片段追踪] 素材区间记录手动清空: asset_id=%s", asset_id) -def make_reset_callback(db: Session, used_segments: dict) -> Callable[[str], None]: - """构造给 _calc_random_start_time 用的 reset 回调. +# ── 素材余量/可用性计算（Task H：素材库角标 + smart-match 过滤）────────────── - 回调同时清空持久化 metadata 和内存中的 used_segments，使重试随机能覆盖全素材。 +# 判定「是否还有空闲可切区间」时使用的最小片段时长（秒）：空闲段长于此值才视为可切 +_MIN_FREE_CLIP_DURATION = 3.0 + + +def _merge_intervals(intervals: list[tuple[float, float]]) -> list[tuple[float, float]]: + """合并重叠/相接的时间区间，返回升序不重叠区间列表。""" + if not intervals: + return [] + ordered = sorted((float(a), float(b)) for a, b in intervals if b > a) + merged: list[tuple[float, float]] = [ordered[0]] + for start, end in ordered[1:]: + last_start, last_end = merged[-1] + if start <= last_end: + merged[-1] = (last_start, max(last_end, end)) + else: + merged.append((start, end)) + return merged + + +def _has_free_gap(used: list[tuple[float, float]], total: float, min_free: float = _MIN_FREE_CLIP_DURATION) -> bool: + """素材 [0, total] 中是否存在长度 ≥ min_free 的空闲段（考虑边缘间隙）。""" + if total <= 0: + return False + # 历史区间按边缘间隙扩边后判定空闲（与选片冲突检测同一口径） + expanded = [(max(0.0, s - SEGMENT_EDGE_GAP), min(total, e + SEGMENT_EDGE_GAP)) for s, e in used] + merged = _merge_intervals(expanded) + cursor = 0.0 + for start, end in merged: + if start - cursor >= min_free: + return True + cursor = max(cursor, end) + return total - cursor >= min_free + + +def compute_asset_availability( + model: "AssetModel | None", + min_free_clip_duration: float = _MIN_FREE_CLIP_DURATION, +) -> dict | None: + """计算单个素材的余量与可用性（纯函数，不读写 DB）。 + + Returns: + 视频素材返回 ``{"used_duration", "available_duration", "used_ratio", "usable"}``； + 非视频 / 无 model / 无时长信息返回 None（调用方按可用处理，零影响）。 + + usable=False 条件（与受控复用机制一致）： + 零重复可切区间已耗尽（不存在 ≥ min_free 的空闲段）且 + 所有历史区间 use_count 均达 MAX_RANGE_USE_COUNT 上限（无区间可复用）。 + """ + if model is None: + return None + file_type = getattr(model, "file_type", None) or getattr(model, "mime_type", "") or "" + if file_type != "video" and not str(file_type).startswith("video/"): + return None + total = float(getattr(model, "duration", 0.0) or 0.0) + if total <= 0: + return None + + meta = _read_meta(model) + raw_ranges = meta.get(USED_RANGES_KEY) or [] + + intervals: list[tuple[float, float]] = [] + use_counts: list[int] = [] + for r in raw_ranges: + try: + start = float(r["start"]) + end = float(r["end"]) + except (KeyError, TypeError, ValueError): + continue + if end <= start: + continue + intervals.append((start, end)) + try: + use_counts.append(int(r.get("use_count", 1))) + except (TypeError, ValueError): + use_counts.append(1) + + merged = _merge_intervals(intervals) + used_duration = round(sum(e - s for s, e in merged), 3) + used_duration = min(used_duration, total) + available_duration = round(max(total - used_duration, 0.0), 3) + used_ratio = round(min(used_duration / total, 1.0), 4) + + has_free = _has_free_gap(intervals, total, min_free_clip_duration) + if has_free: + usable = True + else: + # 空闲段耗尽：仅当存在历史区间且全部达复用上限时才判定不可用； + # 无历史区间（理论上不会走到，因为 has_free=True）按可用处理 + if not use_counts: + usable = True + else: + usable = any(uc < MAX_RANGE_USE_COUNT for uc in use_counts) + + return { + "used_duration": used_duration, + "available_duration": available_duration, + "used_ratio": used_ratio, + "usable": usable, + } + + +def find_reusable_range( + db: Session, + asset_id: str, + clip_duration: float, + asset_total: float, + *, + max_use_count: int = MAX_RANGE_USE_COUNT, +) -> tuple[float, float] | None: + """受控复用：在素材历史区间中选一条可复用区间返回 (start, end)。 + + 选择规则： + 1. 仅选 ``use_count < max_use_count`` 的历史区间； + 2. 优先返回能完整容纳当前 clip_duration（起点后不越素材边界）的最久未用区间； + 3. 没有能容纳的，则返回 last_used_at 最老（或缺失 last_used_at 的旧数据优先） + 且 use_count 最低的区间起点（可能与其他历史区间重叠，属降级复用）； + 4. 无任何可复用区间（记录为空或全部达上限）返回 None。 + + 本函数只读不写；复用次数的累加由后续 record_used_segments 完成。 + """ + model = _get_model(db, asset_id) + if model is None: + return None + meta = _read_meta(model) + ranges = [r for r in (meta.get(USED_RANGES_KEY) or []) if int(r.get("use_count", 1)) < max_use_count] + if not ranges: + return None + + def _last_used(r: dict) -> str: + return str(r.get("last_used_at") or r.get("created_at") or "") + + max_start = max(0.0, asset_total - clip_duration) + # 2. 能完整容纳当前片段的候选：按 last_used_at 升序（最久未用优先） + fit = sorted( + [r for r in ranges if float(r["start"]) <= max_start + 1e-6], + key=_last_used, + ) + if fit: + start = min(float(fit[0]["start"]), max_start) + return (start, start + clip_duration) + + # 3. 降级：最久未用 + use_count 最低的区间起点 + fallback = sorted(ranges, key=lambda r: (_last_used(r), int(r.get("use_count", 1))))[0] + start = min(float(fallback["start"]), max_start) + return (start, start + clip_duration) + + +def make_reuse_callback( + db: Session, + asset_durations: dict[str, float], + reused_tracker: dict[str, float] | None = None, + assigned_tracker: dict[str, float] | None = None, + ratio_limit: float = REUSE_RATIO_LIMIT, +) -> Callable[[str, float], tuple[float, float] | None]: + """构造给 ``_calc_random_start_time`` 用的受控复用回调. + + Args: + db: SQLAlchemy session + asset_durations: 素材 ID -> 总时长（回调需要素材总时长做边界约束） + reused_tracker: 可选的 ``{asset_id: 累计复用时长}``，回调成功返回复用区间时 + 会把本次片段时长累加进去，供调用方统计成片复用占比（15% 阈值）。 + assigned_tracker: 可选的 ``{asset_id: 已分配片段总时长}``，配合 ratio_limit + 在复用前预判：若复用本片段后占比 (reused + clip_duration) / + (assigned + clip_duration) 超过 ratio_limit，则拒绝复用、返回 None + （保证成片复用占比不超阈值）。 + ratio_limit: 单条成片复用时长占比上限，默认 15%。 + + Returns: + 回调函数 ``(asset_id, clip_duration) -> (start, end) | None``。 + 回调内吞掉 DB 异常返回 None，不影响主生成流程。 """ - def _reset(asset_id: str) -> None: + def _reuse(asset_id: str, clip_duration: float) -> tuple[float, float] | None: try: - reset_used_segments(db, asset_id) + total = float(asset_durations.get(asset_id, 0.0) or 0.0) + if total <= 0: + return None + # 占比闸门：预判复用本片段后是否超限（仅当调用方提供了 assigned tracker） + if assigned_tracker is not None: + assigned = float(assigned_tracker.get(asset_id, 0.0) or 0.0) + reused_amt = float((reused_tracker or {}).get(asset_id, 0.0) or 0.0) + if assigned > 0 and (reused_amt + clip_duration) / (assigned + clip_duration) > ratio_limit: + logger.info( + "[片段追踪] 复用占比预判超 %.0f%% 阈值，拒绝复用: asset_id=%s " + "reused=%.1f assigned=%.1f clip=%.1f", + ratio_limit * 100, + asset_id, + reused_amt, + assigned, + clip_duration, + ) + return None + result = find_reusable_range(db, asset_id, clip_duration, total) except Exception: - logger.warning("[片段追踪] reset 持久化记录失败: asset_id=%s", asset_id, exc_info=True) - used_segments.pop(asset_id, None) + logger.warning("[片段追踪] 受控复用查询异常: asset_id=%s", asset_id, exc_info=True) + return None + if result is not None and reused_tracker is not None: + reused_tracker[asset_id] = reused_tracker.get(asset_id, 0.0) + clip_duration + return result - return _reset + return _reuse diff --git a/apps/api/app/services/edit_plan_service.py b/apps/api/app/services/edit_plan_service.py index a87060234..5ed03ae07 100755 --- a/apps/api/app/services/edit_plan_service.py +++ b/apps/api/app/services/edit_plan_service.py @@ -9,6 +9,12 @@ from __future__ import annotations import logging from typing import Any, Dict, List, Optional +from app.services.asset_segment_tracker import ( + REUSE_RATIO_LIMIT, + get_used_segments, + make_reuse_callback, + record_used_segments, +) from sqlalchemy.orm import Session from packages.adapters.sqlalchemy_impl import ( @@ -453,6 +459,116 @@ class EditPlanService: logger.exception("事务性替换片段失败: plan_id=%s", plan_id) raise + def clone_plan_for_variant( + self, + source_plan_id: str, + *, + created_by_user_id: str = "", + name_suffix: str = "变体", + reuse_tracker: Optional[dict] = None, + ) -> EditPlan: + """为批量/多预览场景克隆一份独立 plan，片段起点全部重算（受控随机/复用）。 + + 复制源 plan 的模板归属、config 与片段结构（asset_id / duration / clip_type / + order 不变），每个片段重新调用 ``_calc_random_start_time``：读取素材持久化的 + 历史已用区间避让，耗尽时受控复用（use_count<3、最久未用），从而保证 N 条 + 成片片段区间互不相同，且复用占比受控。 + + - 不替换/不修改源 plan，源 plan 保留用户手动编辑结果。 + - 片段区间记录（record_used_segments）随新片段写入素材 metadata，与新 plan + 同事务；复用历史区间时由 record 自动累加 use_count。 + - 克隆的 clips 复用区间累计时长写入 reuse_tracker（可选），供调用方统计占比。 + + Raises: + ValueError: 源 plan 不存在或无可用片段。 + """ + from packages.adapters.sqlalchemy_impl.models import AssetModel + from packages.domain.plan_generator_utils import _calc_random_start_time + + 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}") + + # 创建新 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 ""), + ) + + # 素材时长映射（O(N) 单查） + asset_ids = list({c.asset_id for c in clips if c.asset_id}) + db = self._clip_repo.session + durations: dict[str, float] = {} + if asset_ids: + for m in db.query(AssetModel).filter(AssetModel.id.in_(asset_ids)).all(): + durations[m.id] = float(getattr(m, "duration", 0.0) or 0.0) + + used_segments = get_used_segments(db, asset_ids) + reused: dict[str, float] = reuse_tracker if reuse_tracker is not None else {} + asset_assigned: dict[str, float] = {} + # 回调内部预判复用后占比超 15% 则拒绝复用（calc 返回 None → 保留原起点） + reuse_cb = make_reuse_callback(db, durations, reused, assigned_tracker=asset_assigned) + + clips_data: list[dict] = [] + for i, c in enumerate(clips): + aid = c.asset_id + dur = float(c.duration or 0.0) + total = durations.get(aid, 0.0) + if aid and total > 0 and dur > 0: + # 复用占比闸门：本片段尚未分配，检查当前已复用占比 + # reused / assigned 是否超 15%，超则该素材不再分配（保留原起点）； + # assigned=0（首个片段）放行，reused=0 时不误拦正常分配 + assigned = asset_assigned.get(aid, 0.0) + eff_dur = min(dur, total) + reused_amt = reused.get(aid, 0.0) + ratio_blocked = assigned > 0 and reused_amt / assigned > REUSE_RATIO_LIMIT + start = None + if not ratio_blocked: + start = _calc_random_start_time(aid, eff_dur, durations, used_segments, on_exhausted=reuse_cb) + if start is None: + start = float(c.start_time or 0.0) + asset_assigned[aid] = assigned + eff_dur + used_segments.setdefault(aid, []).append((start, start + eff_dur)) + record_used_segments(db, aid, start, start + eff_dur, new_plan.id) + else: + start = float(c.start_time or 0.0) + + clips_data.append( + { + "order": c.order if c.order is not None else i, + "asset_id": aid, + "start_time": start, + "duration": dur, + "clip_type": c.clip_type, + } + ) + + # 事务性写入新 plan 的片段（内部统一 commit/rollback） + self.replace_all_clips_transactional(new_plan.id, clips_data) + logger.info( + "克隆变体 plan: source=%s new=%s clips=%d", + source_plan_id, + new_plan.id, + len(clips_data), + ) + return new_plan + # ── 片段分割与合并 ────────────────────────────────────────────────────── def split_clip(self, clip_id: str, split_time: float) -> Dict[str, Any]: diff --git a/packages/domain/plan_generator_utils.py b/packages/domain/plan_generator_utils.py index 96babf114..3e57596ac 100755 --- a/packages/domain/plan_generator_utils.py +++ b/packages/domain/plan_generator_utils.py @@ -207,7 +207,7 @@ def _calc_random_start_time( clip_duration: float, asset_durations: dict[str, float] | None, used_segments: dict[str, list[tuple[float, float]]] | None = None, - on_exhausted: Callable[[str], None] | None = None, + on_exhausted: Callable[[str, float], tuple[float, float] | None] | None = None, ) -> float | None: """计算随机 start_time，避开已使用的时间段. @@ -220,8 +220,10 @@ def _calc_random_start_time( clip_duration: 片段时长（秒） asset_durations: 素材 ID -> 时长映射 used_segments: {asset_id: [(start1, end1), (start2, end2), ...]} 已使用的时间段 - on_exhausted: 100 次随机都找不到空闲区间时的回调（入参 asset_id）。 - 通常用于清空该素材的历史使用记录实现“轮回重置”；回调后会再随机重试一次。 + on_exhausted: 100 次随机都找不到空闲区间时的受控复用回调，入参为 + (asset_id, clip_duration)，返回 (start, end) 复用区间或 None。 + 历史记录永不自动清空；回调返回 None（全部达上限/复用占比超闸门）时 + 本函数返回 None，由调用方轮询下一个素材或报错，不做重叠降级。 Returns: 随机 start_time 或 None @@ -261,22 +263,36 @@ def _calc_random_start_time( if not overlap: return candidate - # 100 次都找不到空闲区间：触发轮回重置回调（清空历史使用记录）后再随机重试一次 + # 100 次都找不到空闲区间：进入受控复用，回调从历史区间中选最久未用且 + # 使用次数未达上限的区间返回（历史记录永不自动清空） if on_exhausted is not None: try: - on_exhausted(asset_id) + reused = on_exhausted(asset_id, clip_duration) except Exception: logger.warning( - "on_exhausted 轮回重置回调异常: asset_id=%s", + "on_exhausted 受控复用回调异常: asset_id=%s", asset_id, exc_info=True, ) - retry = random.uniform(0.0, max_start) - if not used_segments or asset_id not in used_segments: - return retry + reused = None + if reused is not None: + reuse_start, reuse_end = reused + # 边界保护：不越素材末尾、不为负 + reuse_start = max(0.0, min(float(reuse_start), max_start)) + logger.info( + "素材可用区间耗尽，受控复用历史区间: asset_id=%s start=%.2f end=%.2f", + asset_id, + reuse_start, + reuse_end, + ) + return reuse_start + # 回调存在但拒绝复用（区间全部达 use_count 上限，或复用占比将超 15% 闸门）： + # 返回 None，由调用方轮询下一个素材；绝不能末尾/0.0 降级——那会把片段 + # 放回到已用过的画面，违反区间避让与重复率控制原则 + return None - # 如果尝试多次仍找不到，缩短时长使用素材末尾 - # 找到最后一个已使用段之后的可用空间 + # 未提供 on_exhausted 回调（向后兼容）：降级使用素材末尾空闲位置； + # 末尾也已占满时返回 0.0（旧行为，仅无持久化追踪的调用方会走到这里） last_used_end = 0.0 for _seg_start, seg_end in used: last_used_end = max(last_used_end, seg_end) @@ -285,7 +301,6 @@ def _calc_random_start_time( # 返回从最后使用点开始的位置 return min(last_used_end, max_start) - # 实在没有空间，返回0（可能会重叠，但至少能执行） return 0.0 diff --git a/tests/unit/test_asset_availability.py b/tests/unit/test_asset_availability.py new file mode 100755 index 000000000..40e62f53c --- /dev/null +++ b/tests/unit/test_asset_availability.py @@ -0,0 +1,376 @@ +"""Task H 单测：素材余量四字段（used_duration/available_duration/used_ratio/usable）。 + +覆盖： +1. compute_asset_availability 纯函数各分支（无区间/未满/可复用/全达上限/非视频/无时长/区间合并/扩边判定）； +2. _asset_availability_fields 路由辅助（视频有值、非视频 None+usable=True、异常零影响）； +3. _to_asset_response 四字段注入； +4. smart_match_assets 结果层过滤 usable=false。 +""" + +import json +import os +import sys +from pathlib import Path +from types import SimpleNamespace +from unittest.mock import MagicMock + +import pytest + +REPO_ROOT = Path(__file__).resolve().parents[2] +if str(REPO_ROOT) not in sys.path: + sys.path.insert(0, str(REPO_ROOT)) + +from app.api.routes.assets import ( # noqa: E402 + _asset_availability_fields, + _to_asset_response, + smart_match_assets, +) +from app.schemas.asset import SmartMatchRequest # noqa: E402 +from app.services.asset_segment_tracker import ( # noqa: E402 + MAX_RANGE_USE_COUNT, + SEGMENT_EDGE_GAP, + compute_asset_availability, +) + +VIDEO_DURATION = 60.0 + + +def _make_asset(duration=VIDEO_DURATION, ranges=None, file_type="video", classification_result=None): + """构造测试用 Asset-like 对象。 + + ranges: list of dicts（used_time_ranges 条目），会自动写入 classification_result JSON。 + """ + if classification_result is None and ranges is not None: + classification_result = json.dumps({"used_time_ranges": ranges}) + return SimpleNamespace( + id="asset-test", + project_id="proj-1", + library_id="lib-1", + name="测试素材", + storage_key="key/test-asset.mp4", + thumbnail_url=None, + mime_type="video/mp4" if file_type == "video" else "audio/mpeg", + metadata={}, + file_size=1000, + duration=duration, + width=1080, + height=1920, + fps=30, + codec="h264", + status=SimpleNamespace(value="ready"), + classification_status=SimpleNamespace(value="completed"), + quality_score=90.0, + created_at=__import__("datetime").datetime(2026, 8, 1, 12, 0, 0), + uploaded_by_user_id="user-1", + tag_ids=[], + file_type=file_type, + classification_result=classification_result, + ) + + +def _range(start, end, use_count=1): + return { + "start": start, + "end": end, + "plan_id": "plan-1", + "created_at": "2026-08-29T10:00:00", + "use_count": use_count, + "last_used_at": "2026-08-29T10:00:00", + } + + +# ── compute_asset_availability 纯函数 ───────────────────────────────────────── + + +class TestComputeAssetAvailability: + def test_no_ranges_fully_usable(self): + """无历史区间：used=0, ratio=0, usable=True。""" + info = compute_asset_availability(_make_asset(ranges=[])) + assert info is not None + assert info["used_duration"] == 0.0 + assert info["available_duration"] == VIDEO_DURATION + assert info["used_ratio"] == 0.0 + assert info["usable"] is True + + def test_none_model_returns_none(self): + assert compute_asset_availability(None) is None + + def test_non_video_returns_none(self): + """非视频（音频）返回 None，路由层按可用处理。""" + info = compute_asset_availability(_make_asset(file_type="audio")) + assert info is None + + def test_zero_duration_returns_none(self): + info = compute_asset_availability(_make_asset(duration=0.0)) + assert info is None + + def test_partial_usage_usable(self): + """使用 10s，剩余 50s 空闲（≥3s），usable=True。""" + info = compute_asset_availability(_make_asset(ranges=[_range(5.0, 15.0)])) + assert info["used_duration"] == pytest.approx(10.0, abs=0.01) + assert info["available_duration"] == pytest.approx(50.0, abs=0.01) + assert info["used_ratio"] == pytest.approx(10.0 / 60.0, abs=0.001) + assert info["usable"] is True + + def test_overlapping_ranges_merged(self): + """重叠区间合并后计算 used_duration，不重复计时。""" + info = compute_asset_availability(_make_asset(ranges=[_range(0.0, 10.0), _range(5.0, 20.0)])) + # 合并后 [0,20] → 20s + assert info["used_duration"] == pytest.approx(20.0, abs=0.01) + assert info["used_ratio"] == pytest.approx(20.0 / 60.0, abs=0.001) + + def test_full_coverage_but_reusable(self): + """区间铺满全片（无空闲段），但 use_count 未达上限 → usable=True（受控复用）。""" + info = compute_asset_availability( + _make_asset( + duration=10.0, + ranges=[_range(0.0, 10.0, use_count=1)], + ) + ) + assert info["used_duration"] == pytest.approx(10.0, abs=0.01) + assert info["available_duration"] == 0.0 + assert info["usable"] is True + + def test_exhausted_not_usable(self): + """无空闲段 且 所有区间 use_count 达上限 → usable=False。""" + info = compute_asset_availability( + _make_asset( + duration=10.0, + ranges=[_range(0.0, 10.0, use_count=MAX_RANGE_USE_COUNT)], + ) + ) + assert info["usable"] is False + assert info["available_duration"] == 0.0 + assert info["used_ratio"] == pytest.approx(1.0, abs=0.001) + + def test_exhausted_multiple_ranges_all_capped(self): + """多个区间铺满、全部达上限 → usable=False；任一未满即 usable=True。""" + info_capped = compute_asset_availability( + _make_asset( + duration=20.0, + ranges=[ + _range(0.0, 10.0, use_count=MAX_RANGE_USE_COUNT), + _range(10.0, 20.0, use_count=MAX_RANGE_USE_COUNT), + ], + ) + ) + assert info_capped["usable"] is False + + info_partial = compute_asset_availability( + _make_asset( + duration=20.0, + ranges=[ + _range(0.0, 10.0, use_count=MAX_RANGE_USE_COUNT), + _range(10.0, 20.0, use_count=MAX_RANGE_USE_COUNT - 1), + ], + ) + ) + assert info_partial["usable"] is True + + def test_edge_gap_consumed_not_usable(self): + """区间未物理铺满，但扩边（+0.3s）后空闲段 <3s → 视为无空闲段； + 区间 use_count 均达上限 → usable=False。""" + # 10s 素材：[0, 4.0] 与 [4.6, 10]，物理空闲 [4.0,4.6] 仅 0.6s， + # 扩边后左区间延至 4.3、右区间起于 4.3，空闲被吃掉 + info = compute_asset_availability( + _make_asset( + duration=10.0, + ranges=[ + _range(0.0, 4.0, use_count=MAX_RANGE_USE_COUNT), + _range(4.6, 10.0, use_count=MAX_RANGE_USE_COUNT), + ], + ) + ) + assert info["usable"] is False + + def test_large_gap_remains_usable(self): + """区间之间留有 ≥3s 空闲段（扩边后仍 ≥3s）→ usable=True。""" + # [0,2] 扩边到 [0,2.3]，[5.3,10] 扩边前为 [5,10] 扩边起 4.7；空闲 [2.3,4.7]=2.4s <3 + # 改用更大间隙：[0,2] 与 [6,10]，扩边后空闲 [2.3,5.7]=3.4s ≥3 + info = compute_asset_availability( + _make_asset( + duration=10.0, + ranges=[ + _range(0.0, 2.0, use_count=MAX_RANGE_USE_COUNT), + _range(6.0, 10.0, use_count=MAX_RANGE_USE_COUNT), + ], + ) + ) + assert info["usable"] is True + + def test_invalid_ranges_skipped(self): + """脏数据（缺 start/end、end<=start、use_count 非法）不崩溃，合法区间照常计算。""" + info = compute_asset_availability( + _make_asset( + duration=30.0, + ranges=[ + {"start": "bad"}, + {"start": 5.0, "end": 3.0}, + "junk", + _range(0.0, 10.0, use_count="not-a-number"), + ], + ) + ) + assert info is not None + assert info["used_duration"] == pytest.approx(10.0, abs=0.01) + # use_count 非法按 1 处理 → 未达上限，且空闲段充足 + assert info["usable"] is True + + def test_broken_classification_json_treated_as_unused(self): + """classification_result 是非法 JSON 时按无历史区间处理。""" + info = compute_asset_availability(_make_asset(classification_result="not-json{{{")) + assert info is not None + assert info["used_duration"] == 0.0 + assert info["usable"] is True + + def test_segment_edge_gap_constant(self): + """边缘间隙常量为 0.3s（与 MediaKit 冲突检测同口径）。""" + assert SEGMENT_EDGE_GAP == 0.3 + + +# ── 路由层辅助：_asset_availability_fields / _to_asset_response ────────────── + + +class TestAssetAvailabilityFields: + def test_video_asset_returns_values(self): + fields = _asset_availability_fields(_make_asset(ranges=[_range(0.0, 10.0)])) + assert fields["usable"] is True + assert fields["used_duration"] == pytest.approx(10.0, abs=0.01) + assert fields["available_duration"] == pytest.approx(50.0, abs=0.01) + assert fields["used_ratio"] is not None + + def test_non_video_returns_none_fields_usable_true(self): + fields = _asset_availability_fields(_make_asset(file_type="audio")) + assert fields["used_duration"] is None + assert fields["available_duration"] is None + assert fields["used_ratio"] is None + assert fields["usable"] is True + + def test_exception_falls_back_to_zero_impact(self, monkeypatch): + """compute 抛异常时路由层兜底：None 字段 + usable=True，不影响响应。""" + import app.api.routes.assets as assets_module + + def _boom(_model): + raise RuntimeError("unexpected") + + monkeypatch.setattr(assets_module, "compute_asset_availability", _boom) + fields = _asset_availability_fields(_make_asset()) + assert fields["used_duration"] is None + assert fields["usable"] is True + + +class TestToAssetResponseInjectsFields: + def _storage_stub(self): + svc = MagicMock() + svc.get_download_url.return_value = "https://example.com/signed" + return svc + + def test_video_response_carries_availability_fields(self): + asset = _make_asset(ranges=[_range(0.0, 12.0)]) + resp = _to_asset_response(asset, storage_service=self._storage_stub()) + assert resp.usable is True + assert resp.used_duration == pytest.approx(12.0, abs=0.01) + assert resp.available_duration == pytest.approx(48.0, abs=0.01) + assert resp.used_ratio == pytest.approx(0.2, abs=0.01) + + def test_exhausted_asset_response_usable_false(self): + asset = _make_asset( + duration=10.0, + ranges=[_range(0.0, 10.0, use_count=MAX_RANGE_USE_COUNT)], + ) + resp = _to_asset_response(asset, storage_service=self._storage_stub()) + assert resp.usable is False + assert resp.used_ratio == pytest.approx(1.0, abs=0.001) + + def test_non_video_response_fields_none_usable_true(self): + asset = _make_asset(file_type="audio") + resp = _to_asset_response(asset, storage_service=self._storage_stub()) + assert resp.used_duration is None + assert resp.available_duration is None + assert resp.used_ratio is None + assert resp.usable is True + + +# ── smart_match_assets 结果层过滤 ──────────────────────────────────────────── + + +def _exhausted_asset(asset_id): + """构造一个 usable=false 的视频素材：10s 铺满、区间 use_count 均达上限。""" + a = _make_asset( + duration=10.0, + ranges=[_range(0.0, 10.0, use_count=MAX_RANGE_USE_COUNT)], + ) + a.id = asset_id + a.name = f"exhausted-{asset_id}" + return a + + +def _fresh_asset(asset_id, duration=60.0): + a = _make_asset(duration=duration, ranges=[]) + a.id = asset_id + a.name = f"fresh-{asset_id}" + return a + + +class TestSmartMatchFiltersExhausted: + def _call(self, assets): + lib_repo = MagicMock() + lib_repo.get.return_value = SimpleNamespace(project_id="proj-1") + asset_repo = MagicMock() + asset_repo.find_by_library_and_file_type.return_value = assets + project_repo = MagicMock() + project = MagicMock() + project.can_access.return_value = True + project_repo.find_by_id.return_value = project + + user = SimpleNamespace(id="user-1") + auth_user = SimpleNamespace(user=user) + + # storage_service 在 _to_asset_response 内 get_storage_service()，patch 掉 + import app.api.routes.assets as assets_module + + svc = MagicMock() + svc.get_download_url.return_value = "https://example.com/signed" + original_get_storage = assets_module.get_storage_service + assets_module.get_storage_service = lambda: svc + try: + resp = smart_match_assets( + SmartMatchRequest(library_id="lib-1", kind="video"), + authenticated_user=auth_user, + asset_repository=asset_repo, + asset_library_repository=lib_repo, + project_repository=project_repo, + ) + finally: + assets_module.get_storage_service = original_get_storage + return resp + + def test_exhausted_assets_excluded(self): + """smart-match 结果中 usable=false 的素材被剔除，新鲜素材保留。""" + assets = [ + _exhausted_asset("a-exhausted-1"), + _exhausted_asset("a-exhausted-2"), + _fresh_asset("a-fresh-1"), + ] + resp = self._call(assets) + returned_ids = {item.asset.id for item in resp.items} + assert "a-fresh-1" in returned_ids + assert "a-exhausted-1" not in returned_ids + assert "a-exhausted-2" not in returned_ids + # total_candidates 是过滤前的候选总数 + assert resp.total_candidates == 3 + # 返回的素材全部 usable=True + assert all(item.asset.usable for item in resp.items) + + def test_all_exhausted_returns_empty(self): + """全部素材已用尽时返回空列表（不报错，前端显示空结果）。""" + assets = [_exhausted_asset("a-ex-1"), _exhausted_asset("a-ex-2")] + resp = self._call(assets) + assert resp.items == [] + assert resp.total_candidates == 2 + + def test_fresh_assets_all_returned(self): + assets = [_fresh_asset("a-1"), _fresh_asset("a-2")] + resp = self._call(assets) + assert len(resp.items) == 2 + assert all(item.asset.usable for item in resp.items) diff --git a/tests/unit/test_asset_segment_tracker.py b/tests/unit/test_asset_segment_tracker.py index 754bcfb6d..9174226a0 100644 --- a/tests/unit/test_asset_segment_tracker.py +++ b/tests/unit/test_asset_segment_tracker.py @@ -1,12 +1,14 @@ -"""素材片段使用记录追踪服务测试（asset_segment_tracker）. +"""素材片段使用记录追踪 + 受控复用机制测试（asset_segment_tracker）. 覆盖： - get_used_segments 聚合 metadata 中持久化的区间 -- record_used_segments 追加记录（不 commit，保留原有 metadata 字段） -- remove_used_segment 匹配删除（tolerance + plan_id） -- reset_used_segments 清空轮回（其他字段不动） -- make_reset_callback 同时清持久化和内存 -- _calc_random_start_time 的 on_exhausted 轮回回调 +- record_used_segments 追加新记录（use_count=1，保留原有 metadata 字段） +- record_used_segments 复用同一区间时累加 use_count / 刷新 last_used_at +- remove_used_segment 匹配删除（tolerance + plan_id，旧数据按时间匹配） +- reset_used_segments 清空（其他字段不动） +- find_reusable_range：选最久未用且 use_count<3 的区间；全部达上限返回 None +- make_reuse_callback：返回复用区间、累加 reused_tracker、DB 异常返回 None +- _calc_random_start_time：100 次避不开时调用复用回调返回历史区间（不再清空历史） """ from __future__ import annotations @@ -24,8 +26,12 @@ sys.path.insert(0, str(Path(__file__).resolve().parents[2] / "apps" / "api")) import pytest from app.services import asset_segment_tracker as ast from app.services.asset_segment_tracker import ( + MAX_RANGE_USE_COUNT, + REUSE_RATIO_LIMIT, + SEGMENT_EDGE_GAP, + find_reusable_range, get_used_segments, - make_reset_callback, + make_reuse_callback, record_used_segments, remove_used_segment, reset_used_segments, @@ -60,16 +66,16 @@ class _EqExpr: self._target_id = target_id self._models = models + def with_for_update(self): + # 模拟 SQLAlchemy Query.with_for_update() 链式返回自身 + return self + def first(self): return self._models.get(self._target_id) class FakeSession: - """模拟 db：db.query(Model).filter(Model.id.in_(ids)).all() / .filter(Model.id == id).first()。 - - tracker 模块里的 AssetModel 被 monkeypatch 为 FakeModel 类， - 这里用挂在类上的伪 column 对象接住 in_ / __eq__。 - """ + """模拟 db：db.query(Model).filter(Model.id.in_(ids)).all() / .filter(Model.id == id).first()。""" class _Col: def __init__(self, models): @@ -93,7 +99,6 @@ class FakeSession: return expr q = _Q() - # 让 tracker 里 AssetModel.id 能取到伪 column _model.id = col return q @@ -103,19 +108,27 @@ class FakeSession: @pytest.fixture def patched_model(monkeypatch): - """把 tracker 模块内的 AssetModel 替换为 FakeModel（供 FakeSession 挂伪 column）。""" monkeypatch.setattr(ast, "AssetModel", FakeModel) -@pytest.fixture -def models(): - return {} - - def _db(models): return FakeSession(models) +def _ranges(db, aid="a1"): + model = db._models[aid] + return json.loads(model.classification_result)["used_time_ranges"] + + +# ── 配置常量 ────────────────────────────────────────────────────────────────── + + +def test_config_constants(): + assert MAX_RANGE_USE_COUNT == 3 + assert REUSE_RATIO_LIMIT == 0.15 + assert SEGMENT_EDGE_GAP == 0.3 + + # ── get_used_segments ───────────────────────────────────────────────────────── @@ -125,124 +138,138 @@ def test_get_used_segments_aggregates_ranges(patched_model): "a1", { "used_time_ranges": [ - {"start": 1.0, "end": 5.0, "plan_id": "p1"}, + {"start": 1.0, "end": 5.0, "plan_id": "p1", "use_count": 2}, {"start": 9.0, "end": 12.0, "plan_id": "p2"}, ] }, ), - "a2": FakeModel("a2", {"other": 1}), # 无区间记录 - "a3": FakeModel("a3"), # metadata 为空 + "a2": FakeModel("a2", {"other": 1}), + "a3": FakeModel("a3"), } db = _db(models) - - result = get_used_segments(db, ["a1", "a2", "a3", "missing"]) - - assert result == {"a1": [(1.0, 5.0), (9.0, 12.0)]} + assert get_used_segments(db, ["a1", "a2", "a3", "missing"]) == {"a1": [(1.0, 5.0), (9.0, 12.0)]} -def test_get_used_segments_empty_input(patched_model): +def test_get_used_segments_empty(patched_model): assert get_used_segments(_db({}), []) == {} # ── record_used_segments ────────────────────────────────────────────────────── -def test_record_appends_and_no_commit(patched_model): - models = {"a1": FakeModel("a1", {"generation_use_count": 3})} +def test_record_appends_new_range_with_use_count_one(patched_model): + models = {"a1": FakeModel("a1", {"generation_use_count": 48, "review_status": "pending_review"})} db = _db(models) - - record_used_segments(db, "a1", 2.0, 6.5, "plan-x") - - meta = models["a1"].meta() - assert meta["generation_use_count"] == 3 # 原有字段保留 + record_used_segments(db, "a1", 12.5, 20.3, "plan-x") + meta = json.loads(models["a1"].classification_result) + assert meta["generation_use_count"] == 48 + assert meta["review_status"] == "pending_review" ranges = meta["used_time_ranges"] assert len(ranges) == 1 - assert ranges[0]["start"] == 2.0 - assert ranges[0]["end"] == 6.5 + assert ranges[0]["start"] == 12.5 and ranges[0]["end"] == 20.3 assert ranges[0]["plan_id"] == "plan-x" - assert "created_at" in ranges[0] - assert db.commits == 0 # 不自行 commit（事务由调用方控制） + assert ranges[0]["use_count"] == 1 + assert "created_at" in ranges[0] and "last_used_at" in ranges[0] + assert db.commits == 0 # 不自行 commit -def test_record_multiple_appends_in_order(patched_model): - models = {"a1": FakeModel("a1")} +def test_record_reuse_same_range_increments_use_count(patched_model): + """新片段与历史区间高度重叠（复用）→ 累加 use_count，不新增记录。""" + models = { + "a1": FakeModel( + "a1", + { + "used_time_ranges": [ + { + "start": 10.0, + "end": 20.0, + "plan_id": "p1", + "use_count": 1, + "created_at": "2026-01-01T00:00:00+00:00", + "last_used_at": "2026-01-01T00:00:00+00:00", + }, + ] + }, + ) + } db = _db(models) - - record_used_segments(db, "a1", 0.0, 4.0, "p1") - record_used_segments(db, "a1", 10.0, 14.0, "p1") - - ranges = models["a1"].meta()["used_time_ranges"] - assert [r["start"] for r in ranges] == [0.0, 10.0] + # 同一起点复用（find_reusable_range 返回的就是历史区间起点） + record_used_segments(db, "a1", 10.0, 20.0, "p2") + ranges = _ranges(db) + assert len(ranges) == 1 + assert ranges[0]["use_count"] == 2 + assert ranges[0]["last_used_at"] != "2026-01-01T00:00:00+00:00" -def test_record_missing_asset_is_noop(patched_model): +def test_record_distinct_range_appends(patched_model): + models = { + "a1": FakeModel( + "a1", + { + "used_time_ranges": [ + {"start": 10.0, "end": 20.0, "plan_id": "p1", "use_count": 1}, + ] + }, + ) + } + db = _db(models) + record_used_segments(db, "a1", 25.0, 35.0, "p2") + ranges = _ranges(db) + assert len(ranges) == 2 + assert ranges[1]["use_count"] == 1 + + +def test_record_missing_asset_no_raise(patched_model): db = _db({}) - record_used_segments(db, "ghost", 0.0, 1.0, "p1") # 不抛异常 + record_used_segments(db, "ghost", 1.0, 2.0, "p") # 不抛异常 # ── remove_used_segment ─────────────────────────────────────────────────────── -def test_remove_matching_segment(patched_model): - models = {"a1": FakeModel("a1")} - db = _db(models) - record_used_segments(db, "a1", 0.0, 4.0, "p1") - record_used_segments(db, "a1", 10.0, 14.0, "p1") - - removed = remove_used_segment(db, "a1", 0.0, 4.0, plan_id="p1") - - assert removed is True - ranges = models["a1"].meta()["used_time_ranges"] - assert len(ranges) == 1 - assert ranges[0]["start"] == 10.0 - - -def test_remove_not_found_returns_false(patched_model): - models = {"a1": FakeModel("a1")} - db = _db(models) - record_used_segments(db, "a1", 0.0, 4.0, "p1") - - assert remove_used_segment(db, "a1", 99.0, 100.0, plan_id="p1") is False - - -def test_remove_respects_tolerance(patched_model): - models = { - "a1": FakeModel("a1", {"used_time_ranges": [{"start": 5.0, "end": 9.0, "plan_id": "p1"}]}), - "a2": FakeModel("a2", {"used_time_ranges": [{"start": 5.0, "end": 9.0, "plan_id": "p1"}]}), - } - db = _db(models) - - # 偏差 0.3 秒，在 tolerance=0.5 内 → 删除成功 - assert remove_used_segment(db, "a1", 5.3, 8.7, plan_id="p1") is True - # 偏差 2 秒，超出 tolerance → 删除失败 - assert remove_used_segment(db, "a2", 7.0, 11.0, plan_id="p1") is False - - -def test_remove_plan_id_must_match(patched_model): - models = {"a1": FakeModel("a1", {"used_time_ranges": [{"start": 5.0, "end": 9.0, "plan_id": "plan-A"}]})} - db = _db(models) - - # 时间匹配但 plan_id 不同 → 不删除 - assert remove_used_segment(db, "a1", 5.0, 9.0, plan_id="plan-B") is False - assert len(models["a1"].meta()["used_time_ranges"]) == 1 - - -def test_remove_legacy_record_without_plan_id(patched_model): - """旧数据记录没有 plan_id 字段时，MediaKit 移动片段仍能按时间匹配删除（防容量泄漏）。""" +def test_remove_matching_range(patched_model): models = { "a1": FakeModel( "a1", - {"used_time_ranges": [{"start": 5.0, "end": 9.0}]}, # 旧记录无 plan_id + { + "used_time_ranges": [ + {"start": 1.0, "end": 5.0, "plan_id": "p1"}, + {"start": 9.0, "end": 12.0, "plan_id": "p2"}, + ] + }, ) } db = _db(models) - - # 传入 plan_id，但记录本身无 plan_id → 按时间匹配，允许删除 - assert remove_used_segment(db, "a1", 5.0, 9.0, plan_id="plan-new") is True - assert models["a1"].meta()["used_time_ranges"] == [] + assert remove_used_segment(db, "a1", 1.0, 5.0, plan_id="p1") is True + assert len(_ranges(db)) == 1 + assert _ranges(db)[0]["start"] == 9.0 -# ── reset_used_segments ─────────────────────────────────────────────────────── +def test_remove_plan_mismatch_keeps_range(patched_model): + models = {"a1": FakeModel("a1", {"used_time_ranges": [{"start": 1.0, "end": 5.0, "plan_id": "p1"}]})} + db = _db(models) + assert remove_used_segment(db, "a1", 1.0, 5.0, plan_id="other") is False + assert len(_ranges(db)) == 1 + + +def test_remove_legacy_range_without_plan_id(patched_model): + """旧数据记录缺 plan_id → 按时间匹配可删除。""" + models = { + "a1": FakeModel( + "a1", + { + "used_time_ranges": [ + {"start": 2.0, "end": 12.0, "created_at": "2026-01-01T00:00:00"}, + ] + }, + ) + } + db = _db(models) + assert remove_used_segment(db, "a1", 2.0, 12.0, plan_id="plan-new") is True + assert _ranges(db) == [] + + +# ── reset_used_segments（仅运维/测试）───────────────────────────────────────── def test_reset_clears_ranges_keeps_other_fields(patched_model): @@ -250,74 +277,185 @@ def test_reset_clears_ranges_keeps_other_fields(patched_model): "a1": FakeModel( "a1", { - "generation_use_count": 9, - "used_time_ranges": [{"start": 1, "end": 2}], + "generation_use_count": 3, + "used_time_ranges": [ + {"start": 1.0, "end": 5.0}, + ], }, ) } db = _db(models) - reset_used_segments(db, "a1") - - meta = models["a1"].meta() + meta = json.loads(models["a1"].classification_result) assert meta["used_time_ranges"] == [] - assert meta["generation_use_count"] == 9 - assert db.commits == 0 + assert meta["generation_use_count"] == 3 -# ── make_reset_callback ─────────────────────────────────────────────────────── +# ── find_reusable_range：受控复用选择 ───────────────────────────────────────── -def test_reset_callback_clears_persisted_and_memory(patched_model): - models = {"a1": FakeModel("a1", {"used_time_ranges": [{"start": 0, "end": 30}]})} +def test_find_reusable_prefers_oldest_unused(patched_model): + """选 last_used_at 最老、use_count 未达上限的区间；能容纳 clip_duration。""" + models = { + "a1": FakeModel( + "a1", + { + "used_time_ranges": [ + {"start": 0.0, "end": 8.0, "use_count": 1, "last_used_at": "2026-08-01T00:00:00+00:00"}, + { + "start": 10.0, + "end": 20.0, + "use_count": 1, + "last_used_at": "2026-01-01T00:00:00+00:00", + }, # 最久未用 + ] + }, + ) + } db = _db(models) - - used_segments = {"a1": [(0.0, 30.0)], "a2": [(1.0, 2.0)]} - cb = make_reset_callback(db, used_segments) - cb("a1") - - assert "a1" not in used_segments # 内存清空 - assert "a2" in used_segments # 其他素材不受影响 - assert models["a1"].meta()["used_time_ranges"] == [] + result = find_reusable_range(db, "a1", clip_duration=5.0, asset_total=30.0) + assert result is not None + start, end = result + assert start == 10.0 and end == 15.0 -# ── _calc_random_start_time 轮回回调 ────────────────────────────────────────── +def test_find_reusable_excludes_max_use_count(patched_model): + """use_count 达到上限（3）的区间不再参与复用；全部达上限返回 None。""" + models = { + "a1": FakeModel( + "a1", + { + "used_time_ranges": [ + {"start": 0.0, "end": 10.0, "use_count": 3, "last_used_at": "2026-01-01T00:00:00"}, + ] + }, + ) + } + db = _db(models) + assert find_reusable_range(db, "a1", 5.0, 30.0) is None -def test_calc_random_start_invokes_reset_when_exhausted(): - """素材区间被占满（100 次随机必重叠）→ 触发 on_exhausted，重置后重试成功。""" +def test_find_reusable_fourth_use_rejected(patched_model): + """同区间复用第 4 次被拒绝：use_count=2 的可复用，use_count=3 的不可复用。""" + models = { + "a1": FakeModel( + "a1", + { + "used_time_ranges": [ + {"start": 0.0, "end": 10.0, "use_count": 2, "last_used_at": "2026-03-01T00:00:00"}, + {"start": 10.0, "end": 20.0, "use_count": 3, "last_used_at": "2026-01-01T00:00:00"}, + ] + }, + ) + } + db = _db(models) + result = find_reusable_range(db, "a1", 5.0, 30.0) + # 只能选 use_count=2 的区间（start=0），不能选 use_count=3 的（虽然它更老） + assert result is not None and result[0] == 0.0 + + +def test_find_reusable_clamps_to_asset_bounds(patched_model): + """历史区间起点 + clip_duration 会越素材末尾时，起点钳制到 max_start。""" + models = { + "a1": FakeModel( + "a1", + { + "used_time_ranges": [ + {"start": 25.0, "end": 30.0, "use_count": 1, "last_used_at": "2026-01-01T00:00:00"}, + ] + }, + ) + } + db = _db(models) + result = find_reusable_range(db, "a1", clip_duration=10.0, asset_total=30.0) + assert result is not None + start, end = result + assert end <= 30.0 + 1e-6 and start >= 0.0 + + +def test_find_reusable_no_ranges_returns_none(patched_model): + models = {"a1": FakeModel("a1", {"other": 1})} + db = _db(models) + assert find_reusable_range(db, "a1", 5.0, 30.0) is None + + +# ── make_reuse_callback ─────────────────────────────────────────────────────── + + +def test_reuse_callback_returns_range_and_tracks_duration(patched_model): + models = { + "a1": FakeModel( + "a1", + { + "used_time_ranges": [ + {"start": 10.0, "end": 20.0, "use_count": 1, "last_used_at": "2026-01-01T00:00:00"}, + ] + }, + ) + } + db = _db(models) + reused: dict[str, float] = {} + cb = make_reuse_callback(db, {"a1": 30.0}, reused) + result = cb("a1", 8.0) + assert result is not None and result[0] == 10.0 + assert reused["a1"] == 8.0 # 复用时长累加 + + +def test_reuse_callback_db_error_returns_none(patched_model): + class BoomSession: + def query(self, _m): + raise RuntimeError("db down") + + reused: dict[str, float] = {} + cb = make_reuse_callback(BoomSession(), {"a1": 30.0}, reused) + assert cb("a1", 8.0) is None # 异常被吞，返回 None + assert reused == {} + + +# ── _calc_random_start_time 与受控回调集成 ──────────────────────────────────── + + +def test_calc_random_start_uses_reuse_callback_when_exhausted(monkeypatch): + """素材区间被占满、100 次随机找不到空位时，调用复用回调返回历史区间。""" + import packages.domain.plan_generator_utils as pgu + + monkeypatch.setattr(pgu.random, "uniform", lambda a, b: 0.5) # 固定候选点必撞区间 + durations = {"a1": 30.0} - used = {"a1": [(0.0, 10.0), (10.0, 20.0), (20.0, 30.0)]} - reset_called = [] + used = {"a1": [(0.0, 30.0)]} # 全占满 + calls = [] - def _on_exhausted(asset_id): - reset_called.append(asset_id) - used.pop(asset_id, None) # 模拟轮回清空 + def reuse_cb(asset_id, clip_duration): + calls.append((asset_id, clip_duration)) + return (10.0, 18.0) - result = _calc_random_start_time("a1", 10.0, durations, used, on_exhausted=_on_exhausted) - - assert reset_called == ["a1"] - assert result is not None - assert 0.0 <= result <= 20.0 # max_start = 30 - 10 + result = _calc_random_start_time("a1", 8.0, durations, used, on_exhausted=reuse_cb) + assert calls == [("a1", 8.0)] + assert result == 10.0 -def test_calc_random_start_no_callback_keeps_legacy_fallback(): - """不传 on_exhausted 时保持旧降级行为，不报错。""" +def test_calc_random_start_reuse_callback_none_returns_none(monkeypatch): + """复用回调返回 None（区间全部达上限/复用占比超闸门）→ calc 返回 None。 + + 新机制下不做末尾/0.0 重叠降级（那会把片段放回已用过的画面）， + 由调用方轮询下一个素材或报 400；历史记录不被清空。 + """ + import packages.domain.plan_generator_utils as pgu + + monkeypatch.setattr(pgu.random, "uniform", lambda a, b: 0.5) + durations = {"a1": 30.0} - used = {"a1": [(0.0, 10.0), (10.0, 20.0), (20.0, 30.0)]} + used = {"a1": [(0.0, 30.0)]} + used_before = list(used["a1"]) + result = _calc_random_start_time("a1", 8.0, durations, used, on_exhausted=lambda aid, d: None) + assert result is None + assert used["a1"] == used_before # 历史记录未被清空 - result = _calc_random_start_time("a1", 10.0, durations, used) - assert result is not None +def test_calc_random_start_no_callback_backward_compatible(monkeypatch): + """不传 on_exhausted 时行为与旧版兼容（100 次失败走降级）。""" + import packages.domain.plan_generator_utils as pgu - -def test_calc_random_start_with_space_does_not_reset(): - """有充足空闲区间时不触发 reset。""" - durations = {"a1": 100.0} - used = {"a1": [(0.0, 50.0)]} - reset_called = [] - - result = _calc_random_start_time("a1", 5.0, durations, used, on_exhausted=lambda aid: reset_called.append(aid)) - - assert reset_called == [] + monkeypatch.setattr(pgu.random, "uniform", lambda a, b: 0.5) + result = _calc_random_start_time("a1", 8.0, {"a1": 30.0}, {"a1": [(0.0, 30.0)]}) assert result is not None diff --git a/tests/unit/test_batch_no_plan_guard.py b/tests/unit/test_batch_no_plan_guard.py new file mode 100755 index 000000000..ef29f0bcd --- /dev/null +++ b/tests/unit/test_batch_no_plan_guard.py @@ -0,0 +1,95 @@ +"""AI Review 回归：批量生成 count>1 但 source_edit_plan_id 为空时不应 IndexError。 + +变体 plan 预克隆仅在 source_edit_plan_id 非空时执行；无源 plan 时 +variant_plan_ids 为空，循环中禁止索引访问，各任务走自身随机选片流程。 +""" + +import sys +from pathlib import Path +from types import SimpleNamespace +from unittest.mock import MagicMock, patch + +REPO_ROOT = Path(__file__).resolve().parents[2] +if str(REPO_ROOT) not in sys.path: + sys.path.insert(0, str(REPO_ROOT)) + + +def _make_user(): + return SimpleNamespace(user=SimpleNamespace(id="user-1")) + + +def _make_request(count): + from app.schemas.generation_task import CreateGenerationTaskRequest + + return CreateGenerationTaskRequest( + project_id="proj-1", + asset_library_id="lib-1", + strategy_id="one_take", + asset_ids=["a1"], + count=count, + source_edit_plan_id="", # 关键：无源 plan（空字符串为假值） + ) + + +class TestBatchNoSourcePlanNoIndexError: + def test_count3_without_source_plan_creates_three_tasks(self): + """count=3 且无 source_edit_plan_id：不克隆、不 IndexError、创建 3 个任务。""" + 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 + + 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 = 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) diff --git a/tests/unit/test_clone_plan_variant.py b/tests/unit/test_clone_plan_variant.py new file mode 100755 index 000000000..32e6a670b --- /dev/null +++ b/tests/unit/test_clone_plan_variant.py @@ -0,0 +1,201 @@ +"""clone_plan_for_variant 单元测试（Task G 验收项：批量 N 条视频片段独立）。 + +验证： +- 同一源 plan 克隆 3 次产出 3 个不同 plan_id，各自片段起点不同 +- 源 plan 的片段不被修改 +- 模板/config/时长结构被复制 +- 复用占比闸门触发时保留原起点（不重复抽取） +- 源 plan 无片段时抛出 ValueError +""" + +from __future__ import annotations + +import os +import sys +from pathlib import Path +from unittest.mock import MagicMock, patch + +os.environ.setdefault("JWT_SECRET_KEY", "unit-test-secret-key-for-testing") +os.environ.setdefault("DATABASE_URL", "sqlite:///test.db") + +import pytest + +sys.path.insert(0, str(Path(__file__).resolve().parents[2] / "apps" / "api")) +sys.path.insert(0, str(Path(__file__).resolve().parent)) # tests/unit，便于复用同目录 stub + +# 复用 test_edit_plan_service 里的内存 stub 仓储 +from test_edit_plan_service import ( # noqa: E402 + StubEditPlanClipRepository, + StubEditPlanRepository, + _make_service, +) + +from packages.domain.edit_plan_clip import EditPlanClip + + +@pytest.fixture +def svc_with_source(): + """构造带源 plan + 3 个片段的 service（stub 仓储）。""" + svc = _make_service() + # clone 用 self._clip_repo.session 拿 db；stub 无 session，补一个 MagicMock + svc._clip_repo.session = MagicMock() + + source = svc.create_plan(template_id="tpl-001", name="源计划", total_duration=15.0) + + for i in range(3): + clip = EditPlanClip.create( + plan_id=source.id, + clip_type="main", + order=i, + asset_id=f"a{i % 2 + 1}", # a1, a2, a1 + start_time=float(i * 5), + duration=5.0, + ) + svc._clip_repo.create(clip) + return svc, source + + +def _clone_with_fake_calc(svc, source, starts, *, used=None): + """用受控的 calc 起点列表执行一次克隆。 + + starts: 每次 _calc_random_start_time 返回的起点（按片段顺序）。 + 返回 (new_plan, replace_all 调用的 clips_data, calc 调用记录)。 + """ + calc_calls: list[dict] = [] + + def fake_calc(asset_id, clip_duration, durations, used_segments, on_exhausted=None): + idx = len(calc_calls) + calc_calls.append({"asset_id": asset_id, "clip_duration": clip_duration, "on_exhausted": on_exhausted}) + return starts[idx] + + with ( + patch( + "app.services.edit_plan_service.get_used_segments", + return_value=used or {}, + ), + patch( + "app.services.edit_plan_service.make_reuse_callback", + return_value=lambda aid, d: None, + ), + patch( + "app.services.edit_plan_service.record_used_segments", + return_value=None, + ), + patch( + "packages.domain.plan_generator_utils._calc_random_start_time", + side_effect=fake_calc, + ), + patch.object(svc, "replace_all_clips_transactional", return_value=3) as mock_replace, + patch( + "packages.adapters.sqlalchemy_impl.models.AssetModel", + create=True, + ) as mock_asset_model, + ): + # db.query(AssetModel).filter(...).all() → 返回带 duration 的 mock 素材 + m1 = MagicMock(id="a1") + m1.duration = 60.0 + m2 = MagicMock(id="a2") + m2.duration = 60.0 + svc._clip_repo.session.query.return_value.filter.return_value.all.return_value = [m1, m2] + new_plan = svc.clone_plan_for_variant(source.id, created_by_user_id="u1", name_suffix="变体") + clips_data = mock_replace.call_args.args[1] + return new_plan, clips_data, calc_calls + + +class TestClonePlanForVariant: + def test_three_clones_produce_distinct_plans_and_starts(self, svc_with_source): + """克隆 3 次：3 个不同 plan_id，片段起点互不相同（Task G 验收）。""" + svc, source = svc_with_source + start_sets = [ + [10.0, 20.0, 30.0], + [11.0, 21.0, 31.0], + [12.0, 22.0, 32.0], + ] + plans = [] + all_clips = [] + for starts in start_sets: + new_plan, clips_data, _ = _clone_with_fake_calc(svc, source, starts) + plans.append(new_plan) + all_clips.append(clips_data) + + # 3 个不同 plan_id，且都不等于源 plan + plan_ids = {p.id for p in plans} + assert len(plan_ids) == 3 + assert source.id not in plan_ids + + # 每次克隆的起点各自不同 + for clips_data, starts in zip(all_clips, start_sets, strict=True): + assert [c["start_time"] for c in clips_data] == starts + + # 三次克隆的起点集合互不相同 + assert {tuple(c["start_time"] for c in clips) for clips in all_clips} == { + (10.0, 20.0, 30.0), + (11.0, 21.0, 31.0), + (12.0, 22.0, 32.0), + } + + def test_source_plan_not_modified(self, svc_with_source): + """克隆不修改源 plan 及其片段（保留用户手动编辑）。""" + svc, source = svc_with_source + source_clips_before = sorted( + [(c.order, c.asset_id, c.start_time, c.duration) for c in svc._clip_repo.list_by_plan(source.id)] + ) + source_name_before = source.name + + _clone_with_fake_calc(svc, source, [9.0, 19.0, 29.0]) + _clone_with_fake_calc(svc, source, [8.0, 18.0, 28.0]) + + source_clips_after = sorted( + [(c.order, c.asset_id, c.start_time, c.duration) for c in svc._clip_repo.list_by_plan(source.id)] + ) + assert source_clips_after == source_clips_before + assert svc._plan_repo.get(source.id).name == source_name_before + + def test_clone_copies_structure(self, svc_with_source): + """克隆复制 template_id / config / total_duration / 片段素材与时长。""" + svc, source = svc_with_source + source.config = {"mode": "ONE_TAKE"} + new_plan, clips_data, _ = _clone_with_fake_calc(svc, source, [10.0, 20.0, 30.0]) + + assert new_plan.template_id == source.template_id + assert new_plan.total_duration == source.total_duration + assert new_plan.config == {"mode": "ONE_TAKE"} + assert "变体" in new_plan.name + # 片段素材与时长结构保持 + assert [c["asset_id"] for c in clips_data] == ["a1", "a2", "a1"] + assert all(c["duration"] == 5.0 for c in clips_data) + assert [c["order"] for c in clips_data] == [0, 1, 2] + + def test_clone_uses_reuse_callback(self, svc_with_source): + """克隆时 calc 传入了 on_exhausted 受控复用回调（耗尽时复用而非清空历史）。""" + svc, source = svc_with_source + _, _, calc_calls = _clone_with_fake_calc(svc, source, [10.0, 20.0, 30.0]) + assert len(calc_calls) == 3 + for call in calc_calls: + assert call["on_exhausted"] is not None + + def test_clone_ratio_blocked_keeps_original_start(self, svc_with_source): + """复用占比闸门触发（calc 返回 None）时保留源片段原起点。""" + svc, source = svc_with_source + # 第 3 个片段 calc 返回 None（模拟复用占比超 15% 拒绝复用） + new_plan, clips_data, _ = _clone_with_fake_calc(svc, source, [10.0, 20.0, None]) # type: ignore[list-item] + starts = [c["start_time"] for c in clips_data] + assert starts[0] == 10.0 + assert starts[1] == 20.0 + # 第 3 片段保留源起点（源 order=2 → start_time=10.0） + assert starts[2] == 10.0 + + def test_clone_empty_source_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="无片段"): + svc.clone_plan_for_variant(empty.id, name_suffix="变体") + + def test_clone_nonexistent_source_raises(self): + """源 plan 不存在时抛出 ValueError。""" + svc = _make_service() + svc._clip_repo.session = MagicMock() + with pytest.raises(ValueError, match="不存在"): + svc.clone_plan_for_variant("no-such-plan", name_suffix="变体") diff --git a/tests/unit/test_editor_clips_random_start.py b/tests/unit/test_editor_clips_random_start.py index 459200a92..6501a6fa7 100644 --- a/tests/unit/test_editor_clips_random_start.py +++ b/tests/unit/test_editor_clips_random_start.py @@ -5,7 +5,7 @@ - 素材不足时同一素材轮询切多个片段 - 随机 start_time + used_segments 去重 - 素材时长不足时 clip duration 缩短 -- 素材时长为 0 时抛 400 +- 素材时长全部为 0/缺失时抛 400「素材可切区间不足」；混合池中零时长素材被跳过 - 使用 replace_all_clips_transactional 原子性替换 - order 从 0 开始 - start_time=None 时抛出 400 @@ -98,8 +98,8 @@ def _mock_segment_tracker(): return_value=None, ), patch( - "app.api.routes.templates_editor.clips.make_reset_callback", - return_value=lambda asset_id: None, + "app.api.routes.templates_editor.clips.make_reuse_callback", + return_value=lambda asset_id, clip_duration: None, ), patch( "app.api.routes.templates_editor.clips.remove_used_segment", @@ -276,7 +276,11 @@ class TestEditorClipsDurationAndStartTime: @patch("app.api.routes.templates_editor.clips.get_storage_service") def test_zero_duration_asset_raises_400(self, mock_storage): - """素材时长为 0 时应抛出 400，而不是创建无效片段。""" + """所有素材时长均为 0 时轮询无可用素材，抛出 400「素材可切区间不足」。 + + 新轮询逻辑下零时长素材被跳过（而非立即报错）；全部素材都被跳过时 + 返回 400，不创建无效片段。 + """ from app.api.routes.templates_editor.clips import ( create_clips_from_assets_editor, ) @@ -302,7 +306,49 @@ class TestEditorClipsDurationAndStartTime: ) assert exc_info.value.status_code == 400 - assert "时长" in exc_info.value.detail + assert "素材可切区间不足" in exc_info.value.detail + + @patch("app.api.routes.templates_editor.clips.get_storage_service") + def test_zero_duration_asset_skipped_in_mixed_pool(self, mock_storage): + """素材池混合零时长与正常素材时，零时长素材被跳过、正常素材承担片段。""" + from app.api.routes.templates_editor.clips import ( + create_clips_from_assets_editor, + ) + from app.api.routes.templates_editor.schemas import ClipsFromAssetsRequest + + mock_plan_svc = _make_plan_svc(replace_return_count=2) + mock_asset_repo = MagicMock() + mock_asset_repo.get = MagicMock( + side_effect=lambda aid: { + "zero": _make_mock_asset("zero", 0.0), + "good": _make_mock_asset("good", 30.0), + }[aid] + ) + + body = ClipsFromAssetsRequest(asset_ids=["zero", "good"], required_clips_count=2) + + with ( + _patch_segments(_segments(2)), + patch( + "app.api.routes.templates_editor.clips._calc_random_start_time", + side_effect=[5.0, 12.0], + ), + ): + create_clips_from_assets_editor( + template_id="tpl-001", + body=body, + background_tasks=MagicMock(), + plan_id=TEST_PLAN_ID, + services=(MagicMock(), mock_plan_svc), + asset_repo=mock_asset_repo, + db=MagicMock(), + current_user=_make_auth_user(), + ) + + clips_data = _get_clips_data_from_call(mock_plan_svc) + assert len(clips_data) == 2 + # 所有片段都分配给正常素材，零时长素材被跳过 + assert all(c["asset_id"] == "good" for c in clips_data) @patch("app.api.routes.templates_editor.clips.get_storage_service") def test_missing_duration_asset_raises_400(self, mock_storage): @@ -450,7 +496,7 @@ class TestEditorClipsErrorHandling: @patch("app.api.routes.templates_editor.clips.get_storage_service") def test_none_start_time_raises_400(self, mock_storage): - """_calc_random_start_time 返回 None 时应抛出 HTTPException 400。""" + """所有素材 calc 均返回 None（区间耗尽且复用被拒）→ 轮询失败抛 400。""" from app.api.routes.templates_editor.clips import ( create_clips_from_assets_editor, ) @@ -458,7 +504,7 @@ class TestEditorClipsErrorHandling: mock_plan_svc = _make_plan_svc() mock_asset_repo = MagicMock() - # 素材有 duration 但 random 返回 None + # 素材有 duration 但 calc 返回 None（模拟可用区间耗尽、复用被闸门拒绝） mock_asset_repo.get = MagicMock(return_value=_make_mock_asset("a1", 30.0)) body = ClipsFromAssetsRequest(asset_ids=["a1"]) @@ -483,7 +529,9 @@ class TestEditorClipsErrorHandling: ) assert exc_info.value.status_code == 400 - assert "时长" in exc_info.value.detail + assert "素材可切区间不足" in exc_info.value.detail + # 复用被拒导致无起点时，不应创建任何片段 + assert not mock_plan_svc.replace_all_clips_transactional.called @patch("app.api.routes.templates_editor.clips.get_storage_service") def test_transactional_replace_exception_propagates(self, mock_storage): @@ -512,3 +560,190 @@ class TestEditorClipsErrorHandling: db=MagicMock(), current_user=_make_auth_user(), ) + + +class TestReuseRatioGate: + """素材区间耗尽后的受控复用与 15% 占比闸门（路由级）。""" + + @staticmethod + def _make_calc_with_reuse(normal_starts, reused_durations): + """构造模拟「区间耗尽后受控复用」的 _calc_random_start_time。 + + normal_starts: list[float | None]，前 N 次调用返回的空闲起点； + 返回 None 表示随机找不到空闲 → 触发 on_exhausted 复用回调。 + 回调被调用时返回复用区间（固定 0.0 起点），复用片段时长由路由累加到 + reused_durations；回调内部占比预判超 15% 时返回 None（calc 随之 None）。 + """ + calls = {"i": 0} + + def fake_calc(asset_id, clip_duration, durations, used_segments, on_exhausted=None): + i = calls["i"] + calls["i"] += 1 + if i < len(normal_starts) and normal_starts[i] is not None: + return normal_starts[i] + # 空闲耗尽 → 走受控复用回调（回调返回 (start, end) 元组，calc 取起点） + if on_exhausted is not None: + result = on_exhausted(asset_id, clip_duration) + return result[0] if result else None + return None + + return fake_calc, calls + + @patch("app.api.routes.templates_editor.clips.get_storage_service") + def test_reused_clip_ratio_within_threshold(self, mock_storage): + """素材 60s、片段 5s：前 12 个用空闲区间，第 13 个复用， + 复用占比 5/(12*5+5)=7.7% ≤ 15%，正常创建 13 个片段。""" + from app.api.routes.templates_editor.clips import ( + create_clips_from_assets_editor, + ) + from app.api.routes.templates_editor.schemas import ClipsFromAssetsRequest + + mock_plan_svc = _make_plan_svc(replace_return_count=13) + mock_asset_repo = MagicMock() + mock_asset_repo.get = MagicMock(return_value=_make_mock_asset("a1", 60.0)) + body = ClipsFromAssetsRequest(asset_ids=["a1"], required_clips_count=13) + + reused: dict = {} + + def reuse_cb(aid, dur): + # 模拟真实回调：返回复用区间前记录复用时长 + reused[aid] = reused.get(aid, 0.0) + dur + return (0.0, dur) + + # 前 12 次分配空闲起点；第 13 次 calc 直接走回调（normal_starts 越界 → None → 回调） + normal_starts = [float(i * 5) for i in range(12)] + fake_calc, _ = self._make_calc_with_reuse(normal_starts, reused) + with ( + _patch_segments(_segments(13, dur_min=5.0, dur_max=5.0)), + patch("app.api.routes.templates_editor.clips._calc_random_start_time", side_effect=fake_calc), + patch("app.api.routes.templates_editor.clips.make_reuse_callback", return_value=reuse_cb), + ): + create_clips_from_assets_editor( + template_id="tpl-001", + body=body, + background_tasks=MagicMock(), + plan_id=TEST_PLAN_ID, + services=(MagicMock(), mock_plan_svc), + asset_repo=mock_asset_repo, + db=MagicMock(), + current_user=_make_auth_user(), + ) + clips_data = _get_clips_data_from_call(mock_plan_svc) + assert len(clips_data) == 13 + # 1 个复用片段，占比 1/13 ≈ 7.7% ≤ 15% + assert reused.get("a1", 0.0) == 5.0 + + @patch("app.api.routes.templates_editor.clips.get_storage_service") + def test_reuse_ratio_exceeded_returns_400(self, mock_storage): + """复用占比将超 15% 时回调拒绝复用 → 无可用素材 → 400「素材可切区间不足」。 + + 60s 素材、5s 片段：前 12 个空闲、随后复用占比累计；当 (reused+d)/(assigned+d) + 超过 15% 时回调返回 None，calc 返回 None，轮询无素材 → 400。 + """ + from app.api.routes.templates_editor.clips import ( + create_clips_from_assets_editor, + ) + from app.api.routes.templates_editor.schemas import ClipsFromAssetsRequest + + mock_plan_svc = _make_plan_svc(replace_return_count=0) + mock_asset_repo = MagicMock() + mock_asset_repo.get = MagicMock(return_value=_make_mock_asset("a1", 60.0)) + body = ClipsFromAssetsRequest(asset_ids=["a1"], required_clips_count=20) + + # 模拟真实回调：累计复用时长，预判超 15% 拒绝 + reused: dict = {} + assigned: dict = {} + + def fake_reuse_cb(aid, clip_duration): + a = assigned.get(aid, 0.0) + r = reused.get(aid, 0.0) + if a > 0 and (r + clip_duration) / (a + clip_duration) > 0.15: + return None # 占比闸门拒绝 + reused[aid] = r + clip_duration + return (0.0, clip_duration) + + def fake_calc(asset_id, clip_duration, durations, used_segments, on_exhausted=None): + a = assigned.get(asset_id, 0.0) + # 前 12 个片段（60s/5s）有空闲区间 + if a < 60.0: + start = a + assigned[asset_id] = a + clip_duration + return start + # 之后空闲耗尽 → 复用 + if on_exhausted is not None: + result = on_exhausted(asset_id, clip_duration) + if result is not None: + assigned[asset_id] = assigned.get(asset_id, 0.0) + clip_duration + return result[0] if result else None + return None + + with ( + _patch_segments(_segments(20, dur_min=5.0, dur_max=5.0)), + patch("app.api.routes.templates_editor.clips._calc_random_start_time", side_effect=fake_calc), + patch("app.api.routes.templates_editor.clips.make_reuse_callback", return_value=fake_reuse_cb), + ): + with pytest.raises(HTTPException) as exc_info: + create_clips_from_assets_editor( + template_id="tpl-001", + body=body, + background_tasks=MagicMock(), + plan_id=TEST_PLAN_ID, + services=(MagicMock(), mock_plan_svc), + asset_repo=mock_asset_repo, + db=MagicMock(), + current_user=_make_auth_user(), + ) + + assert exc_info.value.status_code == 400 + assert "素材可切区间不足" in exc_info.value.detail + # 闸门在复用占比达上限时拒绝：60s 空闲 + 至多 ~15% 复用 + assert reused.get("a1", 0.0) <= 12.0 # 10.0 或 15.0 以内，不会无限复用 + # 未创建任何片段（整批失败） + assert not mock_plan_svc.replace_all_clips_transactional.called + + @patch("app.api.routes.templates_editor.clips.get_storage_service") + def test_calc_none_falls_through_to_next_asset(self, mock_storage): + """一个素材区间耗尽且复用被拒（calc 返回 None）时，轮询到下一个可用素材。""" + from app.api.routes.templates_editor.clips import ( + create_clips_from_assets_editor, + ) + from app.api.routes.templates_editor.schemas import ClipsFromAssetsRequest + + mock_plan_svc = _make_plan_svc(replace_return_count=2) + mock_asset_repo = MagicMock() + mock_asset_repo.get = MagicMock( + side_effect=lambda aid: { + "exhausted": _make_mock_asset("exhausted", 60.0), + "fresh": _make_mock_asset("fresh", 60.0), + }[aid] + ) + body = ClipsFromAssetsRequest(asset_ids=["exhausted", "fresh"], required_clips_count=2) + + def fake_calc(asset_id, clip_duration, durations, used_segments, on_exhausted=None): + if asset_id == "exhausted": + # 空闲耗尽 + 回调拒绝 → None + return on_exhausted(asset_id, clip_duration) if on_exhausted else None + return 8.0 # 新鲜素材正常返回 + + with ( + _patch_segments(_segments(2, dur_min=5.0, dur_max=5.0)), + patch("app.api.routes.templates_editor.clips._calc_random_start_time", side_effect=fake_calc), + patch( + "app.api.routes.templates_editor.clips.make_reuse_callback", + return_value=lambda aid, d: None, # 复用始终被拒 + ), + ): + create_clips_from_assets_editor( + template_id="tpl-001", + body=body, + background_tasks=MagicMock(), + plan_id=TEST_PLAN_ID, + services=(MagicMock(), mock_plan_svc), + asset_repo=mock_asset_repo, + db=MagicMock(), + current_user=_make_auth_user(), + ) + clips_data = _get_clips_data_from_call(mock_plan_svc) + assert len(clips_data) == 2 + # 耗尽素材被跳过，两个片段都分配给新鲜素材 + assert all(c["asset_id"] == "fresh" for c in clips_data) diff --git a/tests/unit/test_generation_preview.py b/tests/unit/test_generation_preview.py index bb97b7f2e..696e889aa 100644 --- a/tests/unit/test_generation_preview.py +++ b/tests/unit/test_generation_preview.py @@ -612,6 +612,8 @@ def _make_task( task.extra_meta = extra_meta or {} task.asset_ids = asset_ids or [] task.created_by_user_id = "test_user_001" + # 默认无关联编辑计划：涉及克隆变体的测试自行设置并 mock EditPlanService + task.source_edit_plan_id = None return task diff --git a/tests/unit/test_mediakit_conflicts.py b/tests/unit/test_mediakit_conflicts.py new file mode 100755 index 000000000..5c5a7cf96 --- /dev/null +++ b/tests/unit/test_mediakit_conflicts.py @@ -0,0 +1,68 @@ +"""MediaKit 智能选片挪点冲突检测测试（Task G 验收项 1）。 + +覆盖 _recommended_time_conflicts： +- 区间重叠判定（含 0.3s 边缘间隙扩边） +- 不冲突场景（间隔大于边缘间隙） +- 边缘间隙可配置 +""" + +from __future__ import annotations + +import os +import sys +from pathlib import Path + +os.environ.setdefault("JWT_SECRET_KEY", "unit-test-secret-key-for-testing") +os.environ.setdefault("DATABASE_URL", "sqlite:///test.db") + +sys.path.insert(0, str(Path(__file__).resolve().parents[2] / "apps" / "api")) + +from app.api.routes.templates_editor.clips import ( + SEGMENT_EDGE_GAP, + _recommended_time_conflicts, +) + + +class TestRecommendedTimeConflicts: + def test_overlapping_range_conflicts(self): + """推荐区间与已用区间直接重叠 → 冲突。""" + assert _recommended_time_conflicts(10.0, 5.0, [(12.0, 17.0)]) is True + + def test_identical_range_conflicts(self): + assert _recommended_time_conflicts(10.0, 5.0, [(10.0, 15.0)]) is True + + def test_touching_endpoint_conflicts_due_to_edge_gap(self): + """首尾紧贴（推荐 15 开始，已用 [10,15]）：0.3s 扩边内 → 冲突。""" + assert _recommended_time_conflicts(15.0, 5.0, [(10.0, 15.0)]) is True + + def test_gap_within_edge_gap_conflicts(self): + """间隔 0.2s（< 0.3s 边缘间隙）→ 冲突。""" + assert _recommended_time_conflicts(15.2, 5.0, [(10.0, 15.0)]) is True + + def test_gap_beyond_edge_gap_no_conflict(self): + """间隔 0.5s（> 0.3s 边缘间隙）→ 不冲突。""" + assert _recommended_time_conflicts(15.5, 5.0, [(10.0, 15.0)]) is False + + def test_far_apart_no_conflict(self): + """相隔很远 → 不冲突。""" + assert _recommended_time_conflicts(20.0, 5.0, [(0.0, 5.0)]) is False + + def test_empty_used_no_conflict(self): + assert _recommended_time_conflicts(10.0, 5.0, []) is False + + def test_any_one_range_conflicts(self): + """多个已用区间，任一冲突即返回 True。""" + used = [(0.0, 5.0), (10.0, 15.0), (20.0, 25.0)] + assert _recommended_time_conflicts(12.0, 2.0, used) is True + assert _recommended_time_conflicts(6.0, 2.0, used) is False + + def test_custom_edge_gap(self): + """edge_gap 可配置：gap=0 时紧贴不冲突（端点相接不算重叠）。""" + # edge_gap=0：15.0 开始与已用 [10,15] 端点相接，区间判定 start 0.3）→ 不冲突 + assert _recommended_time_conflicts(10.5, 5.0, [(0.0, 10.0)]) is False def test_conflict_multiple_used(self): from app.api.routes.templates_editor.clips import _recommended_time_conflicts @@ -78,8 +83,10 @@ class TestRecommendedTimeConflicts: used = [(0.0, 5.0), (10.0, 15.0), (20.0, 25.0)] # 推荐 [6, 11] 与 [10, 15] 冲突 assert _recommended_time_conflicts(6.0, 5.0, used) is True - # 推荐 [15, 20] 不冲突 - assert _recommended_time_conflicts(15.0, 5.0, used) is False + # 推荐 [15, 20] 与 [10, 15] 首尾相接：0.3s 扩边内 → 冲突 + assert _recommended_time_conflicts(15.0, 5.0, used) is True + # 空闲段 [5.3, 9.7] 长 4.4s：推荐 [5.5, 9.5]（dur=4）与三区间扩边均不接触 + assert _recommended_time_conflicts(5.5, 4.0, used) is False # ── _get_mediakit_recommendations 单元测试 ────────────────────────────────── diff --git a/tests/unit/test_preview_edit_plan_association.py b/tests/unit/test_preview_edit_plan_association.py index 60aa89869..21c41e826 100644 --- a/tests/unit/test_preview_edit_plan_association.py +++ b/tests/unit/test_preview_edit_plan_association.py @@ -13,6 +13,7 @@ from __future__ import annotations import os import sys from dataclasses import dataclass, field +from types import SimpleNamespace from typing import Any, Optional from unittest.mock import MagicMock, patch @@ -180,17 +181,19 @@ class TestPreviewEditPlanAutoAssociation: "packages.adapters.sqlalchemy_impl.edit_plan_repository.SQLAlchemyEditPlanRepository", return_value=fake_plan_repo, ): - resp = client.post( - "/api/v1/generation/preview", - json=_make_request_body(source_edit_plan_id=""), - ) + with patch("app.services.edit_plan_service.EditPlanService") as MockPlanSvc: + MockPlanSvc.return_value.clone_plan_for_variant.return_value = SimpleNamespace(id="plan-clone-001") + resp = client.post( + "/api/v1/generation/preview", + json=_make_request_body(source_edit_plan_id=""), + ) assert resp.status_code == 201 - # 找到 store 中的 task 并验证 source_edit_plan_id 被设置 + # 找到 store 中的 task 并验证 source_edit_plan_id 被设置（自动关联后再克隆为独立 plan） tasks = list(gen_task_repo._store.values()) assert len(tasks) == 1 task = tasks[0] - assert task.source_edit_plan_id == "plan-auto-001" + assert task.source_edit_plan_id == "plan-clone-001" @patch( "app.api.routes.generation_preview._resolve_strategy_id_from_template", @@ -209,16 +212,18 @@ class TestPreviewEditPlanAutoAssociation: client: TestClient, gen_task_repo: StubGenerationTaskRepository, ): - """前端已传 source_edit_plan_id 时，不应触发自动关联""" - resp = client.post( - "/api/v1/generation/preview", - json=_make_request_body(source_edit_plan_id="plan-explicit-001"), - ) + """前端已传 source_edit_plan_id 时，不触发自动关联，但仍克隆独立变体 plan""" + with patch("app.services.edit_plan_service.EditPlanService") as MockPlanSvc: + MockPlanSvc.return_value.clone_plan_for_variant.return_value = SimpleNamespace(id="plan-clone-explicit") + resp = client.post( + "/api/v1/generation/preview", + json=_make_request_body(source_edit_plan_id="plan-explicit-001"), + ) assert resp.status_code == 201 tasks = list(gen_task_repo._store.values()) assert len(tasks) == 1 - assert tasks[0].source_edit_plan_id == "plan-explicit-001" + assert tasks[0].source_edit_plan_id == "plan-clone-explicit" @patch( "app.api.routes.generation_preview._resolve_strategy_id_from_template",