From cd8fb98c1bb3cb7e36d9a139be1c7d82f9813c05 Mon Sep 17 00:00:00 2001 From: CI Bot Date: Thu, 23 Jul 2026 10:36:19 +0800 Subject: [PATCH 1/2] =?UTF-8?q?feat(#632):=20=E9=85=8D=E9=9F=B3=E7=B4=A0?= =?UTF-8?q?=E6=9D=90=E5=BA=93=E9=9F=B3=E9=A2=91=E6=9E=B6=E6=9E=84=E7=BB=9F?= =?UTF-8?q?=E4=B8=80=EF=BC=8C=E8=BF=81=E7=A7=BB=E5=88=B0audio=E5=9B=BE?= =?UTF-8?q?=E5=B1=82=E6=B7=B7=E9=9F=B3?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - UnifiedRenderService 增加 voiceover_audio_path 参数 - 新增 _maybe_add_voice_library_layer 方法，配音素材库音频加到audio图层 - 与TTS配音、BGM共享统一混音链路，支持音量平衡 - RenderAdapter 透传 voiceover_audio_path 参数 - 删除一键生成路径中后处理 _mux_audio_track 调用 - 行为：从替换音轨改为混音叠加，与TTS配音行为一致 --- .../worker/video_processing/render_adapter.py | 6 ++ .../unified_render_service.py | 68 +++++++++++++++++++ apps/worker/worker_app/tasks/generation.py | 14 +--- 3 files changed, 77 insertions(+), 11 deletions(-) diff --git a/apps/worker/video_processing/render_adapter.py b/apps/worker/video_processing/render_adapter.py index d1abb4089..8d84f9735 100755 --- a/apps/worker/video_processing/render_adapter.py +++ b/apps/worker/video_processing/render_adapter.py @@ -483,6 +483,7 @@ class RenderAdapter: progress_cb: ProgressCallback | None = None, rendered_clip_ids: list[str] | None = None, failed_clip_ids: list[str] | None = None, + voiceover_audio_path: str | None = None, ) -> RenderAdapterResult: """执行统一渲染核心流程（BGM + ASR + 渲染 + 缩略图 + 上传）。 @@ -491,6 +492,7 @@ class RenderAdapter: Args: rendered_clip_ids: 成功下载/准备的 clip id 列表（render_plan 从下载阶段传入） failed_clip_ids: 失败的 clip id 列表 + voiceover_audio_path: 配音素材库音频本地路径（一键生成场景使用） Returns: RenderAdapterResult @@ -525,6 +527,7 @@ class RenderAdapter: output_height=output_height, bgm_path=bgm_path, asr_service=asr_service, + voiceover_audio_path=voiceover_audio_path, ) result = render_svc.render() @@ -593,6 +596,7 @@ class RenderAdapter: job_id: str = "", work_dir: Path | None = None, progress_cb: ProgressCallback | None = None, + voiceover_audio_path: str | None = None, ) -> RenderAdapterResult: """使用内存中的 plan/clips/asset_path_map 直接渲染。 @@ -606,6 +610,7 @@ class RenderAdapter: job_id: 关联的 Job ID work_dir: 工作目录，不传则用临时目录 progress_cb: 进度回调 + voiceover_audio_path: 配音素材库音频本地路径 Returns: RenderAdapterResult @@ -649,6 +654,7 @@ class RenderAdapter: plan_id=actual_plan_id, job_id=job_id, progress_cb=progress_cb, + voiceover_audio_path=voiceover_audio_path, ) except subprocess.CalledProcessError as exc: diff --git a/apps/worker/video_processing/unified_render_service.py b/apps/worker/video_processing/unified_render_service.py index 67401439a..45a3e0c85 100755 --- a/apps/worker/video_processing/unified_render_service.py +++ b/apps/worker/video_processing/unified_render_service.py @@ -175,6 +175,7 @@ class UnifiedRenderService: transition_duration: float = DEFAULT_TRANSITION_DURATION, asr_service: Any = None, # ASRService 实例，用于自动生成字幕 bgm_path: str | None = None, # BGM 本地文件路径 + voiceover_audio_path: str | None = None, # 配音素材库音频本地路径 ): self.plan = plan self.clips = clips @@ -186,6 +187,7 @@ class UnifiedRenderService: self.transition_duration = transition_duration self.asr_service = asr_service self.bgm_path = bgm_path + self.voiceover_audio_path = voiceover_audio_path self._transition_engine = TransitionEngine(default_duration=transition_duration) self._speed_engine = SpeedEngine() self._asr_timeline_cache: Any = None # ASR 字幕结果缓存，避免重复调用 @@ -226,6 +228,9 @@ class UnifiedRenderService: # 3.5 TTS 配音生成（如果配置了） self._maybe_add_voiceover_layer(layers, video_duration=video_duration) + # 3.6 配音素材库音频（如果传入了本地路径） + self._maybe_add_voice_library_layer(layers, video_duration=video_duration) + # 4. 生成 ASS 字幕文件（如果有 title/subtitle 配置） ass_path = self._maybe_generate_ass(video_duration) @@ -843,6 +848,69 @@ class UnifiedRenderService: logger.warning("TTS 配音异常，跳过: %s", e) return False + def _maybe_add_voice_library_layer( + self, + layers: list[RenderLayer], + *, + video_duration: float, + ) -> bool: + """将配音素材库音频作为整段配音加到 audio 图层. + + 与 TTS 配音共享同一套 audio 图层混音架构， + 支持与 BGM、TTS 的音量平衡，不再走独立的后处理 mux 链路。 + + Returns: + 是否成功添加了配音音轨 + """ + if not self.voiceover_audio_path: + return False + + audio_path = Path(self.voiceover_audio_path) + if not audio_path.exists() or audio_path.stat().st_size == 0: + logger.warning("配音素材库音频文件不存在或为空，跳过: %s", self.voiceover_audio_path) + return False + + try: + # 找到或创建 audio 图层 + audio_layer = None + for layer in layers: + if layer.role == "audio": + audio_layer = layer + break + + if audio_layer is None: + from video_processing.unified_render_service import _LAYER_Z_INDEX # type: ignore + + z_index = _LAYER_Z_INDEX.get("audio", 2) + audio_layer = RenderLayer(role="audio", z_index=z_index) + layers.append(audio_layer) + + # 配音素材作为整段配音：从 0 开始，覆盖整个视频时长 + # 音频不足视频时长时，混音层会按实际长度处理（amix 不自动循环） + vo_clip = ResolvedClip( + clip_id="voice_library_main", + asset_id="voice_library", + local_path=audio_path, + clip_type="audio", + order=len(audio_layer.clips), + start_time=0.0, + duration=video_duration, + config={"volume": 1.0, "voice_library": True}, + actual_duration=video_duration, + ) + audio_layer.clips.append(vo_clip) + + logger.info( + "配音素材库音频已添加到 audio 图层: plan_id=%s duration=%.2fs", + self.plan.id, + video_duration, + ) + return True + + except Exception as e: + logger.warning("配音素材库音频添加失败，跳过: %s", e) + return False + @staticmethod def _resolve_watermark_config(plan_config: dict[str, Any] | None) -> WatermarkConfig | None: """从 plan config 中解析水印配置，兼容两种存储格式. diff --git a/apps/worker/worker_app/tasks/generation.py b/apps/worker/worker_app/tasks/generation.py index 6b1e1bd37..d165295ea 100755 --- a/apps/worker/worker_app/tasks/generation.py +++ b/apps/worker/worker_app/tasks/generation.py @@ -1238,6 +1238,7 @@ def _render_video( plan_id=f"gen_{task_id}", job_id=task_id, work_dir=temp_path, + voiceover_audio_path=voice_path, ) finally: db.close() @@ -1256,17 +1257,8 @@ def _render_video( render_duration, ) - # 配音混音（素材库音频，后处理混音） - if voice_path: - final_path = temp_path / f"final-{task_id}.mp4" - try: - _mux_audio_track(render_output_path, voice_path, final_path) - output_path = final_path - except Exception as mux_err: - logger.warning("[task_id=%s] [混音] 音频混合失败，使用无音频版本: %s", task_id, mux_err) - output_path = render_output_path - else: - output_path = render_output_path + # 配音素材库音频已在统一渲染引擎内部通过 audio 图层混音处理 + output_path = render_output_path return output_path, render_duration -- 2.54.0 From 898372c9042099a061bba0424f1ed1af657f294d Mon Sep 17 00:00:00 2001 From: CI Bot Date: Thu, 23 Jul 2026 10:54:56 +0800 Subject: [PATCH 2/2] =?UTF-8?q?feat(#584):=20=E6=99=BA=E8=83=BD=E5=8C=B9?= =?UTF-8?q?=E9=85=8D=E8=A7=86=E9=A2=91=E7=B4=A0=E6=9D=90=E5=A2=9E=E5=BC=BA?= =?UTF-8?q?=20=E2=80=94=20=E5=A4=9A=E7=BB=B4=E5=BA=A6=E8=AF=84=E5=88=86+?= =?UTF-8?q?=E5=A4=9A=E6=A0=B7=E6=80=A7=E4=BF=9D=E8=AF=81+=E9=A1=B9?= =?UTF-8?q?=E7=9B=AE=E7=BA=A7=E6=94=AF=E6=8C=81?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 新增 SmartAssetSelector 智能素材选择服务 - 评分维度：质量分(50%) + 分辨率(20%) + 时长合理性(20%) + 码率(10%) - 时长多样性保证：短/中/长三分桶配额，避免全是同类型素材 - 最低质量分门槛：自动过滤低于30分的低质量素材 - 项目级模式也支持 random/smart 自动选取（之前只有素材库模式支持） - 28个单元测试覆盖评分、选择逻辑、多样性、边界情况 --- apps/api/app/api/routes/generation_tasks.py | 32 +- apps/api/app/services/smart_asset_selector.py | 329 ++++++++++++++ tests/unit/test_smart_asset_selector.py | 413 ++++++++++++++++++ 3 files changed, 762 insertions(+), 12 deletions(-) create mode 100755 apps/api/app/services/smart_asset_selector.py create mode 100755 tests/unit/test_smart_asset_selector.py diff --git a/apps/api/app/api/routes/generation_tasks.py b/apps/api/app/api/routes/generation_tasks.py index cfc395d0b..e982bb248 100755 --- a/apps/api/app/api/routes/generation_tasks.py +++ b/apps/api/app/api/routes/generation_tasks.py @@ -30,6 +30,7 @@ from app.schemas.generation_task import ( GenerationTaskResponse, ListGenerationTasksResponse, ) +from app.services.smart_asset_selector import SmartAssetSelector from fastapi import APIRouter, Depends, HTTPException from packages.application import ( @@ -105,7 +106,7 @@ def _select_assets_from_library( Args: assets: 素材库中所有素材（Asset 实体列表） - mode: 选取模式 — all=全部, random=随机, smart=按质量评分 + mode: 选取模式 — all=全部, random=随机, smart=智能匹配（多维度评分+多样性） count: 选取数量，0 表示全部（仅 random/smart 模式有效） Returns: @@ -123,17 +124,10 @@ def _select_assets_from_library( return [a.id for a in selected] if mode == "smart": - # 按质量分降序排列（质量分高的优先），质量分相同时按时长降序 - sorted_assets = sorted( - ready_video_assets, - key=lambda a: ( - a.quality_score if a.quality_score is not None else 0.0, - a.duration if a.duration is not None else 0.0, - ), - reverse=True, - ) - selected = sorted_assets if count <= 0 else sorted_assets[:count] - return [a.id for a in selected] + # 智能匹配：多维度综合评分 + 时长多样性保证 + selector = SmartAssetSelector() + result = selector.select(ready_video_assets, count=count, ensure_diversity=True) + return result.selected_ids # 默认 all 模式：返回全部 ready 视频素材 return [a.id for a in ready_video_assets] @@ -224,6 +218,20 @@ def create_generation_task( mode=request.asset_select_mode, count=request.asset_select_count, ) + elif project_id and not resolved_asset_ids and request.asset_select_mode in ("random", "smart"): + # 项目级模式：未指定 asset_ids 且选择了 random/smart 模式时，也自动选取 + assets = asset_repository.find_by_project(project_id) + if assets: + resolved_asset_ids = _select_assets_from_library( + assets, + mode=request.asset_select_mode, + count=request.asset_select_count, + ) + if not resolved_asset_ids: + raise HTTPException( + status_code=422, + detail="当前项目没有符合条件的视频素材，请先上传并等待导入完成后再生成。", + ) use_case = CreateGenerationTaskUseCase(generation_task_repository) count = request.count diff --git a/apps/api/app/services/smart_asset_selector.py b/apps/api/app/services/smart_asset_selector.py new file mode 100755 index 000000000..8e781f508 --- /dev/null +++ b/apps/api/app/services/smart_asset_selector.py @@ -0,0 +1,329 @@ +"""SmartAssetSelector — 智能素材选择服务. + +根据多维度评分从素材库中自动选择最优视频素材， +用于一键生成等需要自动选取素材的场景。 + +评分维度（加权求和，总分 0-1）： + - 质量分（quality_score）：权重 0.5 — 来自人工或AI的质量评分 + - 分辨率适配：权重 0.2 — 分辨率越接近 1080p 得分越高 + - 时长合理性：权重 0.2 — 3-30 秒区间最佳，过短/过长扣分 + - 码率质量：权重 0.1 — 用文件大小/时长估算，码率适中得分高 + +特性： + - 最低质量分门槛：自动过滤低质量素材 + - 时长多样性：保证选出的素材时长分布均匀（短/中/长各占一定比例） + - 兼容全部模式：素材库模式和项目模式都可用 +""" + +from __future__ import annotations + +import logging +import math +from dataclasses import dataclass + +logger = logging.getLogger(__name__) + +# ── 评分权重 ────────────────────────────────────────────────────────────────── +_WEIGHT_QUALITY = 0.5 +_WEIGHT_RESOLUTION = 0.2 +_WEIGHT_DURATION = 0.2 +_WEIGHT_BITRATE = 0.1 + +# ── 评分参数 ────────────────────────────────────────────────────────────────── +_TARGET_WIDTH = 1920 # 目标分辨率宽度基准 +_TARGET_HEIGHT = 1080 # 目标分辨率高度基准 +_MIN_QUALITY_SCORE = 30.0 # 最低质量分门槛（低于此值的素材直接排除） +_OPTIMAL_DURATION_MIN = 3.0 # 最佳时长区间（秒） +_OPTIMAL_DURATION_MAX = 30.0 + +# ── 多样性分桶 ─────────────────────────────────────────────────────────────── +_SHORT_BUCKET_MAX = 5.0 # 短素材：< 5s +_MEDIUM_BUCKET_MAX = 15.0 # 中素材：5-15s +# 长素材：> 15s + + +@dataclass +class SmartSelectResult: + """智能选择结果.""" + + selected_ids: list[str] + total_candidates: int + filtered_out: int # 被质量门槛过滤的数量 + avg_score: float + details: list[AssetScoreDetail] + + +@dataclass +class AssetScoreDetail: + """单个素材的评分详情.""" + + asset_id: str + total_score: float + quality_score: float + resolution_score: float + duration_score: float + bitrate_score: float + duration: float | None + + +class SmartAssetSelector: + """智能素材选择器. + + 从一组素材中按综合评分选择最优的 N 个， + 同时保证时长分布的多样性。 + """ + + def __init__( + self, + min_quality_score: float = _MIN_QUALITY_SCORE, + target_width: int = _TARGET_WIDTH, + target_height: int = _TARGET_HEIGHT, + ): + self.min_quality_score = min_quality_score + self.target_width = target_width + self.target_height = target_height + + # ── 公开方法 ────────────────────────────────────────────────────────────── + + def select( + self, + assets: list, + count: int = 0, + *, + ensure_diversity: bool = True, + ) -> SmartSelectResult: + """从素材列表中智能选择最优素材. + + Args: + assets: Asset 实体列表（需要有 id/quality_score/width/height/duration/file_size 属性） + count: 选取数量，0 表示全部符合条件的 + ensure_diversity: 是否保证时长多样性（默认开启） + + Returns: + SmartSelectResult 选择结果 + """ + # 1. 过滤：只保留 ready 状态的视频素材 + 最低质量分门槛 + candidates = [] + filtered_out = 0 + for asset in assets: + status = getattr(asset, "status", None) + status_val = status.value if hasattr(status, "value") else str(status) + if status_val != "ready": + continue + mime_type = getattr(asset, "mime_type", "") or "" + if not mime_type.startswith("video"): + continue + quality = getattr(asset, "quality_score", None) + if quality is not None and quality < self.min_quality_score: + filtered_out += 1 + continue + candidates.append(asset) + + if not candidates: + return SmartSelectResult( + selected_ids=[], + total_candidates=0, + filtered_out=filtered_out, + avg_score=0.0, + details=[], + ) + + # 2. 对每个候选素材评分 + scored: list[AssetScoreDetail] = [] + for asset in candidates: + detail = self._score_asset(asset) + scored.append(detail) + + # 3. 按总分降序排列 + scored.sort(key=lambda d: d.total_score, reverse=True) + + # 4. 多样性选择（如果需要且数量有限制） + if ensure_diversity and count > 0 and len(scored) > count: + selected = self._diverse_selection(scored, count) + else: + # 无数量限制或不要求多样性，直接按排名取 + selected = scored if count <= 0 else scored[:count] + + avg_score = sum(d.total_score for d in selected) / len(selected) if selected else 0.0 + + result = SmartSelectResult( + selected_ids=[d.asset_id for d in selected], + total_candidates=len(candidates), + filtered_out=filtered_out, + avg_score=avg_score, + details=selected, + ) + + logger.info( + "智能素材选择完成: 候选=%d, 过滤=%d, 选中=%d, 平均分=%.3f", + result.total_candidates, + result.filtered_out, + len(result.selected_ids), + result.avg_score, + ) + return result + + # ── 内部方法 ────────────────────────────────────────────────────────────── + + def _score_asset(self, asset) -> AssetScoreDetail: + """对单个素材进行多维度评分.""" + # 质量分 + quality = getattr(asset, "quality_score", None) + quality_score = (quality / 100.0) if quality is not None else 0.5 + + # 分辨率评分：越接近目标分辨率得分越高 + width = getattr(asset, "width", None) + height = getattr(asset, "height", None) + resolution_score = self._score_resolution(width, height) + + # 时长评分：在最佳区间内得分高，过短过长扣分 + duration = getattr(asset, "duration", None) + duration_score = self._score_duration(duration) + + # 码率评分：用 file_size/duration 估算，适中得分高 + file_size = getattr(asset, "file_size", 0) or 0 + bitrate_score = self._score_bitrate(file_size, duration) + + # 加权总分 + total = ( + _WEIGHT_QUALITY * quality_score + + _WEIGHT_RESOLUTION * resolution_score + + _WEIGHT_DURATION * duration_score + + _WEIGHT_BITRATE * bitrate_score + ) + + return AssetScoreDetail( + asset_id=asset.id, + total_score=round(total, 4), + quality_score=round(quality_score, 4), + resolution_score=round(resolution_score, 4), + duration_score=round(duration_score, 4), + bitrate_score=round(bitrate_score, 4), + duration=duration, + ) + + def _score_resolution(self, width: int | None, height: int | None) -> float: + """分辨率评分：越接近目标分辨率得分越高，低于480p扣分严重.""" + if width is None or height is None or width <= 0 or height <= 0: + return 0.5 # 未知分辨率给中评分 + + target_pixels = self.target_width * self.target_height + actual_pixels = width * height + + # 计算像素数比例 + ratio = actual_pixels / target_pixels + + if ratio >= 1.0: + # 高于或等于目标分辨率：满分，略高不扣分（4K也给满分） + return 1.0 + else: + # 低于目标分辨率：线性衰减，但最低不低于 0.1 + # 例如：720p (921600) / 1080p (2073600) = 0.44 → 得分 0.6 + score = 0.3 + 0.7 * ratio + return max(0.1, min(1.0, score)) + + def _score_duration(self, duration: float | None) -> float: + """时长评分：3-30秒最佳，过短或过长都扣分.""" + if duration is None or duration <= 0: + return 0.5 # 未知时长给中评分 + + if _OPTIMAL_DURATION_MIN <= duration <= _OPTIMAL_DURATION_MAX: + # 最佳区间：满分 + return 1.0 + + if duration < _OPTIMAL_DURATION_MIN: + # 太短：线性衰减，1秒以下给 0.3 + ratio = duration / _OPTIMAL_DURATION_MIN + return 0.3 + 0.7 * ratio + + # 太长：每超过最佳区间上限10秒扣 0.1 分，最低 0.2 + excess = duration - _OPTIMAL_DURATION_MAX + penalty = min(0.8, excess / 10.0 * 0.1) + return max(0.2, 1.0 - penalty) + + def _score_bitrate(self, file_size: int, duration: float | None) -> float: + """码率评分：根据文件大小和时长估算码率，适中得分高.""" + if not file_size or not duration or duration <= 0: + return 0.5 # 未知给中评分 + + # 估算码率（bps） + bitrate = (file_size * 8) / duration + + # 最佳码率范围：2-8 Mbps + optimal_low = 2_000_000 # 2 Mbps + optimal_high = 8_000_000 # 8 Mbps + + if optimal_low <= bitrate <= optimal_high: + return 1.0 + + if bitrate < optimal_low: + # 码率太低：线性衰减 + ratio = bitrate / optimal_low + return 0.3 + 0.7 * ratio + + # 码率太高（文件太大）：适度扣分 + excess = bitrate / optimal_high - 1.0 + penalty = min(0.5, excess * 0.2) + return max(0.5, 1.0 - penalty) + + def _diverse_selection(self, scored: list[AssetScoreDetail], count: int) -> list[AssetScoreDetail]: + """多样性选择：按时长分桶，保证每个桶都有素材. + + 策略： + 1. 按时长分为三桶：短(<5s)、中(5-15s)、长(>15s) + 2. 每个桶配额 = max(1, count / 3) + 3. 先从每桶按配额取最高分的 + 4. 剩余名额从全局最高分中取（不重复） + """ + # 分桶 + short_bucket = [d for d in scored if d.duration is not None and d.duration < _SHORT_BUCKET_MAX] + medium_bucket = [ + d + for d in scored + if d.duration is not None and _SHORT_BUCKET_MAX <= d.duration < _MEDIUM_BUCKET_MAX + ] + long_bucket = [d for d in scored if d.duration is not None and d.duration >= _MEDIUM_BUCKET_MAX] + unknown_bucket = [d for d in scored if d.duration is None] + + buckets = [short_bucket, medium_bucket, long_bucket] + bucket_names = ["short", "medium", "long"] + + # 每个桶基础配额（至少1个，如果桶非空且需要的话） + base_quota = max(1, count // 3) + + selected: list[AssetScoreDetail] = [] + selected_ids: set[str] = set() + + # 先按配额从每个桶取 + for bucket, name in zip(buckets, bucket_names): + quota = min(base_quota, len(bucket)) + if quota <= 0: + continue + # 桶内已经按分数排好序了，直接取前 quota 个 + for item in bucket[:quota]: + if item.asset_id not in selected_ids: + selected.append(item) + selected_ids.add(item.asset_id) + if len(selected) >= count: + return selected + + # 剩余名额：从全局（未被选中的）中按分数高低取 + remaining_needed = count - len(selected) + if remaining_needed > 0: + for item in scored: + if item.asset_id not in selected_ids: + selected.append(item) + selected_ids.add(item.asset_id) + if len(selected) >= count: + break + + # 如果还不够（不应该发生），加上未知时长的 + if len(selected) < count and unknown_bucket: + for item in unknown_bucket: + if item.asset_id not in selected_ids: + selected.append(item) + selected_ids.add(item.asset_id) + if len(selected) >= count: + break + + return selected[:count] diff --git a/tests/unit/test_smart_asset_selector.py b/tests/unit/test_smart_asset_selector.py new file mode 100755 index 000000000..ed90981e6 --- /dev/null +++ b/tests/unit/test_smart_asset_selector.py @@ -0,0 +1,413 @@ +"""SmartAssetSelector 智能素材选择服务单元测试.""" + +from __future__ import annotations + +import unittest +from dataclasses import dataclass + +from app.services.smart_asset_selector import ( + SmartAssetSelector, + _MEDIUM_BUCKET_MAX, + _SHORT_BUCKET_MAX, +) + + +@dataclass +class MockAsset: + """模拟 Asset 实体.""" + + id: str + quality_score: float | None = None + width: int | None = None + height: int | None = None + duration: float | None = None + file_size: int = 0 + mime_type: str = "video/mp4" + status: str = "ready" + + @property + def status_value(self) -> str: + return self.status + + +class TestSmartAssetSelectorScoring(unittest.TestCase): + """评分维度测试.""" + + def setUp(self): + self.selector = SmartAssetSelector() + + def test_quality_score_normalization(self): + """质量分正确归一化到 0-1.""" + asset_high = MockAsset(id="1", quality_score=90.0) + asset_low = MockAsset(id="2", quality_score=30.0) + asset_none = MockAsset(id="3", quality_score=None) + + detail_high = self.selector._score_asset(asset_high) + detail_low = self.selector._score_asset(asset_low) + detail_none = self.selector._score_asset(asset_none) + + # 90分 → 0.9 × 0.5权重 = 0.45 基础贡献 + self.assertAlmostEqual(detail_high.quality_score, 0.9, delta=0.01) + # 30分 → 0.3 × 0.5权重 = 0.15 基础贡献 + self.assertAlmostEqual(detail_low.quality_score, 0.3, delta=0.01) + # 无质量分给默认 0.5 + self.assertAlmostEqual(detail_none.quality_score, 0.5, delta=0.01) + + def test_resolution_score_1080p_full(self): + """1080p 分辨率得满分.""" + asset = MockAsset(id="1", width=1920, height=1080) + detail = self.selector._score_asset(asset) + self.assertAlmostEqual(detail.resolution_score, 1.0, delta=0.01) + + def test_resolution_score_4k_full(self): + """4K 也得满分（高于目标分辨率不扣分）.""" + asset = MockAsset(id="1", width=3840, height=2160) + detail = self.selector._score_asset(asset) + self.assertAlmostEqual(detail.resolution_score, 1.0, delta=0.01) + + def test_resolution_score_720p_lower(self): + """720p 低于 1080p，得分低于 1.""" + asset = MockAsset(id="1", width=1280, height=720) + detail = self.selector._score_asset(asset) + self.assertLess(detail.resolution_score, 1.0) + self.assertGreater(detail.resolution_score, 0.3) + + def test_resolution_score_none(self): + """分辨率未知给中评分.""" + asset = MockAsset(id="1", width=None, height=None) + detail = self.selector._score_asset(asset) + self.assertAlmostEqual(detail.resolution_score, 0.5, delta=0.01) + + def test_duration_score_optimal(self): + """最佳时长区间内得满分.""" + asset = MockAsset(id="1", duration=10.0) + detail = self.selector._score_asset(asset) + self.assertAlmostEqual(detail.duration_score, 1.0, delta=0.01) + + def test_duration_score_too_short(self): + """时长过短扣分.""" + asset = MockAsset(id="1", duration=1.0) + detail = self.selector._score_asset(asset) + self.assertLess(detail.duration_score, 1.0) + + def test_duration_score_too_long(self): + """时长过长扣分.""" + asset = MockAsset(id="1", duration=120.0) + detail = self.selector._score_asset(asset) + self.assertLess(detail.duration_score, 1.0) + + def test_duration_score_none(self): + """时长未知给中评分.""" + asset = MockAsset(id="1", duration=None) + detail = self.selector._score_asset(asset) + self.assertAlmostEqual(detail.duration_score, 0.5, delta=0.01) + + def test_total_score_weighted_sum(self): + """总分是各维度的加权和.""" + asset = MockAsset( + id="1", + quality_score=100.0, # 1.0 × 0.5 = 0.5 + width=1920, # 1.0 × 0.2 = 0.2 + height=1080, + duration=10.0, # 1.0 × 0.2 = 0.2 + file_size=10_000_000, # ~8Mbps，10秒 → 约 1.0 × 0.1 = 0.1 + ) + detail = self.selector._score_asset(asset) + # 理论上接近 1.0 + self.assertGreater(detail.total_score, 0.85) + self.assertLessEqual(detail.total_score, 1.0) + + +class TestSmartAssetSelectorSelection(unittest.TestCase): + """选择逻辑测试.""" + + def setUp(self): + self.selector = SmartAssetSelector(min_quality_score=0) # 测试时关闭质量门槛 + + def _make_assets(self, count: int, base_quality: float = 80.0) -> list[MockAsset]: + assets = [] + for i in range(count): + assets.append( + MockAsset( + id=f"asset_{i}", + quality_score=base_quality - i * 5, # 质量递减 + width=1920, + height=1080, + duration=10.0 + i, + file_size=5_000_000 + i * 100_000, + ) + ) + return assets + + def test_select_all_when_count_zero(self): + """count=0 时返回全部符合条件的.""" + assets = self._make_assets(10) + result = self.selector.select(assets, count=0) + self.assertEqual(len(result.selected_ids), 10) + self.assertEqual(result.total_candidates, 10) + + def test_select_top_n(self): + """返回指定数量的 top N.""" + assets = self._make_assets(10) + result = self.selector.select(assets, count=3) + self.assertEqual(len(result.selected_ids), 3) + # 最高分的应该是 asset_0（质量分最高） + self.assertEqual(result.selected_ids[0], "asset_0") + + def test_select_more_than_available(self): + """请求数量超过候选数量时返回全部.""" + assets = self._make_assets(5) + result = self.selector.select(assets, count=10) + self.assertEqual(len(result.selected_ids), 5) + + def test_filter_non_ready(self): + """非 ready 状态的素材被过滤.""" + assets = [ + MockAsset(id="1", quality_score=90.0, status="ready"), + MockAsset(id="2", quality_score=80.0, status="processing"), + MockAsset(id="3", quality_score=70.0, status="ready"), + ] + result = self.selector.select(assets, count=0) + self.assertEqual(len(result.selected_ids), 2) + self.assertIn("1", result.selected_ids) + self.assertIn("3", result.selected_ids) + self.assertNotIn("2", result.selected_ids) + + def test_filter_non_video(self): + """非视频素材被过滤.""" + assets = [ + MockAsset(id="1", quality_score=90.0, mime_type="video/mp4"), + MockAsset(id="2", quality_score=80.0, mime_type="image/jpeg"), + MockAsset(id="3", quality_score=70.0, mime_type="video/quicktime"), + ] + result = self.selector.select(assets, count=0) + self.assertEqual(len(result.selected_ids), 2) + + def test_min_quality_filter(self): + """最低质量分门槛过滤.""" + selector = SmartAssetSelector(min_quality_score=60.0) + assets = [ + MockAsset(id="1", quality_score=90.0), + MockAsset(id="2", quality_score=50.0), # 低于门槛 + MockAsset(id="3", quality_score=70.0), + MockAsset(id="4", quality_score=30.0), # 低于门槛 + ] + result = selector.select(assets, count=0) + self.assertEqual(len(result.selected_ids), 2) + self.assertEqual(result.filtered_out, 2) + self.assertIn("1", result.selected_ids) + self.assertIn("3", result.selected_ids) + + def test_empty_input(self): + """空输入返回空结果.""" + result = self.selector.select([], count=5) + self.assertEqual(result.selected_ids, []) + self.assertEqual(result.total_candidates, 0) + self.assertEqual(result.avg_score, 0.0) + + def test_sorted_by_score_descending(self): + """结果按总分降序排列.""" + assets = self._make_assets(5) + result = self.selector.select(assets, count=0, ensure_diversity=False) + scores = [d.total_score for d in result.details] + # 应该是降序 + self.assertEqual(scores, sorted(scores, reverse=True)) + + +class TestSmartAssetSelectorDiversity(unittest.TestCase): + """多样性选择测试.""" + + def setUp(self): + self.selector = SmartAssetSelector(min_quality_score=0) + + def _make_assets(self, count: int, base_quality: float = 80.0) -> list[MockAsset]: + assets = [] + for i in range(count): + assets.append( + MockAsset( + id=f"asset_{i}", + quality_score=base_quality - i * 5, + width=1920, + height=1080, + duration=10.0 + i, + file_size=5_000_000 + i * 100_000, + ) + ) + return assets + + def test_diversity_all_short(self): + """全是短素材时不报错，正常返回.""" + assets = [] + for i in range(10): + assets.append( + MockAsset( + id=f"short_{i}", + quality_score=80.0 + i, + width=1920, + height=1080, + duration=2.0 + i * 0.1, # 都 < 5s + file_size=1_000_000, + ) + ) + result = self.selector.select(assets, count=5, ensure_diversity=True) + self.assertEqual(len(result.selected_ids), 5) + + def test_diversity_mixed_buckets(self): + """混合时长素材时，各桶都有代表.""" + assets = [] + # 短素材（质量分高） + for i in range(5): + assets.append( + MockAsset( + id=f"short_{i}", + quality_score=95.0 - i, + width=1920, + height=1080, + duration=3.0, + file_size=2_000_000, + ) + ) + # 中素材（质量分中等） + for i in range(5): + assets.append( + MockAsset( + id=f"medium_{i}", + quality_score=85.0 - i, + width=1920, + height=1080, + duration=10.0, + file_size=5_000_000, + ) + ) + # 长素材（质量分低） + for i in range(5): + assets.append( + MockAsset( + id=f"long_{i}", + quality_score=75.0 - i, + width=1920, + height=1080, + duration=60.0, + file_size=20_000_000, + ) + ) + + result = self.selector.select(assets, count=6, ensure_diversity=True) + selected = result.selected_ids + + # 6个素材，每个桶至少有1个（基础配额 max(1, 6//3)=2） + short_count = sum(1 for sid in selected if sid.startswith("short_")) + medium_count = sum(1 for sid in selected if sid.startswith("medium_")) + long_count = sum(1 for sid in selected if sid.startswith("long_")) + + # 每个桶至少1个 + self.assertGreaterEqual(short_count, 1) + self.assertGreaterEqual(medium_count, 1) + self.assertGreaterEqual(long_count, 1) + self.assertEqual(len(selected), 6) + + def test_diversity_disabled_returns_top(self): + """关闭多样性时，直接返回 top N（可能全是短素材）.""" + assets = [] + # 短素材（质量分最高） + for i in range(10): + assets.append( + MockAsset( + id=f"short_{i}", + quality_score=95.0 - i, + width=1920, + height=1080, + duration=3.0, + file_size=2_000_000, + ) + ) + # 长素材（质量分低） + for i in range(5): + assets.append( + MockAsset( + id=f"long_{i}", + quality_score=70.0, + width=1920, + height=1080, + duration=60.0, + file_size=20_000_000, + ) + ) + + result = self.selector.select(assets, count=5, ensure_diversity=False) + selected = result.selected_ids + # 全是短素材（因为质量分高） + self.assertTrue(all(s.startswith("short_") for s in selected)) + + def test_avg_score_calculated(self): + """平均分正确计算.""" + assets = self._make_assets(3) + result = self.selector.select(assets, count=3, ensure_diversity=False) + expected_avg = sum(d.total_score for d in result.details) / 3 + self.assertAlmostEqual(result.avg_score, expected_avg, delta=0.001) + + +class TestSmartAssetSelectorEdgeCases(unittest.TestCase): + """边界情况测试.""" + + def setUp(self): + self.selector = SmartAssetSelector(min_quality_score=0) + + def _make_assets(self, count: int, base_quality: float = 80.0) -> list[MockAsset]: + assets = [] + for i in range(count): + assets.append( + MockAsset( + id=f"asset_{i}", + quality_score=base_quality - i * 5, + width=1920, + height=1080, + duration=10.0 + i, + file_size=5_000_000 + i * 100_000, + ) + ) + return assets + + def test_single_asset(self): + """单个素材正常返回.""" + assets = [MockAsset(id="1", quality_score=80.0, width=1920, height=1080, duration=10.0)] + result = self.selector.select(assets, count=1) + self.assertEqual(len(result.selected_ids), 1) + self.assertEqual(result.selected_ids[0], "1") + + def test_zero_width_height(self): + """宽高为0时按未知处理.""" + asset = MockAsset(id="1", width=0, height=0) + detail = self.selector._score_asset(asset) + self.assertAlmostEqual(detail.resolution_score, 0.5, delta=0.01) + + def test_negative_duration(self): + """负时长按未知处理.""" + asset = MockAsset(id="1", duration=-5.0) + detail = self.selector._score_asset(asset) + self.assertAlmostEqual(detail.duration_score, 0.5, delta=0.01) + + def test_zero_file_size_with_duration(self): + """文件大小为0时码率评分中等.""" + asset = MockAsset(id="1", file_size=0, duration=10.0) + detail = self.selector._score_asset(asset) + self.assertAlmostEqual(detail.bitrate_score, 0.5, delta=0.01) + + def test_bitrate_score_optimal(self): + """最佳码率范围得满分.""" + # 5 Mbps × 10秒 = 6.25 MB → file_size = 6,250,000 bytes + asset = MockAsset(id="1", file_size=6_250_000, duration=10.0) + detail = self.selector._score_asset(asset) + self.assertAlmostEqual(detail.bitrate_score, 1.0, delta=0.01) + + def test_details_match_selected_ids(self): + """details 列表和 selected_ids 顺序一致.""" + assets = self._make_assets(5) + result = self.selector.select(assets, count=3, ensure_diversity=False) + self.assertEqual(len(result.details), 3) + for i, aid in enumerate(result.selected_ids): + self.assertEqual(result.details[i].asset_id, aid) + + +if __name__ == "__main__": + unittest.main() -- 2.54.0