"""EditPlanService — 剪辑计划管理业务逻辑. 封装 EditPlan 和 EditPlanClip 的 CRUD 操作、状态机流转、 以及渲染生成流程，提供统一的业务接口供 API 路由层调用。 """ from __future__ import annotations import logging from typing import Any, Dict, List, Optional from app.services.asset_segment_tracker import ( REUSE_RATIO_LIMIT, get_used_segments, make_reuse_callback, record_used_segments, ) from sqlalchemy.orm import Session from packages.adapters.sqlalchemy_impl import ( SQLAlchemyEditPlanClipRepository, SQLAlchemyEditPlanRepository, SQLAlchemyGenerationTaskRepository, ) from packages.domain.clip_operations import calculate_merge as _calc_merge from packages.domain.clip_operations import calculate_shift_orders as _calc_shift_orders from packages.domain.clip_operations import calculate_split as _calc_split from packages.domain.clip_operations import validate_merge_clips as _validate_merge from packages.domain.clip_operations import validate_split_time as _validate_split from packages.domain.edit_plan import EditPlan, EditPlanStatus from packages.domain.edit_plan_clip import EditPlanClip, EditPlanClipStatus logger = logging.getLogger(__name__) class EditPlanService: """剪辑计划管理服务 职责: - 剪辑计划 CRUD（创建、查询、更新、删除） - 剪辑片段管理（增删改查、分配素材） - 状态机流转（draft → editing → rendering → completed/failed） - 渲染生成流程（触发 Celery 任务、查询进度） """ def __init__(self, db: Session) -> None: self._plan_repo = SQLAlchemyEditPlanRepository(db) self._clip_repo = SQLAlchemyEditPlanClipRepository(db) self._generation_task_repo = SQLAlchemyGenerationTaskRepository(db) # ── 剪辑计划 CRUD ────────────────────────────────────────────────────── def get_plan(self, plan_id: str) -> Optional[EditPlan]: """获取计划详情""" return self._plan_repo.get(plan_id) def get_plan_or_raise(self, plan_id: str) -> EditPlan: """获取计划，不存在则抛出 ValueError""" plan = self._plan_repo.get(plan_id) if plan is None: raise ValueError(f"剪辑计划不存在: {plan_id}") return plan def create_plan( self, template_id: str, name: str, *, config: Optional[dict[str, Any]] = None, total_duration: float = 0.0, project_id: str = "", created_by_user_id: str = "", ) -> EditPlan: """创建剪辑计划（基础 CRUD，供内部测试与脚本使用） Raises: ValueError: 参数校验失败 """ plan = EditPlan.create( template_id=template_id, name=name, config=config, total_duration=total_duration, project_id=project_id, created_by_user_id=created_by_user_id, ) created = self._plan_repo.create(plan) logger.info("创建剪辑计划: id=%s name=%s", created.id, created.name) return created def _auto_resume_editing(self, plan_id: str) -> None: """如果计划处于 completed/failed 状态，自动切回 editing（编辑操作前置）""" plan = self._plan_repo.get(plan_id) if plan is None: return if plan.status in (EditPlanStatus.COMPLETED, EditPlanStatus.FAILED): try: plan.resume_editing() self._plan_repo.update(plan) logger.info("自动重新编辑: plan_id=%s", plan_id) except ValueError: pass def update_plan( self, plan_id: str, *, name: Optional[str] = None, config: Optional[dict[str, Any]] = None, total_duration: Optional[float] = None, ) -> EditPlan: """更新计划基础字段 Raises: ValueError: 计划不存在 """ existing = self.get_plan_or_raise(plan_id) # 自动从 completed/failed 切回 editing self._auto_resume_editing(plan_id) existing = self.get_plan_or_raise(plan_id) updated = EditPlan( id=existing.id, template_id=existing.template_id, name=name.strip() if name is not None else existing.name, status=existing.status, total_duration=total_duration if total_duration is not None else existing.total_duration, source_edit_plan_id=existing.source_edit_plan_id, project_id=existing.project_id, created_by_user_id=existing.created_by_user_id, config=config if config is not None else existing.config, created_at=existing.created_at, updated_at=existing.updated_at, ) result = self._plan_repo.update(updated) logger.info("更新剪辑计划: id=%s", plan_id) return result # ── 状态机流转 ────────────────────────────────────────────────────────── def transition_status(self, plan_id: str, target_status: EditPlanStatus) -> EditPlan: """流转计划状态 状态流转规则: - draft → editing (start_editing) - editing → rendering (start_rendering) - rendering → completed (mark_completed) - rendering → failed (mark_failed) - failed → draft (reset_to_draft) Raises: ValueError: 计划不存在或状态流转非法 """ plan = self.get_plan_or_raise(plan_id) # 如果已是目标状态，直接返回 if plan.status == target_status: return plan # 根据目标状态调用对应的状态机方法 # EDITING 支持从 draft / completed / failed 进入 if target_status == EditPlanStatus.EDITING: if plan.status == EditPlanStatus.DRAFT: plan.start_editing() elif plan.status in (EditPlanStatus.COMPLETED, EditPlanStatus.FAILED): plan.resume_editing() else: raise ValueError(f"无法从 {plan.status} 切换到 {target_status}") result = self._plan_repo.update(plan) logger.info( "状态流转: plan_id=%s %s → %s", plan_id, plan.status, target_status, ) return result transition_map = { EditPlanStatus.RENDERING: plan.start_rendering, EditPlanStatus.COMPLETED: plan.mark_completed, EditPlanStatus.FAILED: plan.mark_failed, EditPlanStatus.DRAFT: plan.reset_to_draft, } transition_fn = transition_map.get(target_status) if transition_fn is None: raise ValueError(f"无效的目标状态: {target_status}") transition_fn() result = self._plan_repo.update(plan) logger.info( "状态流转: plan_id=%s %s → %s", plan_id, plan.status, target_status, ) return result # ── 剪辑片段管理 ──────────────────────────────────────────────────────── def list_clips( self, plan_id: str, *, status: Optional[EditPlanClipStatus] = None, skip: int = 0, limit: int = 100, ) -> List[EditPlanClip]: """列出计划的片段""" # 确保计划存在 self.get_plan_or_raise(plan_id) return self._clip_repo.list_by_plan(plan_id, status=status, skip=skip, limit=limit) def count_clips( self, plan_id: str, *, status: Optional[EditPlanClipStatus] = None, ) -> int: """统计片段数量""" return self._clip_repo.count(plan_id=plan_id, status=status) def get_clip(self, clip_id: str) -> Optional[EditPlanClip]: """获取片段详情""" return self._clip_repo.get(clip_id) def get_clip_or_raise(self, clip_id: str) -> EditPlanClip: """获取片段，不存在则抛出 ValueError""" clip = self._clip_repo.get(clip_id) if clip is None: raise ValueError(f"片段不存在: {clip_id}") return clip def create_clip( self, plan_id: str, clip_type: str, order: int, *, template_clip_config_id: str = "", asset_id: str = "", text_content: str = "", start_time: float = 0.0, duration: float = 0.0, transition_effect: str = "cut", transition_duration: float = 0.0, playback_speed: float = 1.0, config: Optional[dict[str, Any]] = None, ) -> EditPlanClip: """创建片段 Raises: ValueError: 计划不存在或参数校验失败 """ # 确保计划存在 self.get_plan_or_raise(plan_id) # 自动从 completed/failed 切回 editing self._auto_resume_editing(plan_id) clip = EditPlanClip.create( plan_id=plan_id, clip_type=clip_type, order=order, template_clip_config_id=template_clip_config_id, asset_id=asset_id, text_content=text_content, start_time=start_time, duration=duration, transition_effect=transition_effect, transition_duration=transition_duration, playback_speed=playback_speed, config=config, ) created = self._clip_repo.create(clip) logger.info( "创建片段: id=%s plan_id=%s clip_type=%s order=%d", created.id, plan_id, created.clip_type, created.order, ) return created def update_clip( self, clip_id: str, *, clip_type: Optional[str] = None, order: Optional[int] = None, asset_id: Optional[str] = None, text_content: Optional[str] = None, start_time: Optional[float] = None, duration: Optional[float] = None, transition_effect: Optional[str] = None, transition_duration: Optional[float] = None, playback_speed: Optional[float] = None, config: Optional[dict[str, Any]] = None, ) -> EditPlanClip: """更新片段 Raises: ValueError: 片段不存在 """ existing = self.get_clip_or_raise(clip_id) # 自动从 completed/failed 切回 editing self._auto_resume_editing(existing.plan_id) # 速度边界钳制 if playback_speed is not None: if playback_speed <= 0: playback_speed = 1.0 elif playback_speed < 0.25: playback_speed = 0.25 elif playback_speed > 4.0: playback_speed = 4.0 updated = EditPlanClip( id=existing.id, plan_id=existing.plan_id, clip_type=clip_type.strip() if clip_type is not None else existing.clip_type, order=order if order is not None else existing.order, template_clip_config_id=existing.template_clip_config_id, asset_id=asset_id.strip() if asset_id is not None else existing.asset_id, text_content=text_content.strip() if text_content is not None else existing.text_content, start_time=start_time if start_time is not None else existing.start_time, duration=duration if duration is not None else existing.duration, transition_effect=( transition_effect.strip() if transition_effect is not None else existing.transition_effect ), transition_duration=( transition_duration if transition_duration is not None else existing.transition_duration ), playback_speed=playback_speed if playback_speed is not None else existing.playback_speed, status=existing.status, config=config if config is not None else existing.config, created_at=existing.created_at, updated_at=existing.updated_at, ) result = self._clip_repo.update(updated) logger.info("更新片段: id=%s", clip_id) return result def assign_asset(self, clip_id: str, asset_id: str) -> EditPlanClip: """为片段分配素材 Raises: ValueError: 片段不存在或 asset_id 为空 """ clip = self.get_clip_or_raise(clip_id) # 自动从 completed/failed 切回 editing self._auto_resume_editing(clip.plan_id) clip.assign_asset(asset_id) result = self._clip_repo.update(clip) logger.info("分配素材: clip_id=%s asset_id=%s", clip_id, asset_id) return result def delete_clip(self, clip_id: str) -> bool: """删除片段 Returns: bool: 是否删除成功 """ deleted = self._clip_repo.delete(clip_id) if deleted: logger.info("删除片段: id=%s", clip_id) return deleted def delete_all_clips(self, plan_id: str) -> int: """删除计划下所有片段 Returns: int: 删除的片段数量 """ count = self._clip_repo.delete_by_plan(plan_id) logger.info("删除所有片段: plan_id=%s count=%d", plan_id, count) return count def replace_all_clips_transactional( self, plan_id: str, clips_data: list[dict], ) -> int: """事务性地替换所有片段：清空→创建→标记ready，单事务保证原子性。 Args: plan_id: 计划 ID clips_data: 片段数据列表，每项包含 asset_id/start_time/duration/order Returns: int: 创建的片段数量 Raises: Exception: 任何步骤失败时自动回滚 """ from packages.adapters.sqlalchemy_impl.models import EditPlanClipModel db = self._clip_repo.session try: # 1. 清空现有 clips（不 commit） deleted_count = db.query(EditPlanClipModel).filter(EditPlanClipModel.plan_id == plan_id).delete() # 2. 批量创建新 clips（不 commit） for i, clip_item in enumerate(clips_data): order = clip_item.get("order") or i clip = EditPlanClip.create( plan_id=plan_id, clip_type=clip_item.get("clip_type", "main"), order=order, asset_id=clip_item.get("asset_id", ""), text_content=clip_item.get("text_content", ""), start_time=clip_item.get("start_time", 0.0), duration=clip_item.get("duration", 0.0), transition_effect=clip_item.get("transition_effect", "cut"), transition_duration=clip_item.get("transition_duration", 0.0), playback_speed=clip_item.get("playback_speed", 1.0), config=clip_item.get("config") or None, ) model = EditPlanClipModel( id=clip.id, plan_id=clip.plan_id, clip_type=clip.clip_type, order=clip.order, asset_id=clip.asset_id, text_content=clip.text_content, start_time=clip.start_time, duration=clip.duration, transition_effect=clip.transition_effect, transition_duration=clip.transition_duration, playback_speed=clip.playback_speed, status=clip.status.value, config=clip.config, ) db.add(model) # flush 让新建 clip 写入当前事务（未 commit），后续查询才能找到它们 db.flush() # 3. 标记有 asset_id 的 clips 为 ready（不 commit） pending_with_asset = ( db.query(EditPlanClipModel) .filter( EditPlanClipModel.plan_id == plan_id, EditPlanClipModel.status == "pending", EditPlanClipModel.asset_id != "", ) .all() ) for m in pending_with_asset: m.status = "ready" # 4. 一次性提交 db.commit() logger.info( "事务性替换片段: plan_id=%s deleted=%d created=%d", plan_id, deleted_count, len(clips_data), ) return len(clips_data) except Exception: db.rollback() logger.exception("事务性替换片段失败: plan_id=%s", plan_id) raise def reselect_plan_for_variant( self, source_plan_id: str, candidate_asset_ids: list[str], *, created_by_user_id: str = "", name_suffix: str = "变体", voice_duration: float = 0.0, rng=None, ) -> EditPlan: """为批量变体生成独立 plan：完整重跑单视频选片流程（#1743）。 与 clone_plan_for_variant（只重算起点、素材/顺序不变）不同，本方法： - 源 plan 片段骨架（clip_type/order/duration/文案/转场）保留； - 素材池 shuffle 随机分配 + main 片段顺序洗牌； - 起点走场景镜头洗牌/随机起点/历史区间避让（与单视频同一入口）； - 批次内同素材区间重叠 >20% 自动重选起点； - 新片段区间 record_used_segments 写回素材 metadata（跨变体/跨任务避让）。 Args: source_plan_id: 源 plan（任务 0 / 预览源）。 candidate_asset_ids: 素材池（源 plan 素材 ∪ 批次素材）。 created_by_user_id: 新 plan 归属用户。 name_suffix: plan 名后缀。 rng: 可选随机数（测试注入种子）。 Raises: ValueError: 源 plan 不存在/无片段、素材池为空或时长全未知。 """ from packages.adapters.sqlalchemy_impl.models import AssetModel from packages.domain.plan_generator_utils import extract_scene_points_from_metadata from packages.domain.variant_plan_selector import reselect_clips_for_variant source = self.get_plan_or_raise(source_plan_id) # 分页读取源 plan 全部片段 clips: List[EditPlanClip] = [] skip, page = 0, 500 while True: batch = self._clip_repo.list_by_plan(source_plan_id, skip=skip, limit=page) if not batch: break clips.extend(batch) if len(batch) < page: break skip += page if not clips: raise ValueError(f"源 plan 无片段，无法生成变体: {source_plan_id}") source_clips_data: list[dict[str, Any]] = [ { "order": c.order if c.order is not None else i, "asset_id": c.asset_id, "start_time": float(c.start_time or 0.0), "duration": float(c.duration or 0.0), "clip_type": c.clip_type, "playback_speed": float(c.playback_speed or 1.0), "transition_effect": c.transition_effect, "transition_duration": float(c.transition_duration or 0.0), "text_content": c.text_content or "", "config": c.config or {}, } for i, c in enumerate(clips) ] db = self._clip_repo.session # #1749：配音时长 → 每段目标段长（片段数=模板片段数定死；素材不足由渲染末帧冻结铺满） target_durations: list[float] | None = None try: voice = float(voice_duration or 0.0) except (TypeError, ValueError): voice = 0.0 if voice > 0 and source_clips_data: from packages.domain.voice_duration_planner import plan_clip_durations _effects: list[str | None] = [c.get("transition_effect") for c in source_clips_data] _tdurs: list[float] = [float(c.get("transition_duration") or 0.0) for c in source_clips_data] target_durations = plan_clip_durations( len(source_clips_data), voice, transition_effects=_effects, transition_durations=_tdurs, ) if target_durations: for _c, _d in zip(source_clips_data, target_durations, strict=False): _c["duration"] = _d # 素材池 = 源 plan 素材 ∪ 调用方传入素材（去重保序） pool_ids: list[str] = [] seen = set() for aid in [c.asset_id for c in clips if c.asset_id] + list(candidate_asset_ids or []): if aid and aid not in seen: seen.add(aid) pool_ids.append(aid) # 时长 + 场景点 durations: dict[str, float] = {} scene_points: dict[str, list[float]] = {} if pool_ids: for m in db.query(AssetModel).filter(AssetModel.id.in_(pool_ids)).all(): durations[m.id] = float(getattr(m, "duration", 0.0) or 0.0) pts = extract_scene_points_from_metadata(getattr(m, "metadata", None)) if pts: scene_points[m.id] = pts historical = get_used_segments(db, pool_ids) # 创建新 plan（复制模板归属与 config） new_plan = self.create_plan( template_id=source.template_id, name=f"{source.name or '剪辑计划'} · {name_suffix}", config=dict(source.config or {}), total_duration=source.total_duration, project_id=source.project_id or "", created_by_user_id=created_by_user_id or (source.created_by_user_id or ""), ) # 批次内区间：以源 plan（变体 0）片段为初始避让对象 batch_segments: dict[str, list[tuple[float, float]]] = {} for c in clips: if c.asset_id and float(c.duration or 0) > 0: st = float(c.start_time or 0.0) batch_segments.setdefault(c.asset_id, []).append((st, st + float(c.duration))) clips_data = reselect_clips_for_variant( source_clips_data, pool_ids, asset_durations=durations, asset_scene_points=scene_points, historical_used_segments=historical, batch_segments=batch_segments, target_durations=target_durations, rng=rng, ) # 片段区间写回素材 metadata（与落库同事务；replace_all_clips_transactional 内 commit） for item in clips_data: aid = item.get("asset_id", "") if aid: st = float(item.get("start_time", 0.0)) record_used_segments(db, aid, st, st + float(item.get("duration", 0.0)), new_plan.id) self.replace_all_clips_transactional(new_plan.id, clips_data) logger.info( "变体独立选片完成: source=%s new=%s clips=%d assets=%d", source_plan_id, new_plan.id, len(clips_data), len(pool_ids), ) return new_plan def clone_plan_for_variant( self, source_plan_id: str, *, created_by_user_id: str = "", name_suffix: str = "变体", reuse_tracker: Optional[dict] = None, ) -> EditPlan: """为批量/多预览场景克隆一份独立 plan，片段起点全部重算（受控随机/复用）。 复制源 plan 的模板归属、config 与片段结构（asset_id / duration / clip_type / order 不变），每个片段重新调用 ``_calc_random_start_time``：读取素材持久化的 历史已用区间避让，耗尽时受控复用（use_count<3、最久未用），从而保证 N 条 成片片段区间互不相同，且复用占比受控。 - 不替换/不修改源 plan，源 plan 保留用户手动编辑结果。 - 片段区间记录（record_used_segments）随新片段写入素材 metadata，与新 plan 同事务；复用历史区间时由 record 自动累加 use_count。 - 克隆的 clips 复用区间累计时长写入 reuse_tracker（可选），供调用方统计占比。 Raises: ValueError: 源 plan 不存在或无可用片段。 """ from packages.adapters.sqlalchemy_impl.models import AssetModel from packages.domain.plan_generator_utils import _calc_random_start_time source = self.get_plan_or_raise(source_plan_id) # 分页读取源 plan 全部片段 clips: List[EditPlanClip] = [] skip, page = 0, 500 while True: batch = self._clip_repo.list_by_plan(source_plan_id, skip=skip, limit=page) if not batch: break clips.extend(batch) if len(batch) < page: break skip += page if not clips: raise ValueError(f"源 plan 无片段，无法克隆变体: {source_plan_id}") # 创建新 plan（复制模板归属与 config） new_plan = self.create_plan( template_id=source.template_id, name=f"{source.name or '剪辑计划'} · {name_suffix}", config=dict(source.config or {}), total_duration=source.total_duration, project_id=source.project_id or "", created_by_user_id=created_by_user_id or (source.created_by_user_id or ""), ) # 素材时长映射（O(N) 单查） asset_ids = list({c.asset_id for c in clips if c.asset_id}) db = self._clip_repo.session durations: dict[str, float] = {} if asset_ids: for m in db.query(AssetModel).filter(AssetModel.id.in_(asset_ids)).all(): durations[m.id] = float(getattr(m, "duration", 0.0) or 0.0) used_segments = get_used_segments(db, asset_ids) reused: dict[str, float] = reuse_tracker if reuse_tracker is not None else {} asset_assigned: dict[str, float] = {} # 回调内部预判复用后占比超 15% 则拒绝复用（calc 返回 None → 保留原起点） reuse_cb = make_reuse_callback(db, durations, reused, assigned_tracker=asset_assigned) clips_data: list[dict] = [] for i, c in enumerate(clips): aid = c.asset_id dur = float(c.duration or 0.0) total = durations.get(aid, 0.0) if aid and total > 0 and dur > 0: # 复用占比闸门：本片段尚未分配，检查当前已复用占比 # reused / assigned 是否超 15%，超则该素材不再分配（保留原起点）； # assigned=0（首个片段）放行，reused=0 时不误拦正常分配 assigned = asset_assigned.get(aid, 0.0) eff_dur = min(dur, total) reused_amt = reused.get(aid, 0.0) ratio_blocked = assigned > 0 and reused_amt / assigned > REUSE_RATIO_LIMIT start = None if not ratio_blocked: start = _calc_random_start_time(aid, eff_dur, durations, used_segments, on_exhausted=reuse_cb) if start is None: start = float(c.start_time or 0.0) asset_assigned[aid] = assigned + eff_dur used_segments.setdefault(aid, []).append((start, start + eff_dur)) record_used_segments(db, aid, start, start + eff_dur, new_plan.id) else: start = float(c.start_time or 0.0) clips_data.append( { "order": c.order if c.order is not None else i, "asset_id": aid, "start_time": start, "duration": dur, "clip_type": c.clip_type, } ) # 事务性写入新 plan 的片段（内部统一 commit/rollback） self.replace_all_clips_transactional(new_plan.id, clips_data) logger.info( "克隆变体 plan: source=%s new=%s clips=%d", source_plan_id, new_plan.id, len(clips_data), ) return new_plan # ── #1749 配音时长分配 / 素材时长查询 / 批量变体 plan 确保 ────────────── def get_asset_durations(self, asset_ids: list[str]) -> dict[str, float]: """批量查询素材时长（秒），O(N) 单查；缺失/异常返回 0.0。""" from packages.adapters.sqlalchemy_impl.models import AssetModel ids = [a for a in dict.fromkeys(asset_ids or []) if a] if not ids: return {} db = self._clip_repo.session out: dict[str, float] = {} for m in db.query(AssetModel).filter(AssetModel.id.in_(ids)).all(): try: out[m.id] = float(getattr(m, "duration", 0.0) or 0.0) except (TypeError, ValueError): out[m.id] = 0.0 return out def apply_voice_duration_to_plan(self, plan_id: str, voice_duration: float) -> Optional[EditPlan]: """把配音时长分配到 plan 的每段（#1749）。 - 片段数保持不变（= 模板片段数，定死）； - 每段 duration 按 voice_duration_planner 分配（含转场重叠扣减）； - 素材短于段长 → start_time 钳制为 0（末帧冻结由渲染侧 tpad/apad 铺满）； - plan.total_duration 回写为成片净时长（≈ 配音时长）； - 幂等：配音时长相同则分配结果不变，可重复调用。 无配音（<=0）或无片段时直接返回 None，不报错。 """ try: voice = float(voice_duration or 0.0) except (TypeError, ValueError): return None if voice <= 0: return None plan = self.get_plan(plan_id) if plan is None: return None clips: List[EditPlanClip] = [] skip, page = 0, 500 while True: batch = self._clip_repo.list_by_plan(plan_id, skip=skip, limit=page) if not batch: break clips.extend(batch) if len(batch) < page: break skip += page if not clips: return None clips.sort(key=lambda c: (c.order if c.order is not None else 0)) from packages.domain.voice_duration_planner import plan_clip_durations, total_output_duration # #1764：从 plan config 读取节奏模板 rhythm_template = None if plan and hasattr(plan, "config") and plan.config: rhythm_template = plan.config.get("rhythm_template") # #1768：先获取素材时长，传入 plan_clip_durations 用于最大片段钳制 asset_ids = [c.asset_id for c in clips if c.asset_id] durations = self.get_asset_durations(asset_ids) asset_durations_for_plan = [durations.get(c.asset_id, 0.0) for c in clips] target = plan_clip_durations( len(clips), voice, transition_effects=[c.transition_effect for c in clips], transition_durations=[float(c.transition_duration or 0.0) for c in clips], rhythm_template=rhythm_template, asset_durations=asset_durations_for_plan, ) if not target: return None clips_data: list[dict] = [] for i, c in enumerate(clips): dur = float(target[i]) total = durations.get(c.asset_id, 0.0) start = float(c.start_time or 0.0) if c.asset_id and total > 0: # 素材短于段长：起点钳 0，段长超出部分渲染侧末帧冻结 max_start = max(0.0, total - min(dur, total)) start = min(start, max_start) clips_data.append( { "order": c.order if c.order is not None else i, "asset_id": c.asset_id or "", "start_time": round(start, 3), "duration": dur, "clip_type": c.clip_type, "playback_speed": float(c.playback_speed or 1.0), "transition_effect": c.transition_effect, "transition_duration": float(c.transition_duration or 0.0), "text_content": c.text_content or "", "config": c.config or {}, } ) self.replace_all_clips_transactional(plan_id, clips_data) net = total_output_duration( target, transition_effects=[c.transition_effect for c in clips], transition_durations=[float(c.transition_duration or 0.0) for c in clips], ) try: plan.total_duration = net db = self._clip_repo.session db.commit() except Exception: db.rollback() logger.exception("回写 plan.total_duration 失败（不阻断）: plan_id=%s", plan_id) logger.info( "配音时长分配完成: plan=%s clips=%d voice=%.2fs 成片净时长=%.2fs", plan_id, len(clips), voice, net, ) return plan def ensure_variant_plans( self, source_plan_id: str, count: int, candidate_asset_ids: list[str], *, created_by_user_id: str = "", voice_durations: Optional[list[float]] = None, rng=None, ) -> list[str]: """确保批量 N 个变体各自拥有独立 plan（#1749 批量正式生成/预览共用）。 - 变体 0：clone 源 plan（不污染源 plan，片段独立可改），并按配音分配段长； - 变体 1..N-1：reselect_plan_for_variant 完整重跑选片（素材级去重）； - voice_durations：每个变体的配音时长（独立配音各自时长；统一配音同值）； 缺省/为 0 时不分配（段长保持骨架/模板值）。 Returns: plan_id 列表，长度 == count，index 即 variant_index。 """ import random as _random rng = rng or _random.Random() plan_ids: list[str] = [] # 变体 0：clone（片段结构同源 plan，起点重算），不污染源 plan plan0 = self.clone_plan_for_variant( source_plan_id, created_by_user_id=created_by_user_id, name_suffix="变体1", ) v0_voice = 0.0 if voice_durations and len(voice_durations) > 0: try: v0_voice = float(voice_durations[0] or 0.0) except (TypeError, ValueError): v0_voice = 0.0 if v0_voice > 0: try: self.apply_voice_duration_to_plan(plan0.id, v0_voice) except Exception: logger.exception("变体0 配音分配失败（不阻断）: plan=%s", plan0.id) plan_ids.append(plan0.id) # 变体 1..N-1：独立选片 for i in range(1, count): voice = 0.0 if voice_durations and i < len(voice_durations): try: voice = float(voice_durations[i] or 0.0) except (TypeError, ValueError): voice = 0.0 variant = self.reselect_plan_for_variant( source_plan_id, candidate_asset_ids, created_by_user_id=created_by_user_id, name_suffix=f"变体{i + 1}", voice_duration=voice, rng=rng, ) plan_ids.append(variant.id) # #1764：为每个变体生成独立节奏模板（让批量视频片段时长分布不同） from packages.domain.voice_duration_planner import RHYTHM_TEMPLATES, adapt_template_length clip_count = 0 if voice_durations and len(voice_durations) > 0: # 从源 plan 获取片段数 source_plan = self.get_plan(source_plan_id) if source_plan and hasattr(source_plan, "clips"): clip_count = len(list(source_plan.clips)) if source_plan.clips else 0 rhythm_templates_for_variants = [] if clip_count > 0: for idx in range(len(plan_ids)): # 每个变体用不同的 seed 选择节奏模板 variant_seed = rng.randint(0, 999999) template = adapt_template_length(RHYTHM_TEMPLATES[variant_seed % len(RHYTHM_TEMPLATES)], clip_count) rhythm_templates_for_variants.append(template) logger.info("变体 %d 节奏模板: plan=%s template=%s", idx, plan_ids[idx], template) # #1767：BGM 池差异化分配（让批量变体使用不同 BGM / 段落 / 音量） from packages.domain.bgm_pool import allocate_bgm_pool_for_variants source_bgm_config = {} source_plan = self.get_plan(source_plan_id) if source_plan and source_plan.config: source_bgm_config = source_plan.config.get("bgm", {}) or {} variant_seeds_for_bgm = [rng.randint(0, 999999) for _ in plan_ids] bgm_pool_assignments = allocate_bgm_pool_for_variants(source_bgm_config, variant_seeds_for_bgm) # 为每个变体生成独立视觉扰动参数（让批量视频画面本身更不同） from packages.domain.variant_plan_selector import generate_visual_perturbation for idx, pid in enumerate(plan_ids): try: perturbation = generate_visual_perturbation(rng) # 变体 0 不做 hflip（保持预览 plan 原始画面方向） if idx == 0: perturbation["hflip"] = False config_update = {"visual_perturbation": perturbation} # #1764：写入节奏模板 if idx < len(rhythm_templates_for_variants): config_update["rhythm_template"] = rhythm_templates_for_variants[idx] # #1765：写入像素级扰动滤镜 from packages.domain.variant_plan_selector import generate_pixel_perturbation pixel_pert = generate_pixel_perturbation(rng) config_update["pixel_perturbation"] = pixel_pert # #1767：写入 BGM 池分配（覆盖 bgm 配置中的 preset_id / audio_offset / volume_adjust_db） if idx < len(bgm_pool_assignments): existing_bgm = dict((source_plan.config or {}).get("bgm", {}) or {}) existing_bgm.update(bgm_pool_assignments[idx]) config_update["bgm"] = existing_bgm self.update_plan_config(pid, config_update) logger.info( "变体 %d 视觉扰动+像素扰动+BGM池: plan=%s vis=%s pix=%s bgm=%s", idx, pid, perturbation, pixel_pert, bgm_pool_assignments[idx] if idx < len(bgm_pool_assignments) else None, ) except Exception: logger.exception("变体 %d 视觉扰动生成失败（不阻断）: plan=%s", idx, pid) # 标记所有变体 plan 的 clips 为 ready（已分配素材+起点，语义上就是 ready） for pid in plan_ids: try: self.mark_clips_ready(pid) except Exception: logger.exception("标记 clips ready 失败（不阻断）: plan=%s", pid) return plan_ids # ── 片段分割与合并 ────────────────────────────────────────────────────── def split_clip(self, clip_id: str, split_time: float) -> Dict[str, Any]: """将一个片段从指定位置分割为两个片段 Args: clip_id: 要分割的片段 ID split_time: 分割点（相对于片段起始的秒数），必须在 (0, duration) 范围内 Returns: dict: {"left_clip": EditPlanClip, "right_clip": EditPlanClip} Raises: ValueError: 片段不存在、分割时间越界 """ clip = self.get_clip_or_raise(clip_id) plan_id = clip.plan_id # 纯逻辑：校验 + 计算 _validate_split(split_time, clip.duration) split = _calc_split( duration=clip.duration, split_time=split_time, start_time=clip.start_time, ) self._auto_resume_editing(plan_id) original_order = clip.order # 更新左半部分（原片段） clip.duration = split.left_duration left_clip = self._clip_repo.update(clip) # 后面片段的 order 全部 +1（给右半部分腾位置） all_clips = self._clip_repo.list_by_plan(plan_id) shifts = _calc_shift_orders( all_clips, threshold_order=original_order, shift=1, excluded_ids={clip_id}, id_attr="id", order_attr="order", ) for c, new_order in shifts: c.order = new_order self._clip_repo.update(c) # 创建右半部分新片段（继承原片段的大部分属性） right_config = dict(clip.config) if clip.config else {} # 素材裁剪信息 if clip.asset_id: # 右半部分从 split_time 开始播放 right_config["trim_start"] = split.right_trim_start # 左半部分在 split_time 处结束 left_config = dict(left_clip.config) if left_clip.config else {} left_config["trim_end"] = split.left_trim_end left_clip.config = left_config left_clip = self._clip_repo.update(left_clip) right_clip = EditPlanClip.create( plan_id=plan_id, clip_type=clip.clip_type, order=original_order + 1, template_clip_config_id=clip.template_clip_config_id, asset_id=clip.asset_id, text_content=clip.text_content, start_time=split.right_start_time, duration=split.right_duration, transition_effect=clip.transition_effect, transition_duration=clip.transition_duration, playback_speed=clip.playback_speed, config=right_config, ) created_right = self._clip_repo.create(right_clip) logger.info( "分割片段: clip_id=%s plan_id=%s split_time=%.3fs left_dur=%.3fs right_dur=%.3fs", clip_id, plan_id, split_time, split.left_duration, split.right_duration, ) return { "left_clip": left_clip, "right_clip": created_right, } def merge_clips(self, clip_ids: List[str]) -> EditPlanClip: """合并多个连续片段为一个片段 Args: clip_ids: 要合并的片段 ID 列表（至少2个），必须属于同一个计划且 order 连续 Returns: EditPlanClip: 合并后的新片段 Raises: ValueError: 数量不足、不属于同一计划、不连续、类型不一致 """ if len(clip_ids) < 2: raise ValueError("至少需要 2 个片段才能合并") # 读取所有片段 clips = [] for cid in clip_ids: clip = self.get_clip_or_raise(cid) clips.append(clip) # 纯逻辑：校验 + 计算 plan_id, first_order = _validate_merge(clips) merge = _calc_merge(clips) self._auto_resume_editing(plan_id) # 更新第一个片段（保留它作为合并结果） first_clip = sorted(clips, key=lambda c: c.order)[0] first_clip.duration = merge.total_duration first_clip.text_content = merge.merged_text first_clip.config = merge.merged_config # 转场保留第一个的（合并后的入点转场） # playback_speed 取第一个的 merged_clip = self._clip_repo.update(first_clip) # 删除其余片段 rest_ids = [c.id for c in clips if c.id != merged_clip.id] for cid in rest_ids: self._clip_repo.delete(cid) # 后面的片段 order 前移 (len - 1) 位 all_clips = self._clip_repo.list_by_plan(plan_id) shifts = _calc_shift_orders( all_clips, threshold_order=first_order, shift=-merge.shift_amount, excluded_ids={merged_clip.id}, id_attr="id", order_attr="order", ) for c, new_order in shifts: c.order = new_order self._clip_repo.update(c) logger.info( "合并片段: plan_id=%s count=%d total_duration=%.3fs", plan_id, len(clips), merge.total_duration, ) return merged_clip # ── 渲染生成流程 ──────────────────────────────────────────────────────── def get_generation_status(self, plan_id: str) -> Dict[str, Any]: """获取渲染进度状态 Returns: dict: { "plan": EditPlan, "clips": List[EditPlanClip], "generation_task_id": Optional[str], "generation_task_status": Optional[str], "progress": float, "error_message": str, } Raises: ValueError: 计划不存在 """ plan = self.get_plan_or_raise(plan_id) clips = self._clip_repo.list_by_plan(plan_id) # 从 plan.config 中获取 generation_task_id generation_task_id = plan.config.get("generation_task_id") generation_task_status = None progress = 0.0 error_message = "" if generation_task_id: task = self._generation_task_repo.get(generation_task_id) if task: generation_task_status = task.status.value if hasattr(task.status, "value") else task.status progress = getattr(task, "progress", 0.0) or 0.0 error_message = getattr(task, "error_message", "") or "" return { "plan": plan, "clips": clips, "generation_task_id": generation_task_id, "generation_task_status": generation_task_status, "progress": progress, "error_message": error_message, } def can_generate(self, plan_id: str) -> tuple[bool, str]: """检查是否可以触发渲染 包含最后一道防线的自动修复： - 如果 clips 存在但都没有 asset_id，且 config.asset_ids 非空， 直接在内部执行素材分配，不再依赖前置 fallback 链路。 Returns: tuple: (can_generate, reason) """ plan = self.get_plan_or_raise(plan_id) # 检查状态 if plan.status != EditPlanStatus.EDITING: return False, "请先编辑并保存模板后再生成视频" # 检查是否有片段 clips = self._clip_repo.list_by_plan(plan_id) if not clips: return False, "请先添加片段后再生成视频" # 检查是否至少有一个片段分配了素材 has_asset = any(c.asset_id for c in clips) config_asset_ids_count = len((plan.config or {}).get("asset_ids", [])) clips_with_asset_count = sum(1 for c in clips if c.asset_id) logger.info( "can_generate 诊断: plan=%s status=%s total_clips=%d clips_with_asset=%d config_asset_ids_count=%d", plan_id, plan.status, len(clips), clips_with_asset_count, config_asset_ids_count, ) if not has_asset: # ── 最后防线：自动从 config.asset_ids 分配素材 ── config_asset_ids = (plan.config or {}).get("asset_ids", []) if config_asset_ids: logger.warning( "can_generate 最后防线触发: plan=%s clips=%d 均无素材，从 config.asset_ids(%d个) 自动分配", plan_id, len(clips), len(config_asset_ids), ) clips_without_asset = [c for c in clips if not c.asset_id] assigned_count = 0 for i, clip in enumerate(clips_without_asset): asset_idx = i % len(config_asset_ids) try: self.assign_asset(clip.id, config_asset_ids[asset_idx]) assigned_count += 1 except Exception as exc: logger.warning( "can_generate 最后防线: plan=%s clip=%s 分配素材 %s 失败: %s", plan_id, clip.id, config_asset_ids[asset_idx], exc, ) logger.info( "can_generate 最后防线: plan=%s 已为 %d/%d 个片段分配素材", plan_id, assigned_count, len(clips_without_asset), ) # 重新加载 clips 验证分配结果 clips = self._clip_repo.list_by_plan(plan_id) if not any(c.asset_id for c in clips): return False, "没有可渲染的就绪片段，自动修复后仍未分配素材" else: logger.warning( "can_generate 失败: plan=%s clips=%d 均无素材，且 config.asset_ids 为空，无法自动修复", plan_id, len(clips), ) return False, "没有可渲染的就绪片段，请确保已选择素材" return True, "" def mark_clips_ready(self, plan_id: str) -> int: """将已分配素材的 pending 片段标记为 ready 只标记同时满足以下条件的片段： - status == PENDING - asset_id 非空（已分配素材） Returns: int: 标记的片段数量 """ clips = self._clip_repo.list_by_plan( plan_id, status=EditPlanClipStatus.PENDING, ) count = 0 for clip in clips: if clip.asset_id: clip.mark_ready() self._clip_repo.update(clip) count += 1 logger.info( "标记片段就绪: plan_id=%s marked=%d total_pending=%d", plan_id, count, len(clips), ) return count def update_plan_config(self, plan_id: str, config_updates: Dict[str, Any]) -> EditPlan: """更新计划配置（合并更新） Args: plan_id: 计划 ID config_updates: 要合并的配置 Returns: 更新后的计划 """ plan = self.get_plan_or_raise(plan_id) # 自动从 completed/failed 切回 editing self._auto_resume_editing(plan_id) plan = self.get_plan_or_raise(plan_id) new_config = {**plan.config, **config_updates} updated = EditPlan( id=plan.id, template_id=plan.template_id, name=plan.name, status=plan.status, total_duration=plan.total_duration, source_edit_plan_id=plan.source_edit_plan_id, project_id=plan.project_id, created_by_user_id=plan.created_by_user_id, config=new_config, created_at=plan.created_at, updated_at=plan.updated_at, ) return self._plan_repo.update(updated)