From 18ea6cfadb5f42415ef438620cce5ccb58689dbe Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E7=81=B5=E5=BA=94?= Date: Wed, 1 Jul 2026 17:45:43 +0800 Subject: [PATCH] =?UTF-8?q?feat(phase8):=20AutoClipService=20=E6=99=BA?= =?UTF-8?q?=E8=83=BD=E9=80=89=E7=89=87=E6=9C=8D=E5=8A=A1=20(=E4=BB=BB?= =?UTF-8?q?=E5=8A=A1=202.07)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 新增 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 | 2 + apps/api/app/services/auto_clip_service.py | 340 +++++++++++++++++ .../sqlalchemy_impl/asset_repository.py | 48 +++ packages/ports/asset_repository.py | 17 + tests/unit/test_auto_clip_service.py | 347 ++++++++++++++++++ 5 files changed, 754 insertions(+) create mode 100644 apps/api/app/services/auto_clip_service.py create mode 100644 tests/unit/test_auto_clip_service.py diff --git a/apps/api/app/services/__init__.py b/apps/api/app/services/__init__.py index b05a02004..c001ef37c 100644 --- a/apps/api/app/services/__init__.py +++ b/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", ] diff --git a/apps/api/app/services/auto_clip_service.py b/apps/api/app/services/auto_clip_service.py new file mode 100644 index 000000000..bb945b16b --- /dev/null +++ b/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 diff --git a/packages/adapters/sqlalchemy_impl/asset_repository.py b/packages/adapters/sqlalchemy_impl/asset_repository.py index ce1e8c9b7..a8cd61a07 100644 --- a/packages/adapters/sqlalchemy_impl/asset_repository.py +++ b/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: diff --git a/packages/ports/asset_repository.py b/packages/ports/asset_repository.py index 409dcb514..e5643a77f 100644 --- a/packages/ports/asset_repository.py +++ b/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 diff --git a/tests/unit/test_auto_clip_service.py b/tests/unit/test_auto_clip_service.py new file mode 100644 index 000000000..f21c03d7b --- /dev/null +++ b/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"]