feat(models): 期1 模特库地基 — Model 顶级实体 + 模特库页 + 审核范围扩展

后端:
- 新增 Model(模特)团队级实体:形象图+三视图+声线(声线尾期),官方模板标签,软删
- Asset.Category 枚举一次加全(model_portrait/tri_view/voice/model_tryon/platform_kit/free_create/storyboard),逐期接线
- 送审范围 person → REVIEW_CATEGORIES{person,tri_view,storyboard}(review.py + adminpanel 队列),图片趴不送审;当前行为不变(forward-compat)
- /api/models/ ModelLibraryViewSet:本团队∪官方模板、官方/我的 tab、真人上传、软删(官方不可删)、跨团队资产引用防越权
- backfill_models 命令把历史成套模特(kind=model + 项目流命名配对)归集成 Model,dry-run 默认

前端:
- 左菜单新增「模特库」顶级入口(person 图标)+ 路由 /models
- 模特库页:全部/官方模板/我的模特 tab + 真人上传 + 模特卡(形象图+三视图缩略+官方标签),仅用 design token

测试:assets 10/10 绿;tsc+build 全绿;无头走查 0 console error(全部5/官方3/我的2,官方标签达标)
基线既有 7 失败(ai/projects provider mock 漂移)零新增

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
seaislee1209
2026-06-20 22:53:46 +08:00
co-authored by Claude Opus 4.8
parent 1c9143ce61
commit 44e90233ff
23 changed files with 1003 additions and 12 deletions
+4 -4
View File
@@ -289,9 +289,9 @@ _REVIEW_STATUSES = {"", "processing", "active", "failed"}
@api_view(["GET"])
@permission_classes([IsPlatformAdmin])
def admin_asset_reviews(request):
"""跨团队真人(person)资产审核队列。?review_status=none|processing|active|failed 过滤(none=未送审)。"""
"""跨团队送审资产审核队列(角色定妆照/三视图/分镜图)。?review_status=none|processing|active|failed 过滤(none=未送审)。"""
qs = (
Asset.objects.filter(category=Asset.Category.PERSON, is_deleted=False)
Asset.objects.filter(category__in=Asset.REVIEW_CATEGORIES, is_deleted=False)
.select_related("team")
.prefetch_related("files")
.order_by("-created_at")
@@ -314,7 +314,7 @@ def admin_asset_reviews(request):
def admin_asset_reviews_submit(request):
"""批量送审(也用于失败重试):对给定 person 资产逐个 submit_asset_for_review。"""
ids = request.data.get("asset_ids") or []
assets = list(Asset.objects.filter(id__in=ids, category=Asset.Category.PERSON, is_deleted=False))
assets = list(Asset.objects.filter(id__in=ids, category__in=Asset.REVIEW_CATEGORIES, is_deleted=False))
for asset in assets:
submit_asset_for_review(asset)
statuses = {str(a.id): a.review_status for a in Asset.objects.filter(id__in=ids)}
@@ -333,7 +333,7 @@ def admin_asset_reviews_submit(request):
def admin_asset_reviews_poll(request):
"""轮询审核中(processing)资产的最新状态。给 asset_ids 则只轮询这些,否则轮询全平台 processing。"""
ids = request.data.get("asset_ids")
qs = Asset.objects.filter(category=Asset.Category.PERSON, review_status="processing", is_deleted=False)
qs = Asset.objects.filter(category__in=Asset.REVIEW_CATEGORIES, review_status="processing", is_deleted=False)
if ids:
qs = qs.filter(id__in=ids)
statuses = {}
@@ -0,0 +1,145 @@
"""把历史「成套模特(形象图 + 三视图)」归集成 Model 实体,顺带报告/清理孤儿错图。
两条历史来源都收编:
1) 模特库生成(model_library.generate_model):metadata.kind="model" 的 PERSON 资产,
成对的 view=frontal / view=three_view 共享同一 brief、同团队 → 按 brief 配对。
2) 项目立绘流(run_triview_task):三视图资产带 metadata.triview_of=<立绘 asset id> → 立绘=形象图、它=三视图。
幂等(局部):已被某个 Model 引用过的 portrait/triview 资产会跳过,同一张图不会被重复归组。
注意这是 **一次性数据工具**:同名前缀若有多余的重生版本(例如同项目多次出图),再次运行会把剩余版本另配成新组。
所以跑一次即可,别反复跑。默认 **dry-run** 只打印计划;--apply 才真正写库。清理孤儿用 --soft-delete-orphans(只置 is_deleted,可逆)。
"""
from collections import defaultdict
from django.core.management.base import BaseCommand
from django.db import transaction
from apps.assets.models import Asset, Model
class Command(BaseCommand):
help = "把历史成套模特(形象图+三视图)归集成 Model 实体;报告并可软删孤儿错图。"
def add_arguments(self, parser):
parser.add_argument("--apply", action="store_true", help="真正写库(默认 dry-run 只打印)")
parser.add_argument(
"--soft-delete-orphans",
action="store_true",
help="把无法配对、且无文件的 model 类孤儿资产置 is_deleted=True(可逆)",
)
def handle(self, *args, **opts):
apply = opts["apply"]
soft_del = opts["soft_delete_orphans"]
self.stdout.write(self.style.WARNING("=== backfill_models %s ===" % ("APPLY" if apply else "DRY-RUN")))
# 已被 Model 引用的资产(幂等跳过)
used = set()
for pid, tid in Model.objects.values_list("portrait_asset_id", "triview_asset_id"):
if pid:
used.add(str(pid))
if tid:
used.add(str(tid))
plans: list[dict] = [] # {name, portrait, triview, source}
# —— 来源 2:项目立绘流(triview_of) ——
triviews = Asset.objects.filter(metadata__triview_of__isnull=False, is_deleted=False)
for tv in triviews:
portrait_id = str((tv.metadata or {}).get("triview_of") or "")
if not portrait_id or str(tv.id) in used or portrait_id in used:
continue
portrait = Asset.objects.filter(id=portrait_id, is_deleted=False).first()
if portrait is None:
continue
plans.append(
{"name": (portrait.name or "模特").split("·")[0][:255], "portrait": portrait, "triview": tv, "source": "project"}
)
used.update({str(tv.id), portrait_id})
# —— 来源 1:模特库生成(kind=model,按 team+brief 配对 frontal/three_view) ——
groups: dict[tuple, dict[str, list[Asset]]] = defaultdict(lambda: {"frontal": [], "three_view": []})
models_qs = Asset.objects.filter(metadata__kind="model", is_deleted=False).order_by("created_at")
for a in models_qs:
if str(a.id) in used:
continue
view = (a.metadata or {}).get("view")
if view in ("frontal", "three_view"):
groups[(a.team_id, (a.metadata or {}).get("brief", ""))][view].append(a)
for (team_id, brief), bucket in groups.items():
frontals, tvs = bucket["frontal"], bucket["three_view"]
for portrait, triview in zip(frontals, tvs):
if str(portrait.id) in used or str(triview.id) in used:
continue
plans.append(
{"name": (portrait.name or brief or "模特").split("·")[0][:255], "portrait": portrait, "triview": triview, "source": "model_lib"}
)
used.update({str(portrait.id), str(triview.id)})
# —— 来源 3:项目立绘流按命名后缀配对(立绘 "X-person" ↔ 三视图 "X-三视图",同前缀同团队) ——
# 历史项目流的 person 资产没写 triview_of,但命名成对;按名前缀收编成套的。
PORT_SUF = ("-person", "-立绘")
TRI_SUF = ("-三视图", "·三视图")
def _strip(name: str):
for s in TRI_SUF:
if name.endswith(s):
return name[: -len(s)], "tri"
for s in PORT_SUF:
if name.endswith(s):
return name[: -len(s)], "port"
return None, None
named: dict[tuple, dict[str, Asset]] = defaultdict(dict)
for a in Asset.objects.filter(category="person", is_deleted=False).order_by("created_at"):
if str(a.id) in used:
continue
prefix, role = _strip(a.name or "")
if role is None:
continue
slot = named[(a.team_id, prefix)]
slot.setdefault(role, a) # 同前缀多版本取最早一张,避免错配
for (team_id, prefix), slot in named.items():
portrait, triview = slot.get("port"), slot.get("tri")
if not portrait or not triview or str(portrait.id) in used or str(triview.id) in used:
continue
plans.append({"name": prefix.strip(" ·-")[:255] or "模特", "portrait": portrait, "triview": triview, "source": "named"})
used.update({str(portrait.id), str(triview.id)})
# —— 孤儿:model 类资产没进任何配对(疑似早期错图 / 半成品) ——
orphans = [
a for a in models_qs
if str(a.id) not in used and (a.metadata or {}).get("view") in ("frontal", "three_view")
]
no_file_orphans = [a for a in orphans if not a.files.exists()]
# —— 打印计划 ——
self.stdout.write(f" 待建 Model:{len(plans)}")
for p in plans:
self.stdout.write(f" [{p['source']}] {p['name']} portrait={p['portrait'].id} triview={p['triview'].id}")
self.stdout.write(f" 孤儿(model 类未配对):{len(orphans)};其中无文件:{len(no_file_orphans)}")
if not apply:
self.stdout.write(self.style.WARNING("DRY-RUN 结束,未写库。加 --apply 执行。"))
return
created = 0
with transaction.atomic():
for p in plans:
Model.objects.create(
team_id=p["portrait"].team_id,
created_by_id=p["portrait"].created_by_id,
name=p["name"],
source=Model.Source.AI,
portrait_asset=p["portrait"],
triview_asset=p["triview"],
metadata={"backfill": p["source"]},
)
created += 1
soft_deleted = 0
if soft_del:
for a in no_file_orphans:
a.is_deleted = True
a.save(update_fields=["is_deleted", "updated_at"])
soft_deleted += 1
self.stdout.write(self.style.SUCCESS(f"完成:建 {created} 个 Model;软删孤儿 {soft_deleted if soft_del else 0} 张。"))
@@ -0,0 +1,46 @@
# Generated by Django 5.1.15 on 2026-06-20 14:32
import django.db.models.deletion
import uuid
from django.conf import settings
from django.db import migrations, models
class Migration(migrations.Migration):
dependencies = [
('accounts', '0005_invitation_kind_alter_invitation_team'),
('assets', '0005_asset_assets_asse_team_id_4b1b57_idx_and_more'),
migrations.swappable_dependency(settings.AUTH_USER_MODEL),
]
operations = [
migrations.AlterField(
model_name='asset',
name='category',
field=models.CharField(choices=[('person', 'Person'), ('scene', 'Scene'), ('product_image', 'Product Image'), ('video_clip', 'Video Clip'), ('final_video', 'Final Video'), ('upload', 'Upload'), ('uncategorized', 'Uncategorized'), ('model_portrait', 'Model Portrait'), ('tri_view', 'Tri View'), ('voice', 'Voice'), ('model_tryon', 'Model Try-on'), ('platform_kit', 'Platform Kit'), ('free_create', 'Free Create'), ('storyboard', 'Storyboard')], default='uncategorized', max_length=32),
),
migrations.CreateModel(
name='Model',
fields=[
('id', models.UUIDField(default=uuid.uuid4, editable=False, primary_key=True, serialize=False)),
('created_at', models.DateTimeField(auto_now_add=True)),
('updated_at', models.DateTimeField(auto_now=True)),
('name', models.CharField(max_length=255)),
('is_official', models.BooleanField(default=False)),
('source', models.CharField(choices=[('ai', 'AI Generated'), ('upload', 'Upload')], default='ai', max_length=16)),
('description', models.TextField(blank=True)),
('metadata', models.JSONField(blank=True, default=dict)),
('is_deleted', models.BooleanField(default=False)),
('created_by', models.ForeignKey(blank=True, null=True, on_delete=django.db.models.deletion.SET_NULL, related_name='created_%(class)s_set', to=settings.AUTH_USER_MODEL)),
('portrait_asset', models.ForeignKey(blank=True, null=True, on_delete=django.db.models.deletion.SET_NULL, related_name='model_as_portrait', to='assets.asset')),
('team', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, related_name='%(class)s_set', to='accounts.team')),
('triview_asset', models.ForeignKey(blank=True, null=True, on_delete=django.db.models.deletion.SET_NULL, related_name='model_as_triview', to='assets.asset')),
('voice_asset', models.ForeignKey(blank=True, null=True, on_delete=django.db.models.deletion.SET_NULL, related_name='model_as_voice', to='assets.asset')),
],
options={
'ordering': ['-created_at'],
'indexes': [models.Index(fields=['team', 'is_official', '-created_at'], name='assets_mode_team_id_0caf77_idx')],
},
),
]
+9
View File
@@ -0,0 +1,9 @@
"""模特库路由(顶级实体,挂在 /api/models/)。与 AssetViewSet(/api/assets/)分开,避免 "" 前缀 pk 冲突。"""
from rest_framework.routers import DefaultRouter
from .views import ModelLibraryViewSet
router = DefaultRouter()
router.register("", ModelLibraryViewSet, basename="model")
urlpatterns = router.urls
+51 -1
View File
@@ -18,13 +18,24 @@ class Asset(TeamOwnedModel):
SYSTEM = "system", "System"
class Category(models.TextChoices):
PERSON = "person", "Person"
PERSON = "person", "Person" # 视频趴·角色定妆照(送审)
SCENE = "scene", "Scene"
PRODUCT_IMAGE = "product_image", "Product Image"
VIDEO_CLIP = "video_clip", "Video Clip"
FINAL_VIDEO = "final_video", "Final Video"
UPLOAD = "upload", "Upload"
UNCATEGORIZED = "uncategorized", "Uncategorized"
# ↓ 模特库 + 资产模型重构(2026-06-20)新增,逐期接线。
MODEL_PORTRAIT = "model_portrait", "Model Portrait" # 模特形象图(图片域·不送审)
TRI_VIEW = "tri_view", "Tri View" # 三视图(送审)
VOICE = "voice", "Voice" # 模特声线(占位·尾期)
MODEL_TRYON = "model_tryon", "Model Try-on" # 模特上身图(图片趴·不送审)
PLATFORM_KIT = "platform_kit", "Platform Kit" # 平台套图(图片趴)
FREE_CREATE = "free_create", "Free Create" # 自由创作(图片趴)
STORYBOARD = "storyboard", "Storyboard" # 分镜图(视频趴·送审)
# 送审范围(火山人像审核):视频趴含人脸 = 角色定妆照 / 三视图 / 分镜图。图片趴(上身图/套图/创作)不送审。
REVIEW_CATEGORIES = ("person", "tri_view", "storyboard")
name = models.CharField(max_length=255)
asset_type = models.CharField(max_length=24, choices=Type.choices)
@@ -56,6 +67,45 @@ class Asset(TeamOwnedModel):
return self.name
class Model(TeamOwnedModel):
"""模特(顶级实体 = 模特库)· 团队级、可复用。
一个完整模特 = 形象图 + 三视图 + 声线(声线尾期做)。图片线(上身图)+ 视频线(角色)都能引用它。
来源:官方预制(is_official=True,打「官方模板」标签)/ 商家自建(AI 生成或真人上传)。
portrait/triview/voice 指向具体 Asset(复用资产表,不重复存文件)。
"""
class Source(models.TextChoices):
AI = "ai", "AI Generated"
UPLOAD = "upload", "Upload"
name = models.CharField(max_length=255)
is_official = models.BooleanField(default=False) # 官方模板标签
source = models.CharField(max_length=16, choices=Source.choices, default=Source.AI)
portrait_asset = models.ForeignKey(
Asset, on_delete=models.SET_NULL, null=True, blank=True, related_name="model_as_portrait"
)
triview_asset = models.ForeignKey(
Asset, on_delete=models.SET_NULL, null=True, blank=True, related_name="model_as_triview"
)
voice_asset = models.ForeignKey( # 声线,尾期接线,先留空
Asset, on_delete=models.SET_NULL, null=True, blank=True, related_name="model_as_voice"
)
description = models.TextField(blank=True)
metadata = models.JSONField(default=dict, blank=True)
is_deleted = models.BooleanField(default=False)
class Meta:
indexes = [
# 模特库默认按团队 + 新→旧;官方模板优先靠前由序列化/查询控制
models.Index(fields=["team", "is_official", "-created_at"]),
]
ordering = ["-created_at"]
def __str__(self) -> str:
return self.name
class AssetReviewGroup(TimeStampedModel):
"""一团队一火山人像素材组(真人资产审核统一上传到这里,单组可放 500 万)。"""
+2 -2
View File
@@ -61,7 +61,7 @@ def get_or_create_team_group(team) -> AssetReviewGroup:
def submit_asset_for_review(asset: Asset) -> None:
"""真人资产静默送审:建组(若无)→ 传素材 → 标 processing。出错只记日志,不抛。"""
if not assets_client.is_enabled() or asset.category != Asset.Category.PERSON:
if not assets_client.is_enabled() or asset.category not in Asset.REVIEW_CATEGORIES:
return
url = _asset_url(asset)
if not url:
@@ -112,7 +112,7 @@ def poll_team_reviews(team) -> dict:
"""轮询该团队所有「审核中」真人资产,更新状态。返回 {asset_id: status}。"""
out: dict[str, str] = {}
pending = Asset.objects.filter(
team=team, category=Asset.Category.PERSON, review_status="processing", is_deleted=False
team=team, category__in=Asset.REVIEW_CATEGORIES, review_status="processing", is_deleted=False
)
for asset in pending:
out[str(asset.id)] = poll_asset_review(asset)
+69 -1
View File
@@ -1,7 +1,7 @@
from django.conf import settings
from rest_framework import serializers
from .models import Asset, AssetFile
from .models import Asset, AssetFile, Model
from .storage import TosStorage
@@ -105,3 +105,71 @@ class AssetUploadSerializer(serializers.Serializer):
category = serializers.ChoiceField(choices=Asset.Category.choices, default=Asset.Category.UPLOAD)
description = serializers.CharField(required=False, allow_blank=True)
def _asset_preview(asset) -> str:
"""取资产主文件的公读直链(模特卡缩略图用)。无文件返回 """""
if asset is None:
return ""
f = asset.files.filter(is_primary=True).first() or asset.files.first()
if f is None:
return ""
if f.preview_url:
return f.preview_url
if not f.object_key or not settings.TOS.get("endpoint"):
return ""
try:
return _tos().public_url(object_key=f.object_key, bucket=f.bucket or None)
except Exception:
return ""
class ModelLibrarySerializer(serializers.ModelSerializer):
"""模特库卡片:实体字段 + 形象图/三视图缩略图直链 + 是否已挂声线。"""
portrait = serializers.SerializerMethodField()
triview = serializers.SerializerMethodField()
has_voice = serializers.SerializerMethodField()
class Meta:
model = Model
fields = [
"id",
"name",
"is_official",
"source",
"description",
"metadata",
"portrait_asset",
"triview_asset",
"voice_asset",
"portrait",
"triview",
"has_voice",
"created_at",
"updated_at",
]
read_only_fields = ["id", "is_official", "source", "created_at", "updated_at"]
def get_portrait(self, obj):
return _asset_preview(obj.portrait_asset)
def get_triview(self, obj):
return _asset_preview(obj.triview_asset)
def get_has_voice(self, obj):
return bool(obj.voice_asset_id)
def validate(self, attrs):
"""防越权:引用的形象图/三视图/声线资产必须属于当前团队。"""
from apps.common.api import get_current_team
request = self.context.get("request")
if request is None:
return attrs
team = get_current_team(request.user)
for field in ("portrait_asset", "triview_asset", "voice_asset"):
asset = attrs.get(field)
if asset is not None and asset.team_id != team.id:
raise serializers.ValidationError({field: "资产不属于当前团队"})
return attrs
+110
View File
@@ -0,0 +1,110 @@
from io import BytesIO
from types import SimpleNamespace
from unittest.mock import patch
from django.test import TestCase
from rest_framework.test import APIClient
from apps.accounts.models import Team, TeamMember, User
from .models import Asset, Model
def _mk_team(username, team_name):
user = User.objects.create_user(username=username, password="x")
team = Team.objects.create(name=team_name, owner=user)
TeamMember.objects.create(team=team, user=user, role=TeamMember.Role.OWNER)
return user, team
class ModelLibraryApiTests(TestCase):
"""模特库:本团队模特 ∪ 官方模板;官方跨团队可见;软删;官方不可删;团队隔离。"""
def setUp(self):
self.user, self.teamA = _mk_team("ua", "TeamA")
self.other, self.teamB = _mk_team("ub", "TeamB")
self.client = APIClient()
self.client.force_authenticate(self.user)
self.mine = Model.objects.create(team=self.teamA, name="我的模特")
self.official = Model.objects.create(team=self.teamB, name="官方小美", is_official=True)
self.bs_other = Model.objects.create(team=self.teamB, name="别队私有") # B 的私有,A 看不到
def test_list_returns_mine_plus_official(self):
ids = {m["id"] for m in self.client.get("/api/models/").json()["results"]}
self.assertIn(str(self.mine.id), ids)
self.assertIn(str(self.official.id), ids) # 官方跨团队可见
self.assertNotIn(str(self.bs_other.id), ids) # 别队私有不可见
def test_tab_official_and_mine(self):
off = {m["id"] for m in self.client.get("/api/models/?tab=official").json()["results"]}
self.assertEqual(off, {str(self.official.id)})
mine = {m["id"] for m in self.client.get("/api/models/?tab=mine").json()["results"]}
self.assertEqual(mine, {str(self.mine.id)})
def test_official_flag_in_payload(self):
row = next(m for m in self.client.get("/api/models/").json()["results"] if m["id"] == str(self.official.id))
self.assertTrue(row["is_official"])
def test_delete_is_soft(self):
res = self.client.delete(f"/api/models/{self.mine.id}/")
self.assertEqual(res.status_code, 204)
self.mine.refresh_from_db()
self.assertTrue(self.mine.is_deleted)
ids = {m["id"] for m in self.client.get("/api/models/").json()["results"]}
self.assertNotIn(str(self.mine.id), ids)
def test_official_cannot_be_deleted(self):
res = self.client.delete(f"/api/models/{self.official.id}/")
self.assertEqual(res.status_code, 403)
self.official.refresh_from_db()
self.assertFalse(self.official.is_deleted)
def test_cannot_delete_other_team_model(self):
res = self.client.delete(f"/api/models/{self.bs_other.id}/")
self.assertIn(res.status_code, (403, 404)) # 别队私有:不在可见 qs → 404,或命中守卫 → 403
self.bs_other.refresh_from_db()
self.assertFalse(self.bs_other.is_deleted)
def test_create_rejects_cross_team_asset(self):
b_asset = Asset.objects.create(team=self.teamB, name="别队图", asset_type="image", source="upload")
res = self.client.post("/api/models/", {"name": "盗用", "portrait_asset": str(b_asset.id)}, format="json")
self.assertEqual(res.status_code, 400)
def test_upload_real_person_creates_asset_and_model(self):
stored = SimpleNamespace(object_key="teams/x/models/a.png", bucket="b", content_type="image/png", size_bytes=10)
with patch("apps.assets.views.TosStorage") as Tos:
Tos.return_value.upload_fileobj.return_value = stored
f = BytesIO(b"img")
f.name = "lady.png"
res = self.client.post("/api/models/upload/", {"name": "上传妹", "file": f}, format="multipart")
self.assertEqual(res.status_code, 201)
body = res.json()
self.assertEqual(body["source"], "upload")
model = Model.objects.get(id=body["id"])
self.assertIsNotNone(model.portrait_asset)
self.assertEqual(model.portrait_asset.category, Asset.Category.MODEL_PORTRAIT)
class ReviewScopeTests(TestCase):
"""送审范围 = 角色定妆照(person)/ 三视图(tri_view)/ 分镜图(storyboard);图片趴不送审。"""
def test_review_categories_constant(self):
self.assertEqual(Asset.REVIEW_CATEGORIES, ("person", "tri_view", "storyboard"))
def test_poll_team_reviews_only_scans_review_categories(self):
from apps.assets import review
_, team = _mk_team("uc", "TeamC")
keep = {}
for cat in ("person", "tri_view", "storyboard"):
a = Asset.objects.create(team=team, name=cat, asset_type="image", source="ai_generated",
category=cat, review_status="processing")
keep[cat] = str(a.id)
# 图片趴 / 场景:不应进送审队列
for cat in ("model_tryon", "model_portrait", "scene"):
Asset.objects.create(team=team, name=cat, asset_type="image", source="ai_generated",
category=cat, review_status="processing")
with patch("apps.assets.review.assets_client.is_enabled", return_value=False):
out = review.poll_team_reviews(team)
self.assertEqual(set(out.keys()), set(keep.values()))
+91 -3
View File
@@ -7,16 +7,17 @@ from django.db.models import Q
from django.http import StreamingHttpResponse
from rest_framework import status
from rest_framework.decorators import action
from rest_framework.exceptions import PermissionDenied, ValidationError
from rest_framework.pagination import PageNumberPagination
from rest_framework.parsers import FormParser, MultiPartParser
from rest_framework.parsers import FormParser, JSONParser, MultiPartParser
from rest_framework.response import Response
from rest_framework.views import APIView
from rest_framework.viewsets import ModelViewSet
from apps.common.api import TeamScopedViewSetMixin, get_current_team
from .models import Asset, AssetFile
from .serializers import AssetSerializer, AssetUploadSerializer
from .models import Asset, AssetFile, Model
from .serializers import AssetSerializer, AssetUploadSerializer, ModelLibrarySerializer
from .storage import TosStorage
@@ -141,6 +142,93 @@ class AssetViewSet(TeamScopedViewSetMixin, ModelViewSet):
return response
class ModelLibraryViewSet(ModelViewSet):
"""模特库(顶级实体)· 团队级 + 官方模板跨团队可见。
list 返回「本团队模特 官方模板」;?tab=official 只看官方、?tab=mine 只看自建。
真人上传走 upload action(传图 → 建 model_portrait 资产 + Model 实体)。删除 = 软删(官方不可删)。"""
serializer_class = ModelLibrarySerializer
pagination_class = AssetPagination
parser_classes = [JSONParser, MultiPartParser, FormParser]
search_fields = ["name", "description"]
def get_team(self):
return get_current_team(self.request.user)
def get_queryset(self):
team = self.get_team()
qs = (
Model.objects.filter(Q(team=team) | Q(is_official=True), is_deleted=False)
.prefetch_related("portrait_asset__files", "triview_asset__files")
)
tab = self.request.query_params.get("tab")
if tab == "official":
qs = qs.filter(is_official=True)
elif tab == "mine":
qs = qs.filter(team=team, is_official=False)
if self.request.query_params.get("q"):
q = self.request.query_params["q"]
qs = qs.filter(Q(name__icontains=q) | Q(description__icontains=q))
# 官方模板靠前,再按新→旧
return qs.order_by("-is_official", "-created_at")
def perform_create(self, serializer):
serializer.save(team=self.get_team(), created_by=self.request.user)
def perform_destroy(self, instance):
if instance.is_official:
raise PermissionDenied("官方模特不可删除")
if instance.team_id != self.get_team().id:
raise PermissionDenied("无权删除其他团队的模特")
instance.is_deleted = True
instance.save(update_fields=["is_deleted", "updated_at"])
@action(detail=False, methods=["post"], url_path="upload", parser_classes=[MultiPartParser, FormParser])
def upload(self, request):
"""真人上传:一张人像图 → 建 model_portrait 资产 + Model(source=upload)。三视图/声线后续补。"""
upload = request.FILES.get("file")
if upload is None:
raise ValidationError({"file": "请上传一张模特形象图"})
team = self.get_team()
name = (request.data.get("name") or Path(upload.name).stem or "模特")[:255]
asset_id = uuid.uuid4()
suffix = Path(upload.name).suffix.lower() or ".png"
object_key = f"teams/{team.id}/models/{asset_id}{suffix}"
with transaction.atomic():
stored = TosStorage().upload_fileobj(
fileobj=upload.file,
object_key=object_key,
content_type=upload.content_type or "image/png",
)
portrait = Asset.objects.create(
id=asset_id,
team=team,
created_by=request.user,
name=f"{name}·形象图",
asset_type=Asset.Type.IMAGE,
source=Asset.Source.UPLOAD,
category=Asset.Category.MODEL_PORTRAIT,
metadata={"kind": "model", "view": "frontal", "source": "upload"},
)
AssetFile.objects.create(
asset=portrait,
object_key=stored.object_key,
bucket=stored.bucket,
content_type=stored.content_type,
size_bytes=stored.size_bytes,
is_primary=True,
)
model = Model.objects.create(
team=team,
created_by=request.user,
name=name,
source=Model.Source.UPLOAD,
portrait_asset=portrait,
metadata={"source": "upload"},
)
return Response(ModelLibrarySerializer(model).data, status=status.HTTP_201_CREATED)
class AssetUploadView(APIView):
parser_classes = [MultiPartParser, FormParser]