优化脚本

This commit is contained in:
Azmat@qq.com
2026-08-28 14:45:38 +08:00
parent 849b14a0f6
commit c0a0b9603b
37 changed files with 541 additions and 1893 deletions
+1 -2
View File
@@ -32,8 +32,7 @@ PROMPT_PLACEHOLDERS = {
"person_triview": [],
"product_triview": ["商品", "补充"],
"scene": ["场景描述"],
"storyboard_frame": ["设定", "场景上下文", "时长", "脚本", "补充"],
"video_segment": ["设定", "分镜", "脚本", "时长"],
"video_segment": ["设定", "风格", "脚本", "时长"],
}
+4 -2
View File
@@ -242,13 +242,15 @@ class AdminQualityWordTests(TestCase):
self.assertNotIn("停用词", words)
def test_prompt_templates_list_update_permission(self):
"""提示词模板:列表(6 条,带 label/placeholders)+ 改正文比例 + 权限 + 审计 + 生成侧立即生效。"""
"""提示词模板:列表(5 条,带 label/placeholders)+ 改正文比例 + 权限 + 审计 + 生成侧立即生效。
分镜图(storyboard_frame)随故事板一起下线,不再出现在后台可编辑列表里。"""
from apps.ai.services import build_product_triview_prompt_refs
self.assertEqual(self.nc.get("/api/admin/prompt-templates/").status_code, 403)
lst = self.ac.get("/api/admin/prompt-templates/")
self.assertEqual(lst.status_code, 200)
self.assertEqual(len(lst.data), 6) # 视频线 6
self.assertEqual(len(lst.data), 5) # 视频线 5(分镜图已下线)
self.assertNotIn("storyboard_frame", [r["key"] for r in lst.data])
row = next(r for r in lst.data if r["key"] == "product_triview")
self.assertEqual(row["label"], "商品三视图")
self.assertIn("商品", row["placeholders"])
+2 -1
View File
@@ -329,7 +329,8 @@ def admin_quality_word_detail(request, word_id):
@permission_classes([IsPlatformAdmin])
def admin_prompt_templates(request):
"""列视频线 6 条提示词模板(固定 key,数量小不分页,按 key 排序)。"""
qs = PromptTemplate.objects.all().order_by("key")
# 故事板已下线,分镜图提示词不再出现在后台可编辑列表里(历史行留在库里不动)
qs = PromptTemplate.objects.exclude(key=PromptTemplate.Key.STORYBOARD_FRAME).order_by("key")
return Response(PromptTemplateSerializer(qs, many=True).data)
@@ -0,0 +1,34 @@
"""故事板下线 + 视频提示词回归代码默认。
背景(踩过的坑):`render_prompt()` 里**库里的 PromptTemplate 行优先于代码里的写死默认**。
`video_segment` 那行长期停留在旧版正文,导致代码默认一路加强(秒级分镜、物理常识…)线上却一直
用着那条短提示词。故事板下线后正文又要大改,再往库里写一份全文只会重演同一出。
因此这里把 `video_segment` 的正文清空 —— 按 admin 页写明的语义「留空 → 回落写死默认」,
让它此后永远跟着 `build_video_segment_prompt` 的默认走;超管仍可随时在后台填正文覆盖。
同时删掉已下线的 `storyboard_frame` 行。
"""
from django.db import migrations
VIDEO_LEGACY_MARKER = "{分镜}" # 故事板时代的占位符;只有还带着它的行才算「没人改过的旧默认」
def forwards(apps, schema_editor):
PromptTemplate = apps.get_model("ai", "PromptTemplate")
for row in PromptTemplate.objects.filter(key="video_segment"):
# 只清「故事板时代」的正文;超管自己写过的新正文(不含 {分镜})原样保留,不越权覆盖
if VIDEO_LEGACY_MARKER in (row.template or ""):
row.template = ""
row.save(update_fields=["template"])
PromptTemplate.objects.filter(key="storyboard_frame").delete()
def backwards(apps, schema_editor):
# 回滚只补回分镜图那行的存在性(正文留空 = 回落当时的写死默认),不还原已清空的视频正文
PromptTemplate = apps.get_model("ai", "PromptTemplate")
PromptTemplate.objects.get_or_create(key="storyboard_frame", defaults={"template": "", "ratio": "portrait"})
class Migration(migrations.Migration):
dependencies = [("ai", "0031_script_beat_prompt_templates")]
operations = [migrations.RunPython(forwards, backwards)]
+59 -17
View File
@@ -137,15 +137,21 @@ _PERSONA_KEY_BY_LABEL = {label: key for key, label in PERSONA_LABELS.items()}
NARRATION_CHARS_PER_SECOND = 5.2
NARRATION_CHARS_PER_SECOND_MIN = 4.3
NARRATION_CHARS_HARD_CAP = 78
VISUAL_CHARS_MIN = 72
# 故事板下线后,visual 是出片模型唯一的画面依据,门槛从「够写一句」抬到「够当导演说明书」
VISUAL_CHARS_MIN = 110
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 = (
"特写", "近景", "中近景", "中景", "全景", "远景", "胸上", "过肩",
"手持", "跟拍", "俯拍", "仰拍", "推近", "拉远", "",
"大特写", "特写", "近景", "中近景", "中景", "全景", "远景", "胸上", "过肩",
)
_CAMERA_MOVE_MARKERS = (
"手持", "跟拍", "跟随", "俯拍", "仰拍", "平视", "推近", "推进", "推镜",
"拉远", "拉回", "拉镜", "摇镜", "横摇", "移镜", "平移", "环绕", "升降",
"固定机位", "定机位", "固定镜头", "过肩", "第一人称", "越肩",
)
_MINOR_CHARACTER_RE = re.compile(
r"(?:婴儿|宝宝|宝贝|幼儿|儿童|小孩|小朋友|未成年|男童|女童|baby|toddler|infant)",
@@ -501,20 +507,47 @@ _CREATIVE_DIRECTION = """
输出前自问:遮住商品名后,这是不是仍然像一个真实的人在讲自己刚遇到的一件事?
如果像广告口号,重写得更具体、更有立场。
### visual 必须写成「导演说明书」,不是一句画面摘要
每个 segment 的 `visual` 采用下面的层级;这不是给用户看的散文,而是会原样交给故事板和视频模型的执行指令:
### visual 必须写成「导演说明书」,因为它是出片模型唯一的画面依据
**这条最重要:流程里没有分镜图了。** 你写的 `visual` 不会再经过一张关键帧图去"翻译"
而是**连同角色/商品/场景参考图一起,原样交给视频生成模型**。你没写清楚的,模型就自己编;
你写含糊的,画面就含糊。所以 `visual` 要写到「一个不认识这个商品的摄影师,照着就能拍」的程度。
每个 segment 的 `visual` 采用下面的层级,逐栏写满:
```
【本镜任务】这一段要让观众看懂的变化或悬念。
【光线氛围】光源方向与性质(窗光/顶光/台灯/逆光);色温冷暖;明暗对比;整体色调。全片各镜保持一致。
【声音】台词/旁白状态;音效;背景音乐;字幕。没有就写「无」;没有配乐写「无配乐,仅同期声」。
【画面内容】
0-3s:景别;机位;运镜;人物/商品动作;情绪或信息变化。
3-8s:景别;机位;运镜;手与商品的空间关系;可见细节。
8-12s:景别;机位;运镜;动作结果或卖点证据。
0-3s:景别;机位高度与角度;运镜;主体在画面中的位置;具体动作;情绪或信息变化。
3-8s:景别;机位;运镜;手与商品的空间关系(哪只手、握哪里、从哪个方向入画);可见细节。
8-12s:景别;机位;运镜;动作结果或卖点证据(看得见的变化:颜色/质地/形态/前后差别)
12-15s:景别;机位;运镜;反应/悬念/自然收束。
```
每条秒级分镜都要明确写出 **景别、机位、运镜、动作、信息变化** 五项中的至少四项。
输入如果是 `【镜头 01】` 导演分镜稿,把原稿的景别、机位、运镜、动作、表情、音效、背景音乐、字幕、备注折进对应栏,不要压成一句画面摘要。
同一角色、商品、场景的外观一律引用 entities 里的既定设定,不在每一镜随意换发型、服装、包装、光线或地点。
每条秒级分镜必须同时写出这五项,不能省:
1. **景别** —— 大特写/特写/近景/中近景/中景/全景(15 秒内至少切两次,别一个景别拍到底)
2. **机位** —— 高度与角度:平视/俯拍/仰拍/过肩/桌面视角
3. **运镜** —— 手持跟拍/推近/拉远/横摇/环绕/固定机位(**每镜至少出现一个运镜词**,否则画面会死)
4. **动作** —— 具体到肢体:谁、用哪只手、对什么、做了什么动作,动作要连贯可拍
5. **信息变化** —— 这 3–5 秒里画面上多了什么、变了什么
### 只写拍得出来的东西
- **禁止抽象词**:「展示商品」「呈现质感」「体现高级感」「氛围到位」「传递温暖」——
这些不是画面,是评价。要写成「杯壁上的水珠顺着往下滑」「她眉头松开,肩膀塌下来」。
- **禁止心理描写**:模型拍不出「她内心很纠结」,要写「她拿起又放下,手指在包装边缘停了两秒」。
- **禁止画面里出现文字**:不要写字幕、花字、标题、价格贴片、UI、弹窗、对比图表;
商品包装上原有的真实文字除外。
- **禁止一镜内跨场景/跨时间蒙太奇**:15 秒是**一个连续的动作链**,同一地点、同一光线、
同一套衣服。要换环境就换到下一镜,并在 entity_refs 里换成另一个 scene。
### 一致性靠文字锁死
同一角色、商品、场景的外观一律引用 entities 里的既定设定;不在每一镜随意换发型、服装、
包装、光线、地点或色调。每一镜的【光线氛围】要和相邻镜衔接得上(同一场戏不能上一镜暖黄台灯、
下一镜冷白日光)。商品的用法要符合物理常识:热饮有蒸汽、液体会流动、包装要先打开才能取出内容物。
输入如果是 `【镜头 01】` 导演分镜稿,把原稿的景别、机位、运镜、动作、表情、音效、背景音乐、字幕、备注
折进对应栏,不要压成一句画面摘要。
"""
# 这些句式几乎总会让首屏像模板广告。只作为生成后的最后一道门,不替代模型的创作判断。
@@ -698,12 +731,16 @@ def build_agent_messages(
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"
f"【秒级分镜】流程里没有分镜图了,visual 会**原样交给出片模型**,写多细画面就有多准。"
f"每个 {SEGMENT_DURATION_MAX} 秒场必须拆成 3–5 个分镜,visual 按「导演说明书」写成多行:"
"先写【本镜任务】、【光线氛围】(光源方向/色温/明暗/色调,各镜保持一致)、【声音】、【画面内容】,再写秒级分镜;"
f"每条格式`0-3s:景别;机位;运镜;主体位置与具体动作;信息变化`,最后一条接到 {SEGMENT_DURATION_MAX}s。"
"每条必须带景别(15 秒内至少切两次)+ 机位 + 运镜词(手持/跟拍/推近/拉远/环绕/固定机位,缺了画面会死),"
"并写清哪只手、从哪个方向入画、握商品哪里,以及真实用法"
"(茶=热水+蒸汽+茶汤变色,禁止茶包丢进冷白开;禁止悬浮肢体、商品凭空出现)。"
"只写拍得出来的东西:禁止「展示商品/呈现质感/体现高级感」这类评价词,禁止心理描写,"
"禁止画面里出现字幕花字标题价签,禁止一镜内跨场景跨时间。"
f"禁止一句空画面撑满 {SEGMENT_DURATION_MAX} 秒。允许另给 beats 数组,后端会折进 visual。\n"
)
if fmt == "oral":
writing_line = (
@@ -1025,6 +1062,11 @@ def assert_intra_shot_beats(seg: dict, index: int, duration: int) -> None:
markers = [mark for mark in _SHOT_SIZE_MARKERS if mark in visual]
if len(set(markers)) < 2:
raise ValueError(f"{index + 1} 镜 15 秒内至少要切两次景别,并写进秒级分镜")
# 故事板下线后没有关键帧兜底,机位/运镜只能靠这段文字告诉出片模型
if not any(mark in visual for mark in _CAMERA_MOVE_MARKERS):
raise ValueError(
f"{index + 1} 镜必须写清机位与运镜(手持/跟拍/推近/拉远/环绕/固定机位…),否则出片只会拍成呆板静止画面"
)
# 模型每次生成都可能换字段名(scene/screenDescription/visual…、dialogue/lines/caption…),
+33 -427
View File
@@ -41,10 +41,6 @@ from apps.projects.models import (
ProjectStage,
ScriptSegment,
ScriptVersion,
StoryboardFrame,
StoryboardShot,
StoryboardShotVersion,
StoryboardVersion,
Timeline,
VideoSegment,
VideoSegmentVersion,
@@ -67,26 +63,6 @@ def get_default_model(capability: str) -> ModelConfig:
return qs.filter(is_default=True).order_by("created_at").first() or qs.order_by("created_at").first()
def get_storyboard_image_model() -> ModelConfig | None:
"""故事板出图只走 GPT 图像模型(gpt-image / gpt-image-2)。
不回落默认图像模型,也不走火山 Seedream:用户明确要求故事板无论怎样都不改成其他生图模型。
"""
qs = (
ModelConfig.objects.select_related("provider")
.filter(
capability=ModelConfig.Capability.IMAGE,
status=ModelConfig.Status.ACTIVE,
provider__status="active",
name__icontains="gpt-image",
)
)
return (
qs.filter(name="gpt-image-2").order_by("created_at").first()
or qs.order_by("created_at").first()
)
def resolve_image_model(key: str | None) -> "ModelConfig | None":
"""前端「生图模型选择」→ ModelConfig。用户显式选的可以是 disabled 模型(故不按 status 过滤)。
· "volcano" → 火山官方 Seedream(取最新一版)
@@ -901,20 +877,6 @@ NO_EMBEDDED_CAPTIONS_REQUIREMENT = (
"只保留商品包装上参考图中原有的真实物理文字与标识。旁白和环境音可以保留,但不要把声音转成画面文字。"
)
# 故事板是创作者确认镜头节奏的导演稿,因此必须由 image-2 直接生成完整排版图;
# 视频提示词会明确要求只提取其中的角色、商品、场景、动作和节奏,忽略版式文字。
STORYBOARD_DIRECTOR_LAYOUT_REQUIREMENT = (
"【导演故事板成图硬性要求】输出必须是一张完整、正式、可交付的电商短视频导演故事板图片,而不是无说明的拼图或单张海报。"
"整体为竖版 9:16,采用清爽专业的影视分镜排版:顶部 1 行标题区,标出“导演故事板”、本段时长和画幅;"
"主体严格按照本段脚本的秒级分镜,优先整理为 3 个连续时间段(例如 0–5s、5–10s、10–15s;若脚本时长不同则按实际时间划分)。"
"每个时间段必须包含三部分:左侧约 22% 宽的浅色信息栏,清晰写时间、景别、机位/运镜和简短动作;右侧约 78% 的写实主画面;"
"画面底部横条使用扬声器图标与“口播/旁白:”呈现该时间段对应旁白。各段按时间自上而下排列,边框、间距、对齐统一。"
"右侧画面要真实展现该秒级分镜的不同关键动作,人物脸、服装、商品外观、配色和场景连续一致;商品用法必须正确,包装真实文字和 Logo 不得改造。"
"故事板中的中文标注必须工整、简洁、可读,只写脚本已有的时间、机位、动作与旁白,不虚构价格、功效、活动或品牌信息。"
"禁止只给一张静态人物图、禁止无文字的四格拼图、禁止把三段画面重复成同一姿势。"
)
def enforce_no_embedded_captions(prompt: str) -> str:
"""所有视频入口最终汇入这里,避免某个入口漏传“无字幕”导致模型自行加花字。"""
base = (prompt or "").strip()
@@ -1470,20 +1432,6 @@ def project_output_spec(project) -> dict:
}
def _storyboard_canvas_phrase(ratio: str) -> str:
r = (ratio or "9:16").strip() or "9:16"
if r in {"9:16", "3:4"}:
return f"电商竖屏 {r}"
if r == "1:1":
return f"电商方形 {r}"
return f"电商横屏 {r}"
def _apply_storyboard_output_ratio(text: str, project) -> str:
phrase = _storyboard_canvas_phrase(project_output_spec(project)["aspect_ratio"])
return (text or "").replace("电商竖屏 9:16", phrase)
def _sync_timeline_output_spec(project, *, aspect_ratio: str, resolution: str) -> None:
from apps.ai.video_pricing import get_resolution
@@ -2702,125 +2650,49 @@ def _segment_script_text(segment, entities=None, *, with_dialogue: bool = False,
return "\n".join(parts)
def build_storyboard_frame_prompt(project, segment, extra_prompt: str = "") -> str:
"""单镜导演故事板提示词(无参考图时的文本版)。与 refs 版共用 admin「提示词·分镜图」模板(设定为空)。
extra_prompt = 整张风格提示词(原 StoryboardVersion.prompt,现存项目级)。"""
dur = segment.duration_seconds or 15
default = (
"{设定}根据以下脚本生成一个导演故事板,用于指导 seedance 的视频生成。\n{场景上下文}\n"
"【分镜脚本】(本段时长约 {时长} 秒)\n{脚本}\n"
"请严格保持各参考图中角色的同一张脸、同一商品的外观与配色;"
"电商竖屏 9:16 导演故事板,一镜一图,画面清晰,可直接指导视频生成。"
"脚本里的 visual 是本段秒级分镜清单;请画商品用法正确、最能看清商品的那一拍作为关键帧。"
"禁止画出违背常识的用法(例如把茶包丢进冷白开、悬浮的手)。{补充}"
)
rendered = render_prompt(
"storyboard_frame", default,
设定="",
场景上下文=_scene_context(project),
时长=dur,
脚本=(_segment_script_text(segment) or f"{segment.sort_order + 1}"),
补充=(("\n" + extra_prompt.strip()) if extra_prompt else ""),
)
cleaned = "\n".join(line for line in rendered.split("\n") if line.strip())
return _apply_storyboard_output_ratio(
f"{cleaned}\n{STORYBOARD_DIRECTOR_LAYOUT_REQUIREMENT}", project
)
def build_video_segment_prompt(project, video_segment, scene, refs, user_prompt: str = "") -> str:
"""单段视频提示词(用户钦定 @图N 格式):
【设定】@图N 点名 角色/场景/商品;【分镜】根据 @图(分镜图) 生成;【脚本】本镜脚本
【设定】@图N 点名 角色/场景/商品;【风格】全片统一的风格锚点;【脚本】本镜导演说明书(秒级分镜 + 台词)
★ 去掉故事板中间层后,导演信息不再经由一张分镜图传递,而是**全靠这份提示词直达出片模型**,
因此这里必须自足:参考图一致性约束 + 秒级执行纪律 + 物理常识 + 成片硬性规则,一条都不能少。
refs 顺序与传给 seedance 的 reference_images 一致(@图N 对齐不错位)。"""
product_name = (getattr(project.product, "title", "") or "商品").strip()
setup_parts = []
storyboard_idx = None
for i, r in enumerate(refs or []):
n = i + 1
if r.get("type") == "storyboard":
storyboard_idx = n
else:
setup_parts.append(f"@图{n}{r.get('label') or ''}({_ENTITY_TYPE_CN.get(r.get('type'), '参考')})")
setup_parts.append(f"@图{n}{r.get('label') or ''}({_ENTITY_TYPE_CN.get(r.get('type'), '参考')})")
# 视频脚本:用带说话人的台词(原样照脚本),不带「商品露出」那行(露出靠模板尾句统一要求)
entities = (project.metadata or {}).get("script_entities", [])
script_text = _segment_script_text(scene, entities, with_dialogue=True, with_exposure=False) if scene is not None else ""
extra = (user_prompt or "").strip()
if extra:
script_text = (script_text + "\n" + extra).strip() if script_text else extra
# 正文可在 admin「提示词·视频」页改。占位符:{设定}=@图N点名、{分镜}=分镜图引用、{脚本}=本段脚本、{时长}。
# 正文可在 admin「提示词·视频」页改。占位符:{设定}=@图N点名、{风格}=全片风格锚点、{脚本}=本段脚本、{时长}。
# 时长/比例靠 API 参数传(duration/ratio),不写进正文;尾句给风格 + 音效/字幕要求。
default = (
"{设定}{分镜}【脚本】{脚本}\n"
"严格按脚本里的秒级分镜切换景别和动作,商品用法必须真实,不要诡异姿势或错误容器。"
"如果参考图是带时间、机位、旁白标注的导演故事板,只提取其中的角色、商品、场景、动作和时间顺序;"
"不要把边框、标题、时间码、扬声器图标或任何故事板文字生成进视频。"
"电商带货短视频,商品露出清晰,节奏有转化感。不要字幕,不要背景音乐,但是要有音效,逼真的音效。"
"{设定}{风格}【脚本】{脚本}\n"
"【执行纪律】严格按脚本里的秒级分镜拍:每一个时间段都要拍到,景别、机位、运镜、动作按写的来;"
"不要把整段并成一个静止长镜头,也不要加脚本里没有的转场、人物或道具。"
"\n【一致性】参考图是本片唯一的视觉依据:角色保持同一张脸、同一发型、同一套服装;"
"商品保持参考图的外形、配色、材质、比例与包装上的真实文字标识,不得改造、换色、换包装或凭空加配件;"
"场景保持同一空间、同一陈设、同一光线方向与色温。多镜之间人物与商品必须看起来是同一次拍摄。"
"\n【物理常识】手从画面内自然入画,禁止悬浮肢体、反关节、商品凭空出现或消失;"
"商品用法必须是真人会做的(茶/咖啡用热水、有蒸汽与茶汤渐染;护肤品挤出并涂抹;食品打开并入口),"
"禁止诡异姿势、错误容器或违背常识的操作。"
"\n【画质】真人实拍质感,自然光影,肤色与材质真实,焦点始终落在脚本指定的主体上,画面稳定不糊。"
"\n电商带货短视频,商品露出清晰,节奏有转化感。不要字幕,不要背景音乐,但是要有音效,逼真的音效。"
)
rendered = render_prompt(
"video_segment", default,
设定=("【设定】" + "".join(setup_parts) + "\n" if setup_parts else ""),
分镜=(f"分镜】根据@图{storyboard_idx}分镜图生成「{product_name}」短视频\n" if storyboard_idx is not None else ""),
风格=(f"风格】{_scene_context(project)}\n" if _scene_context(project) else ""),
分镜="", # 兼容:旧模板行里可能还留着 {分镜}(故事板时代的占位符),渲染成空串而不是原样漏出
脚本=(script_text or f"{video_segment.sort_order + 1}"),
时长=video_segment.target_duration_seconds,
)
return enforce_no_embedded_captions("\n".join(line for line in rendered.split("\n") if line.strip()))
def ensure_storyboard_shots(project) -> list:
"""确保「采用版分镜数」= StoryboardShot 数:每镜一个 shot(对标视频段)。
缺则按 sort_order 建、补绑 script_segment;多出且从没出过图的尾 shot 裁掉(已出图的不动)。"""
adopted_script = project.script_versions.filter(is_adopted=True).prefetch_related("segments").first()
if adopted_script is None:
return []
segs = list(adopted_script.segments.order_by("sort_order"))
shots = {s.sort_order: s for s in project.storyboard_shots.all()}
for seg in segs:
shot = shots.get(seg.sort_order)
if shot is None:
shots[seg.sort_order] = StoryboardShot.objects.create(
project=project, script_segment=seg, sort_order=seg.sort_order, prompt=seg.visual_prompt or "")
elif shot.script_segment_id != seg.id:
shot.script_segment = seg
shot.save(update_fields=["script_segment", "updated_at"])
target = len(segs)
for order, shot in list(shots.items()):
if order >= target and not shot.versions.exists():
shot.delete()
del shots[order]
return [shots[k] for k in sorted(shots)]
def submit_storyboard(*, project, user, prompt: str = "", shot_ids: list | None = None) -> list:
"""故事板·提交(分镜制,对标视频「开始生成/单段重跑」):确保每镜一个 shot,把目标场标 QUEUED,
真正出图由 poll_storyboard 起后台线程。返回受影响的 shots。
- shot_ids 给定 → 只(重)生成这些场(单场重跑);
- 不给 → 还没出片的场;若全部已出片 → 视为整批重跑,全部重出。
prompt = 整张风格提示词,存项目级 metadata,供每场出图带上(原 StoryboardVersion.prompt 的去处)。"""
adopted_script = project.script_versions.filter(is_adopted=True).first()
if adopted_script is None:
raise ValueError("script must be adopted before generating storyboard")
if get_storyboard_image_model() is None:
raise ValueError("故事板只使用 GPT 图像模型,当前没有启用 gpt-image-2")
if prompt:
meta = dict(project.metadata or {})
if meta.get("storyboard_prompt") != prompt:
meta["storyboard_prompt"] = prompt
project.metadata = meta
project.save(update_fields=["metadata", "updated_at"])
shots = ensure_storyboard_shots(project)
if shot_ids is not None:
want = {str(x) for x in shot_ids}
targets = [s for s in shots if str(s.id) in want]
else:
pending = [s for s in shots if s.status != StoryboardShot.Status.SUCCEEDED or s.adopted_version_id is None]
targets = pending if pending else shots # 全部已出片 = 整批重跑
for s in targets:
s.status = StoryboardShot.Status.QUEUED
s.error_message = ""
s.save(update_fields=["status", "error_message", "updated_at"])
return targets
_ENTITY_TYPE_CN = {"character": "角色", "scene": "场景", "product": "商品"}
@@ -2846,9 +2718,9 @@ def _product_reference_image(project, groups: list | None = None) -> dict | None
return None
def _storyboard_reference_images(project, segment) -> list[dict]:
def _segment_reference_images(project, segment) -> list[dict]:
"""按本镜 entity_refs 取参考图(角色 / 场景 已采用基础资产)+ **无条件带上商品参考图**,
gpt-image-2 多图合成 @图N。返回 [{url,label,type}],最多 4 张(角色/场景 ≤3 + 商品 1)。
出片模型 @图N 锁脸 / 锁商品 / 锁场景。返回 [{url,label,type}],最多 4 张(角色/场景 ≤3 + 商品 1)。
商品不靠 entity_refs(预创建真实商品、不从脚本提),统一用商品三视图 / 主图带上,从根上保证
商品参考永不缺失。依赖提取步落进 metadata 的 script_entities。"""
entities = {
@@ -2911,65 +2783,6 @@ def _storyboard_reference_images(project, segment) -> list[dict]:
return out[:4]
def build_storyboard_frame_prompt_refs(project, segment, refs: list[dict], extra_prompt: str = "") -> str:
"""参考图合成版(用户钦定格式):顶部 @图N 点名每张参考图(角色/商品/场景),再给导演故事板指令 + 分镜脚本。
@图N 顺序与传给 gpt-image-2 的参考图顺序严格一致(refs 即 image_edit 的 images 顺序)。"""
if not refs:
return build_storyboard_frame_prompt(project, segment, extra_prompt)
setup = "".join(
f"@图{i + 1}{r['label']}({_ENTITY_TYPE_CN.get(r.get('type'), '参考')})" for i, r in enumerate(refs)
)
dur = segment.duration_seconds or 15
# 正文可在 admin「提示词」页改。占位符:{设定}=@图N点名、{场景上下文}、{时长}、{脚本}=本镜脚本、{补充}=本版附加。
default = (
"{设定}根据以下脚本生成一个导演故事板,用于指导 seedance 的视频生成。\n{场景上下文}\n"
"【分镜脚本】(本段时长约 {时长} 秒)\n{脚本}\n"
"请严格保持各参考图中角色的同一张脸、同一商品的外观与配色;"
"电商竖屏 9:16 导演故事板,一镜一图,画面清晰,可直接指导视频生成。"
"脚本里的 visual 是本段秒级分镜清单;请画商品用法正确、最能看清商品的那一拍作为关键帧。"
"禁止画出违背常识的用法(例如把茶包丢进冷白开、悬浮的手)。{补充}"
)
rendered = render_prompt(
"storyboard_frame", default,
设定=(f"【设定】{setup}\n" if setup else ""),
场景上下文=_scene_context(project),
时长=dur,
脚本=(_segment_script_text(segment) or f"{segment.sort_order + 1}"),
补充=(("\n" + extra_prompt.strip()) if extra_prompt else ""),
)
cleaned = "\n".join(line for line in rendered.split("\n") if line.strip())
return _apply_storyboard_output_ratio(
f"{cleaned}\n{STORYBOARD_DIRECTOR_LAYOUT_REQUIREMENT}", project
)
def _is_transient_error(exc: Exception) -> bool:
"""网络抖动/超时类瞬时错误(可重试),区别于内容审核拦截、参数非法等确定性失败。
中转站(tokenssr 等)偶发 Read timeout / 连接重置会无谓掐掉单帧,这类才重试。
★ 不重试 400:实测故事板的 400 主因是 gpt-image-2 内容审核拦截(moderation_blocked,
safety_violations=[sexual],如脚本写到「蕾丝胸罩/抚摸/贴身」)—— 同一画面/文案重试还是被拦,
重试只会拖延报错、空耗调用。这类应「快速失败 + 友好提示」让用户改提示词,而非闷头重试。"""
msg = str(exc).lower()
return any(
k in msg
for k in ("timed out", "timeout", "connection", "reset by peer", "temporarily",
"bad gateway", "502", "503", "504", "remotedisconnected", "max retries")
)
# 审核类别英文 → 给用户的中文说明(摘自第三方 safety_violations 字段)。命不中则原样保留英文。
_MODERATION_CATEGORY_CN = {
"sexual": "性暗示 / 露骨",
"sexual/minors": "涉及未成年的性内容",
"violence": "暴力",
"violence/graphic": "血腥暴力",
"self-harm": "自残",
"hate": "仇恨",
"harassment": "骚扰",
"illicit": "违禁",
}
def _extract_moderation_categories(raw: str) -> list[str]:
"""从原始报错里抽出被审核命中的类别(如 safety_violations=[sexual] / "categories":["sexual"]),
译成中文标签。抽不到返回 []。供友好提示点名真实类别,而非泛化的「疑似敏感内容」。"""
@@ -3061,192 +2874,6 @@ def notify_generation_failure(
logger.exception("notify_generation_failure failed for task %s", getattr(task, "id", "?"))
def _call_image_with_retry(fn, *, attempts: int = 2, base_delay: float = 2.0):
"""对一次出图网络调用做有界重试:仅瞬时错误重试(指数退避),确定性失败立即抛出。
出图无副作用(失败=没拿到图),重试安全;成功一次即返回。
attempts 默认 2(一次重试):单次 HTTP 超时上限 300s,2 次≈10min,须 < poll 的「在途锁过期窗口」
(STORYBOARD_INFLIGHT_STALE_MINUTES)否则 worker 重试期间任务被判僵尸 → 重复起线程 + 重复扣费。"""
import time
last: Exception | None = None
for i in range(attempts):
try:
return fn()
except Exception as exc: # noqa: BLE001
last = exc
if i == attempts - 1 or not _is_transient_error(exc):
raise
time.sleep(base_delay * (i + 1))
raise last # 理论不可达(循环内已 return/raise)
def _storyboard_shot_worker(task_id, shot_id, user_id) -> None:
"""后台线程:为一个 StoryboardShot 出一张图 → 落成一条 StoryboardShotVersion 并采用。HTTP 永远秒回。"""
from django.db import connections
from apps.accounts.models import User
try:
task = AITask.objects.select_related("model_config__provider").get(id=task_id)
shot = StoryboardShot.objects.select_related("project__team", "script_segment").get(id=shot_id)
user = User.objects.get(id=user_id)
project = shot.project
segment = shot.script_segment
reservation = task.credit_reservation
extra_prompt = (project.metadata or {}).get("storyboard_prompt", "") or ""
spec = project_output_spec(project)
frame_ratio = spec["aspect_ratio"]
frame_size = _ratio_to_image_size(frame_ratio)
task.status = AITask.Status.SUBMITTED
task.save(update_fields=["status", "updated_at"])
try:
# 故事板无论任务上挂了什么模型、是否开了路由,都只走 GPT 图像;失败也不换 Seedream。
model_config = get_storyboard_image_model()
if model_config is None:
raise ValueError("故事板只使用 GPT 图像模型,当前没有启用 gpt-image-2")
provider = get_image_provider(model_config)
refs = _storyboard_reference_images(project, segment) if segment is not None else []
ref_urls = [r["url"] for r in refs]
if ref_urls and hasattr(provider, "image_edit"):
# gpt-image-2 多图参考:必须用 refs 版提示词(点名「参考图N=角色/场景/商品」+锁脸锁商品)
frame_prompt = build_storyboard_frame_prompt_refs(project, segment, refs, extra_prompt)
else:
frame_prompt = (
build_storyboard_frame_prompt(project, segment, extra_prompt) if segment is not None
else (task.request_payload.get("prompt") or "")
)
if ref_urls and hasattr(provider, "image_edit"):
response = _call_image_with_retry(
lambda: provider.image_edit(
model=model_config.name,
prompt=frame_prompt,
images=ref_urls,
size=frame_size,
)
)
else:
response = _call_image_with_retry(
lambda: provider.image_generation(
model=model_config.name,
endpoint=model_config.endpoint,
prompt=frame_prompt,
)
)
media = provider.extract_first_media_url(response)
asset = _store_generated_media(
team=project.team, user=user, project=project, task=task, media=media,
name=f"{project.name}-storyboard-{shot.sort_order + 1}",
category=Asset.Category.STORYBOARD, asset_type=Asset.Type.IMAGE,
)
with transaction.atomic():
task.status = AITask.Status.SUCCEEDED
task.response_payload = response
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)
# 落一条新版本并采用(反采用同 shot 其余版本)= 该场历史 +1,采用最新
version = StoryboardShotVersion.objects.create(
shot=shot, task=task, asset=asset,
prompt=(segment.visual_prompt if segment is not None else ""), is_adopted=True,
)
shot.versions.exclude(id=version.id).update(is_adopted=False)
StoryboardShot.objects.filter(id=shot.id).update(
adopted_version=version, status=StoryboardShot.Status.SUCCEEDED, error_message="", updated_at=timezone.now())
# 合规:分镜图含人脸,视频生成前必须过火山审核 → 事务提交后静默送审(best-effort)
from apps.assets.review import submit_asset_for_review
transaction.on_commit(lambda a=asset: submit_asset_for_review(a))
except Exception as exc: # noqa: BLE001 — 失败回滚额度,标记任务+shot 失败供 poll 上报
raw = str(exc)
public_error = classify_generation_error(
exc, operation="storyboard_generate", reference_id=str(task.id)
)
task.status = AITask.Status.FAILED
task.error_message = raw[:2000]
task.completed_at = timezone.now()
task.save(update_fields=["status", "error_message", "completed_at", "updated_at"])
release_credit(reservation=reservation, reason=raw[:200])
# 重跑失败时保留旧 adopted_version(画面不丢),只把状态标 FAILED + 友好提示供前端显示
StoryboardShot.objects.filter(id=shot.id).update(
status=StoryboardShot.Status.FAILED, error_message=public_error.fallback_message, updated_at=timezone.now())
# 落一条失败通知,正文带上第三方服务商 API 的原始报错(真因),供用户/排查直接查看
notify_generation_failure(
task=task, project=project, recipient=user,
stage_label=f"故事板·场 {shot.sort_order + 1}", raw=raw, hint=public_error.fallback_message,
)
finally:
connections.close_all() # 释放该线程的 DB 连接
def poll_storyboard(*, project, user) -> dict:
"""异步故事板·轮询(秒回):为 QUEUED/在制 shot 起线程出图;报告进度。永不阻塞在 ARK 调用上。
返回 {status: generating|succeeded|failed, done, total}。done = 已出片(有采用版且 SUCCEEDED)的场数。"""
import threading
from django.conf import settings as dj_settings
shots = list(project.storyboard_shots.select_related("script_segment").order_by("sort_order"))
if not shots:
return {"status": "succeeded", "done": 0, "total": 0}
total = len(shots)
done = sum(1 for s in shots if s.adopted_version_id is not None and s.status == StoryboardShot.Status.SUCCEEDED)
active = [s for s in shots if s.status in (StoryboardShot.Status.QUEUED, StoryboardShot.Status.RUNNING)]
if not active:
failed = [s for s in shots if s.status == StoryboardShot.Status.FAILED]
if failed:
return {"status": "failed", "done": done, "total": total, "error": failed[0].error_message or "storyboard shot failed"}
return {"status": "succeeded", "done": done, "total": total}
# 每镜独立「占位锁」:近 N 分钟内有 CREATED/RESERVED/SUBMITTED 任务的 shot = 在生成中(僵尸超时后释放)。
STORYBOARD_MAX_PARALLEL = int(getattr(dj_settings, "STORYBOARD_MAX_PARALLEL", 4))
stale_minutes = int(getattr(dj_settings, "STORYBOARD_INFLIGHT_STALE_MINUTES", 12))
stale_cutoff = timezone.now() - timedelta(minutes=stale_minutes)
inflight_shot_ids = {
str(v)
for v in AITask.objects.filter(
project=project, task_type=AITask.Type.STORYBOARD,
status__in=[AITask.Status.CREATED, AITask.Status.RESERVED, AITask.Status.SUBMITTED],
created_at__gte=stale_cutoff,
).values_list("request_payload__storyboard_shot", flat=True)
if v
}
model_config = get_storyboard_image_model()
if model_config is None:
return {"status": "failed", "done": done, "total": total, "error": "故事板只使用 GPT 图像模型,当前没有启用 gpt-image-2"}
extra_prompt = (project.metadata or {}).get("storyboard_prompt", "") or ""
spawnable = [s for s in active if str(s.id) not in inflight_shot_ids]
slots = max(0, STORYBOARD_MAX_PARALLEL - len(inflight_shot_ids))
for shot in spawnable[:slots]:
segment = shot.script_segment
task = create_ai_task(
project=project, user=user, task_type=AITask.Type.STORYBOARD, model_config=model_config,
request_payload={
"model": model_config.name, "endpoint": model_config.endpoint,
"prompt": build_storyboard_frame_prompt(project, segment, extra_prompt) if segment is not None else "",
"storyboard_shot": str(shot.id),
},
)
StoryboardShot.objects.filter(id=shot.id).update(status=StoryboardShot.Status.RUNNING, updated_at=timezone.now())
threading.Thread(
target=_storyboard_shot_worker, args=(str(task.id), str(shot.id), str(user.id)), daemon=True
).start()
return {"status": "generating", "done": done, "total": total}
def adopt_storyboard_shot_version(*, shot: StoryboardShot, version: StoryboardShotVersion) -> None:
"""采用某场的某个历史版本(对标 adopt-video-version):反采用同场其余版本,置该场为采用版+SUCCEEDED。"""
shot.versions.exclude(id=version.id).update(is_adopted=False)
if not version.is_adopted:
version.is_adopted = True
version.save(update_fields=["is_adopted", "updated_at"])
shot.adopted_version = version
shot.status = StoryboardShot.Status.SUCCEEDED
shot.error_message = ""
shot.save(update_fields=["adopted_version", "status", "error_message", "updated_at"])
def _asset_preview_url(asset) -> str:
"""资产主文件的可公开访问 URL(已写绝对 URL 优先,否则实时签 TOS GET)。"""
if asset is None:
@@ -3274,8 +2901,10 @@ def _seedance_ref_url(raw_url: str, review_status: str = "", review_remote_id: s
def _video_reference_images(project, video_segment) -> list[dict]:
"""视频参考图(带类型,供 @图N):角色/场景/商品 基础资产 + 本镜故事板帧
顺序:角色 → 场景 → 商品 → 分镜图(与用户钦定 @图1角色@图2场景@图3商品@图4分镜图 一致)。
"""视频参考图(带类型,供 @图N):角色 / 场景 / 商品 基础资产
顺序:角色 → 场景 → 商品(与提示词里 @图1角色@图2场景@图3商品 一致)。
★ 故事板已从流程中去掉,不再有「分镜图」这一张:导演信息全部走提示词
(build_video_segment_prompt),参考图只负责锁脸 / 锁商品 / 锁场景。
返回 [{url,label,type}];url 对过审人脸资产为 asset:// 素材库引用。都取不到时兜底商品图。"""
out: list[dict] = []
scene = None
@@ -3283,21 +2912,10 @@ def _video_reference_images(project, video_segment) -> list[dict]:
if adopted_script is not None:
scene = adopted_script.segments.filter(sort_order=video_segment.sort_order).first()
if scene is not None:
refs = list(_storyboard_reference_images(project, scene)) # 角色/商品/场景 实体图
refs = list(_segment_reference_images(project, scene)) # 角色/商品/场景 实体图
_vord = {"character": 0, "scene": 1, "product": 2}
refs.sort(key=lambda r: _vord.get(r.get("type"), 9))
out = refs
# 末位追加本镜故事板分镜图(@图N 末位 = 分镜图):取该 sort_order 的 shot 的采用版资产
shot = (
project.storyboard_shots.filter(sort_order=video_segment.sort_order, adopted_version__isnull=False).select_related("adopted_version__asset").first()
or project.storyboard_shots.filter(adopted_version__isnull=False).select_related("adopted_version__asset").order_by("sort_order").first()
)
if shot is not None and shot.adopted_version_id:
frame_asset = shot.adopted_version.asset
url = _asset_preview_url(frame_asset)
if url:
out.append({"url": url, "label": "分镜图", "type": "storyboard",
"review_status": frame_asset.review_status, "review_remote_id": frame_asset.review_remote_id})
if not out:
product_group = (
project.base_asset_groups.filter(kind=BaseAssetGroup.Kind.PRODUCT, adopted_asset__isnull=False)
@@ -3314,19 +2932,17 @@ def _video_reference_images(project, video_segment) -> list[dict]:
def collect_video_review_blockers(project, only_segment: "VideoSegment | None" = None) -> list[dict]:
"""点「生成视频」前的过审闸:列出本次将生成的段里,含真人脸的参考图(人物立绘 + 该镜故事板分镜)
**尚未过审(review_status != 'active')** 的项。火山视频对含真人脸的图,只接受素材库已过审的引用,
否则报 InputImageSensitiveContentDetected。这里在调火山前先拦,弹窗指明是哪一镜的哪个人物/分镜未过审。
"""点「生成视频」前的过审闸:列出本次将生成的段里,含真人脸的参考图(人物立绘)
**尚未过审(review_status != 'active')** 的项。火山视频对含真人脸的图,只接受素材库已过审的引用,
否则报 InputImageSensitiveContentDetected。这里在调火山前先拦,弹窗指明是哪一镜的哪个人物未过审。
★ 故事板去掉后不再有「分镜图」这一类待审资产,闸口只剩人物立绘。
only_segment 给定 → 只校验该段(单段重跑);为 None → 校验全部「未出片」段(整批生成)。
返回 [{video_segment_id, sort_order, scene_no, kind:'person'|'storyboard', name, asset_id, review_status}];
返回 [{video_segment_id, sort_order, scene_no, kind:'person', name, asset_id, review_status}];
空列表 = 全部已过审,可放行。"""
segs = [only_segment] if only_segment is not None else list(project.video_segments.order_by("sort_order"))
adopted_script = project.script_versions.filter(is_adopted=True).prefetch_related("segments").first()
shots_by_order = {
s.sort_order: s
for s in project.storyboard_shots.filter(adopted_version__isnull=False).select_related("adopted_version__asset")
}
blockers: list[dict] = []
for seg in segs:
# 整批生成只校验还没出片的段;单段重跑则无论状态都校验(用户主动要重出这一段)
@@ -3336,7 +2952,7 @@ def collect_video_review_blockers(project, only_segment: "VideoSegment | None" =
scene = adopted_script.segments.filter(sort_order=seg.sort_order).first() if adopted_script else None
# 人物立绘(与视频实际取图同一套逻辑,保证「拦的」就是「会传给火山的」)
if scene is not None:
for ref in _storyboard_reference_images(project, scene):
for ref in _segment_reference_images(project, scene):
if ref.get("type") != "character":
continue
if (ref.get("review_status") or "") != "active":
@@ -3345,16 +2961,6 @@ def collect_video_review_blockers(project, only_segment: "VideoSegment | None" =
"kind": "person", "name": ref.get("label") or "人物",
"asset_id": str(ref.get("asset_id") or ""), "review_status": ref.get("review_status") or "",
})
# 该镜故事板分镜图(分镜图里也有同一张脸,是另一个需过审的资产)= 对应 shot 的采用版资产
shot = shots_by_order.get(seg.sort_order)
if shot is not None and shot.adopted_version_id:
frame_asset = shot.adopted_version.asset
if (frame_asset.review_status or "") != "active":
blockers.append({
"video_segment_id": str(seg.id), "sort_order": seg.sort_order, "scene_no": scene_no,
"kind": "storyboard", "name": f"{scene_no} 分镜图",
"asset_id": str(frame_asset.id), "review_status": frame_asset.review_status or "",
})
return blockers
@@ -437,10 +437,10 @@ class NarrationLimitTests(SimpleTestCase):
class ShotDensityTests(SimpleTestCase):
_VISUAL = (
"0-3s:近景,女主对镜头抬眼,工位键盘还亮着,窗外是下午的光\n"
"3-8s:手持跟到桌面右手把茶包放进盛了热水的玻璃杯,水面起蒸汽\n"
"8-12s杯壁特写,茶汤从浅金慢慢变深,标签贴在杯沿\n"
"12-15s拉回中近景,女主喝一口,眼神松下来"
"0-3s:近景;平视;固定机位;女主在画面右侧对镜头抬眼,工位键盘还亮着,窗外是下午的光\n"
"3-8s中近景;侧后方过肩;手持跟到桌面右手从画面右侧入画,把茶包放进盛了热水的玻璃杯,水面起蒸汽\n"
"8-12s特写;俯拍杯口;缓慢推近;茶汤从浅金慢慢变深,标签贴在杯沿\n"
"12-15s:中近景;平视;拉回;女主双手捧杯喝一口,眉头松开肩膀塌下来"
)
def test_short_oral_narration_is_rejected(self):
@@ -493,8 +493,9 @@ class ShotDensityTests(SimpleTestCase):
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:
@@ -1,312 +0,0 @@
from decimal import Decimal
from unittest.mock import Mock, patch
import requests
from django.test import TestCase, override_settings
from apps.accounts.models import Team, TeamMember, User
from apps.ai.models import AITask, ModelConfig, ModelProvider
from apps.ai.services import _storyboard_shot_worker, poll_storyboard
from apps.assets.models import Asset
from apps.billing.models import CreditAccount, CreditLedger
from apps.products.models import Product
from apps.projects.models import (
Project,
ScriptSegment,
ScriptVersion,
StoryboardShot,
StoryboardShotVersion,
)
def _metadata(*, outbound=True, base_cost="0.50", max_refs=9):
return {
"routing": {"fallback_on_failure": outbound, "fallback_candidate": True},
"capabilities": {
"operations": ["image_generate", "image_edit"],
"features": [],
"reference_modes": ["none", "single", "multiple"],
"max_reference_images": max_refs,
"aspect_ratios": ["1:1", "3:4", "4:5", "9:16", "16:9"],
},
"pricing": {"base_cost_yuan": base_cost},
}
@override_settings(
STORYBOARD_MAX_PARALLEL=4,
CACHES={"default": {"BACKEND": "django.core.cache.backends.locmem.LocMemCache"}},
)
class StoryboardRoutingTests(TestCase):
def setUp(self):
ModelConfig.objects.filter(capability=ModelConfig.Capability.IMAGE).update(
status=ModelConfig.Status.DISABLED
)
self.user = User.objects.create_user(username="storyboard-routing", password="x")
self.team = Team.objects.create(name="Storyboard Routing", 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="故事板路由项目",
)
script = ScriptVersion.objects.create(
project=self.project,
title="脚本",
content="结构化脚本",
is_adopted=True,
)
self.segment = ScriptSegment.objects.create(
script_version=script,
sort_order=0,
visual_prompt="人物在商品旁展示",
)
self.shot = StoryboardShot.objects.create(
project=self.project,
script_segment=self.segment,
sort_order=0,
status=StoryboardShot.Status.QUEUED,
)
self.refs = [
{"url": "http://example.test/person.png", "label": "女主", "type": "person"},
{"url": "http://example.test/product.png", "label": "商品", "type": "product"},
]
self.provider_mocks = {}
patch("apps.ai.services.get_image_provider", side_effect=self._provider_for).start()
patch(
"apps.ai.services._storyboard_reference_images",
side_effect=lambda project, segment: list(self.refs),
).start()
patch(
"apps.ai.services.build_storyboard_frame_prompt_refs",
return_value="带参考图编号与锁定约束的故事板提示词",
).start()
patch(
"apps.ai.services.build_storyboard_frame_prompt",
return_value="无参考图故事板提示词",
).start()
patch("apps.ai.services._store_generated_media", side_effect=self._store_media).start()
patch("apps.assets.review.submit_asset_for_review").start()
patch("apps.ai.services.notify_generation_failure").start()
self.addCleanup(patch.stopall)
def provider(self, name, priority):
return ModelProvider.objects.create(
name=name,
display_name=name,
status=ModelProvider.Status.ACTIVE,
metadata={"routing": {"fallback_priority": priority}},
)
def model(self, provider, name, *, outbound=True, base_cost="0.50", max_refs=9):
return ModelConfig.objects.create(
provider=provider,
name=name,
display_name=name,
capability=ModelConfig.Capability.IMAGE,
endpoint="images/generations",
unit_price=Decimal("20"),
status=ModelConfig.Status.ACTIVE,
metadata=_metadata(outbound=outbound, base_cost=base_cost, max_refs=max_refs),
)
@staticmethod
def _new_provider_mock():
provider = Mock()
response = {"data": [{"url": "http://example.test/storyboard.png"}]}
provider.image_edit.return_value = response
provider.image_generation.return_value = response
provider.extract_first_media_url.side_effect = lambda value: value["data"][0]["url"]
return provider
@staticmethod
def _new_generation_only_provider_mock():
provider = Mock(spec=["image_generation", "extract_first_media_url"])
provider.image_generation.return_value = {
"data": [{"url": "http://example.test/storyboard.png"}]
}
provider.extract_first_media_url.side_effect = lambda value: value["data"][0]["url"]
return provider
def _provider_for(self, model):
return self.provider_mocks.setdefault(model.id, self._new_provider_mock())
def _store_media(self, **kwargs):
return Asset.objects.create(
team=kwargs["team"],
created_by=kwargs["user"],
name=kwargs["name"],
asset_type=kwargs["asset_type"],
source=Asset.Source.AI_GENERATED,
category=kwargs["category"],
origin_task=kwargs["task"],
)
def enqueue(self, *, run=True):
with patch("threading.Thread"):
result = poll_storyboard(project=self.project, user=self.user)
self.assertEqual(result["status"], "generating")
task = AITask.objects.filter(project=self.project, task_type=AITask.Type.STORYBOARD).latest(
"created_at"
)
if run:
_storyboard_shot_worker(str(task.id), str(self.shot.id), str(self.user.id))
return task
@staticmethod
def ledger_count(task, ledger_type):
return CreditLedger.objects.filter(task=task, ledger_type=ledger_type).count()
def test_multi_reference_success_records_9_16_and_adopts_one_shot_version(self):
primary = self.model(self.provider("storyboard-primary", 20), "gpt-image-2")
task = self.enqueue()
task.refresh_from_db()
self.shot.refresh_from_db()
self.assertEqual(task.status, AITask.Status.SUCCEEDED)
self.assertNotIn("model_routing_v1", task.request_payload)
self.assertEqual(task.model_config_id, primary.id)
self.assertEqual(task.model_config.name, "gpt-image-2")
call = self.provider_mocks[primary.id].image_edit.call_args
self.assertEqual(
call.kwargs["images"],
["http://example.test/person.png", "http://example.test/product.png"],
)
self.assertEqual(call.kwargs["prompt"], "带参考图编号与锁定约束的故事板提示词")
self.assertEqual(call.kwargs["size"], "1024x1536")
self.assertEqual(call.kwargs["model"], "gpt-image-2")
self.assertIsNotNone(self.shot.adopted_version_id)
self.assertEqual(self.shot.adopted_version.task_id, task.id)
self.assertEqual(self.shot.versions.count(), 1)
self.assertEqual(self.ledger_count(task, CreditLedger.Type.RESERVE), 1)
self.assertEqual(self.ledger_count(task, CreditLedger.Type.CHARGE), 1)
def test_wizard_aspect_ratio_drives_storyboard_size(self):
self.project.metadata = {"wizard": {"aspect_ratio": "16:9", "resolution": "480p"}}
self.project.save(update_fields=["metadata"])
primary = self.model(self.provider("storyboard-wide", 20), "gpt-image-2")
task = self.enqueue()
call = self.provider_mocks[primary.id].image_edit.call_args
self.assertEqual(call.kwargs["size"], "1536x864")
def test_gpt_failure_does_not_fallback_to_seedream(self):
primary = self.model(
self.provider("storyboard-gpt", 100),
"gpt-image-2",
base_cost="0.25",
)
seedream = self.model(
self.provider("storyboard-volcano-fallback", 10),
"seedream-4-5-251128",
outbound=False,
base_cost="0.75",
)
self.provider_mocks[primary.id] = self._new_provider_mock()
self.provider_mocks[primary.id].image_edit.side_effect = requests.ConnectionError("offline")
self.provider_mocks[seedream.id] = self._new_provider_mock()
task = self.enqueue()
task.refresh_from_db()
self.assertEqual(task.status, AITask.Status.FAILED)
self.assertEqual(task.model_config_id, primary.id)
self.provider_mocks[seedream.id].image_edit.assert_not_called()
self.provider_mocks[seedream.id].image_generation.assert_not_called()
self.assertEqual(self.ledger_count(task, CreditLedger.Type.RESERVE), 1)
self.assertEqual(self.ledger_count(task, CreditLedger.Type.CHARGE), 0)
self.assertEqual(self.ledger_count(task, CreditLedger.Type.RELEASE), 1)
def test_all_candidates_fail_releases_and_keeps_old_adopted_version(self):
primary = self.model(self.provider("storyboard-all-primary", 100), "gpt-image-2")
self.model(self.provider("storyboard-volcano-keep", 10), "seedream-4-5-251128", outbound=False)
old_asset = Asset.objects.create(
team=self.team,
created_by=self.user,
name="旧故事板",
asset_type=Asset.Type.IMAGE,
source=Asset.Source.AI_GENERATED,
category=Asset.Category.STORYBOARD,
)
old_version = StoryboardShotVersion.objects.create(
shot=self.shot,
asset=old_asset,
prompt="旧画面",
is_adopted=True,
)
self.shot.adopted_version = old_version
self.shot.save(update_fields=["adopted_version", "updated_at"])
self.provider_mocks[primary.id] = self._new_provider_mock()
self.provider_mocks[primary.id].image_edit.side_effect = requests.ConnectionError("offline")
task = self.enqueue()
task.refresh_from_db()
self.shot.refresh_from_db()
self.assertEqual(task.status, AITask.Status.FAILED)
self.assertEqual(self.shot.adopted_version_id, old_version.id)
self.assertEqual(self.shot.versions.count(), 1)
self.assertEqual(self.ledger_count(task, CreditLedger.Type.RESERVE), 1)
self.assertEqual(self.ledger_count(task, CreditLedger.Type.CHARGE), 0)
self.assertEqual(self.ledger_count(task, CreditLedger.Type.RELEASE), 1)
self.assertEqual(CreditAccount.objects.get(team=self.team).reserved_balance, Decimal("0"))
def test_no_reference_uses_image_generation(self):
primary = self.model(self.provider("storyboard-no-ref", 20), "gpt-image-2")
self.refs = []
task = self.enqueue()
task.refresh_from_db()
self.provider_mocks[primary.id].image_edit.assert_not_called()
call = self.provider_mocks[primary.id].image_generation.call_args
self.assertEqual(call.kwargs["prompt"], "无参考图故事板提示词")
self.assertEqual(call.kwargs["model"], "gpt-image-2")
def test_seedream_is_ignored_even_if_it_is_the_default_image_model(self):
seedream = self.model(
self.provider("storyboard-volcano-default", 10), "seedream-4-5-251128", outbound=False
)
gpt = self.model(self.provider("storyboard-gpt-default", 20), "gpt-image-2")
seedream.is_default = True
seedream.save(update_fields=["is_default"])
self.provider_mocks[seedream.id] = self._new_generation_only_provider_mock()
task = self.enqueue()
task.refresh_from_db()
self.assertEqual(task.status, AITask.Status.SUCCEEDED)
self.assertEqual(task.model_config_id, gpt.id)
self.provider_mocks[seedream.id].image_generation.assert_not_called()
self.provider_mocks[gpt.id].image_edit.assert_called_once()
def test_poll_creates_one_independent_task_and_reservation_per_shot(self):
gpt = self.model(self.provider("storyboard-batch", 20), "gpt-image-2")
second_segment = ScriptSegment.objects.create(
script_version=self.segment.script_version,
sort_order=1,
visual_prompt="第二镜",
)
second_shot = StoryboardShot.objects.create(
project=self.project,
script_segment=second_segment,
sort_order=1,
status=StoryboardShot.Status.QUEUED,
)
with patch("threading.Thread"):
result = poll_storyboard(project=self.project, user=self.user)
self.assertEqual(result, {"status": "generating", "done": 0, "total": 2})
tasks = list(
AITask.objects.filter(project=self.project, task_type=AITask.Type.STORYBOARD).order_by(
"created_at"
)
)
self.assertEqual(len(tasks), 2)
self.assertEqual(
{task.request_payload["storyboard_shot"] for task in tasks},
{str(self.shot.id), str(second_shot.id)},
)
self.assertTrue(all("model_routing_v1" not in task.request_payload for task in tasks))
self.assertTrue(all(task.model_config_id == gpt.id for task in tasks))
self.assertTrue(all(self.ledger_count(task, CreditLedger.Type.RESERVE) == 1 for task in tasks))
+4 -3
View File
@@ -29,9 +29,10 @@ def project_stage_label(project):
return {
"script": "Stage 1 · 脚本",
"base_assets": "Stage 2 · 基础资产",
"storyboard": "Stage 3 · 故事板",
"video": "Stage 4 · 视频",
"export": "Stage 5 · 导出",
# storyboard 已下线,只为老项目的历史 current_stage 兜底显示
"storyboard": "Stage 2 · 基础资产",
"video": "Stage 3 · 视频",
"export": "Stage 4 · 导出",
}.get(project.current_stage, "Stage 1 · 脚本")
+3 -91
View File
@@ -15,10 +15,6 @@ from .models import (
ScriptSegment,
ScriptTemplate,
ScriptVersion,
StoryboardFrame,
StoryboardShot,
StoryboardShotVersion,
StoryboardVersion,
SubtitleTrack,
Timeline,
TimelineClip,
@@ -140,84 +136,6 @@ class BaseAssetGroupSerializer(serializers.ModelSerializer):
return {str(asset.id): _asset_preview_url(asset) for asset in obj.candidate_assets.all()}
class StoryboardFrameSerializer(serializers.ModelSerializer):
asset_url = serializers.SerializerMethodField()
# 分镜图含人脸,后端自动送火山人像审核;暴露审核态让前端挂盾牌(与角色/场景卡同款)
review_status = serializers.SerializerMethodField()
review_error = serializers.SerializerMethodField()
class Meta:
model = StoryboardFrame
fields = ["id", "script_segment", "asset", "asset_url", "sort_order", "prompt", "review_status", "review_error"]
read_only_fields = fields
def get_asset_url(self, obj) -> str:
return _asset_preview_url(obj.asset)
def get_review_status(self, obj) -> str:
return obj.asset.review_status if obj.asset_id else ""
def get_review_error(self, obj) -> str:
return (obj.asset.review_error or "") if obj.asset_id else ""
class StoryboardVersionSerializer(serializers.ModelSerializer):
frames = StoryboardFrameSerializer(many=True, read_only=True)
class Meta:
model = StoryboardVersion
fields = ["id", "prompt", "is_adopted", "frames", "created_at", "updated_at"]
read_only_fields = fields
# ── 故事板分镜制(对标 VideoSegment/Version):每镜一个 shot,采用版缩略图 + 各自历史版本 ──
class StoryboardShotSerializer(serializers.ModelSerializer):
adopted_asset = serializers.SerializerMethodField()
adopted_asset_url = serializers.SerializerMethodField()
# 分镜图含人脸,后端自动送火山人像审核;暴露采用版审核态让前端挂盾牌
review_status = serializers.SerializerMethodField()
review_error = serializers.SerializerMethodField()
versions = serializers.SerializerMethodField()
class Meta:
model = StoryboardShot
fields = ["id", "script_segment", "sort_order", "status", "error_message", "prompt",
"adopted_version", "adopted_asset", "adopted_asset_url", "review_status", "review_error", "versions"]
read_only_fields = fields
def get_adopted_asset(self, obj):
v = obj.adopted_version
return str(v.asset_id) if v and v.asset_id else None
def get_adopted_asset_url(self, obj) -> str:
v = obj.adopted_version
return _asset_preview_url(v.asset) if v is not None else ""
def get_review_status(self, obj) -> str:
v = obj.adopted_version
return (getattr(v.asset, "review_status", "") or "") if (v and v.asset_id) else ""
def get_review_error(self, obj) -> str:
v = obj.adopted_version
return (getattr(v.asset, "review_error", "") or "") if (v and v.asset_id) else ""
def get_versions(self, obj):
versions = sorted(obj.versions.all(), key=lambda v: (v.created_at is None, v.created_at), reverse=True)
return [
{
"id": str(v.id),
"asset": str(v.asset_id) if v.asset_id else None,
"asset_url": _asset_preview_url(v.asset) if v.asset_id else "",
"prompt": v.prompt,
"is_adopted": v.is_adopted,
"review_status": (getattr(v.asset, "review_status", "") or "") if v.asset_id else "",
"review_error": (getattr(v.asset, "review_error", "") or "") if v.asset_id else "",
"created_at": v.created_at.isoformat() if v.created_at else "",
}
for v in versions
]
class VideoSegmentVersionSerializer(serializers.ModelSerializer):
class Meta:
model = VideoSegmentVersion
@@ -389,15 +307,13 @@ class QuickCreateJobSerializer(serializers.ModelSerializer):
]
def get_phase_index(self, obj) -> int:
# 前端步:脚本 / 资产 / 故事板 / 视频。当前步未完成,不算 done。
if obj.phase == QuickCreateJob.Phase.PRODUCTION:
message = obj.message or ""
return 3 if "视频" in message else 2
# 前端步:脚本 / 资产 / 视频(故事板已下线)。当前步未完成,不算 done。
return {
QuickCreateJob.Phase.PRODUCT: 0,
QuickCreateJob.Phase.SCRIPT: 0,
QuickCreateJob.Phase.ASSETS: 1,
QuickCreateJob.Phase.COMPLETE: 3,
QuickCreateJob.Phase.PRODUCTION: 2,
QuickCreateJob.Phase.COMPLETE: 2,
}.get(obj.phase, 0)
def get_settings(self, obj) -> dict:
@@ -554,8 +470,6 @@ class ProjectSerializer(serializers.ModelSerializer):
video_segments = VideoSegmentSerializer(many=True, read_only=True)
script_versions = ScriptVersionSerializer(many=True, read_only=True)
base_asset_groups = BaseAssetGroupSerializer(many=True, read_only=True)
storyboard_versions = StoryboardVersionSerializer(many=True, read_only=True) # 过渡期保留(旧整版,新生成不再写)
storyboard_shots = StoryboardShotSerializer(many=True, read_only=True)
timeline = TimelineSerializer(read_only=True)
# 合成成片地址(最新一次成功拼接):视频阶段的「播放成片 / 下载成片」直接用它
final_video_url = serializers.SerializerMethodField()
@@ -578,8 +492,6 @@ class ProjectSerializer(serializers.ModelSerializer):
"stages",
"script_versions",
"base_asset_groups",
"storyboard_versions",
"storyboard_shots",
"video_segments",
"timeline",
"final_video_url",
+19 -12
View File
@@ -5,14 +5,23 @@ from django.utils import timezone
from apps.projects.models import Project, ProjectStage, ScriptVersion, VideoSegment
# 故事板已从流程中去掉:脚本 → 基础资产 → 视频 → 导出。
# ProjectStage.Stage.STORYBOARD 枚举保留(老项目的历史行还引用它),但不再出现在流水线里。
STAGE_ORDER = [
ProjectStage.Stage.SCRIPT,
ProjectStage.Stage.BASE_ASSETS,
ProjectStage.Stage.STORYBOARD,
ProjectStage.Stage.VIDEO,
ProjectStage.Stage.EXPORT,
]
# 老项目可能正停在已下线的 storyboard 阶段;按「基础资产之后」对待,让它们能正常进视频。
LEGACY_STAGE_FALLBACK = {ProjectStage.Stage.STORYBOARD: ProjectStage.Stage.BASE_ASSETS}
def normalize_stage(stage: str) -> str:
"""把已下线的阶段名归一到现役阶段,供索引/比较使用。"""
return LEGACY_STAGE_FALLBACK.get(stage, stage)
@dataclass(frozen=True)
class StageTransition:
@@ -22,24 +31,22 @@ class StageTransition:
reason: str = ""
def can_enter_stage(current_stage: str, target_stage: str, allow_skip_storyboard: bool = True) -> StageTransition:
def can_enter_stage(current_stage: str, target_stage: str) -> StageTransition:
if target_stage not in STAGE_ORDER:
return StageTransition(current_stage, target_stage, False, "unknown target stage")
current_index = STAGE_ORDER.index(current_stage) if current_stage in STAGE_ORDER else -1
current = normalize_stage(current_stage)
current_index = STAGE_ORDER.index(current) if current in STAGE_ORDER else -1
target_index = STAGE_ORDER.index(target_stage)
if target_index <= current_index + 1:
return StageTransition(current_stage, target_stage, True)
if allow_skip_storyboard and current_stage == ProjectStage.Stage.BASE_ASSETS and target_stage == ProjectStage.Stage.VIDEO:
return StageTransition(current_stage, target_stage, True)
return StageTransition(current_stage, target_stage, False, "stage prerequisite is not satisfied")
def initialize_project_pipeline(project: Project, *, placeholder_segments: int = 4) -> None:
"""建立专业创作和极速成片共用的阶段、视频占位数据。幂等,可安全重试。"""
"""建立专业创作和一键成片共用的阶段、视频占位数据。幂等,可安全重试。"""
for stage_name in STAGE_ORDER:
ProjectStage.objects.get_or_create(project=project, stage=stage_name)
for index in range(placeholder_segments):
@@ -94,7 +101,7 @@ def sync_video_segments_to_script(project: Project, script: ScriptVersion) -> No
def adopt_script_version(project: Project, script: ScriptVersion) -> None:
"""采用脚本并推进到资产阶段,供专业创作按钮和极速编排共同调用。"""
"""采用脚本并推进到资产阶段,供专业创作按钮和一键成片编排共同调用。"""
ScriptVersion.objects.filter(project=project).exclude(id=script.id).update(is_adopted=False)
if not script.is_adopted:
script.is_adopted = True
@@ -111,9 +118,9 @@ def adopt_script_version(project: Project, script: ScriptVersion) -> None:
project.save(update_fields=["current_stage", "status", "failure_reason", "updated_at"])
def finish_storyboard_stage(project: Project) -> None:
"""故事板全部成功后推进到视频阶段,并再次校准片段数量。"""
stage, _ = ProjectStage.objects.get_or_create(project=project, stage=ProjectStage.Stage.STORYBOARD)
def finish_base_assets_stage(project: Project) -> None:
"""基础资产齐备后直接推进到视频阶段(故事板已下线),并再次校准片段数量。"""
stage, _ = ProjectStage.objects.get_or_create(project=project, stage=ProjectStage.Stage.BASE_ASSETS)
stage.status = ProjectStage.Status.SUCCEEDED
stage.completed_at = timezone.now()
stage.error_message = ""
@@ -127,7 +134,7 @@ def finish_storyboard_stage(project: Project) -> None:
def finish_video_stage(project: Project) -> bool:
"""全部片段成功即完成专业创作的视频阶段。"""
"""全部片段成功即完成视频阶段。"""
segments = list(project.video_segments.values_list("status", "adopted_version_id"))
if not segments or not all(status == VideoSegment.Status.SUCCEEDED and adopted for status, adopted in segments):
return False
@@ -1,7 +1,10 @@
"""极速成片”自动编排。
"""一键成片”自动编排。
这不是另一套生成实现脚本基础资产三视图故事板视频和合成全部调用专业创作现有服务
这不是另一套生成实现脚本基础资产三视图视频和合成全部调用专业创作现有服务
本模块只负责按状态推进并把子任务 ID 持久化到 QuickCreateJob方便刷新后恢复进度
故事板已从流程中去掉资产齐备后直接提交视频导演信息全部由脚本的秒级分镜
build_video_segment_prompt 直达出片模型
"""
from __future__ import annotations
@@ -27,8 +30,6 @@ from apps.ai.services import (
create_export_job,
generate_base_asset,
generate_person_triview,
poll_storyboard,
submit_storyboard,
submit_video_segment,
video_segment_has_inflight_task,
)
@@ -48,7 +49,7 @@ from apps.projects.models import (
from apps.projects.services.pipeline import (
STAGE_ORDER,
adopt_script_version,
finish_storyboard_stage,
finish_base_assets_stage,
finish_video_stage,
sync_video_segments_to_script,
)
@@ -305,9 +306,7 @@ def _restore_project_after_orchestrator_timeout(job: QuickCreateJob) -> None:
project = job.project
if project.status != Project.Status.FAILED or _videos_have_started(job):
return
shots = list(project.storyboard_shots.all())
storyboard_ready = bool(shots) and all(shot.status == "succeeded" and shot.adopted_version_id for shot in shots)
if project.current_stage != ProjectStage.Stage.VIDEO and not storyboard_ready:
if project.current_stage != ProjectStage.Stage.VIDEO:
return
project.status = Project.Status.VIDEOING
project.failure_reason = ""
@@ -804,7 +803,7 @@ def _all_portraits_have_triview(project: Project) -> bool:
return all(_portrait_has_triview(project, group.adopted_asset_id) for group in portraits)
def _finish_assets_and_enter_storyboard(job: QuickCreateJob) -> int:
def _finish_assets_and_enter_video(job: QuickCreateJob) -> int:
for group in job.project.base_asset_groups.filter(adopted_asset__isnull=False):
if group.kind != BaseAssetGroup.Kind.PRODUCT:
group_meta = dict(group.metadata or {})
@@ -816,14 +815,14 @@ def _finish_assets_and_enter_storyboard(job: QuickCreateJob) -> int:
stage.completed_at = timezone.now()
stage.error_message = ""
stage.save(update_fields=["status", "completed_at", "error_message", "updated_at"])
job.project.current_stage = ProjectStage.Stage.STORYBOARD
job.project.status = Project.Status.STORYBOARDING
job.project.current_stage = ProjectStage.Stage.VIDEO
job.project.status = Project.Status.VIDEOING
job.project.save(update_fields=["current_stage", "status", "updated_at"])
_save_job(
job,
phase=QuickCreateJob.Phase.PRODUCTION,
progress=72,
message="正在生成故事板与镜头画面",
message="素材已就绪,正在准备生成视频",
)
return 1
@@ -874,7 +873,7 @@ def _advance_assets(job: QuickCreateJob) -> int | None:
metadata.pop("triview_task_ids", None)
metadata.pop("triview_ready", None)
_save_job(job, metadata=metadata)
return _finish_assets_and_enter_storyboard(job)
return _finish_assets_and_enter_video(job)
def _reviews_ready(job: QuickCreateJob) -> bool | None:
@@ -896,7 +895,7 @@ def _reviews_ready(job: QuickCreateJob) -> bool | None:
metadata["reviews_skipped"] = True
_save_job(job, metadata=metadata, message="质量检查暂时不可用,继续生成视频")
return True
_save_job(job, metadata=metadata, progress=82, message="故事板已完成,正在进行视频素材质量检查")
_save_job(job, metadata=metadata, progress=82, message="素材已就绪,正在进行视频素材质量检查")
return False
blockers = collect_video_review_blockers(job.project)
if not blockers:
@@ -925,7 +924,7 @@ def _reviews_ready(job: QuickCreateJob) -> bool | None:
asset = Asset.objects.filter(team=job.team, id=item["asset_id"]).first()
if asset is not None:
submit_asset_for_review(asset)
_save_job(job, progress=82, message="故事板已完成,正在进行视频素材质量检查")
_save_job(job, progress=82, message="素材已就绪,正在进行视频素材质量检查")
return False
@@ -963,7 +962,7 @@ def _start_videos(job: QuickCreateJob) -> None:
submit_video_segment(
video_segment=segment,
user=job.created_by or job.project.created_by,
prompt="一键成片自动生成,严格遵循本镜故事板与脚本。",
prompt="一键成片自动生成,严格遵循本镜脚本的秒级分镜",
model_config_id=settings["video_model_config_id"] or None,
aspect_ratio=settings["aspect_ratio"],
resolution=settings["resolution"],
@@ -1075,64 +1074,10 @@ def _complete(job: QuickCreateJob) -> None:
def _advance_production(job: QuickCreateJob) -> int | None:
metadata = dict(job.metadata or {})
if (
metadata.get("storyboard_started")
and job.project.current_stage != ProjectStage.Stage.VIDEO
and not job.project.storyboard_shots.exists()
):
metadata["storyboard_started"] = False
_save_job(job, metadata=metadata, message="正在生成故事板与镜头画面")
if not (job.metadata or {}).get("storyboard_started"):
with transaction.atomic():
job = QuickCreateJob.objects.select_for_update().select_related("project").get(id=job.id)
metadata = dict(job.metadata or {})
if not metadata.get("storyboard_started"):
metadata["storyboard_started"] = True
_save_job(job, metadata=metadata, progress=74, message="正在生成故事板与镜头画面")
should_submit = True
else:
should_submit = False
if should_submit:
try:
submit_storyboard(project=job.project, user=job.created_by or job.project.created_by, prompt="")
stage, _ = ProjectStage.objects.get_or_create(project=job.project, stage=ProjectStage.Stage.STORYBOARD)
stage.status = ProjectStage.Status.RUNNING
stage.save(update_fields=["status", "updated_at"])
except Exception as exc:
metadata = dict(job.metadata or {})
metadata.pop("storyboard_started", None)
metadata["internal_error"] = str(exc)[:2000]
hard = any(
token in str(exc).lower()
for token in ("insufficient credit", "no active", "not configured")
)
retries = int(metadata.get("storyboard_submit_retries") or 0)
if not hard and retries < 3:
metadata["storyboard_submit_retries"] = retries + 1
_save_job(job, metadata=metadata, message="正在继续生成故事板…")
return POLL_DELAY_SECONDS
_save_job(job, metadata=metadata)
raise
"""出片阶段:过审闸 → 提交各镜视频 → 合成导出。
故事板已下线资产齐备后就直接提交视频不再有中间出图环节"""
if job.project.current_stage != ProjectStage.Stage.VIDEO:
result = poll_storyboard(project=job.project, user=job.created_by or job.project.created_by)
if result.get("status") == "failed":
retries = int((job.metadata or {}).get("storyboard_fail_retries") or 0)
if retries >= 2:
fail_quick_create(job, _public_error(str(result.get("error") or "故事板生成失败")))
return None
metadata = dict(job.metadata or {})
metadata["storyboard_fail_retries"] = retries + 1
metadata["storyboard_started"] = False
_save_job(job, metadata=metadata, message="故事板未成功,正在重试…")
return 1
if result.get("status") != "succeeded":
total = max(1, int(result.get("total") or 1))
done = int(result.get("done") or 0)
_save_job(job, progress=min(80, 74 + round(done / total * 6)), message=f"正在生成故事板({done}/{total}")
return POLL_DELAY_SECONDS
finish_storyboard_stage(job.project)
finish_base_assets_stage(job.project)
review_state = _reviews_ready(job)
if review_state is None:
@@ -1142,7 +1087,23 @@ def _advance_production(job: QuickCreateJob) -> int | None:
job.refresh_from_db(fields=["metadata"])
if not (job.metadata or {}).get("video_started"):
_start_videos(job)
try:
_start_videos(job)
except Exception as exc: # noqa: BLE001
# 提交视频失败:硬错(余额不足 / 没有可用模型 / 未配置)直接抛给上层落可恢复终态;
# 其余(网络抖动、素材还没写完)给 3 次机会,别让一次失败把整单打死。
metadata = dict(job.metadata or {})
metadata["internal_error"] = str(exc)[:2000]
hard = any(
token in str(exc).lower()
for token in ("insufficient credit", "no active", "not configured")
)
retries = int(metadata.get("video_submit_retries") or 0)
if hard or retries >= 3:
_save_job(job, metadata=metadata)
raise
metadata["video_submit_retries"] = retries + 1
_save_job(job, metadata=metadata, message="正在继续申请生成视频…")
return POLL_DELAY_SECONDS
segments = list(job.project.video_segments.order_by("sort_order"))
@@ -1480,7 +1441,7 @@ def _advance_without_quick_queue(job_id: str) -> None:
def resume_quick_create(job: QuickCreateJob) -> QuickCreateJob:
"""从失败处接着跑:已完成的脚本/资产/故事板保留,只补没做完的步骤。"""
"""从失败处接着跑:已完成的脚本 / 资产 / 视频保留,只补没做完的步骤。"""
job.refresh_from_db()
if job.status == QuickCreateJob.Status.SUCCEEDED:
return job
@@ -1498,12 +1459,11 @@ def resume_quick_create(job: QuickCreateJob) -> QuickCreateJob:
metadata.pop("review_poll_retries", None)
metadata.pop("next_advance_at", None)
metadata.pop("video_fail_retries", None)
metadata.pop("video_submit_retries", None)
metadata.pop("asset_fail_retries", None)
metadata.pop("asset_fail_retries_by_key", None)
metadata.pop("storyboard_fail_retries", None)
metadata.pop("triview_fail_retries", None)
metadata.pop("triview_fail_retries_by_portrait", None)
metadata.pop("storyboard_submit_retries", None)
if metadata.pop("triview_skipped", None):
metadata.pop("triview_task_ids", None)
metadata.pop("triview_ready", None)
@@ -1633,7 +1593,7 @@ def recover_quick_create(job: QuickCreateJob) -> None:
def advance_quick_create(job_id: str) -> int | None:
"""推进一个状态并返回下次轮询秒数;返回 None 表示终态。
一次只做一段识别商品 脚本(独立任务) 资产 故事板/视频/合成
一次只做一段识别商品 脚本(独立任务) 资产 视频/合成
脚本不再在编排任务里同步吃完整条 SSE避免页面一直停在推荐脚本方向
"""
job = QuickCreateJob.objects.select_related("project__product", "created_by", "team").get(id=job_id)
+17 -14
View File
@@ -94,7 +94,7 @@ class QuickCreateApiTests(TestCase):
self.assertEqual(project.metadata["wizard"]["persona"], "bestie")
self.assertEqual(project.metadata["wizard"]["resolution"], "720p")
self.assertEqual(project.metadata["wizard"]["video_model_config_id"], str(video_model_id))
self.assertEqual(project.stages.count(), 5)
self.assertEqual(project.stages.count(), 4)
self.assertEqual(project.video_segments.count(), 2)
self.assertTrue(product.selling_points.exists())
job = QuickCreateJob.objects.get(project=project)
@@ -794,7 +794,7 @@ class QuickCreateCoordinatorTests(TestCase):
self.project.save(update_fields=["status", "current_stage", "updated_at"])
self.job.status = QuickCreateJob.Status.RUNNING
self.job.phase = QuickCreateJob.Phase.PRODUCTION
self.job.metadata = {"storyboard_started": True}
self.job.metadata = {}
self.job.save(update_fields=["status", "phase", "metadata", "updated_at"])
fail_quick_create(self.job, "一键成片暂未完成,请稍后重试或进入专业模式查看", internal_error="Timeout reading from socket")
@@ -825,7 +825,7 @@ class QuickCreateCoordinatorTests(TestCase):
self.project.save(update_fields=["status", "current_stage", "failure_reason", "updated_at"])
self.job.status = QuickCreateJob.Status.FAILED
self.job.phase = QuickCreateJob.Phase.PRODUCTION
self.job.metadata = {"storyboard_started": True}
self.job.metadata = {}
self.job.save(update_fields=["status", "phase", "metadata", "updated_at"])
recover_quick_create(self.job)
@@ -841,7 +841,7 @@ class QuickCreateCoordinatorTests(TestCase):
self.job.status = QuickCreateJob.Status.FAILED
self.job.phase = QuickCreateJob.Phase.PRODUCTION
self.job.error_message = "脚本已保留,后续步骤遇到网络波动。点重试会从上次进度继续"
self.job.metadata = {"storyboard_started": True, "internal_error": "Timeout reading from socket"}
self.job.metadata = {"internal_error": "Timeout reading from socket"}
self.job.save(update_fields=["status", "phase", "error_message", "metadata", "updated_at"])
recover_quick_create(self.job)
@@ -850,12 +850,14 @@ class QuickCreateCoordinatorTests(TestCase):
self.assertEqual(self.job.metadata.get("transient_retries"), 1)
enqueue.assert_called_once()
# 出片阶段的提交本身在别处测;这里只验「从失败处继续」的编排语义,故把提交打桩掉
@patch("apps.projects.services.quick_create._start_videos")
@patch("apps.projects.tasks.advance_quick_create_task.apply_async")
def test_resume_failed_production_job_keeps_progress(self, enqueue):
def test_resume_failed_production_job_keeps_progress(self, enqueue, _start_videos):
self.job.status = QuickCreateJob.Status.FAILED
self.job.phase = QuickCreateJob.Phase.PRODUCTION
self.job.error_message = "一键成片暂未完成,请稍后重试或进入专业模式查看"
self.job.metadata = {"storyboard_started": True, "transient_retries": 8, "video_fail_retries": 2}
self.job.metadata = {"transient_retries": 8, "video_fail_retries": 2}
self.job.save(update_fields=["status", "phase", "error_message", "metadata", "updated_at"])
resume_quick_create(self.job)
@@ -1075,7 +1077,7 @@ class QuickCreateCoordinatorTests(TestCase):
self.assertEqual(self.job.phase, QuickCreateJob.Phase.PRODUCTION)
self.assertIsNone(group.task_id)
def test_failed_triview_does_not_block_storyboard(self):
def test_failed_triview_does_not_block_video(self):
base_ids = self._ready_asset_tasks()
_group, portrait = self._portrait_group()
failed = self._script_task(AITask.Status.FAILED, key="triview-failed", error_message="image_edit timeout")
@@ -1188,7 +1190,7 @@ class QuickCreateCoordinatorTests(TestCase):
self.assertEqual(self.job.phase, QuickCreateJob.Phase.PRODUCTION)
@patch("apps.projects.tasks.advance_quick_create_task.apply_async")
def test_resume_failed_assets_continues_into_storyboard(self, enqueue):
def test_resume_failed_assets_continues_into_video(self, enqueue):
base_ids = self._ready_asset_tasks()
self.job.status = QuickCreateJob.Status.FAILED
self.job.phase = QuickCreateJob.Phase.ASSETS
@@ -1227,7 +1229,7 @@ class QuickCreateCoordinatorTests(TestCase):
segment.save(update_fields=["adopted_version", "status", "updated_at"])
self.job.status = QuickCreateJob.Status.RUNNING
self.job.phase = QuickCreateJob.Phase.PRODUCTION
self.job.metadata = {"storyboard_started": True, "video_started": True}
self.job.metadata = {"video_started": True}
self.job.save(update_fields=["status", "phase", "metadata", "updated_at"])
self.project.current_stage = ProjectStage.Stage.VIDEO
self.project.save(update_fields=["current_stage", "updated_at"])
@@ -1266,19 +1268,20 @@ class QuickCreateCoordinatorTests(TestCase):
self.job.save(update_fields=["status", "phase", "progress", "updated_at"])
data = QuickCreateJobSerializer(self.job).data
self.assertEqual(data["phase_index"], 3)
self.assertEqual(data["phase_index"], 2)
self.assertEqual(data["result"]["video_url"], "https://cdn.example/quick.mp4")
self.assertEqual(data["result"]["final_video_url"], "")
self.assertEqual(data["result"]["duration_seconds"], 15)
def test_phase_index_matches_four_step_ui(self):
def test_phase_index_matches_three_step_ui(self):
"""前端进度条三步:脚本 / 资产 / 视频(故事板已下线)。"""
cases = [
(QuickCreateJob.Phase.PRODUCT, "正在识别商品", 0),
(QuickCreateJob.Phase.SCRIPT, "正在生成分镜脚本…", 0),
(QuickCreateJob.Phase.ASSETS, "正在生成商品、模特与场景资产", 1),
(QuickCreateJob.Phase.PRODUCTION, "正在生成故事板与镜头画面", 2),
(QuickCreateJob.Phase.PRODUCTION, "正在生成视频(1/2", 3),
(QuickCreateJob.Phase.COMPLETE, "视频已生成", 3),
(QuickCreateJob.Phase.PRODUCTION, "素材已就绪,正在准备生成视频", 2),
(QuickCreateJob.Phase.PRODUCTION, "正在生成视频(1/2", 2),
(QuickCreateJob.Phase.COMPLETE, "视频已生成", 2),
]
for phase, message, expected in cases:
self.job.phase = phase
@@ -202,7 +202,7 @@ class ScriptTemplateTests(TestCase):
self.assertEqual(wizard["total_duration"], 30)
self.assertIn("第 4 镜 · 5s · CTA", wizard["template_outline"])
# 新项目仍走完整流水线初始化,不是「复制旧项目」
self.assertEqual(project.stages.count(), 5)
self.assertEqual(project.stages.count(), 4)
self.assertEqual(project.script_versions.count(), 0)
self.assertEqual(ScriptTemplate.objects.get(id=template_id).usage_count, 1)
+27 -80
View File
@@ -14,10 +14,6 @@ from apps.projects.models import (
ProjectStage,
ScriptSegment,
ScriptVersion,
StoryboardFrame,
StoryboardShot,
StoryboardShotVersion,
StoryboardVersion,
SubtitleTrack,
Timeline,
TimelineClip,
@@ -67,14 +63,13 @@ class ProjectApiTests(TestCase):
project = Project.objects.get(id=response.data["id"])
self.assertEqual(project.team, self.team)
self.assertEqual(project.created_by, self.user)
self.assertEqual(project.stages.count(), 5)
self.assertEqual(project.stages.count(), 4)
self.assertEqual(project.video_segments.count(), 4)
self.assertEqual(
list(project.stages.values_list("stage", flat=True)),
[
ProjectStage.Stage.SCRIPT,
ProjectStage.Stage.BASE_ASSETS,
ProjectStage.Stage.STORYBOARD,
ProjectStage.Stage.VIDEO,
ProjectStage.Stage.EXPORT,
],
@@ -574,76 +569,29 @@ class ProjectApiTests(TestCase):
capability=ModelConfig.Capability.IMAGE, endpoint="images/generations", unit_price="1.0000",
)
def _mk_shot(self, project, script, order, *, review_status="active"):
"""建一个已出片的 StoryboardShot(+ 已采用版本,资产审核态可控),供过审/采用测试。"""
def _mk_person(self, project, script, order, *, review_status="active"):
"""给某镜挂一个「已采用的角色立绘」(审核态可控),供过审测试。
故事板下线后,含真人脸的待审资产只剩人物立绘"""
from apps.projects.models import BaseAssetGroup
seg = script.segments.filter(sort_order=order).first() or ScriptSegment.objects.create(
script_version=script, sort_order=order, narration=f"n{order}", visual_prompt=f"v{order}")
a = Asset.objects.create(team=self.team, created_by=self.user, name=f"sb{order}", asset_type="image",
source="ai_generated", category=Asset.Category.STORYBOARD, review_status=review_status)
shot = StoryboardShot.objects.create(project=project, script_segment=seg, sort_order=order,
status=StoryboardShot.Status.SUCCEEDED)
ver = StoryboardShotVersion.objects.create(shot=shot, asset=a, is_adopted=True)
shot.adopted_version = ver
shot.save(update_fields=["adopted_version"])
return shot, ver, a
seg.entity_refs = ["c1"]
seg.save(update_fields=["entity_refs"])
meta = dict(project.metadata or {})
meta["script_entities"] = [{"id": "c1", "type": "character", "name": "女主"}]
project.metadata = meta
project.save(update_fields=["metadata"])
asset = Asset.objects.create(team=self.team, created_by=self.user, name=f"person{order}", asset_type="image",
source="ai_generated", category=Asset.Category.PERSON, review_status=review_status)
AssetFile.objects.create(asset=asset, object_key=f"p{order}.png", bucket="b",
content_type="image/png", preview_url=f"http://x/p{order}.png", is_primary=True)
BaseAssetGroup.objects.create(project=project, kind=BaseAssetGroup.Kind.PERSON,
adopted_asset=asset, metadata={"label": "女主", "adopt": "adopted"})
return asset
def test_submit_storyboard_creates_one_shot_per_segment_all_queued(self):
"""开始生成故事板:确保每镜一个 shot、全部置 QUEUED(真正出图交给 poll)。"""
from apps.ai.services import submit_storyboard
self._mk_image_model()
project = Project.objects.create(team=self.team, created_by=self.user, product=self.product, name="P")
script = ScriptVersion.objects.create(project=project, is_adopted=True)
for i in range(3):
ScriptSegment.objects.create(script_version=script, sort_order=i, narration=f"n{i}", visual_prompt=f"v{i}")
targets = submit_storyboard(project=project, user=self.user, prompt="风格")
self.assertEqual(project.storyboard_shots.count(), 3)
self.assertEqual(len(targets), 3)
self.assertTrue(all(s.status == StoryboardShot.Status.QUEUED for s in project.storyboard_shots.all()))
project.refresh_from_db()
self.assertEqual((project.metadata or {}).get("storyboard_prompt"), "风格") # 整张风格存项目级
def test_submit_storyboard_single_shot_only_targets_that_shot(self):
"""单场重跑:只把指定 shot 置 QUEUED,其余已出片的场不动。"""
from apps.ai.services import submit_storyboard
self._mk_image_model()
project = Project.objects.create(team=self.team, created_by=self.user, product=self.product, name="P")
script = ScriptVersion.objects.create(project=project, is_adopted=True)
for i in range(2):
ScriptSegment.objects.create(script_version=script, sort_order=i, narration=f"n{i}", visual_prompt=f"v{i}")
shot0, _, _ = self._mk_shot(project, script, 0)
shot1, _, _ = self._mk_shot(project, script, 1)
submit_storyboard(project=project, user=self.user, shot_ids=[str(shot1.id)])
shot0.refresh_from_db(); shot1.refresh_from_db()
self.assertEqual(shot0.status, StoryboardShot.Status.SUCCEEDED) # 没动
self.assertEqual(shot1.status, StoryboardShot.Status.QUEUED) # 只重跑这场
def test_adopt_storyboard_shot_version_switches_adopted(self):
"""采用某场的历史版本:切换该场采用的分镜图。"""
self._mk_image_model()
project = Project.objects.create(team=self.team, created_by=self.user, product=self.product, name="P")
script = ScriptVersion.objects.create(project=project, is_adopted=True)
ScriptSegment.objects.create(script_version=script, sort_order=0, narration="n", visual_prompt="v")
shot, v1, _ = self._mk_shot(project, script, 0)
a2 = Asset.objects.create(team=self.team, created_by=self.user, name="sb-v2", asset_type="image",
source="ai_generated", category=Asset.Category.STORYBOARD)
v2 = StoryboardShotVersion.objects.create(shot=shot, asset=a2, is_adopted=False)
res = self.client.post(
f"/api/projects/{project.id}/adopt-storyboard-shot-version/",
{"shot_id": str(shot.id), "version_id": str(v2.id)}, format="json",
)
self.assertEqual(res.status_code, 200)
shot.refresh_from_db(); v1.refresh_from_db(); v2.refresh_from_db()
self.assertEqual(shot.adopted_version_id, v2.id)
self.assertTrue(v2.is_adopted)
self.assertFalse(v1.is_adopted)
def test_video_review_precheck_flags_unreviewed_storyboard_then_clears(self):
"""生成视频前的过审闸:某镜故事板分镜未过审(非 active)→ precheck 列为 blocker;
def test_video_review_precheck_flags_unreviewed_person_then_clears(self):
"""生成视频前的过审闸:某镜的人物立绘未过审(非 active)→ precheck 列为 blocker;
过审(active) blockers 清空,可放行生成"""
project = Project.objects.create(
team=self.team, created_by=self.user, product=self.product, name="P",
@@ -652,21 +600,21 @@ class ProjectApiTests(TestCase):
VideoSegment.objects.create(project=project, sort_order=0, target_duration_seconds=15)
script = ScriptVersion.objects.create(project=project, is_adopted=True)
ScriptSegment.objects.create(script_version=script, sort_order=0, narration="n", visual_prompt="v")
_, _, frame_asset = self._mk_shot(project, script, 0, review_status="")
person_asset = self._mk_person(project, script, 0, review_status="")
res = self.client.post(f"/api/projects/{project.id}/video-review-precheck/", format="json")
self.assertEqual(res.status_code, 200)
blockers = res.json()["blockers"]
self.assertTrue(any(b["kind"] == "storyboard" and b["scene_no"] == 1 for b in blockers))
self.assertTrue(any(b["kind"] == "person" and b["scene_no"] == 1 for b in blockers))
# 过审后再查 → 不再拦
frame_asset.review_status = "active"
frame_asset.save(update_fields=["review_status"])
person_asset.review_status = "active"
person_asset.save(update_fields=["review_status"])
res2 = self.client.post(f"/api/projects/{project.id}/video-review-precheck/", format="json")
self.assertEqual(res2.json()["blockers"], [])
def test_video_review_precheck_skips_succeeded_segments_in_batch(self):
"""整批校验(不传 segment_id)只看未出片的段:已 succeeded 的段即便分镜未过审也不拦。"""
"""整批校验(不传 segment_id)只看未出片的段:已 succeeded 的段即便人物未过审也不拦。"""
project = Project.objects.create(
team=self.team, created_by=self.user, product=self.product, name="P",
current_stage=ProjectStage.Stage.VIDEO,
@@ -674,7 +622,7 @@ class ProjectApiTests(TestCase):
seg = VideoSegment.objects.create(project=project, sort_order=0, status=VideoSegment.Status.SUCCEEDED)
script = ScriptVersion.objects.create(project=project, is_adopted=True)
ScriptSegment.objects.create(script_version=script, sort_order=0, narration="n", visual_prompt="v")
self._mk_shot(project, script, 0, review_status="")
self._mk_person(project, script, 0, review_status="")
# 整批:succeeded 段跳过 → 不拦
res = self.client.post(f"/api/projects/{project.id}/video-review-precheck/", format="json")
@@ -936,7 +884,6 @@ class WorkerGateTests(TestCase):
cases = [
(f"/api/projects/{self.project.id}/generate-base-asset/", {"kind": "person"}),
(f"/api/projects/{self.project.id}/generate-triview/", {"portrait_asset_id": "x"}),
(f"/api/projects/{self.project.id}/generate-storyboard/", {}),
(f"/api/projects/{self.project.id}/submit-video-segment/", {"video_segment_id": "x"}),
("/api/ai/generate-image/", {"prompt": "测试"}),
]
+9 -114
View File
@@ -27,17 +27,14 @@ from apps.ai.script_agent import (
from apps.ai.services import (
DEFAULT_VOICEOVER_VOICE,
VOICEOVER_VOICES,
adopt_storyboard_shot_version,
create_export_job,
generate_base_asset,
generate_person_triview,
get_default_model,
get_inflight_extraction,
poll_storyboard,
poll_video_segment,
regenerate_script_segment,
submit_extract_entities,
submit_storyboard,
submit_video_segment,
synthesize_project_voiceover,
)
@@ -60,10 +57,6 @@ from .models import (
ScriptSegment,
ScriptTemplate,
ScriptVersion,
StoryboardFrame,
StoryboardShot,
StoryboardShotVersion,
StoryboardVersion,
SubtitleTrack,
Timeline,
TimelineClip,
@@ -78,14 +71,12 @@ from .serializers import (
QuickCreateJobSerializer,
ScriptTemplateSerializer,
ScriptVersionSerializer,
StoryboardVersionSerializer,
VideoSegmentVersionSerializer,
is_playable_video,
)
from .services.export import run_export_job_in_thread
from .services.pipeline import (
adopt_script_version,
finish_storyboard_stage,
initialize_project_pipeline,
sync_video_segments_to_script,
)
@@ -182,8 +173,8 @@ def promote_base_asset_stage_if_ready(project: Project) -> bool:
stage, _ = ProjectStage.objects.get_or_create(project=project, stage=ProjectStage.Stage.BASE_ASSETS)
stage.status = ProjectStage.Status.SUCCEEDED
stage.save(update_fields=["status", "updated_at"])
project.current_stage = ProjectStage.Stage.STORYBOARD
project.status = Project.Status.STORYBOARDING
project.current_stage = ProjectStage.Stage.VIDEO
project.status = Project.Status.VIDEOING
project.save(update_fields=["current_stage", "status", "updated_at"])
return True
@@ -197,7 +188,7 @@ VIDEO_IS_FINAL_STAGE = True
def _ensure_video_stage(project: Project) -> None:
"""幂等地把项目推进到「视频」阶段(生成视频片段时调用)。current_stage 没到 VIDEO 时,
settle_video_completion 会直接返回 全段出片也不收口卡在故事板(ZWQ#15)。"""
settle_video_completion 会直接返回 全段出片也不收口卡在上一(ZWQ#15)。"""
if project.current_stage != ProjectStage.Stage.VIDEO:
project.current_stage = ProjectStage.Stage.VIDEO
if project.status != Project.Status.COMPLETED:
@@ -252,10 +243,6 @@ class ProjectViewSet(TeamScopedViewSetMixin, ModelViewSet):
"base_asset_groups",
"base_asset_groups__adopted_asset__files",
"base_asset_groups__candidate_assets__files",
"storyboard_versions",
"storyboard_versions__frames__asset__files",
"storyboard_shots__adopted_version__asset__files",
"storyboard_shots__versions__asset__files",
"timeline__clips__asset__files",
).all()
serializer_class = ProjectSerializer
@@ -304,7 +291,7 @@ class ProjectViewSet(TeamScopedViewSetMixin, ModelViewSet):
def retrieve(self, request, *args, **kwargs):
# 详情加载(=进入流水线页)时自愈视频片段数:历史项目在「采用前增删分镜」未同步,
# 或项目创建固定铺的 4 段从未被收口,会让视频步骤的片段数与故事板/采用版分镜对不上。
# 或项目创建固定铺的 4 段从未被收口,会让视频步骤的片段数与采用版分镜对不上。
# 按采用版分镜数收口一次(幂等;只裁从未生成过的尾段,生成中/已出片的段绝不动)。
# 用轻量查询判定 + 同步,再交给 super 做重 prefetch 序列化(避免重查询跑两遍)。
proj = (
@@ -326,7 +313,7 @@ class ProjectViewSet(TeamScopedViewSetMixin, ModelViewSet):
if adopted_script is not None:
self._sync_video_segments_to_script(proj, adopted_script)
# 自愈卡阶段:历史项目「视频片段全部出片并采用」却因 current_stage 没推到 VIDEO 而永远停在
# 「3/4 故事板」、不收口(ZWQ#15)。进详情时若检测到全段已出片,就补推到视频阶段再收口。
# 停在上一格、不收口(ZWQ#15)。进详情时若检测到全段已出片,就补推到视频阶段再收口。
segs = list(VideoSegment.objects.filter(project=proj).values_list("status", "adopted_version_id"))
if segs and all(s == VideoSegment.Status.SUCCEEDED and a is not None for s, a in segs):
_ensure_video_stage(proj)
@@ -343,7 +330,7 @@ class ProjectViewSet(TeamScopedViewSetMixin, ModelViewSet):
.order_by("-updated_at")
)
# 列表:轻查询(只 select_related product + 计数注解),不做详情那串 12 个重 prefetch
# ——原列表把每个项目的 阶段/片段/故事板/时间线/资产文件全拉出,20 个项目实测 ~2s。
# ——原列表把每个项目的 阶段/片段/时间线/资产文件全拉出,20 个项目实测 ~2s。
if self.action == "list":
qs = (
Project.objects.select_related("product", "product__cover_asset", "timeline", "quick_create_job")
@@ -725,7 +712,6 @@ class ProjectViewSet(TeamScopedViewSetMixin, ModelViewSet):
"project__product__cover_asset__files",
"project__script_versions",
"project__base_asset_groups",
"project__storyboard_shots__adopted_version__asset__files",
"project__video_segments__adopted_version__asset__files",
"project__timeline__export_jobs__output_asset__files",
)
@@ -1256,21 +1242,6 @@ class ProjectViewSet(TeamScopedViewSetMixin, ModelViewSet):
已出片的段不改时长改了会跟已渲染的成片对不上"""
sync_video_segments_to_script(project, script)
def _sync_storyboard_shots_to_script(self, project: Project, script: ScriptVersion) -> None:
"""采用版分镜数变化时,同步 StoryboardShot 数量(与视频段同策略,按位置对齐):
多出且尾部从未出过图 shot 裁掉(已出图的不动);不主动建出图时 ensure_storyboard_shots 按需补"""
if not script.is_adopted:
return
target = script.segments.count()
shots = list(project.storyboard_shots.order_by("sort_order"))
while len(shots) > target:
tail = shots[-1]
if tail.status == StoryboardShot.Status.NOT_STARTED and not tail.versions.exists():
tail.delete()
shots.pop()
else:
break
@action(detail=True, methods=["post"], url_path="update-script-segment")
def update_script_segment(self, request, pk=None):
project = self.get_object()
@@ -1341,7 +1312,6 @@ class ProjectViewSet(TeamScopedViewSetMixin, ModelViewSet):
seg.sort_order = index
seg.save(update_fields=["sort_order", "updated_at"])
self._sync_video_segments_to_script(project, script)
self._sync_storyboard_shots_to_script(project, script)
return Response(ScriptVersionSerializer(script).data, status=status.HTTP_201_CREATED)
@action(detail=True, methods=["post"], url_path="delete-script-segment")
@@ -1353,15 +1323,6 @@ class ProjectViewSet(TeamScopedViewSetMixin, ModelViewSet):
)
script = segment.script_version
# 允许删空:用户可以把镜头全删了(再整体重新生成或手动加),不再强制「至少保留一个分镜」。
# 删本场脚本时,它对应的故事板分镜帧也一起删 —— frame.script_segment 是 SET_NULL,
# 不主动删会留下孤儿帧,故事板里仍显示这一(已删的)场。各版本受影响的帧统一收口后重排 sort_order。
affected_storyboards = list(StoryboardVersion.objects.filter(frames__script_segment=segment).distinct())
StoryboardFrame.objects.filter(script_segment=segment).delete()
for sb in affected_storyboards:
for index, frame in enumerate(sb.frames.order_by("sort_order")):
if frame.sort_order != index:
frame.sort_order = index
frame.save(update_fields=["sort_order", "updated_at"])
segment.delete()
# 删后重排 sort_order:收集变动的镜一次性 bulk_update,替代逐条 save(远程库少跑 N 个往返 → 删除快很多)
resort = []
@@ -1372,7 +1333,6 @@ class ProjectViewSet(TeamScopedViewSetMixin, ModelViewSet):
if resort:
ScriptSegment.objects.bulk_update(resort, ["sort_order"])
self._sync_video_segments_to_script(project, script)
self._sync_storyboard_shots_to_script(project, script)
return Response(ScriptVersionSerializer(script).data)
# ── Stage 4 · 视频版本采用(详情弹窗里切历史版) ──
@@ -1402,74 +1362,9 @@ class ProjectViewSet(TeamScopedViewSetMixin, ModelViewSet):
settle_video_completion(project)
return Response(ProjectSerializer(project).data)
@action(detail=True, methods=["post"], url_path="generate-storyboard")
def generate_storyboard_action(self, request, pk=None):
"""开始生成故事板(全部场):确保每镜一个 shot 并标 QUEUED;逐场出图交给 poll-storyboard 起线程。
全部已出片时再点 = 整批重跑(全部场重出)不推进阶段"""
require_worker()
project = self.get_object()
submit_storyboard(project=project, user=request.user, prompt=request.data.get("prompt", ""))
stage, _ = ProjectStage.objects.get_or_create(project=project, stage=ProjectStage.Stage.STORYBOARD)
stage.status = ProjectStage.Status.RUNNING
stage.save(update_fields=["status", "updated_at"])
return Response({"status": "queued"}, status=status.HTTP_202_ACCEPTED)
@action(detail=True, methods=["post"], url_path="rerun-storyboard-shot")
def rerun_storyboard_shot_action(self, request, pk=None):
"""单场重跑:只重出该 shot 的图 —— 新增一条历史版本并采用,其余场不动。出图交给 poll-storyboard。"""
require_worker()
project = self.get_object()
shot_id = request.data.get("shot_id")
if not project.storyboard_shots.filter(id=shot_id).exists():
return Response({"detail": "shot not found"}, status=status.HTTP_404_NOT_FOUND)
submit_storyboard(project=project, user=request.user, prompt=request.data.get("prompt", ""), shot_ids=[shot_id])
stage, _ = ProjectStage.objects.get_or_create(project=project, stage=ProjectStage.Stage.STORYBOARD)
stage.status = ProjectStage.Status.RUNNING
stage.save(update_fields=["status", "updated_at"])
return Response({"status": "queued"}, status=status.HTTP_202_ACCEPTED)
@action(detail=True, methods=["post"], url_path="adopt-storyboard-shot-version")
def adopt_storyboard_shot_version_action(self, request, pk=None):
"""采用某场的某个历史版本(对标 adopt-video-version):切换该场采用的分镜图。"""
project = self.get_object()
shot = StoryboardShot.objects.filter(project=project, id=request.data.get("shot_id")).first()
if shot is None:
return Response({"detail": "shot not found"}, status=status.HTTP_404_NOT_FOUND)
version = StoryboardShotVersion.objects.filter(shot=shot, id=request.data.get("version_id")).first()
if version is None:
return Response({"detail": "version not found"}, status=status.HTTP_404_NOT_FOUND)
adopt_storyboard_shot_version(shot=shot, version=version)
return Response(ProjectSerializer(self.get_object()).data)
@action(detail=True, methods=["post"], url_path="poll-storyboard")
def poll_storyboard_action(self, request, pk=None):
"""异步故事板·轮询:为在制场起线程出图(单次 ARK 调用 ~20s)。全部完成 → 推进到 VIDEO 阶段。"""
project = self.get_object()
result = poll_storyboard(project=project, user=request.user)
if result.get("status") == "succeeded":
finish_storyboard_stage(project)
http_status = status.HTTP_200_OK if result.get("status") == "succeeded" else status.HTTP_202_ACCEPTED
return Response(result, status=http_status)
@action(detail=True, methods=["post"], url_path="skip-storyboard")
@transaction.atomic
def skip_storyboard(self, request, pk=None):
project = self.get_object()
stage, _ = ProjectStage.objects.get_or_create(project=project, stage=ProjectStage.Stage.STORYBOARD)
stage.status = ProjectStage.Status.SKIPPED
stage.save(update_fields=["status", "updated_at"])
# 跳过故事板直接进视频时,同样把视频片段数对齐到采用版分镜数(已生成的段不动)
adopted_script = project.script_versions.filter(is_adopted=True).order_by("-created_at").first()
if adopted_script is not None:
self._sync_video_segments_to_script(project, adopted_script)
project.current_stage = ProjectStage.Stage.VIDEO
project.status = Project.Status.VIDEOING
project.save(update_fields=["current_stage", "status", "updated_at"])
return Response(ProjectSerializer(project).data)
@action(detail=True, methods=["post"], url_path="video-review-precheck")
def video_review_precheck(self, request, pk=None):
"""点「生成视频」前的过审校验:返回未过审(非 active)的人物立绘 / 故事板分镜清单。
"""点「生成视频」前的过审校验:返回未过审(非 active)的人物立绘清单。
body 可选 video_segment_id:只校验该段(单段重跑);不传 = 校验全部未出片段
blockers = 全部已过审,前端放行生成"""
from apps.ai.services import collect_video_review_blockers
@@ -1511,9 +1406,9 @@ class ProjectViewSet(TeamScopedViewSetMixin, ModelViewSet):
poll_video_segment_task.apply_async(args=[str(segment.id)], countdown=30)
except Exception: # noqa: BLE001
logger.error("poll_video_segment_task enqueue failed; relying on client polling", exc_info=True)
# 生成视频即代表已进入视频阶段:某些路径(跳过故事板/异常)可能没把 current_stage 推到 VIDEO,
# 生成视频即代表已进入视频阶段:某些路径(异常/老数据)可能没把 current_stage 推到 VIDEO,
# 而 settle_video_completion 只在 current_stage==VIDEO 时收口 → 不补推的话,全段出片后项目会永远停在
# 「3/4 故事板」、永不 completed(ZWQ#15)。这里幂等补推。
# 停在上一格、永不 completed(ZWQ#15)。这里幂等补推。
_ensure_video_stage(project)
# 重跑某段 → 该段回到在制;若项目此前已 completed,退回 videoing
settle_video_completion(project)
@@ -15,7 +15,7 @@ description: >
你是一个**实体提取 agent**。输入是一份**已经定稿的分镜脚本** + **商品信息**
你的任务:从脚本里认出**角色(出镜的人)**和**场景(画面发生的地点 / 环境)**,
为每个角色 / 场景写好生图描述,并标出**每个分镜镜头里出现了哪些角色 / 场景**。
你的产出会直接进入 AirShelf 流水线的下游(生成角色立绘 / 场景图 → 故事板 → Seedance 生视频),
你的产出会直接进入 AirShelf 流水线的下游(生成角色立绘 / 场景图 → 连同商品参考图一起交给 Seedance 直接出片),
因此你**只输出一个结构化 JSON 对象,绝不输出散文、解释或 markdown 代码块外的任何文字**。
> **模型无关声明**:本技能不依赖任何特定模型的能力或语气。无论运行在豆包 / GPT / Gemini /
@@ -38,7 +38,7 @@
**为什么这么写(避坑点评)**
- ❌ 没有任何 `product` 实体——包是商品,系统会用真实主图,不在这里出。
- ❌ 女主、闺蜜身上、手上**都没有包**,连镜3「背着包出门」也**不在角色描述里写包**:角色立绘是「干净的人」,包到了故事板 / 视频阶段才靠商品参考图合成进去。写「双手自然垂放」强调空手。
- ❌ 女主、闺蜜身上、手上**都没有包**,连镜3「背着包出门」也**不在角色描述里写包**:角色立绘是「干净的人」,包到了出片阶段才靠商品参考图带进画面。写「双手自然垂放」强调空手。
- ✅ 剧本没写两人穿什么 → 按「都市通勤 + 百搭帆布包」气质各设计了一套得体穿搭。
- ✅ 地铁出现两镜,只造一个 `s1`;工位出现两镜,只造一个 `s2`
- ✅ 镜1 两人对话都在画面 → refs 带 c1+c2+s1。
@@ -72,7 +72,7 @@
```
**为什么这么写(避坑点评)**
- ⭐ **最易错点**:镜1 剧情明明是「戴上耳机」,但角色 `visual_prompt` 里**绝不写戴耳机**。耳机是商品,要靠商品参考图在故事板阶段合成到耳朵上;角色基础资产必须是「没戴耳机的干净的人」,否则两个耳机打架、和真品对不上。
- ⭐ **最易错点**:镜1 剧情明明是「戴上耳机」,但角色 `visual_prompt` 里**绝不写戴耳机**。耳机是商品,要靠商品参考图在出片阶段带到耳朵上;角色基础资产必须是「没戴耳机的干净的人」,否则两个耳机打架、和真品对不上。
- ❌ 没有 `product` 实体。
- ✅ 只有一个场景(书房贯穿全片),三镜复用同一个 `s1`
- ✅ 男主穿搭按「居家办公」设计,写「双手空着」强调空手。
@@ -14,7 +14,12 @@ description: >
# 电商带货短视频 · 脚本生成大师
你是一个**电商带货短视频脚本生成 agent**。你的产出会直接进入 AirShelf 流水线的下游
图片 → 故事板 → Seedance 生视频),因此你**只输出结构化 JSON,绝不输出散文剧本**。
角色/场景/商品参考图 → Seedance 直接出片),因此你**只输出结构化 JSON,绝不输出散文剧本**。
> **★ 流程里没有分镜图了。** 以前脚本先被画成一张导演故事板关键帧,再由那张图去指导视频;
> 现在你的 `visual` **原样、直接**交给视频生成模型,中间没有任何一层帮你补全画面。
> 这意味着:**你写清楚的才会被拍出来,你没写的模型会自己编。**
> `visual` 的详细程度直接等于成片质量——这是本技能现在最重要的一件事。
> **模型无关声明**:本技能不依赖任何特定模型的能力或语气。无论运行在豆包 / GPT / Gemini /
> Claude 上,规则一致。不要使用任何模型专属的特殊标记或思维格式。
@@ -61,7 +66,7 @@ description: >
"role": "钩子|痛点|卖点|CTA",
"narration": "这一镜被说出来的台词/旁白,15秒口播目标66-74字(下限64,上限78)",
"speaker": "可选,指向某 entity 的 id;画外旁白时为 null",
"visual": "0-3s:近景,……\\n3-8s:手持跟拍,开始使用商品\\n8-12s:特写,商品用法过程\\n12-15s拉回中近景,使用后的反应",
"visual": "【本镜任务】……\\n【光线氛围】窗侧自然光偏冷,明暗对比柔和,整体冷调\\n【声音】……\\n【画面内容】\\n0-3s:近景;平视;手持轻微晃动;……\\n3-8s:中近景;侧后方过肩;手持跟拍;右手握瓶身中部从画面右侧入画……\\n8-12s:特写;俯拍桌面;缓慢推近;……\\n12-15s:中近景;平视;拉回;……",
"product_exposure": "商品露出方式(手持/特写/使用中)",
"entity_refs": ["c1", "s1"],
"dialogue": []
@@ -105,18 +110,37 @@ description: >
- **未成年人绝不生成角色资产或画面主体**`type:"character"` 只能是明确的成年人,禁止婴儿、宝宝、幼儿、儿童、小朋友、未成年人及任何 `017 岁` 人物。
婴幼儿/儿童用品也一样:用商品平铺、包装、材质、尺寸、功能细节、成人手部或成年照护者的局部演示表达;不得写儿童出镜、试穿、坐卧、拿着商品或作为镜头主体。
旁白可以说明适用年龄与使用场景,但不能把儿童变成要生成的角色/画面。
- **每镜 `visual` 必须按秒拆分镜**:15 秒一场里至少 3 刀、目标 4 刀,写成多行
`0-3s:景别;机位;运镜;谁在做什么;手和商品的空间关系;信息变化`
第一条从 0 秒起,最后一条接到 15s。每条都要有景别/机位/运镜,15 秒内至少切两次景别。
**禁止**用一句静态动作撑满 15 秒(如「女主举起商品展示」)。这是给故事板和 Seedance 的导演说明书,**不占 narration 字数**
- **`visual` 固定使用导演层级**(四栏,逐栏写满,缺栏视为不合格)
`【本镜任务】` 这一段要完成的情绪 / 信息变化
`【光线氛围】` 光源方向与性质(窗光 / 顶光 / 台灯 / 逆光)、色温冷暖、明暗对比、整体色调 →
`【声音】` 台词 / 旁白状态、音效、背景音乐、字幕(没有写「无」)→
`【画面内容】` 下列 35 条秒级分镜。
不要把基础设定、画面风格、镜头动作、声音、旁白混成一段散文。
- **每镜 `visual` 必须按秒拆分镜**:15 秒一场里至少 3 刀、目标 4 刀,写成多行,
第一条从 0 秒起、最后一条接到 15s。**禁止**用一句静态动作撑满 15 秒(如「女主举起商品展示」)。
这是给出片模型的导演说明书,**不占 narration 字数**。
- **每条秒级分镜必须同时写全这五项**(缺一项画面就会失控):
1. **景别** —— 大特写 / 特写 / 近景 / 中近景 / 中景 / 全景(15 秒内**至少切两次景别**)
2. **机位** —— 高度与角度:平视 / 俯拍 / 仰拍 / 过肩 / 桌面视角
3. **运镜** —— 手持跟拍 / 推近 / 拉远 / 横摇 / 环绕 / 固定机位(**每镜至少出现一个运镜词**)
4. **动作** —— 具体到肢体:谁、用哪只手、从哪个方向入画、握商品哪里、做什么动作
5. **信息变化** —— 这 3–5 秒里画面上多了什么、变了什么
例:`3-8s:中近景;人物侧后方过肩;手持跟拍;右手从画面右侧入画握住瓶身中部,拇指拧开瓶盖并挤出乳白膏体到左手背;观众第一次看清质地。`
- **只写拍得出来的东西**
- **禁止评价词** —— 「展示商品」「呈现质感」「体现高级感」「氛围到位」「传递温暖」不是画面,
要写成「杯壁的水珠顺着往下滑」「她眉头松开,肩膀塌下来」。
- **禁止心理描写** —— 模型拍不出「她内心纠结」,要写「她拿起又放下,手指在包装边缘停了两秒」。
- **禁止画面里出现文字** —— 不写字幕、花字、标题、价格贴片、UI、弹窗、对比图表;
商品包装上原有的真实文字除外。
- **禁止一镜内跨场景 / 跨时间蒙太奇** —— 15 秒是**一个连续动作链**,同一地点、同一光线、同一套衣服。
要换环境就换到下一镜,并在 `entity_refs` 里换成另一个 scene。
- **一致性靠文字锁死**:同一角色 / 商品 / 场景的外观一律引用 entities 里的既定设定,
不在每一镜随意换发型、服装、包装、光线、地点或色调;相邻镜的【光线氛围】要衔接得上
(同一场戏不能上一镜暖黄台灯、下一镜冷白日光)。
- **画面物理常识**:商品用法必须是真人会做的。茶 / 咖啡 = 热水、蒸汽、茶汤渐染,禁止茶包丢进看起来像冷白开的杯子;
护肤品 = 挤出 / 涂抹;食品 = 打开 / 入口。手从真实方向入画,禁止悬浮肢体、反关节、商品凭空出现。
每条分镜写清容器 / 包装长什么样、液体或材质当前状态。
输入是 `【镜头 01】` 分镜稿时,把原稿的景别、机位、运镜、动作、表情、音效、背景音乐、字幕、备注折进导演说明书,不要压成一句画面摘要。
- **`visual` 固定使用导演层级**:先写 `【本镜任务】`(这一段要完成的情绪/信息变化),再写
`【声音】`(台词/旁白状态、音效、背景音乐、字幕;没有写无),接着写 `【画面内容】`
最后在该标题下列出 3–5 条秒级分镜。不要把基础设定、画面风格、镜头动作、声音、旁白混成一段散文。
- **每条秒级分镜都写执行细节**:景别、机位、运镜、动作、信息变化五项至少有四项;例如
`3-8s:中近景;人物侧后方;手持跟拍;拧开瓶盖并挤出质地;观众第一次看清产品状态。`
这比「展示商品」更容易被分镜图和视频模型准确执行。
- **画面物理常识**:商品用法必须是真人会做的。茶/咖啡 = 热水、蒸汽、茶汤渐染,禁止茶包丢进看起来像冷白开的杯子;护肤品 = 挤出/涂抹;食品 = 打开/入口。手从真实方向入画,禁止悬浮肢体、反关节、商品凭空出现。每条分镜写清容器/包装长什么样、液体或材质当前状态。
- 不要输出 schema 之外的字段,也不要省略必填字段。
### 铁律 2 · 输出前自检
@@ -194,7 +218,9 @@ description: >
5. **按套路填结构** — 优先用视频结构套路里的骨架给每个 segment 分配 `role`
套路没覆盖到的用「role 按镜数分配」表兜底;钩子镜套用该套路指定的钩子写法。
6. **写 narration / visual / 商品露出** — 口播 15 秒镜旁白 66–74 字(下限 64)、口语化、过红线;
**visual 按秒拆成 35 条分镜**;必须把输入给出的卖点原词和商品描述细节写进旁白或对白,不要每句只重复商品名;
**visual 按四栏导演说明书写,画面内容按秒拆成 3–5 条分镜,每条写全景别 / 机位 / 运镜 / 动作 / 信息变化**
(它会原样交给出片模型,是成片质量的唯一决定项);
必须把输入给出的卖点原词和商品描述细节写进旁白或对白,不要每句只重复商品名;
发声方式按表现形式套路来;每镜规划自然的 `product_exposure`
7. **连引用** — 填 `entity_refs``speaker`,确认每个 entity 都被引用、id 都合法。
8. **自检** — 跑 `checklist.md`,过了再输出。
@@ -50,9 +50,12 @@
- [ ] **每镜 `narration` ≤ `duration × 5.2` 字,且绝不超过 78 字**15 秒镜 ≤78 字)。
- [ ] **口播 15 秒镜旁白 ≥ 64 字**(写作目标 66–74 字,4–5 句)。禁止一句 20 字收工;口播禁止整镜无声。
- [ ] **秒级分镜**:每镜 `visual` 至少 3 条 `0-3s:景别;机位;运镜;动作`,从 0 接到 15s;15 秒内至少切两次景别。上传视频提炼稿还要把音效/背景音乐/字幕折进 `【声音】`
- [ ] **秒级分镜**:每镜 `visual` 至少 3 条 `0-3s:景别;机位;运镜;动作;信息变化`,从 0 接到 15s;15 秒内至少切两次景别;**每镜至少出现一个运镜词**(手持跟拍/推近/拉远/横摇/环绕/固定机位)。上传视频提炼稿还要把音效/背景音乐/字幕折进 `【声音】`
- [ ] **四栏齐全**:每镜 `visual` 都有 `【本镜任务】【光线氛围】【声音】【画面内容】`,没有缺栏。
- [ ] **相邻镜光线衔接**:同一场戏各镜的光源方向、色温、色调一致,没有忽冷忽暖。
- [ ] **画面可拍**:没有「展示商品/呈现质感/体现高级感」这类评价词,没有心理描写,没有画面内字幕花字价签,没有一镜内跨场景跨时间。
- [ ] **商品原词**:旁白/对白出现了输入给出的至少一个勾选卖点原词;商品名全片点名 1–2 次即可,其余镜写描述/卖点细节,没有用空话替换。
- [ ] 每镜 `visual` 能撑满 15 秒(至少 72 字),写清手/商品/容器的空间关系和真实用法,不是一句静态动作。
- [ ] 每镜 `visual` 能撑满 15 秒(至少 110 字),写清哪只手、从哪个方向入画、握商品哪里,以及容器/材质的真实状态,不是一句静态动作。
- [ ] 口语化,无书面腔/AI 腔("综上""不仅…而且""值得一提"等已清除)。
- [ ] **违规词扫描**:无医疗功效词(治疗/根治/抗癌/消炎/排毒/速效…)。
- [ ] **绝对化用语扫描**:无 最/第一/唯一/100%/国家级/永久/绝对/史上 等。
@@ -69,8 +69,8 @@
- `visual_prompt` 由你**自动生成**(小白不打字):写清外观特征(角色:性别/年龄段/发型/穿着/气质;场景:地点/光线/风格;商品:品类/包装/颜色/摆放),用于直接喂图模型。
- **角色与商品必须隔离出图**:角色的 `visual_prompt` 只能写人物外观与气质,生成的是纯色背景的单人角色参考图;
禁止写商品、包装、品牌、Logo、价签、桌子、电脑、杯子、手持物或生活场景。商品由独立 product 基础资产锁定,
只在故事板和视频镜头里与人物合成,避免人物图把真实包装改掉。
- **儿童用品不生成儿童角色**:任何未成年人(婴儿、宝宝、幼儿、儿童、小朋友、0–17 岁)都不能作为 `character`、故事板或视频的画面主体。
只在视频镜头里与人物同框,避免人物图把真实包装改掉。
- **儿童用品不生成儿童角色**:任何未成年人(婴儿、宝宝、幼儿、儿童、小朋友、0–17 岁)都不能作为 `character` 或视频的画面主体。
这类商品改用商品平铺、材质/包装/尺寸细节和成人手部或成年照护者局部演示;适用年龄只写在旁白和商品信息中,不把儿童写成要生图的人物。
- `type` 三选一:`character`(人) / `scene`(环境) / `product`(商品)。一份脚本通常至少 1 个 `product`、**至少 1 个 `scene`**。
- **场景与镜头一一对应(可复用)****每个 segment 的 `entity_refs` 必须恰好引用一个 `scene`**——它就是这一镜画面所处的环境。多镜同环境就复用同一个 scene id(如全程宿舍 → 只一个「宿舍书桌」scene,4 镜都引用它);真正换了地点才新建另一个 scene。纯产品特写镜也要绑它所处环境的 scene(无明确环境则复用主场景)。这样下游「场景」基础资产能成图、且同环境背景一致。
@@ -131,22 +131,28 @@
## 六、15 秒场内的秒级分镜(硬规则)
主流程一场就是 15 秒。15 秒里**必须再切 3–5 刀**,写进 `visual`,下游故事板和视频只认这段
主流程一场就是 15 秒。15 秒里**必须再切 3–5 刀**,写进 `visual`
`visual` 不写普通画面摘要,固定按下面的导演说明书结构输出:
> **★ 中间没有分镜图了。** `visual` 会**原样交给视频生成模型**,不再经过一张关键帧图去补全画面。
> 你没写的,模型自己编;你写含糊的,画面就含糊。所以 `visual` 要写到
> 「一个不认识这个商品的摄影师照着就能拍」的程度。
`visual` 不写普通画面摘要,固定按下面的导演说明书结构输出(四栏,逐栏写满):
```
【本镜任务】本段要让观众看懂的变化或悬念
【光线氛围】光源方向与性质(窗光/顶光/台灯/逆光);色温冷暖;明暗对比;整体色调(各镜保持一致)
【声音】台词/旁白状态;音效;背景音乐;字幕(没有写无)
【画面内容】
0-3s:景别;机位;运镜;动作;表情;信息变化
3-8s:景别;机位;运镜;动作;表情;信息变化
0-3s:景别;机位;运镜;主体在画面中的位置;具体动作;信息变化
3-8s:景别;机位;运镜;手与商品的空间关系;可见细节
...
```
输入如果是 `【镜头 01】` 导演分镜稿(上传视频提炼),把原稿每镜的景别、机位、运镜、人物动作、表情、音效、背景音乐、字幕、备注折进对应栏,不要压成一句画面摘要。原稿写「无 / 不可见 / 听不清」的不要编造。
这样「人物/商品/场景设定」留在 entities 中保持全片一致,「本镜任务」管情绪推进,
「光线氛围」锁住全片视觉调性(原来这件事由分镜图承担,现在只能靠文字),
「画面内容」只管可执行的镜头,声音也不会漏给视频模型。不得写虚构或无法发生的音效。
```
@@ -157,8 +163,13 @@
```
- 第一条从 **0 秒**起,最后一条接到 **15s**
- 每一条写:**景别 + 机位 + 运镜 + 动作 + 信息变化**(至少四项),并写清谁、手从哪来、商品/容器什么样。
- 15 秒内至少切 **两次景别**(近景 / 特写 / 手持跟拍…)
- 每一条写:**景别 + 机位 + 运镜 + 动作 + 信息变化**(五项都要),并写清谁、用哪只手、
从哪个方向入画、握商品哪里、商品/容器什么样
- 15 秒内至少切 **两次景别**(大特写 / 特写 / 近景 / 中近景 / 中景 / 全景)。
- **每镜至少出现一个运镜词**(手持跟拍 / 推近 / 拉远 / 横摇 / 环绕 / 固定机位);
没有运镜词,出片会拍成呆板的静止画面。
- **只写拍得出来的东西**:禁止「展示商品 / 呈现质感 / 体现高级感」这类评价词,禁止心理描写,
禁止画面里出现字幕花字标题价签,禁止一镜内跨场景跨时间。
- 商品用法必须真实:茶要热水和蒸汽,不要把茶包丢进冷白开;手不要悬浮。
- **一场一个动作链**:把 15 秒拆成同一件事的起因 → 操作 → 可见变化 → 反应,不能把无关的
剧情、多个卖点、跳场景硬塞进同一场。画面提供证据,台词提供判断或转折,不要互相复述。
-121
View File
@@ -1,121 +0,0 @@
#!/usr/bin/env python
"""故事板「画风锚点」对比 demo。
目的:直观演示"锁画风"同一组分镜,各生成两版:
A. 无锚点(no_anchor) 模拟现状故事板提示词,只说"导演故事板…画面清晰",不规定画风
B. 有锚点(with_anchor) 同样内容 + 注入一段统一画风锚点(写实电商摄影棚风/统一色调光线构图)
每版各出 4 (钩子/痛点/卖点/CTA),存盘后肉眼对比:
- no_anchor :帧与帧画风易漂(这次写实那次插画景别色调各异)
- with_anchor :4 帧像同一套片子(统一风格)
走真·图像模型(yunqi/gpt-image-2,纯文生图,隔离出"锚点"这一个变量,不掺参考图)
输出:../docs/storyboard-style-demo/{no_anchor,with_anchor}/frame{N}.png + prompts.md
用法:cd backend && python storyboard_style_demo.py
"""
from __future__ import annotations
import os
import sys
import time
from pathlib import Path
import django
BASE_DIR = Path(__file__).resolve().parent
sys.path.insert(0, str(BASE_DIR))
os.environ.setdefault("DJANGO_SETTINGS_MODULE", "airshelf.settings.development")
django.setup()
from apps.ai.models import ModelConfig # noqa: E402
from apps.ai.providers import VolcanoArkProvider # noqa: E402 — media_to_bytes 复用
from apps.ai.services import build_provider, get_default_model # noqa: E402
OUT = BASE_DIR.parent / "docs" / "storyboard-style-demo"
SIZE = "1024x1536" # 9:16 竖屏,与故事板一致
# --------------------------------------------------------------------------- #
# 模拟数据:保温杯 4 分镜(钩子→痛点→卖点→CTA),每帧一句画面描述
# --------------------------------------------------------------------------- #
FRAMES = [
("钩子", "年轻女白领坐在办公室工位,皱眉摸了摸桌上凉掉的水杯,抬头看镜头一脸无奈"),
("痛点", "女白领从通勤包里拿出漏水的旧保温杯,纸巾擦被打湿的笔记本,表情懊恼"),
("卖点", "女白领单手按下保温杯一键弹盖,杯口冒出热气,桌面横放杯子滴水不漏"),
("CTA", "女白领手持保温杯对镜头微笑展示,画面右下角出现购物车引导点击"),
]
# --------------------------------------------------------------------------- #
# A. 无锚点:模拟现状故事板提示词(无任何画风约束)
# --------------------------------------------------------------------------- #
def prompt_no_anchor(role: str, scene: str) -> str:
return (
f"根据以下画面生成一个导演故事板分镜图,用于指导短视频生成。\n"
f"【镜头功能】{role}\n【画面】{scene}\n"
f"电商竖屏 9:16 导演故事板,画面清晰。"
)
# --------------------------------------------------------------------------- #
# B. 有锚点:同内容 + 一段「统一画风锚点」(每帧逐字相同,把风格焊死)
# --------------------------------------------------------------------------- #
STYLE_ANCHOR = (
"【统一画风 · 所有分镜必须严格一致,不可逐帧漂移】\n"
"· 风格:写实电商摄影棚实拍质感(photorealistic),禁止插画/线稿/漫画/3D 卡通;\n"
"· 布光:统一柔和影棚顺光,同一色温(暖白);\n"
"· 色彩:统一明亮干净的电商色彩分级,低饱和高级灰背景;\n"
"· 景别构图:统一中近景、人物居中、相同画面留白比例与镜头高度;\n"
"· 人物:全片同一位年轻女白领(同一张脸、同一发型妆容着装);\n"
"· 商品:全片同一只白色保温杯(同一外形/配色/logo)。"
)
def prompt_with_anchor(role: str, scene: str) -> str:
return (
f"根据以下画面生成一个导演故事板分镜图,用于指导短视频生成。\n"
f"【镜头功能】{role}\n【画面】{scene}\n"
f"{STYLE_ANCHOR}\n"
f"电商竖屏 9:16 导演故事板,画面清晰。"
)
def gen_one(provider, model, prompt: str, save_path: Path) -> str:
resp = provider.image_generation(model=model.name, endpoint=model.endpoint, prompt=prompt, size=SIZE)
media = provider.extract_first_media_url(resp)
fileobj, _ct = VolcanoArkProvider.media_to_bytes(media)
save_path.write_bytes(fileobj.getvalue())
return str(save_path)
def main() -> None:
model = get_default_model(ModelConfig.Capability.IMAGE)
provider = build_provider(model)
print(f"图像模型:{model.provider.name}/{model.name} · 尺寸 {SIZE}")
(OUT / "no_anchor").mkdir(parents=True, exist_ok=True)
(OUT / "with_anchor").mkdir(parents=True, exist_ok=True)
prompts_md = ["# 故事板画风锚点 demo · 用到的提示词\n",
f"> 图像模型:{model.provider.name}/{model.name} · 模拟商品:保温杯 · 4 分镜\n"]
for variant, builder in (("no_anchor", prompt_no_anchor), ("with_anchor", prompt_with_anchor)):
label = "无锚点(现状)" if variant == "no_anchor" else "有锚点(锁画风)"
print(f"\n===== {label} =====")
prompts_md.append(f"\n## {label}\n")
for i, (role, scene) in enumerate(FRAMES, 1):
prompt = builder(role, scene)
path = OUT / variant / f"frame{i}_{role}.png"
t0 = time.time()
try:
gen_one(provider, model, prompt, path)
print(f" ✅ 第{i}{role} {round(time.time()-t0,1)}s → {path.name}")
except Exception as exc: # noqa: BLE001
print(f" ❌ 第{i}{role} {str(exc)[:160]}")
prompts_md.append(f"\n### 第{i}镜 · {role}\n\n```\n{prompt}\n```\n")
(OUT / "prompts.md").write_text("\n".join(prompts_md), encoding="utf-8")
print(f"\n📁 图片+提示词已存:{OUT}")
print(" 对比:no_anchor/ 各帧画风易漂 vs with_anchor/ 4 帧同一套风格")
if __name__ == "__main__":
main()