import logging from pathlib import Path import uuid from django.db import transaction from django.db.models import Count from django.http import JsonResponse, StreamingHttpResponse from rest_framework import status from rest_framework.decorators import action from rest_framework.parsers import FormParser, MultiPartParser from rest_framework.renderers import BaseRenderer from rest_framework.response import Response from rest_framework.viewsets import ModelViewSet from apps.ai.models import AITask, ModelConfig from apps.ai.providers import TtsNotConfigured from apps.ai.script_agent import stream_script_agent from apps.ai.services import ( DEFAULT_VOICEOVER_VOICE, VOICEOVER_VOICES, create_export_job, generate_base_asset, generate_person_triview, generate_storyboard_frame, get_default_model, poll_video_segment, regenerate_script_segment, submit_storyboard, submit_video_segment, synthesize_project_voiceover, ) from apps.assets.models import Asset, AssetFile from apps.assets.serializers import AssetFileSerializer from apps.assets.storage import TosStorage from apps.common.api import TeamScopedViewSetMixin from apps.common.celery_health import require_worker from .models import ( BaseAssetGroup, BgmTrack, ExportJob, Project, ProjectStage, ScriptSegment, ScriptVersion, StoryboardFrame, StoryboardVersion, SubtitleTrack, Timeline, TimelineClip, VideoSegment, VideoSegmentVersion, ) from .serializers import ( BaseAssetGroupSerializer, ExportJobSerializer, ProjectListSerializer, ProjectSerializer, ScriptVersionSerializer, StoryboardVersionSerializer, VideoSegmentVersionSerializer, ) from .services.export import run_export_job_in_thread from .services.pipeline import STAGE_ORDER from .tasks import poll_video_segment_task logger = logging.getLogger(__name__) class ServerSentEventRenderer(BaseRenderer): """让 DRF 内容协商接受 Accept: text/event-stream(否则流式端点直接 406)。 实际响应由视图返回 StreamingHttpResponse 直接下发,这个 renderer 只用于通过协商。""" media_type = "text/event-stream" format = "event-stream" charset = None def render(self, data, accepted_media_type=None, renderer_context=None): return data def _store_uploaded_asset(*, team, user, upload, asset_type: str, category: str, name: str) -> Asset: """把上传的文件落到 TOS,建 Asset+AssetFile(主文件)。供上传视频段 / 上传 BGM 复用。""" suffix = Path(upload.name).suffix.lower() or (".mp4" if asset_type == Asset.Type.VIDEO else ".mp3") asset_id = uuid.uuid4() object_key = f"teams/{team.id}/uploads/{asset_id}{suffix}" stored = TosStorage().upload_fileobj( fileobj=upload.file, object_key=object_key, content_type=upload.content_type or "application/octet-stream", ) asset = Asset.objects.create( id=asset_id, team=team, created_by=user, name=name, asset_type=asset_type, source=Asset.Source.UPLOAD, category=category, ) AssetFile.objects.create( asset=asset, object_key=stored.object_key, bucket=stored.bucket, content_type=stored.content_type, size_bytes=stored.size_bytes, is_primary=True, ) return asset def promote_base_asset_stage_if_ready(project: Project) -> bool: adopted_kind_count = ( project.base_asset_groups.filter(adopted_asset__isnull=False).values("kind").distinct().count() ) if adopted_kind_count < 3: return False stage, _ = ProjectStage.objects.get_or_create(project=project, stage=ProjectStage.Stage.BASE_ASSETS) stage.status = ProjectStage.Status.SUCCEEDED stage.save(update_fields=["status", "updated_at"]) project.current_stage = ProjectStage.Stage.STORYBOARD project.status = Project.Status.STORYBOARDING project.save(update_fields=["current_stage", "status", "updated_at"]) return True class ProjectViewSet(TeamScopedViewSetMixin, ModelViewSet): queryset = Project.objects.select_related("product", "timeline").prefetch_related( "stages", "video_segments", "video_segments__adopted_version__asset__files", "video_segments__versions__asset__files", "script_versions", "script_versions__segments", "base_asset_groups", "base_asset_groups__adopted_asset__files", "base_asset_groups__candidate_assets__files", "storyboard_versions", "storyboard_versions__frames__asset__files", "timeline__clips__asset__files", ).all() serializer_class = ProjectSerializer search_fields = ["name", "product__title"] ordering_fields = ["created_at", "updated_at", "name"] def get_serializer_class(self): # 列表用轻量序列化(只给列表/仪表盘/侧栏要的字段);详情/写操作用全量 ProjectSerializer return ProjectListSerializer if self.action == "list" else ProjectSerializer def get_queryset(self): # 列表:轻查询(只 select_related product + 计数注解),不做详情那串 12 个重 prefetch # ——原列表把每个项目的 阶段/片段/故事板/时间线/资产文件全拉出,20 个项目实测 ~2s。 if self.action == "list": qs = ( Project.objects.select_related("product", "product__cover_asset") .prefetch_related("product__cover_asset__files") .annotate( script_version_count=Count("script_versions", distinct=True), video_segment_count=Count("video_segments", distinct=True), ) ) return qs.filter(**{self.team_field: self.get_team()}).order_by("-updated_at") return super().get_queryset() @action(detail=False, methods=["get"]) def summary(self, request): """项目统计(总数 + 各状态计数),供仪表盘/侧栏徽标——只跑 COUNT,不拉项目对象。""" base = Project.objects.filter(team=self.get_team()) by_status = {row["status"]: row["n"] for row in base.values("status").annotate(n=Count("id"))} return Response({"total": base.count(), "by_status": by_status}) @transaction.atomic def perform_create(self, serializer): project = serializer.save(team=self.get_team(), created_by=self.request.user) for stage in STAGE_ORDER: ProjectStage.objects.create(project=project, stage=stage) for index in range(4): VideoSegment.objects.create(project=project, sort_order=index, target_duration_seconds=15) @action(detail=True, methods=["post"], url_path="script-agent-stream", renderer_classes=[ServerSentEventRenderer]) def script_agent_stream(self, request, pk=None): """对话式脚本 agent · 流式(SSE)。出稿 + 改稿一体,多模型可选。 请求体:mode(auto|theme|revise)、prompt、model_config_id、selling_point_ids、 base_version_id(改稿)、aspect_ratio、total_duration。 响应:text/event-stream,逐帧吐 tool/delta/draft/saved/done/error。""" project = self.get_object() mode = str(request.data.get("mode") or "auto") prompt = str(request.data.get("prompt") or "") selling_point_ids = request.data.get("selling_point_ids") or [] base_version_id = request.data.get("base_version_id") or None aspect_ratio = str(request.data.get("aspect_ratio") or "9:16") try: total_duration = int(request.data.get("total_duration") or 60) except (TypeError, ValueError): total_duration = 60 target_index = request.data.get("target_index") try: target_index = int(target_index) if target_index is not None else None except (TypeError, ValueError): target_index = None model_config = None requested = request.data.get("model_config_id") if requested: model_config = ( ModelConfig.objects.select_related("provider") .filter(id=requested, capability=ModelConfig.Capability.TEXT, status=ModelConfig.Status.ACTIVE) .first() ) if model_config is None: model_config = get_default_model(ModelConfig.Capability.TEXT) if model_config is None: # 纯 Django 响应:绕开 DRF 渲染(此 action 只挂了 SSE renderer) return JsonResponse({"detail": "没有可用的文本模型,请先在模型库配置"}, status=400) stream = stream_script_agent( project=project, user=request.user, model_config=model_config, mode=mode, user_prompt=prompt, selling_point_ids=selling_point_ids, base_version_id=base_version_id, aspect_ratio=aspect_ratio, total_duration=total_duration, target_index=target_index, ) response = StreamingHttpResponse(stream, content_type="text/event-stream") response["Cache-Control"] = "no-cache" response["X-Accel-Buffering"] = "no" # 关 nginx 缓冲,保证逐帧下发 return response @action(detail=True, methods=["post"], url_path="adopt-script") @transaction.atomic def adopt_script(self, request, pk=None): project = self.get_object() script_id = request.data.get("script_version_id") script = ScriptVersion.objects.select_for_update().get(project=project, id=script_id) ScriptVersion.objects.filter(project=project).update(is_adopted=False) script.is_adopted = True script.save(update_fields=["is_adopted", "updated_at"]) stage, _ = ProjectStage.objects.get_or_create(project=project, stage=ProjectStage.Stage.SCRIPT) stage.status = ProjectStage.Status.SUCCEEDED stage.save(update_fields=["status", "updated_at"]) project.current_stage = ProjectStage.Stage.BASE_ASSETS project.status = Project.Status.ASSETING project.save(update_fields=["current_stage", "status", "updated_at"]) return Response(ScriptVersionSerializer(script).data) @action(detail=True, methods=["post"], url_path="generate-base-asset") def generate_base_asset_action(self, request, pk=None): require_worker() project = self.get_object() kind = request.data.get("kind") if kind not in BaseAssetGroup.Kind.values: return Response({"detail": "invalid base asset kind"}, status=status.HTTP_400_BAD_REQUEST) try: task = generate_base_asset(project=project, user=request.user, kind=kind, prompt=request.data.get("prompt", ""), label=request.data.get("label", "")) except ValueError as exc: # 无可用模型 / 余额不足等,立即反馈 return Response({"detail": str(exc)}, status=status.HTTP_400_BAD_REQUEST) stage, _ = ProjectStage.objects.get_or_create(project=project, stage=ProjectStage.Stage.BASE_ASSETS) stage.status = ProjectStage.Status.NEEDS_REVIEW stage.save(update_fields=["status", "updated_at"]) # 异步:出图在 worker 里跑,秒回 RESERVED 任务,前端轮询 /api/ai/generate-image/?ids=… 取结果后刷新项目 return Response({"task": {"id": str(task.id), "status": task.status}}, status=status.HTTP_202_ACCEPTED) @action(detail=True, methods=["get"], url_path="pending-assets") def pending_assets(self, request, pk=None): """本项目在途的基础资产出图任务(立绘/商品图/三视图)。前端进基础资产趴时据此重建 「生成中」占位卡 loading —— 出图在 worker 跑,刷新页面后内存态 genBusy 丢了,用它从后端认领恢复。""" project = self.get_object() inflight = [AITask.Status.CREATED, AITask.Status.RESERVED, AITask.Status.SUBMITTED, AITask.Status.POLLING] image_types = [AITask.Type.PRODUCT_IMAGE, AITask.Type.PERSON_IMAGE, AITask.Type.SCENE_IMAGE] tasks = AITask.objects.filter(project=project, task_type__in=image_types, status__in=inflight) pending = [] for task in tasks: payload = task.request_payload or {} pending.append({ "id": str(task.id), "kind": payload.get("kind") or "", "label": payload.get("label") or "", "is_triview": bool(payload.get("triview_of")), "status": task.status, }) return Response({"pending": pending}) @action(detail=True, methods=["post"], url_path="poll-reviews") def poll_reviews(self, request, pk=None): """轮询本团队真人资产的火山审核状态(绿盾 active / 红标 failed)。前端基础资产趴定时调,刷新徽章。""" project = self.get_object() from apps.assets.review import poll_team_reviews return Response({"reviews": poll_team_reviews(project.team)}) @action(detail=True, methods=["post"], url_path="adopt-base-asset") @transaction.atomic def adopt_base_asset(self, request, pk=None): project = self.get_object() group = BaseAssetGroup.objects.select_for_update().get(project=project, id=request.data.get("group_id")) asset_id = request.data.get("asset_id") if not group.candidate_assets.filter(id=asset_id).exists(): return Response({"detail": "asset is not a candidate"}, status=status.HTTP_400_BAD_REQUEST) group.adopted_asset_id = asset_id group.save(update_fields=["adopted_asset", "updated_at"]) promote_base_asset_stage_if_ready(project) return Response(BaseAssetGroupSerializer(group).data) @action(detail=True, methods=["post"], url_path="attach-base-asset") @transaction.atomic def attach_base_asset(self, request, pk=None): """流程步骤4 · 用演员库现有资产替换某基础资产卡:把团队内现有 Asset 挂为该组候选并采用。 seed 占位卡还没有 group,此时不传 group_id,改传 kind+label:按 label 命中同实体组,没有则据 label 建组再挂(不出图)。""" project = self.get_object() group = None group_id = request.data.get("group_id") if group_id: group = BaseAssetGroup.objects.select_for_update().filter(project=project, id=group_id).first() if group is None: return Response({"detail": "group not found"}, status=status.HTTP_404_NOT_FOUND) else: # 无 group(seed 占位卡):据 kind+label 复用/新建实体组,把所选演员资产挂成该 tag 的立绘 kind = request.data.get("kind") if kind not in BaseAssetGroup.Kind.values: return Response({"detail": "group_id or valid kind is required"}, status=status.HTTP_400_BAD_REQUEST) label = str(request.data.get("label") or "").strip() # 按 label 命中同实体组(排除三视图组);商品=该项目唯一非三视图组;都没有则据 label 新建 candidates = [g for g in project.base_asset_groups.filter(kind=kind).order_by("created_at") if not (g.metadata or {}).get("triview_of")] if kind == BaseAssetGroup.Kind.PRODUCT: group = candidates[0] if candidates else None elif label: group = next((g for g in candidates if (g.metadata or {}).get("label") == label), None) if group is None: group_meta = {"label": label} if label else {} group = BaseAssetGroup.objects.create(project=project, kind=kind, metadata=group_meta) group = BaseAssetGroup.objects.select_for_update().get(id=group.id) asset = Asset.objects.filter(team=project.team, id=request.data.get("asset_id")).first() if asset is None: return Response({"detail": "asset not found"}, status=status.HTTP_404_NOT_FOUND) group.candidate_assets.add(asset) group.adopted_asset_id = asset.id group.save(update_fields=["adopted_asset", "updated_at"]) promote_base_asset_stage_if_ready(project) return Response(BaseAssetGroupSerializer(group).data) @action(detail=True, methods=["post"], url_path="generate-triview") def generate_triview(self, request, pk=None): """流程步骤4 · 据某一版立绘资产生成它配套的三视图(image_edit 锁角色一致性,三视图绑定该立绘)。""" require_worker() # 异步出图依赖 worker 兜底执行,没 worker 直接拒绝(否则任务永远 RESERVED) project = self.get_object() portrait_asset = Asset.objects.filter(team=project.team, id=request.data.get("portrait_asset_id")).first() if portrait_asset is None: return Response({"detail": "portrait asset not found"}, status=status.HTTP_404_NOT_FOUND) try: task = generate_person_triview(project=project, user=request.user, portrait_asset=portrait_asset) except ValueError as exc: return Response({"detail": str(exc)}, status=status.HTTP_400_BAD_REQUEST) # 异步:image_edit 出图在 worker 里跑,秒回 RESERVED 任务,前端轮询取结果后刷新项目即见三视图 return Response({"task": {"id": str(task.id), "status": task.status}}, status=status.HTTP_202_ACCEPTED) # ── Stage 1 · 镜头脚本逐字段编辑 / 增删分镜 ── def _sync_video_segments_to_script(self, project: Project, script: ScriptVersion) -> None: """采用版分镜数变化时,同步 VideoSegment 数量:不足则尾部补 NOT_STARTED, 多出且尾部是「从未生成过」的段则裁掉(已生成的段绝不动)。""" if not script.is_adopted: return target = script.segments.count() 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)): segments.append(VideoSegment.objects.create(project=project, sort_order=next_order, target_duration_seconds=15)) next_order += 1 @action(detail=True, methods=["post"], url_path="update-script-segment") def update_script_segment(self, request, pk=None): project = self.get_object() segment = ScriptSegment.objects.get(id=request.data.get("segment_id"), script_version__project=project) changed = [] for field in ("narration", "visual_prompt"): if field in request.data: setattr(segment, field, str(request.data.get(field) or "").strip()) changed.append(field) if "duration_seconds" in request.data: try: segment.duration_seconds = max(1, min(60, int(request.data["duration_seconds"]))) changed.append("duration_seconds") except (TypeError, ValueError): pass if changed: segment.save(update_fields=[*changed, "updated_at"]) return Response(ScriptVersionSerializer(segment.script_version).data) @action(detail=True, methods=["post"], url_path="rerun-script-segment") def rerun_script_segment(self, request, pk=None): """单条分镜 AI 重跑:只重新生成该 ScriptSegment 的旁白+画面提示(其它镜不动),按现有计费扣费。""" project = self.get_object() segment_id = request.data.get("segment_id") if not segment_id: return Response({"detail": "segment_id is required"}, status=status.HTTP_400_BAD_REQUEST) try: segment = ScriptSegment.objects.get(id=segment_id, script_version__project=project) except ScriptSegment.DoesNotExist: return Response({"detail": "script segment not found"}, status=status.HTTP_404_NOT_FOUND) try: script = regenerate_script_segment( project=project, user=request.user, segment=segment, instruction=str(request.data.get("instruction") or "").strip(), ) except ValueError as exc: return Response({"detail": str(exc)}, status=status.HTTP_400_BAD_REQUEST) return Response(ScriptVersionSerializer(script).data) @action(detail=True, methods=["post"], url_path="add-script-segment") @transaction.atomic def add_script_segment(self, request, pk=None): project = self.get_object() after = ScriptSegment.objects.select_related("script_version").get( id=request.data.get("after_segment_id"), script_version__project=project ) script = after.script_version segments = list(script.segments.order_by("sort_order")) insert_at = next(i for i, s in enumerate(segments) if s.id == after.id) + 1 created = ScriptSegment.objects.create( script_version=script, sort_order=insert_at, duration_seconds=int(request.data.get("duration_seconds") or 15), narration=str(request.data.get("narration") or "").strip(), visual_prompt=str(request.data.get("visual_prompt") or "").strip(), ) segments.insert(insert_at, created) for index, seg in enumerate(segments): if seg.sort_order != index: seg.sort_order = index seg.save(update_fields=["sort_order", "updated_at"]) self._sync_video_segments_to_script(project, script) return Response(ScriptVersionSerializer(script).data, status=status.HTTP_201_CREATED) @action(detail=True, methods=["post"], url_path="delete-script-segment") @transaction.atomic def delete_script_segment(self, request, pk=None): project = self.get_object() segment = ScriptSegment.objects.select_related("script_version").get( id=request.data.get("segment_id"), script_version__project=project ) script = segment.script_version if script.segments.count() <= 1: return Response({"detail": "至少保留一个分镜"}, status=status.HTTP_400_BAD_REQUEST) # 删本场脚本时,它对应的故事板分镜帧也一起删 —— frame.script_segment 是 SET_NULL, # 不主动删会留下孤儿帧,故事板里仍显示这一(已删的)场。各版本受影响的帧统一收口后重排 sort_order。 affected_storyboards = list(StoryboardVersion.objects.filter(frames__script_segment=segment).distinct()) StoryboardFrame.objects.filter(script_segment=segment).delete() for sb in affected_storyboards: for index, frame in enumerate(sb.frames.order_by("sort_order")): if frame.sort_order != index: frame.sort_order = index frame.save(update_fields=["sort_order", "updated_at"]) segment.delete() for index, seg in enumerate(script.segments.order_by("sort_order")): if seg.sort_order != index: seg.sort_order = index seg.save(update_fields=["sort_order", "updated_at"]) self._sync_video_segments_to_script(project, script) return Response(ScriptVersionSerializer(script).data) # ── Stage 4 · 视频版本采用(详情弹窗里切历史版) ── @action(detail=True, methods=["post"], url_path="adopt-video-version") @transaction.atomic def adopt_video_version(self, request, pk=None): project = self.get_object() segment = VideoSegment.objects.get(project=project, id=request.data.get("video_segment_id")) version = VideoSegmentVersion.objects.get(video_segment=segment, id=request.data.get("version_id")) segment.versions.update(is_adopted=False) version.is_adopted = True version.save(update_fields=["is_adopted", "updated_at"]) segment.adopted_version = version segment.status = VideoSegment.Status.SUCCEEDED segment.error_message = "" segment.save(update_fields=["adopted_version", "status", "error_message", "updated_at"]) # 时间线草稿(剪辑台/拼接导出都读它)若还引用本场旧版本的资产,必须跟随切换, # 否则采用新版本后预览和导出仍是旧片段;裁剪点是对旧素材设的,一并复位 timeline = Timeline.objects.filter(project=project).first() if timeline is not None and version.asset_id: segment_asset_ids = list(segment.versions.values_list("asset_id", flat=True)) timeline.clips.filter(asset_id__in=segment_asset_ids).exclude(asset_id=version.asset_id).update( asset_id=version.asset_id, trim_start_ms=0, trim_end_ms=None ) return Response(ProjectSerializer(project).data) @action(detail=True, methods=["post"], url_path="generate-storyboard") def generate_storyboard_action(self, request, pk=None): """异步故事板·提交:快速创建版本(不在此生图、不推进阶段)。前端随后轮询 poll-storyboard 逐帧生成。""" require_worker() project = self.get_object() storyboard = submit_storyboard(project=project, user=request.user, prompt=request.data.get("prompt", "")) stage, _ = ProjectStage.objects.get_or_create(project=project, stage=ProjectStage.Stage.STORYBOARD) stage.status = ProjectStage.Status.RUNNING stage.save(update_fields=["status", "updated_at"]) return Response(StoryboardVersionSerializer(storyboard).data, status=status.HTTP_202_ACCEPTED) @action(detail=True, methods=["post"], url_path="poll-storyboard") def poll_storyboard_action(self, request, pk=None): """异步故事板·轮询:每次生成下一帧(单次 ARK 调用 ~20s)。全部完成 → 推进到 VIDEO 阶段。""" project = self.get_object() result = generate_storyboard_frame(project=project, user=request.user) if result.get("status") == "succeeded": stage, _ = ProjectStage.objects.get_or_create(project=project, stage=ProjectStage.Stage.STORYBOARD) stage.status = ProjectStage.Status.SUCCEEDED stage.save(update_fields=["status", "updated_at"]) project.current_stage = ProjectStage.Stage.VIDEO project.status = Project.Status.VIDEOING project.save(update_fields=["current_stage", "status", "updated_at"]) http_status = status.HTTP_200_OK if result.get("status") == "succeeded" else status.HTTP_202_ACCEPTED return Response(result, status=http_status) @action(detail=True, methods=["post"], url_path="skip-storyboard") @transaction.atomic def skip_storyboard(self, request, pk=None): project = self.get_object() stage, _ = ProjectStage.objects.get_or_create(project=project, stage=ProjectStage.Stage.STORYBOARD) stage.status = ProjectStage.Status.SKIPPED stage.save(update_fields=["status", "updated_at"]) project.current_stage = ProjectStage.Stage.VIDEO project.status = Project.Status.VIDEOING project.save(update_fields=["current_stage", "status", "updated_at"]) return Response(ProjectSerializer(project).data) @action(detail=True, methods=["post"], url_path="submit-video-segment") def submit_video_segment_action(self, request, pk=None): # 前置闸:无 worker 禁止提交——提交到 ARK 后若无人轮询,结果悬在云端、预扣额度冻结。 require_worker() project = self.get_object() segment = VideoSegment.objects.get(project=project, id=request.data.get("video_segment_id")) submit_video_segment(video_segment=segment, user=request.user, prompt=request.data.get("prompt", "")) # worker 在线由上面的闸保证;此处入队失败只剩极小窗口(刚提交完 broker 闪断), # 任务已在 ARK,前端轮询仍可兜底取回,故不让提交 500。 try: poll_video_segment_task.apply_async(args=[str(segment.id)], countdown=30) except Exception: # noqa: BLE001 logger.error("poll_video_segment_task enqueue failed; relying on client polling", exc_info=True) return Response(ProjectSerializer(project).data, status=status.HTTP_202_ACCEPTED) @action(detail=True, methods=["post"], url_path="poll-video-segment") def poll_video_segment_action(self, request, pk=None): project = self.get_object() segment = VideoSegment.objects.get(project=project, id=request.data.get("video_segment_id")) version = poll_video_segment(video_segment=segment, user=request.user) if version is None: return Response({"status": segment.status}, status=status.HTTP_202_ACCEPTED) return Response(VideoSegmentVersionSerializer(version).data) @action(detail=True, methods=["post"], url_path="submit-export") @transaction.atomic def submit_export(self, request, pk=None): project = self.get_object() missing_segments = project.video_segments.filter(adopted_version__isnull=True).count() if missing_segments: return Response( {"detail": f"{missing_segments} video segments are not ready for export"}, status=status.HTTP_400_BAD_REQUEST, ) timeline, _ = Timeline.objects.get_or_create( project=project, defaults={"name": f"{project.name} Timeline", "duration_seconds": 60}, ) if not timeline.clips.exists(): start_ms = 0 for segment in project.video_segments.select_related("adopted_version__asset").order_by("sort_order"): if segment.adopted_version_id: TimelineClip.objects.create( timeline=timeline, asset=segment.adopted_version.asset, sort_order=segment.sort_order, start_ms=start_ms, duration_ms=segment.target_duration_seconds * 1000, ) start_ms += segment.target_duration_seconds * 1000 # 清理僵尸任务:后台线程跑导出时若 Pod 被重启/OOM 杀掉,旧任务会永远停在 RUNNING、 # 前端无限轮询。新建前把本时间线超过 5 分钟仍未结束的旧任务标失败,避免界面一直转圈。 from django.utils import timezone from datetime import timedelta ExportJob.objects.filter( timeline=timeline, status__in=[ExportJob.Status.QUEUED, ExportJob.Status.RUNNING], updated_at__lt=timezone.now() - timedelta(minutes=5), ).update(status=ExportJob.Status.FAILED, error_message="任务中断(服务重启或超时),已自动失败,请重新导出") export_job = create_export_job(timeline=timeline, user=request.user) stage, _ = ProjectStage.objects.get_or_create(project=project, stage=ProjectStage.Stage.EXPORT) stage.status = ProjectStage.Status.RUNNING stage.save(update_fields=["status", "updated_at"]) project.current_stage = ProjectStage.Stage.EXPORT project.status = Project.Status.EXPORTING project.save(update_fields=["current_stage", "status", "updated_at"]) # 后台线程跑真实 ffmpeg 拼接(无需 Celery worker);前端轮询 poll-export 取进度/成片。 transaction.on_commit(lambda: run_export_job_in_thread(str(export_job.id))) return Response(ExportJobSerializer(export_job).data, status=status.HTTP_202_ACCEPTED) @action(detail=True, methods=["post", "get"], url_path="poll-export") def poll_export_action(self, request, pk=None): """拼接导出·轮询:回最新导出任务的状态/进度/成片 URL。成片就绪时把 EXPORT 阶段标记成功。""" project = self.get_object() timeline = getattr(project, "timeline", None) export_job = timeline.export_jobs.order_by("-created_at").first() if timeline is not None else None if export_job is None: return Response({"status": "not_started", "progress": 0, "output_url": ""}) output_url = "" output = export_job.output_asset if output is not None: primary = output.files.filter(is_primary=True).first() or output.files.first() if primary is not None: output_url = AssetFileSerializer().get_preview_url(primary) if export_job.status == ExportJob.Status.SUCCEEDED: stage, _ = ProjectStage.objects.get_or_create(project=project, stage=ProjectStage.Stage.EXPORT) if stage.status != ProjectStage.Status.SUCCEEDED: stage.status = ProjectStage.Status.SUCCEEDED stage.save(update_fields=["status", "updated_at"]) elif export_job.status == ExportJob.Status.FAILED: stage, _ = ProjectStage.objects.get_or_create(project=project, stage=ProjectStage.Stage.EXPORT) if stage.status != ProjectStage.Status.FAILED: stage.status = ProjectStage.Status.FAILED stage.error_message = export_job.error_message stage.save(update_fields=["status", "error_message", "updated_at"]) return Response({ "status": export_job.status, "progress": export_job.progress, "output_asset": str(output.id) if output is not None else None, "output_url": output_url, "error_message": export_job.error_message, }) @action(detail=True, methods=["post"], url_path="upload-video-segment", parser_classes=[MultiPartParser, FormParser]) @transaction.atomic def upload_video_segment_action(self, request, pk=None): """上传自带视频替换某段:落 TOS → Asset(video) → VideoSegmentVersion → 采用并标记完成。""" project = self.get_object() upload = request.data.get("file") if upload is None: return Response({"detail": "no file uploaded"}, status=status.HTTP_400_BAD_REQUEST) segment = VideoSegment.objects.filter(project=project, id=request.data.get("video_segment_id")).first() if segment is None: return Response({"detail": "video segment not found"}, status=status.HTTP_404_NOT_FOUND) asset = _store_uploaded_asset( team=project.team, user=request.user, upload=upload, asset_type=Asset.Type.VIDEO, category=Asset.Category.VIDEO_CLIP, name=f"{project.name}-上传-{segment.sort_order + 1}", ) version = VideoSegmentVersion.objects.create( video_segment=segment, asset=asset, prompt="用户上传", is_adopted=True, metadata={"source": "upload"}, ) segment.adopted_version = version segment.status = VideoSegment.Status.SUCCEEDED segment.error_message = "" segment.save(update_fields=["adopted_version", "status", "error_message", "updated_at"]) return Response(ProjectSerializer(project).data, status=status.HTTP_201_CREATED) @action(detail=True, methods=["post"], url_path="upload-bgm", parser_classes=[MultiPartParser, FormParser]) @transaction.atomic def upload_bgm_action(self, request, pk=None): """上传 BGM 音频:落 TOS → Asset(audio) → 设为 timeline 的(唯一)BGM 轨。""" project = self.get_object() upload = request.data.get("file") if upload is None: return Response({"detail": "no file uploaded"}, status=status.HTTP_400_BAD_REQUEST) timeline, _ = Timeline.objects.get_or_create( project=project, defaults={"name": f"{project.name} Timeline", "duration_seconds": 60} ) asset = _store_uploaded_asset( team=project.team, user=request.user, upload=upload, asset_type=Asset.Type.AUDIO, category=Asset.Category.UPLOAD, name=f"{project.name}-BGM", ) volume = int(request.data.get("volume") or 60) timeline.bgm_tracks.all().delete() BgmTrack.objects.create(timeline=timeline, asset=asset, volume=max(0, min(100, volume)), start_ms=0) return Response(ProjectSerializer(self.get_object()).data, status=status.HTTP_201_CREATED) # ── Stage 5 · 旁白配音(TTS):每镜旁白合成语音资产,导出时作为人声轨混在 BGM 之上 ── @action(detail=True, methods=["get", "post"], url_path="generate-voiceover") def generate_voiceover_action(self, request, pk=None): project = self.get_object() if request.method == "GET": return Response({"voices": VOICEOVER_VOICES, "default_voice": DEFAULT_VOICEOVER_VOICE}) try: voiceover = synthesize_project_voiceover( project=project, user=request.user, items=request.data.get("items") or [], voice_type=request.data.get("voice_type") or DEFAULT_VOICEOVER_VOICE, speed_ratio=float(request.data.get("speed_ratio") or 1.0), ) except (TtsNotConfigured, ValueError) as exc: return Response({"detail": str(exc)}, status=status.HTTP_400_BAD_REQUEST) return Response({"voiceover": voiceover}, status=status.HTTP_201_CREATED) @action(detail=True, methods=["post", "put"], url_path="save-timeline") @transaction.atomic def save_timeline_action(self, request, pk=None): """保存草稿:整体持久化时间轴编辑态(片段顺序/裁剪、字幕样式与内容、BGM 音量、转场、草稿元数据)。""" project = self.get_object() timeline, _ = Timeline.objects.get_or_create( project=project, defaults={"name": f"{project.name} Timeline", "duration_seconds": 60} ) data = request.data clips = data.get("clips") if isinstance(clips, list): valid_asset_ids = set( Asset.objects.filter(team=project.team, id__in=[c.get("asset") for c in clips if c.get("asset")]) .values_list("id", flat=True) ) timeline.clips.all().delete() start_ms = 0 for index, clip in enumerate(clips): asset_id = clip.get("asset") if not asset_id or str(asset_id) not in {str(a) for a in valid_asset_ids}: continue duration = int(clip.get("duration_ms") or 15000) TimelineClip.objects.create( timeline=timeline, asset_id=asset_id, sort_order=index, start_ms=start_ms, duration_ms=duration, trim_start_ms=int(clip.get("trim_start_ms") or 0), trim_end_ms=clip.get("trim_end_ms"), ) start_ms += duration timeline.duration_seconds = max(1, round(start_ms / 1000)) subtitle = data.get("subtitle") if isinstance(subtitle, dict): track = timeline.subtitle_tracks.first() or SubtitleTrack(timeline=timeline) track.enabled = bool(subtitle.get("enabled", True)) style = dict(track.style or {}) if subtitle.get("style_key"): style["key"] = subtitle["style_key"] track.style = style if isinstance(subtitle.get("content"), list): track.content = subtitle["content"] track.save() bgm = data.get("bgm") if isinstance(bgm, dict): track = timeline.bgm_tracks.first() if bgm.get("clear"): timeline.bgm_tracks.all().delete() elif track is not None and bgm.get("volume") is not None: track.volume = max(0, min(100, int(bgm["volume"]))) track.save(update_fields=["volume", "updated_at"]) metadata = dict(timeline.metadata or {}) if isinstance(data.get("transition"), dict): metadata["transition"] = {"type": str(data["transition"].get("type", "none"))} if isinstance(data.get("draft"), dict): metadata["draft"] = data["draft"] if isinstance(data.get("voiceover"), dict): vo_patch = data["voiceover"] existing = metadata.get("voiceover") if vo_patch.get("clear"): metadata.pop("voiceover", None) elif isinstance(existing, dict): if vo_patch.get("enabled") is not None: existing["enabled"] = bool(vo_patch["enabled"]) # 拖动字幕块后回写每句语音的句内起点(按 asset id 对位,只动 offset_ms) if isinstance(vo_patch.get("items"), list): offsets = {} for it in vo_patch["items"]: if isinstance(it, dict) and it.get("asset") is not None and it.get("offset_ms") is not None: try: offsets[str(it["asset"])] = max(0, int(it["offset_ms"])) except (TypeError, ValueError): continue for it in existing.get("items") or []: if str(it.get("asset")) in offsets: it["offset_ms"] = offsets[str(it.get("asset"))] metadata["voiceover"] = existing timeline.metadata = metadata timeline.save(update_fields=["metadata", "duration_seconds", "updated_at"]) return Response(ProjectSerializer(self.get_object()).data)