feat(core): 基础资产 Agent 化 — 商品三视图(image_edit 参考主图)+ 人物立绘/三视图异步生成 + 演员库/详情

- 商品三视图:有真实商品主图时走 image_edit,以主图为参考锁包装一致(品牌字/配色/外形/Logo)
- 人物:据「某一版立绘」异步生成配套三视图(image_edit,worker 内跑),立绘/三视图各存版本可切换
- 基础资产出图改异步(run_base_asset_task / generate_triview_task),Web 层不被慢出图占住
- 前端:演员库浏览/新增、人物详情(立绘+三视图+版本切换/下载/查看大图)、按 busyKey 的单卡并发 loading(genBusy,替代全局 genBusyKey,各按钮互不阻塞)
- dev:本地连云端 MySQL 复用连接(CONN_MAX_AGE/health check/connect_timeout),仅 development 生效
- 含 projects 测试补充

tsc + py_compile + 26 后端测试通过。

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
zyc
2026-06-17 15:10:22 +08:00
co-authored by Claude Opus 4.8
parent da0c7edfe2
commit 296888f0bf
11 changed files with 361 additions and 59 deletions
+110 -2
View File
@@ -218,11 +218,118 @@ class ProjectApiTests(TestCase):
{"kind": "person", "prompt": "26岁都市女性", "label": "女主"},
format="json",
)
self.assertEqual(response.status_code, 201)
# 异步:秒回 202 + 任务;EAGER 下 worker 任务就地执行,组已落库,label 进 metadata
self.assertEqual(response.status_code, 202)
self.assertIn("task", response.data)
from apps.projects.models import BaseAssetGroup
group = BaseAssetGroup.objects.get(id=response.data["id"])
group = BaseAssetGroup.objects.get(project=project, kind=BaseAssetGroup.Kind.PERSON)
self.assertEqual(group.metadata.get("label"), "女主")
@patch("apps.ai.services._store_generated_media")
@patch("apps.ai.services.get_image_provider")
def test_product_base_asset_uses_cover_image_as_reference(self, get_provider, store_media):
"""商品三视图:商品有主图时走 image_edit 以主图为参考(锁包装一致),而非纯文生图。"""
from apps.assets.models import AssetFile
ModelConfig.objects.create(
provider=self.provider, name="img-model", display_name="Img",
capability=ModelConfig.Capability.IMAGE, endpoint="images/generations", unit_price="1.0000",
)
# 商品主图资产(带可访问 preview_url)
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="k.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"])
out = Asset.objects.create(
team=self.team, created_by=self.user, name="商品三视图",
asset_type=Asset.Type.IMAGE, source=Asset.Source.AI_GENERATED, category=Asset.Category.PRODUCT_IMAGE,
)
store_media.return_value = out
provider = get_provider.return_value
provider.image_edit.return_value = {"data": [{"url": "http://x/tri.png"}]}
provider.extract_first_media_url.return_value = "http://x/tri.png"
project = Project.objects.create(team=self.team, created_by=self.user, product=self.product, name="P")
response = self.client.post(
f"/api/projects/{project.id}/generate-base-asset/",
{"kind": "product", "prompt": "商品三视图"},
format="json",
)
self.assertEqual(response.status_code, 202) # 异步:秒回任务,EAGER 下 worker 就地出图
# 走 image_edit 且把商品主图 URL 作为参考图传入,未走纯文生图
provider.image_edit.assert_called_once()
self.assertEqual(provider.image_edit.call_args.kwargs["images"], ["http://x/cover.png"])
provider.image_generation.assert_not_called()
@patch("apps.ai.services._store_generated_media")
@patch("apps.ai.services.get_image_provider")
def test_product_base_asset_falls_back_without_cover(self, get_provider, store_media):
"""商品无主图时回落纯文生图(image_generation),不报错。"""
ModelConfig.objects.create(
provider=self.provider, name="img-model", display_name="Img",
capability=ModelConfig.Capability.IMAGE, endpoint="images/generations", unit_price="1.0000",
)
out = Asset.objects.create(
team=self.team, created_by=self.user, name="商品三视图",
asset_type=Asset.Type.IMAGE, source=Asset.Source.AI_GENERATED, category=Asset.Category.PRODUCT_IMAGE,
)
store_media.return_value = out
provider = get_provider.return_value
provider.image_generation.return_value = {"data": [{"url": "http://x/tri.png"}]}
provider.extract_first_media_url.return_value = "http://x/tri.png"
project = Project.objects.create(team=self.team, created_by=self.user, product=self.product, name="P")
response = self.client.post(
f"/api/projects/{project.id}/generate-base-asset/",
{"kind": "product", "prompt": "商品三视图"},
format="json",
)
self.assertEqual(response.status_code, 202) # 异步:秒回任务,EAGER 下 worker 就地出图
provider.image_generation.assert_called_once()
provider.image_edit.assert_not_called()
@patch("apps.ai.services._store_generated_media")
@patch("apps.ai.services.get_image_provider")
def test_triview_async_binds_group_to_portrait(self, get_provider, store_media):
"""三视图异步:提交秒回 202;EAGER 下 worker 就地以立绘为参考 image_edit,生成的组 triview_of 绑定该立绘。"""
from apps.assets.models import AssetFile
from apps.projects.models import BaseAssetGroup
ModelConfig.objects.create(
provider=self.provider, name="img-model", display_name="Img",
capability=ModelConfig.Capability.IMAGE, endpoint="images/generations", unit_price="1.0000",
)
portrait = 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=portrait, object_key="p.png", bucket="b", content_type="image/png", preview_url="http://x/portrait.png", is_primary=True)
tri = 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,
)
store_media.return_value = tri
provider = get_provider.return_value
provider.image_edit.return_value = {"data": [{"url": "http://x/tri.png"}]}
provider.extract_first_media_url.return_value = "http://x/tri.png"
project = Project.objects.create(team=self.team, created_by=self.user, product=self.product, name="P")
response = self.client.post(
f"/api/projects/{project.id}/generate-triview/",
{"portrait_asset_id": str(portrait.id)},
format="json",
)
self.assertEqual(response.status_code, 202)
provider.image_edit.assert_called_once()
self.assertEqual(provider.image_edit.call_args.kwargs["images"], ["http://x/portrait.png"])
group = BaseAssetGroup.objects.get(project=project, kind=BaseAssetGroup.Kind.PERSON)
self.assertEqual(group.metadata.get("triview_of"), str(portrait.id))
self.assertEqual(group.adopted_asset_id, tri.id)
def test_extract_cast_and_scenes_parsing_is_robust(self):
"""纯函数:_coerce_tag_entries 去重/容错;无可用模型时 extract 返回空且不抛。"""
from apps.ai.services import _coerce_tag_entries, extract_cast_and_scenes
@@ -418,6 +525,7 @@ class WorkerGateTests(TestCase):
def test_generation_endpoints_blocked_without_worker(self):
cases = [
(f"/api/projects/{self.project.id}/generate-base-asset/", {"kind": "person"}),
(f"/api/projects/{self.project.id}/generate-triview/", {"portrait_asset_id": "x"}),
(f"/api/projects/{self.project.id}/generate-storyboard/", {}),
(f"/api/projects/{self.project.id}/submit-video-segment/", {"video_segment_id": "x"}),
("/api/ai/generate-image/", {"prompt": "测试"}),