0469272bd6
CI/CD Pipeline / Check if frontend-only change (push) Has been skipped
CI/CD Pipeline / Check push changed paths (pull_request) Has been skipped
CI/CD Pipeline / Dedup Check - skip PR tests when covered by push pipeline (push) Successful in 1s
CI/CD Pipeline / PR Build API Image (push) Has been skipped
CI/CD Pipeline / Dedup Check - skip PR tests when covered by push pipeline (pull_request) Successful in 2s
CI/CD Pipeline / Build Staging API Image (pull_request) Has been skipped
CI/CD Pipeline / Build Staging Worker Image (pull_request) Has been skipped
CI/CD Pipeline / Check if frontend-only change (pull_request) Successful in 2s
CI/CD Pipeline / Build Staging Web Image (pull_request) Has been skipped
CI/CD Pipeline / Frontend Lint (push) Has been skipped
CI/CD Pipeline / Validate - Security (pull_request) Has been skipped
CI/CD Pipeline / Validate - Python (mypy + alembic) (pull_request) Has been skipped
CI/CD Pipeline / Validate - Style (pull_request) Has been skipped
CI/CD Pipeline / Unit Tests (pull_request) Has been skipped
CI/CD Pipeline / Integration Tests (pull_request) Has been skipped
CI/CD Pipeline / Frontend Lint (pull_request) Has been skipped
CI/CD Pipeline / Frontend Unit Tests (pull_request) Has been skipped
CI/CD Pipeline / PR Build Web Image (push) Has been skipped
CI/CD Pipeline / PR Build Worker Image (push) Has been skipped
CI/CD Pipeline / Check push changed paths (push) Successful in 9s
CI/CD Pipeline / Retag skipped Staging API Image (pull_request) Has been skipped
CI/CD Pipeline / Retag skipped Staging Web Image (pull_request) Has been skipped
CI/CD Pipeline / Retag skipped Staging Worker Image (pull_request) Has been skipped
CI/CD Pipeline / PR Build API Image (pull_request) Successful in 22s
CI/CD Pipeline / Build Production API Image (pull_request) Has been skipped
CI/CD Pipeline / Build Production Web Image (pull_request) Has been skipped
CI/CD Pipeline / Build Production Worker Image (pull_request) Has been skipped
CI/CD Pipeline / Build Staging API Image (push) Successful in 38s
CI/CD Pipeline / Deploy Staging (Watchtower auto-deploy) (pull_request) Has been skipped
CI/CD Pipeline / Deploy Production (pull_request) Has been skipped
CI/CD Pipeline / Production Browser E2E (pull_request) Has been skipped
CI/CD Pipeline / Staging E2E Tests (pull_request) Has been skipped
CI/CD Pipeline / Validate - Python (mypy + alembic) (push) Successful in 1m59s
CI/CD Pipeline / Staging API Integration Tests (pull_request) Has been skipped
CI/CD Pipeline / ACR Image Cleanup (pull_request) Has been skipped
CI/CD Pipeline / Build Staging Worker Image (push) Successful in 33s
CI/CD Pipeline / Canary Release to Production (pull_request) Has been skipped
CI/CD Pipeline / PR Build Worker Image (pull_request) Successful in 2m23s
Preview Deploy / Deploy Preview Environment (pull_request) Successful in 2m24s
CI/CD Pipeline / Integration Tests (push) Successful in 2m40s
CI/CD Pipeline / PR Build Web Image (pull_request) Successful in 2m31s
CI/CD Pipeline / Build Staging Web Image (push) Successful in 1m16s
CI/CD Pipeline / Retag skipped Staging API Image (push) Has been skipped
CI/CD Pipeline / Retag skipped Staging Web Image (push) Has been skipped
CI/CD Pipeline / Retag skipped Staging Worker Image (push) Has been skipped
CI/CD Pipeline / CI Gate (pull_request) Successful in 3s
CI/CD Pipeline / Validate - Style (push) Successful in 3m6s
CI/CD Pipeline / Deploy Staging (Watchtower auto-deploy) (push) Successful in 47s
PR Automation / Auto Approve on CI Green (pull_request) Successful in 3m42s
PR Automation / Auto Merge on CI Green + Approved (pull_request) Has been skipped
CI/CD Pipeline / ACR Image Cleanup (push) Successful in 1m48s
CI/CD Pipeline / Validate - Security (push) Successful in 5m28s
AI Code Review / AI Code Review (pull_request) Failing after 5m35s
CI/CD Pipeline / Frontend Unit Tests (push) Successful in 6m5s
CI/CD Pipeline / Staging API Integration Tests (push) Successful in 4m22s
CI/CD Pipeline / Staging E2E Tests (push) Failing after 5m6s
CI/CD Pipeline / Unit Tests (push) Successful in 8m54s
CI/CD Pipeline / Build Production API Image (push) Has been skipped
CI/CD Pipeline / Build Production Web Image (push) Has been skipped
CI/CD Pipeline / Build Production Worker Image (push) Has been skipped
CI/CD Pipeline / CI Gate (push) Has been skipped
CI/CD Pipeline / Deploy Production (push) Has been skipped
CI/CD Pipeline / Production Browser E2E (push) Has been skipped
CI/CD Pipeline / Canary Release to Production (push) Has been skipped
Co-authored-by: xiaoxia <dev@xiaoxiajianji.com> Co-committed-by: xiaoxia <dev@xiaoxiajianji.com>
1040 lines
39 KiB
Python
Executable File
1040 lines
39 KiB
Python
Executable File
"""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)
|