931 lines
42 KiB
Python
931 lines
42 KiB
Python
"""对话式脚本生成 agent(出稿 + 改稿一体,多模型可选,流式 SSE)。
|
|
|
|
设计:
|
|
- 加载电商脚本 skill(SKILL.md + references)作为领域知识系统提示词;模型无关。
|
|
- 3 种输入模式(全自动 / 一句话 / 改稿)收敛到同一份结构化 ScriptDraft(铁律1契约)。
|
|
- 流式:边生成边吐「工具卡 + 思考」事件,给前端真 agent 体感;JSON 由后端可靠抽取,不靠模型排版。
|
|
- 计费走现有 AITask + 额度预扣(reserve→charge/release),与 generate_project_script 一致。
|
|
|
|
SSE 事件(每帧 `data: {json}\n\n`,json 带 type):
|
|
tool {id,label?,status:running|done|error} —— 工具卡(加载skill/分析商品/生成分镜/提取实体/自检)
|
|
reasoning {text} —— 推理模型思考流(逐字,纯展示,不进答案)
|
|
delta {text} —— 模型自然语言前言(JSON 部分不外露)
|
|
draft {draft} —— 规范化后的 ScriptDraft(前端结构化渲染)
|
|
saved {script_version_id, version} —— 已落库的 ScriptVersion(含 segments/metadata)
|
|
summary {text} —— 模型自己写的收尾交付语(当 AI 回复气泡,替代写死的「已生成」)
|
|
done {} —— 结束
|
|
error {detail,error} —— 失败(已回滚额度；error 为安全业务错误对象)
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import re
|
|
from decimal import Decimal
|
|
from functools import lru_cache
|
|
from pathlib import Path
|
|
|
|
from django.conf import settings
|
|
from django.utils import timezone
|
|
|
|
from apps.ai.generation_errors import classify_generation_error
|
|
from apps.ai.models import AITask, ModelConfig
|
|
from apps.billing.services.ledger import charge_reserved_credit, release_credit
|
|
|
|
VALID_TONES = ["种草", "测评", "剧情", "痛点"]
|
|
VALID_ROLES = ["钩子", "痛点", "卖点", "CTA"]
|
|
VALID_ENTITY_TYPES = ["character", "scene", "product"]
|
|
DURATION_TIERS = [15, 30, 60, 90]
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# skill 加载(缓存)
|
|
# --------------------------------------------------------------------------- #
|
|
def _skill_dir() -> Path:
|
|
override = getattr(settings, "ECOMMERCE_SKILL_DIR", None)
|
|
if override:
|
|
return Path(override)
|
|
# skills 已随后端打进镜像(core/backend/skills);优先 BASE_DIR/skills,回落仓库根(本地/旧布局)。
|
|
# 旧版只看仓库根 → 镜像里没有(构建上下文是 ./core/backend)→ 提示词为空、退化兜底。详见 services._skills_root。
|
|
base = Path(settings.BASE_DIR)
|
|
for cand in (base / "skills", base.parent.parent / "skills"):
|
|
if (cand / "ecommerce-video-script").is_dir():
|
|
return cand / "ecommerce-video-script"
|
|
return base / "skills" / "ecommerce-video-script"
|
|
|
|
|
|
@lru_cache(maxsize=1)
|
|
def load_ecommerce_skill() -> str:
|
|
"""读取 SKILL.md + 全部 references 拼成系统提示词(领域知识)。缺文件不致命,尽量给。"""
|
|
skill_dir = _skill_dir()
|
|
parts: list[str] = []
|
|
main = skill_dir / "SKILL.md"
|
|
if main.exists():
|
|
parts.append(main.read_text(encoding="utf-8"))
|
|
ref_dir = skill_dir / "references"
|
|
if ref_dir.exists():
|
|
for ref in sorted(ref_dir.glob("*.md")):
|
|
parts.append(f"\n\n===== references/{ref.name} =====\n\n{ref.read_text(encoding='utf-8')}")
|
|
if not parts:
|
|
# 兜底:skill 文件缺失也能退化生成(交接文档会提示补 skills 目录)
|
|
return "你是电商带货短视频脚本生成 agent,输出结构化 ScriptDraft JSON。"
|
|
return "".join(parts)
|
|
|
|
|
|
# 运行时输出协议:优先级高于 skill 里的「只输出 JSON / 不展示思考」,只为流式体感放开一句前言。
|
|
_OUTPUT_PROTOCOL = """
|
|
|
|
---
|
|
|
|
## 运行时输出协议(AirShelf 流式展示专用,优先级高于技能正文的「只输出 JSON」)
|
|
|
|
严格按以下顺序输出,不要有别的内容:
|
|
1. 先用 **1 句中文口语**告诉用户你正在做什么(≤40 字,例:「在为这款保温杯生成 4 镜痛点脚本…」),让用户看到进展;
|
|
2. 紧接着输出**且仅输出一个** ```json 代码块,内容为符合技能契约(铁律1)的 ScriptDraft 对象;
|
|
3. json 代码块**收尾之后另起一行**,用 **1–2 句中文口语**跟用户交付这一版:做了什么、为什么这么改、可以怎么接着调(像同事汇报,**别复述 JSON 字段、别再写代码块**)。这句会作为你的回复气泡展示给用户。
|
|
|
|
### 字段名锚定(硬性 · 下游靠它取数,跑偏即数据全空)
|
|
- 分镜数组的键名**必须**叫 `segments`(禁止用 scenes / script / shots / 分镜 等同义词)。
|
|
- 每镜口播键名**必须**叫 `narration`(禁止用 voiceover / audio / line)。
|
|
- 每镜画面键名**必须**叫 `visual`,且为**一句话字符串**(禁止用 scene / screen / 画面,也禁止写成 {setting,camera,...} 对象)。
|
|
- 即使你额外附带了 scenes / shots 等创作结构,也**必须同时**给出标准 `segments` 数组,并把口播填进 `narration`、画面填进 `visual`,否则视为不合格。
|
|
"""
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# 提示词构建(3 模式)
|
|
# --------------------------------------------------------------------------- #
|
|
def _product_context(project, selling_point_ids: list[str] | None) -> str:
|
|
product = project.product
|
|
selling_points = product.selling_points.all()
|
|
if selling_point_ids:
|
|
selling_points = selling_points.filter(id__in=selling_point_ids)
|
|
selling_text = "\n".join(f"- {sp.title}:{sp.detail}" for sp in selling_points)
|
|
return (
|
|
f"商品标题:{product.title}\n"
|
|
f"品牌:{product.brand or '未填写'}\n"
|
|
f"类目:{product.category or '未填写'}\n"
|
|
f"目标人群:{product.target_audience or '未填写'}\n"
|
|
f"商品描述:{product.description or '未填写'}\n"
|
|
f"卖点:\n{selling_text or '未勾选卖点,请根据商品信息自行提炼。'}"
|
|
)
|
|
|
|
|
|
def build_agent_messages(
|
|
*,
|
|
project,
|
|
mode: str,
|
|
user_prompt: str,
|
|
selling_point_ids: list[str] | None,
|
|
base_draft: dict | None,
|
|
aspect_ratio: str,
|
|
total_duration: int,
|
|
target_index: int | None = None,
|
|
) -> list[dict[str, str]]:
|
|
system = load_ecommerce_skill() + _OUTPUT_PROTOCOL
|
|
head = (
|
|
f"【画幅】{aspect_ratio}\n"
|
|
f"【总时长】{total_duration} 秒(每 15 秒一镜,共 {total_duration // 15} 镜)\n"
|
|
f"【商品信息】\n{_product_context(project, selling_point_ids)}"
|
|
)
|
|
if mode == "revise" and base_draft and target_index is not None:
|
|
# 精准改一镜:读全脚本上下文,只重写第 N 镜,强制与前后镜衔接;其余镜后端会强制保持原样。
|
|
user = (
|
|
f"【任务】只重写第 {target_index + 1} 镜(共 {len(base_draft.get('segments', []))} 镜),其余镜保持不变。\n"
|
|
f"{head}\n\n"
|
|
f"【现有完整脚本 JSON(读它保证与前后镜衔接)】\n{json.dumps(base_draft, ensure_ascii=False)}\n\n"
|
|
f"【对第 {target_index + 1} 镜的修改意见】{user_prompt.strip() or '让这一镜更有吸引力、表达更清晰。'}\n\n"
|
|
"仍输出**完整** ScriptDraft(我只会采用第 "
|
|
f"{target_index + 1} 镜的改动);若意见涉及角色对白,就给这一镜填 dialogue。"
|
|
)
|
|
elif mode == "revise" and base_draft:
|
|
user = (
|
|
"【任务】改稿(模式③):在保留用户原意的前提下,增强钩子/节奏/卖点/CTA,并归一化到契约 JSON。\n"
|
|
f"{head}\n\n"
|
|
f"【现有脚本 JSON】\n{json.dumps(base_draft, ensure_ascii=False)}\n\n"
|
|
f"【用户修改意见】{user_prompt.strip() or '让整体更有吸引力、转化感更强,并保持各镜衔接连贯。'}\n\n"
|
|
"请输出修订后的**完整** ScriptDraft。"
|
|
)
|
|
elif mode == "theme" or (user_prompt and user_prompt.strip()):
|
|
user = (
|
|
"【任务】一句话主题扩写(模式②):以用户主题为脚本主轴,其余自动补全。\n"
|
|
f"{head}\n\n"
|
|
f"【用户主题】{user_prompt.strip()}\n\n"
|
|
"请按技能流程一次性产出 ScriptDraft。"
|
|
)
|
|
else:
|
|
user = (
|
|
"【任务】全自动(模式①):仅凭商品与前置条件,自动定档/选 tone/造 entity/填黄金结构。\n"
|
|
f"{head}\n\n"
|
|
"请按技能流程一次性产出 ScriptDraft。"
|
|
)
|
|
return [{"role": "system", "content": system}, {"role": "user", "content": user}]
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# JSON 抽取 + 契约规范化(模型无关,后端兜底)
|
|
# --------------------------------------------------------------------------- #
|
|
def _balanced_object(text: str) -> str | None:
|
|
"""从首个 '{' 起按括号深度扫描,返回第一个配平的 {...}(忽略字符串内的括号)。
|
|
避免 rfind('}') 在 JSON 后还有含花括号的散文时越界截出非法片段。"""
|
|
start = text.find("{")
|
|
if start == -1:
|
|
return None
|
|
depth = 0
|
|
in_str = False
|
|
esc = False
|
|
for i in range(start, len(text)):
|
|
c = text[i]
|
|
if in_str:
|
|
if esc:
|
|
esc = False
|
|
elif c == "\\":
|
|
esc = True
|
|
elif c == '"':
|
|
in_str = False
|
|
continue
|
|
if c == '"':
|
|
in_str = True
|
|
elif c == "{":
|
|
depth += 1
|
|
elif c == "}":
|
|
depth -= 1
|
|
if depth == 0:
|
|
return text[start : i + 1]
|
|
return None
|
|
|
|
|
|
def _looks_like_draft(blob: str) -> bool:
|
|
try:
|
|
d = json.loads(blob)
|
|
except (ValueError, TypeError):
|
|
return False
|
|
return isinstance(d, dict) and ("segments" in d or "hook" in d)
|
|
|
|
|
|
def _extract_json(text: str) -> str | None:
|
|
"""抽取 ScriptDraft JSON。容错:模型可能先给示例 ```json 块再给正式块,
|
|
故取**最后一个**含 segments/hook 的合法围栏块;都不像草稿再退而取末个配平对象;无围栏再裸扫。"""
|
|
fences = re.findall(r"```(?:json)?\s*(.+?)```", text, re.DOTALL)
|
|
for block in reversed(fences):
|
|
obj = _balanced_object(block)
|
|
if obj and _looks_like_draft(obj):
|
|
return obj
|
|
for block in reversed(fences):
|
|
obj = _balanced_object(block)
|
|
if obj:
|
|
return obj
|
|
return _balanced_object(text)
|
|
|
|
|
|
def _nearest_duration(value) -> int:
|
|
try:
|
|
value = int(value)
|
|
except (TypeError, ValueError):
|
|
return 60
|
|
if value in DURATION_TIERS:
|
|
return value
|
|
return min(DURATION_TIERS, key=lambda t: abs(t - value))
|
|
|
|
|
|
# 模型每次生成都可能换字段名(scene/screenDescription/visual…、dialogue/lines/caption…),
|
|
# 与其逐一追变体,不如「优先键命中 → 否则按关键词模糊匹配」通用解析。SKIP 掉明显的非内容键,
|
|
# 避免误抓(shotNo/duration/bgMusic/note 等)。
|
|
_PICK_SKIP_KEYS = {
|
|
"shotno", "shotsize", "shotsizetype", "duration", "duration_seconds", "starttime", "endtime",
|
|
"timerange", "time_range", "bgmusic", "bg_music", "music", "sound", "sfx", "note", "notes",
|
|
"tips", "index", "role", "speaker", "entity_refs", "product_exposure", "id", "no", "transition",
|
|
# 标题/编号类:含 scene 字样会误命中画面 fuzzy,显式跳过(注意:不跳裸 scene,它常=画面)
|
|
"scene_id", "sceneid", "scene_no", "sceneno", "scene_number", "scenenumber",
|
|
"scene_title", "scenetitle", "title", "scene_title_type",
|
|
}
|
|
_VISUAL_EXACT = ("visual", "visual_prompt", "visual_description", "screen_description", "screendescription", "screen", "picture", "shot_description", "scene", "画面")
|
|
_VISUAL_FUZZY = ("visual", "screen", "picture", "scene", "画面", "镜头描述", "分镜画面")
|
|
_NARRATION_EXACT = ("narration", "voiceover", "voice_over", "vo", "line", "caption", "subtitle", "speech", "口播", "旁白")
|
|
_NARRATION_FUZZY = ("narrat", "voice", "旁白", "口播", "台词", "dialog", "caption", "subtitle", "字幕", "speech", "monolog", "audio")
|
|
|
|
|
|
def _flatten_text(val) -> str:
|
|
"""把 字符串/字典/列表 里的文本拍平成一句。模型常把 visual 写成 {setting,camera,key_shots}
|
|
对象、把口播写成数组,这里统一抽成纯文本,避免结构化值被当空丢弃(全空根因之一)。"""
|
|
if isinstance(val, str):
|
|
return val.strip()
|
|
if isinstance(val, dict):
|
|
return " · ".join(p for p in (_flatten_text(v) for v in val.values()) if p)
|
|
if isinstance(val, (list, tuple)):
|
|
return " ".join(p for p in (_flatten_text(v) for v in val) if p)
|
|
return ""
|
|
|
|
|
|
def _pick_field(seg: dict, exact: tuple[str, ...], fuzzy: tuple[str, ...]) -> str:
|
|
"""从一镜里取某类文本字段:先按优先键精确命中,再按关键词在剩余键里模糊匹配(跳过非内容键)。
|
|
值允许是 字符串/对象/数组(嵌套结构拍平成一句),不再只认裸字符串。"""
|
|
for key in exact:
|
|
if key in seg:
|
|
txt = _flatten_text(seg.get(key))
|
|
if txt:
|
|
return txt
|
|
for key, val in seg.items():
|
|
kl = str(key).lower()
|
|
if kl in _PICK_SKIP_KEYS:
|
|
continue
|
|
if any(f in kl for f in fuzzy):
|
|
txt = _flatten_text(val)
|
|
if txt:
|
|
return txt
|
|
return ""
|
|
|
|
|
|
# 模型给分镜数组的键名五花八门(segments/scenes/script/shots…),且常同时给一个**空的**
|
|
# segments 骨架 + 真内容放在 scenes 里。所以不能「见 segments 是 list 就用」,要在所有候选里
|
|
# 挑「能解析出最多非空旁白/画面」的那个。segments(契约本名)排第一,同分时优先。
|
|
_SEGMENT_ARRAY_KEYS = (
|
|
"segments", "shots", "scenes", "script", "shot_list", "shotlist", "shotList",
|
|
"scene_list", "scenes_list", "storyboard", "分镜", "镜头",
|
|
)
|
|
|
|
|
|
def _resolve_segments(draft: dict) -> list:
|
|
"""在所有候选数组键里挑内容最丰富的分镜数组(按非空旁白/画面条数打分)。"""
|
|
best: list = []
|
|
best_score = -1
|
|
for key in _SEGMENT_ARRAY_KEYS:
|
|
arr = draft.get(key)
|
|
if not isinstance(arr, list) or not arr:
|
|
continue
|
|
dict_items = [s for s in arr if isinstance(s, dict)]
|
|
if not dict_items:
|
|
continue
|
|
score = sum(
|
|
1
|
|
for s in dict_items
|
|
if _pick_field(s, _NARRATION_EXACT, _NARRATION_FUZZY)
|
|
or _pick_field(s, _VISUAL_EXACT, _VISUAL_FUZZY)
|
|
)
|
|
if score > best_score: # 严格大于 → 同分保留更靠前的键(segments 优先)
|
|
best, best_score = arr, score
|
|
return best
|
|
|
|
|
|
def normalize_draft(raw_text: str, *, aspect_ratio: str, total_duration: int) -> dict:
|
|
"""把模型输出抽成 JSON 并按铁律1契约规范化。宽容:小问题就地修,不轻易抛错。"""
|
|
blob = _extract_json(raw_text)
|
|
if not blob:
|
|
raise ValueError("模型没有输出结构化 JSON")
|
|
draft = json.loads(blob)
|
|
if not isinstance(draft, dict):
|
|
raise ValueError("脚本 JSON 顶层不是对象")
|
|
|
|
# 兼容模型不按契约的常见变体:① 把内容裹进 {"ScriptDraft": {...}} 外壳;
|
|
# ② 用 basicInfo(camelCase)放时长/比例;③ 用 shots 代替 segments。
|
|
# 解开外壳并把别名拍平到契约字段,避免「找不到 segments → 全填空占位镜」(旁白/画面全空)。
|
|
for wrapper in ("ScriptDraft", "script_draft", "scriptDraft", "draft"):
|
|
inner = draft.get(wrapper)
|
|
if isinstance(inner, dict):
|
|
draft = {**inner, **{k: v for k, v in draft.items() if k != wrapper}}
|
|
break
|
|
basic = draft.get("basicInfo") if isinstance(draft.get("basicInfo"), dict) else {}
|
|
if basic:
|
|
draft.setdefault("total_duration", basic.get("totalDuration") or basic.get("total_duration"))
|
|
draft.setdefault("aspect_ratio", basic.get("aspectRatio") or basic.get("aspect_ratio"))
|
|
draft.setdefault("theme", basic.get("theme"))
|
|
# 在所有候选键里挑内容最丰富的分镜数组(空 segments 骨架会被更丰富的 scenes/shots 顶替)
|
|
draft["segments"] = _resolve_segments(draft)
|
|
|
|
draft["aspect_ratio"] = (draft.get("aspect_ratio") or aspect_ratio or "9:16").strip()
|
|
dur = _nearest_duration(draft.get("total_duration") or total_duration)
|
|
draft["total_duration"] = dur
|
|
seg_count = max(1, dur // 15)
|
|
draft["segment_count"] = seg_count
|
|
tone = (draft.get("tone") or "").strip()
|
|
draft["tone"] = tone if tone in VALID_TONES else "种草"
|
|
draft["hook"] = (draft.get("hook") or "").strip()
|
|
|
|
# entities 规范化:补 id / ref_index,过滤非法 type
|
|
entities = draft.get("entities") if isinstance(draft.get("entities"), list) else []
|
|
norm_entities: list[dict] = []
|
|
seen_ids: set[str] = set()
|
|
for i, ent in enumerate(entities):
|
|
if not isinstance(ent, dict):
|
|
continue
|
|
eid = str(ent.get("id") or f"e{i + 1}").strip() or f"e{i + 1}"
|
|
while eid in seen_ids:
|
|
eid = f"{eid}_{i}"
|
|
seen_ids.add(eid)
|
|
etype = (ent.get("type") or "").strip()
|
|
if etype not in VALID_ENTITY_TYPES:
|
|
etype = "character"
|
|
norm_entities.append(
|
|
{
|
|
"id": eid,
|
|
"type": etype,
|
|
"name": (ent.get("name") or eid).strip(),
|
|
"visual_prompt": (ent.get("visual_prompt") or "").strip(),
|
|
"ref_index": ent.get("ref_index") if isinstance(ent.get("ref_index"), int) else i + 1,
|
|
"voice_ref": ent.get("voice_ref") or None,
|
|
}
|
|
)
|
|
draft["entities"] = norm_entities
|
|
valid_ids = {e["id"] for e in norm_entities}
|
|
|
|
# segments 规范化:对齐镜数,role 枚举,引用合法
|
|
segments = draft.get("segments") if isinstance(draft.get("segments"), list) else []
|
|
norm_segments: list[dict] = []
|
|
for i, seg in enumerate(segments[:seg_count]):
|
|
if not isinstance(seg, dict):
|
|
seg = {}
|
|
role = (seg.get("role") or "").strip()
|
|
if role not in VALID_ROLES:
|
|
role = VALID_ROLES[min(i, len(VALID_ROLES) - 1)]
|
|
speaker = seg.get("speaker")
|
|
speaker = speaker if (speaker in valid_ids) else None
|
|
refs = [r for r in (seg.get("entity_refs") or []) if r in valid_ids]
|
|
# 对白(剧情向):[{speaker(合法 entity id 或 null=旁白), line}];默认空 = 纯口播。
|
|
# 模型变体的口播字段五花八门:dialogue(字符串/数组)/ lines(数组)/ 每项 line|text|content。
|
|
# 这里把任意数组形态归一成结构化对白,字符串形态留给下面当整句旁白。
|
|
raw_dialogue = seg.get("dialogue")
|
|
dialogue = []
|
|
dialogue_array = raw_dialogue if isinstance(raw_dialogue, list) else (seg.get("lines") if isinstance(seg.get("lines"), list) else [])
|
|
for d in dialogue_array:
|
|
if not isinstance(d, dict):
|
|
continue
|
|
line = (d.get("line") or d.get("text") or d.get("content") or "").strip()
|
|
if not line:
|
|
continue
|
|
sp = d.get("speaker")
|
|
dialogue.append({"speaker": sp if sp in valid_ids else None, "line": line})
|
|
# 旁白:结构化对白/lines 优先,其次整句字符串 dialogue,再退到通用字段解析(narration/voiceover/caption/字幕…)
|
|
narration = ""
|
|
if isinstance(raw_dialogue, str) and raw_dialogue.strip():
|
|
narration = raw_dialogue.strip()
|
|
elif dialogue:
|
|
narration = " ".join(d["line"] for d in dialogue) # 扁平拼接,兼容下游字幕/配音
|
|
if not narration:
|
|
narration = _pick_field(seg, _NARRATION_EXACT, _NARRATION_FUZZY)
|
|
# 画面:通用解析(visual/scene/screenDescription/画面… 都能命中,跳过 shotNo/bgMusic 等)
|
|
visual = _pick_field(seg, _VISUAL_EXACT, _VISUAL_FUZZY)
|
|
norm_segments.append(
|
|
{
|
|
"index": i,
|
|
"duration": 15,
|
|
"role": role,
|
|
"narration": narration,
|
|
"speaker": speaker,
|
|
"visual": visual,
|
|
"product_exposure": (seg.get("product_exposure") or "").strip(),
|
|
"entity_refs": refs,
|
|
"dialogue": dialogue,
|
|
}
|
|
)
|
|
# 不足镜数则补占位镜(极少发生,避免下游镜数对不上)
|
|
while len(norm_segments) < seg_count:
|
|
i = len(norm_segments)
|
|
norm_segments.append(
|
|
{
|
|
"index": i,
|
|
"duration": 15,
|
|
"role": VALID_ROLES[min(i, len(VALID_ROLES) - 1)],
|
|
"narration": "",
|
|
"speaker": None,
|
|
"visual": "",
|
|
"product_exposure": "",
|
|
"entity_refs": [],
|
|
"dialogue": [],
|
|
}
|
|
)
|
|
if not norm_segments:
|
|
raise ValueError("脚本没有任何分镜")
|
|
draft["segments"] = norm_segments
|
|
return draft
|
|
|
|
|
|
def _merge_single_segment(base: dict, new: dict, idx: int, aspect_ratio: str, total_duration: int) -> dict:
|
|
"""精准改一镜:以基准稿为底,只用新稿的第 idx 镜替换,其余镜逐字保持;合并新稿引入的新 entity(对白可能加角色)。再整体规范化。"""
|
|
merged = json.loads(json.dumps(base)) # 深拷贝
|
|
base_ids = {e.get("id") for e in merged.get("entities", []) if isinstance(e, dict)}
|
|
for e in new.get("entities", []):
|
|
if isinstance(e, dict) and e.get("id") and e["id"] not in base_ids:
|
|
merged.setdefault("entities", []).append(e)
|
|
base_ids.add(e["id"])
|
|
new_segs = new.get("segments", [])
|
|
target = next((s for s in new_segs if isinstance(s, dict) and s.get("index") == idx), None)
|
|
if target is None and 0 <= idx < len(new_segs):
|
|
target = new_segs[idx]
|
|
if not isinstance(target, dict):
|
|
# 模型没产出目标镜(没按 N 镜输出)→ 抛错让上层释放额度+报错,而不是静默返回 base 空转计费
|
|
raise ValueError(f"模型未产出第 {idx + 1} 镜的改动,请重试")
|
|
segs = merged.get("segments", [])
|
|
if isinstance(target, dict) and 0 <= idx < len(segs):
|
|
target = dict(target)
|
|
target["index"] = idx
|
|
segs[idx] = target
|
|
merged["segments"] = segs
|
|
return normalize_draft(json.dumps(merged, ensure_ascii=False), aspect_ratio=aspect_ratio, total_duration=total_duration)
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# 落库
|
|
# --------------------------------------------------------------------------- #
|
|
def _map_entities_to_project_metadata(project, entities: list[dict]) -> None:
|
|
"""把结构化 entities 回填到 project.metadata,复用下游已有的 cast/scenes/*_prompts 接线
|
|
(脚本页标签 + 基础资产 seed + 故事板 @图N)。只在有内容时覆盖,空结果不清旧标签。"""
|
|
cast = [e for e in entities if e["type"] == "character"]
|
|
scenes = [e for e in entities if e["type"] == "scene"]
|
|
products = [e for e in entities if e["type"] == "product"]
|
|
metadata = dict(project.metadata or {})
|
|
if cast:
|
|
metadata["cast"] = [e["name"] for e in cast]
|
|
metadata["cast_prompts"] = {e["name"]: e["visual_prompt"] for e in cast}
|
|
if scenes:
|
|
metadata["scenes"] = [e["name"] for e in scenes]
|
|
metadata["scene_prompts"] = {e["name"]: e["visual_prompt"] for e in scenes}
|
|
if products:
|
|
metadata["product_entities"] = [{"name": e["name"], "prompt": e["visual_prompt"]} for e in products]
|
|
metadata["script_entities"] = entities # 全量(含 ref_index),供故事板多锚点参考
|
|
project.metadata = metadata
|
|
project.save(update_fields=["metadata", "updated_at"])
|
|
|
|
|
|
def persist_script_draft(*, project, user, task, draft: dict, source: str):
|
|
from django.db import transaction
|
|
|
|
from apps.projects.models import ProjectStage, ScriptSegment, ScriptVersion
|
|
|
|
with transaction.atomic():
|
|
script = ScriptVersion.objects.create(
|
|
project=project,
|
|
task=task,
|
|
title=(draft.get("hook") or "AI 脚本")[:128],
|
|
content=json.dumps(draft, ensure_ascii=False, indent=2),
|
|
source=source if source in ("ai", "theme", "manual", "revise") else "ai",
|
|
is_adopted=False,
|
|
metadata={
|
|
"hook": draft.get("hook", ""),
|
|
"tone": draft.get("tone", ""),
|
|
"aspect_ratio": draft.get("aspect_ratio", "9:16"),
|
|
"total_duration": draft.get("total_duration", 60),
|
|
"segment_count": draft.get("segment_count", 4),
|
|
"entities": draft.get("entities", []),
|
|
},
|
|
)
|
|
for seg in draft["segments"]:
|
|
ScriptSegment.objects.create(
|
|
script_version=script,
|
|
sort_order=seg["index"],
|
|
duration_seconds=seg.get("duration", 15),
|
|
narration=seg.get("narration", ""),
|
|
visual_prompt=seg.get("visual", ""),
|
|
role=seg.get("role", ""),
|
|
speaker=seg.get("speaker") or "",
|
|
product_exposure=seg.get("product_exposure", ""),
|
|
entity_refs=seg.get("entity_refs") or [],
|
|
dialogue=seg.get("dialogue") or [],
|
|
product_points=[],
|
|
)
|
|
_map_entities_to_project_metadata(project, draft.get("entities", []))
|
|
stage, _ = ProjectStage.objects.get_or_create(project=project, stage=ProjectStage.Stage.SCRIPT)
|
|
stage.status = ProjectStage.Status.NEEDS_REVIEW
|
|
stage.save(update_fields=["status", "updated_at"])
|
|
return script
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# 流式编排
|
|
# --------------------------------------------------------------------------- #
|
|
def _sse(obj: dict) -> str:
|
|
return f"data: {json.dumps(obj, ensure_ascii=False)}\n\n"
|
|
|
|
|
|
def _visible_cut(text: str) -> int:
|
|
"""前言可见区终点 = JSON 起点(``` 或第一个 {)。之后的内容不外露,只在后端解析。"""
|
|
cands = []
|
|
for marker in ("```", "{"):
|
|
i = text.find(marker)
|
|
if i != -1:
|
|
cands.append(i)
|
|
return min(cands) if cands else len(text)
|
|
|
|
|
|
def _closing_summary(raw: str) -> str:
|
|
"""模型在 json 之后写的收尾交付语 = 给用户的回复气泡(像同事汇报「这版做了啥/怎么接着调」)。
|
|
取最后一个 json 对象之后的文字,去掉收尾围栏与空白;没写或残留花括号就返回空(前端兜底)。"""
|
|
js = _extract_json(raw)
|
|
tail = ""
|
|
if js:
|
|
idx = raw.rfind(js)
|
|
if idx != -1:
|
|
tail = raw[idx + len(js):]
|
|
tail = re.sub(r"`+", " ", tail).strip() # 去掉 json 收尾的 ``` 围栏
|
|
if "{" in tail or len(tail) < 4:
|
|
return ""
|
|
return tail[:160]
|
|
|
|
|
|
def stream_script_agent(
|
|
*,
|
|
project,
|
|
user,
|
|
model_config: ModelConfig,
|
|
mode: str = "auto",
|
|
user_prompt: str = "",
|
|
selling_point_ids: list[str] | None = None,
|
|
base_version_id: str | None = None,
|
|
aspect_ratio: str = "9:16",
|
|
total_duration: int = 60,
|
|
target_index: int | None = None,
|
|
):
|
|
"""生成 SSE 帧字符串的同步生成器,供 StreamingHttpResponse 包裹。
|
|
target_index 非空 = 精准只改第 N 镜(读全脚本上下文,后端强制保留其余镜原样)。"""
|
|
from apps.ai.services import create_ai_task, stream_routed_text_request
|
|
|
|
yield _sse({"type": "tool", "id": "skill", "label": "加载电商脚本技能", "status": "running"})
|
|
skill_loaded = bool(load_ecommerce_skill())
|
|
yield _sse({"type": "tool", "id": "skill", "status": "done" if skill_loaded else "error"})
|
|
|
|
yield _sse({"type": "tool", "id": "analyze", "label": f"分析商品:{project.product.title}", "status": "running"})
|
|
base_draft = None
|
|
if mode == "revise" and base_version_id:
|
|
base_draft = _load_base_draft(project, base_version_id)
|
|
if base_draft is None:
|
|
target_index = None # 没有基准稿就退回整版生成,单镜改无从谈起
|
|
# 改稿以基准稿的时长/镜数为准,避免请求侧默认值(前端可能硬编码 60)把 90s/6镜稿的尾镜挤掉
|
|
effective_duration = (base_draft.get("total_duration") if base_draft else None) or total_duration
|
|
# 精准改一镜:镜号越界直接报错返回,绝不建任务/扣费(避免计费空转的静默 no-op)
|
|
if target_index is not None and base_draft is not None:
|
|
seg_n = len(base_draft.get("segments", []))
|
|
if not (0 <= target_index < seg_n):
|
|
yield _sse({"type": "error", "detail": f"镜号越界:第 {target_index + 1} 镜(共 {seg_n} 镜)"})
|
|
return
|
|
messages = build_agent_messages(
|
|
project=project,
|
|
mode=mode,
|
|
user_prompt=user_prompt,
|
|
selling_point_ids=selling_point_ids,
|
|
base_draft=base_draft,
|
|
aspect_ratio=aspect_ratio,
|
|
total_duration=effective_duration, # 改稿用基准稿时长,prompt head 才不会误导模型镜数(否则模型按60s只出4镜)
|
|
target_index=target_index,
|
|
)
|
|
yield _sse({"type": "tool", "id": "analyze", "status": "done"})
|
|
|
|
task_type = AITask.Type.SCRIPT_OPTIMIZATION if mode == "revise" else AITask.Type.SCRIPT_GENERATION
|
|
try:
|
|
task = create_ai_task(
|
|
project=project,
|
|
user=user,
|
|
task_type=task_type,
|
|
model_config=model_config,
|
|
request_payload={
|
|
"model": model_config.name,
|
|
"endpoint": model_config.endpoint,
|
|
"mode": mode,
|
|
"aspect_ratio": aspect_ratio,
|
|
"total_duration": total_duration,
|
|
"base_version_id": str(base_version_id or ""),
|
|
"target_index": target_index,
|
|
"model_routing_v1": True,
|
|
},
|
|
)
|
|
except Exception as exc: # noqa: BLE001 — 多为额度不足
|
|
internal_kind = "user_credit_insufficient" if str(exc).strip().lower() == "insufficient credit" else ""
|
|
yield _sse(_script_error_event(exc, internal_kind=internal_kind))
|
|
return
|
|
reservation = task.credit_reservation
|
|
# 额度是否已结算(charge 成功 / release 失败)。客户端中途断连时,生成器被 .close() 抛
|
|
# GeneratorExit —— 它是 BaseException 不是 Exception,普通 except 抓不到,会让预扣额度冻结。
|
|
# 故用 try/finally 兜底:任何未结算路径(含断连)都释放预扣。
|
|
settled = False
|
|
try:
|
|
yield _sse({"type": "tool", "id": "generate", "label": "按黄金结构生成分镜", "status": "running"})
|
|
full: list[str] = []
|
|
shown = 0
|
|
forwarding = True
|
|
try:
|
|
task.status = AITask.Status.SUBMITTED
|
|
task.submitted_at = timezone.now()
|
|
task.save(update_fields=["status", "submitted_at", "updated_at"])
|
|
def validate_script_text(raw_text: str) -> dict:
|
|
candidate = normalize_draft(
|
|
raw_text,
|
|
aspect_ratio=aspect_ratio,
|
|
total_duration=effective_duration,
|
|
)
|
|
if target_index is not None and base_draft:
|
|
return _merge_single_segment(
|
|
base_draft,
|
|
candidate,
|
|
target_index,
|
|
aspect_ratio,
|
|
effective_duration,
|
|
)
|
|
return candidate
|
|
|
|
routed_stream = stream_routed_text_request(
|
|
task=task,
|
|
primary_model=model_config,
|
|
messages=messages,
|
|
streaming=True,
|
|
structured_output=True,
|
|
business_operation="script_generate",
|
|
temperature=0.85,
|
|
validate_text=validate_script_text,
|
|
request_summary={
|
|
"mode": mode,
|
|
"target_index": target_index,
|
|
"base_version_id": str(base_version_id or ""),
|
|
"aspect_ratio": aspect_ratio,
|
|
"total_duration": effective_duration,
|
|
},
|
|
)
|
|
while True:
|
|
try:
|
|
ev = next(routed_stream)
|
|
except StopIteration as completed:
|
|
routed = completed.value
|
|
break
|
|
et = ev.get("type")
|
|
if et == "reasoning":
|
|
# 思考流:推理模型在出 JSON 前会先想很久,把思考逐字下发给前端(像对话一样可见),
|
|
# 不进 full(不是答案正文,纯展示)。这是「卡在生成分镜」假死的根因修复。
|
|
rpiece = ev.get("text") or ""
|
|
if rpiece:
|
|
yield _sse({"type": "reasoning", "text": rpiece})
|
|
continue
|
|
if et == "delta":
|
|
full.append(ev["text"])
|
|
if forwarding:
|
|
text = "".join(full)
|
|
cut = _visible_cut(text)
|
|
if cut < len(text):
|
|
forwarding = False
|
|
visible = text[:cut]
|
|
if len(visible) > shown:
|
|
piece = visible[shown:]
|
|
shown = len(visible)
|
|
if piece.strip():
|
|
yield _sse({"type": "delta", "text": piece})
|
|
elif et == "done":
|
|
continue
|
|
raw, _provider_response, draft = routed.value
|
|
except Exception as exc: # noqa: BLE001
|
|
_fail_task(task, reservation, str(exc))
|
|
settled = True
|
|
yield _sse({"type": "tool", "id": "generate", "status": "error"})
|
|
yield _sse(_script_error_event(exc, reference_id=str(task.id)))
|
|
return
|
|
|
|
yield _sse({"type": "tool", "id": "generate", "status": "done"})
|
|
yield _sse(
|
|
{
|
|
"type": "tool",
|
|
"id": "extract",
|
|
"label": f"提取实体 {len(draft['entities'])} 个 · {len(draft['segments'])} 镜",
|
|
"status": "done",
|
|
}
|
|
)
|
|
yield _sse({"type": "tool", "id": "check", "label": "自检:镜数 / ≤55字 / 违规词", "status": "done"})
|
|
yield _sse({"type": "draft", "draft": draft})
|
|
|
|
from django.db import transaction
|
|
|
|
try:
|
|
with transaction.atomic():
|
|
task.status = AITask.Status.SUCCEEDED
|
|
task.response_payload = {"raw": raw[:8000]}
|
|
task.actual_cost = task.estimated_cost
|
|
task.completed_at = timezone.now()
|
|
task.save(update_fields=["status", "response_payload", "actual_cost", "completed_at", "updated_at"])
|
|
charge_reserved_credit(reservation=reservation, actual_amount=task.actual_cost)
|
|
source = "revise" if mode == "revise" else ("theme" if mode == "theme" else "ai")
|
|
script = persist_script_draft(project=project, user=user, task=task, draft=draft, source=source)
|
|
settled = True # charge 已提交
|
|
except Exception as exc: # noqa: BLE001 — 落库失败:atomic 已回滚 charge,补释放预留
|
|
_fail_task(task, reservation, f"保存脚本失败:{exc}")
|
|
settled = True
|
|
yield _sse(_script_error_event(exc, reference_id=str(task.id), internal_kind="processing_failed"))
|
|
return
|
|
|
|
from apps.projects.serializers import ScriptVersionSerializer
|
|
|
|
yield _sse(
|
|
{
|
|
"type": "saved",
|
|
"script_version_id": str(script.id),
|
|
"version": ScriptVersionSerializer(script).data,
|
|
}
|
|
)
|
|
# 模型自己写的收尾交付语 → AI 回复气泡(没写则前端兜底默认句)
|
|
summary = _closing_summary(raw)
|
|
if summary:
|
|
yield _sse({"type": "summary", "text": summary})
|
|
yield _sse({"type": "done"})
|
|
finally:
|
|
# 断连(GeneratorExit)或任何 settled=False 的退出路径:释放预扣,避免额度冻结
|
|
if not settled:
|
|
_fail_task(task, reservation, "stream aborted (client disconnected)")
|
|
|
|
|
|
def _fail_task(task, reservation, message: str) -> None:
|
|
try:
|
|
task.status = AITask.Status.FAILED
|
|
task.error_message = message[:2000]
|
|
task.completed_at = timezone.now()
|
|
task.save(update_fields=["status", "error_message", "completed_at", "updated_at"])
|
|
finally:
|
|
try:
|
|
release_credit(reservation=reservation, reason=message[:200])
|
|
except Exception: # noqa: BLE001
|
|
pass
|
|
|
|
|
|
def _script_error_event(
|
|
exc: Exception,
|
|
*,
|
|
reference_id: str | None = None,
|
|
internal_kind: str = "",
|
|
) -> dict:
|
|
"""脚本 SSE 的安全失败帧；原始异常只落任务记录/日志，不回传普通用户。"""
|
|
public_error = classify_generation_error(
|
|
exc,
|
|
operation="script_generate",
|
|
internal_kind=internal_kind,
|
|
reference_id=reference_id,
|
|
)
|
|
return {
|
|
"type": "error",
|
|
"detail": public_error.fallback_message,
|
|
"error": public_error.as_dict(),
|
|
}
|
|
|
|
|
|
def _load_base_draft(project, base_version_id: str) -> dict | None:
|
|
from apps.projects.models import ScriptVersion
|
|
|
|
try:
|
|
version = ScriptVersion.objects.get(project=project, id=base_version_id)
|
|
except (ScriptVersion.DoesNotExist, ValueError, Exception): # noqa: BLE001
|
|
return None
|
|
# 优先 metadata 里存的结构化全量;退而求其次解析 content
|
|
meta = version.metadata or {}
|
|
if meta.get("entities") is not None or meta.get("hook"):
|
|
try:
|
|
return json.loads(version.content)
|
|
except (ValueError, TypeError):
|
|
pass
|
|
try:
|
|
return json.loads(version.content)
|
|
except (ValueError, TypeError):
|
|
return None
|
|
|
|
|
|
def _draft_from_version(version) -> dict:
|
|
"""从 ScriptVersion 的 DB 行(segments + metadata)重建 ScriptDraft —— 比解析可能已 stale 的 content 可靠
|
|
(用户增删/改镜后 content 不一定同步)。total_duration 按真实镜数算,避免 normalize 按 stale 值截/补镜。"""
|
|
meta = version.metadata or {}
|
|
segs = list(version.segments.order_by("sort_order"))
|
|
return {
|
|
"hook": meta.get("hook", ""),
|
|
"tone": meta.get("tone", ""),
|
|
"aspect_ratio": meta.get("aspect_ratio", "9:16"),
|
|
"total_duration": max(int(meta.get("total_duration") or 0), 15 * len(segs)) or 60,
|
|
"segment_count": len(segs),
|
|
"entities": meta.get("entities", []),
|
|
"segments": [
|
|
{
|
|
"index": s.sort_order,
|
|
"duration": s.duration_seconds or 15,
|
|
"role": s.role or "",
|
|
"narration": s.narration or "",
|
|
"speaker": s.speaker or None,
|
|
"visual": s.visual_prompt or "",
|
|
"product_exposure": s.product_exposure or "",
|
|
"entity_refs": s.entity_refs or [],
|
|
"dialogue": s.dialogue or [],
|
|
}
|
|
for s in segs
|
|
],
|
|
}
|
|
|
|
|
|
def regenerate_segment_via_agent(*, project, user, model_config: ModelConfig, segment, instruction: str = ""):
|
|
"""非流式·精准改一镜(「场次刷新」按钮复用 agent 单镜逻辑):读全脚本上下文,只重写该镜,落新 ScriptVersion。
|
|
与 stream_script_agent 的 target_index 分支同源,但同步返回(不走 SSE)。计费 reserve→charge/release 闭环。"""
|
|
from django.db import transaction
|
|
|
|
from apps.ai.services import create_ai_task, execute_routed_text_request
|
|
from apps.billing.services.ledger import charge_reserved_credit
|
|
|
|
base_draft = _draft_from_version(segment.script_version) # 用 DB 行重建基准,别用可能 stale 的 content
|
|
target_index = segment.sort_order
|
|
aspect_ratio = (base_draft.get("aspect_ratio") or "9:16").strip()
|
|
seg_n = len(base_draft.get("segments") or [])
|
|
if not (0 <= target_index < seg_n):
|
|
raise ValueError(f"镜号越界:第 {target_index + 1} 镜(共 {seg_n} 镜)")
|
|
total_duration = base_draft["total_duration"] # 已按真实镜数算,normalize 不会截掉用户增删后的镜
|
|
|
|
messages = build_agent_messages(
|
|
project=project,
|
|
mode="revise",
|
|
user_prompt=instruction,
|
|
selling_point_ids=None,
|
|
base_draft=base_draft,
|
|
aspect_ratio=aspect_ratio,
|
|
total_duration=total_duration,
|
|
target_index=target_index,
|
|
)
|
|
task = create_ai_task(
|
|
project=project,
|
|
user=user,
|
|
task_type=AITask.Type.SCRIPT_OPTIMIZATION,
|
|
model_config=model_config,
|
|
request_payload={
|
|
"model": model_config.name,
|
|
"endpoint": model_config.endpoint,
|
|
"mode": "revise",
|
|
"target_index": target_index,
|
|
"model_routing_v1": True,
|
|
},
|
|
)
|
|
reservation = task.credit_reservation
|
|
# 每条真实尝试的平台成本由统一执行器累计；用户积分仍只结算这一条脚本任务。
|
|
task.base_cost = Decimal("0")
|
|
task.save(update_fields=["base_cost", "updated_at"])
|
|
# 实际平台成本由每条 AIModelAttempt 累加；用户积分仍只结算这一条逻辑任务。
|
|
task.base_cost = Decimal("0")
|
|
task.save(update_fields=["base_cost", "updated_at"])
|
|
try:
|
|
task.status = AITask.Status.SUBMITTED
|
|
task.submitted_at = timezone.now()
|
|
task.save(update_fields=["status", "submitted_at", "updated_at"])
|
|
def validate_segment_text(raw_text: str) -> dict:
|
|
candidate = normalize_draft(raw_text, aspect_ratio=aspect_ratio, total_duration=total_duration)
|
|
return _merge_single_segment(base_draft, candidate, target_index, aspect_ratio, total_duration)
|
|
|
|
routed = execute_routed_text_request(
|
|
task=task,
|
|
primary_model=model_config,
|
|
messages=messages,
|
|
streaming=False,
|
|
structured_output=True,
|
|
business_operation="script_generate",
|
|
temperature=0.3,
|
|
validate_text=validate_segment_text,
|
|
request_summary={
|
|
"mode": "revise",
|
|
"target_index": target_index,
|
|
"base_version_id": str(segment.script_version_id),
|
|
},
|
|
)
|
|
raw, _response, draft = routed.value
|
|
with transaction.atomic():
|
|
task.status = AITask.Status.SUCCEEDED
|
|
task.response_payload = {"raw": raw[:8000]}
|
|
task.actual_cost = task.estimated_cost
|
|
task.completed_at = timezone.now()
|
|
task.save(update_fields=["status", "response_payload", "actual_cost", "completed_at", "updated_at"])
|
|
charge_reserved_credit(reservation=reservation, actual_amount=task.actual_cost)
|
|
script = persist_script_draft(project=project, user=user, task=task, draft=draft, source="revise")
|
|
return script
|
|
except Exception as exc:
|
|
_fail_task(task, reservation, str(exc) or "单镜重跑失败")
|
|
raise
|