yingqing/core/backend/apps/projects/views.py

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 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")
            .first()
        )
        if proj is not None:
            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", ""))
        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)