azmat1127/yingqing
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

优化脚本

Browse Source
This commit is contained in:
Azmat@qq.com
2026-08-21 10:28:35 +08:00
parent 6c588a7fee
commit c2d3a786a3
23 changed files with 1461 additions and 287 deletions
Download Patch File Download Diff File Expand all files Collapse all files
core/backend/apps/ai/generation_errors.py
+11
View File
@@ -404,6 +404,17 @@ def classify_generation_error(
return _build_error("invalid_input", operation, reference_id=reference_id)
if _contains(compact_code, "postprocesserror"):
return _build_error("processing_failed", operation, reference_id=reference_id)
if _contains(
signal,
"没有用上商品卖点原词",
"没有出现品牌原词",
"必须原词用上",
"旁白太短",
"画面描写太短",
"按秒拆分镜",
"切两次景别",
):
return _build_error("processing_failed", operation, reference_id=reference_id)
return _build_error("unknown", operation, reference_id=reference_id)
core/backend/apps/ai/migrations/0031_script_beat_prompt_templates.py
+47
View File
@@ -0,0 +1,47 @@
from django.db import migrations
STORYBOARD = (
"{设定}根据以下脚本生成一个导演故事板，用于指导 seedance 的视频生成。\n{场景上下文}\n"
"【分镜脚本】(本段时长约 {时长} 秒)\n{脚本}\n"
"请严格保持各参考图中角色的同一张脸、同一商品的外观与配色；"
"电商竖屏 9:16 导演故事板，一镜一图，画面清晰，可直接指导视频生成。"
"脚本里的 visual 是本段秒级分镜清单；请画商品用法正确、最能看清商品的那一拍作为关键帧。"
"禁止画出违背常识的用法（例如把茶包丢进冷白开、悬浮的手）。{补充}"
)
STORYBOARD_OLD = (
"{设定}根据以下脚本生成一个导演故事板，用于指导 seedance 的视频生成。\n{场景上下文}\n"
"【分镜脚本】(本段时长约 {时长} 秒)\n{脚本}\n"
"请严格保持各参考图中角色的同一张脸、同一商品的外观与配色；"
"电商竖屏 9:16 导演故事板，一镜一图，画面清晰，可直接指导视频生成。{补充}"
)
VIDEO = (
"{设定}{分镜}【脚本】{脚本}\n"
"严格按照脚本里的秒级分镜切换景别和动作，商品用法必须真实，不要诡异姿势或错误容器。"
"电商带货短视频，商品露出清晰，节奏有转化感。不要字幕，不要背景音乐，但是要有音效，逼真的音效。"
)
VIDEO_OLD = (
"{设定}{分镜}【脚本】{脚本}\n"
"电商带货短视频，商品露出清晰，节奏有转化感。不要字幕，不要背景音乐，但是要有音效，逼真的音效。"
)
def _set(apps, key, template):
PromptTemplate = apps.get_model("ai", "PromptTemplate")
PromptTemplate.objects.filter(key=key).update(template=template)
def forwards(apps, schema_editor):
_set(apps, "storyboard_frame", STORYBOARD)
_set(apps, "video_segment", VIDEO)
def backwards(apps, schema_editor):
_set(apps, "storyboard_frame", STORYBOARD_OLD)
_set(apps, "video_segment", VIDEO_OLD)
class Migration(migrations.Migration):
dependencies = [("ai", "0030_alter_aitask_task_type")]
operations = [migrations.RunPython(forwards, backwards)]
core/backend/apps/ai/script_agent.py
+545 -94
View File
@@ -19,8 +19,8 @@ SSE 事件(每帧 `data: {json}\n\n`,json 带 type):
from __future__ import annotations
import json
import math
import re
import uuid
from decimal import Decimal
from functools import lru_cache
from pathlib import Path
@@ -36,10 +36,11 @@ VALID_TONES = ["种草", "测评", "剧情", "痛点"]
VALID_ROLES = ["钩子", "痛点", "卖点", "CTA"]
VALID_ENTITY_TYPES = ["character", "scene", "product"]
# 时长:总时长 5–60 秒按 5 秒步进;单镜 4–15 秒(15 是出片模型硬上限,越界下游直接拒片)。
TOTAL_DURATION_MIN = 5
# 时长:主流程每镜固定 15 秒;总时长只能是 15/30/45/60。
# 4–15 仍是出片模型合法区间,手改单镜 / 精准改一镜(旧稿)才用。
TOTAL_DURATION_MIN = 15
TOTAL_DURATION_MAX = 60
TOTAL_DURATION_STEP = 5
TOTAL_DURATION_STEP = 15
SEGMENT_DURATION_MIN = 4
SEGMENT_DURATION_MAX = 15
DEFAULT_TOTAL_DURATION = 30
@@ -58,22 +59,84 @@ 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}
# 各结构能压到的最短总时长(须落在 15 秒步进上),见 playbooks/combo-matrix.md。
STRUCTURE_MIN_DURATION: dict[str, int] = {"pain": 15, "contrast": 15, "review": 30, "scene": 30}
# 可懂语速上限 3.5 字/秒 —— 旁白字数按这一镜自己的秒数算,不再全场 55 字一刀切。
# 设定卡人物 key → 中文(与前端 WIZ_PERSONA_LABEL / 模板 coerce_persona 对齐)。
PERSONA_LABELS: dict[str, str] = {
"urban": "都市白领女性",
"bestie": "闺蜜种草",
"ceo": "总裁亲选",
"reviewer": "专业测评师",
"mom": "实用宝妈",
"genz": "学生党",
}
PERSONA_BRIEFS: dict[str, str] = {
"urban": "25–32岁都市白领,工位或下班回家,说话像跟同事吐槽,不要主播腔",
"bestie": "闺蜜分享口吻,带点兴奋,爱用「你懂的」「我跟你讲」",
"ceo": "利落、判断句、少形容词,像拍板不是带货",
"reviewer": "先讲怎么试的再给结论,允许提一个小缺点才可信",
"mom": "带娃/家务间隙,讲省事,孩子或家人能沾边",
"genz": "宿舍或通勤,短句,像给朋友发语音条",
}
_PERSONA_KEY_BY_LABEL = {label: key for key, label in PERSONA_LABELS.items()}
# 可懂语速上限 3.5 字/秒;15 秒口播还要有下限,否则模型写一句 20 字就收,撑不满。
NARRATION_CHARS_PER_SECOND = 3.5
NARRATION_CHARS_PER_SECOND_MIN = 2.4
NARRATION_CHARS_HARD_CAP = 55
VISUAL_CHARS_MIN = 72
SHOT_BEATS_MIN = 3
BEAT_SPAN_RE = re.compile(
r"(?P<start>\d{1,2})\s*[-–—~到至]\s*(?P<end>\d{1,2})\s*(?:s|秒)?\s*[：:]",
re.IGNORECASE,
)
_SHOT_SIZE_MARKERS = (
"特写", "近景", "中近景", "中景", "全景", "远景", "胸上", "过肩",
"手持", "跟拍", "俯拍", "仰拍", "推近", "拉远", "摇",
)
_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 coerce_persona(value: str | None) -> str:
"""把设定卡 key 或中文标签收成 key;空/未知原样(未知时当补充描述用)。"""
raw = (value or "").strip()
if not raw:
return ""
if raw in PERSONA_LABELS:
return raw
return _PERSONA_KEY_BY_LABEL.get(raw, raw)
def persona_label(value: str | None) -> str:
key = coerce_persona(value)
return PERSONA_LABELS.get(key, key)
def _wizard_meta(project) -> dict:
meta = getattr(project, "metadata", None) or {}
wizard = meta.get("wizard") if isinstance(meta, dict) else None
return wizard if isinstance(wizard, dict) else {}
def _resolve_selling_point_ids(project, selling_point_ids: list | None) -> list[str]:
if selling_point_ids:
return [str(item) for item in selling_point_ids if item]
raw = _wizard_meta(project).get("selling_point_ids") or []
if not isinstance(raw, list):
return []
return [str(item) for item in raw if item]
def _resolve_persona(project, persona: str | None) -> str:
return coerce_persona(persona) or coerce_persona(_wizard_meta(project).get("persona"))
def combo_keys(value_format, value_structure) -> tuple[str, str]:
"""把「中文标签或 ASCII key」都归一成 key。落库存的是中文,请求传的是 key,两边都要认。"""
fmt = _FORMAT_KEY_BY_LABEL.get(value_format, value_format)
@@ -101,6 +164,61 @@ def narration_limit(duration: int) -> int:
return max(1, min(NARRATION_CHARS_HARD_CAP, int(duration * NARRATION_CHARS_PER_SECOND)))
def narration_floor(duration: int) -> int:
"""15 秒口播至少要说到这个字数,再短就撑不满镜头。"""
cap = narration_limit(duration)
return max(1, min(cap - 6, int(duration * NARRATION_CHARS_PER_SECOND_MIN)))
def _compact_len(text: str) -> int:
return len(re.sub(r"\s+", "", text or ""))
def _speech_text(seg: dict) -> str:
dialogue = seg.get("dialogue") if isinstance(seg.get("dialogue"), list) else []
lines = [
str(item.get("line") or "")
for item in dialogue
if isinstance(item, dict) and (item.get("line") or "").strip()
]
if lines:
return "".join(lines)
return str(seg.get("narration") or "")
def min_beats_for_duration(duration: int) -> int:
if duration >= 12:
return SHOT_BEATS_MIN
if duration >= 8:
return 2
return 0
def parse_visual_beats(text: str) -> list[tuple[int, int, str]]:
"""从 visual 文本里抽出「0-3s：…」分镜。接得上才能喂下游视频。"""
raw = (text or "").strip()
if not raw:
return []
matches = list(BEAT_SPAN_RE.finditer(raw))
if not matches:
return []
beats: list[tuple[int, int, str]] = []
for index, match in enumerate(matches):
start, end = int(match.group("start")), int(match.group("end"))
if end < start:
start, end = end, start
body_to = matches[index + 1].start() if index + 1 < len(matches) else len(raw)
body = re.sub(r"[\n\r]+", " ", raw[match.end():body_to]).strip(" ；;，,、")
if start == end or not body:
continue
beats.append((start, end, body))
return beats
def format_visual_beats(beats: list[tuple[int, int, str]]) -> str:
return "\n".join(f"{start}-{end}s：{body}" for start, end, body in beats)
# --------------------------------------------------------------------------- #
# skill 加载(缓存)
# --------------------------------------------------------------------------- #
@@ -174,14 +292,16 @@ _OUTPUT_PROTOCOL = """
## 运行时输出协议(AirShelf 流式展示专用,优先级高于技能正文的「只输出 JSON」)
严格按以下顺序输出,不要有别的内容:
1. 先用 **1 句中文口语**告诉用户你正在做什么(≤40 字,例:「在为这款保温杯生成 4 镜痛点脚本…」),让用户看到进展;
1. 先用 **1 句中文口语**告诉用户你正在做什么(≤40 字,例:「在为这款保温杯生成 2 镜痛点脚本…」),让用户看到进展;
2. 紧接着输出**且仅输出一个** ```json 代码块,内容为符合技能契约(铁律1)的 ScriptDraft 对象;
3. json 代码块**收尾之后另起一行**,用 **1–2 句中文口语**跟用户交付这一版:做了什么、为什么这么改、可以怎么接着调(像同事汇报,**别复述 JSON 字段、别再写代码块**)。这句会作为你的回复气泡展示给用户。
### 字段名锚定(硬性 · 下游靠它取数,跑偏即数据全空)
- 分镜数组的键名**必须**叫 `segments`(禁止用 scenes / script / shots / 分镜 等同义词)。
- 每镜口播键名**必须**叫 `narration`(禁止用 voiceover / audio / line)。
- 每镜画面键名**必须**叫 `visual`,且为**一句话字符串**(禁止用 scene / screen / 画面,也禁止写成 {setting,camera,...} 对象)。
- 每镜画面键名**必须**叫 `visual`(禁止用 scene / screen / 画面 当键名)。
- `visual` 是**多行字符串**,每行一条秒级分镜:`0-3s：近景，……`(禁止只写一句静态动作,也禁止只给 setting/camera 对象而不给 visual)。
- 允许额外给 `beats`/`shots` 数组,后端会折进 visual;有数组也必须能折成「起-止秒」格式。
- 即使你额外附带了 scenes / shots 等创作结构,也**必须同时**给出标准 `segments` 数组,并把口播填进 `narration`、画面填进 `visual`,否则视为不合格。
"""
@@ -189,25 +309,87 @@ _OUTPUT_PROTOCOL = """
# --------------------------------------------------------------------------- #
# 提示词构建(3 模式)
# --------------------------------------------------------------------------- #
def _product_context(project, selling_point_ids: list[str] | None) -> str:
def _specs_lines(specs) -> str:
if not isinstance(specs, dict) or not specs:
return ""
skip = {"source"}
labels = {"price": "价格"}
lines: list[str] = []
for key, val in specs.items():
if key in skip or val in (None, "", [], {}):
continue
if isinstance(val, (dict, list)):
rendered = json.dumps(val, ensure_ascii=False)
else:
rendered = str(val).strip()
if not rendered:
continue
lines.append(f"- {labels.get(key, key)}:{rendered}")
return "\n".join(lines)
def _product_facts(project, selling_point_ids: list[str] | None):
"""商品事实 + 本次勾选的卖点(供提示词和落库前原词校验共用)。
新建向导历史上把卖点**标题**(如「茶」)写进 metadata.wizard.selling_point_ids,
字段名叫 ids 但不是 UUID。按 id__in 过滤会直接 ValidationError 把 SSE 打崩。
这里 UUID 和标题都认;对不上就回落全部卖点,绝不抛。
"""
product = project.product
selling_points = product.selling_points.all()
if selling_point_ids:
selling_points = selling_points.filter(id__in=selling_point_ids)
selling_text = "\n".join(f"- {sp.title}:{sp.detail}" for sp in selling_points)
keys = [str(item).strip() for item in _resolve_selling_point_ids(project, selling_point_ids) if str(item).strip()]
all_points = list(product.selling_points.all())
if not keys:
return product, all_points
uuid_keys: set[str] = set()
title_keys: set[str] = set()
for key in keys:
try:
uuid_keys.add(str(uuid.UUID(key)))
except (ValueError, AttributeError, TypeError):
title_keys.add(key)
matched = [
point
for point in all_points
if str(getattr(point, "id", "") or "") in uuid_keys
or (getattr(point, "title", "") or "").strip() in title_keys
]
return product, matched or all_points
def _product_context(project, selling_point_ids: list[str] | None, persona: str | None = None) -> str:
product, selling_points = _product_facts(project, selling_point_ids)
selling_text = "\n".join(f"- {sp.title}:{sp.detail or sp.title}" for sp in selling_points)
business_type = getattr(product, "business_type", "") or "ecommerce"
if business_type == "local_life":
type_line = "业务类型:本地生活(团购/到店核销,无实物,按虚拟商品走主流程 SOP)\n"
else:
type_line = "业务类型:电商(实物商品)\n"
specs_text = _specs_lines(getattr(product, "specs", None) or {})
persona_key = _resolve_persona(project, persona)
persona_text = persona_label(persona_key)
persona_brief = PERSONA_BRIEFS.get(persona_key, "")
must = [f"商品名「{product.title}」"] if product.title else []
if product.brand:
must.append(f"品牌「{product.brand}」")
must.extend(f"卖点「{sp.title.strip()}」" for sp in selling_points if (sp.title or "").strip())
must_line = "、".join(must) if must else "无(根据商品描述自行提炼,禁止空话套话)"
specs_block = f"\n{specs_text}" if specs_text else "未填写"
desc = (product.description or "").strip() or "未填写"
return (
f"商品标题:{product.title}\n"
f"品牌:{product.brand or '未填写'}\n"
f"{type_line}"
f"品类:{product.category or '未填写'}\n"
f"目标人群:{product.target_audience or '未填写'}\n"
f"商品描述:{product.description or '未填写'}\n"
f"卖点:\n{selling_text or '未勾选卖点,请根据商品信息自行提炼。'}"
f"人物设定:{persona_text or '未指定,按目标人群自行定一个具体身份'}\n"
f"人物口吻:{persona_brief or '按人物设定自己定一个具体身份,不要用万能主播腔'}\n"
f"商品描述:{desc}\n"
f"规格:{specs_block}\n"
f"本次必须用上的卖点:\n{selling_text or '未勾选卖点,请根据商品信息自行提炼,禁止空话。'}\n"
f"【必须原词用上】旁白/对白里要出现:{must_line}。"
f"禁止用「补水/好用/值得买/宝藏」这类空卖点替换上面的原词。"
f"商品名全片点名 1–2 次即可,其余镜用卖点原词和可感知细节(口感/气味/动作/使用场景),不要每句重复商品名。"
f"描述若未填写,就从标题+卖点把使用感写具体,禁止只喊商品名。"
)
@@ -223,23 +405,80 @@ def build_agent_messages(
presentation_format: str = DEFAULT_PRESENTATION_FORMAT,
video_structure: str = DEFAULT_VIDEO_STRUCTURE,
target_index: int | None = None,
persona: str | None = None,
) -> list[dict[str, str]]:
fmt, structure = coerce_combo(presentation_format, video_structure)
total = coerce_total_duration(total_duration)
if target_index is not None:
try:
total = int(total_duration)
except (TypeError, ValueError):
total = DEFAULT_TOTAL_DURATION
if total <= 0:
total = DEFAULT_TOTAL_DURATION
else:
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)
shot_n = len(suggested)
extra = (user_prompt or "").strip()
if target_index is not None and base_draft:
existing_n = len(base_draft.get("segments") or [])
duration_line = (
f"【分镜时长】保持现有 {existing_n} 镜和每镜秒数,"
f"只改第 {target_index + 1} 镜的文案与画面,不要加减镜、不要改时长。\n"
)
else:
duration_line = (
f"【分镜时长】每镜必须 {SEGMENT_DURATION_MAX} 秒,一共 {shot_n} 镜;"
f"禁止写成 8/10/12 这种不等长,也禁止加减镜数。\n"
)
speech_floor = narration_floor(SEGMENT_DURATION_MAX)
speech_cap = narration_limit(SEGMENT_DURATION_MAX)
beats_line = (
f"【秒级分镜】每个 {SEGMENT_DURATION_MAX} 秒场必须拆成 3–5 个分镜,visual 写成多行,"
f"格式`0-3s：景别，谁在做什么`,最后一条接到 {SEGMENT_DURATION_MAX}s。"
"每条写清手/商品/容器的空间关系和真实用法"
"(茶=热水+蒸汽+茶汤变色,禁止茶包丢进冷白开;手从真实方向入画,禁止悬浮肢体)。"
"禁止一句空画面撑满 15 秒。允许另给 beats 数组,后端会折进 visual。\n"
)
if fmt == "oral":
writing_line = (
f"【写法硬约束】口播每镜必须说满 {speech_floor}–{speech_cap} 字(2–4 句短句,"
f"禁止一句 20 字收工);visual 至少 {VISUAL_CHARS_MIN} 字。"
"第一镜用人物口吻交代身份,不要万能主播腔。\n"
)
elif fmt == "drama":
writing_line = (
f"【写法硬约束】对白每镜合计 {speech_floor}–{speech_cap} 字;"
f"visual 至少 {VISUAL_CHARS_MIN} 字。\n"
)
else:
writing_line = (
f"【写法硬约束】有人声的镜口播/对白 {speech_floor}–{speech_cap} 字,允许个别镜纯画面;"
f"visual 至少 {VISUAL_CHARS_MIN} 字。\n"
)
structure_line = ""
if structure == "scene":
structure_line = (
"【场景种草骨架】按镜走完:场景建立(商品先不出现)→商品被使用着入场→场景因它更好/软CTA。"
"每镜写出具体时刻+地点+光/声音/物件。口播时每镜仍要说满字数,留白靠停顿不是靠少写字。\n"
)
combo_line = (
f"严格按已加载的「{PRESENTATION_FORMATS[fmt]} × {VIDEO_STRUCTURES[structure]}」套路写,"
f"不要串成别的结构的套话。\n"
)
head = (
f"【画幅】{aspect_ratio}\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)}"
f"{duration_line}"
f"{writing_line}"
f"{beats_line}"
f"{structure_line}"
f"{combo_line}"
f"【商品信息】\n{_product_context(project, selling_point_ids, persona)}"
)
if mode == "revise" and base_draft and target_index is not None:
# 精准改一镜:读全脚本上下文,只重写第 N 镜,强制与前后镜衔接;其余镜后端会强制保持原样。
@@ -247,7 +486,7 @@ def build_agent_messages(
f"【任务】只重写第 {target_index + 1} 镜(共 {len(base_draft.get('segments', []))} 镜),其余镜保持不变。\n"
f"{head}\n\n"
f"【现有完整脚本 JSON(读它保证与前后镜衔接)】\n{json.dumps(base_draft, ensure_ascii=False)}\n\n"
f"【对第 {target_index + 1} 镜的修改意见】{user_prompt.strip() or '让这一镜更有吸引力、表达更清晰。'}\n\n"
f"【对第 {target_index + 1} 镜的修改意见】{extra or '让这一镜更有吸引力、表达更清晰。'}\n\n"
"仍输出**完整** ScriptDraft(我只会采用第 "
f"{target_index + 1} 镜的改动);若意见涉及角色对白,就给这一镜填 dialogue。"
)
@@ -256,21 +495,25 @@ def build_agent_messages(
"【任务】改稿(模式③):在保留用户原意的前提下,增强钩子/节奏/卖点/CTA,并归一化到契约 JSON。\n"
f"{head}\n\n"
f"【现有脚本 JSON】\n{json.dumps(base_draft, ensure_ascii=False)}\n\n"
f"【用户修改意见】{user_prompt.strip() or '让整体更有吸引力、转化感更强,并保持各镜衔接连贯。'}\n\n"
f"【用户修改意见】{extra or '让整体更有吸引力、转化感更强,并保持各镜衔接连贯。'}\n\n"
"请输出修订后的**完整** ScriptDraft。"
)
elif mode == "theme" or (user_prompt and user_prompt.strip()):
elif mode == "theme":
user = (
"【任务】一句话主题扩写(模式②):以用户主题为脚本主轴,其余自动补全。\n"
"【任务】一句话主题扩写(模式②):以用户主题为脚本主轴,但商品事实和必须原词仍要全部用上。\n"
f"{head}\n\n"
f"【用户主题】{user_prompt.strip()}\n\n"
f"【用户主题】{extra or '按商品最强卖点选题'}\n\n"
"请按技能流程一次性产出 ScriptDraft。"
)
else:
extra_block = f"\n\n【补充要求】{extra}\n" if extra else "\n"
user = (
"【任务】全自动(模式①):仅凭商品与前置条件,按指定的表现形式与视频结构套路"
"自动定镜数/选 tone/造 entity/填结构骨架。\n"
f"{head}\n\n"
"【任务】全自动(模式①):仅凭上面的商品事实与前置条件,按指定的表现形式与视频结构套路"
"自动定镜/选 tone/造 entity/填结构骨架。"
"不要另起一个空主题;钩子、痛点、卖点必须能对上这份商品,而不是品类套话。"
"每镜旁白要能撑满 15 秒;画面必须按秒拆分镜,够导演在 15 秒里切 3–5 刀。\n"
f"{head}"
f"{extra_block}\n"
"请按技能流程一次性产出 ScriptDraft。"
)
return [{"role": "system", "content": system}, {"role": "user", "content": user}]
@@ -333,7 +576,7 @@ def _extract_json(text: str) -> str | None:
def coerce_total_duration(value) -> int:
"""总时长夹到 5–60 秒、5 秒步进。空值/0/非法输入一律回落默认 30。"""
"""总时长夹到 15–60 秒、15 秒步进(15/30/45/60)。空值/0/非法输入一律回落默认 30。"""
if value in (None, "", 0):
return DEFAULT_TOTAL_DURATION
try:
@@ -347,21 +590,11 @@ def coerce_total_duration(value) -> int:
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 秒——
这只是**兜底**,模型自己给的不等长时长只要合法就照用。
"""
def plan_segment_durations(total_duration: int, presentation_format: str | None = None) -> list[int]:
"""主流程每镜固定 15 秒。镜数 = 总时长 / 15。presentation_format 保留签名以免调用方改动。"""
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)]
count = max(1, total // SEGMENT_DURATION_MAX)
return [SEGMENT_DURATION_MAX] * count
def plan_roles(count: int) -> list[str]:
@@ -376,45 +609,106 @@ def plan_roles(count: int) -> list[str]:
def _fit_segment_durations(raw: list, total: int, presentation_format: str) -> list[int]:
"""采纳模型给的每镜时长(允许不等长),非法就修;修不动就整组回落到 plan_segment_durations。
"""主流程强制每镜 15 秒。模型给多少秒都丢掉,按总时长切成 N 个 15。"""
return plan_segment_durations(total, presentation_format)
合法定义:每镜 4–15 秒的整数,且加总 == 总时长。模型很容易把总数算错一两秒,
所以先夹单镜范围,再把差额摊到还有余量的镜上,尽量保住模型的节奏意图。
"""
if not raw:
return plan_segment_durations(total, presentation_format)
def _layout_durations(preserve_layout: dict | None, count: int) -> list[int] | None:
"""精准改一镜:保住原稿每镜秒数,避免 3 镜旧稿被压成 2×15。"""
if not preserve_layout:
return None
raw = preserve_layout.get("durations")
if not isinstance(raw, list) or len(raw) != count:
return None
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
durations.append(max(SEGMENT_DURATION_MIN, min(SEGMENT_DURATION_MAX, seconds or SEGMENT_DURATION_MAX)))
return durations
def _script_plaintext(draft: dict) -> str:
parts = [str(draft.get("hook") or "")]
for seg in draft.get("segments") or []:
if not isinstance(seg, dict):
continue
parts.append(str(seg.get("narration") or ""))
parts.append(str(seg.get("visual") or ""))
parts.append(str(seg.get("product_exposure") or ""))
for item in seg.get("dialogue") or []:
if isinstance(item, dict):
parts.append(str(item.get("line") or ""))
return "".join(parts)
def _fact_mentioned(text: str, fact: str) -> bool:
fact = (fact or "").strip()
if not fact:
return False
if fact in text:
return True
# 超长卖点/标题模型常念前半截,允许前 4 字命中
return len(fact) >= 6 and fact[:4] in text
def assert_product_facts_used(draft: dict, *, brand: str = "", selling_titles: list[str] | None = None) -> None:
"""整版生成时至少用上一个勾选卖点原词。单字卖点(如「茶」)也算。
品牌不强制:口播经常只喊卖点不念品牌,硬卡会把已经写对的稿整单作废。
"""
del brand # 保留调用方签名,避免改一串入口
text = _script_plaintext(draft)
titles = [t.strip() for t in (selling_titles or []) if (t or "").strip()]
if titles and not any(_fact_mentioned(text, title) for title in titles):
raise ValueError("脚本没有用上商品卖点原词,请按【必须原词用上】重写")
def assert_shot_density(draft: dict, presentation_format: str = DEFAULT_PRESENTATION_FORMAT) -> None:
"""口播/短剧每镜要说满、画面要写厚;Vlog 允许个别镜无声,但开口就不能偷懒。"""
fmt, _ = coerce_combo(_FORMAT_KEY_BY_LABEL.get(presentation_format, presentation_format), DEFAULT_VIDEO_STRUCTURE)
allow_silent = fmt == "vlog"
for index, seg in enumerate(draft.get("segments") or []):
if not isinstance(seg, dict):
continue
try:
duration = int(seg.get("duration") or SEGMENT_DURATION_MAX)
except (TypeError, ValueError):
duration = SEGMENT_DURATION_MAX
visual_n = _compact_len(str(seg.get("visual") or ""))
if visual_n < VISUAL_CHARS_MIN:
raise ValueError(f"第 {index + 1} 镜画面描写太短,撑不满 {duration} 秒")
assert_intra_shot_beats(seg, index, duration)
speech_n = _compact_len(_speech_text(seg))
if speech_n == 0 and allow_silent:
continue
floor = narration_floor(duration)
if speech_n < floor:
raise ValueError(f"第 {index + 1} 镜旁白太短,口播至少要说到 {floor} 字才能撑满 {duration} 秒")
def assert_intra_shot_beats(seg: dict, index: int, duration: int) -> None:
"""15 秒场必须按秒拆出分镜,否则下游视频只能对着一句空描述乱编。"""
need = min_beats_for_duration(duration)
if need <= 0:
return
visual = str(seg.get("visual") or "")
beats = parse_visual_beats(visual)
if len(beats) < need:
raise ValueError(
f"第 {index + 1} 镜画面必须按秒拆分镜,至少 {need} 条「0-3s：景别，动作」,接到 {duration}s"
)
if beats[0][0] > 1:
raise ValueError(f"第 {index + 1} 镜画面必须按秒拆分镜,第一条要从 0 秒起")
if beats[-1][1] < duration - 1:
raise ValueError(f"第 {index + 1} 镜画面必须按秒拆分镜,最后一条要接到 {duration}s")
markers = [mark for mark in _SHOT_SIZE_MARKERS if mark in visual]
if len(set(markers)) < 2:
raise ValueError(f"第 {index + 1} 镜 15 秒内至少要切两次景别,并写进秒级分镜")
# 模型每次生成都可能换字段名(scene/screenDescription/visual…、dialogue/lines/caption…),
# 与其逐一追变体,不如「优先键命中 → 否则按关键词模糊匹配」通用解析。SKIP 掉明显的非内容键,
# 避免误抓(shotNo/duration/bgMusic/note 等)。
@@ -422,6 +716,7 @@ _PICK_SKIP_KEYS = {
"shotno", "shotsize", "shotsizetype", "duration", "duration_seconds", "starttime", "endtime",
"timerange", "time_range", "bgmusic", "bg_music", "music", "sound", "sfx", "note", "notes",
"tips", "index", "role", "speaker", "entity_refs", "product_exposure", "id", "no", "transition",
"beats", "timeline",
# 标题/编号类:含 scene 字样会误命中画面 fuzzy,显式跳过(注意:不跳裸 scene,它常=画面)
"scene_id", "sceneid", "scene_no", "sceneno", "scene_number", "scenenumber",
"scene_title", "scenetitle", "title", "scene_title_type",
@@ -463,6 +758,113 @@ def _pick_field(seg: dict, exact: tuple[str, ...], fuzzy: tuple[str, ...]) -> st
return ""
def _opt_int(value) -> int | None:
try:
if value is None or value == "":
return None
return int(value)
except (TypeError, ValueError):
return None
def _beat_from_dict(item: dict) -> dict | None:
action = _flatten_text(
item.get("action")
or item.get("shot")
or item.get("visual")
or item.get("description")
or item.get("画面")
or ""
)
shot_size = str(item.get("shot_size") or item.get("camera") or item.get("景别") or "").strip()
start = _opt_int(item.get("start") if item.get("start") is not None else item.get("from"))
end = _opt_int(item.get("end") if item.get("end") is not None else item.get("to"))
time_s = str(item.get("time") or item.get("t") or "").strip()
if start is None and end is None and time_s:
parsed = parse_visual_beats(f"{time_s}：{action or '画面'}")
if parsed:
start, end, parsed_body = parsed[0]
action = action or parsed_body
if not action and not shot_size:
return None
piece = f"{shot_size}，{action}" if shot_size and action else (action or shot_size)
return {"start": start, "end": end, "action": piece}
def coerce_beats(raw) -> list[dict]:
if raw is None or raw == "":
return []
if isinstance(raw, str):
return [{"start": start, "end": end, "action": body} for start, end, body in parse_visual_beats(raw)]
if isinstance(raw, dict):
nested = raw.get("beats") or raw.get("shots") or raw.get("timeline") or raw.get("key_shots") or raw.get("分镜")
if nested:
return coerce_beats(nested)
beat = _beat_from_dict(raw)
return [beat] if beat else []
if isinstance(raw, list):
out: list[dict] = []
for item in raw:
if isinstance(item, dict):
nested = item.get("beats") or item.get("shots")
if nested and not (item.get("action") or item.get("shot") or item.get("visual")):
out.extend(coerce_beats(nested))
continue
beat = _beat_from_dict(item)
if beat:
out.append(beat)
else:
out.extend(coerce_beats(item))
return out
return []
def _assign_beat_times(beats: list[dict], duration: int) -> list[tuple[int, int, str]]:
n = len(beats)
if n == 0:
return []
out: list[tuple[int, int, str]] = []
for index, beat in enumerate(beats):
start = beat.get("start")
end = beat.get("end")
if start is None:
start = int(round(index * duration / n))
if end is None:
end = duration if index == n - 1 else int(round((index + 1) * duration / n))
if end <= start:
end = start + 1
out.append((int(start), int(end), str(beat.get("action") or "").strip()))
return [item for item in out if item[2]]
def compose_segment_visual(seg: dict, duration: int = SEGMENT_DURATION_MAX) -> str:
"""把 beats 数组或已写好的秒级 visual 收成多行「0-3s：…」。"""
visual_raw = seg.get("visual")
if isinstance(visual_raw, str) and visual_raw.strip():
parsed = parse_visual_beats(visual_raw)
if len(parsed) >= 2:
return format_visual_beats(parsed)
candidates = [
seg.get("beats"),
seg.get("timeline"),
seg.get("分镜"),
visual_raw if isinstance(visual_raw, (list, dict)) else None,
]
shots_raw = seg.get("shots")
if isinstance(shots_raw, list) and shots_raw:
sample = shots_raw[0]
looks_like_nested_segments = isinstance(sample, dict) and (
sample.get("narration") or sample.get("role") in VALID_ROLES
)
if not looks_like_nested_segments:
candidates.append(shots_raw)
for raw in candidates:
assigned = _assign_beat_times(coerce_beats(raw), duration)
if len(assigned) >= 2:
return format_visual_beats(assigned)
return _pick_field(seg, _VISUAL_EXACT, _VISUAL_FUZZY)
# 模型给分镜数组的键名五花八门(segments/scenes/script/shots…),且常同时给一个**空的**
# segments 骨架 + 真内容放在 scenes 里。所以不能「见 segments 是 list 就用」,要在所有候选里
# 挑「能解析出最多非空旁白/画面」的那个。segments(契约本名)排第一,同分时优先。
@@ -501,6 +903,7 @@ def normalize_draft(
total_duration: int,
presentation_format: str = DEFAULT_PRESENTATION_FORMAT,
video_structure: str = DEFAULT_VIDEO_STRUCTURE,
preserve_layout: dict | None = None,
) -> dict:
"""把模型输出抽成 JSON 并按铁律1契约规范化。宽容:小问题就地修,不轻易抛错。"""
blob = _extract_json(raw_text)
@@ -564,13 +967,17 @@ def normalize_draft(
draft["entities"] = norm_entities
valid_ids = {e["id"] for e in norm_entities}
# segments 规范化:镜数交给模型(只夹进合法区间),role 枚举,引用合法
# segments 规范化:主流程镜数 = 总时长/15;精准改一镜则保住原稿镜数。
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))
layout_count = 0
if preserve_layout:
try:
layout_count = int(preserve_layout.get("count") or 0)
except (TypeError, ValueError):
layout_count = 0
expected = layout_count or max(1, dur // SEGMENT_DURATION_MAX)
segments = segments[:expected]
seg_count = expected
role_plan = plan_roles(seg_count)
norm_segments: list[dict] = []
for i, seg in enumerate(segments):
@@ -604,12 +1011,12 @@ def normalize_draft(
narration = " ".join(d["line"] for d in dialogue) # 扁平拼接,兼容下游字幕/配音
if not narration:
narration = _pick_field(seg, _NARRATION_EXACT, _NARRATION_FUZZY)
# 画面:通用解析(visual/scene/screenDescription/画面… 都能命中,跳过 shotNo/bgMusic 等)
visual = _pick_field(seg, _VISUAL_EXACT, _VISUAL_FUZZY)
# 画面:优先收成秒级分镜;beats 数组会折进 visual,下游故事板/视频直接读这一段。
visual = compose_segment_visual(seg)
norm_segments.append(
{
"index": i,
"duration": seg.get("duration"), # 先原样收着,等镜数定了再统一夹进 4–15 秒并配平总时长
"duration": seg.get("duration"), # 先原样收着,后面统一写成 15 或保住原稿秒数
"role": role,
"narration": narration,
"speaker": speaker,
@@ -638,16 +1045,36 @@ def normalize_draft(
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
kept = _layout_durations(preserve_layout, len(norm_segments))
if kept:
fitted = kept
draft["total_duration"] = sum(fitted)
else:
fitted = plan_segment_durations(dur, fmt)
if len(fitted) != len(norm_segments):
fitted = [SEGMENT_DURATION_MAX] * len(norm_segments)
draft["total_duration"] = sum(fitted)
for index, (norm, seconds) in enumerate(zip(norm_segments, fitted)):
norm["duration"] = seconds
if index < len(segments) and isinstance(segments[index], dict):
composed = compose_segment_visual(segments[index], seconds)
if composed:
norm["visual"] = composed
draft["segments"] = norm_segments
draft["segment_count"] = len(norm_segments)
return draft
def _preserve_layout_from(draft: dict | None) -> dict | None:
if not draft:
return None
segs = [s for s in (draft.get("segments") or []) if isinstance(s, dict)]
if not segs:
return None
return {"count": len(segs), "durations": [s.get("duration") for s in segs]}
def _merge_single_segment(
base: dict,
new: dict,
@@ -677,12 +1104,14 @@ def _merge_single_segment(
target["index"] = idx
segs[idx] = target
merged["segments"] = segs
layout = _preserve_layout_from(base)
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,
preserve_layout=layout,
)
@@ -802,6 +1231,7 @@ def stream_script_agent(
video_structure: str = DEFAULT_VIDEO_STRUCTURE,
target_index: int | None = None,
entry_source: str = "",
persona: str | None = None,
):
"""生成 SSE 帧字符串的同步生成器,供 StreamingHttpResponse 包裹。
target_index 非空 = 精准只改第 N 镜(读全脚本上下文,后端强制保留其余镜原样)。"""
@@ -820,15 +1250,24 @@ def stream_script_agent(
if base_draft is None:
target_index = None # 没有基准稿就退回整版生成,单镜改无从谈起
# 改稿以基准稿的时长/镜数为准,避免请求侧默认值把长稿的尾镜挤掉
effective_duration = coerce_total_duration(
(base_draft.get("total_duration") if base_draft else None) or total_duration
)
# 精准改一镜不把总时长夹成 15 步进,否则旧的不等长稿会被提示词误导切镜。
if target_index is not None and base_draft is not None:
effective_duration = int(base_draft.get("total_duration") or total_duration or DEFAULT_TOTAL_DURATION)
else:
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", []))
if not (0 <= target_index < seg_n):
yield _sse({"type": "error", "detail": f"镜号越界:第 {target_index + 1} 镜(共 {seg_n} 镜)"})
return
selling_point_ids = _resolve_selling_point_ids(project, selling_point_ids)
persona = _resolve_persona(project, persona)
product, selling_points = _product_facts(project, selling_point_ids)
selling_titles = [sp.title for sp in selling_points]
layout = _preserve_layout_from(base_draft) if target_index is not None else None
messages = build_agent_messages(
project=project,
mode=mode,
@@ -840,6 +1279,7 @@ def stream_script_agent(
presentation_format=fmt,
video_structure=structure,
target_index=target_index,
persona=persona,
)
yield _sse({"type": "tool", "id": "analyze", "status": "done"})
@@ -886,6 +1326,7 @@ def stream_script_agent(
total_duration=effective_duration,
presentation_format=fmt,
video_structure=structure,
preserve_layout=layout,
)
if target_index is not None and base_draft:
return _merge_single_segment(
@@ -897,6 +1338,13 @@ def stream_script_agent(
fmt,
structure,
)
if mode != "revise":
assert_product_facts_used(
candidate,
brand=getattr(product, "brand", "") or "",
selling_titles=selling_titles,
)
assert_shot_density(candidate, fmt)
return candidate
routed_stream = stream_routed_text_request(
@@ -1069,7 +1517,7 @@ def _draft_from_version(version) -> dict:
(用户增删/改镜后 content 不一定同步)。total_duration 按各镜真实秒数加总,避免 normalize 按 stale 值截/补镜。"""
meta = version.metadata or {}
segs = list(version.segments.order_by("sort_order"))
# 镜可以不等长了,总时长必须按实际相加(旧写法 15×镜数 会在不等长稿上算出错误总时长)
# 总时长按各镜真实秒数加总(旧稿可能不等长;新稿每镜 15 秒)。
actual_total = sum(s.duration_seconds or SEGMENT_DURATION_MAX for s in segs)
return {
"hook": meta.get("hook", ""),
@@ -1119,13 +1567,14 @@ def regenerate_segment_via_agent(*, project, user, model_config: ModelConfig, se
project=project,
mode="revise",
user_prompt=instruction,
selling_point_ids=None,
selling_point_ids=_resolve_selling_point_ids(project, None),
base_draft=base_draft,
aspect_ratio=aspect_ratio,
total_duration=total_duration,
presentation_format=fmt,
video_structure=structure,
target_index=target_index,
persona=_resolve_persona(project, None),
)
task = create_ai_task(
project=project,
@@ -1152,12 +1601,14 @@ 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:
layout = _preserve_layout_from(base_draft)
candidate = normalize_draft(
raw_text,
aspect_ratio=aspect_ratio,
total_duration=total_duration,
presentation_format=fmt,
video_structure=structure,
preserve_layout=layout,
)
return _merge_single_segment(
base_draft, candidate, target_index, aspect_ratio, total_duration, fmt, structure
core/backend/apps/ai/services.py
+13 -4
View File
@@ -2508,7 +2508,8 @@ def _segment_script_text(segment, entities=None, *, with_dialogue: bool = False,
parts = []
visual = (segment.visual_prompt or "").strip()
if visual:
parts.append(f"画面:{visual}")
label = "秒级分镜(按时间切镜,商品用法必须真实,禁止诡异动作/错误容器)" if ("s：" in visual or "s:" in visual or "秒" in visual[:12]) else "画面"
parts.append(f"{label}:\n{visual}")
dialogue = getattr(segment, "dialogue", None) if with_dialogue else None
name_by_id = {e.get("id"): (e.get("name") or "").strip() for e in (entities or []) if isinstance(e, dict)}
lines = []
@@ -2540,7 +2541,9 @@ def build_storyboard_frame_prompt(project, segment, extra_prompt: str = "") -> s
"{设定}根据以下脚本生成一个导演故事板，用于指导 seedance 的视频生成。\n{场景上下文}\n"
"【分镜脚本】(本段时长约 {时长} 秒)\n{脚本}\n"
"请严格保持各参考图中角色的同一张脸、同一商品的外观与配色；"
"电商竖屏 9:16 导演故事板，一镜一图，画面清晰，可直接指导视频生成。{补充}"
"电商竖屏 9:16 导演故事板，一镜一图，画面清晰，可直接指导视频生成。"
"脚本里的 visual 是本段秒级分镜清单；请画商品用法正确、最能看清商品的那一拍作为关键帧。"
"禁止画出违背常识的用法（例如把茶包丢进冷白开、悬浮的手）。{补充}"
)
rendered = render_prompt(
"storyboard_frame", default,
@@ -2574,7 +2577,11 @@ def build_video_segment_prompt(project, video_segment, scene, refs, user_prompt:
script_text = (script_text + "\n" + extra).strip() if script_text else extra
# 正文可在 admin「提示词·视频」页改。占位符:{设定}=@图N点名、{分镜}=分镜图引用、{脚本}=本段脚本、{时长}。
# 时长/比例靠 API 参数传(duration/ratio),不写进正文;尾句给风格 + 音效/字幕要求。
default = "{设定}{分镜}【脚本】{脚本}\n电商带货短视频，商品露出清晰，节奏有转化感。不要字幕，不要背景音乐，但是要有音效，逼真的音效。"
default = (
"{设定}{分镜}【脚本】{脚本}\n"
"严格按照脚本里的秒级分镜切换景别和动作，商品用法必须真实，不要诡异姿势或错误容器。"
"电商带货短视频，商品露出清晰，节奏有转化感。不要字幕，不要背景音乐，但是要有音效，逼真的音效。"
)
rendered = render_prompt(
"video_segment", default,
设定=("【设定】" + "，".join(setup_parts) + "。\n" if setup_parts else ""),
@@ -2744,7 +2751,9 @@ def build_storyboard_frame_prompt_refs(project, segment, refs: list[dict], extra
"{设定}根据以下脚本生成一个导演故事板，用于指导 seedance 的视频生成。\n{场景上下文}\n"
"【分镜脚本】(本段时长约 {时长} 秒)\n{脚本}\n"
"请严格保持各参考图中角色的同一张脸、同一商品的外观与配色；"
"电商竖屏 9:16 导演故事板，一镜一图，画面清晰，可直接指导视频生成。{补充}"
"电商竖屏 9:16 导演故事板，一镜一图，画面清晰，可直接指导视频生成。"
"脚本里的 visual 是本段秒级分镜清单；请画商品用法正确、最能看清商品的那一拍作为关键帧。"
"禁止画出违背常识的用法（例如把茶包丢进冷白开、悬浮的手）。{补充}"
)
rendered = render_prompt(
"storyboard_frame", default,
core/backend/apps/ai/test_generation_errors.py
+8
View File
@@ -322,3 +322,11 @@ class GenerationErrorClassifierTests(SimpleTestCase):
self.assertEqual(input_error.code, "invalid_input")
self.assertEqual(processing_error.code, "processing_failed")
self.assertNotIn("storage", processing_error.fallback_message)
def test_short_script_copy_is_processing_failed_not_unknown(self):
error = classify_generation_error(
ValueError("第 1 镜旁白太短,口播至少要说到 36 字才能撑满 15 秒"),
operation="script_generate",
)
self.assertEqual(error.code, "processing_failed")
self.assertNotEqual(error.code, "unknown")
core/backend/apps/ai/test_script_duration_combo.py
+309 -33
View File
@@ -1,32 +1,42 @@
"""二期第2段:时长自由化 + 表现形式 × 视频结构 的纯函数单测(不碰 DB / 不调模型)。
"""主流程:每镜固定 15 秒 + 表现形式 × 视频结构 的纯函数单测(不碰 DB / 不调模型)。
覆盖三件容易悄悄坏掉的事:
1. 切镜结果必须「每镜 4–15 秒」且「加总精确等于总时长」—— 越界下游出片会直接拒。
2. 模型给的每镜时长可能是错的(加总对不上/单镜越界),后端必须能修回来而不是原样落库。
1. 切镜结果必须「每镜 15 秒」且「加总精确等于总时长」—— 不再允许 8/10/12。
2. 模型给的每镜时长可能是错的,后端必须改写成 15×N,而不是原样落库。
3. 套路必须按组合**选择性**加载,全量灌进去会让系统提示词翻倍且互相打架。
"""
from unittest.mock import Mock
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,
VISUAL_CHARS_MIN,
_fit_segment_durations,
_product_facts,
allowed_structures,
assert_product_facts_used,
assert_shot_density,
build_agent_messages,
coerce_combo,
coerce_total_duration,
combo_keys,
compose_segment_visual,
load_ecommerce_skill,
narration_floor,
narration_limit,
normalize_draft,
parse_visual_beats,
plan_roles,
plan_segment_durations,
)
ALL_TOTALS = list(range(TOTAL_DURATION_MIN, TOTAL_DURATION_MAX + 1, 5))
ALL_TOTALS = [15, 30, 45, 60]
class TotalDurationCoercionTests(SimpleTestCase):
@@ -37,28 +47,24 @@ class TotalDurationCoercionTests(SimpleTestCase):
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(7), 15)
self.assertEqual(coerce_total_duration(20), 15)
self.assertEqual(coerce_total_duration(25), 30)
self.assertEqual(coerce_total_duration(45), 45)
class SegmentPlanningTests(SimpleTestCase):
def test_every_total_and_format_yields_legal_shots(self):
def test_every_total_and_format_yields_fixed_15s_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)
self.assertEqual(durations, [SEGMENT_DURATION_MAX] * (total // 15))
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_format_no_longer_changes_shot_count(self):
self.assertEqual(plan_segment_durations(60, "vlog"), plan_segment_durations(60, "oral"))
self.assertEqual(plan_segment_durations(30, "drama"), [15, 15])
def test_roles_follow_shot_count(self):
self.assertEqual(plan_roles(1), ["钩子"])
@@ -76,32 +82,195 @@ class SegmentPlanningTests(SimpleTestCase):
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_uneven_durations_are_forced_to_15(self):
self.assertEqual(_fit_segment_durations([12, 8, 10], 30, "oral"), [15, 15])
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_too_many_shots_are_collapsed(self):
self.assertEqual(_fit_segment_durations([15, 15, 15, 15], 30, "oral"), [15, 15])
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))
self.assertEqual(_fit_segment_durations([99, 1], 30, "drama"), [15, 15])
def test_empty_falls_back_to_planner(self):
self.assertEqual(_fit_segment_durations([], 20, "vlog"), plan_segment_durations(20, "vlog"))
self.assertEqual(_fit_segment_durations([], 30, "vlog"), plan_segment_durations(30, "vlog"))
def test_garbage_values_never_escape_the_legal_range(self):
def test_garbage_values_never_escape_15s_grid(self):
for raw in ([None, None], ["a", "b"], [0, 0, 0], [3, 3], [40, 40, 40]):
for total in (10, 30, 60):
for total in ALL_TOTALS:
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)
self.assertTrue(all(d == SEGMENT_DURATION_MAX for d in fitted))
class NormalizeDurationTests(SimpleTestCase):
def test_extra_shots_are_truncated_to_duration_grid(self):
raw = {
"segments": [
{"role": "钩子", "narration": "口播1", "visual": "画面1", "duration": 12},
{"role": "痛点", "narration": "口播2", "visual": "画面2", "duration": 8},
{"role": "CTA", "narration": "口播3", "visual": "画面3", "duration": 10},
]
}
import json
draft = normalize_draft(json.dumps(raw, ensure_ascii=False), aspect_ratio="9:16", total_duration=30)
self.assertEqual(len(draft["segments"]), 2)
self.assertEqual([s["duration"] for s in draft["segments"]], [15, 15])
def test_preserve_layout_keeps_legacy_shot_count(self):
import json
raw = {
"segments": [
{"role": "钩子", "narration": "口播1", "visual": "画面1"},
{"role": "痛点", "narration": "口播2", "visual": "画面2"},
{"role": "CTA", "narration": "口播3", "visual": "画面3"},
]
}
draft = normalize_draft(
json.dumps(raw, ensure_ascii=False),
aspect_ratio="9:16",
total_duration=30,
preserve_layout={"count": 3, "durations": [12, 8, 10]},
)
self.assertEqual(len(draft["segments"]), 3)
self.assertEqual([s["duration"] for s in draft["segments"]], [12, 8, 10])
self.assertEqual(draft["total_duration"], 30)
class ProductFactTests(SimpleTestCase):
def test_missing_selling_point_is_rejected(self):
with self.assertRaises(ValueError):
assert_product_facts_used(
{"hook": "夏天好热", "segments": [{"narration": "补水很重要"}]},
selling_titles=["控油12小时"],
)
def test_selling_point_original_word_passes(self):
assert_product_facts_used(
{"hook": "控油12小时真的扛得住", "segments": [{"narration": "晚上八点还不油"}]},
selling_titles=["控油12小时"],
)
def test_single_character_selling_point_counts(self):
assert_product_facts_used(
{"hook": "这杯茶真香", "segments": [{"narration": "下午一杯茶"}]},
selling_titles=["茶"],
)
def test_brand_is_not_required(self):
assert_product_facts_used(
{"hook": "控油12小时", "segments": [{"narration": "控油12小时"}]},
brand="AirBrand",
selling_titles=["控油12小时"],
)
class ProductFactsLookupTests(SimpleTestCase):
def _project(self, points):
class _QS(list):
def filter(self, **kwargs):
raise AssertionError("不要再用 id__in 过滤,标题不是 UUID")
product = Mock()
product.selling_points.all.return_value = _QS(points)
project = Mock()
project.product = product
project.metadata = {}
return project
def test_title_keys_match_without_uuid_filter(self):
tea = Mock(id="11111111-1111-1111-1111-111111111111", title="茶", detail="茶香")
oil = Mock(id="22222222-2222-2222-2222-222222222222", title="控油", detail="x")
_product, matched = _product_facts(self._project([tea, oil]), ["茶"])
self.assertEqual([item.title for item in matched], ["茶"])
def test_uuid_keys_still_match(self):
tea = Mock(id="11111111-1111-1111-1111-111111111111", title="茶", detail="茶香")
oil = Mock(id="22222222-2222-2222-2222-222222222222", title="控油", detail="x")
_product, matched = _product_facts(self._project([tea, oil]), [str(tea.id)])
self.assertEqual([item.title for item in matched], ["茶"])
class PromptAssemblyTests(SimpleTestCase):
def _project(self):
sp = Mock(title="控油12小时", detail="到晚八点")
class _QS(list):
def filter(self, **kwargs):
return self
product = Mock()
product.title = "玻尿酸精华"
product.brand = "AirBrand"
product.business_type = "ecommerce"
product.category = "美妆"
product.target_audience = "油皮"
product.description = "夜间修护"
product.specs = {"price": "99", "source": "product-create-upload"}
product.selling_points.all.return_value = _QS([sp])
project = Mock()
project.product = product
project.metadata = {"wizard": {"persona": "urban", "selling_point_ids": ["1"]}}
return project
def test_auto_mode_with_prompt_is_not_theme(self):
messages = build_agent_messages(
project=self._project(),
mode="auto",
user_prompt="节奏紧凑,适合短视频投放",
selling_point_ids=["1"],
base_draft=None,
aspect_ratio="9:16",
total_duration=30,
persona="urban",
)
user = messages[1]["content"]
self.assertIn("全自动", user)
self.assertNotIn("一句话主题扩写", user)
self.assertIn("【补充要求】节奏紧凑,适合短视频投放", user)
self.assertIn("控油12小时", user)
self.assertIn("都市白领女性", user)
self.assertIn("价格:99", user)
self.assertIn("每镜必须 15 秒", user)
self.assertIn("一共 2 镜", user)
self.assertIn("必须原词用上", user)
self.assertIn("跟同事吐槽", user)
self.assertIn("禁止一句 20 字收工", user)
self.assertIn("秒级分镜", user)
def test_scene_oral_prompt_locks_skeleton_and_density(self):
messages = build_agent_messages(
project=self._project(),
mode="auto",
user_prompt="",
selling_point_ids=["1"],
base_draft=None,
aspect_ratio="9:16",
total_duration=45,
presentation_format="oral",
video_structure="scene",
persona="urban",
)
user = messages[1]["content"]
self.assertIn("场景种草骨架", user)
self.assertIn("口播每镜必须说满", user)
self.assertIn("一共 3 镜", user)
self.assertIn("秒级分镜", user)
def test_theme_mode_still_exists(self):
messages = build_agent_messages(
project=self._project(),
mode="theme",
user_prompt="主打熬夜党",
selling_point_ids=None,
base_draft=None,
aspect_ratio="9:16",
total_duration=15,
)
self.assertIn("一句话主题扩写", messages[1]["content"])
self.assertIn("主打熬夜党", messages[1]["content"])
class NarrationLimitTests(SimpleTestCase):
@@ -113,6 +282,107 @@ class NarrationLimitTests(SimpleTestCase):
def test_never_exceeds_hard_cap(self):
self.assertLessEqual(narration_limit(60), 55)
def test_fifteen_second_floor_is_about_thirty_six(self):
self.assertEqual(narration_floor(15), 36)
self.assertLess(narration_floor(15), narration_limit(15))
class ShotDensityTests(SimpleTestCase):
_VISUAL = (
"0-3s：近景，女主对镜头抬眼，工位键盘还亮着，窗外是下午的光\n"
"3-8s：手持跟到桌面，右手把茶包放进盛了热水的玻璃杯，水面起蒸汽\n"
"8-12s：杯壁特写，茶汤从浅金慢慢变深，标签贴在杯沿\n"
"12-15s：拉回中近景，女主喝一口，眼神松下来"
)
def test_short_oral_narration_is_rejected(self):
with self.assertRaises(ValueError) as ctx:
assert_shot_density(
{
"segments": [
{
"duration": 15,
"narration": "下午三点来杯100numur茶解乏。",
"visual": self._VISUAL,
}
]
},
"oral",
)
self.assertIn("旁白太短", str(ctx.exception))
def test_oral_meeting_floor_passes(self):
narration = "下午三点工位犯困，键盘都敲不利索。我倒了杯茶，第一口是回甘不是苦，整个人才慢慢醒过来。"
self.assertGreaterEqual(len(narration.replace(" ", "")), narration_floor(15))
self.assertGreaterEqual(len(self._VISUAL.replace(" ", "")), VISUAL_CHARS_MIN)
assert_shot_density(
{"segments": [{"duration": 15, "narration": narration, "visual": self._VISUAL}]},
"oral",
)
def test_short_visual_is_rejected(self):
with self.assertRaises(ValueError) as ctx:
assert_shot_density(
{
"segments": [
{
"duration": 15,
"narration": "下午三点工位犯困，键盘都敲不利索。我倒了杯茶，第一口是回甘不是苦，整个人才慢慢醒过来。",
"visual": "女主举起杯子。",
}
]
},
"oral",
)
self.assertIn("画面描写太短", str(ctx.exception))
def test_vlog_allows_silent_shot(self):
assert_shot_density(
{"segments": [{"duration": 15, "narration": "", "visual": self._VISUAL}]},
"vlog",
)
def test_paragraph_visual_without_timestamps_is_rejected(self):
narration = "下午三点工位犯困，键盘都敲不利索。我倒了杯茶，第一口是回甘不是苦，整个人才慢慢醒过来。"
long_visual = (
"近景对镜头，女主从键盘抬眼，端起杯子喝一口，再切杯壁特写，蒸汽从杯口升起，"
"茶汤慢慢变深，窗外下午的光打在桌面上，手没有离开杯柄，整段都不写起止秒。"
)
self.assertGreaterEqual(len(long_visual.replace(" ", "")), VISUAL_CHARS_MIN)
with self.assertRaises(ValueError) as ctx:
assert_shot_density(
{"segments": [{"duration": 15, "narration": narration, "visual": long_visual}]},
"oral",
)
self.assertIn("按秒拆分镜", str(ctx.exception))
def test_beats_array_flattens_into_visual(self):
import json
raw = {
"segments": [
{
"role": "钩子",
"narration": "下午三点工位犯困，键盘都敲不利索。",
"beats": [
{"start": 0, "end": 3, "shot_size": "近景", "action": "女主对镜头抬眼，键盘还亮着"},
{"start": 3, "end": 8, "shot_size": "手持", "action": "右手把茶包放进热水杯，蒸汽升起"},
{"start": 8, "end": 12, "shot_size": "特写", "action": "茶汤从浅金慢慢变深"},
{"start": 12, "end": 15, "shot_size": "中近景", "action": "女主喝一口眼神松开"},
],
}
]
}
draft = normalize_draft(json.dumps(raw, ensure_ascii=False), aspect_ratio="9:16", total_duration=15)
visual = draft["segments"][0]["visual"]
self.assertIn("0-3s：", visual)
self.assertIn("近景", visual)
self.assertGreaterEqual(len(parse_visual_beats(visual)), 3)
def test_compose_keeps_timestamped_visual(self):
visual = compose_segment_visual({"visual": self._VISUAL})
self.assertGreaterEqual(len(parse_visual_beats(visual)), 3)
class ComboTests(SimpleTestCase):
def test_drama_cannot_pick_review(self):
@@ -160,3 +430,9 @@ class SkillLoadingTests(SimpleTestCase):
def test_skill_text_is_substantial(self):
# skills 没随镜像打进去时会退化成一句兜底,这里守住「提示词没丢」
self.assertGreater(len(load_ecommerce_skill("oral", "pain")), 5000)
def test_skill_now_requires_fixed_15s(self):
text = load_ecommerce_skill("oral", "pain")
self.assertIn("每镜固定 15 秒", text)
self.assertIn("商品原词", text)
self.assertNotIn("单镜 4–15 秒，允许不等长", text)
core/backend/apps/projects/views.py
+8 -5
View File
@@ -518,14 +518,13 @@ class ProjectViewSet(TeamScopedViewSetMixin, ModelViewSet):
@action(detail=True, methods=["post"], url_path="script-agent-stream", renderer_classes=[ServerSentEventRenderer])
def script_agent_stream(self, request, pk=None):
"""对话式脚本 agent · 流式(SSE)。出稿 + 改稿一体,多模型可选。
请求体:mode(auto|theme|revise)、prompt、model_config_id、selling_point_ids、
base_version_id(改稿)、aspect_ratio、total_duration(5-60,5 秒步进)、
请求体: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 "")
selling_point_ids = request.data.get("selling_point_ids") or []
base_version_id = request.data.get("base_version_id") or None
aspect_ratio = str(request.data.get("aspect_ratio") or "9:16")
# 非法值一律由 agent 侧 coerce 兜底(夹区间/回落默认),这里不做 400,避免生成被参数噪声打断
@@ -534,6 +533,9 @@ class ProjectViewSet(TeamScopedViewSetMixin, ModelViewSet):
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
@@ -568,6 +570,7 @@ class ProjectViewSet(TeamScopedViewSetMixin, ModelViewSet):
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"
@@ -894,8 +897,8 @@ class ProjectViewSet(TeamScopedViewSetMixin, ModelViewSet):
"""采用版分镜数变化时,同步 VideoSegment 的**数量和时长**:不足则尾部补 NOT_STARTED,
多出且尾部是「从未生成过」的段则裁掉(已生成的段绝不动)。
时长同步是二期补的:脚本镜可以不等长了,而出片、计价、时间线读的都是
VideoSegment.target_duration_seconds。不同步的话脚本写了 8 秒、出片仍按 15 秒跑。
时长同步:脚本镜主流程固定 15 秒,而出片、计价、时间线读的都是
VideoSegment.target_duration_seconds。不同步的话脚本写了 15 秒、出片仍按旧默认跑。
已出片的段不改时长——改了会跟已渲染的成片对不上。"""
if not script.is_adopted:
return