saas-backend-agent
5f4bad170a
from-assets（clips.py）与一键生成（plan_generator_service.py）两条 素材选取路径中，smart_match 评分排序注入 0~20 分随机噪声： - 分差 > 20 的素材排名稳定，质量差距显著的素材保持优先级 - 得分接近的素材排名每次浮动，13 条素材选出的 4 条组合不再固定， 排名靠后的低分素材也有机会入选，从素材组合层面降低成片重复率 - 新增 SCORE_RANDOM_NOISE_MAX 常量（smart_match.py）统一定义 - 确定性测试通过 zero-noise patch 消除随机性，新增 3 个噪声有效性测试 （等分素材排名浮动 / 大分差不翻转），全量 14034 测试通过
276 lines
10 KiB
Python
Executable File
276 lines
10 KiB
Python
Executable File
"""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
|