"""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, extract_scene_points_from_metadata, 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, user_id=created_by_user_id, ) # 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, user_id: str = "", ) -> None: """按 editing_mode 将素材分配到 clips（就地修改，未持久化）. 先用 smart_match 评分对素材排序（高分优先），再委托给 plan_generator_utils.distribute_assets 纯函数完成分配。 """ # 预览随机模式：素材顺序已 shuffle，纯随机起点即可，不读 DB 评分/缓存 asset_scene_points: dict[str, list[float]] = {} if not random_selection: # 正式生成：smart_match 评分排序（高分优先）+ 场景切换点缓存 if self._asset_repo: asset_ids = self._sort_assets_by_smart_score(asset_ids) # 读取素材 metadata 中的场景切换点缓存（后台 SceneChange 检测写入）： # 有缓存的素材片段起点从随机镜头段选取，无缓存走随机起点兜底 asset_scene_points = self._fetch_asset_scene_points(asset_ids) # 正式生成也随机重排片段顺序（降重，默认开启无开关） # smart_match 决定选哪些素材，shuffle 只改变分配到 clips 的顺序 asset_ids = list(asset_ids) # 复制避免修改调用方原列表 random.shuffle(asset_ids) # 查询已有视频的已用区间（跨视频避让） external_used_segments = None if user_id and self._clip_repo: try: external_used_segments = self._clip_repo.list_used_segments_by_user(user_id, limit_recent=50) except Exception: logger.warning("跨视频避让查询失败，回退到纯随机", exc_info=True) distribute_assets( clips, asset_ids, editing_mode, random_selection=random_selection, asset_durations=asset_durations, asset_scene_points=asset_scene_points, external_used_segments=external_used_segments, ) def _fetch_asset_scene_points(self, asset_ids: List[str]) -> dict[str, list[float]]: """从素材 metadata 读取场景切换点缓存（无缓存的素材不包含在结果中）。""" points_map: dict[str, list[float]] = {} if not self._asset_repo: return points_map for asset_id in asset_ids: asset = self._asset_repo.get(asset_id) if asset: points = extract_scene_points_from_metadata(getattr(asset, "metadata", None)) if points: points_map[asset_id] = points return points_map 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