大量修改二期功能清单内容

core/backend/apps/ai/free_video.py

@@ -99,6 +99,31 @@ def _refresh_processing_free_asset(free_asset: FreeAsset) -> bool:
     return False
 
 
+def _guard_asset_reference(asset: Asset, label: str) -> None:
+    """三库引用前的审核闸(模块4 · 4.2)。
+
+    平台生成的资产免审直接过;用户上传的必须真过一遍审核才能当生成参考。
+    判据是 Asset.source,**不是**「在不在资产库里」—— 按库免审等于把审核架空:
+    用户传一张图进库、再从自由创作引用出去,就绕过了整套人像审核。
+    """
+    from apps.assets.review import poll_asset_review, reference_review_state, submit_asset_for_review
+
+    state = reference_review_state(asset)
+    if state == "processing":
+        poll_asset_review(asset)  # 实时刷一次,别让用户干等下一轮轮询
+        state = reference_review_state(asset)
+    if state == "allowed":
+        return
+    name = label or asset.name or "未命名"
+    if state == "processing":
+        raise ValueError(f"素材「{name}」正在审核中，请稍后再引用")
+    if state == "failed":
+        raise ValueError(f"素材「{name}」未通过审核，不能用作生成参考")
+    # 从没送过审(资产库上传不自动送审):这里补送一次,用户等审核结果即可,不必回去手动点
+    submit_asset_for_review(asset, force=True)
+    raise ValueError(f"素材「{name}」是上传素材，已提交审核，通过后即可引用")
+
+
 def build_content_items(*, team, prompt: str, mode: str, references: list) -> dict:
     """references → 火山 content_items + api_prompt(@label 已替换)。
 
@@ -244,6 +269,32 @@ def build_content_items(*, team, prompt: str, mode: str, references: list) -> di
                 _remember_library_asset(fa)
             continue
 
+        # 三库引用(模块4 · 4.1):资产库 / 模特库 / 商品库 挑出来的东西最终都是一行 Asset,
+        # 所以后端只认一种 source=asset,三个库的差别全在前端的 picker 上。
+        if source == "asset" and ref.get("asset_id"):
+            asset = Asset.objects.filter(id=ref["asset_id"], team=team, is_deleted=False).first()
+            if asset is None:
+                raise ValueError(f"素材「{label or '未命名'}」不存在或已被删除")
+            _guard_asset_reference(asset, label)
+            from .services import _asset_preview_url, _seedance_ref_url
+
+            raw_url = _asset_preview_url(asset)
+            if not raw_url:
+                raise ValueError(f"素材「{label or asset.name}」没有可用文件，无法引用")
+            # 已登记火山素材库的走 asset://(写实人脸走直链会被 InputImageSensitiveContentDetected 拒)
+            resolved_url = _seedance_ref_url(raw_url, asset.review_status, asset.review_remote_id)
+            kind = asset.asset_type if asset.asset_type in {"image", "video", "audio"} else "image"
+            if mode == "keyframe":
+                if kind != "image":
+                    raise ValueError("首尾帧模式仅支持图片素材")
+                effective_role = role if role in {"first_frame", "last_frame"} else "first_frame"
+            else:
+                effective_role = "reference_video" if kind == "video" else ("reference_audio" if kind == "audio" else "reference_image")
+            asset_type = _push(kind, resolved_url, effective_role, duration)
+            if label and label not in label_to_placeholder:
+                label_to_placeholder[label] = _placeholder_for(asset_type)
+            continue
+
         # 直传素材(已上传 TOS 的直链)
         if ref_type == "image":
             # 参考图模式下所有图 role 必须 reference_image;keyframe 用 first_frame/last_frame

core/backend/apps/ai/migrations/0030_alter_aitask_task_type.py

@@ -0,0 +1,18 @@
+# Generated by Django 5.1.15 on 2026-08-17 09:03
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('ai', '0029_aimodelattempt'),
+    ]
+
+    operations = [
+        migrations.AlterField(
+            model_name='aitask',
+            name='task_type',
+            field=models.CharField(choices=[('script_generation', 'Script Generation'), ('script_optimization', 'Script Optimization'), ('entity_extraction', 'Entity Extraction'), ('video_digest', 'Video Digest'), ('product_image', 'Product Image'), ('person_image', 'Person Image'), ('model_triview', 'Model Triview'), ('scene_image', 'Scene Image'), ('storyboard', 'Storyboard'), ('video_segment', 'Video Segment'), ('voiceover', 'Voiceover'), ('export', 'Export'), ('free_video', 'Free Video')], max_length=48),
+        ),
+    ]

core/backend/apps/ai/models.py

@@ -92,6 +92,7 @@ class AITask(TeamOwnedModel):
         SCRIPT_GENERATION = "script_generation", "Script Generation"
         SCRIPT_OPTIMIZATION = "script_optimization", "Script Optimization"
         ENTITY_EXTRACTION = "entity_extraction", "Entity Extraction"
+        VIDEO_DIGEST = "video_digest", "Video Digest"  # 上传视频提炼:参考视频 → 分镜稿
         PRODUCT_IMAGE = "product_image", "Product Image"
         PERSON_IMAGE = "person_image", "Person Image"
         MODEL_TRIVIEW = "model_triview", "Model Triview"

core/backend/apps/ai/script_agent.py

@@ -19,6 +19,7 @@ SSE 事件(每帧 `data: {json}\n\n`,json 带 type):
 from __future__ import annotations
 
 import json
+import math
 import re
 from decimal import Decimal
 from functools import lru_cache
@@ -34,7 +35,70 @@ from apps.billing.services.ledger import charge_reserved_credit, release_credit
 VALID_TONES = ["种草", "测评", "剧情", "痛点"]
 VALID_ROLES = ["钩子", "痛点", "卖点", "CTA"]
 VALID_ENTITY_TYPES = ["character", "scene", "product"]
-DURATION_TIERS = [15, 30, 60, 90]
+
+# 时长:总时长 5–60 秒按 5 秒步进;单镜 4–15 秒(15 是出片模型硬上限,越界下游直接拒片)。
+TOTAL_DURATION_MIN = 5
+TOTAL_DURATION_MAX = 60
+TOTAL_DURATION_STEP = 5
+SEGMENT_DURATION_MIN = 4
+SEGMENT_DURATION_MAX = 15
+DEFAULT_TOTAL_DURATION = 30
+
+# 表现形式 × 视频结构(二期)。key 用 ASCII 找套路文件,label 是给模型和用户看的中文。
+PRESENTATION_FORMATS: dict[str, str] = {"oral": "口播", "drama": "短剧", "vlog": "Vlog"}
+VIDEO_STRUCTURES: dict[str, str] = {
+    "pain": "痛点解决",
+    "contrast": "前后对比",
+    "review": "测评验证",
+    "scene": "场景种草",
+}
+DEFAULT_PRESENTATION_FORMAT = "oral"
+DEFAULT_VIDEO_STRUCTURE = "pain"
+
+# 唯一禁用组合:短剧 × 测评验证。演出来的实测没有可信度,详见 playbooks/combo-matrix.md。
+FORBIDDEN_COMBOS: set[tuple[str, str]] = {("drama", "review")}
+
+# 表现形式推荐的单镜节奏(秒)。镜数 ≈ 总时长 / 该值,再夹到 4–15 秒的合法区间。
+FORMAT_SHOT_PACE: dict[str, int] = {"oral": 12, "drama": 8, "vlog": 7}
+# 表现形式推荐的默认总时长:口播短平快,短剧要装下三幕,Vlog 要铺氛围。
+FORMAT_DEFAULT_DURATION: dict[str, int] = {"oral": 30, "drama": 45, "vlog": 30}
+# 各结构能压到的最短总时长(低于此值证据/氛围不成立),见 playbooks/combo-matrix.md。
+STRUCTURE_MIN_DURATION: dict[str, int] = {"pain": 15, "contrast": 10, "review": 20, "scene": 20}
+
+# 可懂语速上限 3.5 字/秒 —— 旁白字数按这一镜自己的秒数算,不再全场 55 字一刀切。
+NARRATION_CHARS_PER_SECOND = 3.5
+NARRATION_CHARS_HARD_CAP = 55
+
+
+_FORMAT_KEY_BY_LABEL = {label: key for key, label in PRESENTATION_FORMATS.items()}
+_STRUCTURE_KEY_BY_LABEL = {label: key for key, label in VIDEO_STRUCTURES.items()}
+
+
+def combo_keys(value_format, value_structure) -> tuple[str, str]:
+    """把「中文标签或 ASCII key」都归一成 key。落库存的是中文,请求传的是 key,两边都要认。"""
+    fmt = _FORMAT_KEY_BY_LABEL.get(value_format, value_format)
+    structure = _STRUCTURE_KEY_BY_LABEL.get(value_structure, value_structure)
+    return coerce_combo(fmt, structure)
+
+
+def allowed_structures(fmt: str) -> list[str]:
+    """某表现形式下可选的视频结构 key(1.8 组合联动:换表现形式,结构列表跟着变)。"""
+    fmt = fmt if fmt in PRESENTATION_FORMATS else DEFAULT_PRESENTATION_FORMAT
+    return [key for key in VIDEO_STRUCTURES if (fmt, key) not in FORBIDDEN_COMBOS]
+
+
+def coerce_combo(fmt: str | None, structure: str | None) -> tuple[str, str]:
+    """把任意输入夹成一组合法的(表现形式, 视频结构)。禁用组合回落到该形式的第一个合法结构。"""
+    fmt = fmt if fmt in PRESENTATION_FORMATS else DEFAULT_PRESENTATION_FORMAT
+    structure = structure if structure in VIDEO_STRUCTURES else DEFAULT_VIDEO_STRUCTURE
+    if (fmt, structure) in FORBIDDEN_COMBOS:
+        structure = allowed_structures(fmt)[0]
+    return fmt, structure
+
+
+def narration_limit(duration: int) -> int:
+    """这一镜旁白的字数上限:秒数 × 3.5,且不超过硬上限 55。"""
+    return max(1, min(NARRATION_CHARS_HARD_CAP, int(duration * NARRATION_CHARS_PER_SECOND)))
 
 
 # --------------------------------------------------------------------------- #
@@ -53,9 +117,18 @@ def _skill_dir() -> Path:
     return base / "skills" / "ecommerce-video-script"
 
 
+def _read_ref(path: Path, label: str) -> str:
+    if not path.exists():
+        return ""
+    return f"\n\n===== {label} =====\n\n{path.read_text(encoding='utf-8')}"
+
+
 @lru_cache(maxsize=1)
-def load_ecommerce_skill() -> str:
-    """读取 SKILL.md + 全部 references 拼成系统提示词(领域知识)。缺文件不致命,尽量给。"""
+def _load_skill_base() -> str:
+    """SKILL.md + references 根目录下的通用资料(方法论/钩子库/品类/平台/自检),每次都要。
+
+    playbooks/ 是子目录,glob("*.md") 不会递归到,套路由 load_ecommerce_skill 按组合单独挑。
+    """
     skill_dir = _skill_dir()
     parts: list[str] = []
     main = skill_dir / "SKILL.md"
@@ -64,11 +137,33 @@ def load_ecommerce_skill() -> str:
     ref_dir = skill_dir / "references"
     if ref_dir.exists():
         for ref in sorted(ref_dir.glob("*.md")):
-            parts.append(f"\n\n===== references/{ref.name} =====\n\n{ref.read_text(encoding='utf-8')}")
-    if not parts:
+            parts.append(_read_ref(ref, f"references/{ref.name}"))
+    return "".join(parts)
+
+
+@lru_cache(maxsize=32)
+def load_ecommerce_skill(
+    presentation_format: str = DEFAULT_PRESENTATION_FORMAT,
+    video_structure: str = DEFAULT_VIDEO_STRUCTURE,
+) -> str:
+    """通用资料 + 组合矩阵 + **只挑被选中的那一份表现形式和那一份视频结构**。
+
+    套路全量灌进去会让系统提示词翻倍(每份 2-3K 字),而且模型会在 11 套互相矛盾的
+    镜头语言里挑花眼。只给当前这一组,提示词更短、约束更硬。
+    """
+    fmt, structure = coerce_combo(presentation_format, video_structure)
+    playbooks = _skill_dir() / "references" / "playbooks"
+    parts = [
+        _load_skill_base(),
+        _read_ref(playbooks / "combo-matrix.md", "references/playbooks/combo-matrix.md"),
+        _read_ref(playbooks / f"format-{fmt}.md", f"references/playbooks/format-{fmt}.md"),
+        _read_ref(playbooks / f"structure-{structure}.md", f"references/playbooks/structure-{structure}.md"),
+    ]
+    joined = "".join(parts)
+    if not joined.strip():
         # 兜底:skill 文件缺失也能退化生成(交接文档会提示补 skills 目录)
         return "你是电商带货短视频脚本生成 agent,输出结构化 ScriptDraft JSON。"
-    return "".join(parts)
+    return joined
 
 
 # 运行时输出协议:优先级高于 skill 里的「只输出 JSON / 不展示思考」,只为流式体感放开一句前言。
@@ -119,12 +214,25 @@ def build_agent_messages(
     base_draft: dict | None,
     aspect_ratio: str,
     total_duration: int,
+    presentation_format: str = DEFAULT_PRESENTATION_FORMAT,
+    video_structure: str = DEFAULT_VIDEO_STRUCTURE,
     target_index: int | None = None,
 ) -> list[dict[str, str]]:
-    system = load_ecommerce_skill() + _OUTPUT_PROTOCOL
+    fmt, structure = coerce_combo(presentation_format, video_structure)
+    total = coerce_total_duration(total_duration)
+    system = load_ecommerce_skill(fmt, structure) + _OUTPUT_PROTOCOL
+
+    # 给一组建议时长(不是硬性),模型可以按内容调整,只要每镜 4–15 秒且加总不变。
+    suggested = plan_segment_durations(total, fmt)
+    pace_hint = "+".join(str(d) for d in suggested)
     head = (
         f"【画幅】{aspect_ratio}\n"
-        f"【总时长】{total_duration} 秒(每 15 秒一镜,共 {total_duration // 15} 镜)\n"
+        f"【表现形式】{PRESENTATION_FORMATS[fmt]}(套路见 playbooks/format-{fmt}.md,已加载)\n"
+        f"【视频结构】{VIDEO_STRUCTURES[structure]}(套路见 playbooks/structure-{structure}.md,已加载)\n"
+        f"【总时长】{total} 秒\n"
+        f"【分镜时长】单镜 4–15 秒,可以不等长;各镜相加必须精确等于 {total} 秒。\n"
+        f"　建议切成 {len(suggested)} 镜({pace_hint}),这是按「{PRESENTATION_FORMATS[fmt]}」的节奏算的;\n"
+        f"　你可以按内容调整镜数与每镜长短(该长的给足、该短的压短),但必须守住上面两条硬约束。\n"
         f"【商品信息】\n{_product_context(project, selling_point_ids)}"
     )
     if mode == "revise" and base_draft and target_index is not None:
@@ -154,7 +262,8 @@ def build_agent_messages(
         )
     else:
         user = (
-            "【任务】全自动(模式①):仅凭商品与前置条件,自动定档/选 tone/造 entity/填黄金结构。\n"
+            "【任务】全自动(模式①):仅凭商品与前置条件,按指定的表现形式与视频结构套路"
+            "自动定镜数/选 tone/造 entity/填结构骨架。\n"
             f"{head}\n\n"
             "请按技能流程一次性产出 ScriptDraft。"
         )
@@ -217,14 +326,87 @@ def _extract_json(text: str) -> str | None:
     return _balanced_object(text)
 
 
-def _nearest_duration(value) -> int:
+def coerce_total_duration(value) -> int:
+    """总时长夹到 5–60 秒、5 秒步进。空值/0/非法输入一律回落默认 30。"""
+    if value in (None, "", 0):
+        return DEFAULT_TOTAL_DURATION
     try:
         value = int(value)
     except (TypeError, ValueError):
-        return 60
-    if value in DURATION_TIERS:
-        return value
-    return min(DURATION_TIERS, key=lambda t: abs(t - value))
+        return DEFAULT_TOTAL_DURATION
+    if value <= 0:
+        return DEFAULT_TOTAL_DURATION
+    value = max(TOTAL_DURATION_MIN, min(TOTAL_DURATION_MAX, value))
+    stepped = int(round(value / TOTAL_DURATION_STEP) * TOTAL_DURATION_STEP)
+    return max(TOTAL_DURATION_MIN, min(TOTAL_DURATION_MAX, stepped))
+
+
+def plan_segment_durations(total_duration: int, presentation_format: str) -> list[int]:
+    """把总时长切成每镜 4–15 秒、加总精确等于总时长的一组时长。
+
+    镜数按表现形式的推荐节奏定(口播 12s/镜、短剧 8s/镜、Vlog 7s/镜),再夹进
+    ceil(total/15) ~ total//4 的合法区间。余数摊到前面几镜,所以镜与镜之间最多差 1 秒——
+    这只是**兜底**,模型自己给的不等长时长只要合法就照用。
+    """
+    total = coerce_total_duration(total_duration)
+    fmt = presentation_format if presentation_format in FORMAT_SHOT_PACE else DEFAULT_PRESENTATION_FORMAT
+    count_min = math.ceil(total / SEGMENT_DURATION_MAX)
+    count_max = max(count_min, total // SEGMENT_DURATION_MIN)
+    count = max(1, round(total / FORMAT_SHOT_PACE[fmt]))
+    count = max(count_min, min(count_max, count))
+    base, remainder = divmod(total, count)
+    return [base + 1 if i < remainder else base for i in range(count)]
+
+
+def plan_roles(count: int) -> list[str]:
+    """镜数 → role 序列。通用规则:首钩子、次痛点、末 CTA,中间全是卖点。"""
+    if count <= 1:
+        return ["钩子"]
+    if count == 2:
+        return ["钩子", "卖点"]
+    if count == 3:
+        return ["钩子", "卖点", "CTA"]
+    return ["钩子", "痛点"] + ["卖点"] * (count - 3) + ["CTA"]
+
+
+def _fit_segment_durations(raw: list, total: int, presentation_format: str) -> list[int]:
+    """采纳模型给的每镜时长(允许不等长),非法就修;修不动就整组回落到 plan_segment_durations。
+
+    合法定义:每镜 4–15 秒的整数,且加总 == 总时长。模型很容易把总数算错一两秒,
+    所以先夹单镜范围,再把差额摊到还有余量的镜上,尽量保住模型的节奏意图。
+    """
+    if not raw:
+        return plan_segment_durations(total, presentation_format)
+
+    durations: list[int] = []
+    for value in raw:
+        try:
+            seconds = int(value)
+        except (TypeError, ValueError):
+            seconds = 0
+        durations.append(max(SEGMENT_DURATION_MIN, min(SEGMENT_DURATION_MAX, seconds or SEGMENT_DURATION_MIN)))
+
+    # 镜数本身就装不下总时长(太少会超 15s/镜,太多会低于 4s/镜)→ 模型节奏不可用,整组重排。
+    count = len(durations)
+    if not (count * SEGMENT_DURATION_MIN <= total <= count * SEGMENT_DURATION_MAX):
+        return plan_segment_durations(total, presentation_format)
+
+    diff = total - sum(durations)
+    while diff != 0:
+        step = 1 if diff > 0 else -1
+        # 每轮只给「还有余量」的镜加/减 1 秒,均匀铺开,避免把某一镜顶到边界
+        movable = [
+            i for i, d in enumerate(durations)
+            if (step > 0 and d < SEGMENT_DURATION_MAX) or (step < 0 and d > SEGMENT_DURATION_MIN)
+        ]
+        if not movable:
+            return plan_segment_durations(total, presentation_format)
+        for i in movable:
+            if diff == 0:
+                break
+            durations[i] += step
+            diff -= step
+    return durations
 
 
 # 模型每次生成都可能换字段名(scene/screenDescription/visual…、dialogue/lines/caption…),
@@ -306,7 +488,14 @@ def _resolve_segments(draft: dict) -> list:
     return best
 
 
-def normalize_draft(raw_text: str, *, aspect_ratio: str, total_duration: int) -> dict:
+def normalize_draft(
+    raw_text: str,
+    *,
+    aspect_ratio: str,
+    total_duration: int,
+    presentation_format: str = DEFAULT_PRESENTATION_FORMAT,
+    video_structure: str = DEFAULT_VIDEO_STRUCTURE,
+) -> dict:
     """把模型输出抽成 JSON 并按铁律1契约规范化。宽容:小问题就地修,不轻易抛错。"""
     blob = _extract_json(raw_text)
     if not blob:
@@ -332,10 +521,12 @@ def normalize_draft(raw_text: str, *, aspect_ratio: str, total_duration: int) ->
     draft["segments"] = _resolve_segments(draft)
 
     draft["aspect_ratio"] = (draft.get("aspect_ratio") or aspect_ratio or "9:16").strip()
-    dur = _nearest_duration(draft.get("total_duration") or total_duration)
+    # 总时长以「请求参数」为准:模型经常把它算错,而下游出片/计价都按这个数走。
+    dur = coerce_total_duration(total_duration)
     draft["total_duration"] = dur
-    seg_count = max(1, dur // 15)
-    draft["segment_count"] = seg_count
+    fmt, structure = coerce_combo(presentation_format, video_structure)
+    draft["presentation_format"] = PRESENTATION_FORMATS[fmt]
+    draft["video_structure"] = VIDEO_STRUCTURES[structure]
     tone = (draft.get("tone") or "").strip()
     draft["tone"] = tone if tone in VALID_TONES else "种草"
     draft["hook"] = (draft.get("hook") or "").strip()
@@ -367,15 +558,21 @@ def normalize_draft(raw_text: str, *, aspect_ratio: str, total_duration: int) ->
     draft["entities"] = norm_entities
     valid_ids = {e["id"] for e in norm_entities}
 
-    # segments 规范化:对齐镜数,role 枚举,引用合法
+    # segments 规范化:镜数交给模型(只夹进合法区间),role 枚举,引用合法
     segments = draft.get("segments") if isinstance(draft.get("segments"), list) else []
+    # 镜数上下限由「单镜 4–15 秒」倒推:少于 count_min 会有镜超 15 秒,多于 count_max 会有镜不足 4 秒。
+    count_min = math.ceil(dur / SEGMENT_DURATION_MAX)
+    count_max = max(count_min, dur // SEGMENT_DURATION_MIN)
+    segments = segments[:count_max]
+    seg_count = max(count_min, len(segments))
+    role_plan = plan_roles(seg_count)
     norm_segments: list[dict] = []
-    for i, seg in enumerate(segments[:seg_count]):
+    for i, seg in enumerate(segments):
         if not isinstance(seg, dict):
             seg = {}
         role = (seg.get("role") or "").strip()
         if role not in VALID_ROLES:
-            role = VALID_ROLES[min(i, len(VALID_ROLES) - 1)]
+            role = role_plan[i]
         speaker = seg.get("speaker")
         speaker = speaker if (speaker in valid_ids) else None
         refs = [r for r in (seg.get("entity_refs") or []) if r in valid_ids]
@@ -406,7 +603,7 @@ def normalize_draft(raw_text: str, *, aspect_ratio: str, total_duration: int) ->
         norm_segments.append(
             {
                 "index": i,
-                "duration": 15,
+                "duration": seg.get("duration"),  # 先原样收着,等镜数定了再统一夹进 4–15 秒并配平总时长
                 "role": role,
                 "narration": narration,
                 "speaker": speaker,
@@ -416,14 +613,14 @@ def normalize_draft(raw_text: str, *, aspect_ratio: str, total_duration: int) ->
                 "dialogue": dialogue,
             }
         )
-    # 不足镜数则补占位镜(极少发生,避免下游镜数对不上)
+    # 不足下限则补占位镜(极少发生,避免出现超过 15 秒的镜导致下游拒片)
     while len(norm_segments) < seg_count:
         i = len(norm_segments)
         norm_segments.append(
             {
                 "index": i,
-                "duration": 15,
-                "role": VALID_ROLES[min(i, len(VALID_ROLES) - 1)],
+                "duration": None,
+                "role": role_plan[i],
                 "narration": "",
                 "speaker": None,
                 "visual": "",
@@ -434,11 +631,26 @@ def normalize_draft(raw_text: str, *, aspect_ratio: str, total_duration: int) ->
         )
     if not norm_segments:
         raise ValueError("脚本没有任何分镜")
+
+    # 每镜时长:采纳模型的不等长意图,非法就修,修不动整组回落到按表现形式节奏均切。
+    fitted = _fit_segment_durations([s["duration"] for s in norm_segments], dur, fmt)
+    for seg, seconds in zip(norm_segments, fitted):
+        seg["duration"] = seconds
+
     draft["segments"] = norm_segments
+    draft["segment_count"] = len(norm_segments)
     return draft
 
 
-def _merge_single_segment(base: dict, new: dict, idx: int, aspect_ratio: str, total_duration: int) -> dict:
+def _merge_single_segment(
+    base: dict,
+    new: dict,
+    idx: int,
+    aspect_ratio: str,
+    total_duration: int,
+    presentation_format: str = DEFAULT_PRESENTATION_FORMAT,
+    video_structure: str = DEFAULT_VIDEO_STRUCTURE,
+) -> dict:
     """精准改一镜:以基准稿为底,只用新稿的第 idx 镜替换,其余镜逐字保持;合并新稿引入的新 entity(对白可能加角色)。再整体规范化。"""
     merged = json.loads(json.dumps(base))  # 深拷贝
     base_ids = {e.get("id") for e in merged.get("entities", []) if isinstance(e, dict)}
@@ -459,7 +671,13 @@ def _merge_single_segment(base: dict, new: dict, idx: int, aspect_ratio: str, to
         target["index"] = idx
         segs[idx] = target
     merged["segments"] = segs
-    return normalize_draft(json.dumps(merged, ensure_ascii=False), aspect_ratio=aspect_ratio, total_duration=total_duration)
+    return normalize_draft(
+        json.dumps(merged, ensure_ascii=False),
+        aspect_ratio=aspect_ratio,
+        total_duration=total_duration,
+        presentation_format=presentation_format,
+        video_structure=video_structure,
+    )
 
 
 # --------------------------------------------------------------------------- #
@@ -496,14 +714,17 @@ def persist_script_draft(*, project, user, task, draft: dict, source: str):
             task=task,
             title=(draft.get("hook") or "AI 脚本")[:128],
             content=json.dumps(draft, ensure_ascii=False, indent=2),
-            source=source if source in ("ai", "theme", "manual", "revise") else "ai",
+            source=source if source in ("ai", "theme", "manual", "video", "revise") else "ai",
             is_adopted=False,
             metadata={
                 "hook": draft.get("hook", ""),
                 "tone": draft.get("tone", ""),
                 "aspect_ratio": draft.get("aspect_ratio", "9:16"),
-                "total_duration": draft.get("total_duration", 60),
-                "segment_count": draft.get("segment_count", 4),
+                "total_duration": draft.get("total_duration", DEFAULT_TOTAL_DURATION),
+                "segment_count": draft.get("segment_count", len(draft.get("segments") or [])),
+                # 二期:表现形式 × 视频结构 跟着稿子走,改稿和「保存模板」都要读它
+                "presentation_format": draft.get("presentation_format", ""),
+                "video_structure": draft.get("video_structure", ""),
                 "entities": draft.get("entities", []),
             },
         )
@@ -511,7 +732,7 @@ def persist_script_draft(*, project, user, task, draft: dict, source: str):
             ScriptSegment.objects.create(
                 script_version=script,
                 sort_order=seg["index"],
-                duration_seconds=seg.get("duration", 15),
+                duration_seconds=seg.get("duration") or SEGMENT_DURATION_MAX,
                 narration=seg.get("narration", ""),
                 visual_prompt=seg.get("visual", ""),
                 role=seg.get("role", ""),
@@ -570,15 +791,20 @@ def stream_script_agent(
     selling_point_ids: list[str] | None = None,
     base_version_id: str | None = None,
     aspect_ratio: str = "9:16",
-    total_duration: int = 60,
+    total_duration: int = DEFAULT_TOTAL_DURATION,
+    presentation_format: str = DEFAULT_PRESENTATION_FORMAT,
+    video_structure: str = DEFAULT_VIDEO_STRUCTURE,
     target_index: int | None = None,
+    entry_source: str = "",
 ):
     """生成 SSE 帧字符串的同步生成器,供 StreamingHttpResponse 包裹。
     target_index 非空 = 精准只改第 N 镜(读全脚本上下文,后端强制保留其余镜原样)。"""
     from apps.ai.services import create_ai_task, stream_routed_text_request
 
-    yield _sse({"type": "tool", "id": "skill", "label": "加载电商脚本技能", "status": "running"})
-    skill_loaded = bool(load_ecommerce_skill())
+    fmt, structure = coerce_combo(presentation_format, video_structure)
+
+    yield _sse({"type": "tool", "id": "skill", "label": f"加载套路:{PRESENTATION_FORMATS[fmt]} · {VIDEO_STRUCTURES[structure]}", "status": "running"})
+    skill_loaded = bool(load_ecommerce_skill(fmt, structure))
     yield _sse({"type": "tool", "id": "skill", "status": "done" if skill_loaded else "error"})
 
     yield _sse({"type": "tool", "id": "analyze", "label": f"分析商品:{project.product.title}", "status": "running"})
@@ -587,8 +813,10 @@ def stream_script_agent(
         base_draft = _load_base_draft(project, base_version_id)
     if base_draft is None:
         target_index = None  # 没有基准稿就退回整版生成,单镜改无从谈起
-    # 改稿以基准稿的时长/镜数为准,避免请求侧默认值(前端可能硬编码 60)把 90s/6镜稿的尾镜挤掉
-    effective_duration = (base_draft.get("total_duration") if base_draft else None) or total_duration
+    # 改稿以基准稿的时长/镜数为准,避免请求侧默认值把长稿的尾镜挤掉
+    effective_duration = coerce_total_duration(
+        (base_draft.get("total_duration") if base_draft else None) or total_duration
+    )
     # 精准改一镜:镜号越界直接报错返回,绝不建任务/扣费(避免计费空转的静默 no-op)
     if target_index is not None and base_draft is not None:
         seg_n = len(base_draft.get("segments", []))
@@ -602,7 +830,9 @@ def stream_script_agent(
         selling_point_ids=selling_point_ids,
         base_draft=base_draft,
         aspect_ratio=aspect_ratio,
-        total_duration=effective_duration,  # 改稿用基准稿时长,prompt head 才不会误导模型镜数(否则模型按60s只出4镜)
+        total_duration=effective_duration,  # 改稿用基准稿时长,prompt head 才不会误导模型镜数
+        presentation_format=fmt,
+        video_structure=structure,
         target_index=target_index,
     )
     yield _sse({"type": "tool", "id": "analyze", "status": "done"})
@@ -648,6 +878,8 @@ def stream_script_agent(
                     raw_text,
                     aspect_ratio=aspect_ratio,
                     total_duration=effective_duration,
+                    presentation_format=fmt,
+                    video_structure=structure,
                 )
                 if target_index is not None and base_draft:
                     return _merge_single_segment(
@@ -656,6 +888,8 @@ def stream_script_agent(
                         target_index,
                         aspect_ratio,
                         effective_duration,
+                        fmt,
+                        structure,
                     )
                 return candidate
 
@@ -735,7 +969,14 @@ def stream_script_agent(
                 task.completed_at = timezone.now()
                 task.save(update_fields=["status", "response_payload", "actual_cost", "completed_at", "updated_at"])
                 charge_reserved_credit(reservation=reservation, actual_amount=task.actual_cost)
-                source = "revise" if mode == "revise" else ("theme" if mode == "theme" else "ai")
+                # 三个入口(辅助生成 / 上传脚本 / 上传视频提炼)都走 mode=auto,只有 entry_source
+                # 分得清是哪个来的 —— 脚本卡的「来源」徽标靠它,别一律记成 ai。
+                if mode == "revise":
+                    source = "revise"
+                elif mode == "theme":
+                    source = "theme"
+                else:
+                    source = entry_source if entry_source in {"manual", "video"} else "ai"
                 script = persist_script_draft(project=project, user=user, task=task, draft=draft, source=source)
             settled = True  # charge 已提交
         except Exception as exc:  # noqa: BLE001 — 落库失败:atomic 已回滚 charge,补释放预留
@@ -819,20 +1060,24 @@ def _load_base_draft(project, base_version_id: str) -> dict | None:
 
 def _draft_from_version(version) -> dict:
     """从 ScriptVersion 的 DB 行(segments + metadata)重建 ScriptDraft —— 比解析可能已 stale 的 content 可靠
-    (用户增删/改镜后 content 不一定同步)。total_duration 按真实镜数算,避免 normalize 按 stale 值截/补镜。"""
+    (用户增删/改镜后 content 不一定同步)。total_duration 按各镜真实秒数加总,避免 normalize 按 stale 值截/补镜。"""
     meta = version.metadata or {}
     segs = list(version.segments.order_by("sort_order"))
+    # 镜可以不等长了,总时长必须按实际相加(旧写法 15×镜数 会在不等长稿上算出错误总时长)
+    actual_total = sum(s.duration_seconds or SEGMENT_DURATION_MAX for s in segs)
     return {
         "hook": meta.get("hook", ""),
         "tone": meta.get("tone", ""),
+        "presentation_format": meta.get("presentation_format", ""),
+        "video_structure": meta.get("video_structure", ""),
         "aspect_ratio": meta.get("aspect_ratio", "9:16"),
-        "total_duration": max(int(meta.get("total_duration") or 0), 15 * len(segs)) or 60,
+        "total_duration": actual_total or coerce_total_duration(meta.get("total_duration")),
         "segment_count": len(segs),
         "entities": meta.get("entities", []),
         "segments": [
             {
                 "index": s.sort_order,
-                "duration": s.duration_seconds or 15,
+                "duration": s.duration_seconds or SEGMENT_DURATION_MAX,
                 "role": s.role or "",
                 "narration": s.narration or "",
                 "speaker": s.speaker or None,
@@ -860,7 +1105,9 @@ def regenerate_segment_via_agent(*, project, user, model_config: ModelConfig, se
     seg_n = len(base_draft.get("segments") or [])
     if not (0 <= target_index < seg_n):
         raise ValueError(f"镜号越界:第 {target_index + 1} 镜(共 {seg_n} 镜)")
-    total_duration = base_draft["total_duration"]  # 已按真实镜数算,normalize 不会截掉用户增删后的镜
+    total_duration = base_draft["total_duration"]  # 已按各镜真实秒数加总,normalize 不会截掉用户增删后的镜
+    # 改一镜要沿用原稿的套路,否则重写出来的那一镜镜头语言会跟其余镜打架
+    fmt, structure = combo_keys(base_draft.get("presentation_format"), base_draft.get("video_structure"))
 
     messages = build_agent_messages(
         project=project,
@@ -870,6 +1117,8 @@ def regenerate_segment_via_agent(*, project, user, model_config: ModelConfig, se
         base_draft=base_draft,
         aspect_ratio=aspect_ratio,
         total_duration=total_duration,
+        presentation_format=fmt,
+        video_structure=structure,
         target_index=target_index,
     )
     task = create_ai_task(
@@ -897,8 +1146,16 @@ def regenerate_segment_via_agent(*, project, user, model_config: ModelConfig, se
         task.submitted_at = timezone.now()
         task.save(update_fields=["status", "submitted_at", "updated_at"])
         def validate_segment_text(raw_text: str) -> dict:
-            candidate = normalize_draft(raw_text, aspect_ratio=aspect_ratio, total_duration=total_duration)
-            return _merge_single_segment(base_draft, candidate, target_index, aspect_ratio, total_duration)
+            candidate = normalize_draft(
+                raw_text,
+                aspect_ratio=aspect_ratio,
+                total_duration=total_duration,
+                presentation_format=fmt,
+                video_structure=structure,
+            )
+            return _merge_single_segment(
+                base_draft, candidate, target_index, aspect_ratio, total_duration, fmt, structure
+            )
 
         routed = execute_routed_text_request(
             task=task,

core/backend/apps/ai/test_free_video_asset_ref.py

@@ -0,0 +1,150 @@
+"""自由创作 @引用三库(模块4 · 4.1/4.2)单测:平台资产解析 + 审核闸。
+
+运行:DB_ENGINE=sqlite python manage.py test apps.ai.test_free_video_asset_ref --settings=airshelf.settings.test
+
+关键不变量:引用是否放行只看 Asset.source + review_status,**不看**资产在不在库里 ——
+按库免审等于把审核架空(用户传图进库再引用出去就绕过了人像审核)。
+"""
+from unittest.mock import patch
+
+from django.test import TestCase
+
+from apps.accounts.models import Team, User
+from apps.ai.free_video import build_content_items
+from apps.assets.models import Asset, AssetFile
+from apps.assets.review import reference_review_state
+
+
+def _asset(team, *, source, review_status="", review_remote_id="", category=Asset.Category.UPLOAD):
+    asset = Asset.objects.create(
+        team=team,
+        name="素材",
+        asset_type=Asset.Type.IMAGE,
+        source=source,
+        category=category,
+        review_status=review_status,
+        review_remote_id=review_remote_id,
+    )
+    AssetFile.objects.create(
+        asset=asset,
+        object_key="k/1.png",
+        bucket="b",
+        content_type="image/png",
+        size_bytes=1,
+        preview_url="http://tos/1.png",
+        is_primary=True,
+    )
+    return asset
+
+
+class ReferenceReviewStateTests(TestCase):
+    """审核判定四态。审核服务开着时才生效,关着一律放行(不能拿没配置的机制拦人)。"""
+
+    def setUp(self):
+        self.user = User.objects.create_user(username="refowner", password="p")
+        self.team = Team.objects.create(name="REF", owner=self.user)
+        patch("apps.assets.assets_client.is_enabled", return_value=True).start()
+        self.addCleanup(patch.stopall)
+
+    def test_platform_generated_is_exempt(self):
+        asset = _asset(self.team, source=Asset.Source.AI_GENERATED)
+        self.assertEqual(reference_review_state(asset), "allowed")
+
+    def test_upload_needs_active(self):
+        self.assertEqual(reference_review_state(_asset(self.team, source=Asset.Source.UPLOAD)), "unsubmitted")
+        self.assertEqual(
+            reference_review_state(_asset(self.team, source=Asset.Source.UPLOAD, review_status="processing")),
+            "processing",
+        )
+        self.assertEqual(
+            reference_review_state(_asset(self.team, source=Asset.Source.UPLOAD, review_status="failed")),
+            "failed",
+        )
+        self.assertEqual(
+            reference_review_state(_asset(self.team, source=Asset.Source.UPLOAD, review_status="active")),
+            "allowed",
+        )
+
+    def test_in_library_alone_does_not_exempt(self):
+        """核心:进了资产库 ≠ 审过。只要是上传来源,没审过就不能引用。"""
+        asset = _asset(self.team, source=Asset.Source.UPLOAD)
+        asset.in_library = True
+        asset.save(update_fields=["in_library"])
+        self.assertEqual(reference_review_state(asset), "unsubmitted")
+
+    def test_review_disabled_lets_everything_through(self):
+        patch("apps.assets.assets_client.is_enabled", return_value=False).start()
+        self.assertEqual(reference_review_state(_asset(self.team, source=Asset.Source.UPLOAD)), "allowed")
+
+
+class AssetReferenceBuildTests(TestCase):
+    """source=asset 分支:URL 解析 + 闸门拦截 + @label 映射。"""
+
+    def setUp(self):
+        self.user = User.objects.create_user(username="refbuild", password="p")
+        self.team = Team.objects.create(name="REFB", owner=self.user)
+        patch("apps.assets.assets_client.is_enabled", return_value=True).start()
+        self.addCleanup(patch.stopall)
+
+    def _build(self, asset, label="模特A"):
+        return build_content_items(
+            team=self.team,
+            prompt=f"@{label} 走过来",
+            mode="universal",
+            references=[{"type": "image", "label": label, "asset_id": str(asset.id), "source": "asset"}],
+        )
+
+    def test_platform_asset_resolves_to_preview_url(self):
+        asset = _asset(self.team, source=Asset.Source.AI_GENERATED)
+        built = self._build(asset)
+        self.assertEqual(built["image_n"], 1)
+        self.assertEqual(built["content_items"][0]["image_url"]["url"], "http://tos/1.png")
+        self.assertEqual(built["api_prompt"], "图片1 走过来")
+
+    def test_registered_asset_uses_volcano_asset_scheme(self):
+        """已登记火山素材库的走 asset://,写实人脸传直链会被拒。"""
+        asset = _asset(
+            self.team,
+            source=Asset.Source.AI_GENERATED,
+            category=Asset.Category.PERSON,
+            review_status="active",
+            review_remote_id="asset-abc",
+        )
+        built = self._build(asset)
+        self.assertEqual(built["content_items"][0]["image_url"]["url"], "asset://asset-abc")
+
+    def test_unreviewed_upload_is_blocked_and_submitted(self):
+        asset = _asset(self.team, source=Asset.Source.UPLOAD)
+        with patch("apps.assets.review.submit_asset_for_review", return_value=True) as submit:
+            with self.assertRaises(ValueError) as ctx:
+                self._build(asset)
+        self.assertIn("已提交审核", str(ctx.exception))
+        submit.assert_called_once()
+        self.assertTrue(submit.call_args.kwargs["force"])  # 上传素材不看类目白名单,一律登记
+
+    def test_failed_review_is_blocked(self):
+        asset = _asset(self.team, source=Asset.Source.UPLOAD, review_status="failed")
+        with self.assertRaises(ValueError) as ctx:
+            self._build(asset)
+        self.assertIn("未通过审核", str(ctx.exception))
+
+    def test_other_team_asset_is_not_visible(self):
+        stranger = User.objects.create_user(username="stranger", password="p")
+        other = Team.objects.create(name="OTHER", owner=stranger)
+        asset = _asset(other, source=Asset.Source.AI_GENERATED)
+        with self.assertRaises(ValueError) as ctx:
+            self._build(asset)
+        self.assertIn("不存在", str(ctx.exception))
+
+    def test_keyframe_rejects_non_image(self):
+        asset = _asset(self.team, source=Asset.Source.AI_GENERATED)
+        asset.asset_type = Asset.Type.VIDEO
+        asset.save(update_fields=["asset_type"])
+        with self.assertRaises(ValueError) as ctx:
+            build_content_items(
+                team=self.team,
+                prompt="动起来",
+                mode="keyframe",
+                references=[{"type": "video", "asset_id": str(asset.id), "source": "asset", "role": "first_frame"}],
+            )
+        self.assertIn("首尾帧模式仅支持图片素材", str(ctx.exception))

core/backend/apps/ai/test_script_duration_combo.py

@@ -0,0 +1,162 @@
+"""二期第2段:时长自由化 + 表现形式 × 视频结构 的纯函数单测(不碰 DB / 不调模型)。
+
+覆盖三件容易悄悄坏掉的事:
+1. 切镜结果必须「每镜 4–15 秒」且「加总精确等于总时长」—— 越界下游出片会直接拒。
+2. 模型给的每镜时长可能是错的(加总对不上/单镜越界),后端必须能修回来而不是原样落库。
+3. 套路必须按组合**选择性**加载,全量灌进去会让系统提示词翻倍且互相打架。
+"""
+from django.test import SimpleTestCase
+
+from apps.ai.script_agent import (
+    DEFAULT_TOTAL_DURATION,
+    PRESENTATION_FORMATS,
+    SEGMENT_DURATION_MAX,
+    SEGMENT_DURATION_MIN,
+    TOTAL_DURATION_MAX,
+    TOTAL_DURATION_MIN,
+    VIDEO_STRUCTURES,
+    _fit_segment_durations,
+    allowed_structures,
+    coerce_combo,
+    coerce_total_duration,
+    combo_keys,
+    load_ecommerce_skill,
+    narration_limit,
+    plan_roles,
+    plan_segment_durations,
+)
+
+ALL_TOTALS = list(range(TOTAL_DURATION_MIN, TOTAL_DURATION_MAX + 1, 5))
+
+
+class TotalDurationCoercionTests(SimpleTestCase):
+    def test_empty_and_invalid_fall_back_to_default(self):
+        for raw in (None, "", 0, -5, "abc", object()):
+            self.assertEqual(coerce_total_duration(raw), DEFAULT_TOTAL_DURATION, raw)
+
+    def test_clamped_into_range_and_snapped_to_step(self):
+        self.assertEqual(coerce_total_duration(3), TOTAL_DURATION_MIN)
+        self.assertEqual(coerce_total_duration(999), TOTAL_DURATION_MAX)
+        self.assertEqual(coerce_total_duration(7), 5)
+        self.assertEqual(coerce_total_duration(8), 10)
+        self.assertEqual(coerce_total_duration(45), 45)
+
+
+class SegmentPlanningTests(SimpleTestCase):
+    def test_every_total_and_format_yields_legal_shots(self):
+        for total in ALL_TOTALS:
+            for fmt in PRESENTATION_FORMATS:
+                with self.subTest(total=total, fmt=fmt):
+                    durations = plan_segment_durations(total, fmt)
+                    self.assertEqual(sum(durations), total)
+                    for seconds in durations:
+                        self.assertGreaterEqual(seconds, SEGMENT_DURATION_MIN)
+                        self.assertLessEqual(seconds, SEGMENT_DURATION_MAX)
+
+    def test_faster_format_produces_more_shots(self):
+        # 同样 60 秒,Vlog(碎片)镜数应多于口播(要把话说完)
+        self.assertGreater(
+            len(plan_segment_durations(60, "vlog")),
+            len(plan_segment_durations(60, "oral")),
+        )
+
+    def test_roles_follow_shot_count(self):
+        self.assertEqual(plan_roles(1), ["钩子"])
+        self.assertEqual(plan_roles(2), ["钩子", "卖点"])
+        self.assertEqual(plan_roles(3), ["钩子", "卖点", "CTA"])
+        self.assertEqual(plan_roles(4), ["钩子", "痛点", "卖点", "CTA"])
+        self.assertEqual(plan_roles(6), ["钩子", "痛点", "卖点", "卖点", "卖点", "CTA"])
+
+    def test_roles_always_open_with_hook_and_close_with_cta(self):
+        for count in range(3, 13):
+            plan = plan_roles(count)
+            self.assertEqual(len(plan), count)
+            self.assertEqual(plan[0], "钩子")
+            self.assertEqual(plan[-1], "CTA")
+
+
+class SegmentDurationFittingTests(SimpleTestCase):
+    def test_legal_uneven_durations_are_kept_as_is(self):
+        # 模型的不等长节奏意图只要合法就不该被抹平
+        self.assertEqual(_fit_segment_durations([12, 8, 10], 30, "oral"), [12, 8, 10])
+
+    def test_wrong_total_is_repaired(self):
+        fitted = _fit_segment_durations([15, 15, 15, 15], 30, "oral")
+        self.assertEqual(sum(fitted), 30)
+        self.assertTrue(all(SEGMENT_DURATION_MIN <= d <= SEGMENT_DURATION_MAX for d in fitted))
+
+    def test_out_of_range_shot_is_repaired(self):
+        fitted = _fit_segment_durations([99, 1], 30, "drama")
+        self.assertEqual(sum(fitted), 30)
+        self.assertTrue(all(SEGMENT_DURATION_MIN <= d <= SEGMENT_DURATION_MAX for d in fitted))
+
+    def test_empty_falls_back_to_planner(self):
+        self.assertEqual(_fit_segment_durations([], 20, "vlog"), plan_segment_durations(20, "vlog"))
+
+    def test_garbage_values_never_escape_the_legal_range(self):
+        for raw in ([None, None], ["a", "b"], [0, 0, 0], [3, 3], [40, 40, 40]):
+            for total in (10, 30, 60):
+                with self.subTest(raw=raw, total=total):
+                    fitted = _fit_segment_durations(list(raw), total, "oral")
+                    self.assertEqual(sum(fitted), total)
+                    for seconds in fitted:
+                        self.assertGreaterEqual(seconds, SEGMENT_DURATION_MIN)
+                        self.assertLessEqual(seconds, SEGMENT_DURATION_MAX)
+
+
+class NarrationLimitTests(SimpleTestCase):
+    def test_limit_scales_with_shot_length(self):
+        self.assertEqual(narration_limit(4), 14)
+        self.assertEqual(narration_limit(8), 28)
+        self.assertEqual(narration_limit(15), 52)
+
+    def test_never_exceeds_hard_cap(self):
+        self.assertLessEqual(narration_limit(60), 55)
+
+
+class ComboTests(SimpleTestCase):
+    def test_drama_cannot_pick_review(self):
+        # 演出来的实测没有可信度 —— 这是唯一的禁用组合
+        self.assertNotIn("review", allowed_structures("drama"))
+        self.assertEqual(len(allowed_structures("drama")), 3)
+
+    def test_other_formats_allow_everything(self):
+        for fmt in ("oral", "vlog"):
+            self.assertEqual(len(allowed_structures(fmt)), len(VIDEO_STRUCTURES))
+
+    def test_forbidden_combo_falls_back_instead_of_raising(self):
+        fmt, structure = coerce_combo("drama", "review")
+        self.assertEqual(fmt, "drama")
+        self.assertNotEqual(structure, "review")
+
+    def test_unknown_values_fall_back_to_defaults(self):
+        self.assertEqual(coerce_combo("nope", "nope"), ("oral", "pain"))
+
+    def test_chinese_labels_round_trip_back_to_keys(self):
+        # 落库存的是中文标签,改稿时要能还原成 key 去挑套路文件
+        self.assertEqual(combo_keys("短剧", "痛点解决"), ("drama", "pain"))
+        self.assertEqual(combo_keys("Vlog", "场景种草"), ("vlog", "scene"))
+        self.assertEqual(combo_keys("口播", "测评验证"), ("oral", "review"))
+
+
+class SkillLoadingTests(SimpleTestCase):
+    def test_only_the_selected_playbooks_are_loaded(self):
+        for fmt in PRESENTATION_FORMATS:
+            for structure in allowed_structures(fmt):
+                with self.subTest(fmt=fmt, structure=structure):
+                    text = load_ecommerce_skill(fmt, structure)
+                    self.assertIn(f"format-{fmt}.md", text)
+                    self.assertIn(f"structure-{structure}.md", text)
+                    for other in PRESENTATION_FORMATS:
+                        if other != fmt:
+                            self.assertNotIn(f"format-{other}.md", text)
+                    for other in VIDEO_STRUCTURES:
+                        if other != structure:
+                            self.assertNotIn(f"structure-{other}.md", text)
+
+    def test_combo_matrix_is_always_loaded(self):
+        self.assertIn("combo-matrix.md", load_ecommerce_skill("oral", "pain"))
+
+    def test_skill_text_is_substantial(self):
+        # skills 没随镜像打进去时会退化成一句兜底,这里守住「提示词没丢」
+        self.assertGreater(len(load_ecommerce_skill("oral", "pain")), 5000)

core/backend/apps/ai/test_script_entry_source.py

@@ -0,0 +1,118 @@
+"""脚本「来源」要记准是哪个入口生成的。
+
+三个入口(脚本辅助生成 / 上传脚本 / 上传视频提炼)都走 mode=auto,后端原先一律记成 ai,
+脚本卡上的「来源」徽标因此永远显示「脚本辅助生成」。加了 entry_source 之后锁住这个行为。
+"""
+
+from __future__ import annotations
+
+import json
+from decimal import Decimal
+from unittest.mock import Mock, patch
+
+from django.test import TransactionTestCase
+
+from apps.accounts.models import Team, TeamMember, User
+from apps.ai.models import ModelConfig, ModelProvider
+from apps.ai.script_agent import stream_script_agent
+from apps.billing.models import CreditAccount
+from apps.products.models import Product
+from apps.projects.models import Project, ScriptVersion
+
+
+class ScriptEntrySourceTests(TransactionTestCase):
+    reset_sequences = True
+
+    def setUp(self):
+        ModelConfig.objects.filter(capability=ModelConfig.Capability.TEXT).update(
+            status=ModelConfig.Status.DISABLED
+        )
+        self.user = User.objects.create_user(username="entry-source", password="x")
+        self.team = Team.objects.create(name="Entry Source", owner=self.user)
+        TeamMember.objects.create(team=self.team, user=self.user, role=TeamMember.Role.OWNER)
+        CreditAccount.objects.create(team=self.team, balance=Decimal("1000"))
+        product = Product.objects.create(team=self.team, created_by=self.user, title="测试商品")
+        self.project = Project.objects.create(
+            team=self.team, created_by=self.user, product=product, name="入口来源项目"
+        )
+        provider = ModelProvider.objects.create(
+            name="entry-source-provider",
+            display_name="entry-source-provider",
+            status=ModelProvider.Status.ACTIVE,
+            metadata={"routing": {"fallback_priority": 20}},
+        )
+        self.model_config = ModelConfig.objects.create(
+            provider=provider,
+            name="entry-source-model",
+            display_name="entry-source-model",
+            capability=ModelConfig.Capability.TEXT,
+            endpoint="chat/completions",
+            unit_price=Decimal("10"),
+            status=ModelConfig.Status.ACTIVE,
+            metadata={
+                "routing": {"fallback_on_failure": True, "fallback_candidate": True},
+                "capabilities": {"operations": ["chat"], "features": ["streaming", "structured_output"]},
+                "pricing": {"base_cost_yuan": "0.50"},
+            },
+        )
+        patch("apps.ai.services.get_text_provider", side_effect=self._provider).start()
+        self.addCleanup(patch.stopall)
+
+    def _provider(self, _model):
+        raw = json.dumps(
+            {
+                "hook": "开场钩子",
+                "tone": "自然",
+                "aspect_ratio": "9:16",
+                "total_duration": 15,
+                "segment_count": 1,
+                "entities": [
+                    {"id": "c1", "type": "character", "name": "女主", "visual_prompt": "都市女主", "ref_index": 1}
+                ],
+                "segments": [
+                    {
+                        "index": 0, "duration": 15, "role": "钩子",
+                        "narration": "全新脚本口播", "visual": "女主展示商品",
+                        "speaker": "女主", "product_exposure": "展示",
+                        "entity_refs": ["c1"], "dialogue": [],
+                    }
+                ],
+            },
+            ensure_ascii=False,
+        )
+        provider = Mock()
+        provider.chat_completion_stream.return_value = iter(
+            [{"type": "delta", "text": raw}, {"type": "done"}]
+        )
+        return provider
+
+    def _run(self, **kwargs) -> str:
+        frames = list(
+            stream_script_agent(
+                project=self.project,
+                user=self.user,
+                model_config=self.model_config,
+                mode="auto",
+                user_prompt="生成一版脚本",
+                aspect_ratio="9:16",
+                total_duration=15,
+                **kwargs,
+            )
+        )
+        events = [json.loads(frame.removeprefix("data: ").strip()) for frame in frames]
+        errors = [e for e in events if e.get("type") == "error"]
+        self.assertFalse(errors, f"生成失败:{events}")
+        return ScriptVersion.objects.filter(project=self.project).latest("created_at").source
+
+    def test_video_entry_is_recorded_as_video(self):
+        self.assertEqual(self._run(entry_source="video"), "video")
+
+    def test_upload_script_entry_is_recorded_as_manual(self):
+        self.assertEqual(self._run(entry_source="manual"), "manual")
+
+    def test_assisted_entry_falls_back_to_ai(self):
+        self.assertEqual(self._run(entry_source=""), "ai")
+
+    def test_unknown_entry_source_is_not_trusted(self):
+        """来源直接进 DB 并渲染成徽标,不认的值一律回落 ai,别把请求体原样落库。"""
+        self.assertEqual(self._run(entry_source="../../etc/passwd"), "ai")

core/backend/apps/ai/test_video_digest.py

@@ -0,0 +1,142 @@
+"""1.11 上传视频提炼:抽帧规划、上传校验、多模态消息组装、产出判空。
+
+不碰真模型:模型侧已在真视频上端到端验证过,这里只锁纯函数与边界。
+"""
+
+from __future__ import annotations
+
+import subprocess
+
+from django.core.files.uploadedfile import SimpleUploadedFile
+from django.test import SimpleTestCase
+
+from apps.ai.video_digest import (
+    MAX_DURATION_SECONDS,
+    MAX_FRAMES,
+    MAX_UPLOAD_BYTES,
+    MIN_FRAMES,
+    SECONDS_PER_FRAME,
+    VideoDigestError,
+    VideoFrame,
+    build_digest_messages,
+    frames_from_upload,
+    load_digest_skill,
+    plan_frame_times,
+    validate_digest_text,
+)
+
+
+class FramePlanTests(SimpleTestCase):
+    def test_frame_count_scales_with_duration_within_bounds(self):
+        for duration, expected in [(3, MIN_FRAMES), (10, MIN_FRAMES), (30, 6), (60, 12), (180, MAX_FRAMES)]:
+            self.assertEqual(len(plan_frame_times(duration)), expected, duration)
+
+    def test_short_clip_still_gets_min_frames(self):
+        """5 秒的片子按每 5 秒一帧只有 1 帧,判不出分镜,必须兜到 MIN_FRAMES。"""
+        self.assertEqual(len(plan_frame_times(SECONDS_PER_FRAME)), MIN_FRAMES)
+
+    def test_times_are_inside_the_clip_and_ordered(self):
+        times = plan_frame_times(22.2)
+        self.assertEqual(times, sorted(times))
+        self.assertGreaterEqual(times[0], 0)
+        self.assertLess(times[-1], 22.2)
+
+    def test_samples_midpoints_not_edges(self):
+        """首帧常是黑场、尾帧常是片尾卡片,取每段中点避开。"""
+        self.assertGreater(plan_frame_times(60)[0], 0)
+
+
+class UploadGuardTests(SimpleTestCase):
+    def _upload(self, name: str, size: int = 1024):
+        return SimpleUploadedFile(name, b"x" * size, content_type="video/mp4")
+
+    def test_rejects_non_video_suffix(self):
+        with self.assertRaises(VideoDigestError) as ctx:
+            frames_from_upload(self._upload("script.docx"))
+        self.assertIn("mp4", str(ctx.exception))
+
+    def test_rejects_oversized_file_before_touching_ffmpeg(self):
+        big = self._upload("big.mp4", size=8)
+        big.size = MAX_UPLOAD_BYTES + 1
+        with self.assertRaises(VideoDigestError) as ctx:
+            frames_from_upload(big)
+        self.assertIn("MB", str(ctx.exception))
+
+    def test_rejects_unreadable_file(self):
+        with self.assertRaises(VideoDigestError):
+            frames_from_upload(self._upload("broken.mp4"))
+
+    def test_rejects_clip_longer_than_cap(self):
+        """超长片抽样密度不够,拆出来是错的,宁可让用户先剪。"""
+        with self.assertRaises(VideoDigestError) as ctx:
+            _digest_with_duration(_synth_clip(seconds=1), MAX_DURATION_SECONDS + 1)
+        self.assertIn("分钟", str(ctx.exception))
+
+    def test_accepts_clip_within_cap_and_returns_frames(self):
+        frames, duration = frames_from_upload(_synth_clip(seconds=6))
+        self.assertGreaterEqual(len(frames), MIN_FRAMES)
+        self.assertTrue(all(f.jpeg.startswith(b"\xff\xd8") for f in frames))
+        self.assertAlmostEqual(duration, 6, delta=1)
+
+
+class MessageBuildTests(SimpleTestCase):
+    frames = [VideoFrame(at_seconds=2, jpeg=b"\xff\xd8fake"), VideoFrame(at_seconds=7, jpeg=b"\xff\xd8fake2")]
+
+    def test_system_prompt_is_the_digest_skill(self):
+        messages = build_digest_messages(self.frames, 10)
+        self.assertEqual(messages[0]["role"], "system")
+        self.assertIn("七要素", messages[0]["content"])
+
+    def test_every_frame_is_inlined_with_its_timestamp(self):
+        content = build_digest_messages(self.frames, 10)[1]["content"]
+        images = [c for c in content if c["type"] == "image_url"]
+        self.assertEqual(len(images), 2)
+        for image in images:
+            self.assertTrue(image["image_url"]["url"].startswith("data:image/jpeg;base64,"))
+        stamps = [c["text"] for c in content if c["type"] == "text"]
+        self.assertIn("[第 2 秒]", stamps)
+        self.assertIn("[第 7 秒]", stamps)
+
+    def test_product_hint_reaches_the_model(self):
+        """带上商品,模型才会在存疑里提示「这条结构换到你的商品要改哪镜」。"""
+        content = build_digest_messages(self.frames, 10, product_hint="蓝牙耳机 · 数码3C")[1]["content"]
+        self.assertIn("蓝牙耳机 · 数码3C", content[0]["text"])
+
+    def test_skill_loads_from_disk(self):
+        self.assertIn("画面七要素", load_digest_skill())
+
+
+class DigestValidationTests(SimpleTestCase):
+    def test_rejects_empty_or_apologetic_output(self):
+        for bad in ["", "   ", "抱歉,我无法处理这个请求。"]:
+            with self.assertRaises(ValueError):
+                validate_digest_text(bad)
+
+    def test_rejects_prose_without_shot_blocks(self):
+        with self.assertRaises(ValueError):
+            validate_digest_text("这是一条很好的带货视频,节奏明快,画面精美。" * 5)
+
+    def test_accepts_well_formed_digest(self):
+        good = "【整体结构】\n形式:口播\n" + "【第 1 镜】0-4 秒 · 钩子\n主体:一位女生\n" * 3
+        self.assertEqual(validate_digest_text(f"  {good}  "), good.strip())
+
+
+# --------------------------------------------------------------------------- #
+# helpers
+# --------------------------------------------------------------------------- #
+def _synth_clip(seconds: int) -> SimpleUploadedFile:
+    """用 ffmpeg 造一段纯色小视频当夹具,不依赖外部素材。"""
+    done = subprocess.run(
+        ["ffmpeg", "-v", "error", "-f", "lavfi", "-i", f"color=c=orange:s=320x180:d={seconds}",
+         "-c:v", "libx264", "-pix_fmt", "yuv420p", "-f", "mp4", "-movflags", "frag_keyframe+empty_moov", "-"],
+        capture_output=True, check=True,
+    )
+    return SimpleUploadedFile("clip.mp4", done.stdout, content_type="video/mp4")
+
+
+def _digest_with_duration(clip, duration: float):
+    """把探到的时长顶成 duration,验时长闸门,不必真造一条 3 分钟的片。"""
+    from unittest.mock import patch
+
+    with patch("apps.ai.video_digest.probe_duration", return_value=duration):
+        return frames_from_upload(clip)

core/backend/apps/ai/video_digest.py

@@ -0,0 +1,295 @@
+"""上传视频提炼 —— 参考视频 → 可人工逐镜编辑的中文分镜稿。
+
+链路:ffmpeg 抽帧(均匀采样) → 帧内联进多模态 messages → 走现有文本模型路由 → 纯文本分镜稿。
+
+两个「本来以为要新建、其实已经有」的前提:
+  1. **读图能力**:默认脚本模型(YunQi gemini-3.1-pro)本身是多模态的,OpenAI 兼容的
+     ``content: [{type:"text"},{type:"image_url"}]`` 直接透传即可,不需要新 provider、新模型、新 key。
+  2. **ffmpeg**:第 5 阶段导出早就依赖它,已装进后端镜像(见 Dockerfile)。
+
+**没有音轨**:帧里读不到口播,只能读画面上的字幕。这是刻意取舍——接语音转写要另开火山 ASR 服务、
+另加一套凭证与计价,而带货参考片绝大多数带硬字幕,且口播词下游本来就要按用户自己的商品重写。
+skill 里已要求模型「无字幕就如实写缺失,不许编口播词」。
+"""
+
+from __future__ import annotations
+
+import base64
+import json
+import math
+import shutil
+import subprocess
+import tempfile
+from dataclasses import dataclass
+from functools import lru_cache
+from pathlib import Path
+
+from django.conf import settings
+
+# 上传限制:超了直接 400,不进 ffmpeg,也不花模型钱
+ALLOWED_SUFFIXES = (".mp4", ".mov", ".m4v", ".webm")
+MAX_UPLOAD_BYTES = 200 * 1024 * 1024  # 200 MB
+MAX_DURATION_SECONDS = 180  # 3 分钟。带货参考片远短于此;更长的帧采样密度不够,拆出来也是错的
+
+# 抽帧:每 5 秒一帧,夹在 4~12 帧之间。12 帧 × 512px JPEG ≈ 0.5 MB base64,单次请求扛得住
+SECONDS_PER_FRAME = 5
+MIN_FRAMES = 4
+MAX_FRAMES = 12
+FRAME_WIDTH = 512  # 帧宽上限;分镜拆解看的是构图与景别,不需要原分辨率
+FRAME_QUALITY = 5  # ffmpeg -q:v,2(最好)~31(最差)
+
+_FFMPEG_TIMEOUT = 60
+
+
+class VideoDigestError(ValueError):
+    """用户可见的失败(文件不合格 / ffmpeg 读不动),一律 400。"""
+
+
+@dataclass(frozen=True)
+class VideoFrame:
+    at_seconds: int
+    jpeg: bytes
+
+    def as_data_url(self) -> str:
+        return "data:image/jpeg;base64," + base64.b64encode(self.jpeg).decode("ascii")
+
+
+# --------------------------------------------------------------------------- #
+# skill 加载
+# --------------------------------------------------------------------------- #
+def _skill_dir() -> Path:
+    """与 script_agent._skill_dir 同源:优先 BASE_DIR/skills(镜像内),回落仓库根(本地旧布局)。"""
+    base = Path(settings.BASE_DIR)
+    for cand in (base / "skills", base.parent.parent / "skills"):
+        if (cand / "video-shot-digest").is_dir():
+            return cand / "video-shot-digest"
+    return base / "skills" / "video-shot-digest"
+
+
+@lru_cache(maxsize=1)
+def load_digest_skill() -> str:
+    main = _skill_dir() / "SKILL.md"
+    if main.exists():
+        return main.read_text(encoding="utf-8")
+    # 兜底:skill 丢了也别整条链路挂掉,退化成一句话提示词(产出会明显变差,交接文档已注明须带 skills 目录)
+    return (
+        "你是分镜拆解 agent。输入是一条电商短视频按时间均匀抽出的截帧。"
+        "逐镜还原分镜,每镜写清主体/动作/场景/景别/运镜/光线氛围/商品露出七要素,"
+        "台词只抄画面上的字幕,看不见的不要编。输出中文纯文本。"
+    )
+
+
+# --------------------------------------------------------------------------- #
+# ffmpeg:探时长 + 抽帧
+# --------------------------------------------------------------------------- #
+def _binary(name: str) -> str:
+    found = shutil.which(name)
+    if not found:
+        raise VideoDigestError("服务器暂时无法解析视频,请稍后再试")
+    return found
+
+
+def probe_duration(path: str | Path) -> float:
+    """ffprobe 读时长(秒)。读不到 = 不是能解的视频。"""
+    try:
+        out = subprocess.run(
+            [
+                _binary("ffprobe"), "-v", "error",
+                "-print_format", "json", "-show_format",
+                str(path),
+            ],
+            capture_output=True, timeout=_FFMPEG_TIMEOUT, check=True,
+        ).stdout
+        duration = float(json.loads(out)["format"]["duration"])
+    except VideoDigestError:
+        raise
+    except Exception as exc:  # noqa: BLE001 — ffprobe 各种失败对用户是同一件事
+        raise VideoDigestError("这个视频读不出来,请换一个 mp4 / mov 文件") from exc
+    if duration <= 0:
+        raise VideoDigestError("这个视频读不出来,请换一个 mp4 / mov 文件")
+    return duration
+
+
+def plan_frame_times(duration: float) -> list[int]:
+    """均匀采样时间点。取每段的**中点**,避开首尾黑场与片尾卡片。"""
+    count = max(MIN_FRAMES, min(MAX_FRAMES, math.ceil(duration / SECONDS_PER_FRAME)))
+    step = duration / count
+    return [int(step * (i + 0.5)) for i in range(count)]
+
+
+def extract_frames(path: str | Path, times: list[int]) -> list[VideoFrame]:
+    """逐时间点抽一帧。``-ss`` 放在 ``-i`` 前走关键帧快速定位,每帧约几十毫秒。"""
+    ffmpeg = _binary("ffmpeg")
+    frames: list[VideoFrame] = []
+    for at in times:
+        try:
+            done = subprocess.run(
+                [
+                    ffmpeg, "-v", "error", "-ss", str(at), "-i", str(path),
+                    "-frames:v", "1", "-vf", f"scale={FRAME_WIDTH}:-2",
+                    "-q:v", str(FRAME_QUALITY), "-f", "image2", "-",
+                ],
+                capture_output=True, timeout=_FFMPEG_TIMEOUT, check=True,
+            )
+        except Exception:  # noqa: BLE001 — 单帧抽失败(定位越界等)跳过,别拖垮整次提炼
+            continue
+        if done.stdout:
+            frames.append(VideoFrame(at_seconds=at, jpeg=done.stdout))
+    if not frames:
+        raise VideoDigestError("没能从这个视频里取到画面,请换一个文件")
+    return frames
+
+
+def frames_from_upload(upload) -> tuple[list[VideoFrame], float]:
+    """校验上传文件 → 落临时盘 → 探时长 → 抽帧。临时文件退出即删。"""
+    name = (getattr(upload, "name", "") or "").lower()
+    if not name.endswith(ALLOWED_SUFFIXES):
+        raise VideoDigestError("只支持 mp4 / mov / m4v / webm 四种视频格式")
+    size = getattr(upload, "size", 0) or 0
+    if size > MAX_UPLOAD_BYTES:
+        raise VideoDigestError(f"视频不能超过 {MAX_UPLOAD_BYTES // 1024 // 1024} MB,请压缩后再传")
+
+    suffix = Path(name).suffix or ".mp4"
+    with tempfile.NamedTemporaryFile(suffix=suffix) as tmp:
+        for chunk in upload.chunks():
+            tmp.write(chunk)
+        tmp.flush()
+        duration = probe_duration(tmp.name)
+        if duration > MAX_DURATION_SECONDS:
+            raise VideoDigestError(
+                f"视频不能超过 {MAX_DURATION_SECONDS // 60} 分钟,请剪出要参考的那一段再传"
+            )
+        return extract_frames(tmp.name, plan_frame_times(duration)), duration
+
+
+# --------------------------------------------------------------------------- #
+# 组装多模态消息
+# --------------------------------------------------------------------------- #
+def build_digest_messages(
+    frames: list[VideoFrame],
+    duration: float,
+    *,
+    product_hint: str = "",
+) -> list[dict]:
+    """system = 拆解 skill;user = 时间戳 + 帧图交替,让模型知道每张图在原片的第几秒。"""
+    head = [
+        f"这是一条时长约 {round(duration)} 秒的电商带货短视频,",
+        f"按时间顺序均匀抽了 {len(frames)} 帧。每帧图前面标了它在原片中的时间点。",
+    ]
+    if product_hint:
+        head.append(f"用户接下来想用这条片子的结构去拍自己的商品:{product_hint}。")
+    head.append("请按技能里的输出格式还原它的分镜。")
+
+    content: list[dict] = [{"type": "text", "text": "".join(head)}]
+    for frame in frames:
+        content.append({"type": "text", "text": f"[第 {frame.at_seconds} 秒]"})
+        content.append({"type": "image_url", "image_url": {"url": frame.as_data_url()}})
+    return [
+        {"role": "system", "content": load_digest_skill()},
+        {"role": "user", "content": content},
+    ]
+
+
+def validate_digest_text(text: str) -> str:
+    """模型偶尔吐空 / 吐一句道歉。判空后交给路由层重试或切模型,别把废稿塞给用户。"""
+    cleaned = (text or "").strip()
+    if len(cleaned) < 80 or "【" not in cleaned:
+        raise ValueError("视频拆解结果不完整")
+    return cleaned
+
+
+# --------------------------------------------------------------------------- #
+# 入口:一次真实的计费调用
+# --------------------------------------------------------------------------- #
+def digest_project_video(*, project, user, upload) -> dict:
+    """上传视频 → 分镜稿。抽帧在建任务之前做,文件不合格不占积分。"""
+    from django.db import transaction
+    from django.utils import timezone
+
+    from apps.ai.models import AITask, ModelConfig
+    from apps.ai.services import create_ai_task, execute_routed_text_request, get_default_model
+    from apps.billing.services.ledger import charge_reserved_credit
+
+    frames, duration = frames_from_upload(upload)
+
+    model_config = get_default_model(ModelConfig.Capability.TEXT)
+    if model_config is None:
+        raise VideoDigestError("暂时没有可用的模型,请联系管理员")
+
+    product = getattr(project, "product", None)
+    messages = build_digest_messages(
+        frames,
+        duration,
+        product_hint=" · ".join(
+            filter(None, [getattr(product, "title", ""), getattr(product, "category", "")])
+        ),
+    )
+
+    task = create_ai_task(
+        project=project,
+        user=user,
+        task_type=AITask.Type.VIDEO_DIGEST,
+        model_config=model_config,
+        # 帧是几百 KB base64,绝不进 request_payload(会把 AITask 表撑爆),只记形状
+        request_payload={
+            "model": model_config.name,
+            "endpoint": model_config.endpoint,
+            "duration_seconds": round(duration, 2),
+            "frame_count": len(frames),
+            "frame_times": [f.at_seconds for f in frames],
+        },
+    )
+    reservation = task.credit_reservation
+    try:
+        task.status = AITask.Status.SUBMITTED
+        task.submitted_at = timezone.now()
+        task.save(update_fields=["status", "submitted_at", "updated_at"])
+        routed = execute_routed_text_request(
+            task=task,
+            primary_model=model_config,
+            messages=messages,
+            streaming=False,
+            structured_output=False,
+            business_operation="video_digest",
+            temperature=0.4,
+            validate_text=validate_digest_text,
+            request_summary={"duration_seconds": round(duration, 2), "frame_count": len(frames)},
+        )
+        _text, _response, digest = routed.value
+    except Exception as exc:  # noqa: BLE001
+        _fail_digest_task(task, reservation, str(exc))
+        raise
+
+    with transaction.atomic():
+        task.status = AITask.Status.SUCCEEDED
+        task.response_payload = {"digest": digest[:8000]}
+        task.actual_cost = task.estimated_cost
+        task.completed_at = timezone.now()
+        task.save(update_fields=["status", "response_payload", "actual_cost", "completed_at", "updated_at"])
+        charge_reserved_credit(reservation=reservation, actual_amount=task.actual_cost)
+
+    return {
+        "text": digest,
+        "chars": len(digest),
+        "frames": len(frames),
+        "duration": round(duration, 1),
+        "task_id": str(task.id),
+    }
+
+
+def _fail_digest_task(task, reservation, message: str) -> None:
+    from django.utils import timezone
+
+    from apps.ai.models import AITask
+    from apps.billing.services.ledger import release_credit
+
+    try:
+        task.status = AITask.Status.FAILED
+        task.error_message = message[:2000]
+        task.completed_at = timezone.now()
+        task.save(update_fields=["status", "error_message", "completed_at", "updated_at"])
+    finally:
+        try:
+            release_credit(reservation=reservation, reason=message[:200])
+        except Exception:  # noqa: BLE001
+            pass