feat(core/backend): pipeline continuity + threaded ffmpeg burn-in export + upload/save-timeline

Video pipeline (script→assets→storyboard→video→stitch): - robust split_script_into_segments (4 non-empty scenes), scene-aware storyboard/video prompts - link VideoSegment→ScriptSegment + storyboard-frame reference image (graceful text fallback) - idempotent poll_video_segment (no double-charge on repeated polling) - threaded export (no Celery worker needed) + poll-export endpoint - run_export_job rewritten to filter_complex: per-clip trim, xfade transitions, subtitle burn-in (Pillow PNG overlay; this ffmpeg lacks libass), BGM mix - upload-video-segment / upload-bgm / save-timeline endpoints - serializers embed asset preview URLs (beat assets pagination); Pillow added to requirements Also includes prior uncommitted backend work: account preferences/sessions, billing trend, product/asset endpoints, accounts 0002 migration. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-06-09 14:46:16 +08:00
parent 8959946241
commit 92826dec14
13 changed files with 1229 additions and 97 deletions
@@ -1,4 +1,6 @@
+import re
 import uuid
+from datetime import timedelta
 from decimal import Decimal
 from django.db import transaction
 from django.utils import timezone
@@ -59,13 +61,43 @@ def build_script_prompt(*, project, user_prompt: str, selling_point_ids: list[st
    return [{"role": "system", "content": system}, {"role": "user", "content": user}]


-def split_script_into_segments(content: str) -> list[str]:
-    blocks = [line.strip() for line in content.splitlines() if line.strip()]
-    if len(blocks) >= 4:
-        return blocks[:4]
-    if not content.strip():
-        return [""] * 4
-    return [content.strip()] + [""] * (4 - len(blocks or [content]))
+def split_script_into_segments(content: str, count: int = 4) -> list[str]:
+    """把一段脚本稳健地拆成 `count` 个分镜文本,保证每镜都非空、且所有内容都被分配到某一镜。
+
+    原实现按行 `[:4]`,ARK 返回整段散文时常变成「第1镜有词、2/3/4镜全空」,
+    导致后续故事板帧 / 视频段拿到空提示词,前后内容断裂。这里改为:
+    优先按空行/标号块切,块数够就把全部块均匀分桶;块不够再按句子切;仍不够则补齐。
+    """
+
+    def _bucketize(items: list[str], joiner: str) -> list[str]:
+        buckets: list[list[str]] = [[] for _ in range(count)]
+        per = len(items) / count
+        for index, item in enumerate(items):
+            buckets[min(count - 1, int(index / per))].append(item)
+        return [joiner.join(bucket).strip() for bucket in buckets]
+
+    text = (content or "").strip()
+    if not text:
+        return [""] * count
+
+    # 1) 优先按空行分段;只有一段时退回按行分
+    blocks = [block.strip() for block in re.split(r"\n\s*\n", text) if block.strip()]
+    if len(blocks) < 2:
+        blocks = [line.strip() for line in text.splitlines() if line.strip()]
+    if len(blocks) >= count:
+        return _bucketize(blocks, "\n")
+
+    # 2) 段落不足:按中英文句末标点切句,再均匀分桶
+    sentences = [s.strip() for s in re.split(r"(?<=[。！？!?.;；\n])", text) if s.strip()]
+    if len(sentences) >= count:
+        return _bucketize(sentences, " ")
+
+    # 3) 仍不足:用已有块/句补齐到 count,绝不留空镜
+    base = blocks or sentences or [text]
+    filled = list(base)
+    while len(filled) < count:
+        filled.append(base[-1])
+    return filled[:count]


@transaction.atomic
@@ -242,30 +274,98 @@ def generate_base_asset(*, project, user, kind: str, prompt: str) -> BaseAssetGr
        raise


-def generate_storyboard(*, project, user, prompt: str = "") -> StoryboardVersion:
+def _scene_context(project) -> str:
+    """从商品 + 已采用基础资产提炼一句「风格锚点」,贯穿故事板 / 视频,保证各镜内容一致。"""
+    product = project.product
+    parts = [f"商品:{product.title}"]
+    if product.brand:
+        parts.append(f"品牌:{product.brand}")
+    if product.category:
+        parts.append(f"类目:{product.category}")
+    if getattr(product, "target_audience", ""):
+        parts.append(f"人群:{product.target_audience}")
+    adopted_kinds = set(
+        project.base_asset_groups.filter(adopted_asset__isnull=False).values_list("kind", flat=True)
+    )
+    if BaseAssetGroup.Kind.PERSON in adopted_kinds:
+        parts.append("真人出镜,保持人物一致")
+    if BaseAssetGroup.Kind.SCENE in adopted_kinds:
+        parts.append("统一场景与色调")
+    return " · ".join(parts)
+
+
+def build_storyboard_frame_prompt(project, version, segment) -> str:
+    """单帧故事板提示词:风格锚点 + 本镜画面(回退旁白)+ 版本统一指令。"""
+    visual = (segment.visual_prompt or segment.narration or "").strip()
+    lines = [
+        _scene_context(project),
+        f"第 {segment.sort_order + 1} 镜画面:{visual}" if visual else f"第 {segment.sort_order + 1} 镜",
+    ]
+    if version.prompt:
+        lines.append(version.prompt.strip())
+    lines.append("电商竖屏分镜图,构图清晰,可直接指导视频生成")
+    return "\n".join(line for line in lines if line)
+
+
+def build_video_segment_prompt(project, video_segment, scene, user_prompt: str) -> str:
+    """单段视频提示词:把本镜旁白 + 画面 + 风格锚点织进去,让每个视频片段跟住对应脚本/故事板。"""
+    lines = [_scene_context(project)]
+    if scene is not None:
+        if scene.narration:
+            lines.append(f"旁白:{scene.narration.strip()}")
+        visual = (scene.visual_prompt or scene.narration or "").strip()
+        if visual:
+            lines.append(f"画面:{visual}")
+    if user_prompt:
+        lines.append(user_prompt.strip())
+    lines.append(
+        f"第 {video_segment.sort_order + 1} 段 · {video_segment.target_duration_seconds}s · "
+        "9:16 竖屏电商带货短视频,镜头稳定,商品露出清晰,节奏有转化感"
+    )
+    return "\n".join(line for line in lines if line)
+
+
+def submit_storyboard(*, project, user, prompt: str = "") -> StoryboardVersion:
+    """异步故事板·提交:快速创建(或复用)一个未采用的版本,不在此处生图。逐帧生成交给 generate_storyboard_frame(轮询)。"""
    adopted_script = project.script_versions.filter(is_adopted=True).prefetch_related("segments").first()
    if adopted_script is None:
        raise ValueError("script must be adopted before generating storyboard")
-    model_config = get_default_model(ModelConfig.Capability.IMAGE)
-    if model_config is None:
+    if get_default_model(ModelConfig.Capability.IMAGE) is None:
        raise ValueError("no active image model configured")
+    # 复用尚未完成(未采用)的版本,避免重复提交产生多版本;否则新建
+    version = project.storyboard_versions.filter(is_adopted=False).order_by("-created_at").first()
+    if version is None:
+        version = StoryboardVersion.objects.create(project=project, prompt=prompt)
+    elif prompt and version.prompt != prompt:
+        version.prompt = prompt
+        version.save(update_fields=["prompt", "updated_at"])
+    return version

-    storyboard = StoryboardVersion.objects.create(project=project, prompt=prompt)
-    provider = VolcanoArkProvider(base_url=model_config.provider.base_url or None)
-    for segment in adopted_script.segments.all():
-        task = create_ai_task(
-            project=project,
-            user=user,
-            task_type=AITask.Type.STORYBOARD,
-            model_config=model_config,
-            request_payload={"model": model_config.name, "endpoint": model_config.endpoint, "prompt": segment.visual_prompt},
-        )
+
+def _storyboard_frame_worker(task_id, version_id, segment_id, user_id) -> None:
+    """后台线程:真正调 ARK 生成一帧故事板图并落库。每次 poll 不阻塞在此——HTTP 永远秒回。"""
+    import threading  # noqa: F401 — 仅标注此函数运行在独立线程
+    from django.db import connections
+
+    from apps.accounts.models import User
+
+    try:
+        task = AITask.objects.select_related("model_config__provider").get(id=task_id)
+        version = StoryboardVersion.objects.select_related("project__team").get(id=version_id)
+        segment = ScriptSegment.objects.get(id=segment_id)
+        user = User.objects.get(id=user_id)
+        project = version.project
+        model_config = task.model_config
        reservation = task.credit_reservation
+        task.status = AITask.Status.SUBMITTED
+        task.save(update_fields=["status", "updated_at"])
        try:
+            provider = VolcanoArkProvider(base_url=model_config.provider.base_url or None)
+            frame_prompt = task.request_payload.get("prompt") or build_storyboard_frame_prompt(project, version, segment)
            response = provider.image_generation(
                model=model_config.name,
                endpoint=model_config.endpoint,
-                prompt=f"{prompt}\n{segment.visual_prompt}".strip(),
+                prompt=frame_prompt,
            )
            media = provider.extract_first_media_url(response)
            task.status = AITask.Status.SUCCEEDED
@@ -285,22 +385,141 @@ def generate_storyboard(*, project, user, prompt: str = "") -> StoryboardVersion
                asset_type=Asset.Type.IMAGE,
            )
            StoryboardFrame.objects.create(
-                storyboard=storyboard,
+                storyboard=version,
                script_segment=segment,
                asset=asset,
                sort_order=segment.sort_order,
                prompt=segment.visual_prompt,
            )
-        except Exception as exc:
+        except Exception as exc:  # noqa: BLE001 — 失败回滚额度,标记任务失败供 poll 上报
            task.status = AITask.Status.FAILED
            task.error_message = str(exc)
            task.completed_at = timezone.now()
            task.save(update_fields=["status", "error_message", "completed_at", "updated_at"])
            release_credit(reservation=reservation, reason=str(exc))
-            raise
-    storyboard.is_adopted = True
-    storyboard.save(update_fields=["is_adopted", "updated_at"])
-    return storyboard
+    finally:
+        connections.close_all()  # 释放该线程的 DB 连接
+
+
+def generate_storyboard_frame(*, project, user) -> dict:
+    """异步故事板·轮询(秒回):读取进度;若无帧在生成则后台起线程生成下一帧。永不阻塞在 ARK 调用上。
+    返回 {status: generating|succeeded|failed, done, total, version_id}。全部完成→采用版本。"""
+    import threading
+
+    version = project.storyboard_versions.filter(is_adopted=False).order_by("-created_at").first()
+    adopted_script = project.script_versions.filter(is_adopted=True).prefetch_related("segments").first()
+    if version is None or adopted_script is None:
+        latest = project.storyboard_versions.order_by("-created_at").first()
+        n = latest.frames.count() if latest else 0
+        return {"status": "succeeded", "done": n, "total": n, "version_id": str(latest.id) if latest else ""}
+
+    segments = list(adopted_script.segments.all().order_by("sort_order"))
+    total = len(segments)
+    done_segment_ids = set(version.frames.values_list("script_segment_id", flat=True))
+    done = len(done_segment_ids)
+
+    if done >= total:
+        _finalize_storyboard(project, version)
+        return {"status": "succeeded", "done": total, "total": total, "version_id": str(version.id)}
+
+    # 该版本内是否已有帧在后台生成中(RESERVED/SUBMITTED 的故事板任务即为「占位锁」)。
+    # 仅算「近 3 分钟内」的任务:若进程/线程意外中断留下僵尸任务,超时后不再视为在生成,允许重新发起。
+    stale_cutoff = timezone.now() - timedelta(minutes=3)
+    inflight = AITask.objects.filter(
+        project=project,
+        task_type=AITask.Type.STORYBOARD,
+        status__in=[AITask.Status.CREATED, AITask.Status.RESERVED, AITask.Status.SUBMITTED],
+        request_payload__storyboard_version=str(version.id),
+        created_at__gte=stale_cutoff,
+    ).exists()
+    if inflight:
+        return {"status": "generating", "done": done, "total": total, "version_id": str(version.id)}
+
+    pending = [s for s in segments if s.id not in done_segment_ids]
+    segment = pending[0]
+    # 单帧失败次数上限,避免持续失败时无限重试
+    failed_for_segment = AITask.objects.filter(
+        project=project,
+        task_type=AITask.Type.STORYBOARD,
+        status=AITask.Status.FAILED,
+        request_payload__storyboard_segment=str(segment.id),
+    ).count()
+    if failed_for_segment >= 2:
+        last = AITask.objects.filter(project=project, task_type=AITask.Type.STORYBOARD, status=AITask.Status.FAILED,
+                                     request_payload__storyboard_segment=str(segment.id)).order_by("-created_at").first()
+        return {"status": "failed", "done": done, "total": total, "version_id": str(version.id),
+                "error": last.error_message if last else "storyboard frame failed"}
+
+    model_config = get_default_model(ModelConfig.Capability.IMAGE)
+    task = create_ai_task(
+        project=project,
+        user=user,
+        task_type=AITask.Type.STORYBOARD,
+        model_config=model_config,
+        request_payload={
+            "model": model_config.name,
+            "endpoint": model_config.endpoint,
+            "prompt": build_storyboard_frame_prompt(project, version, segment),
+            "storyboard_version": str(version.id),
+            "storyboard_segment": str(segment.id),
+        },
+    )
+    threading.Thread(
+        target=_storyboard_frame_worker,
+        args=(str(task.id), str(version.id), str(segment.id), str(user.id)),
+        daemon=True,
+    ).start()
+    return {"status": "generating", "done": done, "total": total, "version_id": str(version.id)}
+
+
+def _finalize_storyboard(project, version) -> None:
+    """全部帧就绪:采用该版本(反采用其余版本)。项目阶段推进由视图负责(与原同步实现一致)。"""
+    project.storyboard_versions.exclude(id=version.id).update(is_adopted=False)
+    if not version.is_adopted:
+        version.is_adopted = True
+        version.save(update_fields=["is_adopted", "updated_at"])
+
+
+def _asset_preview_url(asset) -> str:
+    """资产主文件的可公开访问 URL(已写绝对 URL 优先,否则实时签 TOS GET)。"""
+    if asset is None:
+        return ""
+    primary = asset.files.filter(is_primary=True).first() or asset.files.first()
+    if primary is None:
+        return ""
+    if primary.preview_url:
+        return primary.preview_url
+    try:
+        return TosStorage().presigned_get_url(object_key=primary.object_key)
+    except Exception:
+        return ""
+
+
+def _video_reference_images(project, video_segment) -> list[str]:
+    """为本视频段挑一张视觉参考图:优先本镜故事板帧,兜底已采用商品基础资产。"""
+    version = (
+        project.storyboard_versions.filter(is_adopted=True).order_by("-created_at").first()
+        or project.storyboard_versions.order_by("-created_at").first()
+    )
+    if version is not None:
+        frame = (
+            version.frames.filter(sort_order=video_segment.sort_order).first()
+            or version.frames.order_by("sort_order").first()
+        )
+        if frame is not None:
+            url = _asset_preview_url(frame.asset)
+            if url:
+                return [url]
+    product_group = (
+        project.base_asset_groups.filter(kind=BaseAssetGroup.Kind.PRODUCT, adopted_asset__isnull=False)
+        .order_by("-created_at")
+        .first()
+    )
+    if product_group is not None:
+        url = _asset_preview_url(product_group.adopted_asset)
+        if url:
+            return [url]
+    return []


 def submit_video_segment(*, video_segment: VideoSegment, user, prompt: str) -> VideoSegmentVersion | None:
@@ -308,6 +527,20 @@ def submit_video_segment(*, video_segment: VideoSegment, user, prompt: str) -> V
    if model_config is None:
        raise ValueError("no active video model configured")
    project = video_segment.project
+
+    # 衔接:按 sort_order 把视频段绑到对应脚本镜,并织出跟住该镜的提示词。
+    scene = None
+    adopted_script = project.script_versions.filter(is_adopted=True).prefetch_related("segments").first()
+    if adopted_script is not None:
+        scene = adopted_script.segments.filter(sort_order=video_segment.sort_order).first()
+        if scene is not None and video_segment.script_segment_id != scene.id:
+            video_segment.script_segment = scene
+            video_segment.save(update_fields=["script_segment", "updated_at"])
+    final_prompt = build_video_segment_prompt(project, video_segment, scene, prompt)
+
+    # 参考图:优先用本镜故事板帧,其次商品/人物基础资产,给视频做视觉锚点(衔接故事板→视频)。
+    reference_images = _video_reference_images(project, video_segment)
+
    task = create_ai_task(
        project=project,
        user=user,
@@ -316,22 +549,38 @@ def submit_video_segment(*, video_segment: VideoSegment, user, prompt: str) -> V
        request_payload={
            "model": model_config.name,
            "endpoint": model_config.endpoint,
-            "prompt": prompt,
+            "prompt": final_prompt,
            "duration": video_segment.target_duration_seconds,
            "ratio": "9:16",
            "video_segment_id": str(video_segment.id),
+            "reference_images": reference_images,
        },
    )
    try:
        provider = VolcanoArkProvider(base_url=model_config.provider.base_url or None)
-        response = provider.create_video_task(
-            model=model_config.name,
-            endpoint=model_config.endpoint,
-            prompt=prompt,
-            duration=video_segment.target_duration_seconds,
-            ratio="9:16",
-            resolution="720p",
-        )
+        try:
+            response = provider.create_video_task(
+                model=model_config.name,
+                endpoint=model_config.endpoint,
+                prompt=final_prompt,
+                duration=video_segment.target_duration_seconds,
+                ratio="9:16",
+                resolution="720p",
+                reference_images=reference_images or None,
+            )
+        except Exception:
+            # 降级:带参考图被拒时退回纯文生视频(文本里已含本镜旁白/画面,衔接不丢)
+            if not reference_images:
+                raise
+            response = provider.create_video_task(
+                model=model_config.name,
+                endpoint=model_config.endpoint,
+                prompt=final_prompt,
+                duration=video_segment.target_duration_seconds,
+                ratio="9:16",
+                resolution="720p",
+                reference_images=None,
+            )
        task.provider_task_id = str(response.get("id") or response.get("task_id") or "")
        task.response_payload = response
        task.status = AITask.Status.SUBMITTED
@@ -353,6 +602,12 @@ def submit_video_segment(*, video_segment: VideoSegment, user, prompt: str) -> V


 def poll_video_segment(*, video_segment: VideoSegment, user) -> VideoSegmentVersion | None:
+    # 幂等:已完成的段直接回采用版;已失败的段不再 poll。避免对已成功 task 再 poll → 二次建版 / 二次扣费。
+    if video_segment.status == VideoSegment.Status.SUCCEEDED:
+        return video_segment.adopted_version or video_segment.versions.order_by("-created_at").first()
+    if video_segment.status == VideoSegment.Status.FAILED:
+        return None
+
    task = video_segment.versions.order_by("-created_at").first()
    ai_task = None
    if task:
@@ -366,6 +621,12 @@ def poll_video_segment(*, video_segment: VideoSegment, user) -> VideoSegmentVers
    if ai_task is None:
        raise ValueError("no active video generation task")

+    # task 已终态(可能被并发的 worker / 另一次 poll 处理过):直接回已有版,不再调 ARK。
+    if ai_task.status == AITask.Status.SUCCEEDED:
+        return video_segment.versions.filter(task=ai_task).order_by("-created_at").first()
+    if ai_task.status in (AITask.Status.FAILED, AITask.Status.CANCELLED):
+        return None
+
    provider = VolcanoArkProvider(base_url=ai_task.model_config.provider.base_url or None)
    response = provider.poll_video_task(endpoint=ai_task.model_config.endpoint, provider_task_id=ai_task.provider_task_id)
    remote_status = response.get("status")