xiaoxia-saas/apps/api/app/services/auto_clip_service.py

"""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

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) as e:
                logger.warning(f"Operation failed in apps/api/app/services/auto_clip_service.py: {e}", exc_info=True)

        # 分类筛选
        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