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, F, Prefetch, Q
from django.http import HttpResponse, JsonResponse, StreamingHttpResponse
from django.utils import timezone
from rest_framework import status
from rest_framework.decorators import action
from rest_framework.exceptions import APIException, ValidationError
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 (
    SEGMENT_DURATION_MAX,
    SEGMENT_DURATION_MIN,
    combo_keys,
    coerce_total_duration,
    recommend_script_setup,
    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, require_worker_task
from apps.ai.generation_errors import classify_generation_error, public_error_for_task
from apps.ai.video_digest import VideoDigestError, digest_project_video
from apps.products.models import Product, ProductImage, ProductSellingPoint

from .models import (
    BaseAssetGroup,
    BgmTrack,
    ExportJob,
    Project,
    ProjectStage,
    QuickCreateJob,
    ScriptSegment,
    ScriptTemplate,
    ScriptVersion,
    StoryboardFrame,
    StoryboardShot,
    StoryboardShotVersion,
    StoryboardVersion,
    SubtitleTrack,
    Timeline,
    TimelineClip,
    VideoSegment,
    VideoSegmentVersion,
)
from .serializers import (
    BaseAssetGroupSerializer,
    ExportJobSerializer,
    ProjectListSerializer,
    ProjectSerializer,
    QuickCreateJobSerializer,
    ScriptTemplateSerializer,
    ScriptVersionSerializer,
    StoryboardVersionSerializer,
    VideoSegmentVersionSerializer,
    is_playable_video,
)
from .services.export import run_export_job_in_thread
from .services.pipeline import (
    adopt_script_version,
    finish_storyboard_stage,
    initialize_project_pipeline,
    sync_video_segments_to_script,
)
from .services.script_import import ScriptFileError, extract_script_text
from .services.templates import build_template_fields, coerce_persona, coerce_template_combo, render_outline_text
from .tasks import poll_video_segment_task

logger = logging.getLogger(__name__)


class QuickCreateInProgress(APIException):
    status_code = status.HTTP_409_CONFLICT
    default_detail = "该项目正在极速成片中，请到极速成片页查看进度"
    default_code = "quick_create_running"


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 复用。"""
    fallback_suffix = {
        Asset.Type.IMAGE: ".png",
        Asset.Type.VIDEO: ".mp4",
        Asset.Type.AUDIO: ".mp3",
    }.get(asset_type, ".bin")
    suffix = Path(upload.name).suffix.lower() or fallback_suffix
    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", "quick_create_job").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, purged_at__isnull=True,
                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 list(self, request, *args, **kwargs):
        from .services.quick_create import restore_false_failed_quick_creates

        restore_false_failed_quick_creates(self.get_team())
        return super().list(request, *args, **kwargs)

    def retrieve(self, request, *args, **kwargs):
        # 详情加载(=进入流水线页)时自愈视频片段数:历史项目在「采用前增删分镜」未同步,
        # 或项目创建固定铺的 4 段从未被收口,会让视频步骤的片段数与故事板/采用版分镜对不上。
        # 按采用版分镜数收口一次(幂等;只裁从未生成过的尾段,生成中/已出片的段绝不动)。
        # 用轻量查询判定 + 同步,再交给 super 做重 prefetch 序列化(避免重查询跑两遍)。
        proj = (
            Project.objects.filter(
                pk=kwargs.get("pk"),
                is_deleted=False,
                purged_at__isnull=True,
                **{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):
        if self.action in ("trash", "restore", "purge"):
            return (
                Project.objects.select_related("product", "product__cover_asset")
                .prefetch_related("product__cover_asset__files")
                .filter(is_deleted=True, purged_at__isnull=True, **{self.team_field: self.get_team()})
                .order_by("-updated_at")
            )
        # 列表:轻查询(只 select_related product + 计数注解),不做详情那串 12 个重 prefetch
        # ——原列表把每个项目的 阶段/片段/故事板/时间线/资产文件全拉出,20 个项目实测 ~2s。
        if self.action == "list":
            qs = (
                Project.objects.select_related("product", "product__cover_asset", "timeline", "quick_create_job")
                .prefetch_related(
                    "product__cover_asset__files",
                    # 成片地址(final_video_url)只需要「成功的导出任务」,预取到位后列表不再逐项目查库
                    Prefetch(
                        "timeline__export_jobs",
                        queryset=ExportJob.objects.filter(status=ExportJob.Status.SUCCEEDED)
                        .select_related("output_asset")
                        .prefetch_related("output_asset__files")
                        .order_by("-created_at"),
                    ),
                )
                .annotate(
                    script_version_count=Count("script_versions", distinct=True),
                    video_segment_count=Count("video_segments", distinct=True),
                )
            )
            return qs.filter(
                is_deleted=False,
                purged_at__isnull=True,
                **{self.team_field: self.get_team()},
            ).order_by("-updated_at")
        return super().get_queryset().filter(is_deleted=False, purged_at__isnull=True)

    def initial(self, request, *args, **kwargs):
        super().initial(request, *args, **kwargs)
        if request.method in ("GET", "HEAD", "OPTIONS"):
            return
        if self.action in {"create", "destroy"}:
            return
        pk = kwargs.get("pk")
        if not pk:
            return
        if QuickCreateJob.objects.filter(
            project_id=pk,
            status__in=[QuickCreateJob.Status.QUEUED, QuickCreateJob.Status.RUNNING],
        ).exists():
            raise QuickCreateInProgress()

    def perform_destroy(self, instance):
        instance.is_deleted = True
        instance.save(update_fields=["is_deleted", "updated_at"])

    @action(detail=False, methods=["get"], url_path="trash")
    def trash(self, request):
        """垃圾桶:列出本团队已软删且未彻底隐藏的视频项目。"""
        qs = self.filter_queryset(self.get_queryset())
        page = self.paginate_queryset(qs)
        serializer_cls = ProjectListSerializer
        if page is not None:
            return self.get_paginated_response(serializer_cls(page, many=True).data)
        return Response(serializer_cls(qs, many=True).data)

    @action(detail=True, methods=["post"], url_path="restore")
    def restore(self, request, pk=None):
        """从垃圾桶恢复视频项目。"""
        project = self.get_object()
        project.is_deleted = False
        project.purged_at = None
        project.save(update_fields=["is_deleted", "purged_at", "updated_at"])
        return Response(ProjectSerializer(project).data, status=status.HTTP_200_OK)

    @action(detail=True, methods=["delete"], url_path="purge")
    def purge(self, request, pk=None):
        """彻底删除 = 二级软删除:从垃圾桶隐藏,项目 DB 记录保留。"""
        project = self.get_object()
        project.purged_at = timezone.now()
        project.save(update_fields=["purged_at", "updated_at"])
        return Response(status=status.HTTP_204_NO_CONTENT)

    @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(), is_deleted=False, purged_at__isnull=True)
        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})

    @action(
        detail=False,
        methods=["post"],
        url_path="quick-create",
        parser_classes=[MultiPartParser, FormParser],
    )
    def quick_create(self, request):
        """商品名称 + 1–9 张图 → 新建商品与项目，并启动完整自动生产流水线。"""
        require_worker_task("apps.projects.tasks.advance_quick_create_task")
        name = str(request.data.get("name") or "").strip()
        uploads = request.FILES.getlist("images") or request.FILES.getlist("images[]")
        source_product_id = str(request.data.get("source_product_id") or "").strip()
        requested_asset_ids = [
            str(value).strip()
            for value in (request.data.getlist("image_asset_ids") or request.data.getlist("image_asset_ids[]") or [])
            if str(value).strip()
        ]
        if not name:
            return Response({"detail": "请填写商品名称"}, status=status.HTTP_400_BAD_REQUEST)
        if len(name) > 255:
            return Response({"detail": "商品名称不能超过255个字符"}, status=status.HTTP_400_BAD_REQUEST)
        if not uploads and not source_product_id:
            return Response({"detail": "请至少上传一张商品图片"}, status=status.HTTP_400_BAD_REQUEST)
        if len(uploads) > 9:
            return Response({"detail": "商品图片最多上传9张"}, status=status.HTTP_400_BAD_REQUEST)
        allowed_types = {"image/jpeg", "image/png", "image/webp"}
        for upload in uploads:
            if (upload.content_type or "").lower() not in allowed_types:
                return Response({"detail": "仅支持 JPG、PNG 或 WebP 商品图片"}, status=status.HTTP_400_BAD_REQUEST)
            if upload.size > 20 * 1024 * 1024:
                return Response({"detail": "单张商品图片不能超过20MB"}, status=status.HTTP_400_BAD_REQUEST)

        aspect_ratio = str(request.data.get("aspect_ratio") or "9:16")
        resolution = str(request.data.get("resolution") or "720p").lower()
        try:
            total_duration = int(request.data.get("total_duration") or 15)
        except (TypeError, ValueError):
            total_duration = 0
        if aspect_ratio not in {"21:9", "16:9", "4:3", "1:1", "3:4", "9:16"}:
            return Response({"detail": "请选择有效的视频比例"}, status=status.HTTP_400_BAD_REQUEST)
        if resolution not in {"480p", "720p", "1080p", "4k"}:
            return Response({"detail": "请选择有效的视频分辨率"}, status=status.HTTP_400_BAD_REQUEST)
        if total_duration not in {15, 30, 45, 60}:
            return Response({"detail": "视频时长仅支持15、30、45或60秒"}, status=status.HTTP_400_BAD_REQUEST)

        # 极速成片和专业创作使用同一款豆包脚本模型；绝不因默认模型变化而退回 DeepSeek。
        from .services.quick_create import get_quick_script_model

        text_model = get_quick_script_model()
        image_model = get_default_model(ModelConfig.Capability.IMAGE)
        requested_video_model_id = str(request.data.get("video_model_config_id") or "").strip()
        video_model = None
        if requested_video_model_id:
            try:
                video_model_uuid = uuid.UUID(requested_video_model_id)
            except (TypeError, ValueError):
                return Response({"detail": "请选择有效的视频模型"}, status=status.HTTP_400_BAD_REQUEST)
            video_model = (
                ModelConfig.objects.select_related("provider")
                .filter(
                    id=video_model_uuid,
                    capability=ModelConfig.Capability.VIDEO,
                    status=ModelConfig.Status.ACTIVE,
                    provider__status="active",
                )
                .first()
            )
        else:
            video_model = get_default_model(ModelConfig.Capability.VIDEO)

        missing = [
            label
            for model, label in ((image_model, "图像"), (video_model, "视频"))
            if model is None
        ]
        if text_model is None:
            return Response(
                {"detail": "极速成片需要的豆包 Seed 2.1 Pro 模型未启用，请联系管理员配置"},
                status=status.HTTP_503_SERVICE_UNAVAILABLE,
            )
        if missing:
            return Response(
                {"detail": f"当前缺少可用的{'、'.join(missing)}模型，请联系管理员配置"},
                status=status.HTTP_503_SERVICE_UNAVAILABLE,
            )
        capabilities = dict((video_model.metadata or {}).get("capabilities") or {})
        supported_resolutions = set(capabilities.get("resolutions") or (video_model.metadata or {}).get("resolutions") or [])
        supported_ratios = set(capabilities.get("aspect_ratios") or [])
        supported_durations = set(capabilities.get("durations") or (video_model.metadata or {}).get("durations") or [])
        if supported_resolutions and resolution not in supported_resolutions:
            return Response(
                {"detail": f"{video_model.display_name} 不支持 {resolution.upper()}，请更换清晰度或模型"},
                status=status.HTTP_400_BAD_REQUEST,
            )
        if supported_ratios and aspect_ratio not in supported_ratios:
            return Response(
                {"detail": f"{video_model.display_name} 不支持 {aspect_ratio} 比例"},
                status=status.HTTP_400_BAD_REQUEST,
            )
        # 极速成片把总时长拆成多个15秒镜头，视频模型只需支持单镜15秒。
        if supported_durations and 15 not in supported_durations:
            return Response(
                {"detail": f"{video_model.display_name} 不支持15秒单镜生成"},
                status=status.HTTP_400_BAD_REQUEST,
            )

        team = self.get_team()
        reused_assets = []
        if source_product_id:
            try:
                source_uuid = uuid.UUID(source_product_id)
            except (TypeError, ValueError):
                return Response({"detail": "找不到可复用的商品图片"}, status=status.HTTP_400_BAD_REQUEST)
            source = (
                Product.objects.filter(id=source_uuid, team=team)
                .prefetch_related("images__asset")
                .first()
            )
            if source is None:
                return Response({"detail": "找不到可复用的商品图片"}, status=status.HTTP_400_BAD_REQUEST)
            source_images = list(source.images.all())
            by_asset_id = {str(image.asset_id): image.asset for image in source_images}
            if requested_asset_ids:
                if any(asset_id not in by_asset_id for asset_id in requested_asset_ids):
                    return Response({"detail": "找不到可复用的商品图片"}, status=status.HTTP_400_BAD_REQUEST)
                reused_assets = [by_asset_id[asset_id] for asset_id in requested_asset_ids]
            else:
                reused_assets = [image.asset for image in source_images]
            if not reused_assets and not uploads:
                return Response({"detail": "请至少上传一张商品图片"}, status=status.HTTP_400_BAD_REQUEST)
        if len(reused_assets) + len(uploads) > 9:
            return Response({"detail": "商品图片最多上传9张"}, status=status.HTTP_400_BAD_REQUEST)
        category = str(request.data.get("category") or "").strip()
        if not category:
            return Response({"detail": "请选择商品品类"}, status=status.HTTP_400_BAD_REQUEST)
        if category not in Product.ECOMMERCE_CATEGORIES:
            return Response({"detail": "请选择有效的商品品类"}, status=status.HTTP_400_BAD_REQUEST)

        with transaction.atomic():
            assets = list(reused_assets)
            assets.extend(
                _store_uploaded_asset(
                    team=team,
                    user=request.user,
                    upload=upload,
                    asset_type=Asset.Type.IMAGE,
                    category=Asset.Category.PRODUCT_IMAGE,
                    name=upload.name or f"{name}-商品图{index + 1}",
                )
                for index, upload in enumerate(uploads)
            )
            if not assets:
                return Response({"detail": "请至少上传一张商品图片"}, status=status.HTTP_400_BAD_REQUEST)
            product = Product.objects.create(
                team=team,
                created_by=request.user,
                title=name,
                business_type=Product.BusinessType.ECOMMERCE,
                category=category,
                cover_asset=assets[0],
            )
            ProductImage.objects.bulk_create(
                [
                    ProductImage(
                        product=product,
                        asset=asset,
                        sort_order=index,
                        is_primary=index == 0,
                    )
                    for index, asset in enumerate(assets)
                ]
            )
            rec = recommend_script_setup(category, name)
            project = Project.objects.create(
                team=team,
                created_by=request.user,
                product=product,
                name=f"{name} · 极速成片",
                status=Project.Status.SCRIPTING,
                current_stage=ProjectStage.Stage.SCRIPT,
                metadata={
                    "quick_create": True,
                    "wizard": {
                        "aspect_ratio": aspect_ratio,
                        "resolution": resolution,
                        "total_duration": total_duration,
                        "video_model_config_id": str(video_model.id),
                        "video_model_name": video_model.name,
                        "video_model_label": video_model.display_name,
                        "presentation_format": rec["format"],
                        "video_structure": rec["structure"],
                        "persona": rec["persona"],
                    },
                },
            )
            initialize_project_pipeline(project, placeholder_segments=max(1, total_duration // 15))
            if not product.selling_points.exists():
                ProductSellingPoint.objects.create(
                    product=product,
                    title=name,
                    detail="由极速成片根据商品名称自动填写",
                    sort_order=0,
                )
            job = QuickCreateJob.objects.create(
                team=team,
                created_by=request.user,
                project=project,
                status=QuickCreateJob.Status.QUEUED,
                phase=QuickCreateJob.Phase.PRODUCT,
                progress=0,
                message="等待开始极速成片",
            )

        try:
            from .services.quick_create import _enqueue_advance

            _enqueue_advance(job)
        except Exception as exc:  # noqa: BLE001 — broker 极小窗口失败也必须给任务落终态
            from .services.quick_create import fail_quick_create

            logger.exception("quick create enqueue failed for job %s", job.id)
            fail_quick_create(job, "生成队列暂时不可用，请稍后重试", internal_error=str(exc))
            return Response(QuickCreateJobSerializer(job).data, status=status.HTTP_503_SERVICE_UNAVAILABLE)

        job = self._quick_job_queryset().get(id=job.id)
        return Response(QuickCreateJobSerializer(job).data, status=status.HTTP_202_ACCEPTED)

    def _quick_job_queryset(self):
        return (
            QuickCreateJob.objects.select_related("project__product", "project__product__cover_asset", "project__timeline")
            .prefetch_related(
                "project__product__images__asset__files",
                "project__product__cover_asset__files",
                "project__script_versions",
                "project__base_asset_groups",
                "project__storyboard_shots__adopted_version__asset__files",
                "project__video_segments__adopted_version__asset__files",
                "project__timeline__export_jobs__output_asset__files",
            )
            .filter(team=self.get_team())
        )

    @action(detail=False, methods=["get"], url_path=r"quick-create-status/(?P<job_id>[^/.]+)")
    def quick_create_status(self, request, job_id=None):
        job = self._quick_job_queryset().filter(id=job_id).first()
        if job is None:
            return Response({"detail": "极速成片任务不存在"}, status=status.HTTP_404_NOT_FOUND)
        from .services.quick_create import recover_quick_create

        try:
            recover_quick_create(job)
        except Exception:  # noqa: BLE001 — 恢复失败不能把进度接口打成 500
            logger.exception("quick create recover failed for job %s", job.id)
        job = self._quick_job_queryset().get(id=job.id)
        return Response(QuickCreateJobSerializer(job).data)

    @action(detail=False, methods=["post"], url_path=r"quick-create-cancel/(?P<job_id>[^/.]+)")
    def quick_create_cancel(self, request, job_id=None):
        job = self._quick_job_queryset().filter(id=job_id).first()
        if job is None:
            return Response({"detail": "极速成片任务不存在"}, status=status.HTTP_404_NOT_FOUND)
        if job.status == QuickCreateJob.Status.SUCCEEDED:
            return Response({"detail": "已完成的任务不能取消"}, status=status.HTTP_400_BAD_REQUEST)
        from .services.quick_create import cancel_quick_create

        cancel_quick_create(job)
        job = self._quick_job_queryset().get(id=job.id)
        return Response(QuickCreateJobSerializer(job).data)

    @action(detail=False, methods=["post"], url_path=r"quick-create-retry/(?P<job_id>[^/.]+)")
    def quick_create_retry(self, request, job_id=None):
        job = self._quick_job_queryset().filter(id=job_id).first()
        if job is None:
            return Response({"detail": "极速成片任务不存在"}, status=status.HTTP_404_NOT_FOUND)
        if job.status == QuickCreateJob.Status.SUCCEEDED:
            return Response({"detail": "任务已完成"}, status=status.HTTP_400_BAD_REQUEST)
        if job.status == QuickCreateJob.Status.CANCELLED:
            return Response({"detail": "已取消的任务请重新开始"}, status=status.HTTP_400_BAD_REQUEST)
        from .services.quick_create import resume_quick_create

        resume_quick_create(job)
        job = self._quick_job_queryset().get(id=job.id)
        return Response(QuickCreateJobSerializer(job).data)

    @action(detail=False, methods=["get"], url_path="quick-create-history")
    def quick_create_history(self, request):
        from .services.quick_create import restore_false_failed_quick_creates

        restore_false_failed_quick_creates(self.get_team())
        # 进行中 / 未完成的任务回填上方表单；过往列表只放已完成成片。
        jobs = (
            self._quick_job_queryset()
            .filter(
                project__is_deleted=False,
                project__purged_at__isnull=True,
                status=QuickCreateJob.Status.SUCCEEDED,
            )
            .order_by("-created_at")
        )
        return Response({
            "count": jobs.count(),
            "results": QuickCreateJobSerializer(jobs[:30], many=True).data,
        })

    @transaction.atomic
    def perform_create(self, serializer):
        project = serializer.save(team=self.get_team(), created_by=self.request.user)
        self._apply_wizard_template(project)
        # 先铺 4 段占位;真实段数与每段时长在采用脚本时收口。
        initialize_project_pipeline(project, placeholder_segments=4)

    def _apply_wizard_template(self, project) -> None:
        """新建向导选了套路模板 → 后端按模板真值回填 metadata.wizard,并计一次使用。
        前端只传 template_id,套路参数一律以库里的模板为准,避免前后端两份真相。"""
        wizard = (project.metadata or {}).get("wizard") or {}
        template_id = wizard.get("template_id")
        if not template_id:
            return
        template = ScriptTemplate.objects.filter(id=template_id, team=project.team).first()
        if template is None:
            # 模板被删/跨团队 → 静默丢弃这个字段,项目照常建,不阻断创建
            wizard.pop("template_id", None)
            project.metadata = {**(project.metadata or {}), "wizard": wizard}
            project.save(update_fields=["metadata", "updated_at"])
            return
        wizard.update(
            {
                "template_id": str(template.id),
                "template_name": template.name,
                "template_outline": render_outline_text({"outline": template.outline, "cta": template.cta}),
            }
        )
        # 套路参数只在模板真有值时覆盖,空值不要把设定卡的智能推荐顶掉。
        # 旧模板可能存了中文标签(「短剧」),这里归一成 key 再写入 wizard。
        raw_format = getattr(template, "presentation_format", "") or ""
        raw_structure = getattr(template, "video_structure", "") or ""
        if raw_format or raw_structure:
            fmt, structure = coerce_template_combo(raw_format, raw_structure)
            wizard["presentation_format"] = fmt
            wizard["video_structure"] = structure
        persona = coerce_persona(getattr(template, "persona", "") or "")
        if persona:
            wizard["persona"] = persona
        if template.total_duration:
            wizard["total_duration"] = template.total_duration
        project.metadata = {**(project.metadata or {}), "wizard": wizard}
        project.save(update_fields=["metadata", "updated_at"])
        ScriptTemplate.objects.filter(id=template.id).update(usage_count=F("usage_count") + 1)

    @action(detail=True, methods=["post"], url_path="save-as-template")
    def save_as_template(self, request, pk=None):
        """把当前脚本存成团队套路模板(5.1)。只抽套路不抽文案,见 services/templates.py。"""
        project = self.get_object()
        name = str(request.data.get("name") or "").strip()
        if len(name) < 2:
            return Response({"detail": "模板名至少 2 个字"}, status=status.HTTP_400_BAD_REQUEST)
        script_id = request.data.get("script_version_id")
        scripts = ScriptVersion.objects.filter(project=project).prefetch_related("segments")
        script = (
            scripts.filter(id=script_id).first()
            if script_id
            else (scripts.filter(is_adopted=True).order_by("-created_at").first() or scripts.order_by("-created_at").first())
        )
        if script is None:
            return Response({"detail": "这个项目还没有脚本,先生成一版再存模板"}, status=status.HTTP_400_BAD_REQUEST)
        if not script.segments.exists():
            return Response({"detail": "这版脚本没有分镜,存不出套路"}, status=status.HTTP_400_BAD_REQUEST)
        if ScriptTemplate.objects.filter(team=project.team, name=name).exists():
            return Response({"detail": "模板名已存在,换一个"}, status=status.HTTP_400_BAD_REQUEST)
        template = ScriptTemplate.objects.create(
            team=project.team,
            created_by=request.user,
            name=name,
            source_project=project,
            source_script=script,
            **build_template_fields(project=project, script=script),
        )
        return Response(ScriptTemplateSerializer(template).data, status=status.HTTP_201_CREATED)

    @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、persona、
        base_version_id(改稿)、aspect_ratio、total_duration(15/30/45/60)、
        presentation_format(oral|drama|vlog)、video_structure(pain|contrast|review|scene)。
        响应: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 "")
        base_version_id = request.data.get("base_version_id") or None
        aspect_ratio = str(request.data.get("aspect_ratio") or "9:16")
        # 非法值一律由 agent 侧 coerce 兜底(夹区间/回落默认),这里不做 400,避免生成被参数噪声打断
        total_duration = coerce_total_duration(request.data.get("total_duration"))
        presentation_format, video_structure = combo_keys(
            request.data.get("presentation_format"),
            request.data.get("video_structure"),
        )
        wizard = ((project.metadata or {}).get("wizard") if isinstance(project.metadata, dict) else None) or {}
        selling_point_ids = request.data.get("selling_point_ids") or wizard.get("selling_point_ids") or []
        persona = str(request.data.get("persona") or wizard.get("persona") or "")
        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,
            presentation_format=presentation_format,
            video_structure=video_structure,
            target_index=target_index,
            entry_source=str(request.data.get("source") or ""),
            persona=persona,
        )
        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)
        adopt_script_version(project, script)
        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:  # 无脚本 / 无分镜 / 无模型 / 额度不足,立即反馈
            message = str(exc)
            internal_kind = (
                "user_credit_insufficient" if "额度不足" in message
                else "model_unavailable" if "模型" in message
                else "invalid_input"
            )
            public_error = classify_generation_error(
                exc, operation="entity_extract", internal_kind=internal_kind
            )
            return Response(
                {"detail": public_error.fallback_message, "error": public_error.as_dict()},
                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": None, "error_message": "", "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": None, "error_message": "", "entities": []})
        entities = (project.metadata or {}).get("script_entities", []) if last.status == AITask.Status.SUCCEEDED else []
        public_error = public_error_for_task(last, operation="entity_extract")
        return Response(
            {
                "running": False,
                "status": last.status,
                "error": public_error.as_dict() if public_error else None,
                "error_message": public_error.fallback_message if public_error else "",
                "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,
                # 用户点击角色 AI 生成 = 生成立绘并在完成后自动接力三视图。
                # 三视图任务由 worker 创建，页面刷新或离开也不会漏掉。
                auto_triview=kind == BaseAssetGroup.Kind.PERSON,
            )
        except ValueError as exc:  # 无可用模型 / 余额不足等,立即反馈
            internal_kind = "user_credit_insufficient" if str(exc).strip().lower() == "insufficient credit" else ""
            public_error = classify_generation_error(
                exc, operation="base_asset_generate", internal_kind=internal_kind
            )
            return Response(
                {"detail": public_error.fallback_message, "error": public_error.as_dict()},
                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,
            AITask.Status.POSTPROCESSING,
        ]
        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")),
                "triview_of": payload.get("triview_of") or "",
                "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, purged_at__isnull=True)
                .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, purged_at__isnull=True)
            .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:
            internal_kind = "user_credit_insufficient" if str(exc).strip().lower() == "insufficient credit" else ""
            public_error = classify_generation_error(
                exc, operation="triview_generate", internal_kind=internal_kind
            )
            return Response(
                {"detail": public_error.fallback_message, "error": public_error.as_dict()},
                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,
        多出且尾部是「从未生成过」的段则裁掉(已生成的段绝不动)。

        时长同步:脚本镜主流程固定 15 秒,而出片、计价、时间线读的都是
        VideoSegment.target_duration_seconds。不同步的话脚本写了 15 秒、出片仍按旧默认跑。
        已出片的段不改时长——改了会跟已渲染的成片对不上。"""
        sync_video_segments_to_script(project, script)

    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:
                # 单镜必须落在 4–15 秒:出片模型的硬上限,越界下游直接拒片
                segment.duration_seconds = max(
                    SEGMENT_DURATION_MIN, min(SEGMENT_DURATION_MAX, 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=max(
                SEGMENT_DURATION_MIN,
                min(SEGMENT_DURATION_MAX, int(request.data.get("duration_seconds") or SEGMENT_DURATION_MAX)),
            ),
            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":
            finish_storyboard_stage(project)
        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,
        # 任务仍保留原始错误供排障，但普通用户仅收到安全错误对象，不让 500 HTML 导致前端白屏。
        try:
            submit_video_segment(
                video_segment=segment,
                user=request.user,
                prompt=request.data.get("prompt", ""),
                model_config_id=request.data.get("model_config_id") or None,
                aspect_ratio=request.data.get("aspect_ratio") or None,
                resolution=request.data.get("resolution") or None,
            )
        except Exception as exc:  # noqa: BLE001
            public_error = classify_generation_error(exc, operation="video_generate")
            return Response(
                {"detail": public_error.fallback_message, "error": public_error.as_dict()},
                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):
        """把已出片的各场视频用 ffmpeg 拼成一条成片。HTTP 秒回,拼接在后台线程跑,
        前端轮询 poll-export 取进度 / 成片地址。"""
        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},
        )
        # 清理僵尸任务:后台线程跑导出时若 Pod 被重启/OOM 杀掉,旧任务会永远停在 RUNNING、
        # 前端无限轮询。新建前把本时间线超过 5 分钟仍未结束的旧任务标失败,避免界面一直转圈。
        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="任务中断(服务重启或超时),已自动失败,请重新导出")
        # 防重复合成:仍在跑的任务直接复用,不再新建 —— 刷新页面 / 连点两次 / 开两个标签页
        # 都只会有一条 ffmpeg 在跑(拼接吃 CPU,跑两遍纯属浪费,且两条任务会互相覆盖成片)。
        inflight = ExportJob.objects.filter(
            timeline=timeline,
            status__in=[ExportJob.Status.QUEUED, ExportJob.Status.RUNNING],
        ).order_by("-created_at").first()
        if inflight is not None:
            return Response(ExportJobSerializer(inflight).data, status=status.HTTP_202_ACCEPTED)
        if VIDEO_IS_FINAL_STAGE:
            # V1 没有剪辑台(第 5 阶段雪藏),时间线纯由后端按「当前采用的片段」生成。
            # 每次合成都重建:某场重跑/换采用版之后再合成,成片必须跟着换,不能沿用上次的旧片段。
            timeline.clips.all().delete()
        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
        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"])
        if not VIDEO_IS_FINAL_STAGE:
            # V1 视频是末阶段:一旦把 current_stage 推到 export,流水线页会落到被雪藏的第 5 阶段
            # (前端 STAGE_STEPS 只有 4 格)→ 用户点完合成就掉进一个看不见的页面。故 V1 留在视频阶段,
            # 合成进度由 poll-export 内联展示。V2 恢复剪辑台后此处照旧推进。
            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
        # 必须是真能播的视频文件才下发:演示种子数据里有「导出标成功、挂的却是张 PNG 海报」的成片,
        # 直接下发会让「播放成片」打开一个放不出来的图(与 serializers._final_video_url 同一把闸)。
        if output is not None and is_playable_video(output):
            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="extract-script-file", parser_classes=[MultiPartParser, FormParser])
    def extract_script_file_action(self, request, pk=None):
        """上传脚本取正文:docx / txt → 纯文本。只读不落库,交前端填进对话框再走脚本 agent。"""
        self.get_object()  # 走一遍团队权限校验
        upload = request.data.get("file")
        if upload is None:
            return Response({"detail": "no file uploaded"}, status=status.HTTP_400_BAD_REQUEST)
        try:
            name, text = extract_script_text(upload)
        except ScriptFileError as exc:
            return Response({"detail": str(exc)}, status=status.HTTP_400_BAD_REQUEST)
        return Response({"name": name, "chars": len(text), "text": text})

    @action(detail=True, methods=["post"], url_path="extract-script-video", parser_classes=[MultiPartParser, FormParser])
    def extract_script_video_action(self, request, pk=None):
        """上传视频提炼:参考视频 → 中文分镜稿。

        与 extract-script-file 同形状(只读不落脚本),交前端填进对话框让用户逐镜改,
        改完再走脚本 agent 出结构化稿。差别是这条要真调模型,故按一次文本任务计费。
        """
        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)
        try:
            result = digest_project_video(
                project=project,
                user=request.user,
                upload=upload,
                model_config_id=request.data.get("model_config_id") or None,
            )
        except VideoDigestError as exc:
            return Response({"detail": str(exc)}, status=status.HTTP_400_BAD_REQUEST)
        except Exception as exc:  # noqa: BLE001 — 模型/网络失败:走统一安全文案,不回传原始异常
            public_error = classify_generation_error(exc, operation="video_digest")
            return Response(
                {"detail": public_error.fallback_message, "error": public_error.as_dict()},
                status=status.HTTP_502_BAD_GATEWAY,
            )
        return Response({"name": getattr(upload, "name", "") or "参考视频", **result})

    @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:
            message = str(exc)
            internal_kind = (
                "user_credit_insufficient" if "insufficient credit" in message.lower()
                else "model_unavailable" if "audio model" in message.lower()
                else "invalid_input"
            )
            public_error = classify_generation_error(
                exc,
                operation="voiceover_generate",
                internal_kind=internal_kind if not isinstance(exc, TtsNotConfigured) else "",
            )
            return Response(
                {"detail": public_error.fallback_message, "error": public_error.as_dict()},
                status=status.HTTP_400_BAD_REQUEST,
            )
        except Exception as exc:  # noqa: BLE001 — TTS 上游异常已在任务中留原始信息，HTTP 仅返回安全提示。
            public_error = classify_generation_error(exc, operation="voiceover_generate")
            return Response(
                {"detail": public_error.fallback_message, "error": public_error.as_dict()},
                status=status.HTTP_502_BAD_GATEWAY,
            )
        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)


class ScriptTemplateViewSet(TeamScopedViewSetMixin, ModelViewSet):
    """套路模板库(团队隔离)。建模板走 projects/{id}/save-as-template/,这里只做 列 / 改名 / 删。"""

    queryset = ScriptTemplate.objects.select_related("source_project").all()
    serializer_class = ScriptTemplateSerializer
    http_method_names = ["get", "patch", "delete", "head", "options"]

    def perform_update(self, serializer):
        name = str(serializer.validated_data.get("name") or "").strip()
        if len(name) < 2:
            raise ValidationError({"detail": "模板名至少 2 个字"})
        if ScriptTemplate.objects.filter(team=self.get_team(), name=name).exclude(id=serializer.instance.id).exists():
            raise ValidationError({"detail": "模板名已存在,换一个"})
        serializer.save(name=name)