Files
yingqing/core/backend/apps/ai/video_replace.py
T
2026-08-31 10:13:56 +08:00

1662 lines
67 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.
"""视频复刻:参考视频提炼分镜稿 → Seedance 2.5 按稿出片。
流程很短:
1. 用户上传的参考视频只走「提炼提示词」同一套接口,拆成中文分镜稿;
2. 把这份稿(格式与提炼页一模一样)加上换商品/换角色,交给 Seedance 2.5;
3. 只带商品图/角色图。用户原片不传火山,避免素材审核报错。
商品图/人物图过审后用 asset://;审核失败不扣费。
"""
from __future__ import annotations
import logging
import re
import uuid
from decimal import Decimal
from django.conf import settings
from django.db import transaction
from django.db.models import Q
from django.utils import timezone
from apps.assets.models import Asset, Model
from apps.billing.models import CreditAccount
from apps.billing.pricing import quote_video_estimate, video_reserve_amount
from apps.products.models import Product
from .free_video import (
FREE_VIDEO_MODELS,
HIGH_RES_MODEL,
REPLACE_MODEL,
IN_FLIGHT_STATUSES,
RATIOS,
RESOLUTIONS,
_reap_stale_free_video_tasks,
model_duration_range,
serialize_free_video_task,
start_pending_free_video,
submit_free_video,
)
from .models import AITask, ModelConfig
logger = logging.getLogger(__name__)
FEATURE = "video_replace"
REPLACE_MODES = {"product", "character"}
MAX_IMAGES = 9
LEGACY_PROMPT_PREFIX = "[视频复刻]"
# 商品三视图在商品库是 standalone Asset(metadata.view=three_view + product_id),
# 和项目内的 BaseAssetGroup 是两套存储,这里直接取商品库那份。
TRIVIEW_LABEL = "目标商品三视图"
# 复刻走 Seedance 2.5(单次最长 30 秒)。半秒容差同 media_probe(30 秒成片常被探成 30.02)。
REPLACE_REF_DURATION_MAX = 30.5
REVIEW_UNAVAILABLE = "素材审核服务暂不可用,请稍后重试"
REVIEW_FAILED = "参考素材未通过真人合规审核,请更换视频或图片后重试"
REVIEW_SUBMIT_FAILED = "素材提交审核失败,请稍后重试"
class VideoReplaceInProgress(ValueError):
"""本团队已有复刻在跑。视图转 409,并把在跑的那条带回去让前端接上。"""
def __init__(self, task):
super().__init__("已有一个视频正在复刻中,完成或取消后才能再提交")
self.task = task
def get_inflight_video_replace(team):
"""本团队正在跑的复刻任务(含审核中 / 提炼中)。没有返回 None。"""
_reap_stale_free_video_tasks(team=team)
return (
AITask.objects.filter(
team=team,
task_type=AITask.Type.FREE_VIDEO,
status__in=IN_FLIGHT_STATUSES,
is_deleted=False,
purged_at__isnull=True,
)
.filter(video_replace_q())
.order_by("-created_at")
.first()
)
PRODUCT_PROMPT = (
"按下面这份分镜稿生成视频。镜头数量、每镜时长、景别、机位、运镜、人物、场景、口播全部按稿执行,"
"不要增删镜头、不要改顺序。"
"把稿里的原商品换成@目标商品,外观、材质、包装以参考图为准。"
)
CHARACTER_PROMPT = (
"按下面这份分镜稿生成视频。镜头数量、每镜时长、景别、机位、运镜、商品、场景、口播全部按稿执行,"
"不要增删镜头、不要改顺序。"
"把稿里的原人物换成@目标角色,五官、发型、体态以参考图为准。"
)
# 拆解期间的占位提示词。真正的提示词在 worker 拆完后回写。
DIGEST_PENDING_PROMPT = "正在提炼参考视频的分镜稿…"
# 商品库里有三视图时才追加(没有就不提,免得模型去找一张不存在的图)
PRODUCT_TRIVIEW_NOTE = "商品各面以@目标商品三视图为准。"
# 只在用户主动把成片调得比参考片短很多时才出现。时长一致时绝不出现 ——
# 它和「节奏一模一样」是互相打架的。
PRODUCT_CONDENSE_NOTE = (
"参考片约 {source} 秒,本次成片 {output} 秒:整镜省略次要镜头,不要快放、不要把两镜压成一镜。"
)
SHOT_HEADER = "【镜头"
DIGEST_TAIL_SECTION = "【拆解存疑】"
# 整片设定里值得留的(顺序即输出顺序)
KEEP_HEADER_FIELDS = ("画面比例", "整体风格", "主线")
# 喂给 Seedance 的镜头原文栏位。字幕/音效/BGM/备注会诱使模型画花字、加戏,一律丢掉。
SHOT_COPY_FIELDS = ("时间", "时长", "景别", "机位", "运镜", "画面", "人物动作", "人物表情", "台词/旁白")
SHOT_INLINE_FIELDS = ("时间", "时长", "景别", "机位", "运镜")
SHOT_DETAIL_FIELDS = (("画面", "画面"), ("人物动作", "动作"), ("人物表情", "表情"), ("台词/旁白", "口播"))
_EMPTY_VALUES = {"", "无", "不可见", "看不出", "看不清", "听不清"}
# 模型会写「不可见。」「无。」这种带句读的空值,比对前先剥掉尾部标点
_TRAILING_PUNCT = "。.,,、;:!? "
def _is_empty_value(value: str) -> bool:
return value.strip().rstrip(_TRAILING_PUNCT) in _EMPTY_VALUES
def _parse_digest(digest_text: str) -> tuple[dict, list[dict]]:
"""分镜稿纯文本 → (整片设定, 镜头列表)。解析不出镜头返回 ([], []),由调用方回落原文。"""
body = (digest_text or "").strip()
head, sep, _tail = body.partition(DIGEST_TAIL_SECTION)
if sep:
body = head.strip()
header: dict[str, str] = {}
shots: list[dict] = []
current: dict | None = None
for raw in body.split("\n"):
line = raw.strip()
if not line:
continue
if line.startswith(SHOT_HEADER):
current = {}
shots.append(current)
continue
key, sep_char, value = line.partition("")
if not sep_char:
continue
key, value = key.strip(), value.strip()
if current is None:
header[key] = value
else:
current[key] = value
return header, [shot for shot in shots if shot]
def _render_shot_table(header: dict, shots: list[dict]) -> str:
"""镜头表:每镜一行摘要(时间/景别/机位/运镜)+ 画面、动作、表情、口播四行细节。
空值(无 / 不可见)整行省掉 —— 反复出现的「无」只会稀释真正的指令。"""
lines: list[str] = ["【整片设定】"]
for field in KEEP_HEADER_FIELDS:
value = (header.get(field) or "").strip()
if not _is_empty_value(value):
lines.append(f"{field}{value}")
lines.append(f"镜头数:{len(shots)} 镜(下面逐镜照拍,不要增删、不要改顺序)")
lines.append("")
lines.append("【逐镜还原】")
for index, shot in enumerate(shots, start=1):
inline = [shot.get(f, "").strip() for f in SHOT_INLINE_FIELDS]
inline = [v for v in inline if not _is_empty_value(v)]
lines.append(f"镜头 {index:02d} | " + " | ".join(inline) if inline else f"镜头 {index:02d}")
for field, label in SHOT_DETAIL_FIELDS:
value = (shot.get(field) or "").strip()
if not _is_empty_value(value):
lines.append(f" {label}{value}")
return "\n".join(lines)
# —— 逐镜生成 ——
# 一镜一次 Seedance,每段只还原原片对应时间,并把 @参考视频 带上锁切镜。
MIN_SHOT_SECONDS = 4 # 火山单次出片下限
MAX_CLIP_SECONDS = 15 # Seedance 2.0 单次上限;单镜再长也一次出,避免凭空切一刀
MAX_SHOTS = 12 # 再多就是积分黑洞,也超出用户预期
_CLOCK_RE = re.compile(r"(\d{1,2}):(\d{2})")
def _raw_shot_seconds(shot: dict) -> int:
"""分镜稿里这一镜多长,不做上下限裁切。优先「时长」,其次用「时间」区间算。"""
raw = (shot.get("时长") or "").strip()
digits = re.sub(r"[^0-9]", "", raw)
if digits:
return max(1, int(digits))
stamps = _CLOCK_RE.findall(shot.get("时间") or "")
if len(stamps) >= 2:
start = int(stamps[0][0]) * 60 + int(stamps[0][1])
end = int(stamps[1][0]) * 60 + int(stamps[1][1])
if end > start:
return end - start
return 5
def _clip_seconds(raw: int) -> int:
return max(MIN_SHOT_SECONDS, min(MAX_CLIP_SECONDS, int(raw or 0) or MIN_SHOT_SECONDS))
def _pack_shot_groups(shots: list[dict]) -> list[list[dict]]:
"""不足 4 秒的相邻镜合并成一次出片,避免火山拒单,也避免把 2 秒镜硬拉成 4 秒。"""
groups: list[list[dict]] = []
current: list[dict] = []
current_sec = 0
for shot in shots[:MAX_SHOTS]:
sec = _raw_shot_seconds(shot)
if current and current_sec < MIN_SHOT_SECONDS:
current.append(shot)
current_sec += sec
continue
if current:
groups.append(current)
current = [shot]
current_sec = sec
if current:
if groups and current_sec < MIN_SHOT_SECONDS:
groups[-1].extend(current)
else:
groups.append(current)
return groups
def _trim_groups_to_output(groups: list[list[dict]], output_seconds: int) -> list[list[dict]]:
"""用户把成片要得比参考片短很多时,整组省略尾镜,不要压缩画面。"""
if not output_seconds or not groups:
return groups
total = sum(_clip_seconds(sum(_raw_shot_seconds(s) for s in group)) for group in groups)
if total - output_seconds < 3:
return groups
kept: list[list[dict]] = []
acc = 0
for group in groups:
if acc >= output_seconds:
break
kept.append(group)
acc += _clip_seconds(sum(_raw_shot_seconds(s) for s in group))
return kept or groups[:1]
def _verbatim_shot_block(shot: dict) -> str:
"""提炼稿这一镜的原文栏位,不改写、不换标签名。"""
lines = []
for field in SHOT_COPY_FIELDS:
value = (shot.get(field) or "").strip()
if not _is_empty_value(value):
lines.append(f"{field}{value}")
return "\n".join(lines)
def _shot_clock(shot: dict) -> str:
clock = (shot.get("时间") or "").strip()
return clock if clock and not _is_empty_value(clock) else ""
def build_shot_plan(
digest_text: str,
subject_name: str,
*,
has_triview: bool = False,
replace_mode: str = "product",
output_seconds: int = 0,
) -> list[dict]:
"""分镜稿 → 逐镜生成计划。解析不出镜头返回 [],调用方回落单次生成。"""
_header, shots = _parse_digest(digest_text)
if not shots:
return []
groups = _trim_groups_to_output(_pack_shot_groups(shots), output_seconds)
total = len(groups)
plan = []
for index, group in enumerate(groups, start=1):
seconds = _clip_seconds(sum(_raw_shot_seconds(shot) for shot in group))
plan.append({
"index": index,
"seconds": seconds,
"prompt": build_shot_prompt(
group, index=index, total=total, seconds=seconds,
subject_name=subject_name, has_triview=has_triview, replace_mode=replace_mode,
),
})
return plan
def build_shot_prompt(
shots, *, index: int, total: int, seconds: int, setting: str = "",
subject_name: str, has_triview: bool = False, replace_mode: str = "product",
) -> str:
"""2.0 短指令 + 这一镜的提炼原文。@参考视频 才是切镜来源,不要扩写。"""
from .services import enforce_no_embedded_captions
group = shots if isinstance(shots, list) else [shots]
subject = (subject_name or "").strip() or ("目标角色" if replace_mode == "character" else "目标商品")
is_character = replace_mode == "character"
if len(group) == 1:
clock = _shot_clock(group[0])
span = f",对应原片 {clock}" if clock else ""
scope = f"只生成下面这一镜(第 {index}/{total} 镜),时长 {seconds}{span}。不要拍前后镜头,不要额外切镜。"
shot_body = _verbatim_shot_block(group[0])
else:
scope = (
f"按顺序只拍下面 {len(group)} 个连续切点(第 {index}/{total} 段),共 {seconds} 秒。"
"只在这些切点切换,不要加镜。"
)
blocks = []
for offset, shot in enumerate(group, start=1):
clock = _shot_clock(shot)
head = f"切点 {offset}" + (f"(原片 {clock}" if clock else "")
block = _verbatim_shot_block(shot)
blocks.append(f"{head}\n{block}" if block else head)
shot_body = "\n".join(blocks)
if is_character:
swap = (
"将画面中需要替换的原人物完整替换为@目标角色。"
"保留@参考视频的商品、场景、镜头运动、剪辑节奏与口播。"
)
else:
swap_bits = [
"将画面中需要替换的原商品完整替换为@目标商品的外观。",
"保留@参考视频的人物、场景、镜头运动、剪辑节奏与口播氛围。",
]
if has_triview:
swap_bits.append(PRODUCT_TRIVIEW_NOTE)
if subject:
swap_bits.append(f"口播里的旧商品名改说「{subject}」。")
swap = "".join(swap_bits)
parts = [
"使用@参考视频作为镜头、节奏与口播氛围基准。",
scope,
swap,
shot_body,
]
return enforce_no_embedded_captions("\n".join(part for part in parts if part))
def _digest_for_seedance(digest_text: str) -> str:
"""提炼页同款分镜稿。只砍给用户校对的【拆解存疑】,其余原文照传。"""
body = (digest_text or "").strip()
if not body:
raise ValueError("参考视频拆解结果为空,请重试")
head, sep, _tail = body.partition(DIGEST_TAIL_SECTION)
return (head if sep else body).strip()
def build_character_replace_prompt(
digest_text: str,
*,
source_seconds: float = 0,
output_seconds: int = 0,
) -> str:
"""提炼页分镜稿 + 换角色。原片不传 Seedance。"""
from .services import enforce_no_embedded_captions
source = int(round(float(source_seconds or 0)))
output = int(output_seconds or 0)
condense = (
PRODUCT_CONDENSE_NOTE.format(source=source, output=output)
if source and output and source - output >= 3
else ""
)
return enforce_no_embedded_captions(
"\n".join(part for part in (CHARACTER_PROMPT, _digest_for_seedance(digest_text), condense) if part)
)
def compact_digest_for_video(digest_text: str) -> str:
"""分镜稿 → 交给视频模型的镜头表。
解析失败(模型没按 skill 的格式吐)就回落:只砍掉【拆解存疑】,原文照传 ——
宁可啰嗦也不能把镜头信息弄丢。
"""
header, shots = _parse_digest(digest_text)
if not shots:
body = (digest_text or "").strip()
head, sep, _tail = body.partition(DIGEST_TAIL_SECTION)
return (head if sep else body).strip()
return _render_shot_table(header, shots)
def build_product_replace_prompt(
digest_text: str,
subject_name: str,
*,
has_triview: bool = False,
source_seconds: float = 0,
output_seconds: int = 0,
) -> str:
"""提炼页分镜稿 + 换商品。原片不传 Seedance。"""
from .services import enforce_no_embedded_captions
subject = (subject_name or "").strip()
source = int(round(float(source_seconds or 0)))
output = int(output_seconds or 0)
condense = (
PRODUCT_CONDENSE_NOTE.format(source=source, output=output)
if source and output and source - output >= 3
else ""
)
extra = []
if has_triview:
extra.append(PRODUCT_TRIVIEW_NOTE)
if subject:
extra.append(f"口播里的旧商品名改说「{subject}」。")
return enforce_no_embedded_captions(
"\n".join(part for part in (PRODUCT_PROMPT, *extra, _digest_for_seedance(digest_text), condense) if part)
)
def is_video_replace_task(task) -> bool:
payload = task.request_payload or {}
if payload.get("feature") == FEATURE:
return True
return str(payload.get("prompt") or "").startswith(LEGACY_PROMPT_PREFIX)
def video_replace_q() -> Q:
return Q(request_payload__feature=FEATURE) | Q(request_payload__prompt__startswith=LEGACY_PROMPT_PREFIX)
def serialize_video_replace_task(task, *, include_deleted_assets: bool = False) -> dict:
data = serialize_free_video_task(task, include_deleted_assets=include_deleted_assets)
payload = task.request_payload or {}
replace_mode = payload.get("replace_mode") or _legacy_replace_mode(payload.get("prompt") or "")
created = task.status == AITask.Status.CREATED
digesting = created and bool(payload.get("digest_pending"))
reviewing = created and bool(payload.get("review_pending")) and not digesting
plan = list(payload.get("shot_plan") or [])
shot_total = int(payload.get("shot_total") or len(plan) or 0)
shot_index = 0
if plan:
active = next((item for item in plan if item.get("status") in {"submitted", "polling"}), None)
if active is None:
done = sum(1 for item in plan if item.get("status") == "succeeded")
shot_index = done
else:
shot_index = int(active.get("index") or 0)
data.update({
"feature": FEATURE,
"replace_mode": replace_mode,
"subject_name": payload.get("subject_name") or "",
"subject_source": payload.get("subject_source") or "",
"product_id": payload.get("product_id") or "",
"model_id": payload.get("model_id") or "",
"review_stage": "reviewing" if reviewing else "",
"digest_stage": "digesting" if digesting else "",
"digest_text": str(payload.get("digest_text") or ""),
"digest_shots": payload.get("digest_shots") or shot_total or 0,
"digest_source_name": payload.get("digest_source_name") or "",
"digest_source": payload.get("digest_source_ref") or None,
"shot_index": shot_index,
"shot_total": shot_total,
})
return data
def submit_video_replace(*, team, user, params: dict):
"""校验素材 → 送审 → 已过审则直接生成,否则建 CREATED 任务等绿盾。失败抛 ValueError。"""
replace_mode = str(params.get("replace_mode") or "").strip()
if replace_mode not in REPLACE_MODES:
raise ValueError("请选择替换商品或替换角色")
product_id = _optional_uuid(params.get("product_id"), "商品")
model_id = _optional_uuid(params.get("model_id"), "角色")
image_ids = _uuid_list(params.get("image_asset_ids"), "参考图")
has_product = product_id is not None
has_model = model_id is not None
has_temp = bool(image_ids)
if replace_mode == "product":
if has_model:
raise ValueError("商品复刻请选择商品,不要同时选择角色")
if has_product and has_temp:
raise ValueError("请从商品库选择,或临时上传商品图,不要混用")
if not has_product and not has_temp:
raise ValueError("请选择商品或上传商品参考图")
else:
if has_product:
raise ValueError("角色复刻请选择角色,不要同时选择商品")
if has_model and has_temp:
raise ValueError("请从人物库选择,或临时上传角色图,不要混用")
if not has_model and not has_temp:
raise ValueError("请选择角色或上传角色参考图")
# 单飞闸:一个团队同时只允许一条复刻在跑。刷新页面时前端拉状态有空档,
# 用户容易以为「没任务」再点一次,重复扣费。_reap_stale_free_video_tasks 会先回收
# 死任务(CREATED 超 16 分钟等),所以不会被永久锁住。
running = get_inflight_video_replace(team)
if running is not None:
raise VideoReplaceInProgress(running)
video = _team_asset(team, params.get("video_asset_id"), kind=Asset.Type.VIDEO, label="参考视频")
video_seconds = _asset_duration_seconds(video)
if video_seconds > REPLACE_REF_DURATION_MAX:
raise ValueError("参考视频不能超过 30 秒,请剪短后重试")
if has_product:
subject_name, image_refs, subject_source = _product_library_refs(team, product_id)
elif has_model:
subject_name, image_refs, subject_source = _character_library_refs(team, model_id)
else:
noun = "商品" if replace_mode == "product" else "角色"
subject_name, image_refs, subject_source = _temporary_image_refs(team, image_ids, noun=noun)
duration = _output_duration(params.get("duration"), video_seconds)
extra = {
"replace_mode": replace_mode,
"subject_name": subject_name,
"subject_source": subject_source,
"product_id": str(product_id) if product_id else "",
"model_id": str(model_id) if model_id else "",
}
base_params = {
"mode": "universal",
# 固定 Seedance 2.5,不接受前端指定别的档
"model": REPLACE_MODEL,
"aspect_ratio": str(params.get("aspect_ratio") or "9:16"),
"resolution": str(params.get("resolution") or "720p"),
"duration": duration,
"seed": params.get("seed", -1),
"generate_audio": True,
"feature": FEATURE,
}
# 分镜用「提炼提示词」同一套拆。原片只当拆解源,不进审核列表、也不传火山。
from .video_digest import resolve_digest_model_config
if resolve_digest_model_config() is None:
raise ValueError("视频提炼模型未配置,请联系管理员")
# 商品图/角色图仍要过审才能当生成参考。提前送审 → 审核和提炼并行跑,明确不过审的直接 400。
review_state = _ensure_replace_refs_reviewed(team, image_refs)
if review_state == "failed":
raise ValueError(REVIEW_FAILED)
image_refs = _refresh_replace_refs(team, image_refs)
extra.update({
"digest_source_asset_id": str(video.id),
"digest_source_name": video.name or "参考视频",
"digest_source_duration": round(video_seconds, 2),
# 原片快照只给历史卡「原视频」和重新生成用,不进 Seedance references。
"digest_source_ref": _owned_ref(video, kind="video", role="reference_video", label="参考视频"),
"review_pending": review_state != "ready",
})
return _create_reviewing_task(
team=team,
user=user,
params={
**base_params,
"prompt": DIGEST_PENDING_PROMPT,
"references": image_refs,
"extra_payload": extra,
},
digest_pending=True,
)
def advance_video_replace(task):
"""推进复刻:审核中 → 整份分镜稿一次出片。旧逐镜任务也按单次收尾,不再拼接。"""
if not is_video_replace_task(task):
return task
if task.status != AITask.Status.CREATED:
from .free_video import finalize_free_video
return finalize_free_video(task=task)
payload = task.request_payload or {}
if payload.get("digest_pending"):
# worker 还在拆参考视频,提示词都没生成,别当成「审核中」反复送审。
# worker 挂了也不会永远卡着:CREATED 超 16 分钟由 _reap_stale_free_video_tasks 回收退费。
return task
references = _image_refs(payload.get("references") or [])
try:
state = _ensure_replace_refs_reviewed(task.team, references)
except ValueError as exc:
return _fail_reviewing_task(task, str(exc))
if state == "failed":
return _fail_reviewing_task(task, REVIEW_FAILED)
if state != "ready":
return task
refreshed = _refresh_replace_refs(task.team, references)
try:
_assert_replace_refs_ready(refreshed)
except ValueError as exc:
return _fail_reviewing_task(task, str(exc))
with transaction.atomic():
locked = AITask.objects.select_for_update().get(id=task.id)
if locked.status != AITask.Status.CREATED:
return locked
next_payload = dict(locked.request_payload or {})
next_payload["references"] = refreshed
next_payload["review_pending"] = False
locked.request_payload = next_payload
locked.save(update_fields=["request_payload", "updated_at"])
task = locked
return start_replace_generation(task)
def start_replace_generation(task):
"""审核过了:整份分镜稿一次交给 Seedance 2.5。只带商品图/角色图,不传用户原片。"""
payload = dict(task.request_payload or {})
payload["references"] = _image_refs(payload.get("references") or [])
payload.pop("shot_plan", None)
payload.pop("shot_total", None)
task.request_payload = payload
task.save(update_fields=["request_payload", "updated_at"])
return start_pending_free_video(task)
def run_replace_digest(task) -> None:
"""Worker:参考视频 → 提炼页同款分镜稿 → 一次交给 Seedance。不传原片。
商品和角色共用。这一步失败 = 整条复刻失败。此时还没预留视频积分,不扣费。
"""
from apps.ai.video_digest import VideoDigestError, digest_asset_video
if not is_video_replace_task(task) or task.status != AITask.Status.CREATED:
return
payload = task.request_payload or {}
if not payload.get("digest_pending"):
return
try:
asset = _load_replace_asset(
task.team, uuid.UUID(str(payload.get("digest_source_asset_id") or "")), "参考视频"
)
except (ValueError, TypeError):
_fail_reviewing_task(task, "参考视频已失效,请重新上传", error_code="asset_unavailable")
return
replace_mode = str(payload.get("replace_mode") or "product")
subject = str(payload.get("subject_name") or "")
has_triview = any(
(ref or {}).get("label") == TRIVIEW_LABEL for ref in (payload.get("references") or [])
)
source_seconds = float(payload.get("digest_source_duration") or 0)
output_seconds = int(payload.get("duration") or 0)
try:
digest, meta = digest_asset_video(asset=asset, task=task)
if replace_mode == "character":
prompt = build_character_replace_prompt(
digest, source_seconds=source_seconds, output_seconds=output_seconds,
)
else:
prompt = build_product_replace_prompt(
digest, subject, has_triview=has_triview,
source_seconds=source_seconds, output_seconds=output_seconds,
)
except (VideoDigestError, ValueError) as exc:
_fail_reviewing_task(task, str(exc), error_code="processing_failed")
return
except Exception as exc: # noqa: BLE001 — 拆解任何异常都要把任务收尾,别留 CREATED 僵尸
logger.exception("video replace digest failed for %s", task.id)
_fail_reviewing_task(task, f"参考视频拆解失败:{exc}", error_code="processing_failed")
return
with transaction.atomic():
locked = AITask.objects.select_for_update().get(id=task.id)
if locked.status != AITask.Status.CREATED:
return
next_payload = dict(locked.request_payload or {})
next_payload.update(meta)
next_payload["digest_pending"] = False
next_payload["digest_text"] = digest[:32000]
next_payload["prompt"] = prompt
next_payload["references"] = _image_refs(next_payload.get("references") or [])
next_payload.pop("shot_plan", None)
next_payload.pop("shot_total", None)
locked.request_payload = next_payload
locked.save(update_fields=["request_payload", "updated_at"])
task = locked
# 拆完就地推进一次:参考图多半早已过审,能直接提交火山,省掉一轮 8s 轮询。
try:
task = advance_video_replace(task)
except Exception: # noqa: BLE001 — 推进失败交给轮询重试,任务还在 CREATED
logger.warning("video replace advance after digest failed for %s", task.id, exc_info=True)
task.refresh_from_db()
if task.status == AITask.Status.CREATED:
_enqueue_replace_review_poll(task)
def start_pending_replace_shots(task):
"""旧逐镜路径已下线。一律整片一次出,且不传用户原片。"""
return start_replace_generation(task)
def _legacy_start_pending_replace_shots(task):
"""CREATED → 按全部镜头总时长预留积分 → 提交第一镜。"""
from apps.billing.services.ledger import reserve_credit
from .free_video import (
_dispatch_free_video_provider,
_fail_pending_free_video,
build_content_items,
model_duration_range,
)
if task.status != AITask.Status.CREATED:
return task
payload = task.request_payload or {}
plan = list(payload.get("shot_plan") or [])
if not plan:
return start_pending_free_video(task)
first = plan[0]
prompt = str(first.get("prompt") or payload.get("prompt") or "").strip()
mode = str(payload.get("mode") or "universal")
aspect_ratio = str(payload.get("aspect_ratio") or "9:16")
resolution = str(payload.get("resolution") or "720p")
generate_audio = bool(payload.get("generate_audio", True))
search_mode = str(payload.get("search_mode") or "off")
feature = str(payload.get("feature") or FEATURE)
try:
seed = int(payload.get("seed") if payload.get("seed") is not None else -1)
except (TypeError, ValueError):
seed = -1
billed_duration = sum(int(item.get("seconds") or 0) for item in plan) or int(payload.get("duration") or 5)
references = _seedance_references(payload)
try:
built = build_content_items(
team=task.team, prompt=prompt, mode=mode, references=references,
max_ref_seconds=model_duration_range(task.model_config)[1],
)
except ValueError as exc:
message = str(exc)
if "正在审核" in message or "已提交审核" in message or "尚未完成合规审核" in message:
return task
return _fail_pending_free_video(task, message)
tokens, quote = quote_video_estimate(
task.model_config,
aspect_ratio=aspect_ratio,
resolution=resolution,
duration=billed_duration,
references=built["snapshots"],
team=task.team,
)
reserve_amount = video_reserve_amount(quote.points)
with transaction.atomic():
locked = (
AITask.objects.select_for_update()
.select_related("model_config", "model_config__provider", "team", "created_by")
.get(id=task.id)
)
if locked.status != AITask.Status.CREATED:
return locked
try:
reserve_credit(team=locked.team, user=locked.created_by, task=locked, amount=reserve_amount)
except ValueError as exc:
message = "团队余额不足,请充值后重试" if "insufficient credit" in str(exc) else str(exc)
locked.status = AITask.Status.FAILED
locked.error_code = "user_credit_insufficient" if "余额不足" in message else "invalid_input"
locked.error_message = message[:2000]
locked.completed_at = timezone.now()
locked.save(update_fields=["status", "error_code", "error_message", "completed_at", "updated_at"])
return locked
next_payload = dict(locked.request_payload or {})
plan = list(next_payload.get("shot_plan") or plan)
plan[0] = {**plan[0], "status": "submitted"}
next_payload["shot_plan"] = plan
next_payload["prompt"] = prompt
next_payload["api_prompt"] = built["api_prompt"]
next_payload["references"] = built["snapshots"]
next_payload["estimated_tokens"] = tokens
next_payload["review_pending"] = False
next_payload["duration"] = billed_duration
locked.request_payload = next_payload
locked.estimated_cost = quote.points
locked.status = AITask.Status.RESERVED
locked.save(update_fields=["request_payload", "estimated_cost", "status", "updated_at"])
dispatched = _dispatch_free_video_provider(
task=locked,
built=built,
model_config=locked.model_config,
aspect_ratio=aspect_ratio,
duration=int(first.get("seconds") or MIN_SHOT_SECONDS),
resolution=resolution,
generate_audio=generate_audio,
seed=seed,
search_mode=search_mode,
feature=feature,
mode=mode,
poll_countdown=30,
)
return _stamp_current_shot_id(dispatched)
def advance_replace_shots(task):
"""轮询当前这一镜:还在跑就等;失败整单失败;成功则提交下一镜,全部完成再拼接。"""
from .free_video import (
_dispatch_free_video_provider,
build_content_items,
finalize_free_video,
model_duration_range,
)
from .generation_errors import classify_generation_error
from .routing_policy import load_model_routing_policy
from .services import get_video_provider
payload = dict(task.request_payload or {})
plan = list(payload.get("shot_plan") or [])
if not plan:
return finalize_free_video(task=task)
if task.status not in (AITask.Status.RESERVED, AITask.Status.SUBMITTED, AITask.Status.POLLING):
return task
current = _active_shot(plan)
if current is None:
if plan and all(item.get("status") == "succeeded" and item.get("media") for item in plan):
return complete_replace_shots(task)
return task
if not current.get("provider_task_id"):
if task.provider_task_id and current.get("status") in {"submitted", "polling", "queued"}:
current = {**current, "provider_task_id": task.provider_task_id, "status": "submitted"}
_write_shot(task, current)
plan = _read_shot_plan(task)
current = _shot_by_index(plan, current["index"]) or current
else:
return _dispatch_next_replace_shot(task, current)
video_policy = load_model_routing_policy().video
submit_attempt = (
task.model_attempts.filter(status="succeeded", operation="video_generate")
.select_related("model_config__provider")
.order_by("-sequence")
.first()
)
actual_model = submit_attempt.model_config if submit_attempt and submit_attempt.model_config else task.model_config
provider = get_video_provider(actual_model)
response = provider.poll_video_task(
endpoint=actual_model.endpoint,
provider_task_id=current["provider_task_id"],
timeout=video_policy.poll_request_timeout,
)
remote_status = str(response.get("status") or "")
if remote_status in {"queued", "running", "processing", "submitted"}:
if task.status != AITask.Status.POLLING or current.get("status") != "polling":
current = {**current, "status": "polling"}
task = _write_shot(task, current)
if task.status != AITask.Status.POLLING:
task.status = AITask.Status.POLLING
task.save(update_fields=["status", "updated_at"])
return task
if remote_status in {"failed", "expired", "cancelled"}:
err = response.get("error") or {}
code = str(err.get("code") or "")
raw_message = str(err.get("message") or "video generation failed")
public_error = classify_generation_error(
RuntimeError(raw_message),
operation="video_generate",
provider_code=code,
reference_id=str(task.id),
)
return _fail_replace_generation(
task, raw_message, error_code=code[:64] or "CreateTaskError", hint=public_error.fallback_message,
)
try:
media = provider.extract_first_media_url(response)
except Exception as exc: # noqa: BLE001
return _fail_replace_generation(task, str(exc), error_code="PostprocessError")
usage = response.get("usage") or {}
try:
tokens = int(usage.get("total_tokens") or 0)
except (TypeError, ValueError):
tokens = 0
current = {**current, "status": "succeeded", "media": media, "tokens": tokens}
task = _write_shot(task, current)
plan = _read_shot_plan(task)
nxt = next((item for item in plan if item.get("status") in {"queued", "", None}), None)
if nxt is None:
if len(plan) == 1:
return finalize_free_video(task=task)
return complete_replace_shots(task)
return _dispatch_next_replace_shot(task, nxt)
def _dispatch_next_replace_shot(task, shot: dict):
from .free_video import _dispatch_free_video_provider, build_content_items, model_duration_range
payload = dict(task.request_payload or {})
prompt = str(shot.get("prompt") or "").strip()
if not prompt:
return _fail_replace_generation(task, "分镜提示词缺失,请重试", error_code="processing_failed")
try:
built = build_content_items(
team=task.team, prompt=prompt, mode=str(payload.get("mode") or "universal"),
references=_seedance_references(payload),
max_ref_seconds=model_duration_range(task.model_config)[1],
)
except ValueError as exc:
return _fail_replace_generation(task, str(exc), error_code="invalid_input")
with transaction.atomic():
locked = AITask.objects.select_for_update().select_related(
"model_config", "model_config__provider", "team", "created_by",
).get(id=task.id)
if locked.status not in (AITask.Status.RESERVED, AITask.Status.SUBMITTED, AITask.Status.POLLING):
return locked
plan = list((locked.request_payload or {}).get("shot_plan") or [])
target = _shot_by_index(plan, shot["index"])
if target is None or target.get("status") not in {"queued", "", None, "submitted"}:
return locked
if target.get("provider_task_id") and target.get("status") in {"submitted", "polling"}:
return locked
next_payload = dict(locked.request_payload or {})
next_payload["prompt"] = prompt
next_payload["api_prompt"] = built["api_prompt"]
plan = [
{**item, "status": "submitted"} if item.get("index") == shot["index"] else item
for item in plan
]
next_payload["shot_plan"] = plan
locked.request_payload = next_payload
locked.save(update_fields=["request_payload", "updated_at"])
try:
seed = int(payload.get("seed") if payload.get("seed") is not None else -1)
except (TypeError, ValueError):
seed = -1
dispatched = _dispatch_free_video_provider(
task=locked,
built=built,
model_config=locked.model_config,
aspect_ratio=str(payload.get("aspect_ratio") or "9:16"),
duration=int(shot.get("seconds") or MIN_SHOT_SECONDS),
resolution=str(payload.get("resolution") or "720p"),
generate_audio=bool(payload.get("generate_audio", True)),
seed=seed,
search_mode=str(payload.get("search_mode") or "off"),
feature=str(payload.get("feature") or FEATURE),
mode=str(payload.get("mode") or "universal"),
poll_countdown=30,
)
return _stamp_current_shot_id(dispatched)
def complete_replace_shots(task):
"""全部镜头出完:下载 → ffmpeg 拼接 → 转存 TOS → 按 tokens 合计结算。"""
from decimal import Decimal
from apps.billing.pricing import quote_video_actual
from apps.billing.services.ledger import charge_reserved_credit, release_credit
from .free_video import _notify_failure, _store_free_video_media
from .generation_errors import classify_generation_error
with transaction.atomic():
locked = AITask.objects.select_for_update().get(id=task.id)
if locked.status not in (AITask.Status.SUBMITTED, AITask.Status.POLLING):
return locked
locked.status = AITask.Status.POSTPROCESSING
locked.save(update_fields=["status", "updated_at"])
payload = dict(locked.request_payload or {})
plan = list(payload.get("shot_plan") or [])
try:
if len(plan) == 1 and plan[0].get("media"):
_store_free_video_media(task=locked, media=plan[0]["media"])
else:
video_bytes = _concat_shot_media([item.get("media") or "" for item in plan])
_store_free_video_media(task=locked, video_bytes=video_bytes)
total_tokens = 0
for item in plan:
try:
total_tokens += int(item.get("tokens") or 0)
except (TypeError, ValueError):
pass
resolution = payload.get("resolution") or "720p"
if total_tokens > 0:
settle = quote_video_actual(
locked.model_config, tokens=total_tokens, with_video_ref=True,
resolution=resolution, multiplier=Decimal(str(payload.get("price_multiplier") or "1")),
)
actual, base_cost = settle.points, settle.base_cost_yuan
payload["actual_tokens"] = total_tokens
if settle.meta.get("rate"):
payload["points_per_yuan_snapshot"] = settle.meta["rate"]
else:
actual, base_cost = locked.estimated_cost, locked.base_cost
with transaction.atomic():
locked = AITask.objects.select_for_update().get(id=locked.id)
if locked.status != AITask.Status.POSTPROCESSING:
return locked
reservation = locked.credit_reservation
if actual > reservation.amount:
logger.warning(
"video replace task %s actual cost %s exceeds reserved %s, clamped",
locked.id, actual, reservation.amount,
)
actual = reservation.amount
locked.status = AITask.Status.SUCCEEDED
locked.actual_cost = actual
locked.base_cost = base_cost
locked.request_payload = payload
locked.completed_at = timezone.now()
locked.save(
update_fields=["status", "actual_cost", "base_cost", "request_payload", "completed_at", "updated_at"]
)
charge_reserved_credit(reservation=reservation, actual_amount=actual)
return locked
except Exception as exc: # noqa: BLE001
logger.exception("video replace concat failed for task %s", locked.id)
public_error = classify_generation_error(
exc, operation="video_generate", internal_kind="processing_failed",
reference_id=str(locked.id),
)
with transaction.atomic():
locked = AITask.objects.select_for_update().get(id=locked.id)
if locked.status != AITask.Status.POSTPROCESSING:
return locked
locked.status = AITask.Status.FAILED
locked.error_code = "PostprocessError"
locked.error_message = str(exc)[:2000]
locked.completed_at = timezone.now()
locked.save(update_fields=["status", "error_code", "error_message", "completed_at", "updated_at"])
release_credit(reservation=locked.credit_reservation, reason=str(exc)[:200])
_notify_failure(locked, raw=str(exc), hint=public_error.fallback_message)
return locked
def _concat_shot_media(urls: list[str]) -> bytes:
import subprocess
import tempfile
from pathlib import Path
from .providers.volcano import VolcanoArkProvider
clips = [url for url in urls if url]
if not clips:
raise ValueError("没有可拼接的镜头成片")
if len(clips) == 1:
fileobj, _ctype = VolcanoArkProvider.media_to_bytes(clips[0])
return fileobj.getvalue()
with tempfile.TemporaryDirectory(prefix="airshelf-replace-concat-") as tmp:
tmp_dir = Path(tmp)
paths = []
for index, url in enumerate(clips):
fileobj, _ctype = VolcanoArkProvider.media_to_bytes(url)
path = tmp_dir / f"shot{index:02d}.mp4"
path.write_bytes(fileobj.getvalue())
paths.append(path)
output = tmp_dir / "out.mp4"
inputs: list[str] = []
for path in paths:
inputs.extend(["-i", str(path)])
n = len(paths)
parts = [f"[{i}:v]fps=24,format=yuv420p,setsar=1[v{i}]" for i in range(n)]
audio_ok = []
for path in paths:
probe = subprocess.run(
["ffprobe", "-v", "error", "-select_streams", "a",
"-show_entries", "stream=codec_type", "-of", "csv=p=0", str(path)],
capture_output=True, text=True, timeout=30,
)
audio_ok.append(bool((probe.stdout or "").strip()))
for i, path in enumerate(paths):
if audio_ok[i]:
parts.append(f"[{i}:a]aresample=44100,aformat=sample_fmts=fltp:channel_layouts=stereo[a{i}]")
else:
dur_proc = subprocess.run(
["ffprobe", "-v", "error", "-show_entries", "format=duration",
"-of", "csv=p=0", str(path)],
capture_output=True, text=True, timeout=30,
)
try:
dur = max(0.1, float((dur_proc.stdout or "4").strip() or "4"))
except ValueError:
dur = 4.0
parts.append(
f"anullsrc=channel_layout=stereo:sample_rate=44100,atrim=0:{dur:.3f},"
f"asetpts=PTS-STARTPTS,aformat=sample_fmts=fltp:channel_layouts=stereo[a{i}]"
)
concat_in = "".join(f"[v{i}][a{i}]" for i in range(n))
parts.append(f"{concat_in}concat=n={n}:v=1:a=1[v][a]")
cmd = [
"ffmpeg", "-y", *inputs, "-filter_complex", ";".join(parts),
"-map", "[v]", "-map", "[a]",
"-c:v", "libx264", "-pix_fmt", "yuv420p", "-preset", "veryfast", "-threads", "2",
"-c:a", "aac", "-b:a", "192k", "-movflags", "+faststart", str(output),
]
proc = subprocess.run(cmd, capture_output=True, timeout=180)
if proc.returncode != 0 or not output.exists() or not output.stat().st_size:
err = (proc.stderr or b"").decode("utf-8", errors="replace")[-500:]
raise RuntimeError(f"镜头拼接失败:{err or 'ffmpeg 未产出文件'}")
return output.read_bytes()
def _stamp_current_shot_id(task):
"""把刚提交火山拿到的 provider_task_id 写回当前这一镜,下一轮 poll 才找得到。"""
if task.status not in (AITask.Status.SUBMITTED, AITask.Status.POLLING) or not task.provider_task_id:
return task
payload = dict(task.request_payload or {})
plan = list(payload.get("shot_plan") or [])
changed = False
for item in plan:
if item.get("status") in {"submitted", "polling"} and not item.get("provider_task_id"):
item["provider_task_id"] = task.provider_task_id
changed = True
break
if not changed:
return task
payload["shot_plan"] = plan
task.request_payload = payload
task.save(update_fields=["request_payload", "updated_at"])
return task
def _active_shot(plan: list[dict]) -> dict | None:
for item in plan:
if item.get("status") in {"submitted", "polling"}:
return item
for item in plan:
if item.get("status") in {"queued", "", None}:
return item
return None
def _shot_by_index(plan: list[dict], index) -> dict | None:
try:
index = int(index)
except (TypeError, ValueError):
return None
return next((item for item in plan if int(item.get("index") or 0) == index), None)
def _read_shot_plan(task) -> list[dict]:
return list((task.request_payload or {}).get("shot_plan") or [])
def _write_shot(task, shot: dict):
with transaction.atomic():
locked = AITask.objects.select_for_update().get(id=task.id)
payload = dict(locked.request_payload or {})
plan = list(payload.get("shot_plan") or [])
plan = [
shot if int(item.get("index") or 0) == int(shot.get("index") or 0) else item
for item in plan
]
payload["shot_plan"] = plan
if shot.get("provider_task_id"):
locked.provider_task_id = shot["provider_task_id"]
locked.request_payload = payload
locked.save(update_fields=["request_payload", "provider_task_id", "updated_at"])
return locked
def _fail_replace_generation(task, message: str, *, error_code: str = "", hint: str = ""):
from apps.billing.services.ledger import release_credit
from .free_video import _fail_pending_free_video, _notify_failure
if task.status in (AITask.Status.CREATED, AITask.Status.RESERVED):
return _fail_reviewing_task(task, message, error_code=error_code or "processing_failed")
public_hint = hint or message
with transaction.atomic():
locked = AITask.objects.select_for_update().get(id=task.id)
if locked.status in (AITask.Status.SUCCEEDED, AITask.Status.FAILED, AITask.Status.CANCELLED):
return locked
locked.status = AITask.Status.FAILED
locked.error_code = (error_code or "CreateTaskError")[:64]
locked.error_message = message[:2000]
locked.completed_at = timezone.now()
locked.save(update_fields=["status", "error_code", "error_message", "completed_at", "updated_at"])
reservation = getattr(locked, "credit_reservation", None)
if reservation is not None:
release_credit(reservation=reservation, reason=message[:200])
_notify_failure(locked, raw=message, hint=public_hint)
return locked
def _legacy_replace_mode(prompt: str) -> str:
return "character" if prompt.startswith("[视频复刻·角色]") else "product"
def _fail_reviewing_task(task, message: str, *, error_code: str = ""):
"""收尾一个还没提交火山的任务。
``_fail_pending_free_video`` 把错误码写死成 content_rejected —— 那是审核不通过的语义,
前端会渲染成「内容未通过生成审核」且不可重试。拆解失败/素材丢失不是那回事,
这里按实际原因改码,否则模型抖一下会被误报成合规问题,用户白白去换素材。
"""
from .free_video import _fail_pending_free_video
task = _fail_pending_free_video(task, message)
if error_code and task.status == AITask.Status.FAILED and task.error_code != error_code:
task.error_code = error_code
task.save(update_fields=["error_code", "updated_at"])
return task
def _enqueue_replace_review_poll(task):
try:
from .tasks import poll_free_video_task
poll_free_video_task.apply_async(args=[str(task.id), 0], countdown=8)
except Exception: # noqa: BLE001
logger.error("video replace review poll enqueue failed; relying on client polling", exc_info=True)
def _enqueue_replace_digest(task):
try:
from .tasks import run_video_replace_digest_task
run_video_replace_digest_task.delay(str(task.id))
except Exception: # noqa: BLE001
logger.error("video replace digest enqueue failed", exc_info=True)
def _create_reviewing_task(*, team, user, params: dict, digest_pending: bool = False):
"""还不能提交火山时:只建 CREATED 任务,不预留积分。
商品/角色复刻都先拆解参考视频;参考图若还在审核,拆完再等绿盾。
"""
model_name = str(params.get("model") or REPLACE_MODEL)
aspect_ratio = str(params.get("aspect_ratio") or "9:16")
resolution = str(params.get("resolution") or "720p")
try:
duration = int(params.get("duration") or 5)
except (TypeError, ValueError):
raise ValueError("时长参数无效")
if model_name not in FREE_VIDEO_MODELS:
raise ValueError("模型无效")
if aspect_ratio not in RATIOS:
raise ValueError("画面比例无效")
if resolution not in RESOLUTIONS:
raise ValueError("分辨率无效")
low, high = replace_duration_range()
if not low <= duration <= high:
raise ValueError(f"视频时长需在 {low}-{high} 秒之间")
model_config = (
ModelConfig.objects.select_related("provider")
.filter(name=model_name, capability=ModelConfig.Capability.VIDEO, status=ModelConfig.Status.ACTIVE)
.first()
)
if model_config is None:
raise ValueError("视频模型未配置,请联系管理员")
_reap_stale_free_video_tasks(team=team)
max_concurrent = int(getattr(settings, "FREE_VIDEO_MAX_CONCURRENT", 3))
in_flight = AITask.objects.filter(
team=team, task_type=AITask.Type.FREE_VIDEO, status__in=IN_FLIGHT_STATUSES
).count()
if in_flight >= max_concurrent:
raise ValueError(f"当前有 {in_flight} 个视频任务进行中(上限 {max_concurrent}),请等待完成后再提交")
references = params.get("references") or []
tokens, quote = quote_video_estimate(
model_config,
aspect_ratio=aspect_ratio,
resolution=resolution,
duration=duration,
references=references,
team=team,
)
reserve_amount = video_reserve_amount(quote.points)
account = CreditAccount.objects.filter(team=team).first()
available = (account.balance - account.reserved_balance) if account else Decimal("0")
if available < reserve_amount:
raise ValueError("团队余额不足,请充值后重试")
try:
seed = int(params.get("seed") if params.get("seed") is not None else -1)
except (TypeError, ValueError):
seed = -1
extra = params.get("extra_payload") if isinstance(params.get("extra_payload"), dict) else {}
request_payload = {
"feature": FEATURE,
"mode": "universal",
"model": model_name,
"endpoint": model_config.endpoint,
"prompt": params.get("prompt") or "",
"api_prompt": "",
"aspect_ratio": aspect_ratio,
"resolution": resolution,
"duration": duration,
"seed": seed,
"generate_audio": True,
"search_mode": "off",
"estimated_tokens": tokens,
"price_multiplier": quote.meta.get("price_multiplier", "1"),
"points_per_yuan_snapshot": quote.meta.get("rate", ""),
"references": references,
"model_routing_v1": True,
"review_pending": True,
"digest_pending": digest_pending,
}
for key, value in extra.items():
if key in request_payload or value in (None, ""):
continue
request_payload[key] = value
task = AITask.objects.create(
team=team,
created_by=user,
project=None,
task_type=AITask.Type.FREE_VIDEO,
status=AITask.Status.CREATED,
model_config=model_config,
idempotency_key=f"free_video:{team.id}:{uuid.uuid4()}",
request_payload=request_payload,
estimated_cost=quote.points,
base_cost=Decimal("0"),
)
if digest_pending:
_enqueue_replace_digest(task)
else:
_enqueue_replace_review_poll(task)
return task
def _replace_ref_assets(team, references: list) -> list[tuple[dict, Asset]]:
out = []
seen = set()
for ref in references or []:
raw_id = ref.get("asset_id")
if not raw_id:
continue
try:
parsed = uuid.UUID(str(raw_id))
except (TypeError, ValueError):
continue
if parsed in seen:
continue
seen.add(parsed)
asset = _load_replace_asset(team, parsed, ref.get("label") or "参考素材")
out.append((ref, asset))
return out
def _load_replace_asset(team, asset_id: uuid.UUID, label: str) -> Asset:
asset = Asset.objects.filter(id=asset_id, is_deleted=False, purged_at__isnull=True).first()
if asset is None:
raise ValueError(f"{label}不存在或已被删除")
if asset.team_id == team.id:
return asset
if Model.objects.filter(Q(is_official=True), Q(portrait_asset=asset) | Q(triview_asset=asset)).exists():
return asset
raise ValueError(f"{label}不存在或已被删除")
def _ensure_replace_refs_reviewed(team, references: list) -> str:
"""送审/轮询全部参考素材。返回 ready / pending / failed;审核未配置且无 remote_id 抛错。"""
from apps.assets import assets_client
from apps.assets.review import poll_asset_review, submit_asset_for_review
pairs = _replace_ref_assets(team, references)
if not pairs:
raise ValueError("请先上传参考素材")
states = []
for ref, asset in pairs:
label = ref.get("label") or asset.name or "参考素材"
if asset.review_status == "active" and asset.review_remote_id:
states.append("ready")
continue
if not assets_client.is_enabled():
raise ValueError(REVIEW_UNAVAILABLE)
if asset.review_status == "processing" and asset.review_remote_id:
poll_asset_review(asset)
asset.refresh_from_db(fields=["review_status", "review_remote_id", "review_error"])
elif asset.review_status != "active" or not asset.review_remote_id:
ok = submit_asset_for_review(asset, force=True)
asset.refresh_from_db(fields=["review_status", "review_remote_id", "review_error"])
if not ok and not asset.review_remote_id:
# 带上火山给的原因(submit_asset_for_review 会写进 review_error),
# 否则用户只看到「提交审核失败」,不知道是图不合规、拉不到图还是服务抽风。
asset.refresh_from_db(fields=["review_error"])
reason = (asset.review_error or "").strip()
detail = f"{REVIEW_SUBMIT_FAILED}({reason})" if reason else REVIEW_SUBMIT_FAILED
raise ValueError(f"「{label}{detail}")
if asset.review_status == "processing" and asset.review_remote_id:
poll_asset_review(asset)
asset.refresh_from_db(fields=["review_status", "review_remote_id", "review_error"])
if asset.review_status == "active" and asset.review_remote_id:
states.append("ready")
elif asset.review_status == "failed":
states.append("failed")
else:
states.append("pending")
if any(state == "failed" for state in states):
return "failed"
if all(state == "ready" for state in states):
return "ready"
return "pending"
def _refresh_replace_refs(team, references: list) -> list:
"""同一团队走 source=asset;官方跨团队素材把过审 id 写成 resolved_url=asset://。"""
out = []
for ref in references or []:
item = dict(ref)
raw_id = item.get("asset_id")
if not raw_id:
out.append(item)
continue
try:
parsed = uuid.UUID(str(raw_id))
except (TypeError, ValueError):
out.append(item)
continue
try:
asset = _load_replace_asset(team, parsed, item.get("label") or "参考素材")
except ValueError:
out.append(item)
continue
if asset.team_id == team.id:
item["source"] = "asset"
item.pop("resolved_url", None)
elif asset.review_remote_id:
item["source"] = "upload"
item["resolved_url"] = f"asset://{asset.review_remote_id}"
out.append(item)
return out
def _assert_replace_refs_ready(references: list) -> None:
missing = []
for ref in references or []:
raw_id = ref.get("asset_id")
if not raw_id:
continue
try:
parsed = uuid.UUID(str(raw_id))
except (TypeError, ValueError):
continue
asset = Asset.objects.filter(id=parsed).first()
if asset is None or not asset.review_remote_id or asset.review_status != "active":
missing.append(ref.get("label") or "参考素材")
if missing:
raise ValueError("素材尚未完成合规审核,请稍后再试")
def _optional_uuid(value, label: str):
text = str(value or "").strip()
if not text:
return None
try:
return uuid.UUID(text)
except (TypeError, ValueError) as exc:
raise ValueError(f"{label}无效") from exc
def _uuid_list(value, label: str) -> list:
if value in (None, ""):
return []
if not isinstance(value, (list, tuple)):
raise ValueError(f"{label}格式无效")
if len(value) > MAX_IMAGES:
raise ValueError(f"{label}最多 {MAX_IMAGES} 张")
seen = set()
out = []
for item in value:
parsed = _optional_uuid(item, label)
if parsed is None or parsed in seen:
continue
seen.add(parsed)
out.append(parsed)
return out
def _team_asset(team, asset_id, *, kind: str, label: str) -> Asset:
parsed = _optional_uuid(asset_id, label)
if parsed is None:
raise ValueError(f"请先上传{label}")
asset = Asset.objects.filter(id=parsed, team=team, is_deleted=False, purged_at__isnull=True).first()
if asset is None:
raise ValueError(f"{label}不存在或已被删除")
if asset.asset_type != kind:
raise ValueError(f"{label}类型不正确")
return asset
def _asset_duration_seconds(asset: Asset) -> float:
primary = asset.files.filter(is_primary=True).first() or asset.files.first()
if primary is None or not primary.duration_ms:
return 0.0
return primary.duration_ms / 1000.0
def replace_duration_range() -> tuple[int, int]:
"""复刻模型(Seedance 2.5)支持的出片时长区间。模型没配好时回落 4–15,不至于炸。"""
model = (
ModelConfig.objects.filter(
name=REPLACE_MODEL, capability=ModelConfig.Capability.VIDEO, status=ModelConfig.Status.ACTIVE
).first()
)
return model_duration_range(model)
def _output_duration(requested, video_seconds: float) -> int:
low, high = replace_duration_range()
try:
value = int(requested) if requested not in (None, "") else 0
except (TypeError, ValueError):
value = 0
if value:
return min(high, max(low, value))
if video_seconds:
return min(high, max(low, int(round(video_seconds))))
return high
def _owned_ref(asset: Asset, *, kind: str, role: str, label: str) -> dict:
from .services import _asset_preview_url
url = _asset_preview_url(asset)
if not url:
raise ValueError(f"「{label}」没有可用文件")
ref = {
"url": url,
"type": kind,
"role": role,
"label": label,
"source": "asset",
"asset_id": str(asset.id),
}
seconds = _asset_duration_seconds(asset)
if seconds:
ref["duration"] = seconds
return ref
def _image_refs(references) -> list:
"""出片/审核只用图。用户原片哪怕还躺在旧 payload 里,也一律丢掉。"""
return [
dict(item)
for item in (references or [])
if (item or {}).get("type") != "video"
]
def _seedance_references(payload: dict) -> list:
"""出片只带商品图/角色图。用户上传的原片不传 Seedance,避免素材审核报错。"""
return _image_refs(payload.get("references") or [])
def _library_image_ref(asset: Asset, *, team, label: str) -> dict:
from .services import _asset_preview_url, _seedance_ref_url
if asset.team_id == team.id and not asset.is_deleted:
return {
"url": _asset_preview_url(asset) or "",
"type": "image",
"role": "reference_image",
"label": label,
"source": "asset",
"asset_id": str(asset.id),
}
raw = _asset_preview_url(asset)
url = _seedance_ref_url(raw, asset.review_status, asset.review_remote_id)
if not url:
raise ValueError(f"「{label}」没有可用文件")
return {
"url": url,
"type": "image",
"role": "reference_image",
"label": label,
"source": "upload",
"asset_id": str(asset.id),
}
def _product_triview_asset(team, product_id: uuid.UUID):
"""商品库里这个商品的三视图(白底多角度单张 16:9)。没有就返回 None,不挡生成。"""
return (
Asset.objects.filter(
team=team,
is_deleted=False,
purged_at__isnull=True,
metadata__product_id=str(product_id),
metadata__view="three_view",
)
.order_by("-created_at")
.first()
)
def _product_library_refs(team, product_id: uuid.UUID) -> tuple[str, list, str]:
product = (
Product.objects.filter(id=product_id, team=team, purged_at__isnull=True, status=Product.Status.ACTIVE)
.select_related("cover_asset")
.prefetch_related("images__asset")
.first()
)
if product is None:
raise ValueError("商品不存在或已被删除")
triview = _product_triview_asset(team, product_id)
# 三视图也占一张参考图的名额,商品图要给它让位,否则火山那边超 9 张直接拒。
image_budget = MAX_IMAGES - 1 if triview is not None else MAX_IMAGES
assets = []
seen = {triview.id} if triview is not None else set()
for image in product.images.all():
asset = image.asset
if asset is None or asset.id in seen or asset.is_deleted:
continue
seen.add(asset.id)
assets.append(asset)
if len(assets) >= image_budget:
break
if not assets and product.cover_asset_id and product.cover_asset_id not in seen and not product.cover_asset.is_deleted:
assets.append(product.cover_asset)
if not assets:
# 只有三视图、没有任何实拍图时,把三视图顶上来当 @目标商品。
# 否则提示词里的 @目标商品 找不到同名 label,火山拿到的是一句没有指代的字面量。
if triview is None:
raise ValueError("这个商品还没有可用图片")
assets = [triview]
triview = None
refs = [_library_image_ref(asset, team=team, label="目标商品" if index == 0 else f"目标商品{index + 1}") for index, asset in enumerate(assets)]
if triview is not None:
# 放最后:@目标商品 仍指向第一张实拍图,三视图作为「各面长什么样」的补充证据。
refs.append(_library_image_ref(triview, team=team, label=TRIVIEW_LABEL))
return product.title, refs, "library"
def _character_library_refs(team, model_id: uuid.UUID) -> tuple[str, list, str]:
model = (
Model.objects.filter(Q(team=team) | Q(is_official=True), id=model_id, is_deleted=False, purged_at__isnull=True)
.select_related("portrait_asset", "triview_asset")
.first()
)
if model is None:
raise ValueError("角色不存在或已被删除")
assets = []
seen = set()
for asset in (model.portrait_asset, model.triview_asset):
if asset is None or asset.id in seen or asset.is_deleted:
continue
seen.add(asset.id)
assets.append(asset)
if len(assets) >= MAX_IMAGES:
break
if not assets:
raise ValueError("这个角色还没有可用图片")
labels = ["目标角色", "目标角色三视图"]
refs = [_library_image_ref(asset, team=team, label=labels[index] if index < len(labels) else f"目标角色{index + 1}") for index, asset in enumerate(assets)]
return model.name, refs, "library"
def _temporary_image_refs(team, image_ids: list, *, noun: str) -> tuple[str, list, str]:
refs = []
for index, asset_id in enumerate(image_ids):
asset = _team_asset(team, asset_id, kind=Asset.Type.IMAGE, label=f"{noun}参考图")
label = "目标商品" if noun == "商品" else "目标角色"
if index > 0:
label = f"{label}{index + 1}"
refs.append(_owned_ref(asset, kind="image", role="reference_image", label=label))
fallback = "临时商品素材" if noun == "商品" else "临时角色素材"
name = Asset.objects.filter(id=image_ids[0]).values_list("name", flat=True).first() or fallback
subject = name.rsplit(".", 1)[0] if name else fallback
if len(refs) > 1:
subject = f"{subject}{len(refs)}张参考图)"
return subject, refs, "temporary"