import logging from pathlib import Path import uuid from django.db import transaction from django.db.models import Count, Q from django.http import HttpResponse, 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, adopt_storyboard_shot_version, create_export_job, generate_base_asset, generate_person_triview, get_default_model, get_inflight_extraction, poll_storyboard, poll_video_segment, regenerate_script_segment, submit_extract_entities, 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, StoryboardShot, StoryboardShotVersion, 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 _clone_asset_into_team(src: Asset, *, team, user) -> Asset: """把跨团队可见的「官方模特」形象图克隆一份进目标团队:共享同一份 TOS 对象(只新建 Asset/AssetFile DB 行, 不重新上传),使其成为本团队真实资产 —— 可被基础资产组采用、在「我的演员」里列出,不再因 team 隔离报 asset not found。 去掉 metadata 里的 kind=model 标记,让克隆体作为普通「我的演员」person 资产,而不是被误判成模特库预设。""" src_meta = {k: v for k, v in (src.metadata or {}).items() if k != "kind"} clone = Asset.objects.create( team=team, created_by=user, name=src.name, asset_type=src.asset_type, source=src.source, category=src.category, description=src.description, metadata={**src_meta, "cloned_from_asset": str(src.id), "from_official_model": True}, in_library=src.in_library, ) for f in src.files.all(): AssetFile.objects.create( asset=clone, object_key=f.object_key, bucket=f.bucket, content_type=f.content_type, size_bytes=f.size_bytes, checksum=f.checksum, width=f.width, height=f.height, duration_ms=f.duration_ms, preview_url=f.preview_url, is_primary=f.is_primary, ) return clone 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 # V1 雪藏「拼接导出」第 5 阶段(前端 STAGE_STEPS 同步隐藏)→ 视频是当前最后一步。 # 因此「全部片段出片」即视为项目完成,不再依赖导出任务(原来只有 export.py 在拼接成片后 # 才置 COMPLETED,导出一藏起来就没有任何环节会把 videoing 翻成 completed)。 # V2 恢复导出时把此常量置 False:完成重新交回 export.py。 VIDEO_IS_FINAL_STAGE = True def _ensure_video_stage(project: Project) -> None: """幂等地把项目推进到「视频」阶段(生成视频片段时调用)。current_stage 没到 VIDEO 时, settle_video_completion 会直接返回 → 全段出片也不收口、卡在故事板格(ZWQ#15)。""" if project.current_stage != ProjectStage.Stage.VIDEO: project.current_stage = ProjectStage.Stage.VIDEO if project.status != Project.Status.COMPLETED: project.status = Project.Status.VIDEOING project.save(update_fields=["current_stage", "status", "updated_at"]) def settle_video_completion(project: Project) -> bool: """视频阶段收口(幂等): · 全部片段已出片(SUCCEEDED + 有采用版)→ 项目 COMPLETED、VIDEO 阶段标成功; · 否则若项目却仍标着 COMPLETED(用户重跑了某一镜)→ 退回 VIDEOING。 仅当视频为末阶段时生效;非视频阶段不动。返回是否发生状态变更。""" if not VIDEO_IS_FINAL_STAGE: return False if project.current_stage != ProjectStage.Stage.VIDEO: return False # 直接查库,不用 project.video_segments 的 prefetch 缓存:调用方常在刚 save 完某段后调用, # 缓存里仍是旧状态会误判(漏判完成 / 误判未完成)。 segments = list( VideoSegment.objects.filter(project=project).values_list("status", "adopted_version_id") ) if not segments: return False all_done = all( status_ == VideoSegment.Status.SUCCEEDED and adopted_id is not None for status_, adopted_id in segments ) if all_done and project.status != Project.Status.COMPLETED: stage, _ = ProjectStage.objects.get_or_create(project=project, stage=ProjectStage.Stage.VIDEO) if stage.status != ProjectStage.Status.SUCCEEDED: stage.status = ProjectStage.Status.SUCCEEDED stage.save(update_fields=["status", "updated_at"]) project.status = Project.Status.COMPLETED project.save(update_fields=["status", "updated_at"]) return True if not all_done and project.status == Project.Status.COMPLETED: # 重跑某段后又回到在制 → 从「已完成」退回「视频生成中」(用户重跑第 3 镜的情形) project.status = Project.Status.VIDEOING project.save(update_fields=["status", "updated_at"]) return True return False 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", "storyboard_shots__adopted_version__asset__files", "storyboard_shots__versions__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 _seed_product_triview_group(self, *, project_id, product_id, team): """惰性把商品库已有的商品三视图挂进本项目的商品 BaseAssetGroup(幂等)。 商品库走 standalone Asset、视频项目走项目级 BaseAssetGroup,是两套存储;此处做单向复用同步: 进详情时若商品已有 standalone 三视图(metadata.view=three_view + product_id)但项目商品组尚无采用资产, 就 seed 进去 → 第二步基础资产直接显示该三视图,不再「缺三视图」。已在项目内生成过的不覆盖。""" if not product_id: return groups = [ g for g in BaseAssetGroup.objects.filter(project_id=project_id, kind=BaseAssetGroup.Kind.PRODUCT) if not (g.metadata or {}).get("triview_of") ] if any(g.adopted_asset_id for g in groups): return # 项目内已有商品三视图,不覆盖 tri = ( Asset.objects.filter( team=team, is_deleted=False, metadata__product_id=str(product_id), metadata__view="three_view", ) .order_by("-created_at") .first() ) if tri is None: return group = groups[0] if groups else BaseAssetGroup.objects.create( project_id=project_id, kind=BaseAssetGroup.Kind.PRODUCT, metadata={} ) group.candidate_assets.add(tri) group.adopted_asset = tri group.save(update_fields=["adopted_asset", "updated_at"]) def retrieve(self, request, *args, **kwargs): # 详情加载(=进入流水线页)时自愈视频片段数:历史项目在「采用前增删分镜」未同步, # 或项目创建固定铺的 4 段从未被收口,会让视频步骤的片段数与故事板/采用版分镜对不上。 # 按采用版分镜数收口一次(幂等;只裁从未生成过的尾段,生成中/已出片的段绝不动)。 # 用轻量查询判定 + 同步,再交给 super 做重 prefetch 序列化(避免重查询跑两遍)。 proj = ( Project.objects.filter(pk=kwargs.get("pk"), **{self.team_field: self.get_team()}) .only("id", "current_stage", "status", "product_id") .first() ) if proj is not None: # 跨库同步:商品在「商品库」生成过三视图(standalone Asset, view=three_view + product_id), # 但本项目还没商品三视图组 → seed 进项目商品组,让第二步「基础资产」复用、不再「缺三视图」。 self._seed_product_triview_group(project_id=proj.id, product_id=proj.product_id, team=self.get_team()) if proj.current_stage == ProjectStage.Stage.VIDEO: adopted_script = proj.script_versions.filter(is_adopted=True).order_by("-created_at").first() if adopted_script is not None: self._sync_video_segments_to_script(proj, adopted_script) # 自愈卡阶段:历史项目「视频片段全部出片并采用」却因 current_stage 没推到 VIDEO 而永远停在 # 「3/4 故事板」、不收口(ZWQ#15)。进详情时若检测到全段已出片,就补推到视频阶段再收口。 segs = list(VideoSegment.objects.filter(project=proj).values_list("status", "adopted_version_id")) if segs and all(s == VideoSegment.Status.SUCCEEDED and a is not None for s, a in segs): _ensure_video_stage(proj) # 自愈完成态:全部片段已出片却卡在 videoing → 收口成 completed;若有段被重跑则退回 videoing。 settle_video_completion(proj) return super().retrieve(request, *args, **kwargs) 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=True, methods=["get"], url_path="export-clips") def export_clips(self, request, pk=None): """导出全部:把该项目所有已采用的视频片段打成一个 zip(逐段从 TOS 拉取,内存打包后下载)。 片段短(每段 ~15s),内存打包足够;单段拉取失败跳过不毁整包。""" import io import zipfile from urllib.parse import quote import requests project = self.get_object() segs = ( project.video_segments.filter(adopted_version__isnull=False) .select_related("adopted_version__asset") .prefetch_related("adopted_version__asset__files") .order_by("sort_order") ) clips = [] for seg in segs: asset = getattr(seg.adopted_version, "asset", None) if asset is None: continue f = asset.files.filter(is_primary=True).first() or asset.files.first() if f and f.object_key: clips.append((seg.sort_order, f)) if not clips: return Response({"detail": "该项目还没有可导出的视频片段"}, status=status.HTTP_400_BAD_REQUEST) buf = io.BytesIO() storage = TosStorage() packed = 0 with zipfile.ZipFile(buf, "w", zipfile.ZIP_STORED) as zf: for order, f in clips: try: url = storage.presigned_get_url(object_key=f.object_key) data = requests.get(url, timeout=120).content except Exception: # noqa: BLE001 — 单段失败跳过,不毁整包 continue ext = ".mp4" if f.content_type and "/" in f.content_type: sub = f.content_type.rsplit("/", 1)[-1] ext = "." + ("mov" if sub == "quicktime" else sub if sub in ("mp4", "webm") else "mp4") zf.writestr(f"{order + 1:02d}-镜{order + 1}{ext}", data) packed += 1 if packed == 0: return Response({"detail": "视频片段拉取失败,请稍后重试"}, status=status.HTTP_502_BAD_GATEWAY) payload = buf.getvalue() resp = HttpResponse(payload, content_type="application/zip") fname = quote(f"{project.name}-视频素材.zip") resp["Content-Disposition"] = f"attachment; filename=videos.zip; filename*=UTF-8''{fname}" resp["Content-Length"] = str(len(payload)) return resp @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"]) # 采用脚本时把视频片段数对齐到这版分镜数:用户常在「采用前」就增删分镜, # 那些编辑因 _sync 的 is_adopted 闸而未同步到 VideoSegment(项目创建时固定铺了 4 段), # 不在此收口的话视频步骤会一直停在 4 段,与故事板/分镜数对不上。已生成的段绝不动。 self._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.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="extract-entities") def extract_entities_action(self, request, pk=None): """从已定稿脚本提取角色 / 场景实体(独立步,进资产阶段时调)——**异步**:豆包思考模型流式抽取较慢, Web 层只建任务秒回(不再 502),慢活交 worker(run_extract_entities_task)。前端拿 task_id 轮询 extract-status 取结果。已有在途提取则复用(防刷新后重点重复扣费)。落库覆盖 project.metadata + 每镜 entity_refs;**商品不提**(参考图层用真实主图)。""" require_worker() # 异步提取依赖 worker 兜底执行,没 worker 直接拒绝(否则任务永远 RESERVED) project = self.get_object() try: task = submit_extract_entities(project=project, user=request.user) except ValueError as exc: # 无脚本 / 无分镜 / 无模型 / 额度不足,立即反馈 return Response({"detail": str(exc)}, status=status.HTTP_400_BAD_REQUEST) return Response({"task_id": str(task.id), "status": task.status}) @action(detail=True, methods=["get"], url_path="extract-status") def extract_status_action(self, request, pk=None): """实体提取进度(异步)。前端进资产页 / 点提取后据此轮询 + 刷新后重建「提取中」loading: running=有在途任务;否则读最近一条提取任务的成败。成功时回带 entities(供「提取+生成」继续出参考图)。""" project = self.get_object() inflight = get_inflight_extraction(project) if inflight is not None: return Response({"running": True, "status": inflight.status, "error": "", "entities": []}) last = ( AITask.objects.filter(project=project, task_type=AITask.Type.ENTITY_EXTRACTION) .order_by("-created_at") .first() ) if last is None: return Response({"running": False, "status": "", "error": "", "entities": []}) entities = (project.metadata or {}).get("script_entities", []) if last.status == AITask.Status.SUCCEEDED else [] return Response( {"running": False, "status": last.status, "error": last.error_message or "", "entities": entities} ) @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", ""), reference_asset_id=request.data.get("reference_asset_id") or None, # ZWQ#3:角色立绘可让 worker 出图后自动接力生成三视图。前端勾选时传 auto_triview=true; # 不传则维持原行为(只出立绘,用户再手点「AI 生成三视图」)。 auto_triview=bool(request.data.get("auto_triview")), ) 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="set-base-asset-adopt") @transaction.atomic def set_base_asset_adopt(self, request, pk=None): """显性「采用 / 未采用」某角色/场景组(metadata.adopt)。下游只取已采用的;脚本里没有的默认未采用, 但用户强制采用随他。默认态(不写 adopt)由前端按是否在脚本里推导,这里只存用户的覆盖。""" project = self.get_object() group = BaseAssetGroup.objects.select_for_update().filter(project=project, id=request.data.get("group_id")).first() if group is None: return Response({"detail": "group not found"}, status=status.HTTP_404_NOT_FOUND) state = request.data.get("state") if state not in ("adopted", "unadopted"): return Response({"detail": "state must be adopted/unadopted"}, status=status.HTTP_400_BAD_REQUEST) meta = dict(group.metadata or {}) meta["adopt"] = state group.metadata = meta group.save(update_fields=["metadata", "updated_at"]) return Response(BaseAssetGroupSerializer(group).data) @action(detail=True, methods=["post"], url_path="delete-base-asset") @transaction.atomic def delete_base_asset(self, request, pk=None): """删除某角色/场景资产组(连同它的三视图配套组一并删)。资产本身留在资产表,只移除项目内的组引用。""" project = self.get_object() group = BaseAssetGroup.objects.filter(project=project, id=request.data.get("group_id")).first() if group is None: return Response({"detail": "group not found"}, status=status.HTTP_404_NOT_FOUND) # 连带删该角色立绘配套的三视图组(triview_of 指向本组采用的立绘) if group.adopted_asset_id: project.base_asset_groups.filter(metadata__triview_of=str(group.adopted_asset_id)).delete() group.delete() return Response({"deleted": True}) @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 建组再挂(不出图)。 所选资产可能是「官方模特」形象图(跨团队可见但不属本团队)→ 克隆一份进本团队再挂:避免 team 隔离导致 明明选了官方模特却报 asset not found、替换失败。""" project = self.get_object() # 先把要挂的资产定下来(本团队资产,或官方模特形象图克隆),拿不到立即 404 —— 必须在建/找组「之前」, # 否则会留下一个采用资产为空的幽灵角色组(旧版本在 404 return 前就 create 了组,且 @atomic 正常 return 仍会提交)。 asset_id = request.data.get("asset_id") asset = Asset.objects.filter(team=project.team, id=asset_id).first() if asset is None and asset_id: # 跨团队可见的官方模特形象图 / 三视图 → 克隆进本团队再挂 official = ( Asset.objects.filter(id=asset_id, is_deleted=False) .filter(Q(model_as_portrait__is_official=True) | Q(model_as_triview__is_official=True)) .first() ) if official is not None: asset = _clone_asset_into_team(official, team=project.team, user=request.user) if asset is None: return Response({"detail": "asset not found"}, status=status.HTTP_404_NOT_FOUND) 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) group.candidate_assets.add(asset) group.adopted_asset_id = asset.id # 选用/替换即「采用」:显式置 metadata.adopt=adopted,让卡片直接显「已采用」,不必用户再手点采用(ZWQ#7)。 meta = dict(group.metadata or {}) meta["adopt"] = "adopted" group.metadata = meta group.save(update_fields=["adopted_asset", "metadata", "updated_at"]) # 人物立绘/三视图属送审类:选用后自动送审(变绿盾「已过审」),不必用户再手点灰盾(ZWQ#7)。 # 仅当还没过审时提交;best-effort,审核服务不可用也不阻断采用。on_commit 确保资产已落库再送审。 if asset.category in Asset.REVIEW_CATEGORIES and asset.review_status != "active": from apps.assets.review import submit_asset_for_review transaction.on_commit(lambda a=asset: submit_asset_for_review(a)) # ZWQ#4 · 沿用模特库已有三视图:选了「已带三视图」的模特(官方/自有)作角色时, # 把该模特的三视图也挂进项目(triview_of=本组采用立绘 id),前端即判定「已有三视图」、 # 不再要求点「AI 生成三视图」。仅对人物组生效;asset_id 传的是模特的 portrait_asset(原始 id), # 故按原始 portrait_asset 反查 Model;官方模特三视图同立绘一样克隆进本团队再挂(team 隔离)。 if asset.category in (Asset.Category.PERSON, Asset.Category.MODEL_PORTRAIT): self._attach_existing_model_triview(project, group, asset, source_portrait_id=asset_id, user=request.user) promote_base_asset_stage_if_ready(project) return Response(BaseAssetGroupSerializer(group).data) def _attach_existing_model_triview(self, project, portrait_group, portrait_asset, *, source_portrait_id, user): """若被采用的立绘来自「已带三视图」的模特,把该三视图挂进项目并采用(triview_of=立绘 id)。 幂等:已存在该立绘的三视图组就跳过,不重复挂。best-effort,出错不阻断主采用流程。""" from apps.assets.models import Model as ModelEntity # source_portrait_id = 前端传的资产 id(模特的原始 portrait_asset);官方模特会被克隆, # 故 portrait_asset 可能是克隆体。两边都查一遍找到对应 Model。 candidate_ids = {str(portrait_asset.id)} if source_portrait_id: candidate_ids.add(str(source_portrait_id)) model = ( ModelEntity.objects.filter(portrait_asset_id__in=candidate_ids, is_deleted=False) .exclude(triview_asset__isnull=True) .select_related("triview_asset") .first() ) if model is None or model.triview_asset is None: return portrait_key = str(portrait_asset.id) # 已有该立绘的三视图组?幂等跳过。 existing = next( (g for g in project.base_asset_groups.filter(kind=BaseAssetGroup.Kind.PERSON) if (g.metadata or {}).get("triview_of") == portrait_key), None, ) if existing is not None: return tri_asset = model.triview_asset # 官方模特三视图跨团队 → 克隆进本团队再挂(与立绘同款);本团队三视图直接用。 if tri_asset.team_id != project.team_id: tri_asset = _clone_asset_into_team(tri_asset, team=project.team, user=user) tri_group = BaseAssetGroup.objects.create( project=project, kind=BaseAssetGroup.Kind.PERSON, prompt="", metadata={"label": "·三视图", "triview_of": portrait_key, "adopt": "adopted"}, ) tri_group.candidate_assets.add(tri_asset) tri_group.adopted_asset = tri_asset tri_group.save(update_fields=["adopted_asset", "updated_at"]) if tri_asset.category in Asset.REVIEW_CATEGORIES and tri_asset.review_status != "active": from apps.assets.review import submit_asset_for_review transaction.on_commit(lambda a=tri_asset: submit_asset_for_review(a)) @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 def _sync_storyboard_shots_to_script(self, project: Project, script: ScriptVersion) -> None: """采用版分镜数变化时,同步 StoryboardShot 数量(与视频段同策略,按位置对齐): 多出且尾部「从未出过图」的 shot 裁掉(已出图的不动);不主动建——出图时 ensure_storyboard_shots 按需补。""" if not script.is_adopted: return target = script.segments.count() shots = list(project.storyboard_shots.order_by("sort_order")) while len(shots) > target: tail = shots[-1] if tail.status == StoryboardShot.Status.NOT_STARTED and not tail.versions.exists(): tail.delete() shots.pop() else: break @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) self._sync_storyboard_shots_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 # 允许删空:用户可以把镜头全删了(再整体重新生成或手动加),不再强制「至少保留一个分镜」。 # 删本场脚本时,它对应的故事板分镜帧也一起删 —— 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() # 删后重排 sort_order:收集变动的镜一次性 bulk_update,替代逐条 save(远程库少跑 N 个往返 → 删除快很多) resort = [] for index, seg in enumerate(script.segments.order_by("sort_order")): if seg.sort_order != index: seg.sort_order = index resort.append(seg) if resort: ScriptSegment.objects.bulk_update(resort, ["sort_order"]) self._sync_video_segments_to_script(project, script) self._sync_storyboard_shots_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 ) # 采用历史版后该段为已出片 → 先确保在视频阶段,再收口项目为 completed(否则卡 3/4,ZWQ#15) _ensure_video_stage(project) settle_video_completion(project) return Response(ProjectSerializer(project).data) @action(detail=True, methods=["post"], url_path="generate-storyboard") def generate_storyboard_action(self, request, pk=None): """开始生成故事板(全部场):确保每镜一个 shot 并标 QUEUED;逐场出图交给 poll-storyboard 起线程。 全部已出片时再点 = 整批重跑(全部场重出)。不推进阶段。""" require_worker() project = self.get_object() 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({"status": "queued"}, status=status.HTTP_202_ACCEPTED) @action(detail=True, methods=["post"], url_path="rerun-storyboard-shot") def rerun_storyboard_shot_action(self, request, pk=None): """单场重跑:只重出该 shot 的图 —— 新增一条历史版本并采用,其余场不动。出图交给 poll-storyboard。""" require_worker() project = self.get_object() shot_id = request.data.get("shot_id") if not project.storyboard_shots.filter(id=shot_id).exists(): return Response({"detail": "shot not found"}, status=status.HTTP_404_NOT_FOUND) submit_storyboard(project=project, user=request.user, prompt=request.data.get("prompt", ""), shot_ids=[shot_id]) 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({"status": "queued"}, status=status.HTTP_202_ACCEPTED) @action(detail=True, methods=["post"], url_path="adopt-storyboard-shot-version") def adopt_storyboard_shot_version_action(self, request, pk=None): """采用某场的某个历史版本(对标 adopt-video-version):切换该场采用的分镜图。""" project = self.get_object() shot = StoryboardShot.objects.filter(project=project, id=request.data.get("shot_id")).first() if shot is None: return Response({"detail": "shot not found"}, status=status.HTTP_404_NOT_FOUND) version = StoryboardShotVersion.objects.filter(shot=shot, id=request.data.get("version_id")).first() if version is None: return Response({"detail": "version not found"}, status=status.HTTP_404_NOT_FOUND) adopt_storyboard_shot_version(shot=shot, version=version) return Response(ProjectSerializer(self.get_object()).data) @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 = poll_storyboard(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"]) # 进入视频阶段前再收口一次视频片段数,确保与采用版分镜数一致(已生成的段不动) adopted_script = project.script_versions.filter(is_adopted=True).order_by("-created_at").first() if adopted_script is not None: self._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"]) 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"]) # 跳过故事板直接进视频时,同样把视频片段数对齐到采用版分镜数(已生成的段不动) adopted_script = project.script_versions.filter(is_adopted=True).order_by("-created_at").first() if adopted_script is not None: self._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"]) return Response(ProjectSerializer(project).data) @action(detail=True, methods=["post"], url_path="video-review-precheck") def video_review_precheck(self, request, pk=None): """点「生成视频」前的过审校验:返回未过审(非 active)的人物立绘 / 故事板分镜清单。 body 可选 video_segment_id:只校验该段(单段重跑);不传 = 校验全部未出片段。 blockers 空 = 全部已过审,前端放行生成。""" from apps.ai.services import collect_video_review_blockers project = self.get_object() seg = None sid = request.data.get("video_segment_id") if sid: seg = project.video_segments.filter(id=sid).first() blockers = collect_video_review_blockers(project, only_segment=seg) return Response({"blockers": blockers}) @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")) # 火山/中转报错(如人脸需走素材库的 InputImageSensitiveContentDetected)→ 返回真实报错的 JSON, # 前端如实显示(段已被 submit_video_segment 标 FAILED + error_message);绝不让它变 500 HTML → 前端白屏。 try: submit_video_segment(video_segment=segment, user=request.user, prompt=request.data.get("prompt", "")) except Exception as exc: # noqa: BLE001 return Response({"detail": str(exc)}, status=status.HTTP_502_BAD_GATEWAY) # 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) # 生成视频即代表已进入视频阶段:某些路径(跳过故事板/异常)可能没把 current_stage 推到 VIDEO, # 而 settle_video_completion 只在 current_stage==VIDEO 时收口 → 不补推的话,全段出片后项目会永远停在 # 「3/4 故事板」、永不 completed(ZWQ#15)。这里幂等补推。 _ensure_video_stage(project) # 重跑某段 → 该段回到在制;若项目此前已 completed,退回 videoing settle_video_completion(project) 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) # 这一段刚出片 → 先确保已在视频阶段(否则 settle 不收口,卡 3/4),再收口整个项目为 completed _ensure_video_stage(project) settle_video_completion(project) 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)