feat(phase8): AutoClipService 智能选片服务 (任务 2.07)

- 新增 AutoClipService：自动为剪辑计划片段分配最佳素材
- 评分算法：质量分(0.5) + 时长匹配(0.3) + 分类匹配(0.2)
- 支持按 file_type/quality_score/duration/classification/tags 筛选
- AssetRepository 新增 search_candidates() 方法（抽象+SQLAlchemy实现）
- 18 个单元测试全部通过

apps/api/app/services/__init__.py

@@ -1,9 +1,11 @@
 """Service layer exports for Phase 8 模板编排引擎."""
 
+from .auto_clip_service import AutoClipService
 from .edit_plan_service import EditPlanService
 from .edit_template_service import EditTemplateService
 
 __all__ = [
+    "AutoClipService",
     "EditPlanService",
     "EditTemplateService",
 ]

apps/api/app/services/auto_clip_service.py

@@ -0,0 +1,340 @@
+"""AutoClipService — 智能选片服务.
+
+根据模板片段配置 (TemplateClipConfig) 的素材需求 (material_requirements)，
+自动从项目素材库中筛选、评分并分配最佳素材到剪辑计划片段 (EditPlanClip)。
+
+评分规则：
+- 质量分 (quality_score)：权重 0.5
+- 时长匹配度：权重 0.3（越接近目标时长得分越高）
+- 分类匹配度：权重 0.2（分类完全匹配得满分，部分匹配按比例得分）
+"""
+
+from __future__ import annotations
+
+import logging
+from dataclasses import dataclass
+
+from sqlalchemy.orm import Session
+
+from packages.adapters.sqlalchemy_impl import (
+    SQLAlchemyAssetRepository,
+    SQLAlchemyEditPlanClipRepository,
+    SQLAlchemyEditPlanRepository,
+    SQLAlchemyTemplateClipConfigRepository,
+)
+from packages.domain.asset import AssetType
+from packages.domain.classification import AssetClassification
+from packages.domain.edit_plan_clip import EditPlanClip, EditPlanClipStatus
+
+logger = logging.getLogger(__name__)
+
+# ── 评分权重 ──────────────────────────────────────────────────────────────────
+_WEIGHT_QUALITY = 0.5
+_WEIGHT_DURATION = 0.3
+_WEIGHT_CLASSIFICATION = 0.2
+
+
+@dataclass
+class AutoSelectResult:
+    """智能选片结果。"""
+
+    plan_id: str
+    total_clips: int
+    assigned_clips: int
+    unassigned_clips: int
+    details: list[ClipAssignDetail]
+
+
+@dataclass
+class ClipAssignDetail:
+    """单个片段的分配详情。"""
+
+    clip_id: str
+    clip_type: str
+    assigned_asset_id: str | None
+    candidate_count: int
+    score: float | None
+    reason: str
+
+
+class AutoClipService:
+    """智能选片服务 — 自动为剪辑计划片段分配最佳素材。"""
+
+    def __init__(self, db: Session) -> None:
+        self._plan_repo = SQLAlchemyEditPlanRepository(db)
+        self._clip_repo = SQLAlchemyEditPlanClipRepository(db)
+        self._config_repo = SQLAlchemyTemplateClipConfigRepository(db)
+        self._asset_repo = SQLAlchemyAssetRepository(db)
+
+    # ── 公开方法 ──────────────────────────────────────────────────────────────
+
+    def auto_select_assets(self, plan_id: str, project_id: str) -> AutoSelectResult:
+        """为剪辑计划的所有片段自动分配素材。
+
+        流程：
+        1. 获取剪辑计划 → 读取 template_id
+        2. 获取模板的所有片段配置 (TemplateClipConfig)
+        3. 获取计划的所有片段 (EditPlanClip)
+        4. 对每个片段，根据其关联的 config 筛选候选素材并评分
+        5. 将最佳素材分配给片段，标记为 READY
+
+        Args:
+            plan_id: 剪辑计划 ID
+            project_id: 项目 ID（素材所属项目）
+
+        Returns:
+            AutoSelectResult 包含分配统计和每个片段的详情
+
+        Raises:
+            ValueError: 计划不存在
+        """
+        plan = self._plan_repo.get(plan_id)
+        if plan is None:
+            raise ValueError(f"剪辑计划不存在: {plan_id}")
+
+        # 获取模板片段配置（按 order 排序）
+        configs = self._config_repo.list_by_template(plan.template_id)
+        config_map = {c.id: c for c in configs}
+
+        # 获取计划的所有片段
+        clips = self._clip_repo.list_by_plan(plan_id)
+
+        details: list[ClipAssignDetail] = []
+        assigned_count = 0
+
+        for clip in clips:
+            detail = self._assign_single_clip(clip, project_id, config_map)
+            details.append(detail)
+            if detail.assigned_asset_id is not None:
+                assigned_count += 1
+
+        result = AutoSelectResult(
+            plan_id=plan_id,
+            total_clips=len(clips),
+            assigned_clips=assigned_count,
+            unassigned_clips=len(clips) - assigned_count,
+            details=details,
+        )
+        logger.info(
+            "智能选片完成: plan=%s total=%d assigned=%d unassigned=%d",
+            plan_id,
+            result.total_clips,
+            result.assigned_clips,
+            result.unassigned_clips,
+        )
+        return result
+
+    def select_for_clip(self, clip_id: str, project_id: str) -> ClipAssignDetail:
+        """为单个片段选择并分配最佳素材。
+
+        Args:
+            clip_id: 片段 ID
+            project_id: 项目 ID（素材所属项目）
+
+        Returns:
+            ClipAssignDetail 分配详情
+
+        Raises:
+            ValueError: 片段不存在或缺少关联配置
+        """
+        clip = self._clip_repo.get(clip_id)
+        if clip is None:
+            raise ValueError(f"片段不存在: {clip_id}")
+
+        # 获取关联的模板配置
+        config = None
+        if clip.template_clip_config_id:
+            config = self._config_repo.get(clip.template_clip_config_id)
+
+        config_map = {config.id: config} if config else {}
+        return self._assign_single_clip(clip, project_id, config_map)
+
+    # ── 内部方法 ──────────────────────────────────────────────────────────────
+
+    def _assign_single_clip(
+        self,
+        clip: EditPlanClip,
+        project_id: str,
+        config_map: dict[str, object],
+    ) -> ClipAssignDetail:
+        """为单个片段分配素材。"""
+        config = config_map.get(clip.template_clip_config_id) if clip.template_clip_config_id else None
+
+        # 解析素材需求
+        requirements = self._parse_material_requirements(config)
+
+        # 搜索候选素材
+        candidates = self._asset_repo.search_candidates(
+            project_id=project_id,
+            file_type=requirements.get("file_type"),
+            min_quality_score=requirements.get("min_quality_score"),
+            min_duration=requirements.get("min_duration"),
+            max_duration=requirements.get("max_duration"),
+            classification_category=requirements.get("classification_category"),
+            tags=requirements.get("tags"),
+            status="completed",
+            limit=50,
+        )
+
+        if not candidates:
+            return ClipAssignDetail(
+                clip_id=clip.id,
+                clip_type=requirements.get("clip_type", "unknown"),
+                assigned_asset_id=None,
+                candidate_count=0,
+                score=None,
+                reason="无符合条件的候选素材",
+            )
+
+        # 评分并选择最佳素材
+        target_duration = requirements.get("target_duration")
+        target_category = requirements.get("classification_category")
+
+        best_asset = None
+        best_score = -1.0
+        for asset in candidates:
+            score = self._score_candidate(
+                asset,
+                target_duration=target_duration,
+                target_category=target_category,
+            )
+            if score > best_score:
+                best_score = score
+                best_asset = asset
+
+        if best_asset is None:
+            return ClipAssignDetail(
+                clip_id=clip.id,
+                clip_type=requirements.get("clip_type", "unknown"),
+                assigned_asset_id=None,
+                candidate_count=len(candidates),
+                score=None,
+                reason="候选素材评分均不合格",
+            )
+
+        # 分配素材并标记就绪
+        clip.assign_asset(best_asset.id)
+        clip.mark_ready()
+        self._clip_repo.update(clip)
+
+        return ClipAssignDetail(
+            clip_id=clip.id,
+            clip_type=requirements.get("clip_type", "unknown"),
+            assigned_asset_id=best_asset.id,
+            candidate_count=len(candidates),
+            score=round(best_score, 4),
+            reason=f"最佳匹配 (score={best_score:.4f})",
+        )
+
+    @staticmethod
+    def _score_candidate(
+        asset: object,
+        *,
+        target_duration: float | None = None,
+        target_category: str | None = None,
+    ) -> float:
+        """对候选素材评分 (0.0 ~ 1.0)。
+
+        评分维度：
+        - 质量分 (quality_score)：归一化到 0-1，权重 0.5
+        - 时长匹配度：越接近目标时长得分越高，权重 0.3
+        - 分类匹配度：完全匹配得 1.0，无分类得 0.0，权重 0.2
+        """
+        # 质量分 (0-100 → 0-1)
+        quality = getattr(asset, "quality_score", None)
+        quality_score = (quality / 100.0) if quality is not None else 0.5
+
+        # 时长匹配度
+        duration_score = 0.5  # 无目标时长的默认分
+        if target_duration is not None and target_duration > 0:
+            asset_duration = getattr(asset, "duration", None)
+            if asset_duration is not None and asset_duration > 0:
+                ratio = asset_duration / target_duration
+                # 比率越接近 1.0 得分越高，使用高斯衰减
+                duration_score = max(0.0, 1.0 - abs(1.0 - ratio) * 2)
+            # 无时长的素材得 0 分
+            else:
+                duration_score = 0.0
+
+        # 分类匹配度
+        classification_score = 0.0
+        if target_category is not None:
+            metadata = getattr(asset, "metadata", {}) or {}
+            asset_category = metadata.get("category", "")
+            if asset_category == target_category:
+                classification_score = 1.0
+            elif asset_category:
+                # 部分匹配（同大类）给 0.5
+                classification_score = 0.3
+        else:
+            # 无分类要求，所有素材得满分
+            classification_score = 1.0
+
+        total = (
+            _WEIGHT_QUALITY * quality_score
+            + _WEIGHT_DURATION * duration_score
+            + _WEIGHT_CLASSIFICATION * classification_score
+        )
+        return total
+
+    @staticmethod
+    def _parse_material_requirements(config: object | None) -> dict:
+        """从 TemplateClipConfig 解析素材筛选条件。
+
+        将 material_requirements JSON 和 config 自身的时长/类型字段
+        统一转换为 search_candidates 可用的筛选参数。
+        """
+        result: dict = {}
+        if config is None:
+            return result
+
+        # 从 material_requirements 提取筛选条件
+        requirements = getattr(config, "material_requirements", {}) or {}
+        # 素材类型: material_requirements 中的 "type" 字段
+        req_type = requirements.get("type")
+        if req_type and req_type in (AssetType.VIDEO, AssetType.IMAGE, AssetType.AUDIO):
+            result["file_type"] = req_type
+
+        # 最低质量分
+        min_quality = requirements.get("min_quality_score") or requirements.get("min_quality")
+        if min_quality is not None:
+            try:
+                result["min_quality_score"] = float(min_quality)
+            except (TypeError, ValueError):
+                pass
+
+        # 分类筛选
+        category = requirements.get("category") or requirements.get("classification")
+        if category:
+            # 验证是否为有效分类
+            valid_categories = {c.value for c in AssetClassification}
+            if category in valid_categories:
+                result["classification_category"] = category
+
+        # 标签筛选
+        tags = requirements.get("tags")
+        if isinstance(tags, list) and tags:
+            result["tags"] = tags
+
+        # 时长范围：优先使用 config 的 min/max_duration，其次 material_requirements
+        min_dur = getattr(config, "min_duration", None) or requirements.get("min_duration")
+        max_dur = getattr(config, "max_duration", None) or requirements.get("max_duration")
+        if min_dur is not None and min_dur > 0:
+            result["min_duration"] = float(min_dur)
+        if max_dur is not None and max_dur > 0:
+            result["max_duration"] = float(max_dur)
+
+        # 目标时长（用于评分）
+        if min_dur and max_dur:
+            result["target_duration"] = (float(min_dur) + float(max_dur)) / 2
+        elif min_dur:
+            result["target_duration"] = float(min_dur) * 1.2
+        elif max_dur:
+            result["target_duration"] = float(max_dur) * 0.8
+
+        # 片段类型（用于日志）
+        clip_type = getattr(config, "clip_type", None)
+        if clip_type:
+            result["clip_type"] = clip_type.value if hasattr(clip_type, "value") else str(clip_type)
+
+        return result

packages/adapters/sqlalchemy_impl/asset_repository.py

@@ -124,6 +124,54 @@ class SQLAlchemyAssetRepository:
         )
         return int(result or 0)
 
+    def search_candidates(
+        self,
+        project_id: str,
+        *,
+        file_type: str | None = None,
+        min_quality_score: float | None = None,
+        min_duration: float | None = None,
+        max_duration: float | None = None,
+        classification_category: str | None = None,
+        tags: list[str] | None = None,
+        status: str | None = None,
+        limit: int = 50,
+    ) -> list[Asset]:
+        """按筛选条件搜索候选素材，按质量分降序排列。"""
+        query = self.session.query(AssetModel).filter(
+            AssetModel.project_id == project_id,
+        )
+        if file_type is not None:
+            query = query.filter(AssetModel.file_type == file_type)
+        if min_quality_score is not None:
+            query = query.filter(AssetModel.quality_score >= min_quality_score)
+        if min_duration is not None:
+            query = query.filter(AssetModel.duration >= min_duration)
+        if max_duration is not None:
+            query = query.filter(AssetModel.duration <= max_duration)
+        if status is not None:
+            query = query.filter(AssetModel.status == status)
+        if classification_category is not None:
+            # classification_result 是 JSON Text，用 LIKE 匹配 category 字段
+            query = query.filter(
+                AssetModel.classification_result.like(
+                    f'%"{classification_category}"%'
+                )
+            )
+        query = query.order_by(AssetModel.quality_score.desc().nullslast())
+        if limit > 0:
+            query = query.limit(limit)
+        models = query.all()
+        candidates = [self._to_domain(m) for m in models]
+        # 内存中过滤 tags（tags 存在 metadata 中）
+        if tags:
+            tag_set = set(tags)
+            candidates = [
+                a for a in candidates
+                if tag_set.issubset(set(a.metadata.get("tags", [])))
+            ]
+        return candidates
+
     def _to_domain(self, model: AssetModel) -> Asset:
         metadata = {}
         if model.classification_result:

packages/ports/asset_repository.py

@@ -51,3 +51,20 @@ class AssetRepository(ABC):
     @abstractmethod
     def sum_storage_by_project_ids(self, project_ids: list[str]) -> int:
         pass
+
+    @abstractmethod
+    def search_candidates(
+        self,
+        project_id: str,
+        *,
+        file_type: str | None = None,
+        min_quality_score: float | None = None,
+        min_duration: float | None = None,
+        max_duration: float | None = None,
+        classification_category: str | None = None,
+        tags: list[str] | None = None,
+        status: str | None = None,
+        limit: int = 50,
+    ) -> list[Asset]:
+        """按筛选条件搜索候选素材，按质量分降序排列。"""
+        pass

tests/unit/test_auto_clip_service.py

@@ -0,0 +1,347 @@
+"""AutoClipService 单元测试."""
+
+from __future__ import annotations
+
+import sys
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parents[2] / "apps" / "api"))
+
+from dataclasses import dataclass, field
+from enum import Enum
+from typing import Any
+from unittest.mock import MagicMock
+
+import pytest
+
+from app.services.auto_clip_service import AutoClipService, ClipAssignDetail
+
+
+# ── Stub 实体 ─────────────────────────────────────────────────────────────────
+
+
+class _ClipStatus(str, Enum):
+    PENDING = "pending"
+    READY = "ready"
+
+
+class _ClipType(str, Enum):
+    INTRO = "intro"
+    MAIN = "main"
+    TRANSITION = "transition"
+    OUTRO = "outro"
+
+
+@dataclass
+class _StubClip:
+    id: str
+    plan_id: str
+    template_clip_config_id: str | None = None
+    clip_type: _ClipType = _ClipType.MAIN
+    order: int = 0
+    asset_id: str | None = None
+    status: _ClipStatus = _ClipStatus.PENDING
+
+    @property
+    def has_asset(self) -> bool:
+        return self.asset_id is not None
+
+    def assign_asset(self, asset_id: str) -> None:
+        self.asset_id = asset_id
+
+    def mark_ready(self) -> None:
+        if self.status == _ClipStatus.PENDING:
+            self.status = _ClipStatus.READY
+
+
+@dataclass
+class _StubAsset:
+    id: str
+    quality_score: float | None = 80.0
+    duration: float | None = 10.0
+    metadata: dict[str, Any] = field(default_factory=dict)
+
+
+@dataclass
+class _StubConfig:
+    id: str
+    template_id: str
+    clip_type: _ClipType = _ClipType.MAIN
+    order: int = 0
+    min_duration: float | None = None
+    max_duration: float | None = None
+    material_requirements: dict[str, Any] = field(default_factory=dict)
+
+
+@dataclass
+class _StubPlan:
+    id: str
+    template_id: str = "tpl-001"
+
+
+# ── Stub 仓储 ─────────────────────────────────────────────────────────────────
+
+
+class _StubPlanRepo:
+    def __init__(self, plan: _StubPlan | None = None) -> None:
+        self._plan = plan
+
+    def get(self, plan_id: str) -> _StubPlan | None:
+        return self._plan if self._plan and self._plan.id == plan_id else None
+
+
+class _StubClipRepo:
+    def __init__(self, clips: list[_StubClip] | None = None) -> None:
+        self._clips = clips or []
+        self.updated: list[_StubClip] = []
+
+    def get(self, clip_id: str) -> _StubClip | None:
+        for c in self._clips:
+            if c.id == clip_id:
+                return c
+        return None
+
+    def list_by_plan(self, plan_id: str) -> list[_StubClip]:
+        return [c for c in self._clips if c.plan_id == plan_id]
+
+    def update(self, clip: _StubClip) -> _StubClip:
+        self.updated.append(clip)
+        return clip
+
+
+class _StubConfigRepo:
+    def __init__(self, configs: list[_StubConfig] | None = None) -> None:
+        self._configs = configs or []
+
+    def get(self, config_id: str) -> _StubConfig | None:
+        for c in self._configs:
+            if c.id == config_id:
+                return c
+        return None
+
+    def list_by_template(self, template_id: str, **_: Any) -> list[_StubConfig]:
+        return [c for c in self._configs if c.template_id == template_id]
+
+
+class _StubAssetRepo:
+    def __init__(self, candidates: list[_StubAsset] | None = None) -> None:
+        self._candidates = candidates or []
+        self.last_query: dict[str, Any] = {}
+
+    def search_candidates(self, project_id: str, **kwargs: Any) -> list[_StubAsset]:
+        self.last_query = {"project_id": project_id, **kwargs}
+        return list(self._candidates)
+
+
+# ── 构造 Service (注入 stub) ──────────────────────────────────────────────────
+
+
+def _make_service(
+    plan_repo: _StubPlanRepo,
+    clip_repo: _StubClipRepo,
+    config_repo: _StubConfigRepo,
+    asset_repo: _StubAssetRepo,
+) -> AutoClipService:
+    """创建注入 stub 仓储的 AutoClipService（绕过 __init__）。"""
+    svc = AutoClipService.__new__(AutoClipService)
+    svc._plan_repo = plan_repo  # type: ignore[assignment]
+    svc._clip_repo = clip_repo  # type: ignore[assignment]
+    svc._config_repo = config_repo  # type: ignore[assignment]
+    svc._asset_repo = asset_repo  # type: ignore[assignment]
+    return svc
+
+
+# ── 测试：auto_select_assets ──────────────────────────────────────────────────
+
+
+class TestAutoSelectAssets:
+    def test_plan_not_found_raises(self) -> None:
+        svc = _make_service(
+            _StubPlanRepo(None),
+            _StubClipRepo(),
+            _StubConfigRepo(),
+            _StubAssetRepo(),
+        )
+        with pytest.raises(ValueError, match="剪辑计划不存在"):
+            svc.auto_select_assets("bad-id", "proj-1")
+
+    def test_no_clips_returns_empty(self) -> None:
+        plan = _StubPlan(id="plan-1", template_id="tpl-1")
+        svc = _make_service(
+            _StubPlanRepo(plan),
+            _StubClipRepo([]),
+            _StubConfigRepo(),
+            _StubAssetRepo(),
+        )
+        result = svc.auto_select_assets("plan-1", "proj-1")
+        assert result.total_clips == 0
+        assert result.assigned_clips == 0
+        assert result.unassigned_clips == 0
+
+    def test_assigns_best_candidate(self) -> None:
+        plan = _StubPlan(id="plan-1", template_id="tpl-1")
+        config = _StubConfig(
+            id="cfg-1",
+            template_id="tpl-1",
+            clip_type=_ClipType.MAIN,
+            min_duration=8.0,
+            max_duration=12.0,
+            material_requirements={"type": "video", "category": "scenic"},
+        )
+        clip = _StubClip(id="clip-1", plan_id="plan-1", template_clip_config_id="cfg-1")
+        # 两个候选，第一个质量分更高
+        assets = [
+            _StubAsset(id="a1", quality_score=90.0, duration=10.0, metadata={"category": "scenic"}),
+            _StubAsset(id="a2", quality_score=60.0, duration=10.0, metadata={"category": "scenic"}),
+        ]
+        svc = _make_service(
+            _StubPlanRepo(plan),
+            _StubClipRepo([clip]),
+            _StubConfigRepo([config]),
+            _StubAssetRepo(assets),
+        )
+        result = svc.auto_select_assets("plan-1", "proj-1")
+        assert result.assigned_clips == 1
+        assert result.details[0].assigned_asset_id == "a1"
+        assert clip.asset_id == "a1"
+        assert clip.status == _ClipStatus.READY
+
+    def test_no_candidates_marks_unassigned(self) -> None:
+        plan = _StubPlan(id="plan-1", template_id="tpl-1")
+        config = _StubConfig(id="cfg-1", template_id="tpl-1")
+        clip = _StubClip(id="clip-1", plan_id="plan-1", template_clip_config_id="cfg-1")
+        svc = _make_service(
+            _StubPlanRepo(plan),
+            _StubClipRepo([clip]),
+            _StubConfigRepo([config]),
+            _StubAssetRepo([]),  # 无候选
+        )
+        result = svc.auto_select_assets("plan-1", "proj-1")
+        assert result.assigned_clips == 0
+        assert result.unassigned_clips == 1
+        assert result.details[0].assigned_asset_id is None
+        assert "无符合条件" in result.details[0].reason
+
+
+# ── 测试：select_for_clip ─────────────────────────────────────────────────────
+
+
+class TestSelectForClip:
+    def test_clip_not_found_raises(self) -> None:
+        svc = _make_service(
+            _StubPlanRepo(None),
+            _StubClipRepo(),
+            _StubConfigRepo(),
+            _StubAssetRepo(),
+        )
+        with pytest.raises(ValueError, match="片段不存在"):
+            svc.select_for_clip("bad-id", "proj-1")
+
+    def test_assigns_without_config(self) -> None:
+        """片段没有关联 config 时仍可分配（无筛选条件）。"""
+        clip = _StubClip(id="clip-1", plan_id="plan-1", template_clip_config_id=None)
+        assets = [_StubAsset(id="a1", quality_score=70.0, duration=5.0)]
+        svc = _make_service(
+            _StubPlanRepo(None),
+            _StubClipRepo([clip]),
+            _StubConfigRepo(),
+            _StubAssetRepo(assets),
+        )
+        detail = svc.select_for_clip("clip-1", "proj-1")
+        assert detail.assigned_asset_id == "a1"
+
+
+# ── 测试：评分逻辑 ────────────────────────────────────────────────────────────
+
+
+class TestScoring:
+    def test_high_quality_wins(self) -> None:
+        a = _StubAsset(id="a", quality_score=95.0, duration=10.0, metadata={"category": "scenic"})
+        b = _StubAsset(id="b", quality_score=50.0, duration=10.0, metadata={"category": "scenic"})
+        sa = AutoClipService._score_candidate(a, target_duration=10.0, target_category="scenic")
+        sb = AutoClipService._score_candidate(b, target_duration=10.0, target_category="scenic")
+        assert sa > sb
+
+    def test_duration_match_beats_mismatch(self) -> None:
+        a = _StubAsset(id="a", quality_score=80.0, duration=10.0, metadata={})
+        b = _StubAsset(id="b", quality_score=80.0, duration=30.0, metadata={})
+        sa = AutoClipService._score_candidate(a, target_duration=10.0, target_category=None)
+        sb = AutoClipService._score_candidate(b, target_duration=10.0, target_category=None)
+        assert sa > sb
+
+    def test_category_match_beats_mismatch(self) -> None:
+        a = _StubAsset(id="a", quality_score=80.0, duration=10.0, metadata={"category": "scenic"})
+        b = _StubAsset(id="b", quality_score=80.0, duration=10.0, metadata={"category": "tech"})
+        sa = AutoClipService._score_candidate(a, target_duration=10.0, target_category="scenic")
+        sb = AutoClipService._score_candidate(b, target_duration=10.0, target_category="scenic")
+        assert sa > sb
+
+    def test_no_target_category_all_get_full_classification(self) -> None:
+        a = _StubAsset(id="a", quality_score=80.0, duration=10.0, metadata={})
+        score = AutoClipService._score_candidate(a, target_duration=10.0, target_category=None)
+        # classification_score = 1.0 when no target
+        assert score == pytest.approx(0.5 * 0.8 + 0.3 * 1.0 + 0.2 * 1.0)
+
+    def test_no_quality_defaults_to_half(self) -> None:
+        a = _StubAsset(id="a", quality_score=None, duration=10.0, metadata={})
+        score = AutoClipService._score_candidate(a, target_duration=None, target_category=None)
+        assert score == pytest.approx(0.5 * 0.5 + 0.3 * 0.5 + 0.2 * 1.0)
+
+
+# ── 测试：解析素材需求 ────────────────────────────────────────────────────────
+
+
+class TestParseMaterialRequirements:
+    def test_none_config_returns_empty(self) -> None:
+        assert AutoClipService._parse_material_requirements(None) == {}
+
+    def test_extracts_file_type(self) -> None:
+        config = _StubConfig(
+            id="c1", template_id="t1",
+            material_requirements={"type": "video"},
+        )
+        result = AutoClipService._parse_material_requirements(config)
+        assert result["file_type"] == "video"
+
+    def test_extracts_min_quality(self) -> None:
+        config = _StubConfig(
+            id="c1", template_id="t1",
+            material_requirements={"min_quality_score": 60},
+        )
+        result = AutoClipService._parse_material_requirements(config)
+        assert result["min_quality_score"] == 60.0
+
+    def test_extracts_category(self) -> None:
+        config = _StubConfig(
+            id="c1", template_id="t1",
+            material_requirements={"category": "scenic"},
+        )
+        result = AutoClipService._parse_material_requirements(config)
+        assert result["classification_category"] == "scenic"
+
+    def test_invalid_category_ignored(self) -> None:
+        config = _StubConfig(
+            id="c1", template_id="t1",
+            material_requirements={"category": "nonexistent"},
+        )
+        result = AutoClipService._parse_material_requirements(config)
+        assert "classification_category" not in result
+
+    def test_duration_range(self) -> None:
+        config = _StubConfig(
+            id="c1", template_id="t1",
+            min_duration=5.0, max_duration=15.0,
+            material_requirements={},
+        )
+        result = AutoClipService._parse_material_requirements(config)
+        assert result["min_duration"] == 5.0
+        assert result["max_duration"] == 15.0
+        assert result["target_duration"] == 10.0
+
+    def test_tags_extracted(self) -> None:
+        config = _StubConfig(
+            id="c1", template_id="t1",
+            material_requirements={"tags": ["outdoor", "sunset"]},
+        )
+        result = AutoClipService._parse_material_requirements(config)
+        assert result["tags"] == ["outdoor", "sunset"]