Files
yingqing/core/backend/apps/ai/services.py
T
zycandClaude Opus 4.8 77759678a7 fix: 商品库生成的三视图打 view=three_view 标记,刷新不丢 (ZWQ#5 续)
真因:商品库点「AI生成三视图」走 reference_product 路径,产出资产 name=「图片创作」、
metadata 只有 mode/product_id —— 没有 view=three_view,也非 tri_view 类。导致:
①商品库详情三视图面板刷新后认不出(显示「尚未生成」、采用也丢);
②视频项目第二步基础资产读不到「商品已有三视图」。

修:reference_product 资产与视频项目基础资产阶段生成的商品三视图同款标记 ——
category=product_image + 名字含「商品三视图」+ metadata.view=three_view + product_id。
这样 products is_triview(排除 product_image 不误入角色区)、商品库前端回填(view===three_view)、
视频项目商品三视图判定都能认出。需部署生效;存量旧资产无标记不追溯。

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-06-30 11:40:23 +08:00

2561 lines
139 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 整除)。
⚠️ 该网关只接受 1024x1024 / 1024x1536 / 1536x1024 三种固定 size,无法直接产出精确 3:4 / 9:16 / 4:5,
这里只能取最接近的「竖/横/方」近似;要拿到精确像素比例需在出图后做主体保护裁切/补边(见优化文档 §7.1
normalize_output_image,属待部署的后处理项)。修复点:4:5 原会 fallback 成 1024x1024(方图)→ 比例错误,
现归到竖图 1024x1536。"""
return {
"1:1": "1024x1024",
"3:4": "1024x1536", # 近似竖图(网关无精确 3:4)
"4:5": "1024x1536", # 近似竖图(原 fallback 成方图是比例 bug)
"9:16": "1024x1536", # 近似竖图(网关无精确 9:16)
"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)。
# 锁死商品一致性,只让排版/构图/视角/配色基调按张变化。
# 注:仅纯文生图回落路径(无参考图)仍用这个简表;refs 版改用下面的 slot 体系。
_COVER_VARIATIONS = [
"本张:正面居中主图版式,商品占画面主体,纯净背景,经典电商主图构图",
"本张:换一种排版——商品偏置 + 卖点文案分区,场景化背景,杂志感构图",
"本张:特写细节版式,放大商品材质/做工,近景视角,突出质感",
"本张:生活场景套图版式,商品置于真实使用情境,环境光,氛围感构图",
"本张:多角度组合版式,商品换一个朝向/视角,几何分区背景,现代简约风",
"本张:促销封面版式,留出标题/价签区,高对比配色,强视觉冲击构图",
]
# 平台套图优化版(对照「平台套图线上提示词优化版.md」):平台差异 + slot 版式 + 低信息密度上限。
# 平台 id 用规范化 key(前端 dy/tb… 已在 PLATFORM_ID_MAP 里映射成这些 canonical key)。
_PLATFORM_NAMES = {
"taobao": "淘宝", "tmall": "天猫", "jd": "京东", "pdd": "拼多多", "douyin": "抖音电商",
"xhs": "小红书", "kuaishou": "快手", "wechat": "视频号", "amazon": "亚马逊", "1688": "1688",
}
# §4 平台块:每块只描述平台调性 / 版式倾向,统一服从「头图低信息密度」,不写比例(比例由 size 控制)。
_PLATFORM_COVER_BLOCKS = {
"taobao": (
"平台:淘宝。画面像淘宝搜索货架和商品主图,移动端缩略图下商品一眼可识别;"
"常见左上小品牌区、中央或偏右商品主体、底部克制活动条 / 轻卖点条;"
"背景干净但不单调,可用浅灰棚拍 / 浅色墙面 / 试衣间 / 窗边挂拍 / 干净台面;"
"禁止详情页 / 长图 / 平台 UI / 二维码 / 直播间界面 / 虚假价格与未提供折扣。"
),
"tmall": (
"平台:天猫。画面像品牌旗舰店商品首图,品牌感 / 质感 / 留白 / 材质光影更重要;"
"促销感弱于淘宝,不要大红大黄低价感,不要拼多多式强利益区。"
),
"jd": (
"平台:京东。画面清晰 / 可信 / 理性 / 标准,像京东商品主图;白底或浅灰商品图、克制品质短标题、包装 / 配件 / 材质细节;"
"信息区极少,只允许 1 个短标题或 1-2 个短标签;不要参数表 / 功能说明长区 / 详情页式模块,不要小红书滤镜,不要强生活方式过度氛围。"
),
"pdd": (
"平台:拼多多。主体大 / 信息直接 / 利益区明显,像商品主图或活动头图;"
"可有高对比活动区和大标题但只讲一个利益点,不能编造价格 / 优惠 / 折扣 / 销量 / 平台补贴;"
"背景明亮,商品边缘清楚,不要复杂品牌大片感。"
),
"douyin": (
"平台:抖音电商。像信息流商品封面 / 短视频电商封面,近景 / 强裁切 / 真实使用瞬间 / 动作感强;"
"常见背景:衣橱 / 梳妆台 / 卧室 / 开箱 / 手部整理 / 拿取 / 穿搭准备;文案更短、商品主体更大;"
"不要抖音 UI / 播放按钮 / 直播间贴片 / 字幕条,也不要套淘宝底部活动条。"
),
"xhs": (
"平台:小红书。像生活方式笔记封面,自然光 / 真实体验 / 种草氛围;"
"常见背景:卧室 / 衣橱 / 梳妆台 / 桌面 / 浴室 / 旅行收纳 / 通勤准备;标题像用户体验表达,少硬广 / 少参数表 / 少促销条。"
),
"kuaishou": (
"平台:快手。像快手小店商品图 / 直播前置封面素材,真实 / 直接 / 可信;"
"可有桌面展示 / 手持展示 / 打包台 / 家中真实环境 / 开箱场景;不要直播 UI / 主播贴片 / 平台水印。"
),
"wechat": (
"平台:视频号 / 微信小店。画面克制可信,适合社交分享和商品卡;"
"背景可用干净家居 / 礼赠 / 办公 / 生活空间 / 柔和自然光;不要微信聊天界面 / 二维码 / 公众号 UI。"
),
"amazon": (
"平台:亚马逊。第 1 张为 MAIN 合规图:纯白背景、只展示售卖商品本体、无文字 / 无道具 / 无边框 / 无水印,商品占画面主要区域;"
"第 2 张以后可为 Lifestyle / Feature / Detail / Package 辅图;不要 Amazon 徽章 / 评分 / 排名 / 优惠 / 平台 UI。"
),
"1688": (
"平台:1688。像批发采购图,商品清楚,偏规格 / 材质 / 工艺 / 包装 / 供货信息;"
"背景可用白底 / 浅灰 / 工厂台面 / 仓储 / 包装台 / 材料工艺背景;头图仍保持低信息密度,只允许 1 个短标题或 1-2 个短标签;"
"不要做成规格表详情页,不要虚构工厂资质 / 库存 / 起订量 / 认证与价格。"
),
}
# §3 / §3.1 slot 版式骨架(低信息密度版):默认 4 张按 hero→scene→selling→detail 取,8/12 张再补 multi/promo。
_COVER_SLOTS = {
"hero": "正面或四分之三角度的商品 / 上身主视觉,主体占画面 60%-80%,纯净干净背景,经典电商主图构图,无文案",
"scene": "场景主视觉:挂拍 / 衣架 / 生活场景 / 手部整理 / 使用情境,环境自然光,氛围感,无文案或仅 1 个极短标题",
"selling": "轻卖点封面:商品主体居中或偏置,最多 1 个短标题加 1-2 个极短标签,文字区克制,不堆参数",
"detail": "质感特写:放大材质 / 做工 / 关键结构(扣位 / 肩带 / 边缘走线),近景视角,突出质感,无文案或 1 个短标签",
"multi": "多角度组合:商品换一个朝向 / 视角,几何分区干净背景,现代简约风,无文案",
"promo": "促销封面:主体大、利益点单一,底部或角落留 1 条活动短语,高对比配色,不铺满文字、不编造价格",
}
_COVER_SLOT_ORDER = ["hero", "scene", "selling", "detail", "multi", "promo"]
# §3.1 / §8 头图低信息密度上限:所有平台 / 所有模型都必须遵守,防止漂成详情页。
_COVER_LOW_DENSITY = (
"这是一张商品头图,不是详情页:画面第一优先级是商品主体 / 上身效果 / 使用场景,"
"商品或模特主体占画面 60%-80%,文字与装饰不得抢主体;如需文案最多 1 个短标题(≤8 字)加 1-2 个极短标签;"
"严禁参数表 / 规格表 / 长句卖点 / 三段式说明 / 密集图标 / 大量箭头 / 2x3 或 3x3 信息宫格 / 左右对比详情页 / 多屏排版。"
)
# §5 背景反差:避免商品与背景同色相融(无法逐图取色时给通用避让规则)。
_COVER_BG_CONTRAST = (
"背景必须与商品主色形成清楚反差,不能让商品与背景同色相融;"
"浅色商品避免奶白 / 浅米 / 浅粉等近似大面积底,优先冷灰 / 蓝灰 / 浅绿 / 自然木色或明确投影、边缘光;"
"深色商品避免黑 / 深灰 / 深棕大底,优先暖白 / 浅灰 / 浅木色 / 日光窗边;"
"除亚马逊 MAIN 或明确白底图外,不要整组都做成纯白 / 浅灰底。"
)
# §6.1 内衣真人上身强约束:防止被第二件衣服遮挡、防性感化漂移。
_UNDERWEAR_ON_MODEL = (
"这是真人试穿商品头图:成人模特直接穿着参考商品中的内衣,内衣是唯一服饰重点,"
"罩杯 / 肩带 / 下围 / 杯型 / 面料 / 颜色与关键版型须完整可见且与参考商品一致;"
"不要出现 T 恤 / 衬衫 / 吊带背心 / 运动上衣 / 连衣裙 / 制服 / 外套等遮挡内衣的第二件衣服,"
"不要把内衣穿在其他衣物外面或在其下面加衣;最多允许轻薄开衫 / 薄纱松搭在肩臂外侧但不得盖住罩杯 / 肩带 / 下围;"
"正规电商服饰目录拍摄,成人、克制、商品展示向,非性感化、非挑逗姿势,关键结构不被裁切。"
)
# §9 通用负面提示词(平台套图专用,压缩版)。
_COVER_NEGATIVE = (
"请规避:详情页 / 详情长图 / 多屏排版 / 店铺页 / 平台 UI / 直播间界面 / 二维码 / 联系方式 / 水印 / 平台 Logo / 虚假角标;"
"不要重新设计商品或改其包装 / Logo / 配色 / 材质 / 文字 / 图案 / 比例,不要生成相似款或改品类,不要擅自加套装 / 配件 / 赠品;"
"不要出现商品数据外的品牌名 / 英文 Logo / 系列名 / 角落签名 / 模型名 / 伪水印;"
"不要编造价格 / 优惠券 / 满减 / 折扣 / 限时活动 / 销量 / 排名 / 认证 / 功效承诺 / 绝对化用语;"
"不要错别字 / 乱码 / 小字堆叠 / 大段文案;不要让文字遮挡主体;不要让商品与背景同色相融。"
)
# 内衣 / 贴身衣物品类提示(比通用穿戴更窄,用于触发内衣强约束分支)。
_UNDERWEAR_HINTS = ("内衣", "文胸", "胸罩", "bra", "内裤", "泳", "比基尼", "bikini", "睡衣", "塑身", "束身")
def _is_underwear_product(product) -> bool:
blob = f"{getattr(product, 'category', '') or ''} {getattr(product, 'title', '') or ''}".lower()
return any(h in blob for h in _UNDERWEAR_HINTS)
def build_platform_cover_prompt_refs(
product,
*,
has_model: bool = False,
base_prompt: str = "",
platform_id: str = "",
index: int = 0,
count: int = 4, # noqa: ARG001 — 透传保留,后续可据张数扩展 slot 选择
product_ref_count: int = 1,
) -> str:
"""平台套图 image_edit 提示词(refs 优化版,对照「平台套图线上提示词优化版.md」):
参考图1~N=同一件真实商品(锁外形/品牌/配色/Logo/比例),有模特时参考图N+1=出镜模特。
按 `platform_id` 注入平台块、按 `index` 选 slot 版式,统一服从「头图低信息密度」与「背景反差」,
内衣 + 有模特时叠加强约束。只出头图 / 主图 / 封面候选,不再出详情。"""
name = (getattr(product, "title", "") or "商品").strip()
n = max(1, int(product_ref_count or 1))
if n <= 1:
ref_word = "参考图1"
intro = f"参考图1是「{name}」的真实商品参考图。"
else:
rng = f"1-{n}" if n > 2 else "1、2"
ref_word = f"参考图{rng}"
intro = f"参考图{rng}是「{name}」同一件真实商品的不同角度参考图。"
lines = [intro]
lines.append(
f"请严格锁定{ref_word}中商品的外形、颜色、材质、结构、Logo、品牌文字与比例,"
"严禁重新设计 / 改样 / 改品类;这些参考图只锁商品本体,不锁原图里的床品 / 桌面 / 墙面 / 绿植 / 道具与拍摄光线。"
)
# 平台块(canonical key 命中则用 §4 平台块,否则回落平台名 / 通用)
block = _PLATFORM_COVER_BLOCKS.get(platform_id)
pname = _PLATFORM_NAMES.get(platform_id, "")
if block:
lines.append(block)
elif pname:
lines.append(f"请生成一张适合「{pname}」平台的商品头图 / 主图 / 封面候选,统一视觉风格。")
else:
lines.append("请生成一张电商平台商品头图 / 主图 / 封面候选,统一视觉风格。")
# 头图低信息密度上限
lines.append(_COVER_LOW_DENSITY)
# 模特身份 + 内衣强约束
if has_model:
lines.append(
f"参考图{n + 1}是出镜模特,请让这位模特真实展示该商品;"
"模特的五官 / 发型 / 肤色 / 身形须与该图高度一致,不要换人。"
)
if _is_underwear_product(product):
lines.append(_UNDERWEAR_ON_MODEL)
# 本张 slot 版式
slot_key = _COVER_SLOT_ORDER[index % len(_COVER_SLOT_ORDER)]
lines.append("本张版式:" + _COVER_SLOTS[slot_key] + "。")
# 背景反差
lines.append(_COVER_BG_CONTRAST)
if base_prompt and base_prompt.strip():
lines.append(
"用户补充(只影响氛围 / 构图 / 场景 / 光线 / 表达偏好,不得覆盖商品一致性、平台与版式规则):"
+ base_prompt.strip()
)
lines.append(_COVER_NEGATIVE)
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]
# 商品基础资产 = 商品三视图(prompt 即「生成同一件商品的三视图」):名字带「三视图」并打
# metadata.view=three_view + product_id,让商品库的三视图查询(products is_triview / ?product 过滤)
# 能认出并回填——否则视频项目生成的商品三视图同步不回商品库,商品库永远显示「尚未生成」(ZWQ#5)。
is_product = kind == BaseAssetGroup.Kind.PRODUCT
asset_name = f"{project.name}-商品三视图" if is_product else f"{project.name}-{kind}"
asset = _store_generated_media(
team=project.team,
user=user,
project=project,
task=task,
media=media,
name=asset_name,
category=category,
asset_type=Asset.Type.IMAGE,
)
if is_product:
meta = dict(asset.metadata or {})
meta["view"] = "three_view"
if project.product_id:
meta["product_id"] = str(project.product_id)
asset.metadata = meta
asset.save(update_fields=["metadata", "updated_at"])
# 复用同实体的组:追加候选 + 采用最新(版本=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 避免占着事务)。
# 场景 / 商品也送审,因为它们可能出现真人(模特出镜/上身),拿到 remote_id 后视频路才能换 asset:// 引用,
# 否则传原始直链会被火山判「疑似真人」拒。submit_asset_for_review 自身按 REVIEW_CATEGORIES 兜底,非送审类不会真送。
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;已存在该立绘的 Model(尤其「真人上传」模特,upload 时只建了 portrait、
# 还没 triview)→ 回写 triview_asset,否则模特库永远显示「无三视图」(ZWQ#7:AI 角色能更新、真人上传不更新)。
from apps.assets.models import Model as ModelEntity
portrait = Asset.objects.filter(id=asset_key).first()
if portrait is not None:
existing_model = ModelEntity.objects.filter(portrait_asset=portrait, is_deleted=False).first()
if existing_model is None:
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},
)
elif existing_model.triview_asset_id != asset.id:
# 真人上传 / 已有模特补三视图:回写新三视图(应用后模特库即刷新三视图)
existing_model.triview_asset = asset
existing_model.save(update_fields=["triview_asset", "updated_at"])
# 合规(期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, platform_id: 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())
# 平台套图:规范化平台 id(前端 dy/tb… → canonical),用于注入平台版式块(优化版);非 cover 模式忽略。
platform_key = str(platform_id or "").strip() if mode == "cover" else ""
platform_name = _PLATFORM_NAMES.get(platform_key, "")
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, "platform_id": platform_key or None, "platform_name": platform_name or None},
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_urls:
# 平台套图:参考图1~N=商品真实图(多角度,_product_reference_urls 已优先真实上传图/排除 AI 图),
# 有模特则参考图N+1=模特(锁人脸/身形)。platform_id 注入平台版式块(优化版)。
edit_images = product_urls + ([model_url] if model_url else [])
edit_prompt = build_platform_cover_prompt_refs(
product,
has_model=bool(model_url),
base_prompt=prompt,
platform_id=str(payload.get("platform_id") or ""),
index=index,
product_ref_count=len(product_urls),
)
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)
# 商品三视图(reference_product):与视频项目基础资产阶段生成的商品三视图同款标记 ——
# category=product_image + 名字含「商品三视图」+ metadata.view=three_view + product_id。
# 否则商品库自己生成的三视图认不出(无 view 标记),刷新即丢、视频项目第二步也读不到「商品已有三视图」(ZWQ#5 续)。
is_product_triview = bool(payload.get("reference_product"))
asset_label = "商品三视图" if is_product_triview else {"model": "模特上身图", "cover": "平台套图", "image": "图片创作"}.get(mode, mode)
asset_category = Asset.Category.PRODUCT_IMAGE if is_product_triview else category
# 资产元数据:product_id(商品详情页据此只展示该商品素材)+ batch_id(成组)+ mode + model_entity_id(上身图溯源模特库)
asset_meta: dict = {"mode": mode}
if is_product_triview:
asset_meta["view"] = "three_view"
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=asset_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