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

"""PlanGeneratorService — 基于模板+素材自动生成剪辑计划.

核心职责:
- 根据 EditTemplate 的 editing_mode 和 TemplateClipConfig 列表，
  自动生成 EditPlan + EditPlanClip 列表
- 四种模式素材分配策略:
  - ONE_TAKE: 素材顺序分配给 main 类型 clips
  - PIP: 第1个素材→main(全屏背景)，其余→overlay clips
  - VOICE_OVER: 素材→main clips (B-roll)，标记需要配音叠加
  - VOICE_PIP: 第1个→background, 第2个→corner_voice, 其余→b_roll
"""

from __future__ import annotations

import logging
import random
from typing import Any, List

from sqlalchemy.orm import Session

from packages.adapters.sqlalchemy_impl import (
    SQLAlchemyEditPlanClipRepository,
    SQLAlchemyEditPlanRepository,
)
from packages.domain.config_schemas import normalize_plan_config
from packages.domain.edit_plan import EditPlan
from packages.domain.edit_plan_clip import EditPlanClip
from packages.domain.edit_template import EditTemplate
from packages.domain.editing_mode import EditingMode
from packages.domain.plan_generator_utils import (
    create_clips_from_configs,
    distribute_assets,
    generate_default_clips,
    map_clip_types_for_mode,
)
from packages.domain.smart_match import SCORE_RANDOM_NOISE_MAX, score_asset
from packages.domain.template_clip_config import TemplateClipConfig

logger = logging.getLogger(__name__)


# ── 默认片段时长（秒） ────────────────────────────────────────────────────────
_DEFAULT_CLIP_DURATION = 5.0
_DEFAULT_INTRO_DURATION = 3.0
_DEFAULT_OUTRO_DURATION = 3.0


class PlanGeneratorService:
    """剪辑计划生成器

    基于模板 + 素材，自动生成 EditPlan 及 EditPlanClip 列表。
    """

    def __init__(self, db: Session, asset_repo=None) -> None:
        self._plan_repo = SQLAlchemyEditPlanRepository(db)
        self._clip_repo = SQLAlchemyEditPlanClipRepository(db)
        self._asset_repo = asset_repo

    # ── 公开接口 ─────────────────────────────────────────────────────────────

    def generate_from_template(
        self,
        template: EditTemplate,
        clip_configs: List[TemplateClipConfig],
        asset_ids: List[str],
        *,
        project_id: str = "",
        created_by_user_id: str = "",
        name: str = "",
        random_preview: bool = False,
    ) -> dict[str, Any]:
        """基于模板+素材生成剪辑计划

        Args:
            template: 剪辑模板实体
            clip_configs: 模板片段配置列表（可为空，自动生成默认结构）
            asset_ids: 素材 ID 列表
            project_id: 所属项目 ID
            created_by_user_id: 创建者用户 ID
            name: 计划名称（为空则自动取模板名）
            random_preview: 是否启用随机预览模式（随机选素材+随机截取片段）

        Returns:
            dict: {"plan": EditPlan, "clips": List[EditPlanClip]}
        """
        editing_mode = template.editing_mode or EditingMode.ONE_TAKE.value
        plan_name = name.strip() or f"{template.name} - 剪辑计划"

        # 1. 构建 plan config（继承模板的 title/subtitle/bgm，记录 editing_mode）
        plan_config = self._build_plan_config(template, editing_mode)

        # 2. 创建 EditPlan
        plan = EditPlan.create(
            template_id=template.id,
            name=plan_name,
            config=plan_config,
            total_duration=0.0,
            project_id=project_id,
            created_by_user_id=created_by_user_id,
        )
        plan = self._plan_repo.create(plan)
        logger.info(
            "生成剪辑计划: plan_id=%s template=%s mode=%s assets=%d",
            plan.id,
            template.id,
            editing_mode,
            len(asset_ids),
        )

        # 3. 生成片段列表
        if clip_configs:
            clips = self._create_clips_from_configs(plan.id, clip_configs)
            # 模板 clip_config 的 clip_type 是 ClipType 枚举（main/intro/outro 等），
            # 但 PIP / VOICE_PIP 模式需要特定的 clip_type（overlay/background/corner_voice/b_roll）
            # 才能让素材分配和渲染分层正确工作。
            # 这里将 MAIN 类型的片段按顺序映射为对应模式的角色类型。
            self._map_clip_types_for_mode(clips, editing_mode)
        else:
            clips = self._generate_default_clips(plan.id, editing_mode, len(asset_ids))

        # 4. 按 editing_mode 分配素材
        if asset_ids:
            # 获取素材时长信息，用于随机起始时间
            asset_durations = None
            if self._asset_repo:
                asset_durations = self._fetch_asset_durations(asset_ids)
            self._distribute_assets(
                clips,
                asset_ids,
                editing_mode,
                random_selection=random_preview,
                asset_durations=asset_durations,
            )

        # 5. 持久化所有 clips 并计算总时长
        created_clips: List[EditPlanClip] = []
        total_duration = 0.0
        for clip in clips:
            saved = self._clip_repo.create(clip)
            created_clips.append(saved)
            total_duration += saved.duration

        # 6. 更新 plan 的 total_duration
        plan.total_duration = total_duration
        plan = self._plan_repo.update(plan)

        # 7. 流转到 editing 状态
        try:
            plan.start_editing()
            plan = self._plan_repo.update(plan)
        except ValueError as exc:
            logger.warning("计划状态流转失败: plan_id=%s error=%s", plan.id, exc)

        logger.info(
            "剪辑计划生成完成: plan_id=%s clips=%d duration=%.1f",
            plan.id,
            len(created_clips),
            total_duration,
        )

        return {"plan": plan, "clips": created_clips}

    # ── 内部方法 ─────────────────────────────────────────────────────────────

    def _build_plan_config(
        self,
        template: EditTemplate,
        editing_mode: str,
    ) -> dict[str, Any]:
        """从模板配置构建 plan config"""
        template_config = template.config or {}
        plan_config: dict[str, Any] = {
            "editing_mode": editing_mode,
        }
        # 继承模板的 cover/title/subtitle/bgm/export/filter 配置
        for key in ("cover", "title", "subtitle", "bgm", "export", "filter"):
            if key in template_config:
                plan_config[key] = template_config[key]

        return normalize_plan_config(plan_config)

    def _create_clips_from_configs(
        self,
        plan_id: str,
        clip_configs: List[TemplateClipConfig],
    ) -> List[EditPlanClip]:
        """从 TemplateClipConfig 列表创建 EditPlanClip 列表（未持久化）.

        委托给 plan_generator_utils.create_clips_from_configs 纯函数。
        """
        return create_clips_from_configs(plan_id, clip_configs)

    def _map_clip_types_for_mode(self, clips: List[EditPlanClip], editing_mode: str) -> None:
        """将 MAIN 类型片段按 editing_mode 映射为对应角色类型.

        委托给 plan_generator_utils.map_clip_types_for_mode 纯函数。
        """
        map_clip_types_for_mode(clips, editing_mode)

    def _generate_default_clips(
        self,
        plan_id: str,
        editing_mode: str,
        asset_count: int,
    ) -> List[EditPlanClip]:
        """无 clip_configs 时，根据 editing_mode 生成默认 clip 结构.

        委托给 plan_generator_utils.generate_default_clips 纯函数。
        """
        return generate_default_clips(plan_id, editing_mode, asset_count)

    def _distribute_assets(
        self,
        clips: List[EditPlanClip],
        asset_ids: List[str],
        editing_mode: str,
        *,
        random_selection: bool = False,
        asset_durations: dict[str, float] | None = None,
    ) -> None:
        """按 editing_mode 将素材分配到 clips（就地修改，未持久化）.

        先用 smart_match 评分对素材排序（高分优先），再委托给
        plan_generator_utils.distribute_assets 纯函数完成分配。
        """
        # 用 smart_match 评分排序素材：高分（质量好/时长合适/新鲜/未使用）优先
        if self._asset_repo and not random_selection:
            asset_ids = self._sort_assets_by_smart_score(asset_ids)

        distribute_assets(
            clips,
            asset_ids,
            editing_mode,
            random_selection=random_selection,
            asset_durations=asset_durations,
        )

    def _sort_assets_by_smart_score(self, asset_ids: List[str]) -> List[str]:
        """按 smart_match 综合评分降序排列素材 ID（注入随机噪声）。

        评分高的素材（质量好、时长合适、新鲜、使用次数少）倾向排在前面；
        排序时给每个素材的得分注入 0~SCORE_RANDOM_NOISE_MAX 的随机噪声，
        使得分接近的素材排名每次浮动，避免一键生成反复选出相同素材组合，
        从素材组合层面降低成片查重率。分差大于噪声上限时排名保持稳定。
        """
        scored: list[tuple[str, float]] = []
        for asset_id in asset_ids:
            asset = self._asset_repo.get(asset_id)
            if asset:
                score, _ = score_asset(asset)
                scored.append((asset_id, score))
            else:
                scored.append((asset_id, 0.0))
        # 评分 + 随机噪声后按降序排列
        scored.sort(
            key=lambda x: x[1] + random.uniform(0.0, SCORE_RANDOM_NOISE_MAX),
            reverse=True,
        )
        return [aid for aid, _ in scored]

    def _fetch_asset_durations(self, asset_ids: List[str]) -> dict[str, float]:
        """从数据库获取素材时长信息.

        Args:
            asset_ids: 素材 ID 列表

        Returns:
            dict: 素材 ID -> 时长（秒）映射
        """
        durations: dict[str, float] = {}
        for asset_id in asset_ids:
            asset = self._asset_repo.get(asset_id)
            if asset and hasattr(asset, "duration"):
                durations[asset_id] = float(asset.duration or 0.0)
        return durations