144 lines
6.0 KiB
Python
144 lines
6.0 KiB
Python
from dataclasses import dataclass
|
|
|
|
from django.utils import timezone
|
|
|
|
from apps.projects.models import Project, ProjectStage, ScriptVersion, VideoSegment
|
|
|
|
|
|
STAGE_ORDER = [
|
|
ProjectStage.Stage.SCRIPT,
|
|
ProjectStage.Stage.BASE_ASSETS,
|
|
ProjectStage.Stage.STORYBOARD,
|
|
ProjectStage.Stage.VIDEO,
|
|
ProjectStage.Stage.EXPORT,
|
|
]
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class StageTransition:
|
|
current: str
|
|
target: str
|
|
allowed: bool
|
|
reason: str = ""
|
|
|
|
|
|
def can_enter_stage(current_stage: str, target_stage: str, allow_skip_storyboard: bool = True) -> StageTransition:
|
|
if target_stage not in STAGE_ORDER:
|
|
return StageTransition(current_stage, target_stage, False, "unknown target stage")
|
|
|
|
current_index = STAGE_ORDER.index(current_stage) if current_stage in STAGE_ORDER else -1
|
|
target_index = STAGE_ORDER.index(target_stage)
|
|
|
|
if target_index <= current_index + 1:
|
|
return StageTransition(current_stage, target_stage, True)
|
|
|
|
if allow_skip_storyboard and current_stage == ProjectStage.Stage.BASE_ASSETS and target_stage == ProjectStage.Stage.VIDEO:
|
|
return StageTransition(current_stage, target_stage, True)
|
|
|
|
return StageTransition(current_stage, target_stage, False, "stage prerequisite is not satisfied")
|
|
|
|
|
|
def initialize_project_pipeline(project: Project, *, placeholder_segments: int = 4) -> None:
|
|
"""建立专业创作和极速成片共用的阶段、视频占位数据。幂等，可安全重试。"""
|
|
for stage_name in STAGE_ORDER:
|
|
ProjectStage.objects.get_or_create(project=project, stage=stage_name)
|
|
for index in range(placeholder_segments):
|
|
VideoSegment.objects.get_or_create(
|
|
project=project,
|
|
sort_order=index,
|
|
defaults={"target_duration_seconds": 15},
|
|
)
|
|
|
|
|
|
def sync_video_segments_to_script(project: Project, script: ScriptVersion) -> None:
|
|
"""把视频段数量和时长收口到采用脚本；已生成过的段绝不裁剪或改时长。"""
|
|
if not script.is_adopted:
|
|
return
|
|
script_segments = list(script.segments.order_by("sort_order"))
|
|
target = len(script_segments)
|
|
segments = list(project.video_segments.order_by("sort_order"))
|
|
while len(segments) > target:
|
|
tail = segments[-1]
|
|
if tail.status == VideoSegment.Status.NOT_STARTED and not tail.versions.exists():
|
|
tail.delete()
|
|
segments.pop()
|
|
else:
|
|
break
|
|
|
|
next_order = (segments[-1].sort_order + 1) if segments else 0
|
|
for _ in range(target - len(segments)):
|
|
index = len(segments)
|
|
seconds = script_segments[index].duration_seconds if index < target else 15
|
|
segments.append(
|
|
VideoSegment.objects.create(
|
|
project=project,
|
|
sort_order=next_order,
|
|
target_duration_seconds=seconds,
|
|
)
|
|
)
|
|
next_order += 1
|
|
|
|
stale: list[VideoSegment] = []
|
|
for index, video_segment in enumerate(segments):
|
|
if index >= target:
|
|
break
|
|
seconds = script_segments[index].duration_seconds
|
|
if not seconds or video_segment.target_duration_seconds == seconds:
|
|
continue
|
|
if video_segment.status == VideoSegment.Status.SUCCEEDED or video_segment.versions.exists():
|
|
continue
|
|
video_segment.target_duration_seconds = seconds
|
|
stale.append(video_segment)
|
|
if stale:
|
|
VideoSegment.objects.bulk_update(stale, ["target_duration_seconds"])
|
|
|
|
|
|
def adopt_script_version(project: Project, script: ScriptVersion) -> None:
|
|
"""采用脚本并推进到资产阶段，供专业创作按钮和极速编排共同调用。"""
|
|
ScriptVersion.objects.filter(project=project).exclude(id=script.id).update(is_adopted=False)
|
|
if not script.is_adopted:
|
|
script.is_adopted = True
|
|
script.save(update_fields=["is_adopted", "updated_at"])
|
|
sync_video_segments_to_script(project, script)
|
|
stage, _ = ProjectStage.objects.get_or_create(project=project, stage=ProjectStage.Stage.SCRIPT)
|
|
stage.status = ProjectStage.Status.SUCCEEDED
|
|
stage.completed_at = timezone.now()
|
|
stage.error_message = ""
|
|
stage.save(update_fields=["status", "completed_at", "error_message", "updated_at"])
|
|
project.current_stage = ProjectStage.Stage.BASE_ASSETS
|
|
project.status = Project.Status.ASSETING
|
|
project.failure_reason = ""
|
|
project.save(update_fields=["current_stage", "status", "failure_reason", "updated_at"])
|
|
|
|
|
|
def finish_storyboard_stage(project: Project) -> None:
|
|
"""故事板全部成功后推进到视频阶段，并再次校准片段数量。"""
|
|
stage, _ = ProjectStage.objects.get_or_create(project=project, stage=ProjectStage.Stage.STORYBOARD)
|
|
stage.status = ProjectStage.Status.SUCCEEDED
|
|
stage.completed_at = timezone.now()
|
|
stage.error_message = ""
|
|
stage.save(update_fields=["status", "completed_at", "error_message", "updated_at"])
|
|
adopted_script = project.script_versions.filter(is_adopted=True).order_by("-created_at").first()
|
|
if adopted_script is not None:
|
|
sync_video_segments_to_script(project, adopted_script)
|
|
project.current_stage = ProjectStage.Stage.VIDEO
|
|
project.status = Project.Status.VIDEOING
|
|
project.save(update_fields=["current_stage", "status", "updated_at"])
|
|
|
|
|
|
def finish_video_stage(project: Project) -> bool:
|
|
"""全部片段成功即完成专业创作的视频阶段。"""
|
|
segments = list(project.video_segments.values_list("status", "adopted_version_id"))
|
|
if not segments or not all(status == VideoSegment.Status.SUCCEEDED and adopted for status, adopted in segments):
|
|
return False
|
|
stage, _ = ProjectStage.objects.get_or_create(project=project, stage=ProjectStage.Stage.VIDEO)
|
|
stage.status = ProjectStage.Status.SUCCEEDED
|
|
stage.completed_at = timezone.now()
|
|
stage.error_message = ""
|
|
stage.save(update_fields=["status", "completed_at", "error_message", "updated_at"])
|
|
project.current_stage = ProjectStage.Stage.VIDEO
|
|
project.status = Project.Status.COMPLETED
|
|
project.failure_reason = ""
|
|
project.save(update_fields=["current_stage", "status", "failure_reason", "updated_at"])
|
|
return True
|