import json from pathlib import Path from django.conf import settings from django.test import SimpleTestCase from apps.ai.script_agent import normalize_draft class SkillPromptBundlingTests(SimpleTestCase): """守护「skills 必须随后端打进镜像」这条线 —— 历史上 skills 放仓库根、不在 Docker 构建上下文 (`./core/backend`)→ 镜像里没有 → 提示词加载为空 → 提取模型收不到「只输出 JSON」铁律 → 吐散文 → 「提取结果解析失败」。这两条断言锁死:① skills 落在 BASE_DIR 内(随镜像走);② 提示词非空且带铁律。""" def test_skills_dir_bundled_under_base_dir(self): """skills 必须在 BASE_DIR(=core/backend)内,才会被 Dockerfile 的 `COPY . .` 打进镜像。""" for name in ("ecommerce-entity-extract", "ecommerce-video-script"): self.assertTrue( (Path(settings.BASE_DIR) / "skills" / name / "SKILL.md").exists(), f"skills/{name}/SKILL.md 不在 BASE_DIR 内 —— 镜像将丢失该提示词(详见 services._skills_root)", ) def test_entity_extract_prompt_loads_nonempty_with_json_rule(self): """提取系统提示词必须加载到非空内容,且含「只输出 JSON」铁律(空 → 模型不吐 JSON → 解析失败)。""" from apps.ai.services import _load_skill_system_prompt prompt = _load_skill_system_prompt("ecommerce-entity-extract") self.assertGreater(len(prompt), 1000, "提取提示词为空/过短 —— skills 没被正确加载") self.assertIn("JSON", prompt) self.assertIn("entities", prompt) def test_extract_output_contract_is_hardcoded_safety_net(self): """写死的提取输出契约(随代码进镜像)必须存在且含 JSON 铁律 —— 这是 skill 丢失时的兜底, 保证提取永远收到「只输出 JSON」指令(对齐脚本 agent 的 _OUTPUT_PROTOCOL,根治"系统提示词为空"事故)。""" from apps.ai.services import _EXTRACT_OUTPUT_CONTRACT self.assertGreater(len(_EXTRACT_OUTPUT_CONTRACT), 200) for marker in ("entities", "segments", "character", "scene", "JSON"): self.assertIn(marker, _EXTRACT_OUTPUT_CONTRACT) 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 StandaloneCategoryTests(TestCase): """图片趴三类归类(期2):模特上身图→model_tryon / 平台套图→platform_kit / 自由创作→free_create; 生成演员(model 无 product)仍→person(视频角色)。直接跑 worker 函数避开异步 .delay。""" def setUp(self): self.user = User.objects.create_user(username="catowner", password="pass") self.team = Team.objects.create(name="CT", owner=self.user) CreditAccount.objects.create(team=self.team, balance="100.0000") self.product = Product.objects.create(team=self.team, created_by=self.user, title="测试商品") 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): 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 _run(self, mode, *, product_id=None, model_id=None): from apps.ai.services import run_standalone_image_task self._patch_provider() tasks = enqueue_standalone_images( team=self.team, user=self.user, prompt="x", mode=mode, count=1, product_id=product_id, model_id=model_id, model_entity_id="ent-1", ratio="4:5", ) run_standalone_image_task(task_id=str(tasks[0].id)) return Asset.objects.filter(origin_task=tasks[0]).first() def test_model_tryon_category(self): a = self._run("model", product_id=str(self.product.id)) self.assertEqual(a.category, Asset.Category.MODEL_TRYON) self.assertEqual(a.metadata.get("mode"), "model") self.assertTrue(a.metadata.get("batch_id")) # 成组 self.assertEqual(a.metadata.get("model_entity_id"), "ent-1") # 溯源模特库 def test_platform_kit_category(self): a = self._run("cover", product_id=str(self.product.id)) self.assertEqual(a.category, Asset.Category.PLATFORM_KIT) def test_free_create_category(self): a = self._run("image") self.assertEqual(a.category, Asset.Category.FREE_CREATE) def test_generate_actor_stays_person(self): # 生成演员:mode=model 但无 product → 视频角色,仍 person(送审范围内) a = self._run("model") self.assertEqual(a.category, Asset.Category.PERSON) class TriviewAutoEnrollTests(TestCase): """B 路(期3):视频流程新生成的角色立绘 + 三视图成套 → 自动入模特库;幂等不重复建。""" def setUp(self): from apps.projects.models import Project self.user = User.objects.create_user(username="trio", password="p") self.team = Team.objects.create(name="TR", owner=self.user) CreditAccount.objects.create(team=self.team, balance="100.0000") self.product = Product.objects.create(team=self.team, created_by=self.user, title="P") self.project = Project.objects.create(team=self.team, name="项目甲", product=self.product, created_by=self.user) self.portrait = Asset.objects.create( team=self.team, created_by=self.user, name="项目甲-person", asset_type=Asset.Type.IMAGE, source=Asset.Source.AI_GENERATED, category=Asset.Category.PERSON, ) AssetFile.objects.create(asset=self.portrait, object_key="p.png", bucket="b", content_type="image/png", preview_url="http://x/p.png", is_primary=True) def _patch_provider(self): provider = patch("apps.ai.services.get_image_provider").start() prov = provider.return_value prov.image_edit.return_value = {"data": [{"url": "http://x/tri.png"}]} prov.extract_first_media_url.return_value = "http://x/tri.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) def _gen_triview(self): from apps.ai.services import generate_person_triview, run_triview_task task = generate_person_triview(project=self.project, user=self.user, portrait_asset=self.portrait) run_triview_task(task_id=str(task.id)) def test_triview_auto_enrolls_model(self): from apps.assets.models import Model self._patch_provider() self._gen_triview() m = Model.objects.filter(portrait_asset=self.portrait).first() self.assertIsNotNone(m) self.assertIsNotNone(m.triview_asset) # 成套 self.assertTrue(m.metadata.get("auto_enrolled")) self.assertEqual(m.source, Model.Source.AI) # 合规(期3):三视图归 tri_view → 落在送审范围内 self.assertEqual(m.triview_asset.category, Asset.Category.TRI_VIEW) self.assertIn(m.triview_asset.category, Asset.REVIEW_CATEGORIES) def test_auto_enroll_is_idempotent(self): from apps.assets.models import Model self._patch_provider() self._gen_triview() self._gen_triview() # 同一立绘再出一版三视图 self.assertEqual(Model.objects.filter(portrait_asset=self.portrait).count(), 1) 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"), "正在生成脚本…")