azmat1127/yingqing
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

优化视频复刻

Browse Source
This commit is contained in:
Azmat@qq.com
2026-08-27 17:58:48 +08:00
parent d2b786a3cf
commit b3bf159e08
10 changed files with 545 additions and 64 deletions
Download Patch File Download Diff File Expand all files Collapse all files
core/backend/apps/ai/services.py
+15 -2
View File
@@ -901,6 +901,15 @@ NO_EMBEDDED_CAPTIONS_REQUIREMENT = (
"只保留商品包装上参考图中原有的真实物理文字与标识。旁白和环境音可以保留，但不要把声音转成画面文字。"
)
# 故事板图会作为 Seedance 的参考图。导演信息应由页面从结构化脚本渲染，
# 不能烧进参考图，否则边框、时间标签和旁白文字会被视频模型误当成待保留的画面元素。
STORYBOARD_VIDEO_KEYFRAME_REQUIREMENT = (
"【视频关键帧硬性规则】这张图将直接作为视频生成参考图：只生成一张干净、全幅、写实的 9:16 关键画面，"
"画面必须完整呈现本段最关键的角色、商品、场景、动作和空间关系。"
"禁止拼图、多宫格、故事板边框、时间轴、机位标注、旁白文字、标题、价格贴片、字幕、UI 或任何叠加文字；"
"只保留商品包装参考图中原有的真实物理文字与标识。"
)
def enforce_no_embedded_captions(prompt: str) -> str:
"""所有视频入口最终汇入这里，避免某个入口漏传“无字幕”导致模型自行加花字。"""
@@ -2710,7 +2719,9 @@ def build_storyboard_frame_prompt(project, segment, extra_prompt: str = "") -> s
补充=(("\n" + extra_prompt.strip()) if extra_prompt else ""),
)
cleaned = "\n".join(line for line in rendered.split("\n") if line.strip())
return enforce_no_embedded_captions(_apply_storyboard_output_ratio(cleaned, project))
return _apply_storyboard_output_ratio(
f"{cleaned}\n{STORYBOARD_VIDEO_KEYFRAME_REQUIREMENT}", project
)
def build_video_segment_prompt(project, video_segment, scene, refs, user_prompt: str = "") -> str:
@@ -2921,7 +2932,9 @@ def build_storyboard_frame_prompt_refs(project, segment, refs: list[dict], extra
补充=(("\n" + extra_prompt.strip()) if extra_prompt else ""),
)
cleaned = "\n".join(line for line in rendered.split("\n") if line.strip())
return enforce_no_embedded_captions(_apply_storyboard_output_ratio(cleaned, project))
return _apply_storyboard_output_ratio(
f"{cleaned}\n{STORYBOARD_VIDEO_KEYFRAME_REQUIREMENT}", project
)
def _is_transient_error(exc: Exception) -> bool:
core/backend/apps/ai/tasks.py
+13
View File
@@ -41,6 +41,19 @@ def run_video_digest_task(self, task_id: str) -> str:
return task_id
@app.task(bind=True, max_retries=0)
def run_video_replace_digest_task(self, task_id: str) -> str:
"""商品复刻的第一道工序(参考视频 → 分镜稿,Gemini 半分钟起)在 worker 内跑。
失败自己把任务收成 FAILED(见 run_replace_digest),故 max_retries=0,不向上抛重试。"""
from apps.ai.models import AITask
from apps.ai.video_replace import run_replace_digest
task = AITask.objects.select_related("team", "created_by", "model_config").filter(id=task_id).first()
if task is not None:
run_replace_digest(task)
return task_id
@app.task(bind=True, max_retries=0)
def generate_base_asset_task(self, task_id: str) -> str:
"""基础资产(商品/人物/场景立绘)的慢出图在 worker 内跑,Web 层不被占住。
core/backend/apps/ai/test_video_replace.py
+120 -20
View File
@@ -12,12 +12,65 @@ from apps.accounts.models import Team, TeamMember, User
from apps.ai.free_video import submit_free_video
from apps.ai.models import AITask, ModelConfig
from apps.ai.test_free_video import STANDARD, _ark_create_response
from apps.ai.video_replace import PRODUCT_PROMPT, REVIEW_FAILED, REVIEW_UNAVAILABLE, advance_video_replace, submit_video_replace
from apps.ai.video_replace import (
CHARACTER_PROMPT,
DIGEST_PENDING_PROMPT,
REVIEW_FAILED,
REVIEW_UNAVAILABLE,
advance_video_replace,
compact_digest_for_video,
run_replace_digest,
submit_video_replace,
)
from apps.assets.models import Asset, AssetFile, Model
from apps.billing.models import CreditAccount, CreditReservation
from apps.products.models import Product, ProductImage
DIGEST_SAMPLE = """片名：《晨间一杯》
视频类型：口播带货
画面比例：9:16 竖屏
时长：8秒
整体风格：写实、暖色调、清晨感
形式：口播
结构：场景种草
主线：女生早晨用旧牌精华后出门。
【镜头 01】
时间：00:00-00:04
时长：4秒
景别：中近景
机位：平视
运镜：固定
画面：女生站在窗边，把旧牌精华倒进玻璃杯。
人物动作：右手拿瓶，左手扶杯。
人物表情：轻松。
台词/旁白：每天早上我都要来一支旧牌精华。
音效：液体倒入声。
背景音乐：轻快钢琴。
字幕：每天早上来一支
备注：晨光从左侧打入。
【镜头 02】
时间：00:04-00:08
时长：4秒
景别：特写
机位：平视
运镜：缓慢推近
画面：手持瓶身正对镜头，标签清晰。
人物动作：转动瓶身。
人物表情：不可见。
台词/旁白：一支顶三支。
音效：无。
背景音乐：钢琴延续。
字幕：一支顶三支
备注：无。
【拆解存疑】
- 第 2 镜标签文字看不清，请校对品牌名。
"""
def _asset(team, user, *, kind=Asset.Type.IMAGE, name="素材", duration_ms=None, preview="http://tos/1.png", review_status="active", review_remote_id=None):
asset = Asset.objects.create(
team=team,
@@ -51,6 +104,7 @@ class SubmitVideoReplaceTests(TestCase):
self.provider.create_video_task.return_value = _ark_create_response()
patch("apps.ai.services.build_provider", return_value=self.provider).start()
patch("apps.ai.tasks.poll_free_video_task.apply_async").start()
patch("apps.ai.tasks.run_video_replace_digest_task.delay").start()
patch("apps.assets.assets_client.is_enabled", return_value=False).start()
self.addCleanup(patch.stopall)
self.video = _asset(self.team, self.user, kind=Asset.Type.VIDEO, name="参考.mp4", duration_ms=8000, preview="http://tos/video.mp4")
@@ -71,30 +125,72 @@ class SubmitVideoReplaceTests(TestCase):
params.update(over)
return submit_video_replace(team=self.team, user=self.user, params=params)
def test_product_library_uses_backend_prompt_and_feature(self):
def test_product_submit_waits_for_digest_and_keeps_video_out_of_refs(self):
"""商品复刻第一步:秒回 CREATED,参考视频只留作拆解源,不进火山 references。"""
task = self._submit()
self.assertEqual(task.status, AITask.Status.SUBMITTED)
self.assertEqual(task.status, AITask.Status.CREATED)
payload = task.request_payload
self.assertEqual(payload["feature"], "video_replace")
self.assertEqual(payload["replace_mode"], "product")
self.assertEqual(payload["subject_name"], "净颜精华")
self.assertEqual(payload["subject_source"], "library")
self.assertEqual(payload["prompt"], PRODUCT_PROMPT)
self.assertTrue(payload["digest_pending"])
self.assertEqual(payload["prompt"], DIGEST_PENDING_PROMPT)
self.assertEqual(payload["digest_source_asset_id"], str(self.video.id))
labels = [item["label"] for item in payload["references"]]
self.assertIn("参考视频", labels)
self.assertNotIn("参考视频", labels)
self.assertIn("目标商品", labels)
self.assertFalse(self.provider.create_video_task.called)
self.assertEqual(CreditReservation.objects.filter(task=task).count(), 0)
def test_product_digest_becomes_prompt_and_only_images_go_to_ark(self):
"""第二步:worker 拆完 → 分镜稿变提示词 → 只带商品图提交火山。"""
task = self._submit()
with patch(
"apps.ai.video_digest.digest_asset_video",
return_value=(DIGEST_SAMPLE, {"digest_shots": 2, "digest_model": "gemini-3.1-pro-preview"}),
):
run_replace_digest(task)
task.refresh_from_db()
self.assertEqual(task.status, AITask.Status.SUBMITTED)
payload = task.request_payload
self.assertFalse(payload["digest_pending"])
self.assertEqual(payload["digest_shots"], 2)
prompt = payload["prompt"]
self.assertIn("【镜头 01】", prompt) # 分镜稿骨架进了提示词
self.assertIn("倒进玻璃杯", prompt) # 画面描述保留
self.assertIn("@目标商品", prompt) # 商品图引用锚点在
self.assertIn("净颜精华", prompt) # 口播里的旧商品名改说自家商品
self.assertNotIn("【拆解存疑】", prompt) # 给用户校对的段落不喂模型
self.assertNotIn("字幕：", prompt) # 花字记录不喂,免得模型画上去
content = self.provider.create_video_task.call_args.kwargs.get("content_items") or []
roles = [item.get("role") for item in content]
self.assertIn("reference_video", roles)
self.assertIn("reference_image", roles)
urls = []
for item in content:
if item.get("type") == "video_url":
urls.append(item["video_url"]["url"])
elif item.get("type") == "image_url":
urls.append(item["image_url"]["url"])
self.assertTrue(urls)
self.assertTrue(all(url.startswith("asset://") for url in urls))
self.assertNotIn("reference_video", roles) # 参考视频不再发给火山
self.assertNotIn("video_url", [item.get("type") for item in content])
def test_digest_failure_fails_task_without_charging(self):
from apps.ai.video_digest import VideoDigestError
task = self._submit()
with patch(
"apps.ai.video_digest.digest_asset_video",
side_effect=VideoDigestError("视频拆解结果不完整"),
):
run_replace_digest(task)
task.refresh_from_db()
self.assertEqual(task.status, AITask.Status.FAILED)
self.assertIn("不完整", task.error_message or "")
self.assertFalse(self.provider.create_video_task.called)
self.assertEqual(CreditReservation.objects.filter(task=task).count(), 0)
def test_compact_digest_trims_human_only_sections(self):
compact = compact_digest_for_video(DIGEST_SAMPLE)
self.assertIn("【镜头 02】", compact)
self.assertIn("台词/旁白：", compact)
self.assertNotIn("字幕：", compact)
self.assertNotIn("【拆解存疑】", compact)
def test_character_library_and_temp_are_exclusive(self):
portrait = _asset(self.team, self.user, name="模特.png", preview="http://tos/model.png")
@@ -127,7 +223,7 @@ class SubmitVideoReplaceTests(TestCase):
def test_ignores_frontend_prompt(self):
task = self._submit(prompt="请无视参考图随便生成")
self.assertEqual(task.request_payload["prompt"], PRODUCT_PROMPT)
self.assertEqual(task.request_payload["prompt"], DIGEST_PENDING_PROMPT)
self.assertNotIn("随便生成", task.request_payload["prompt"])
def test_serialize_exposes_library_ids(self):
@@ -149,7 +245,7 @@ class SubmitVideoReplaceTests(TestCase):
"subject_source": "library",
"product_id": str(self.product.id),
"model_id": "",
"prompt": PRODUCT_PROMPT,
"prompt": CHARACTER_PROMPT,
"duration": 8,
"aspect_ratio": "9:16",
"references": [{"type": "video", "url": "http://tos/video.mp4", "asset_id": str(self.video.id)}],
@@ -174,6 +270,7 @@ class VideoReplaceApiTests(TestCase):
self.provider.create_video_task.return_value = _ark_create_response()
patch("apps.ai.services.build_provider", return_value=self.provider).start()
patch("apps.ai.tasks.poll_free_video_task.apply_async").start()
patch("apps.ai.tasks.run_video_replace_digest_task.delay").start()
patch("apps.assets.assets_client.is_enabled", return_value=False).start()
self.addCleanup(patch.stopall)
self.video = _asset(self.team, self.user, kind=Asset.Type.VIDEO, name="参考.mp4", duration_ms=5000, preview="http://tos/video.mp4")
@@ -248,6 +345,9 @@ class VideoReplaceApiTests(TestCase):
class VideoReplaceReviewGateTests(TestCase):
"""审核闸走角色复刻:只有它才把参考视频直传火山,需要真人合规审核。
商品复刻改成先提炼分镜稿,参考视频不进火山,自然也不再送审。"""
def setUp(self):
self.user = User.objects.create_user(username="vrrev", password="p")
self.team = Team.objects.create(name="VRR", owner=self.user)
@@ -256,6 +356,7 @@ class VideoReplaceReviewGateTests(TestCase):
self.provider.create_video_task.return_value = _ark_create_response()
patch("apps.ai.services.build_provider", return_value=self.provider).start()
patch("apps.ai.tasks.poll_free_video_task.apply_async").start()
patch("apps.ai.tasks.run_video_replace_digest_task.delay").start()
self.addCleanup(patch.stopall)
self.video = _asset(
self.team, self.user, kind=Asset.Type.VIDEO, name="真人.mp4", duration_ms=8000,
@@ -265,14 +366,13 @@ class VideoReplaceReviewGateTests(TestCase):
self.team, self.user, name="乔尔.png", preview="http://tos/face.png",
review_status="", review_remote_id="",
)
self.product = Product.objects.create(team=self.team, created_by=self.user, title="净颜精华", cover_asset=self.image)
ProductImage.objects.create(product=self.product, asset=self.image, is_primary=True)
self.model = Model.objects.create(team=self.team, name="乔尔", portrait_asset=self.image)
def _submit(self, **over):
params = {
"replace_mode": "product",
"replace_mode": "character",
"video_asset_id": str(self.video.id),
"product_id": str(self.product.id),
"model_id": str(self.model.id),
"model": STANDARD,
"aspect_ratio": "9:16",
"resolution": "480p",
core/backend/apps/ai/video_digest.py
+82
View File
@@ -1344,3 +1344,85 @@ def _fail_digest_task(task, reservation, message: str) -> None:
release_credit(reservation=reservation, reason=message[:200])
except Exception: # noqa: BLE001
pass
# --------------------------------------------------------------------------- #
# 复用入口:视频复刻·商品 拿分镜稿
# --------------------------------------------------------------------------- #
def digest_asset_video(*, asset, task, model_config_id=None) -> tuple[str, dict]:
"""把一条已入库的参考视频拆成中文分镜稿,供「视频复刻·商品」当提示词骨架。
与「提炼提示词」页共用同一份 SKILL.md、同一个 Gemini 3.1 Pro 和同一套质检,
保证两处产出一致;区别只在于这里不另建 VIDEO_DIGEST 任务、也不单独扣积分——
复刻本来就是一次收费,拆解是它的内部工序。模型调用的审计仍记在传入的复刻
task 上(AIModelAttempt),出问题查得到。
"""
from apps.ai.services import execute_routed_text_request
primary = asset.files.filter(is_primary=True).first() or asset.files.first()
if primary is None or not primary.object_key:
raise VideoDigestError("参考视频没有可用文件，请重新上传")
suffix = Path(primary.object_key).suffix.lower()
if suffix not in ALLOWED_SUFFIXES:
suffix = ".mp4"
model_config = resolve_digest_model_config(preferred_id=model_config_id)
if model_config is None:
raise VideoDigestError("视频复刻需要 Gemini 3.1 Pro(会看图)，当前没有启用，请联系管理员")
local_path = ""
try:
local_path = _download_source_to_temp(primary.object_key, suffix)
base_name = (asset.name or "参考视频").rsplit(".", 1)[0]
upload = _FileFromPath(local_path, f"{base_name}{suffix}")
video, frames, duration, extras = digest_input_from_upload(upload)
extras = extras or {}
logger.info(
"replace digest using %s:%s input=%s duration=%.1fs frames=%s",
model_config.provider.name,
model_config.name,
"native_video" if video is not None else "frames",
duration,
len(frames),
)
messages = build_digest_messages(
frames,
duration,
video=video,
aspect_ratio=ratio_label(int(extras.get("width") or 0), int(extras.get("height") or 0)),
file_title=title_from_filename(str(extras.get("file_name") or "")),
)
routed = execute_routed_text_request(
task=task,
primary_model=model_config,
messages=messages,
streaming=True,
structured_output=False,
business_operation="video_digest",
temperature=0.4,
validate_text=lambda text: validate_digest_text(
text, duration=duration, frame_count=len(frames) or (24 if video else 0)
),
extra_body={"max_tokens": DIGEST_MAX_TOKENS},
request_summary={
"duration_seconds": round(duration, 2),
"frame_count": len(frames),
"input": "native_video" if video is not None else "frames",
"for": "video_replace",
},
allow_retry=False,
allow_fallback=False,
)
_text, _response, digest = routed.value
meta = {
"digest_model": model_config.name,
"digest_input": "native_video" if video is not None else "frames",
"digest_frames": len(frames),
"digest_duration": round(duration, 2),
"digest_shots": shot_count(digest),
"digest_ratio": ratio_label(int(extras.get("width") or 0), int(extras.get("height") or 0)),
}
return digest, meta
finally:
if local_path:
Path(local_path).unlink(missing_ok=True)
core/backend/apps/ai/video_replace.py
+191 -27
View File
@@ -44,12 +44,34 @@ REVIEW_UNAVAILABLE = "素材审核服务暂不可用，请稍后重试"
REVIEW_FAILED = "参考素材未通过真人合规审核，请更换视频或图片后重试"
REVIEW_SUBMIT_FAILED = "素材提交审核失败，请稍后重试"
PRODUCT_PROMPT = (
"使用@参考视频作为镜头、节奏与口播氛围基准，"
"将画面中需要替换的原商品完整替换为@目标商品的外观。"
"保留参考视频的人物、场景、镜头运动、剪辑节奏与口播氛围，"
"商品外观、材质、包装必须与参考图一致，不要改变原片构图和人物表演。"
# 商品复刻不再把参考视频直接丢给火山(实测换不动商品,出片跑偏),改走两步:
# 先用「提炼提示词」那一整套(同一份 SKILL.md + 同一个 Gemini 3.1 Pro)把参考视频拆成
# 中文分镜稿,再把分镜稿当导演脚本、连同商品图一起交给 Seedance。
# 角色复刻仍是「参考视频 + 角色图」直传,已验证有效,不要顺手一起改。
PRODUCT_DIGEST_HEAD = (
"下面是参考视频的完整分镜稿，它逐镜记录了这条片子的时间、景别、机位、运镜、画面、"
"人物动作与表情、台词旁白和声音。请把它当作导演脚本，按镜头顺序还原成一条新视频。"
)
PRODUCT_DIGEST_TAIL = (
"【复刻要求】\n"
"1. 严格按上面分镜稿的镜头顺序、时间分配、景别、机位、运镜和剪辑节奏还原全片，不要自行增删镜头。\n"
"2. 分镜稿里出现的原商品，全部替换成 @目标商品；商品的外形、颜色、材质、包装和标识必须与参考图完全一致，"
"不要改造型、不要改配色、不要凭空补细节。\n"
"3. 人物、场景、光线、色调、动作和表情按分镜稿保持不变，只换商品。\n"
"4. 台词/旁白按分镜稿逐字念出；原文提到旧商品名称的地方，改说「{subject}」。\n"
"5. 分镜稿里的「字幕」栏只是对原片的记录，不要把这些文字画到画面上。"
)
# 拆解期间的占位提示词。真正的提示词在 worker 拆完后回写。
DIGEST_PENDING_PROMPT = "正在提炼参考视频的分镜稿…"
# 分镜稿是给人逐镜校对的导演稿,整篇塞给视频模型太长也太杂,按需瘦身:
# ·「字幕」栏是原片花字的记录,留着会诱导模型把字画到画面上,与全站「成片无字幕」冲突;
# ·【拆解存疑】是给用户校对用的,视频模型用不上。
DIGEST_DROP_FIELDS = ("字幕",)
# 瘦身完还超长时再砍这几栏(信息量最低,砍掉不影响镜头结构)。
DIGEST_TRIM_FIELDS = ("音效", "背景音乐", "备注")
DIGEST_TAIL_SECTION = "【拆解存疑】"
MAX_DIGEST_CHARS = 4000
CHARACTER_PROMPT = (
"使用@参考视频作为镜头、节奏与口播氛围基准，"
"将画面中需要替换的原人物完整替换为@目标角色。"
@@ -58,6 +80,39 @@ CHARACTER_PROMPT = (
)
def _drop_digest_fields(lines: list[str], fields: tuple[str, ...]) -> list[str]:
prefixes = tuple(f"{name}：" for name in fields)
return [line for line in lines if not line.strip().startswith(prefixes)]
def compact_digest_for_video(digest_text: str) -> str:
"""分镜稿 → 交给视频模型的精简版。保留镜头结构,砍掉给人看的部分。"""
body = (digest_text or "").strip()
head, sep, _tail = body.partition(DIGEST_TAIL_SECTION)
if sep:
body = head.strip()
lines = _drop_digest_fields(body.split("\n"), DIGEST_DROP_FIELDS)
if sum(len(line) for line in lines) > MAX_DIGEST_CHARS:
lines = _drop_digest_fields(lines, DIGEST_TRIM_FIELDS)
return "\n".join(line for line in lines if line.strip()).strip()
def build_product_replace_prompt(digest_text: str, subject_name: str) -> str:
"""分镜稿 + 商品替换要求 → 交给 Seedance 的完整提示词。
@目标商品 必须与 references 里第一张商品图的 label 对上,否则 build_content_items
不会把它换成火山认的「图片N」指代。
"""
from .services import enforce_no_embedded_captions
body = compact_digest_for_video(digest_text)
if not body:
raise ValueError("参考视频拆解结果为空，请重试")
subject = (subject_name or "").strip() or "目标商品"
tail = PRODUCT_DIGEST_TAIL.format(subject=subject)
return enforce_no_embedded_captions(f"{PRODUCT_DIGEST_HEAD}\n\n{body}\n\n{tail}")
def is_video_replace_task(task) -> bool:
payload = task.request_payload or {}
if payload.get("feature") == FEATURE:
@@ -73,7 +128,9 @@ def serialize_video_replace_task(task, *, include_deleted_assets: bool = False)
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 "")
reviewing = task.status == AITask.Status.CREATED and bool(payload.get("review_pending"))
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
data.update({
"feature": FEATURE,
"replace_mode": replace_mode,
@@ -82,6 +139,12 @@ def serialize_video_replace_task(task, *, include_deleted_assets: bool = False)
"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 0,
"digest_source_name": payload.get("digest_source_name") or "",
"digest_source": payload.get("digest_source_ref") or None,
})
return data
@@ -127,8 +190,48 @@ def submit_video_replace(*, team, user, params: dict):
noun = "商品" if replace_mode == "product" else "角色"
subject_name, image_refs, subject_source = _temporary_image_refs(team, image_ids, noun=noun)
prompt = PRODUCT_PROMPT if replace_mode == "product" else CHARACTER_PROMPT
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",
"model": str(params.get("model") or HIGH_RES_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,
}
if replace_mode == "product":
# 商品复刻:参考视频只喂给提炼模型,不进火山 references(所以也不用送火山审核)。
# 先秒回一个 CREATED 任务,拆解这种慢活(Gemini 半分钟起)交给 worker。
extra.update({
"digest_source_asset_id": str(video.id),
"digest_source_name": video.name or "参考视频",
"digest_source_duration": round(video_seconds, 2),
# 参考视频虽然不进 references,历史卡的「原视频」和「重新生成」仍要拿到它,
# 这里存一份引用快照,免得序列化历史时每条再查一次库。
"digest_source_ref": _owned_ref(video, kind="video", role="reference_video", label="参考视频"),
})
return _create_reviewing_task(
team=team,
user=user,
params={
**base_params,
"prompt": DIGEST_PENDING_PROMPT,
"references": image_refs,
"extra_payload": extra,
},
digest_pending=True,
)
references = [
_owned_ref(video, kind="video", role="reference_video", label="参考视频"),
*image_refs,
@@ -137,25 +240,11 @@ def submit_video_replace(*, team, user, params: dict):
if review_state == "failed":
raise ValueError(REVIEW_FAILED)
references = _refresh_replace_refs(team, references)
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 "",
"review_pending": review_state != "ready",
}
extra["review_pending"] = review_state != "ready"
submit_params = {
"prompt": prompt,
"mode": "universal",
"model": str(params.get("model") or HIGH_RES_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,
**base_params,
"prompt": CHARACTER_PROMPT,
"references": references,
"feature": FEATURE,
"extra_payload": extra,
}
if review_state == "ready":
@@ -174,6 +263,10 @@ def advance_video_replace(task):
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 = list(payload.get("references") or [])
try:
state = _ensure_replace_refs_reviewed(task.team, references)
@@ -202,6 +295,61 @@ def advance_video_replace(task):
return start_pending_free_video(task)
def run_replace_digest(task) -> None:
"""Worker:商品复刻第一道工序——参考视频 → 分镜稿 → 完整提示词 → 走审核提交火山。
这一步失败 = 整条复刻失败。此时还没预留视频积分(CREATED 阶段不预留),不扣费。
"""
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, "参考视频已失效，请重新上传")
return
try:
digest, meta = digest_asset_video(asset=asset, task=task)
prompt = build_product_replace_prompt(digest, str(payload.get("subject_name") or ""))
except (VideoDigestError, ValueError) as exc:
_fail_reviewing_task(task, str(exc))
return
except Exception as exc: # noqa: BLE001 — 拆解任何异常都要把任务收尾,别留 CREATED 僵尸
logger.exception("video replace digest failed for %s", task.id)
_fail_reviewing_task(task, f"参考视频拆解失败：{exc}")
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
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 _legacy_replace_mode(prompt: str) -> str:
return "character" if prompt.startswith("[视频复刻·角色]") else "product"
@@ -221,8 +369,20 @@ def _enqueue_replace_review_poll(task):
logger.error("video replace review poll enqueue failed; relying on client polling", exc_info=True)
def _create_reviewing_task(*, team, user, params: dict):
"""审核未完成:只建 CREATED 任务,不预留积分。"""
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 HIGH_RES_MODEL)
aspect_ratio = str(params.get("aspect_ratio") or "9:16")
resolution = str(params.get("resolution") or "720p")
@@ -294,6 +454,7 @@ def _create_reviewing_task(*, team, user, params: dict):
"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, ""):
@@ -312,7 +473,10 @@ def _create_reviewing_task(*, team, user, params: dict):
estimated_cost=quote.points,
base_cost=Decimal("0"),
)
_enqueue_replace_review_poll(task)
if digest_pending:
_enqueue_replace_digest(task)
else:
_enqueue_replace_review_poll(task)
return task
core/backend/apps/ai/views.py
+3
View File
@@ -863,6 +863,9 @@ class VideoReplaceView(APIView):
def post(self, request):
require_worker()
# 商品复刻的拆解在 worker 里跑;worker 还是旧镜像的话消息会被静默丢弃 → 永久「提炼中」。
if str((request.data or {}).get("replace_mode") or "") == "product":
require_worker_task("apps.ai.tasks.run_video_replace_digest_task")
from .video_replace import serialize_video_replace_task, submit_video_replace
team = get_current_team(request.user)