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-31 10:59:54 +08:00
parent 284748eeff
commit f59e9d0418
20 changed files with 775 additions and 212 deletions
Download Patch File Download Diff File Expand all files Collapse all files
core/backend/apps/ai/test_video_replace.py
+49 -4
View File
@@ -119,6 +119,32 @@ DIGEST_FOUR = """片名：加长
台词/旁白：第四句。
"""
SEMANTIC_SAMPLE = """【成片规则】
参考视频仅保留环境、镜头语言与节奏。@目标商品仅作为新商品外观和包装参考；不得出现原商品、原动作或原台词。
【镜头 01】
时间：00:00-00:04
时长：4秒
景别：中近景
机位：平视
运镜：固定
画面：女生站在窗边，拿起@目标商品观察包装。
人物动作：双手展示商品外观。
人物表情：轻松。
台词/旁白：这是净颜精华。
【镜头 02】
时间：00:04-00:08
时长：4秒
景别：特写
机位：平视
运镜：缓慢推近
画面：镜头拍摄@目标商品包装细节。
人物动作：缓慢转动商品。
人物表情：不可见。
台词/旁白：包装细节清晰可见。
"""
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(
@@ -155,6 +181,8 @@ class SubmitVideoReplaceTests(TestCase):
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()
# 复刻的商品语义规划属于独立文本调用；其余用例只验证编排与出片边界。
patch("apps.ai.video_replace.rewrite_product_semantic_remix", return_value=SEMANTIC_SAMPLE).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")
self.image = _asset(self.team, self.user, name="精华.png", preview="http://tos/product.png")
@@ -209,8 +237,8 @@ class SubmitVideoReplaceTests(TestCase):
prompt = payload["prompt"]
self.assertIn("【镜头 01】", prompt)
self.assertIn("【镜头 02】", prompt)
self.assertIn("倒进玻璃杯", prompt)
self.assertIn("转动瓶身", prompt)
self.assertIn("观察包装", prompt)
self.assertIn("转动商品", prompt)
self.assertIn("台词/旁白", prompt)
self.assertIn("@目标商品", prompt)
self.assertIn("净颜精华", prompt)
@@ -449,6 +477,23 @@ class SubmitVideoReplaceTests(TestCase):
task.refresh_from_db()
return task
def test_product_semantic_remix_uses_facts_not_source_product_actions(self):
from apps.ai.video_replace import build_product_semantic_remix_messages, validate_product_semantic_remix
messages = build_product_semantic_remix_messages(
digest_text=DIGEST_SAMPLE,
product_facts={"商品名称": "蓝牙耳机", "商品品类": "数码家电", "商品描述": "入耳式耳机", "真实卖点": [{"标题": "便携", "说明": ""}]},
duration=8,
aspect_ratio="9:16",
)
request = messages[1]["content"]
self.assertIn("蓝牙耳机", request)
self.assertIn("旧牌精华", request) # 参考动作仅作为要剥离的输入
self.assertIn("商品图片只用于外观", messages[0]["content"])
self.assertEqual(validate_product_semantic_remix(SEMANTIC_SAMPLE), SEMANTIC_SAMPLE.strip())
with self.assertRaisesMessage(ValueError, "商品资料不足"):
validate_product_semantic_remix("MISSING_PRODUCT_FACTS：缺少商品品类")
def test_product_triview_is_sent_with_the_video(self):
"""商品库选商品时,三视图要跟着一起发给火山,并在提示词里被点名。"""
tri = self._make_triview()
@@ -617,8 +662,8 @@ class SubmitVideoReplaceTests(TestCase):
self.assertFalse(task.request_payload.get("shot_plan"))
prompt = task.request_payload["prompt"]
self.assertIn("【镜头 01】", prompt)
self.assertIn("第四镜收束", prompt)
self.assertIn("省略次要镜头", prompt)
self.assertIn("【镜头 02】", prompt)
self.assertNotIn("第一镜开场", prompt)
self.assertEqual(self.provider.create_video_task.call_count, 1)
self.assertEqual(self.provider.create_video_task.call_args.kwargs.get("duration"), 10)
core/backend/apps/ai/video_replace.py
+125 -6
View File
@@ -9,10 +9,12 @@
"""
from __future__ import annotations
import json
import logging
import re
import uuid
from decimal import Decimal
from pathlib import Path
from django.conf import settings
from django.db import transaction
@@ -80,9 +82,8 @@ def get_inflight_video_replace(team):
)
PRODUCT_PROMPT = (
"按下面这份分镜稿生成视频。镜头数量、每镜时长、景别、机位、运镜、人物、场景、口播全部按稿执行，"
"不要增删镜头、不要改顺序。"
"把稿里的原商品换成@目标商品，外观、材质、包装以参考图为准。"
"按下面的商品语义重构分镜生成视频。@目标商品只用于锁定外观、材质与包装。"
"严格遵守每镜的新商品动作、场景与口播；不要恢复参考片原商品、原动作、原卖点或原台词。"
)
CHARACTER_PROMPT = (
"按下面这份分镜稿生成视频。镜头数量、每镜时长、景别、机位、运镜、商品、场景、口播全部按稿执行，"
@@ -344,6 +345,99 @@ def _digest_for_seedance(digest_text: str) -> str:
return (head if sep else body).strip()
def _semantic_skill_dir() -> Path:
"""语义重构 skill 必须在 backend/skills 中，确保镜像内也能加载。"""
base = Path(settings.BASE_DIR)
for candidate in (base / "skills", base.parent.parent / "skills"):
skill = candidate / "product-semantic-video-remix" / "SKILL.md"
if skill.is_file():
return skill
return base / "skills" / "product-semantic-video-remix" / "SKILL.md"
def load_product_semantic_remix_skill() -> str:
path = _semantic_skill_dir()
try:
return path.read_text(encoding="utf-8")
except OSError as exc:
raise ValueError("商品语义重构规则未部署，请联系管理员") from exc
def _product_semantic_facts(product: Product) -> dict:
"""只取商品库中已填写的事实，绝不从商品图片猜用途或功效。"""
selling_points = [
{"标题": point.title.strip(), "说明": point.detail.strip()}
for point in product.selling_points.all()
if point.title.strip() or point.detail.strip()
]
return {
"商品名称": product.title.strip(),
"品牌": product.brand.strip(),
"商品品类": product.category.strip(),
"商品描述": product.description.strip(),
"真实卖点": selling_points,
"目标用户": product.target_audience.strip(),
"规格": product.specs or {},
"资料状态": "完整" if (product.category.strip() and (product.description.strip() or selling_points)) else "不足",
}
def build_product_semantic_remix_messages(*, digest_text: str, product_facts: dict, duration: int, aspect_ratio: str) -> list[dict]:
"""参考片只作为结构输入；商品事实和参考片信息严格分栏，防止语义串台。"""
facts = json.dumps(product_facts, ensure_ascii=False, indent=2)
return [
{"role": "system", "content": load_product_semantic_remix_skill()},
{
"role": "user",
"content": (
"请执行商品语义重构，不要直接把参考分镜改几个名词。\n"
f"成片时长：{duration} 秒；画面比例：{aspect_ratio}。\n\n"
"【新商品的唯一事实来源】\n"
f"{facts}\n\n"
"【参考视频拆解稿，仅可继承角色关系、环境、镜头语言、剪辑节奏和叙事功能】\n"
f"{_digest_for_seedance(digest_text)}\n\n"
"输出给视频模型的中文分镜稿：保留【镜头 01】格式和每镜时间、景别、机位、运镜；"
"每镜的画面、人物动作、台词/旁白必须按新商品真实用途重写。"
"不要输出分析过程、分类标签或 Markdown。"
),
},
]
def validate_product_semantic_remix(text: str) -> str:
"""拒绝模型在事实不足时瞎编，或只给一段泛泛分析而没有可出片分镜。"""
cleaned = (text or "").strip()
if "MISSING_PRODUCT_FACTS" in cleaned:
raise ValueError("商品资料不足：请先在商品库补充品类，以及商品描述或至少一条真实卖点后再复刻")
if len(cleaned) < 120 or "【镜头" not in cleaned:
raise ValueError("商品语义重构结果不完整，请重试")
return cleaned
def rewrite_product_semantic_remix(*, task, digest_text: str, product_facts: dict, duration: int, aspect_ratio: str) -> str:
"""Gemini 负责重写商品语义，Seedance 只负责按这份新分镜出片。"""
from .video_digest import DIGEST_MAX_TOKENS, resolve_digest_model_config
from .services import _collect_extract_text, get_text_provider
model_config = resolve_digest_model_config()
if model_config is None:
raise ValueError("商品语义重构模型未配置，请联系管理员")
provider = get_text_provider(model_config)
text, _payload = _collect_extract_text(
provider,
model_config,
build_product_semantic_remix_messages(
digest_text=digest_text,
product_facts=product_facts,
duration=duration,
aspect_ratio=aspect_ratio,
),
temperature=0.2,
extra_body={"max_tokens": DIGEST_MAX_TOKENS},
)
return validate_product_semantic_remix(text)
def build_character_replace_prompt(
digest_text: str,
*,
@@ -497,11 +591,18 @@ def submit_video_replace(*, team, user, params: dict):
if has_product:
subject_name, image_refs, subject_source = _product_library_refs(team, product_id)
product_facts = _product_semantic_facts_for_id(team, product_id)
elif has_model:
subject_name, image_refs, subject_source = _character_library_refs(team, model_id)
product_facts = {}
else:
noun = "商品" if replace_mode == "product" else "角色"
subject_name, image_refs, subject_source = _temporary_image_refs(team, image_ids, noun=noun)
product_facts = {
"商品名称": subject_name,
"资料状态": "不足",
"说明": "临时上传图片只提供外观，未提供品类、用途或真实卖点。",
}
duration = _output_duration(params.get("duration"), video_seconds)
extra = {
@@ -510,6 +611,7 @@ def submit_video_replace(*, team, user, params: dict):
"subject_source": subject_source,
"product_id": str(product_id) if product_id else "",
"model_id": str(model_id) if model_id else "",
"product_facts": product_facts,
}
base_params = {
"mode": "universal",
@@ -643,10 +745,15 @@ def run_replace_digest(task) -> None:
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,
prompt = rewrite_product_semantic_remix(
task=task,
digest_text=digest,
product_facts=payload.get("product_facts") or {"商品名称": subject, "资料状态": "不足"},
duration=output_seconds,
aspect_ratio=str(payload.get("aspect_ratio") or "9:16"),
)
if has_triview and TRIVIEW_LABEL not in prompt:
prompt = f"{prompt}\n商品各面与材质细节以@{TRIVIEW_LABEL}为准。"
except (VideoDigestError, ValueError) as exc:
_fail_reviewing_task(task, str(exc), error_code="processing_failed")
return
@@ -663,6 +770,7 @@ def run_replace_digest(task) -> None:
next_payload.update(meta)
next_payload["digest_pending"] = False
next_payload["digest_text"] = digest[:32000]
next_payload["semantic_remix"] = replace_mode == "product"
next_payload["prompt"] = prompt
next_payload["references"] = _image_refs(next_payload.get("references") or [])
next_payload.pop("shot_plan", None)
@@ -1583,6 +1691,17 @@ def _product_triview_asset(team, product_id: uuid.UUID):
)
def _product_semantic_facts_for_id(team, product_id: uuid.UUID) -> dict:
product = (
Product.objects.filter(id=product_id, team=team, purged_at__isnull=True, status=Product.Status.ACTIVE)
.prefetch_related("selling_points")
.first()
)
if product is None:
raise ValueError("商品不存在或已被删除")
return _product_semantic_facts(product)
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)