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

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]

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:

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 中解析水印配置，兼容两种存储格式.

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
 

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