"""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 = "变体", 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 = [ { "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 # 素材池 = 源 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, 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 # ── 片段分割与合并 ────────────────────────────────────────────────────── 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)