seaislee1209andClaude Opus 4.8
4c21d85ba1
- 进度区改竖线时间轴:推理模型 reasoning_content 逐字转发(volcano),状态字=模型实时最新一句+秒数+折叠展开思考全文(替代旧 ProgressStream/ThinkingStream) - 场景基础资产出图改 16:9(run_base_asset_task size + 提取提示词横屏) - 已出图立绘主卡不再被在途轮询任务误盖转圈 - 审核态随项目详情持久下发(BaseAssetGroup.adopted_asset_review),刷新不丢徽章 - 设定卡加「← 返回」回三选项;模型选择器固定按 created_at 排序(默认豆包2.0Pro) - 新增 reasoning 转发回归测试 Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
259 lines
15 KiB
Python
259 lines
15 KiB
Python
import json
|
|
|
|
from django.test import SimpleTestCase
|
|
|
|
from apps.ai.script_agent import normalize_draft
|
|
|
|
|
|
class NormalizeDraftTests(SimpleTestCase):
|
|
"""normalize_draft 对模型不按契约输出的容错(防「旁白/画面全空」回归)。"""
|
|
|
|
def test_scriptdraft_shots_variant_fills_narration_and_visual(self):
|
|
"""模型常见变体:{"ScriptDraft":{"basicInfo":..,"shots":[{scene,dialogue,subtitle}]}}。
|
|
必须解开外壳 + 把 shots→segments、scene→画面、dialogue/subtitle→旁白,而非全填空占位镜。"""
|
|
raw = json.dumps({
|
|
"ScriptDraft": {
|
|
"basicInfo": {"totalDuration": 60, "aspectRatio": "9:16", "totalShots": 4},
|
|
"shots": [
|
|
{"shotNo": 1, "duration": 15, "scene": "更衣室扯卡裆旧裤", "dialogue": "卡裆太社死！", "subtitle": "还在卡裆？"},
|
|
{"shotNo": 2, "duration": 15, "scene": "特写拉扯面料回弹", "dialogue": "高弹不变形！", "subtitle": "裸感面料"},
|
|
{"shotNo": 3, "duration": 15, "scene": "健身切通勤", "dialogue": "都能穿！", "subtitle": "一裤多穿"},
|
|
{"shotNo": 4, "duration": 15, "scene": "对镜弹小黄车", "dialogue": "点小黄车抢！", "subtitle": "点击入手"},
|
|
],
|
|
}
|
|
}, ensure_ascii=False)
|
|
draft = normalize_draft(raw, aspect_ratio="9:16", total_duration=60)
|
|
self.assertEqual(len(draft["segments"]), 4)
|
|
self.assertEqual([s["role"] for s in draft["segments"]], ["钩子", "痛点", "卖点", "CTA"])
|
|
for seg in draft["segments"]:
|
|
self.assertTrue(seg["narration"], "旁白不应为空")
|
|
self.assertTrue(seg["visual"], "画面不应为空")
|
|
self.assertEqual(draft["segments"][0]["narration"], "卡裆太社死！")
|
|
self.assertEqual(draft["segments"][0]["visual"], "更衣室扯卡裆旧裤")
|
|
|
|
def test_lines_and_caption_variant_fills_narration(self):
|
|
"""另一种变体(Doubao-Seed-2.0-P):shots 用 lines:[{role,content}] 放口播、caption 放字幕。
|
|
必须把 lines 的 content 当对白/旁白,caption 兜底,而不是只填画面留旁白空。"""
|
|
raw = json.dumps({
|
|
"ScriptDraft": {
|
|
"basicInfo": {"totalDuration": 30, "aspectRatio": "9:16"},
|
|
"shots": [
|
|
{"shotNo": 1, "scene": "化妆台两闺蜜", "lines": [{"role": "女主", "content": "防晒泛白太尴尬！"}, {"role": "闺蜜", "content": "试试这个！"}], "caption": "防晒踩雷？"},
|
|
{"shotNo": 2, "scene": "手背挤膏体", "lines": [{"role": "闺蜜", "content": "质地清透不泛白"}], "caption": "清透质地"},
|
|
],
|
|
}
|
|
}, ensure_ascii=False)
|
|
draft = normalize_draft(raw, aspect_ratio="9:16", total_duration=30)
|
|
self.assertEqual(len(draft["segments"]), 2)
|
|
for seg in draft["segments"]:
|
|
self.assertTrue(seg["narration"], "旁白不应为空")
|
|
self.assertTrue(seg["visual"], "画面不应为空")
|
|
self.assertIn("防晒泛白太尴尬", draft["segments"][0]["narration"])
|
|
self.assertEqual(len(draft["segments"][0]["dialogue"]), 2) # lines→结构化对白
|
|
|
|
def test_generic_resolver_covers_unseen_field_names(self):
|
|
"""模型每次换字段名(scene/screenDescription/画面…、dialogue/lines/caption…)。
|
|
通用解析应「优先键 + 关键词模糊匹配」都能填,且跳过 bgMusic/note 等非内容键。"""
|
|
variants = {
|
|
"canonical": {"total_duration": 15, "segments": [{"role": "钩子", "narration": "口播A", "visual": "画面A"}]},
|
|
"scene+dialogue_str+subtitle": {"total_duration": 15, "shots": [{"scene": "画面B", "dialogue": "口播B", "subtitle": "字幕B"}]},
|
|
"scene+lines+caption": {"total_duration": 15, "shots": [{"scene": "画面C", "lines": [{"role": "女主", "content": "口播C"}], "caption": "字幕C"}]},
|
|
"screenDescription+dialogue+bgMusic+note": {"total_duration": 15, "shots": [{"screenDescription": "画面D", "dialogue": "口播D", "bgMusic": "音乐D", "note": "备注D"}]},
|
|
}
|
|
for name, raw in variants.items():
|
|
draft = normalize_draft(json.dumps(raw, ensure_ascii=False), aspect_ratio="9:16", total_duration=15)
|
|
seg = draft["segments"][0]
|
|
self.assertTrue(seg["narration"], f"{name}: 旁白为空")
|
|
self.assertTrue(seg["visual"], f"{name}: 画面为空")
|
|
# bgMusic/note 不得被误当画面/旁白
|
|
d = normalize_draft(json.dumps(variants["screenDescription+dialogue+bgMusic+note"], ensure_ascii=False), aspect_ratio="9:16", total_duration=15)
|
|
self.assertEqual(d["segments"][0]["visual"], "画面D")
|
|
self.assertEqual(d["segments"][0]["narration"], "口播D")
|
|
|
|
def test_string_dialogue_not_iterated_as_chars(self):
|
|
"""dialogue 为整句字符串时,要当作旁白而不是逐字符遍历。"""
|
|
raw = json.dumps({
|
|
"segments": [{"role": "钩子", "dialogue": "一句完整口播", "visual": "画面"}],
|
|
"total_duration": 15,
|
|
}, ensure_ascii=False)
|
|
draft = normalize_draft(raw, aspect_ratio="9:16", total_duration=15)
|
|
self.assertEqual(draft["segments"][0]["narration"], "一句完整口播")
|
|
self.assertEqual(draft["segments"][0]["dialogue"], []) # 字符串不进结构化对白
|
|
|
|
def test_empty_segments_skeleton_yields_to_richer_scenes(self):
|
|
"""真实回归(全空根因):模型把内容放 scenes/voiceover/visual,却另给一个**空 segments 骨架**。
|
|
必须挑内容最丰富的数组(scenes),而不是见 segments 是 list 就用→落 4 个空镜。"""
|
|
raw = json.dumps({
|
|
"scenes": [
|
|
{"sceneIndex": 1, "sceneTitle": "通勤", "voiceover": "手机一卡效率掉线", "visual": "地铁口拿出亮银色机身"},
|
|
{"sceneIndex": 2, "sceneTitle": "办公", "voiceover": "A19多任务很跟手", "visual": "办公桌俯拍切换应用"},
|
|
],
|
|
"segments": [{"index": 0, "role": "钩子", "narration": "", "visual": ""}, {"index": 1, "role": "痛点", "narration": "", "visual": ""}],
|
|
"total_duration": 30,
|
|
}, ensure_ascii=False)
|
|
draft = normalize_draft(raw, aspect_ratio="9:16", total_duration=30)
|
|
self.assertEqual(draft["segments"][0]["narration"], "手机一卡效率掉线")
|
|
self.assertEqual(draft["segments"][0]["visual"], "地铁口拿出亮银色机身")
|
|
|
|
def test_script_key_with_visual_object_is_flattened(self):
|
|
"""真实回归:数组键叫 script、visual 写成 {setting,camera,key_shots} 对象。
|
|
必须认出 script 数组并把 visual 对象拍平成一句,而非留空。"""
|
|
raw = json.dumps({
|
|
"script": [
|
|
{"scene_id": 1, "scene_title": "通勤", "voiceover": "选手机看重流畅好看",
|
|
"visual": {"setting": "地铁口", "camera": "竖屏手持跟拍", "key_shots": ["拿出亮银色机身"]}},
|
|
],
|
|
"total_duration": 15,
|
|
}, ensure_ascii=False)
|
|
draft = normalize_draft(raw, aspect_ratio="9:16", total_duration=15)
|
|
seg = draft["segments"][0]
|
|
self.assertEqual(seg["narration"], "选手机看重流畅好看")
|
|
self.assertIn("地铁口", seg["visual"])
|
|
self.assertIn("竖屏手持跟拍", seg["visual"])
|
|
|
|
def test_audio_field_fills_narration(self):
|
|
"""真实回归:shots 用 audio 放口播(GPT 变体),旁白曾因 audio 不在词典而留空。"""
|
|
raw = json.dumps({
|
|
"shots": [{"scene": 1, "visual": "咖啡厅办公把玩机身", "audio": "这款亮银色是我的高光决定"}],
|
|
"total_duration": 15,
|
|
}, ensure_ascii=False)
|
|
draft = normalize_draft(raw, aspect_ratio="9:16", total_duration=15)
|
|
self.assertEqual(draft["segments"][0]["narration"], "这款亮银色是我的高光决定")
|
|
self.assertEqual(draft["segments"][0]["visual"], "咖啡厅办公把玩机身")
|
|
|
|
def test_canonical_flat_schema_still_works(self):
|
|
"""契约内的扁平 schema(segments/narration/visual)不受兼容改动影响。"""
|
|
raw = json.dumps({
|
|
"total_duration": 30,
|
|
"segments": [
|
|
{"role": "钩子", "narration": "口播1", "visual": "画面1"},
|
|
{"role": "CTA", "narration": "口播2", "visual": "画面2"},
|
|
],
|
|
}, ensure_ascii=False)
|
|
draft = normalize_draft(raw, aspect_ratio="9:16", total_duration=30)
|
|
self.assertEqual(len(draft["segments"]), 2)
|
|
self.assertEqual(draft["segments"][0]["narration"], "口播1")
|
|
self.assertEqual(draft["segments"][1]["visual"], "画面2")
|
|
|
|
|
|
from io import BytesIO
|
|
from unittest.mock import patch
|
|
|
|
from django.test import TestCase
|
|
|
|
from apps.accounts.models import Team, User
|
|
from apps.ai.providers.volcano import VolcanoArkProvider
|
|
from apps.ai.services import enqueue_standalone_images
|
|
from apps.assets.models import Asset, AssetFile
|
|
from apps.billing.models import CreditAccount
|
|
from apps.products.models import Product
|
|
|
|
|
|
class StandaloneImageReferenceTests(TestCase):
|
|
"""独立生图(平台套图 / 模特上身图)必须把商品真实主图(+ 模特图)作为参考图走 image_edit,
|
|
而不是纯文生图——回归保护 #图片生成没参考商品主图# 这个 bug。
|
|
(图像默认模型由迁移 seed 的 tokenssr:gpt-image-2 提供,get_image_provider 全程 mock。)"""
|
|
|
|
def setUp(self):
|
|
self.user = User.objects.create_user(username="owner", password="pass")
|
|
self.team = Team.objects.create(name="T", owner=self.user)
|
|
CreditAccount.objects.create(team=self.team, balance="100.0000")
|
|
# 商品 + 主图(带可访问 preview_url)
|
|
self.product = Product.objects.create(team=self.team, created_by=self.user, title="南卡 Lite Pro")
|
|
cover = Asset.objects.create(
|
|
team=self.team, created_by=self.user, name="主图", asset_type=Asset.Type.IMAGE,
|
|
source=Asset.Source.UPLOAD, category=Asset.Category.PRODUCT_IMAGE,
|
|
)
|
|
AssetFile.objects.create(asset=cover, object_key="c.png", bucket="b", content_type="image/png", preview_url="http://x/cover.png", is_primary=True)
|
|
self.product.cover_asset = cover
|
|
self.product.save(update_fields=["cover_asset"])
|
|
|
|
def _patch_provider(self):
|
|
# image_edit 返回值 + 媒体落库链路全部 mock,聚焦验证「传了哪些参考图」
|
|
provider = patch("apps.ai.services.get_image_provider").start()
|
|
prov = provider.return_value
|
|
prov.image_edit.return_value = {"data": [{"url": "http://x/out.png"}]}
|
|
prov.image_generation.return_value = {"data": [{"url": "http://x/out.png"}]}
|
|
prov.extract_first_media_url.return_value = "http://x/out.png"
|
|
media = patch("apps.ai.services.VolcanoArkProvider.media_to_bytes").start()
|
|
media.return_value = (BytesIO(b"img"), "image/png")
|
|
store = patch("apps.ai.services.TosStorage").start()
|
|
stored = store.return_value.upload_fileobj.return_value
|
|
stored.object_key, stored.bucket, stored.content_type, stored.size_bytes = "o.png", "b", "image/png", 3
|
|
self.addCleanup(patch.stopall)
|
|
return prov
|
|
|
|
def test_cover_mode_references_product_main_image(self):
|
|
prov = self._patch_provider()
|
|
enqueue_standalone_images(team=self.team, user=self.user, prompt="平台套图", mode="cover", count=1, product_id=str(self.product.id), ratio="4:5")
|
|
prov.image_edit.assert_called_once()
|
|
self.assertEqual(prov.image_edit.call_args.kwargs["images"], ["http://x/cover.png"])
|
|
prov.image_generation.assert_not_called()
|
|
|
|
def test_model_tryon_combines_product_and_model_images(self):
|
|
prov = self._patch_provider()
|
|
# 模特资产(PERSON)带 preview_url
|
|
model_asset = Asset.objects.create(
|
|
team=self.team, created_by=self.user, name="模特", asset_type=Asset.Type.IMAGE,
|
|
source=Asset.Source.AI_GENERATED, category=Asset.Category.PERSON,
|
|
)
|
|
AssetFile.objects.create(asset=model_asset, object_key="m.png", bucket="b", content_type="image/png", preview_url="http://x/model.png", is_primary=True)
|
|
enqueue_standalone_images(team=self.team, user=self.user, prompt="上身图", mode="model", count=1, product_id=str(self.product.id), model_id=str(model_asset.id), ratio="4:5")
|
|
prov.image_edit.assert_called_once()
|
|
self.assertEqual(prov.image_edit.call_args.kwargs["images"], ["http://x/cover.png", "http://x/model.png"])
|
|
prov.image_generation.assert_not_called()
|
|
|
|
def test_cover_mode_falls_back_to_t2i_without_main_image(self):
|
|
prov = self._patch_provider()
|
|
self.product.cover_asset = None
|
|
self.product.save(update_fields=["cover_asset"])
|
|
enqueue_standalone_images(team=self.team, user=self.user, prompt="平台套图", mode="cover", count=1, product_id=str(self.product.id), ratio="4:5")
|
|
prov.image_generation.assert_called_once()
|
|
prov.image_edit.assert_not_called()
|
|
|
|
|
|
class _FakeStreamResp:
|
|
"""模拟 requests 流式响应:支持 with、raise_for_status、可写 encoding、iter_lines。"""
|
|
status_code = 200
|
|
encoding = None
|
|
|
|
def __init__(self, lines):
|
|
self._lines = lines
|
|
|
|
def __enter__(self):
|
|
return self
|
|
|
|
def __exit__(self, *exc):
|
|
return False
|
|
|
|
def raise_for_status(self):
|
|
return None
|
|
|
|
def iter_lines(self, decode_unicode=True): # noqa: ARG002
|
|
yield from self._lines
|
|
|
|
|
|
class ChatStreamReasoningTests(SimpleTestCase):
|
|
"""推理模型(豆包 seed-pro 等)思考期只发 reasoning_content、不发 content。
|
|
provider 必须把它作为独立 `reasoning` 事件转发——否则脚本 agent 思考期零输出 =
|
|
前端「卡在生成分镜」假死。本测试锁住该转发,防回归。"""
|
|
|
|
def test_reasoning_content_forwarded_as_reasoning_event(self):
|
|
def _chunk(delta):
|
|
return "data: " + json.dumps({"choices": [{"delta": delta}]}, ensure_ascii=False)
|
|
|
|
lines = [
|
|
_chunk({"reasoning_content": "先想想"}),
|
|
_chunk({"reasoning_content": "用户要4镜"}),
|
|
_chunk({"content": "正在生成"}),
|
|
_chunk({"content": "脚本…"}),
|
|
"data: [DONE]",
|
|
]
|
|
prov = VolcanoArkProvider(api_key="k", base_url="http://x")
|
|
with patch("apps.ai.providers.volcano.requests.post", return_value=_FakeStreamResp(lines)):
|
|
events = list(prov.chat_completion_stream(model="m", messages=[{"role": "user", "content": "hi"}]))
|
|
|
|
self.assertEqual([e["type"] for e in events], ["reasoning", "reasoning", "delta", "delta", "done"])
|
|
self.assertEqual([e["text"] for e in events if e["type"] == "reasoning"], ["先想想", "用户要4镜"])
|
|
self.assertEqual("".join(e["text"] for e in events if e["type"] == "delta"), "正在生成脚本…")
|