Files
yingqing/core/backend/apps/ai/services.py
T
zycandClaude Opus 4.8 6e7425048c fix: PMC 测试清单未完成项 + YYX#row22 未读角标
PMC#16 账户登录地址不正确:
- 写死的 https://airshelf.com/login(不存在的域名)改成 ${location.origin}/login,
  与邀请链接同源,成员拿到的就是当前真实访问地址。

PMC#17/#18 成员三档额度只有月度生效、日/总纯装饰 → 列表显示与配置对不上:
- TeamMember 新增 daily_credit_limit / total_credit_limit(0=不限,迁移 0006);
- reserve_credit 逐档真实管控(每日/每月/累计总额),口径=窗口内 CHARGE 流水 + 在途预留;
- 成员列表"每日额度"由写死"不限"改成真实日额度+今日已用,并新增"总额度"列;
- 创建/编辑弹窗三档额度真入库真生效;补 3 个计费单测(日/总拦截、0=不限)。

PMC#24 同批生成的图片文字/内容雷同:
- 根因=套图/自由创作同批 N 张用同一 prompt 且无变化(只有模特上身图按 index 变);
- 给平台套图/自由参考/纯文生图都加按 index 的版式变体,避开传 seed(防中转站 400)。

PMC#25 重跑应在原任务上重跑而非新开任务:
- 全批"重跑"改为原地复位同一批次卡;"重跑这张/再生成"改为补图进原卡(不覆盖好图、不新开任务)。

YYX#row22 图片生成未读角标(本仓库既有改动一并提交):
- AITask 新增 read_at(迁移 0020 + 索引一致性迁移 0021);导航栏"图片生成"未读胶囊、
  商品预览右下角未读分数;查看某商品即标记其生成任务已读、清零角标。

注:PMC#14/#15/#20/#22/#23 早前提交已修,需部署重测;PMC#13/#21 为 provider 侧
生成失败,代码侧重试/退费/僵尸回收/错误外露已就位,需部署后看真实 error_message 定位;
PMC#19 收件箱标记与设计稿逐字一致,待截图复现。

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-06-27 18:27:15 +08:00

2371 lines
126 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
import json
import math
import re
import subprocess
import tempfile
import time
import uuid
from datetime import timedelta
from decimal import Decimal
from io import BytesIO
from pathlib import Path
import requests
from django.conf import settings
from django.core.exceptions import ObjectDoesNotExist
from django.db import transaction
from django.utils import timezone
from apps.ai.models import AITask, ModelConfig
from apps.ai.providers import (
OpenAICompatibleProvider,
TtsNotConfigured,
VolcanoArkProvider,
VolcanoTtsProvider,
)
from apps.assets.models import Asset, AssetFile
from apps.assets.storage import TosStorage
from apps.billing.services.ledger import charge_reserved_credit, release_credit, reserve_credit
from apps.projects.models import (
BaseAssetGroup,
ExportJob,
ProjectStage,
ScriptSegment,
ScriptVersion,
StoryboardFrame,
StoryboardShot,
StoryboardShotVersion,
StoryboardVersion,
Timeline,
VideoSegment,
VideoSegmentVersion,
)
def get_default_model(capability: str) -> ModelConfig:
qs = (
ModelConfig.objects.select_related("provider")
.filter(capability=capability, status=ModelConfig.Status.ACTIVE, provider__status="active")
)
# 优先平台超管钦定的默认模型;未钦定则回落「最早创建的 active」(原行为,零回归)
return qs.filter(is_default=True).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(取最新一版)
· "gpt-image"→ gpt-image-2(优先 active provider 的那个)
· "provider:name" 或裸 name → 精确匹配
解析不到返回 None,调用方回落 get_default_model。"""
if not key:
return None
qs = ModelConfig.objects.select_related("provider").filter(capability=ModelConfig.Capability.IMAGE)
if key == "volcano":
# 火山 Seedream:优先版本号最高的(seedream-5 > seedream-4),按 name 倒序
vqs = qs.filter(provider__name__in=OFFICIAL_DIRECT_PROVIDERS)
return vqs.filter(name__icontains="seedream").order_by("-name").first() or vqs.order_by("-name").first()
if key in ("gpt-image", "gpt-image-2"):
return (qs.filter(name__icontains="gpt-image", provider__status="active").first()
or qs.filter(name__icontains="gpt-image").first())
if ":" in key:
pname, mname = key.split(":", 1)
return qs.filter(provider__name=pname, name=mname).first()
return qs.filter(name=key).first()
# 火山官方直连(SeeDream 生图 / Seedance 视频 / 豆包文本)走 ARK SDK;其余 provider 一律
# 视为「OpenAI 兼容中转站」走通用适配器。加/换中转站 = DB 加一行 ModelProvider,零改代码。
# 注意:DB 里火山 provider 实际命名为 "volcengine"(豆包),必须包含,否则会被错路由到中转站。
OFFICIAL_DIRECT_PROVIDERS = {"volcengine", "volcano", "ark", "volcano_ark"}
def resolve_provider_credentials(provider) -> tuple[str | None, str | None]:
"""解析中转站凭证。可插拔顺序:DB(ModelProvider.base_url/api_key)优先 → settings(.env)回退。
两者都不写死;换站只改 DB 这一行,或改 .env 对应项。"""
base_url = (provider.base_url or "").strip() or settings.PROVIDER_BASE_URLS.get(provider.name)
api_key = (getattr(provider, "api_key", "") or "").strip() or settings.PROVIDER_KEYS.get(provider.name)
return (base_url or None), (api_key or None)
def build_provider(model_config: ModelConfig):
"""按 provider.name 分流:火山官方直连 → VolcanoArkProvider;其余 → 通用 OpenAICompatibleProvider。
两条路都走 resolve_provider_credentials,统一 DB→.env 优先级(官方直连的 None 再由 __post_init__ 回退 settings.VOLCANO)。"""
provider = model_config.provider
base_url, api_key = resolve_provider_credentials(provider)
if provider.name in OFFICIAL_DIRECT_PROVIDERS:
# 临时:视频(Seedance)借 AirDrama 账号的 ARK key,使真人素材库 asset:// 引用与 Seedance 同账号、可解析。
# 仅覆盖 VIDEO,图像/文本仍用自有 key。VIDEO_ARK_API_KEY 留空则不覆盖(回落默认)。待自有账号开通素材库后清掉。
video_key = (getattr(settings, "VIDEO_ARK_API_KEY", "") or "").strip()
if video_key and model_config.capability == ModelConfig.Capability.VIDEO:
api_key = video_key
return VolcanoArkProvider(base_url=base_url, api_key=api_key)
# api_version:某些中转站(yunqi)的 images/edits 需 Azure 风格 ?api-version=...。
# DB(ModelProvider.metadata.api_version)优先 → settings(.env)回退,与凭证同款可插拔。
api_version = (provider.metadata or {}).get("api_version") or settings.PROVIDER_API_VERSIONS.get(provider.name)
return OpenAICompatibleProvider(base_url=base_url, api_key=api_key, api_version=api_version or None)
def get_image_provider(model_config: ModelConfig):
return build_provider(model_config)
def get_text_provider(model_config: ModelConfig):
return build_provider(model_config)
def get_video_provider(model_config: ModelConfig):
return build_provider(model_config)
def estimate_cost(model_config: ModelConfig) -> Decimal:
return model_config.unit_price if model_config.unit_price > 0 else Decimal("1.0000")
def parse_segment_fields(block: str) -> tuple[str, str]:
"""从一镜文本里拆出(旁白, 画面)。
模型按 build_script_prompt 的格式输出「旁白:…/画面:…」标签行时精确拆分;
自带脚本/旧格式没有标签则两个字段都用整段(保持旧行为),字幕/故事板各自兜底。
"""
narration_lines: list[str] = []
visual_lines: list[str] = []
current: list[str] | None = None
for raw in (block or "").splitlines():
line = raw.strip()
if not line:
continue
matched = re.match(r"^(旁白|口播|台词|文案)\s*[:]\s*(.*)$", line)
if matched:
current = narration_lines
if matched.group(2):
current.append(matched.group(2))
continue
matched = re.match(r"^(画面|镜头描述|视觉|画面描述)\s*[:]\s*(.*)$", line)
if matched:
current = visual_lines
if matched.group(2):
current.append(matched.group(2))
continue
if re.match(r"^(镜头|分镜|场)\s*\d+", line):
continue # 「镜头N」标题行不计入任何字段
if current is not None:
current.append(line)
narration = " ".join(narration_lines).strip()
visual = " ".join(visual_lines).strip()
if not narration and not visual:
return block.strip(), block.strip()
return narration or visual, visual or narration
def build_cast_scene_extract_prompt(content: str) -> list[dict[str, str]]:
"""轻量抽取提示词:从镜头脚本里提炼人物 / 场景标签,并给每个标签一句可直接生图的画面提示词。"""
system = (
"你是短视频脚本分析助手。请从给定的镜头脚本中提取出现的『人物』和『场景』,"
"并为每个人物 / 场景写一句可直接用于文生图的画面提示词(中文,30 字内,描述外形 / 着装 / 环境 / 光线)。"
"人物提示词用 9:16 竖屏(出镜角色全身);场景提示词用 16:9 横屏(环境 / 背景空镜)。"
"人物指出镜的角色(例:女主、同事、闺蜜);场景指画面发生的地点或环境(例:卫生间、地铁、办公室)。"
"去重,人物与场景各最多 6 个。只输出一个 JSON 对象,不要 markdown 代码块,不要任何额外文字,格式如下:\n"
'{"cast":[{"name":"女主","prompt":"26岁都市女性,自然妆容,米色针织衫,柔和室内光,9:16竖屏"}],'
'"scenes":[{"name":"卫生间","prompt":"现代简约浴室,暖色灯光,干净台面,16:9横屏"}]}'
)
return [{"role": "system", "content": system}, {"role": "user", "content": f"镜头脚本如下:\n{content}".strip()}]
def _coerce_tag_entries(items: object, limit: int = 6) -> tuple[list[str], dict[str, str]]:
"""把模型回的 [{"name","prompt"}] 列表整理成 (标签列表, {标签: 提示词}),去重保序、容错。"""
names: list[str] = []
prompts: dict[str, str] = {}
if not isinstance(items, list):
return names, prompts
for item in items:
if isinstance(item, dict):
name = str(item.get("name") or "").strip()
prompt = str(item.get("prompt") or "").strip()
else:
name, prompt = str(item or "").strip(), ""
if not name or name in ("无", "暂无", "未提及") or name in names:
continue
names.append(name)
if prompt:
prompts[name] = prompt
if len(names) >= limit:
break
return names, prompts
def extract_cast_and_scenes(*, project, user, content: str) -> dict:
"""轻量调一次文本模型,从脚本里抽取人物 / 场景标签及每个标签的建议生图提示词。
出稿后的增益步骤(对齐流程文档「自动从脚本提取信息」),走完整的 AITask + 计费闭环
(reserve→charge/release,任务类型记为 script_optimization)。但全程 best-effort:
无可用模型 / 预扣失败 / 调用失败 / 解析失败都吞掉返回空,绝不阻断脚本生成主流程。
返回 {cast, scenes, cast_prompts, scene_prompts}。
"""
empty = {"cast": [], "scenes": [], "cast_prompts": {}, "scene_prompts": {}}
model_config = get_default_model(ModelConfig.Capability.TEXT)
if model_config is None or not (content or "").strip():
return empty
messages = build_cast_scene_extract_prompt(content)
try:
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, "messages": messages},
)
except Exception:
return empty # 余额不足等预扣失败:跳过提取,不挡出稿
reservation = task.credit_reservation
try:
task.status = AITask.Status.SUBMITTED
task.submitted_at = timezone.now()
task.save(update_fields=["status", "submitted_at", "updated_at"])
provider = build_provider(model_config)
response = provider.chat_completion(model=model_config.name, endpoint=model_config.endpoint, messages=messages)
text = provider.extract_text(response)
# LLM 调用已真实消耗 token → 按成功计费(无论后续能否解析出标签)
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)
match = re.search(r"\{.*\}", text, re.DOTALL) # 容忍模型多裹了 markdown / 解释文字
if not match:
return empty
data = json.loads(match.group(0))
cast, cast_prompts = _coerce_tag_entries(data.get("cast"))
scenes, scene_prompts = _coerce_tag_entries(data.get("scenes"))
return {"cast": cast, "scenes": scenes, "cast_prompts": cast_prompts, "scene_prompts": scene_prompts}
except Exception as exc:
with transaction.atomic():
task.status = AITask.Status.FAILED
task.error_message = str(exc)
task.completed_at = timezone.now()
task.save(update_fields=["status", "error_message", "completed_at", "updated_at"])
release_credit(reservation=reservation, reason=str(exc))
return empty
def _skills_root() -> "Path":
"""skills 根目录。优先 BASE_DIR/skills(= core/backend/skills,随后端打进 Docker 镜像);
回落仓库根 BASE_DIR.parent.parent/skills(本地/旧布局)。
⚠️ 历史坑:镜像由 `./core/backend` 构建,skills 旧在仓库根 → 不在构建上下文 → 镜像里没有 →
`_load_skill_system_prompt` 返回空串 → 提取拿不到「只输出 JSON」铁律 → 模型吐散文 → 解析失败。
现 skills 已挪进 core/backend 随镜像打包;此处仍保留双路径兜底。"""
from pathlib import Path
from django.conf import settings
base = Path(settings.BASE_DIR)
for cand in (base / "skills", base.parent.parent / "skills"):
if cand.is_dir():
return cand
return base / "skills"
def _load_skill_system_prompt(name: str) -> str:
"""读取 skills/<name>/SKILL.md + references/*.md 拼成系统提示词(领域知识)。缺文件返回空串,不致命。"""
skill_dir = _skills_root() / name
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')}")
return "\n".join(parts)
def _normalize_extracted_entities(items: object) -> list[dict]:
"""整理提取出的 entities:只留 character/scene(绝不收 product),保留模型给的 id(供 segment refs 对应),
按 id 去重,角色/场景各最多 6 个,补 ref_index。"""
out: list[dict] = []
if not isinstance(items, list):
return out
seen_ids: set[str] = set()
n_char = n_scene = 0
for it in items:
if not isinstance(it, dict):
continue
typ = str(it.get("type") or "").strip()
if typ not in ("character", "scene"): # 丢弃 product / 未知类型
continue
name = str(it.get("name") or "").strip()
eid = str(it.get("id") or "").strip()
if not name or not eid or eid in seen_ids:
continue
if typ == "character":
n_char += 1
if n_char > 6:
continue
else:
n_scene += 1
if n_scene > 6:
continue
seen_ids.add(eid)
out.append(
{
"id": eid,
"type": typ,
"name": name,
"visual_prompt": str(it.get("visual_prompt") or "").strip(),
"ref_index": len(out) + 1,
}
)
return out
def _normalize_extracted_segment_refs(items: object, valid_ids: set[str]) -> list[dict]:
"""整理每镜 refs:只留指向存活实体(角色/场景)的 id,丢掉被剔除的(如商品/无效 id)。"""
out: list[dict] = []
if not isinstance(items, list):
return out
for it in items:
if not isinstance(it, dict):
continue
idx = it.get("index")
if not isinstance(idx, int):
continue
raw = it.get("entity_refs") or it.get("refs") or []
refs = [r for r in raw if isinstance(r, str) and r in valid_ids]
out.append({"index": idx, "entity_refs": refs})
return out
# 提取步固定锁定豆包 2.0 Pro(与脚本生成同款),不靠 get_default_model 的「最早创建」排序——
# 避免不同环境 DB 创建序漂移把提取路由到别的(中转站)推理模型。取不到再回落默认文本模型。
EXTRACT_TEXT_MODEL_NAME = "doubao-seed-2-0-pro-260215"
def _resolve_extract_model_config():
"""提取实体用的文本模型:优先豆包 2.0 Pro(active 且 provider active),否则回落默认文本模型。"""
pinned = (
ModelConfig.objects.select_related("provider")
.filter(
name=EXTRACT_TEXT_MODEL_NAME,
capability=ModelConfig.Capability.TEXT,
status=ModelConfig.Status.ACTIVE,
provider__status="active",
)
.order_by("created_at")
.first()
)
return pinned or get_default_model(ModelConfig.Capability.TEXT)
def _collect_extract_text(provider, model_config, messages) -> tuple[str, dict]:
"""走与「脚本生成」同一条已在生产验证稳定的流式通道把模型输出收全。
豆包 seed-pro / GPT / Gemini 等思考模型,思考期只发 reasoning_content、正文期才发 content,
流式分支(chat_completion_stream)已分别转发为 `reasoning` / `delta` 事件。这里**只收正文 delta**;
仅当正文为空(整轮被思考占满的极端兜底)才回退已收到的 reasoning。如此从根上避免非流式只读
content 拿到空串 → 解析不到 JSON 的老问题。返回 (text, 存档用精简 payload)。
"""
content_parts: list[str] = []
reasoning_parts: list[str] = []
for ev in provider.chat_completion_stream(
model=model_config.name,
endpoint=model_config.endpoint,
messages=messages,
temperature=0.3, # 结构化抽取要稳:低温降低 JSON 漂移
):
etype = ev.get("type")
if etype == "delta":
content_parts.append(ev.get("text") or "")
elif etype == "reasoning":
reasoning_parts.append(ev.get("text") or "")
content = "".join(content_parts).strip()
reasoning = "".join(reasoning_parts).strip()
text = content or reasoning
payload = {
"streamed": True,
"model": model_config.name,
"content": content,
"reasoning_chars": len(reasoning),
"reasoning_preview": reasoning[:500], # 失败时看一眼模型在想啥/有没有跑题
"system_chars": sum(len(m.get("content") or "") for m in messages if m.get("role") == "system"),
}
return text, payload
# 在途状态:据此判「已有提取在跑」(提交侧防重复扣费 + 前端刷新后重建 loading)
_EXTRACT_INFLIGHT = (
AITask.Status.CREATED,
AITask.Status.RESERVED,
AITask.Status.SUBMITTED,
AITask.Status.POLLING,
)
# 写死的提取输出契约兜底(对齐脚本 agent 的 _OUTPUT_PROTOCOL):随代码进镜像、永远在。
# **只管"格式骨架"**(JSON 形状 / id 规则 / 不提商品 / 只输出 JSON)—— 这是流水线硬底线,skill 丢了也不会塌。
# **不碰任何"创作细则"**(穿搭/空手/字数/visual_prompt 怎么写),那些归 skill 正文独占,避免两处重复→漂移→稀释质量。
_EXTRACT_OUTPUT_CONTRACT = """
---
## 输出契约(硬性 · 仅格式底线)
**只输出且仅输出一个 JSON 对象**(UTF-8,无注释,无 ```json 代码块外的任何散文/解释/markdown),形状如下:
{
"entities": [
{"id": "c1", "type": "character", "name": "角色称呼", "visual_prompt": "一句生图描述"},
{"id": "s1", "type": "scene", "name": "地点名", "visual_prompt": "一句生图描述"}
],
"segments": [{"index": 0, "entity_refs": ["c1", "s1"]}]
}
硬性规则(**仅格式;角色/场景怎么写、穿搭/visual_prompt 细则一律以上文技能正文为准**):
① 只认 character / scene 两类,**绝不输出 product 商品实体**;② 角色 id 用 c1/c2…、场景 id 用 s1/s2…,严格去重;
③ segments 必须覆盖输入每一镜(index 从 0 开始),列出该镜出现的角色/场景 id;④ 整个回复就是那个 JSON,无前后缀。
"""
def get_inflight_extraction(project):
"""本项目正在跑的实体提取任务(最近一条优先);无则 None。"""
return (
AITask.objects.filter(
project=project, task_type=AITask.Type.ENTITY_EXTRACTION, status__in=_EXTRACT_INFLIGHT
)
.order_by("-created_at")
.first()
)
def submit_extract_entities(*, project, user) -> AITask:
"""提交独立实体提取步(**异步**)。读已定稿(或最新)脚本 → 校验 + 建 RESERVED 任务 + 预留额度(秒级),
慢活(豆包思考模型流式抽取,可达数十秒)交给 Celery worker(run_extract_entities_task)跑。
这样 Web 层(gunicorn/nginx)不被数十秒的模型请求占住 → 不再 502。
**防重复扣费**:本项目已有在途提取任务时直接复用它,不新建/不二次预扣
(用户刷新后手痒重点、或并发点击都安全 —— 提取是实体唯一权威来源,本就只需跑一次)。
校验失败(无脚本/无分镜/无模型/额度不足)抛 ValueError 由端点转 400;运行期失败由 worker 记进
task.error_message,前端轮询 extract-status 读取。返回 AITask。"""
from apps.projects.models import ScriptVersion
from apps.ai.tasks import extract_entities_task
script = (
ScriptVersion.objects.filter(project=project, is_adopted=True).order_by("-created_at").first()
or ScriptVersion.objects.filter(project=project).order_by("-created_at").first()
)
if script is None:
raise ValueError("请先生成并定稿脚本,再提取角色 / 场景")
segments = list(script.segments.order_by("sort_order"))
if not segments:
raise ValueError("脚本没有分镜,无法提取")
inflight = get_inflight_extraction(project)
if inflight is not None:
return inflight # 已有提取在跑:复用,绝不二次预扣 / 重复出活
model_config = _resolve_extract_model_config()
if model_config is None:
raise ValueError("没有可用的文本模型")
product = project.product
if product is not None:
sp = "、".join([p.title for p in product.selling_points.all()[:5]])
prod_line = f"名称:{product.title}\n品类:{product.category or '未填'}\n卖点:{sp or '未填'}"
else:
prod_line = "(无商品信息)"
seg_lines = [
f"镜{i} role={s.role or ''} narration={(s.narration or '').strip()} visual={(s.visual_prompt or '').strip()}"
for i, s in enumerate(segments)
]
user_msg = (
f"商品信息:\n{prod_line}\n\n分镜脚本(共 {len(segments)} 镜,index 从 0 开始):\n" + "\n".join(seg_lines)
)
# skill 正文(领域功力)+ 写死的输出契约兜底(保证「只输出 JSON」永远在,即便 skill 丢失)
system = _load_skill_system_prompt("ecommerce-entity-extract") + _EXTRACT_OUTPUT_CONTRACT
messages = [{"role": "system", "content": system}, {"role": "user", "content": user_msg}]
try:
task = create_ai_task(
project=project,
user=user,
task_type=AITask.Type.ENTITY_EXTRACTION,
model_config=model_config,
request_payload={
"model": model_config.name,
"endpoint": model_config.endpoint,
"messages": messages,
"script_id": str(script.id),
},
)
except Exception as exc: # 余额不足等预扣失败
raise ValueError("额度不足,无法提取(请先充值)") from exc
extract_entities_task.delay(str(task.id))
return task
def run_extract_entities_task(*, task_id: str) -> None:
"""Celery worker 内执行实体提取慢活:流式调模型 → 解析 JSON → 落库(覆盖 project.metadata 的
cast/scenes/*_prompts/script_entities + 回填每镜 entity_refs + entities_extracted 标记)并扣费;
失败退费并把可读错误记进 task.error_message(前端轮询 extract-status 读取)。
幂等:只处理 RESERVED 任务,重复投递不会二次出活、二次扣费。"""
from apps.projects.models import ScriptVersion
from apps.ai.script_agent import _map_entities_to_project_metadata
task = AITask.objects.select_related("team", "created_by", "project", "model_config").filter(id=task_id).first()
if task is None or task.status != AITask.Status.RESERVED:
return
project = task.project
payload = task.request_payload or {}
messages = payload.get("messages") or []
model_config = task.model_config
reservation = task.credit_reservation
response: dict = {} # 模型返回的精简存档;失败时也落库(便于事后定位"模型到底吐了啥")
try:
task.status = AITask.Status.SUBMITTED
task.submitted_at = timezone.now()
task.save(update_fields=["status", "submitted_at", "updated_at"])
provider = build_provider(model_config)
text, response = _collect_extract_text(provider, model_config, messages)
match = re.search(r"\{.*\}", text, re.DOTALL)
if not match:
raise ValueError("提取结果解析失败(模型未返回有效 JSON),请重试")
try:
data = json.loads(match.group(0))
except json.JSONDecodeError as exc: # 抓出来给用户可读话术,别把裸解析报错塞进 error_message
raise ValueError("提取结果解析失败,请重试") from exc
entities = _normalize_extracted_entities(data.get("entities"))
if not entities:
raise ValueError("没有从脚本里识别到角色 / 场景,可调整脚本后重试")
seg_refs = _normalize_extracted_segment_refs(data.get("segments"), {e["id"] for e in entities})
script_id = payload.get("script_id")
script = ScriptVersion.objects.filter(id=script_id).first() if script_id else None
if script is None: # 极端兜底:payload 里 script 没了就回落项目当前定稿稿
script = (
ScriptVersion.objects.filter(project=project, is_adopted=True).order_by("-created_at").first()
or ScriptVersion.objects.filter(project=project).order_by("-created_at").first()
)
segments = list(script.segments.order_by("sort_order")) if script is not None else []
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)
# 落库:覆盖 project.metadata + 回填每镜 entity_refs(提取为实体唯一权威来源)。
# entities_extracted 标记:前端资产页提取闸门据此显隐(没走过 → 盖蒙版露按钮;走过 → 露卡片)。
_map_entities_to_project_metadata(project, entities)
md_flag = dict(project.metadata or {})
md_flag["entities_extracted"] = True
project.metadata = md_flag
project.save(update_fields=["metadata", "updated_at"])
refs_by_index = {r["index"]: r["entity_refs"] for r in seg_refs}
for i, seg in enumerate(segments):
new_refs = refs_by_index.get(i, [])
if (seg.entity_refs or []) != new_refs:
seg.entity_refs = new_refs
seg.save(update_fields=["entity_refs", "updated_at"])
except Exception as exc: # noqa: BLE001 — 失败退费并把错误记进 AITask 供前端轮询;不向上抛(避免 celery 重试二次扣费)
# ValueError 是我们给用户写好的可读话术(解析失败 / 没识别到角色等);其余(网络/模型异常)给通用话术。
msg = str(exc) if isinstance(exc, ValueError) else "提取调用失败,请重试"
with transaction.atomic():
task.status = AITask.Status.FAILED
task.error_message = msg
task.response_payload = response # 即便失败也存下模型输出片段(content/reasoning 长度等),供事后定位
task.completed_at = timezone.now()
task.save(update_fields=["status", "response_payload", "error_message", "completed_at", "updated_at"])
release_credit(reservation=reservation, reason=msg)
def split_script_into_segments(content: str, count: int = 4) -> list[str]:
"""把一段脚本稳健地拆成 `count` 个分镜文本,保证每镜都非空、且所有内容都被分配到某一镜。
原实现按行 `[:4]`,ARK 返回整段散文时常变成「第1镜有词、2/3/4镜全空」,
导致后续故事板帧 / 视频段拿到空提示词,前后内容断裂。这里改为:
优先按空行/标号块切,块数够就把全部块均匀分桶;块不够再按句子切;仍不够则补齐。
"""
def _bucketize(items: list[str], joiner: str) -> list[str]:
buckets: list[list[str]] = [[] for _ in range(count)]
per = len(items) / count
for index, item in enumerate(items):
buckets[min(count - 1, int(index / per))].append(item)
return [joiner.join(bucket).strip() for bucket in buckets]
text = (content or "").strip()
if not text:
return [""] * count
# 1) 优先按空行分段;只有一段时退回按行分
blocks = [block.strip() for block in re.split(r"\n\s*\n", text) if block.strip()]
if len(blocks) < 2:
blocks = [line.strip() for line in text.splitlines() if line.strip()]
if len(blocks) >= count:
return _bucketize(blocks, "\n")
# 2) 段落不足:按中英文句末标点切句,再均匀分桶
sentences = [s.strip() for s in re.split(r"(?<=[。!?!?.;\n])", text) if s.strip()]
if len(sentences) >= count:
return _bucketize(sentences, " ")
# 3) 仍不足:用已有块/句补齐到 count,绝不留空镜
base = blocks or sentences or [text]
filled = list(base)
while len(filled) < count:
filled.append(base[-1])
return filled[:count]
@transaction.atomic
def create_ai_task(*, project, user, task_type: str, model_config: ModelConfig, request_payload: dict) -> AITask:
cost = estimate_cost(model_config)
task = AITask.objects.create(
team=project.team,
created_by=user,
project=project,
task_type=task_type,
status=AITask.Status.CREATED,
model_config=model_config,
idempotency_key=f"{task_type}:{project.id}:{uuid.uuid4()}",
request_payload=request_payload,
estimated_cost=cost,
)
reserve_credit(team=project.team, user=user, task=task, amount=cost)
task.status = AITask.Status.RESERVED
task.save(update_fields=["status", "updated_at"])
return task
def regenerate_script_segment(*, project, user, segment, instruction: str = "") -> ScriptVersion:
"""单镜重跑(「场次刷新」按钮):复用脚本 agent 的精准改一镜——读全脚本上下文、只动该镜、保留其余镜,落新 ScriptVersion。
旧的「散文 prompt + 正则解析」整条已废,统一走 agent(结构化 + entity_refs 不丢)。"""
from apps.ai.script_agent import regenerate_segment_via_agent
model_config = get_default_model(ModelConfig.Capability.TEXT)
if model_config is None:
raise ValueError("no active text model configured")
return regenerate_segment_via_agent(
project=project, user=user, model_config=model_config, segment=segment, instruction=instruction
)
def _generate_video_poster(*, video_bytes: bytes, team, project, asset_id) -> "StoredObject | None":
"""用 ffmpeg 抽视频首帧作为封面(poster)并上传 TOS。best-effort:任何失败都返回 None,不影响视频资产落地。"""
if not video_bytes:
return None
try:
with tempfile.TemporaryDirectory(prefix="airshelf-poster-") as tmp:
tmp_dir = Path(tmp)
video_path = tmp_dir / "in.mp4"
poster_path = tmp_dir / "poster.jpg"
video_path.write_bytes(video_bytes)
proc = subprocess.run(
["ffmpeg", "-y", "-ss", "0", "-i", str(video_path), "-frames:v", "1", "-q:v", "3", str(poster_path)],
capture_output=True,
timeout=60,
)
if proc.returncode != 0 or not poster_path.exists():
return None
poster_bytes = poster_path.read_bytes()
if not poster_bytes:
return None
object_key = f"teams/{team.id}/projects/{project.id}/generated/{asset_id}-poster.jpg"
return TosStorage().upload_fileobj(
fileobj=BytesIO(poster_bytes), object_key=object_key, content_type="image/jpeg"
)
except Exception: # noqa: BLE001 — poster 仅用于展示,失败不阻断
return None
def _store_generated_media(*, team, user, project, task, media: str, name: str, category: str, asset_type: str) -> Asset:
fileobj, content_type = VolcanoArkProvider.media_to_bytes(media)
suffix = ".png"
if "video" in content_type:
suffix = ".mp4"
elif "jpeg" in content_type:
suffix = ".jpg"
elif "webp" in content_type:
suffix = ".webp"
asset_id = uuid.uuid4()
object_key = f"teams/{team.id}/projects/{project.id}/generated/{asset_id}{suffix}"
stored = TosStorage().upload_fileobj(fileobj=fileobj, object_key=object_key, content_type=content_type)
asset = Asset.objects.create(
id=asset_id,
team=team,
created_by=user,
name=name,
asset_type=asset_type,
source=Asset.Source.AI_GENERATED,
category=category,
origin_task=task,
)
AssetFile.objects.create(
asset=asset,
object_key=stored.object_key,
bucket=stored.bucket,
content_type=stored.content_type,
size_bytes=stored.size_bytes,
is_primary=True,
)
# 视频资产:额外抽首帧作为封面图,挂成同一 Asset 下的 image 文件,供任务中心/列表显示缩略图
if "video" in content_type:
try:
video_bytes = fileobj.getvalue() if isinstance(fileobj, BytesIO) else b""
except Exception: # noqa: BLE001
video_bytes = b""
poster = _generate_video_poster(video_bytes=video_bytes, team=team, project=project, asset_id=asset_id)
if poster:
AssetFile.objects.create(
asset=asset,
object_key=poster.object_key,
bucket=poster.bucket,
content_type=poster.content_type,
size_bytes=poster.size_bytes,
is_primary=False,
)
return asset
def _find_entity_group(project, kind: str, label: str, group_id: str | None):
"""复用同一实体的基础资产组(版本=candidate_assets,采用=adopted_asset)。
优先 group_id;商品=该项目唯一商品组;人物/场景按 label 命中非三视图组;都没有则 None(新建)。"""
if group_id:
return project.base_asset_groups.filter(id=group_id).first()
candidates = list(project.base_asset_groups.filter(kind=kind).order_by("created_at"))
candidates = [g for g in candidates if not (g.metadata or {}).get("triview_of")] # 三视图组不算实体本体
if kind == BaseAssetGroup.Kind.PRODUCT:
return candidates[0] if candidates else None
lbl = (label or "").strip()
if lbl:
return next((g for g in candidates if (g.metadata or {}).get("label") == lbl), None)
return None
def _ratio_to_image_size(ratio: str) -> str:
"""前端比例 → gpt-image 支持的尺寸(宽高均需被 16 整除)。16:9 用真 16:9 的 1536x864,4:3 用 1536x1024。"""
return {
"1:1": "1024x1024",
"3:4": "1024x1536",
"9:16": "1024x1536",
"4:3": "1536x1024",
"16:9": "1536x864", # 真 16:9(原来误用 1536x1024 = 3:2)
}.get((ratio or "").strip(), "1024x1024")
def _ratio_to_volcano_size(ratio: str) -> str:
"""前端比例 → 火山 Seedream 尺寸(~2K 面积,各边夹在 [1024,4096] 且取 16 的倍数)。
预设比例直接给好尺寸;自定义 W:H 按 2K 面积换算;解析不到回落 '2K'。"""
presets = {
"1:1": "2048x2048",
"3:4": "1728x2304",
"4:3": "2304x1728",
"9:16": "1440x2560",
"16:9": "2560x1440",
# 4:5 必须 ≥ Seedream 5.0 的最小面积 3,686,400 px:旧值 1664x2080=3.46M 会被 ARK 退 400
# (InvalidParameter: image size must be at least 3686400 pixels)。1728x2160 精确 4:5 且 3.73M 达标。
"4:5": "1728x2160",
}
r = (ratio or "").strip()
if r in presets:
return presets[r]
if ":" in r:
try:
w_str, h_str = r.split(":", 1)
w, h = float(w_str), float(h_str)
if w > 0 and h > 0:
scale = math.sqrt((2048 * 2048) / (w * h))
side = lambda v: min(max(int(round(v * scale / 16) * 16), 1024), 4096) # noqa: E731
return f"{side(w)}x{side(h)}"
except (ValueError, ZeroDivisionError):
pass
return "2K"
def _product_cover_url(product) -> str:
"""商品主图 URL:优先 cover_asset,其次标记为主图的商品图,再次首张商品图。无图返回 ''。"""
if product is None:
return ""
if product.cover_asset_id:
url = _asset_preview_url(product.cover_asset)
if url:
return url
image = product.images.filter(is_primary=True).first() or product.images.order_by("sort_order", "created_at").first()
if image is not None:
return _asset_preview_url(image.asset)
return ""
def _product_reference_urls(product, limit: int = 3) -> list[str]:
"""模特上身图的商品参考图(可多张):**真实上传图优先,排除 AI 生成图**——避免拿生成图当真相
再喂回模型造成误差累积。按主图/排序取前 limit 张真实上传图;一张都没有时回落 cover(即便是 AI 图,
至少保证能走 image_edit 而不是纯文生图)。"""
if product is None:
return []
from apps.assets.models import Asset
urls: list[str] = []
seen: set[str] = set()
rels = list(product.images.select_related("asset").all())
rels.sort(key=lambda im: (not im.is_primary, im.sort_order))
for im in rels:
a = im.asset
if a is None or getattr(a, "source", "") == Asset.Source.AI_GENERATED:
continue
u = _asset_preview_url(a)
if u and u not in seen:
seen.add(u)
urls.append(u)
if len(urls) >= limit:
return urls
if not urls: # 无任何真实上传图 → 回落 cover(可能是 AI 图,但好过纯文生图)
cover = _product_cover_url(product)
if cover:
urls.append(cover)
return urls
def quality_words(stage: str, slot: str = "quality") -> list[str]:
"""平台单层质量词配置(QualityWord)。无配置/表不存在 → 返回 [],调用方回落写死值,保证零回归。"""
try:
from apps.ai.models import QualityWord
return list(
QualityWord.objects.filter(stage=stage, slot=slot, enabled=True)
.order_by("sort", "created_at")
.values_list("text", flat=True)
)
except Exception: # noqa: BLE001 — 配置读取失败绝不阻断生成
return []
def quality_suffix(stage: str, fallback: str, slot: str = "quality", sep: str = "") -> str:
"""取该阶段配置的质量词拼成尾串;无配置回落到 fallback(各 builder 的原写死值)。"""
words = quality_words(stage, slot)
return sep.join(words) if words else fallback
_PROMPT_PLACEHOLDER_RE = re.compile(r"\{(\w+)\}", re.UNICODE)
# gpt-image 要求宽高都能被 16 整除;1536x864 = 真 16:9(长边封顶 1536),1536x1024 = 3:2
_RATIO_TO_SIZE = {"1:1": "1024x1024", "portrait": "1024x1536", "landscape": "1536x1024", "16:9": "1536x864"}
def _prompt_template_row(key: str):
"""读 admin 可编辑的提示词模板行(启用的);缺表/缺行/异常一律返回 None(回落写死默认,零回归)。"""
try:
from apps.ai.models import PromptTemplate
return PromptTemplate.objects.filter(key=key, enabled=True).first()
except Exception: # noqa: BLE001 — 表还没迁移/查询异常都不该挡生成
return None
def render_prompt(key: str, default_text: str, **fields) -> str:
"""按 admin 可编辑模板渲染提示词。无模板/停用/正文空 → 回落 default_text(各 builder 的写死值)。
安全替换 {占位符}:已知字段换值,未知占位符**原样保留**(用户改错也不会让生成崩)。占位符名支持中文。"""
row = _prompt_template_row(key)
tpl = row.template if (row and (row.template or "").strip()) else default_text
return _PROMPT_PLACEHOLDER_RE.sub(lambda m: str(fields.get(m.group(1), m.group(0))), tpl)
def prompt_ratio_size(key: str, default_size: str) -> str:
"""模板配的比例 → gpt-image 尺寸(1024x1024 / 1024x1536 / 1536x1024);未配则用 default_size。"""
row = _prompt_template_row(key)
if row and row.ratio:
return _RATIO_TO_SIZE.get(row.ratio, default_size)
return default_size
def build_product_triview_prompt_refs(product, base_prompt: str = "") -> str:
"""商品三视图 image_edit 提示词(refs 版):参考图1=商品真实主图。正文可在 admin「提示词」页改;
{商品}=主图商品名(写死字段,不可删),{补充}=调用方附加文本。"""
name = (getattr(product, "title", "") or "商品").strip()
default = (
"参考图1是「{商品}」的真实商品主图。请严格参照该图的包装外形、品牌文字、配色、Logo 与材质,"
"生成同一件商品的三视图:从左到右依次为正面、侧面、背面,统一光照,纯白背景,16:9 构图。"
"三个视图必须是同一件商品,品牌字样/配色/外形高度一致,不要改动或重新设计包装。{补充}"
)
return render_prompt("product_triview", default, 商品=name, 补充=(base_prompt or "").strip())
# 模特上身图:按品类分「穿戴 / 非穿戴」注入不同的商品-模特关系;每张按序号变化动作/场景/镜头。
_TRYON_WEARABLE_HINTS = ("服", "衣", "裤", "裙", "鞋", "帽", "袜", "围巾", "外套", "卫衣", "内衣", "文胸", "胸罩", "泳", "bra")
_TRYON_VARIATIONS = [
{"action": "模特正面自然展示该商品", "scene": "干净的室内空间", "shot": "半身近景,商品清晰可见"},
{"action": "模特正在穿着 / 使用该商品", "scene": "生活化的真实居家场景", "shot": "侧面角度,突出穿着或使用方式"},
{"action": "模特手持或局部展示商品细节", "scene": "明亮的时尚生活场景", "shot": "中近景,商品占比较高"},
{"action": "模特与商品自然互动", "scene": "温暖时尚的生活场景", "shot": "半身,强调使用情境"},
]
_TRYON_NEGATIVE = (
"不要换人,不要改商品设计,不要改商品颜色与结构,不要生成错误或乱码的 Logo 与文字,"
"不要把商品改成相似款,不要多余的商品堆叠,不要多余文字,不要水印,不要边框,"
"不要低清模糊,不要过度磨皮,不要畸变,不要扭曲身体,不要夸张滤镜"
)
def _is_wearable_product(product) -> bool:
"""据类目/标题判断是否「穿戴类」(服饰鞋帽内衣) → 真实穿到身上;否则非穿戴 → 手持/佩戴/使用。"""
blob = f"{getattr(product, 'category', '') or ''} {getattr(product, 'title', '') or ''}".lower()
return any(h in blob for h in _TRYON_WEARABLE_HINTS)
def build_model_tryon_prompt_refs(product, has_model: bool, base_prompt: str = "", index: int = 0, n_product: int = 1) -> str:
"""模特上身图 image_edit 提示词(refs 版):
参考图1~N=商品真实图(多角度,锁外形/品牌/配色),参考图N+1=选中模特(锁人脸/身形/气质)。
按品类分穿戴/非穿戴(穿戴=真实穿身上、替换原衣;非穿戴=手持/佩戴/使用,不动原衣);
`index` 让每张图动作/场景/镜头不同;`n_product` 让参考图序号自适应。"""
name = (getattr(product, "title", "") or "商品").strip()
n_product = max(1, int(n_product or 1))
# 参考图序号自适应:N 张商品图 → 参考图1~N=商品, 参考图N+1=模特
if n_product <= 1:
intro = f"参考图1是「{name}」的真实商品图,是该商品外观的唯一依据。"
prod_ref = "参考图1"
model_idx = 2
else:
rng = f"1-{n_product}" if n_product > 2 else "1、2"
intro = f"参考图{rng}是「{name}」同一件真实商品的不同角度图,是该商品外观的唯一依据,请综合这些角度还原商品。"
prod_ref = f"参考图{rng}"
model_idx = n_product + 1
if _is_wearable_product(product):
relation = (
f"真实穿着{prod_ref}中的这件商品,替换掉模特原本的衣服,让商品自然合身地穿在身上,"
"而不是放在一旁展示,保持商品的版型、领口、袖型、长度、纹样不变"
)
else:
relation = (
f"自然地手持 / 在合适位置佩戴 / 正在使用{prod_ref}中的这件商品(如为耳机则佩戴在耳朵上),"
"不要改动模特原本的服装,不要把商品强行穿到身上"
)
var = _TRYON_VARIATIONS[index % len(_TRYON_VARIATIONS)]
lines = [intro]
if has_model:
lines.append(f"参考图{model_idx}是出镜模特。请生成参考图{model_idx}中这位模特{relation}的电商详情页效果图。")
lines.append(f"模特的五官、发型、肤色、身形、年龄与气质必须与参考图{model_idx}(模特图)高度一致,不要换人,不要自行生成另一位模特。")
else:
lines.append(f"请生成一位真人模特{relation}的电商详情页效果图。")
lines.append(f"商品的外形、配色、材质、品牌文字与 Logo、图案必须与{prod_ref}(商品图)严格一致,不要重新设计、不要改样、不要生成相似款。")
lines.append(f"本张画面:{var['action']};场景:{var['scene']};镜头:{var['shot']}。")
lines.append(quality_suffix("model_tryon", "自然光、真实质感、干净背景、电商主图构图,人物与商品比例真实协调。"))
if base_prompt and base_prompt.strip():
lines.append(base_prompt.strip())
lines.append("请规避:" + _TRYON_NEGATIVE)
return " ".join(lines)
# 平台套图同批多张要「同款商品、不同版式」,否则 N 张文案/构图雷同(PMC#24)。
# 锁死商品一致性,只让排版/构图/视角/配色基调按张变化。
_COVER_VARIATIONS = [
"本张:正面居中主图版式,商品占画面主体,纯净背景,经典电商主图构图",
"本张:换一种排版——商品偏置 + 卖点文案分区,场景化背景,杂志感构图",
"本张:特写细节版式,放大商品材质/做工,近景视角,突出质感",
"本张:生活场景套图版式,商品置于真实使用情境,环境光,氛围感构图",
"本张:多角度组合版式,商品换一个朝向/视角,几何分区背景,现代简约风",
"本张:促销封面版式,留出标题/价签区,高对比配色,强视觉冲击构图",
]
def build_platform_cover_prompt_refs(product, has_model: bool, base_prompt: str = "", index: int = 0) -> str:
"""平台套图 image_edit 提示词(refs 版):参考图1=商品真实主图(锁外形/品牌/配色/Logo),
有模特时参考图2=出镜模特(锁人脸/身形)。生成电商平台主图 / 封面套图,商品须还原真实包装。
`index` 让同批每张的版式/构图/视角不同,避免 N 张雷同(PMC#24)。"""
name = (getattr(product, "title", "") or "商品").strip()
lines = [f"参考图1是「{name}」的真实商品主图。"]
if has_model:
lines += [
"参考图2是出镜模特。请生成参考图2中的这位模特展示参考图1中商品的电商平台套图(主图 / 封面 / 详情);",
"模特的五官、发型、肤色、身形必须与参考图2高度一致,不要换人;",
]
else:
lines.append("请基于该商品生成电商平台套图(主图 / 封面 / 详情排版),统一视觉风格;")
lines.append(
"商品的外形、品牌文字、配色、Logo、材质必须与参考图1高度一致,严禁改动或重新设计包装;"
"干净背景、电商主图构图、真实质感。"
)
lines.append(_COVER_VARIATIONS[index % len(_COVER_VARIATIONS)] + "。")
if base_prompt and base_prompt.strip():
lines.append(base_prompt.strip())
return " ".join(lines)
def build_free_reference_prompt(base_prompt: str, n_refs: int = 1, index: int = 0) -> str:
"""图片创作自由模式 · 带用户上传参考图时的提示词。
纯把用户原话丢给图生图,模型只会松散借个色调、不会真的保留参考图里的主体(背心/商品/人物),
这正是「没参考我上传的素材」的根因。这里显式把参考图钉成「画面主体的唯一依据」,要求严格保留其
外形/款式/配色/材质/品牌文字/图案,再在此基础上按用户要求创作。"""
base = (base_prompt or "").strip()
if n_refs <= 1:
ref_intro = "参考图是用户提供的素材,是本次画面主体(商品 / 人物 / 物体)的唯一依据。"
ref_word = "参考图"
else:
rng = f"1-{n_refs}" if n_refs > 2 else "1、2"
ref_intro = f"参考图{rng}是用户提供的素材(同一主体的不同角度 / 多个主体),是本次画面主体的唯一依据。"
ref_word = f"参考图{rng}"
lines = [
ref_intro,
f"请严格保留{ref_word}中主体的外形、款式、配色、材质、纹样、品牌文字与 Logo,"
"不要重新设计、不要换款、不要生成相似但不同的物体;主体须与参考图高度一致。",
]
if base:
lines.append(f"在此基础上,按用户要求创作:{base}")
else:
lines.append("在此基础上,生成干净、专业的电商视觉画面。")
# 同批多张要不同构图/视角,否则雷同(PMC#24);主体一致性已在上面钉死,这里只变表现。
if index > 0:
lines.append(_COVER_VARIATIONS[index % len(_COVER_VARIATIONS)] + ",在保持主体一致的前提下换一种表现。")
lines.append("画面真实、构图协调、细节清晰。")
return " ".join(lines)
def build_person_frontal_prompt(description: str = "") -> str:
"""人物正面氛围图提示词:把脚本提取(或用户输入)的人物描述包成统一模板。
用户钦定格式:电商真人模特,氛围正面全身照,<描述>,自然妆容,柔和影棚光,真实质感,单人,纯色背景。"""
desc = (description or "").strip()
# 正文可在 admin「提示词」页改;{描述}=脚本提取/用户输入的人物描述(写死字段)。默认含原质量尾串。
default = "电商真人模特,氛围正面全身照,{描述},自然妆容,柔和影棚光,真实质感,单人,纯色背景"
return render_prompt("person_portrait", default, 描述=desc)
# --------------------------------------------------------------------------- #
# 出图重试(中转站偶发抖动 → 一次失败就「生成不出来」的根因兜底)
# --------------------------------------------------------------------------- #
_IMAGE_GEN_ATTEMPTS = 3
def _is_transient_image_error(exc: Exception) -> bool:
"""判定出图失败是否「瞬时」(值得重试)。瞬时:网络抖动 / 5xx / 429 限流 / 中转站吐空无 media url。
永久(不重试,立即退费报错):4xx 内容违规 / 参数错(如 invalid_image_file)、配置缺失等。"""
if isinstance(exc, (requests.ConnectionError, requests.Timeout)):
return True
if isinstance(exc, requests.HTTPError):
code = getattr(getattr(exc, "response", None), "status_code", 0) or 0
return code >= 500 or code == 429
# extract_first_media_url 在响应没有 url 时抛 ValueError —— 多为中转站偶发空返回,重试常能拿到
if isinstance(exc, ValueError) and "media url" in str(exc).lower():
return True
return False
def _run_image_with_retry(make, *, attempts: int = _IMAGE_GEN_ATTEMPTS):
"""跑一次出图(provider 调用 + 取 media url),瞬时错误按退避重试,永久错误立即抛。
make() 须返回 (response, media)。退避 2s / 4s,总耗时上限 ~6s + 出图本身,worker 内执行对用户无感。"""
for i in range(attempts):
try:
return make()
except Exception as exc: # noqa: BLE001
if i == attempts - 1 or not _is_transient_image_error(exc):
raise
time.sleep(2 * (i + 1))
def generate_base_asset(*, project, user, kind: str, prompt: str, label: str = "", group_id: str | None = None) -> AITask:
"""提交基础资产生成(**异步**):Web 请求只建 RESERVED 任务 + 预留额度(秒级),
慢出图(文生图 / 商品 image_edit)交给 Celery worker(run_base_asset_task)跑。
这样 Web 层(gunicorn)不被 ~30s+ 的出图请求占住 → 健康探针不饿死 → 不再"生成几张就整站 502/卡死"。
返回 RESERVED 的 AITask,前端拿 id 轮询 /api/ai/generate-image/?ids=… 取结果;出图后刷新项目即见新组。"""
from apps.ai.tasks import generate_base_asset_task
model_config = get_default_model(ModelConfig.Capability.IMAGE)
if model_config is None:
raise ValueError("no active image model configured")
provider = get_image_provider(model_config)
# 商品三视图:有真实商品主图 → 走 image_edit 以主图为参考,锁定包装(品牌字/配色/外形/Logo)一致;
# 无主图或当前模型不支持 image_edit → 回落纯文生图(仅凭商品名脑补,不保证还原真实包装)。
product_ref_url = _product_cover_url(project.product) if kind == BaseAssetGroup.Kind.PRODUCT else ""
use_edit = bool(product_ref_url) and hasattr(provider, "image_edit")
if use_edit:
gen_prompt = build_product_triview_prompt_refs(project.product, prompt)
elif kind == BaseAssetGroup.Kind.PERSON:
gen_prompt = build_person_frontal_prompt(prompt) # 人物立绘:包成「电商真人模特/正面全身/纯色背景」统一模板
else:
# 场景图 = 空镜。不论 admin「提示词」页的 scene 模板写没写,都强制叠加「无真人」硬约束,并显式压过
# 场景描述里可能出现的人物词(如「主播在客厅」)—— 否则模型会照着画出真人,下游用这张场景图生成视频时
# 被火山以「输入图像可能包含真实人物」拒绝(ZWQ#13)。已叠加过(含「空镜」)则不重复。
gen_prompt = render_prompt("scene", "{场景描述}", 场景描述=(prompt or "").strip())
if "空镜" not in gen_prompt:
gen_prompt = (
f"{gen_prompt}。"
"重要约束:这是一张空镜场景图,画面里绝对不要出现任何人物 / 真人 / 模特 / 人脸 / 人手 / 人影;"
"即使上文描述里提到了人,也只画对应的环境、空间与陈设,把人物完全省略。干净构图,9:16 竖屏。"
)
payload = {
"model": model_config.name, "endpoint": model_config.endpoint, "prompt": gen_prompt,
"kind": kind, "label": label or "", "group_id": str(group_id) if group_id else "",
"use_edit": use_edit, "reference_image": product_ref_url,
}
task = create_ai_task(
project=project,
user=user,
task_type={
BaseAssetGroup.Kind.PRODUCT: AITask.Type.PRODUCT_IMAGE,
BaseAssetGroup.Kind.PERSON: AITask.Type.PERSON_IMAGE,
BaseAssetGroup.Kind.SCENE: AITask.Type.SCENE_IMAGE,
}[kind],
model_config=model_config,
request_payload=payload,
)
generate_base_asset_task.delay(str(task.id))
return task
def run_base_asset_task(*, task_id: str) -> None:
"""Celery worker 内执行基础资产的慢出图:调模型 → 成功落库扣费并归组 / 失败退费。
幂等:只处理 RESERVED 任务,重复投递不会二次出图、二次扣费。"""
task = AITask.objects.select_related("team", "created_by", "project", "model_config").filter(id=task_id).first()
if task is None or task.status != AITask.Status.RESERVED:
return
project = task.project
user = task.created_by
payload = task.request_payload or {}
kind = payload.get("kind")
prompt = str(payload.get("prompt") or "")
label = str(payload.get("label") or "")
group_id = payload.get("group_id") or None
use_edit = bool(payload.get("use_edit"))
ref_url = str(payload.get("reference_image") or "")
model_config = task.model_config
provider = get_image_provider(model_config)
reservation = task.credit_reservation
try:
def _make():
if use_edit and ref_url:
# 商品三视图:比例可在 admin 改(默认横)
size = prompt_ratio_size("product_triview", "1536x1024")
resp = provider.image_edit(model=model_config.name, prompt=prompt, images=[ref_url], size=size)
else:
# 场景默认横、人物立绘默认竖;两者比例都可在 admin 改
if kind == BaseAssetGroup.Kind.SCENE:
gen_size = prompt_ratio_size("scene", "1536x1024")
else:
gen_size = prompt_ratio_size("person_portrait", "1024x1536")
resp = provider.image_generation(model=model_config.name, endpoint=model_config.endpoint, prompt=prompt, size=gen_size)
return resp, provider.extract_first_media_url(resp)
# 中转站偶发抖动(网络/5xx/限流/空返回)按退避重试,避免「一次失败就生成不出来」;永久错误仍立即退费
response, media = _run_image_with_retry(_make)
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)
category = {
BaseAssetGroup.Kind.PRODUCT: Asset.Category.PRODUCT_IMAGE,
BaseAssetGroup.Kind.PERSON: Asset.Category.PERSON,
BaseAssetGroup.Kind.SCENE: Asset.Category.SCENE,
}[kind]
asset = _store_generated_media(
team=project.team,
user=user,
project=project,
task=task,
media=media,
name=f"{project.name}-{kind}",
category=category,
asset_type=Asset.Type.IMAGE,
)
# 复用同实体的组:追加候选 + 采用最新(版本=candidate_assets,采用=adopted_asset);无则新建
group = _find_entity_group(project, kind, label, group_id)
if group is None:
group_meta = {"label": label.strip()} if label and label.strip() else {}
group = BaseAssetGroup.objects.create(project=project, kind=kind, task=task, prompt=prompt, metadata=group_meta)
elif prompt:
group.prompt = prompt
group.candidate_assets.add(asset)
group.adopted_asset = asset
group.save(update_fields=["adopted_asset", "prompt", "updated_at"])
# 真人资产:事务提交后静默送火山审核(best-effort,网络调用放 on_commit 避免占着事务)
if kind == BaseAssetGroup.Kind.PERSON:
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 — 失败要退费并把错误记进 AITask 供前端轮询读取;不向上抛(避免 celery 重试二次扣费)
task.status = AITask.Status.FAILED
task.error_message = str(exc)
task.completed_at = timezone.now()
task.save(update_fields=["status", "error_message", "completed_at", "updated_at"])
release_credit(reservation=reservation, reason=str(exc))
def generate_person_triview(*, project, user, portrait_asset) -> AITask:
"""流程步骤4 · 据「某一版立绘资产」生成它配套的三视图(**异步**:image_edit 慢,交给 worker)。
Web 请求只建 RESERVED 任务 + 预留额度后秒回;worker 内跑 image_edit 并把三视图归组(run_triview_task)。
三视图与立绘 1:1 绑定:metadata.triview_of=<立绘 asset id>;同一立绘多次=同组追加候选(版本)。"""
from apps.ai.tasks import generate_triview_task
from apps.ai.model_library import THREE_VIEW_PROMPT
if portrait_asset is None:
raise ValueError("该立绘尚未生成,无法据它生成三视图")
asset_key = str(portrait_asset.id)
model_config = get_default_model(ModelConfig.Capability.IMAGE)
if model_config is None:
raise ValueError("no active image model configured")
provider = get_image_provider(model_config)
if not hasattr(provider, "image_edit"):
raise ValueError(f"当前图像模型 {model_config.provider.name}:{model_config.name} 不支持参考图三视图(image_edit)")
ref_url = _asset_preview_url(portrait_asset)
# 人物三视图提示词:正文可在 admin「提示词」页改(无占位符)
tri_prompt = render_prompt("person_triview", THREE_VIEW_PROMPT)
payload = {"model": model_config.name, "prompt": tri_prompt, "kind": "person", "triview_of": asset_key, "reference_image": ref_url}
task = create_ai_task(project=project, user=user, task_type=AITask.Type.PERSON_IMAGE, model_config=model_config, request_payload=payload)
generate_triview_task.delay(str(task.id))
return task
def run_triview_task(*, task_id: str) -> None:
"""Celery worker 内执行三视图慢出图(image_edit 以立绘为参考):成功落库扣费并归到立绘的三视图组 / 失败退费。
幂等:只处理 RESERVED 任务,重复投递不会二次出图、二次扣费。"""
from apps.ai.model_library import THREE_VIEW_PROMPT
task = AITask.objects.select_related("team", "created_by", "project", "model_config").filter(id=task_id).first()
if task is None or task.status != AITask.Status.RESERVED:
return
project = task.project
user = task.created_by
payload = task.request_payload or {}
asset_key = str(payload.get("triview_of") or "")
ref_url = str(payload.get("reference_image") or "")
prompt = str(payload.get("prompt") or THREE_VIEW_PROMPT)
model_config = task.model_config
provider = get_image_provider(model_config)
reservation = task.credit_reservation
try:
def _make():
tri_size = prompt_ratio_size("person_triview", "1536x1024")
resp = provider.image_edit(model=model_config.name, prompt=prompt, images=[ref_url], size=tri_size)
return resp, provider.extract_first_media_url(resp)
# 同基础资产:瞬时抖动重试,永久错误立即退费报错
response, media = _run_image_with_retry(_make)
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)
asset = _store_generated_media(
team=project.team, user=user, project=project, task=task, media=media,
name=f"{project.name}-三视图", category=Asset.Category.TRI_VIEW, asset_type=Asset.Type.IMAGE,
)
# 复用该立绘的三视图组(triview_of==立绘asset id):追加候选 + 采用最新
group = next((g for g in project.base_asset_groups.filter(kind=BaseAssetGroup.Kind.PERSON).order_by("created_at")
if (g.metadata or {}).get("triview_of") == asset_key), None)
if group is None:
group = BaseAssetGroup.objects.create(
project=project, kind=BaseAssetGroup.Kind.PERSON, task=task, prompt=prompt,
metadata={"label": "·三视图", "triview_of": asset_key},
)
group.candidate_assets.add(asset)
group.adopted_asset = asset
group.save(update_fields=["adopted_asset", "updated_at"])
# B 路(期3):视频流程新生成的「角色」立绘 + 三视图成套 → 自动加入模特库(团队级可复用)。
# 幂等:同一立绘只建一次 Model;best-effort,不影响主链路。
from apps.assets.models import Model as ModelEntity
portrait = Asset.objects.filter(id=asset_key).first()
if portrait is not None and not ModelEntity.objects.filter(portrait_asset=portrait).exists():
ModelEntity.objects.create(
team=project.team, created_by=user,
name=(portrait.name or f"{project.name}-角色").split("·")[0].strip()[:255] or "角色",
source=ModelEntity.Source.AI,
portrait_asset=portrait, triview_asset=asset,
metadata={"from_project": str(project.id), "auto_enrolled": True},
)
# 合规(期3):三视图含人脸,视频生成前必须过火山审核 → 事务提交后静默送审(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 — 失败退费 + 错误记进 AITask 供前端轮询读取;不向上抛(避免 celery 重试二次扣费)
task.status = AITask.Status.FAILED
task.error_message = str(exc)
task.completed_at = timezone.now()
task.save(update_fields=["status", "error_message", "completed_at", "updated_at"])
release_credit(reservation=reservation, reason=str(exc))
def _scene_context(project) -> str:
"""从商品 + 已采用基础资产提炼一句「风格锚点」,贯穿故事板 / 视频,保证各镜内容一致。"""
product = project.product
parts = [f"商品:{product.title}"]
if product.brand:
parts.append(f"品牌:{product.brand}")
if product.category:
parts.append(f"类目:{product.category}")
if getattr(product, "target_audience", ""):
parts.append(f"人群:{product.target_audience}")
adopted_kinds = set(
project.base_asset_groups.filter(adopted_asset__isnull=False).values_list("kind", flat=True)
)
if BaseAssetGroup.Kind.PERSON in adopted_kinds:
parts.append("真人出镜,保持人物一致")
if BaseAssetGroup.Kind.SCENE in adopted_kinds:
parts.append("统一场景与色调")
return " · ".join(parts)
def _segment_script_text(segment, entities=None, *, with_dialogue: bool = False, with_exposure: bool = True) -> str:
"""本镜脚本文本(画面 + 台词/口播 + 商品露出),拼进故事板/视频提示词的【分镜脚本】。
with_dialogue:有结构化对白时按「说话人:台词」原样放(说话人 id 用 entities 解析成名字),回退扁平 narration。
with_exposure:是否带「商品露出:…」一行。"""
parts = []
visual = (segment.visual_prompt or "").strip()
if visual:
parts.append(f"画面:{visual}")
dialogue = getattr(segment, "dialogue", None) if with_dialogue else None
name_by_id = {e.get("id"): (e.get("name") or "").strip() for e in (entities or []) if isinstance(e, dict)}
lines = []
if isinstance(dialogue, list):
for d in dialogue:
if not isinstance(d, dict):
continue
line = (d.get("line") or "").strip()
if not line:
continue
spk = name_by_id.get(d.get("speaker")) or ""
lines.append(f"{spk}{line}" if spk else line)
if lines: # 有对白 → 用带说话人的台词
parts.append("台词:\n" + "\n".join(lines))
else: # 无对白 → 回退扁平口播/旁白
narration = (segment.narration or "").strip()
if narration:
parts.append(f"口播/旁白:{narration}")
if with_exposure and getattr(segment, "product_exposure", ""):
parts.append(f"商品露出:{segment.product_exposure.strip()}")
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 导演故事板,一镜一图,画面清晰,可直接指导视频生成。{补充}"
)
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 ""),
)
return "\n".join(line for line in rendered.split("\n") if line.strip())
def build_video_segment_prompt(project, video_segment, scene, refs, user_prompt: str = "") -> str:
"""单段视频提示词(用户钦定 @图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'), '参考')})")
# 视频脚本:用带说话人的台词(原样照脚本),不带「商品露出」那行(露出靠模板尾句统一要求)
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点名、{分镜}=分镜图引用、{脚本}=本段脚本、{时长}。
# 时长/比例靠 API 参数传(duration/ratio),不写进正文;尾句给风格 + 音效/字幕要求。
default = "{设定}{分镜}【脚本】{脚本}\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 ""),
脚本=(script_text or f"第 {video_segment.sort_order + 1} 段"),
时长=video_segment.target_duration_seconds,
)
return "\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_default_model(ModelConfig.Capability.IMAGE) is None:
raise ValueError("no active image model configured")
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": "商品"}
def _product_reference_image(project, groups: list | None = None) -> dict | None:
"""商品参考图:优先已采用的商品三视图(product 基础资产组)→ 否则商品真实主图。无图返回 None。
(商品是预创建的真实商品,不从脚本提取;每镜参考图都无条件带上它。)"""
if groups is None:
groups = list(
project.base_asset_groups.filter(adopted_asset__isnull=False).select_related("adopted_asset")
)
pg = next(
(g for g in groups if g.kind == BaseAssetGroup.Kind.PRODUCT and g.adopted_asset_id),
None,
)
if pg is not None:
url = _asset_preview_url(pg.adopted_asset)
if url:
return {"url": url, "label": "商品", "type": "product",
"review_status": pg.adopted_asset.review_status, "review_remote_id": pg.adopted_asset.review_remote_id}
cover = _product_cover_url(project.product)
if cover:
return {"url": cover, "label": "商品", "type": "product"}
return None
def _storyboard_reference_images(project, segment) -> list[dict]:
"""按本镜 entity_refs 取参考图(角色 / 场景 已采用基础资产)+ **无条件带上商品参考图**,
供 gpt-image-2 多图合成 @图N。返回 [{url,label,type}],最多 4 张(角色/场景 ≤3 + 商品 1)。
商品不靠 entity_refs(预创建真实商品、不从脚本提),统一用商品三视图 / 主图带上,从根上保证
商品参考永不缺失。依赖提取步落进 metadata 的 script_entities。"""
entities = {
e.get("id"): e
for e in (project.metadata or {}).get("script_entities", [])
if isinstance(e, dict)
}
kind_by_type = {
"character": BaseAssetGroup.Kind.PERSON,
"scene": BaseAssetGroup.Kind.SCENE,
}
groups = list(project.base_asset_groups.filter(adopted_asset__isnull=False).select_related("adopted_asset"))
# 下游只取「已采用」的角色/场景组(C/D):metadata.adopt 显性优先,否则按是否在脚本里推导;
# 商品组永远保留(不受采用态影响)。顺带把不在脚本里的(含三视图组 label「·三视图」)排除,
# 避免兜底误抓未采用/三视图组。
_entity_names = {(e.get("name") or "").strip() for e in entities.values() if (e.get("name") or "").strip()}
def _group_adopted(g) -> bool:
if g.kind == BaseAssetGroup.Kind.PRODUCT:
return True
a = (g.metadata or {}).get("adopt")
if a == "adopted":
return True
if a == "unadopted":
return False
return (g.metadata or {}).get("label", "").strip() in _entity_names
groups = [g for g in groups if _group_adopted(g)]
out: list[dict] = []
used: set = set()
for rid in (segment.entity_refs or []):
ent = entities.get(rid)
if not ent:
continue
kind = kind_by_type.get(ent.get("type"))
if kind is None: # 商品 / 未知类型不在此处理,商品统一在末尾无条件带上
continue
name = (ent.get("name") or "").strip()
match = next(
(g for g in groups if g.kind == kind and (g.metadata or {}).get("label", "").strip() == name and g.id not in used),
None,
) or next((g for g in groups if g.kind == kind and g.id not in used), None)
if match:
used.add(match.id)
url = _asset_preview_url(match.adopted_asset)
if url:
# 带上审核态/素材库 ID/资产 ID:视频路按需换成 asset:// 引用、过审闸据此定位送审;图像生成路只读 url 忽略它们
out.append({"url": url, "label": name or _ENTITY_TYPE_CN.get(ent.get("type"), "参考"), "type": ent.get("type"),
"asset_id": match.adopted_asset.id,
"review_status": match.adopted_asset.review_status, "review_remote_id": match.adopted_asset.review_remote_id})
if len(out) >= 3: # 给商品留一个位置(总计最多 4 张)
break
# 商品参考图永远带上:优先已采用商品三视图组 → 否则真实主图
product_ref = _product_reference_image(project, groups)
if product_ref is not None and product_ref["url"] not in {r["url"] for r in out}:
out.append(product_ref)
# 规范 @图N 顺序:角色 → 商品 → 场景(与下游 image_edit 传图顺序一致,标注不错位)
_ord = {"character": 0, "product": 1, "scene": 2}
out.sort(key=lambda r: _ord.get(r.get("type"), 9))
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 导演故事板,一镜一图,画面清晰,可直接指导视频生成。{补充}"
)
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 ""),
)
return "\n".join(line for line in rendered.split("\n") if line.strip())
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")
)
def friendly_generation_error(raw: str) -> str:
"""把模型/中转站的原始报错翻成给用户的中文友好提示(前端直接展示)。原始报错仍记进 AITask 供排查。
覆盖:内容审核拦截 / 超时 / 限流 / 凭证 / 参考图被拒 等;命不中给通用兜底。"""
s = (raw or "").lower()
if any(k in s for k in ("moderation_blocked", "safety system", "safety_violation", "content_policy", "content policy")):
return "画面或文案被内容审核拦截(疑似敏感内容)。请调整脚本措辞(如避免「胸罩 / 内衣 / 抚摸 / 贴身」等直白表述,改用「产品 / 包装展示」),或更换参考图后重试。"
if any(k in s for k in ("timed out", "timeout", "read timed out")):
return "生成超时,可能是网络波动或模型繁忙,请稍后重试。"
if any(k in s for k in ("429", "too many requests", "rate limit", "forbidden", "403")):
return "请求过于频繁,模型暂时限流,请稍候片刻再重试。"
if any(k in s for k in ("api_key", "unauthorized", "401")):
return "图像服务凭证异常,请联系管理员处理。"
if any(k in s for k in ("invalid_image", "invalid image", "image_file")):
return "参考图未被模型接受(可能格式/尺寸问题),请更换参考图后重试。"
if any(k in s for k in ("400", "bad request")):
return "生成请求被模型拒绝,请调整提示词或更换参考图后重试。"
return "生成失败,请重试;若多次失败请联系技术支持。"
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
model_config = task.model_config
reservation = task.credit_reservation
extra_prompt = (project.metadata or {}).get("storyboard_prompt", "") or ""
task.status = AITask.Status.SUBMITTED
task.save(update_fields=["status", "updated_at"])
try:
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)
response = _call_image_with_retry(
lambda: provider.image_edit(model=model_config.name, prompt=frame_prompt, images=ref_urls, size="1024x1536")
)
else:
frame_prompt = (
build_storyboard_frame_prompt(project, segment, extra_prompt) if segment is not None
else (task.request_payload.get("prompt") or "")
)
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)
hint = friendly_generation_error(raw) # 翻成前端友好提示;原始报错(含 yunqi body)留在 task 供排查
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=hint, updated_at=timezone.now())
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_default_model(ModelConfig.Capability.IMAGE)
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:
return ""
primary = asset.files.filter(is_primary=True).first() or asset.files.first()
if primary is None:
return ""
if primary.preview_url:
return primary.preview_url
try:
return TosStorage().presigned_get_url(object_key=primary.object_key)
except Exception:
return ""
def _seedance_ref_url(raw_url: str, review_status: str = "", review_remote_id: str = "") -> str:
"""Seedance 参考图 URL:只要资产已进素材库(有 remote_id)就用 asset:// 素材库引用 —— 火山认的是
「在不在素材库」,与审核态无关(实测 processing 也能引用);未进库才回落原始直链。
火山对写实人脸的原始直链会以 InputImageSensitiveContentDetected 直接拒,必须走同账号素材库引用。
人物立绘 / 分镜图(含脸)进库走 asset://;场景 / 商品(无脸、未送审)保持直链。
注意不能只放过 active:故事板刚重生成时分镜图还是 processing,卡 active 会回落直链 → 视频被火山拒。"""
if review_remote_id:
return f"asset://{review_remote_id}"
return raw_url
def _video_reference_images(project, video_segment) -> list[dict]:
"""视频参考图(带类型,供 @图N):角色/场景/商品 基础资产 + 本镜故事板帧。
顺序:角色 → 场景 → 商品 → 分镜图(与用户钦定 @图1角色@图2场景@图3商品@图4分镜图 一致)。
返回 [{url,label,type}];url 对过审人脸资产为 asset:// 素材库引用。都取不到时兜底商品图。"""
out: list[dict] = []
scene = None
adopted_script = project.script_versions.filter(is_adopted=True).prefetch_related("segments").first()
if adopted_script is not None:
scene = adopted_script.segments.filter(sort_order=video_segment.sort_order).first()
if scene is not None:
refs = list(_storyboard_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)
.order_by("-created_at").first()
)
if product_group is not None:
url = _asset_preview_url(product_group.adopted_asset)
if url:
out.append({"url": url, "label": "商品", "type": "product"})
# 写实人脸(已过审)换素材库引用,避免火山「疑似真人」400 拒 → 退文生 → 人物/场景全错
for r in out:
r["url"] = _seedance_ref_url(r["url"], r.get("review_status", ""), r.get("review_remote_id", ""))
return out
def collect_video_review_blockers(project, only_segment: "VideoSegment | None" = None) -> list[dict]:
"""点「生成视频」前的过审闸:列出本次将生成的段里,含真人脸的参考图(人物立绘 + 该镜故事板分镜)
中**尚未过审(review_status != 'active')** 的项。火山视频对含真人脸的图,只接受素材库已过审的引用,
否则报 InputImageSensitiveContentDetected。这里在调火山前先拦,弹窗指明是哪一镜的哪个人物/分镜未过审。
only_segment 给定 → 只校验该段(单段重跑);为 None → 校验全部「未出片」段(整批生成)。
返回 [{video_segment_id, sort_order, scene_no, kind:'person'|'storyboard', 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:
# 整批生成只校验还没出片的段;单段重跑则无论状态都校验(用户主动要重出这一段)
if only_segment is None and seg.status == VideoSegment.Status.SUCCEEDED:
continue
scene_no = seg.sort_order + 1
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):
if ref.get("type") != "character":
continue
if (ref.get("review_status") or "") != "active":
blockers.append({
"video_segment_id": str(seg.id), "sort_order": seg.sort_order, "scene_no": scene_no,
"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
def submit_video_segment(*, video_segment: VideoSegment, user, prompt: str) -> VideoSegmentVersion | None:
model_config = get_default_model(ModelConfig.Capability.VIDEO)
if model_config is None:
raise ValueError("no active video model configured")
project = video_segment.project
# 衔接:按 sort_order 把视频段绑到对应脚本镜,并织出跟住该镜的提示词。
scene = None
adopted_script = project.script_versions.filter(is_adopted=True).prefetch_related("segments").first()
if adopted_script is not None:
scene = adopted_script.segments.filter(sort_order=video_segment.sort_order).first()
if scene is not None and video_segment.script_segment_id != scene.id:
video_segment.script_segment = scene
video_segment.save(update_fields=["script_segment", "updated_at"])
# 参考图(带类型):角色/场景/商品 基础资产 + 本镜故事板帧;@图N 提示词与传图顺序严格对齐。
refs = _video_reference_images(project, video_segment)
reference_images = [r["url"] for r in refs]
final_prompt = build_video_segment_prompt(project, video_segment, scene, refs, prompt)
task = create_ai_task(
project=project,
user=user,
task_type=AITask.Type.VIDEO_SEGMENT,
model_config=model_config,
request_payload={
"model": model_config.name,
"endpoint": model_config.endpoint,
"prompt": final_prompt,
"duration": video_segment.target_duration_seconds,
"ratio": "9:16",
"video_segment_id": str(video_segment.id),
"reference_images": reference_images,
},
)
try:
provider = build_provider(model_config)
# 不再静默退文生兜底:火山报错(如人脸需走素材库的 InputImageSensitiveContentDetected)直接抛,
# 由下方 except 落 error_message → 前端如实显示真实报错,绝不假装成功出一条没锁脸的视频。
response = provider.create_video_task(
model=model_config.name,
endpoint=model_config.endpoint,
prompt=final_prompt,
duration=video_segment.target_duration_seconds,
ratio="9:16",
resolution="720p",
reference_images=reference_images or None,
)
task.provider_task_id = str(response.get("id") or response.get("task_id") or "")
task.response_payload = response
task.status = AITask.Status.SUBMITTED
task.submitted_at = timezone.now()
task.save(update_fields=["provider_task_id", "response_payload", "status", "submitted_at", "updated_at"])
video_segment.status = VideoSegment.Status.RUNNING
video_segment.save(update_fields=["status", "updated_at"])
return None
except Exception as exc:
task.status = AITask.Status.FAILED
task.error_message = str(exc)
task.completed_at = timezone.now()
task.save(update_fields=["status", "error_message", "completed_at", "updated_at"])
release_credit(reservation=task.credit_reservation, reason=str(exc))
video_segment.status = VideoSegment.Status.FAILED
video_segment.error_message = str(exc)
video_segment.save(update_fields=["status", "error_message", "updated_at"])
raise
def poll_video_segment(*, video_segment: VideoSegment, user) -> VideoSegmentVersion | None:
# 幂等:已完成的段直接回采用版;已失败的段不再 poll。避免对已成功 task 再 poll → 二次建版 / 二次扣费。
if video_segment.status == VideoSegment.Status.SUCCEEDED:
return video_segment.adopted_version or video_segment.versions.order_by("-created_at").first()
if video_segment.status == VideoSegment.Status.FAILED:
return None
# ★ 先找「在途任务」再回退旧版本的任务。旧实现反过来:重跑时段上已有(旧)版本,
# 取到旧版本挂的已成功任务 → 短路返回旧版,在途的新任务永远没人轮询,
# 段永远卡「生成中」、新视频取不回来(实测重跑卡 40 分钟,ARK 侧其实早已生成完)。
ai_task = video_segment.project.ai_tasks.filter(
task_type=AITask.Type.VIDEO_SEGMENT,
request_payload__video_segment_id=str(video_segment.id),
status__in=[AITask.Status.SUBMITTED, AITask.Status.POLLING],
).order_by("-created_at").first()
if ai_task is None:
latest_version = video_segment.versions.order_by("-created_at").first()
ai_task = latest_version.task if latest_version else None
if ai_task is None:
raise ValueError("no active video generation task")
# task 已终态(可能被并发的 worker / 另一次 poll 处理过):直接回已有版,不再调 ARK。
if ai_task.status == AITask.Status.SUCCEEDED:
return video_segment.versions.filter(task=ai_task).order_by("-created_at").first()
if ai_task.status in (AITask.Status.FAILED, AITask.Status.CANCELLED):
return None
provider = build_provider(ai_task.model_config)
response = provider.poll_video_task(endpoint=ai_task.model_config.endpoint, provider_task_id=ai_task.provider_task_id)
remote_status = response.get("status")
if remote_status in {"queued", "running", "processing"}:
# 仍在生成:只在状态首次进入 POLLING 时落一次库。旧实现每次 poll(5s 一次)都把完整
# response JSON 回写远程 MySQL——纯浪费写带宽,终态时反正会存完整 payload。
if ai_task.status != AITask.Status.POLLING:
ai_task.status = AITask.Status.POLLING
ai_task.save(update_fields=["status", "updated_at"])
return None
if remote_status in {"failed", "expired", "cancelled"}:
ai_task.status = AITask.Status.FAILED
ai_task.response_payload = response
ai_task.error_message = response.get("error", {}).get("message", "video generation failed")
ai_task.completed_at = timezone.now()
ai_task.save(update_fields=["status", "response_payload", "error_message", "completed_at", "updated_at"])
release_credit(reservation=ai_task.credit_reservation, reason=ai_task.error_message)
video_segment.status = VideoSegment.Status.FAILED
video_segment.error_message = ai_task.error_message
video_segment.save(update_fields=["status", "error_message", "updated_at"])
return None
media = provider.extract_first_media_url(response)
asset = _store_generated_media(
team=video_segment.project.team,
user=user,
project=video_segment.project,
task=ai_task,
media=media,
name=f"{video_segment.project.name}-segment-{video_segment.sort_order + 1}",
category=Asset.Category.VIDEO_CLIP,
asset_type=Asset.Type.VIDEO,
)
# 终态化必须持锁原子做:两个并发 poll(前端 5s 静默轮询 × 提交后轮询/worker)同时走到这里时,
# 旧实现会同 task 建两个版本 + charge_reserved_credit 双扣费(实测 03:08:25 同秒双版本)。
# select_for_update 锁 task 行,后到者看到 SUCCEEDED 直接回已有版,不再建版/扣费。
with transaction.atomic():
locked_task = AITask.objects.select_for_update().get(id=ai_task.id)
if locked_task.status == AITask.Status.SUCCEEDED:
existing = video_segment.versions.filter(task=locked_task).order_by("-created_at").first()
if existing is not None:
return existing
locked_task.status = AITask.Status.SUCCEEDED
locked_task.response_payload = response
locked_task.actual_cost = locked_task.estimated_cost
locked_task.completed_at = timezone.now()
locked_task.save(update_fields=["status", "response_payload", "actual_cost", "completed_at", "updated_at"])
charge_reserved_credit(reservation=locked_task.credit_reservation, actual_amount=locked_task.actual_cost)
version = VideoSegmentVersion.objects.create(
video_segment=video_segment,
task=locked_task,
asset=asset,
prompt=locked_task.request_payload.get("prompt", ""),
is_adopted=True,
)
video_segment.versions.exclude(id=version.id).update(is_adopted=False)
video_segment.adopted_version = version
video_segment.status = VideoSegment.Status.SUCCEEDED
video_segment.error_message = ""
video_segment.save(update_fields=["adopted_version", "status", "error_message", "updated_at"])
return version
def create_export_job(*, timeline, user) -> ExportJob:
return ExportJob.objects.create(timeline=timeline, status=ExportJob.Status.QUEUED)
# 图片趴三类(模特库+资产模型重构 期2):
# · model + product → model_tryon(模特上身图);model 无 product → person(视频角色「生成演员」,留期3)
# · cover → platform_kit(平台套图);image → free_create(自由创作)
_STANDALONE_CATEGORY = {
"model": Asset.Category.PERSON,
"cover": Asset.Category.PLATFORM_KIT,
"image": Asset.Category.FREE_CREATE,
}
_STANDALONE_TASK_TYPE = {
"model": AITask.Type.PERSON_IMAGE,
"cover": AITask.Type.PRODUCT_IMAGE,
"image": AITask.Type.PRODUCT_IMAGE,
}
def _reap_stale_standalone_image_tasks(*, team) -> None:
"""兜底:worker 崩溃/重启(OOM、部署)可能留下卡在 RESERVED 的出图任务,额度被一直占住、
前端轮询也永远等不到结果。超过 10 分钟(远大于单张真实出图耗时 ~60s)仍 RESERVED 的判为僵尸:
标记失败并退还预留额度。趁每次新提交时顺手回收,无需额外的定时任务(与导出僵尸清理同思路)。"""
cutoff = timezone.now() - timedelta(minutes=10)
stale = AITask.objects.filter(
team=team,
project__isnull=True,
task_type__in=[AITask.Type.PERSON_IMAGE, AITask.Type.PRODUCT_IMAGE],
status=AITask.Status.RESERVED,
updated_at__lt=cutoff,
)
for task in stale:
try:
with transaction.atomic():
task.status = AITask.Status.FAILED
task.error_message = "worker 未在预期时间内完成(僵尸任务自动回收)"
task.completed_at = timezone.now()
task.save(update_fields=["status", "error_message", "completed_at", "updated_at"])
try:
reservation = task.credit_reservation
except ObjectDoesNotExist:
reservation = None
if reservation is not None:
release_credit(reservation=reservation, reason="僵尸出图任务自动回收")
except Exception: # noqa: BLE001 — 单个回收失败不应阻断新任务提交
continue
def enqueue_standalone_images(*, team, user, prompt: str, mode: str = "image", count: int = 1, product_id: str | None = None, reference_product: bool = False, model_id: str | None = None, model_entity_id: str | None = None, ratio: str | None = None, image_model: str | None = None, conversation=None, reference_image_ids: list[str] | None = None) -> list[AITask]:
"""独立生图(图片创作 / 模特上身图 / 平台套图)改为**异步**:本函数在 Web 请求里只做「建任务 +
预留额度」这种秒级的活,真正 ~30s 的 ARK 出图交给 Celery worker(generate_standalone_image_task)。
这样 Web 层(gunicorn)不会被慢出图请求占住 worker → 健康探针不会被饿死 → 根治"几张图就整站 502"。
且任务一旦提交(额度已预留),浏览器关掉 / 断网都不影响——worker 照样把图生成并落库,扣费/退费在
worker 内闭环。返回已 RESERVED 的 AITask 列表,前端拿 id 轮询 GET /api/ai/generate-image/?ids=… 取结果。"""
from apps.ai.tasks import generate_standalone_image_task
_reap_stale_standalone_image_tasks(team=team)
# 用户在工作室选的生图模型(火山 / gpt-image)优先;未选或解析不到则回落系统默认
model_config = resolve_image_model(image_model) or get_default_model(ModelConfig.Capability.IMAGE)
if model_config is None:
raise ValueError("no active image model configured")
task_type = _STANDALONE_TASK_TYPE.get(mode, AITask.Type.PRODUCT_IMAGE)
count = max(1, min(int(count or 1), 12))
ref_ids = [str(r) for r in (reference_image_ids or []) if r]
# 带参考图的图片创作必须走 image_edit(gpt-image)才真的还原上传素材——火山 Seedream 的图生图只松散
# 借色调、不锁主体(与三视图/模特上身图同理,那两条流程也强制 image_edit)。当前模型不支持 image_edit
# 就自动切到 gpt-image-2,否则用户选了火山却传了参考图 → 生成完全不参考素材(用户实测的根因)。
if ref_ids and not hasattr(build_provider(model_config), "image_edit"):
alt = resolve_image_model("gpt-image")
if alt is not None:
model_config = alt
# 本次提交 = 一组(模特上身图组 / 平台套图组):同一 batch_id 串起这批图,前端可成组展示。
batch_id = str(uuid.uuid4())
tasks: list[AITask] = []
for index in range(count):
cost = estimate_cost(model_config)
task = AITask.objects.create(
team=team,
created_by=user,
project=None,
conversation=conversation,
task_type=task_type,
status=AITask.Status.CREATED,
model_config=model_config,
idempotency_key=f"standalone-image:{team.id}:{uuid.uuid4()}",
request_payload={"model": model_config.name, "endpoint": model_config.endpoint, "prompt": prompt, "mode": mode, "index": index, "product_id": str(product_id) if product_id else None, "reference_product": bool(reference_product), "model_id": str(model_id) if model_id else None, "model_entity_id": str(model_entity_id) if model_entity_id else None, "batch_id": batch_id, "ratio": str(ratio) if ratio else None, "reference_image_ids": ref_ids},
estimated_cost=cost,
)
# 预留额度若余额不足会抛 ValueError,在同步的 Web 请求里立刻反馈给前端(不会先建半套任务)
reserve_credit(team=team, user=user, task=task, amount=cost)
task.status = AITask.Status.RESERVED
task.save(update_fields=["status", "updated_at"])
tasks.append(task)
# 额度都预留成功后再统一派发,避免"派发了任务但后面某张预留失败"的半成品状态
for task in tasks:
generate_standalone_image_task.delay(str(task.id))
return tasks
def run_standalone_image_task(*, task_id: str) -> None:
"""Celery worker 内执行**单张**图的慢活:调 ARK → 成功落库扣费 / 失败退费。
幂等:只处理 RESERVED 状态的任务,重复投递(celery retry / 重启重放)不会二次出图、二次扣费。"""
task = AITask.objects.select_related("team", "created_by", "model_config").filter(id=task_id).first()
if task is None or task.status != AITask.Status.RESERVED:
return
team = task.team
user = task.created_by
payload = task.request_payload or {}
prompt = str(payload.get("prompt") or "")
mode = str(payload.get("mode") or "image")
index = int(payload.get("index") or 0)
product_id = payload.get("product_id") or None
# 模特上身图(mode=model 且绑了商品)= 图片趴「模特上身图」(引用模特库),归 model_tryon、不送审;
# 「生成演员」同样走 mode=model 但无 product_id = 视频角色,仍归 person(送审,留期3 收编为「角色」)。
if mode == "model" and product_id:
category = Asset.Category.MODEL_TRYON
else:
category = _STANDALONE_CATEGORY.get(mode, Asset.Category.UNCATEGORIZED)
model_config = task.model_config
provider = get_image_provider(model_config)
reservation = task.credit_reservation
# 出图策略(都优先 image_edit 锁真实素材,模型不支持/无素材才回落纯文生图):
# · 模特上身图(mode=model):参考图1=商品真实主图 + 参考图2=选中模特 → 生成「该模特用该商品」效果图;
# · 平台套图(mode=cover):参考图1=商品真实主图(+ 有模特则参考图2=模特)→ 锁包装一致性出套图;
# · 商品三视图(reference_product):参考图1=商品真实主图 → 锁包装一致性;
# · 其余(图片创作):纯文生图。
product = None
if product_id:
from apps.products.models import Product
product = Product.objects.filter(id=product_id).first()
can_edit = hasattr(provider, "image_edit")
model_url = ""
if payload.get("model_id"):
model_asset = Asset.objects.filter(id=payload.get("model_id")).first()
if model_asset is not None:
model_url = _asset_preview_url(model_asset)
product_url = _product_cover_url(product) if product is not None else ""
# 模特上身图:真实上传图优先、排除 AI 生成图,可多张(多角度更易锁外形/品牌)
product_urls = _product_reference_urls(product, limit=3) if product is not None else []
# 图片创作自由模式:用户上传的参考图(已先传成 Asset)→ 取直链,作多图参考(image_edit / 图生图)
ref_urls: list[str] = []
for rid in (payload.get("reference_image_ids") or []):
ref_asset = Asset.objects.filter(id=rid).first()
if ref_asset is not None:
u = _asset_preview_url(ref_asset)
if u:
ref_urls.append(u)
# 参考图收集与 provider 无关:先把「该用哪些参考图 + 哪条提示词」定下来,再按模型能力选调用方式。
edit_images: list[str] = []
edit_prompt = ""
if mode == "model" and product_urls:
# 模特上身图:参考图1~N=商品真实图(多角度),参考图N+1=模特(模特图可缺则让模型自取真人模特)
edit_images = product_urls + ([model_url] if model_url else [])
edit_prompt = build_model_tryon_prompt_refs(
product, has_model=bool(model_url), base_prompt=prompt,
index=index, n_product=len(product_urls),
)
elif mode == "cover" and product_url:
# 平台套图:参考图1=商品真实主图(锁包装一致性),有模特则参考图2=模特(锁人脸/身形)
edit_images = [product_url] + ([model_url] if model_url else [])
edit_prompt = build_platform_cover_prompt_refs(product, has_model=bool(model_url), base_prompt=prompt, index=index)
elif bool(payload.get("reference_product")) and product_url:
edit_images = [product_url]
edit_prompt = build_product_triview_prompt_refs(product, "")
elif ref_urls:
# 图片创作自由模式:以用户上传的参考图为基底出图。提示词必须显式要求「保留参考图主体」,
# 否则图生图只会松散借个色调、不真的还原上传的素材(=用户反馈的「没参考我的图」)。
edit_images = ref_urls
edit_prompt = build_free_reference_prompt(prompt, n_refs=len(ref_urls), index=index)
use_edit = bool(edit_images)
try:
if use_edit and can_edit:
# gpt-image 等支持 image_edit(多图参考编辑接口)
if payload.get("reference_product"):
size = "1536x1024" # 三视图固定横向
else:
size = _ratio_to_image_size(str(payload.get("ratio") or "")) # 模特图按选中比例
response = provider.image_edit(model=model_config.name, prompt=edit_prompt, images=edit_images, size=size)
elif use_edit:
# 火山 Seedream 无 image_edit:走 image_generation 带 image=参考图(图生图多参考);
# 尺寸按选中比例换算成火山可接受的 ~2K 尺寸(三视图固定横向)
vsize = "2304x1728" if payload.get("reference_product") else _ratio_to_volcano_size(str(payload.get("ratio") or ""))
response = provider.image_generation(model=model_config.name, endpoint=model_config.endpoint, prompt=edit_prompt, image=edit_images, size=vsize)
else:
# 纯文生图同批多张要不同构图,否则雷同(PMC#24);index>0 追加换版式指令。
gen_prompt = prompt
if index > 0:
gen_prompt = f"{prompt}{_COVER_VARIATIONS[index % len(_COVER_VARIATIONS)]},换一种构图与视角。"
response = provider.image_generation(model=model_config.name, endpoint=model_config.endpoint, prompt=gen_prompt)
media = provider.extract_first_media_url(response)
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)
fileobj, content_type = VolcanoArkProvider.media_to_bytes(media)
suffix = ".jpg" if "jpeg" in content_type else (".webp" if "webp" in content_type else ".png")
asset_id = uuid.uuid4()
object_key = f"teams/{team.id}/standalone/{asset_id}{suffix}"
stored = TosStorage().upload_fileobj(fileobj=fileobj, object_key=object_key, content_type=content_type)
asset_label = {"model": "模特上身图", "cover": "平台套图", "image": "图片创作"}.get(mode, mode)
# 资产元数据:product_id(商品详情页据此只展示该商品素材)+ batch_id(成组)+ mode + model_entity_id(上身图溯源模特库)
asset_meta: dict = {"mode": mode}
if product_id:
asset_meta["product_id"] = str(product_id)
if payload.get("batch_id"):
asset_meta["batch_id"] = str(payload["batch_id"])
if payload.get("model_entity_id"):
asset_meta["model_entity_id"] = str(payload["model_entity_id"])
asset = Asset.objects.create(
id=asset_id, team=team, created_by=user, name=f"AI 生成 · {asset_label} · {index + 1}",
asset_type=Asset.Type.IMAGE, source=Asset.Source.AI_GENERATED, category=category, origin_task=task,
metadata=asset_meta,
# 工作台生成的图默认不进资产库列表,只在工作台展示;用户「加入资产库」后才置 True
in_library=False,
)
AssetFile.objects.create(asset=asset, object_key=stored.object_key, bucket=stored.bucket, content_type=stored.content_type, size_bytes=stored.size_bytes, is_primary=True)
except Exception as exc: # noqa: BLE001 — 失败要退费并把错误记进 AITask 供前端轮询读取;不向上抛(避免 celery 重试二次扣费)
task.status = AITask.Status.FAILED
task.error_message = str(exc)
task.completed_at = timezone.now()
task.save(update_fields=["status", "error_message", "completed_at", "updated_at"])
release_credit(reservation=reservation, reason=str(exc))
# ── 旁白配音(TTS):每镜旁白合成一段语音,导出时作为人声轨混在 BGM 之上 ──
# 音色按「语音合成(经典版)」试用包实测可用清单配置;大模型音色(*_bigtts)需另开通「语音合成大模型」服务,当前账号 403
VOICEOVER_VOICES = [
{"key": "BV700_streaming", "label": "灿灿 · 活力女声"},
{"key": "BV034_streaming", "label": "知性姐姐 · 沉稳女声"},
{"key": "BV001_streaming", "label": "通用女声"},
{"key": "BV056_streaming", "label": "阳光男声"},
{"key": "BV102_streaming", "label": "儒雅青年 · 解说男声"},
{"key": "BV002_streaming", "label": "通用男声"},
]
DEFAULT_VOICEOVER_VOICE = VOICEOVER_VOICES[0]["key"]
def synthesize_project_voiceover(*, project, user, items: list[dict], voice_type: str, speed_ratio: float = 1.0) -> dict:
"""每镜旁白 → **逐句** TTS 配音资产(一句一段音频,带句内起点 offset_ms),映射写入
timeline.metadata["voiceover"]。逐句才能支持「拖动字幕块 = 字幕和它的语音一起移动」;
一次调用 = 一个 AITask = 计一次费;任何一句失败则整体失败并释放预留(不留半套配音)。"""
from apps.projects.services.export import _split_subtitle_text
texts = [] # (片段 index, 句序 cue, 句文本)
for n, item in enumerate(items or []):
text = str(item.get("text") or "").strip()
if not text:
continue
idx = int(item.get("index", n))
pieces = _split_subtitle_text(text) or [text]
for j, piece in enumerate(pieces):
texts.append((idx, j, piece))
if not texts:
raise ValueError("没有可配音的旁白文本")
provider = VolcanoTtsProvider()
if not provider.configured:
raise TtsNotConfigured(
"语音合成未配置:请在后端环境变量设置 VOLC_TTS_APPID 和 VOLC_TTS_ACCESS_TOKEN"
"(火山引擎控制台 → 语音技术 → 语音合成大模型 → 创建应用)"
)
voice_type = voice_type or DEFAULT_VOICEOVER_VOICE
model_config = get_default_model(ModelConfig.Capability.AUDIO)
if model_config is None:
raise ValueError("no active audio model configured")
task = create_ai_task(
project=project,
user=user,
task_type=AITask.Type.VOICEOVER,
model_config=model_config,
request_payload={
"voice_type": voice_type,
"speed_ratio": float(speed_ratio or 1.0),
"items": [{"index": idx, "cue": j, "text": text} for idx, j, text in texts],
},
)
reservation = task.credit_reservation
try:
synthesized = []
for idx, j, text in texts:
audio, duration_ms = provider.synthesize(text=text, voice_type=voice_type, speed_ratio=speed_ratio, uid=str(user.id))
synthesized.append((idx, j, text, audio, duration_ms))
with transaction.atomic():
task.status = AITask.Status.SUCCEEDED
task.actual_cost = task.estimated_cost
task.completed_at = timezone.now()
task.response_payload = {"segments": len(synthesized)}
task.save(update_fields=["status", "actual_cost", "completed_at", "response_payload", "updated_at"])
charge_reserved_credit(reservation=reservation, actual_amount=task.actual_cost)
vo_items = []
offset_acc: dict[int, int] = {} # 同一片段内逐句顺排:句 j 的默认起点 = 前面句时长之和
for idx, j, text, audio, duration_ms in synthesized:
asset_id = uuid.uuid4()
object_key = f"teams/{project.team_id}/projects/{project.id}/voiceover/{asset_id}.mp3"
stored = TosStorage().upload_fileobj(fileobj=BytesIO(audio), object_key=object_key, content_type="audio/mpeg")
asset = Asset.objects.create(
id=asset_id, team=project.team, created_by=user,
name=f"配音 · 场 {idx + 1} · 句 {j + 1}", asset_type=Asset.Type.AUDIO,
source=Asset.Source.AI_GENERATED, category=Asset.Category.UNCATEGORIZED,
origin_task=task, description=text,
)
AssetFile.objects.create(
asset=asset, object_key=stored.object_key, bucket=stored.bucket,
content_type=stored.content_type, size_bytes=stored.size_bytes, is_primary=True,
)
offset_ms = offset_acc.get(idx, 0)
offset_acc[idx] = offset_ms + (duration_ms or 0)
vo_items.append({
"index": idx, "cue": j, "text": text, "asset": str(asset.id),
"duration_ms": duration_ms, "offset_ms": offset_ms,
})
timeline, _ = Timeline.objects.get_or_create(
project=project, defaults={"name": f"{project.name} Timeline", "duration_seconds": 60}
)
metadata = dict(timeline.metadata or {})
metadata["voiceover"] = {
"enabled": True,
"voice_type": voice_type,
"speed_ratio": float(speed_ratio or 1.0),
"items": vo_items,
}
timeline.metadata = metadata
timeline.save(update_fields=["metadata", "updated_at"])
return metadata["voiceover"]
except Exception as exc:
task.status = AITask.Status.FAILED
task.error_message = str(exc)
task.completed_at = timezone.now()
task.save(update_fields=["status", "error_message", "completed_at", "updated_at"])
release_credit(reservation=reservation, reason=str(exc))
raise