fix: 自由创作分类入库区分视频与图片展示

core/backend/apps/ai/free_video.py

@@ -695,7 +695,8 @@ def serialize_free_video_task(task: AITask) -> dict:
                 return ""
         return ""
 
-    for asset in task.generated_assets.all():
+    generated_assets = list(task.generated_assets.all())
+    for asset in generated_assets:
         if asset.is_deleted:
             continue
         for f in asset.files.all():
@@ -706,7 +707,7 @@ def serialize_free_video_task(task: AITask) -> dict:
                 video_url = url
             elif not f.is_primary and "image" in (f.content_type or "") and not thumbnail_url:
                 thumbnail_url = url
-    if not video_url:
+    if not video_url and not generated_assets:
         video_url = payload.get("fallback_video_url") or ""
 
     return {

core/backend/apps/ai/tests.py

@@ -178,7 +178,7 @@ from django.test import TestCase
 from rest_framework.test import APIClient
 
 from apps.accounts.models import Team, TeamMember, User
-from apps.ai.models import AITask, ImageConversation, ModelConfig
+from apps.ai.models import AITask, ImageConversation, ModelConfig, ModelProvider
 from apps.ai.providers.volcano import VolcanoArkProvider
 from apps.ai.services import enqueue_standalone_images
 from apps.assets.models import Asset, AssetFile
@@ -186,6 +186,63 @@ from apps.billing.models import CreditAccount
 from apps.products.models import Product
 
 
+class FreeVideoAssetDeletionTests(TestCase):
+    """自由创作视频资产删除后,任务流应降级为无可播成品,不能回退到旧 fallback URL。"""
+
+    def setUp(self):
+        self.user = User.objects.create_user(username="fvdel", password="pass")
+        self.team = Team.objects.create(name="FreeVideoDel", owner=self.user)
+        TeamMember.objects.create(team=self.team, user=self.user, role="owner", status="active")
+        provider = ModelProvider.objects.create(name="fv-provider", display_name="FV Provider")
+        self.model = ModelConfig.objects.create(
+            provider=provider,
+            name="fv-model",
+            display_name="FV Model",
+            capability=ModelConfig.Capability.VIDEO,
+        )
+        self.task = AITask.objects.create(
+            team=self.team,
+            created_by=self.user,
+            task_type=AITask.Type.FREE_VIDEO,
+            status=AITask.Status.SUCCEEDED,
+            model_config=self.model,
+            idempotency_key="free-video-delete-test",
+            request_payload={
+                "prompt": "生成一个广告视频",
+                "fallback_video_url": "http://fallback.example/video.mp4",
+                "fallback_note": "fallback",
+            },
+        )
+        self.asset = Asset.objects.create(
+            team=self.team,
+            created_by=self.user,
+            name="自由创作视频",
+            asset_type=Asset.Type.VIDEO,
+            source=Asset.Source.AI_GENERATED,
+            category=Asset.Category.FREE_CREATE,
+            origin_task=self.task,
+        )
+        AssetFile.objects.create(
+            asset=self.asset,
+            object_key="free-video.mp4",
+            bucket="b",
+            content_type="video/mp4",
+            preview_url="http://tos.example/video.mp4",
+            is_primary=True,
+        )
+
+    def test_deleted_persisted_asset_does_not_fall_back_to_raw_video_url(self):
+        from apps.ai.free_video import serialize_free_video_task
+
+        self.assertEqual(serialize_free_video_task(self.task)["video_url"], "http://tos.example/video.mp4")
+        self.asset.is_deleted = True
+        self.asset.save(update_fields=["is_deleted"])
+        task = AITask.objects.prefetch_related("generated_assets", "generated_assets__files").get(id=self.task.id)
+        data = serialize_free_video_task(task)
+        self.assertEqual(data["video_url"], "")
+        self.assertEqual(data["fallback_note"], "fallback")
+
+
 class StandaloneImageReferenceTests(TestCase):
     """独立生图(平台套图 / 模特上身图)必须把商品真实主图(+ 模特图)作为参考图走 image_edit,
     而不是纯文生图——回归保护 #图片生成没参考商品主图# 这个 bug。

core/backend/apps/assets/tests.py

@@ -220,6 +220,84 @@ class AssetBatchesTests(TestCase):
         self.assertEqual(self.client.get("/api/assets/summary/").json()["tryon"], 2)
 
 
+class FreeCreateSplitTabTests(TestCase):
+    """Free create assets share category=free_create; tabs split by asset_type."""
+
+    def setUp(self):
+        self.user, self.team = _mk_team("ufc", "TeamFC")
+        self.client = APIClient()
+        self.client.force_authenticate(self.user)
+        self.image1 = Asset.objects.create(
+            team=self.team,
+            name="free image 1",
+            asset_type=Asset.Type.IMAGE,
+            source=Asset.Source.AI_GENERATED,
+            category=Asset.Category.FREE_CREATE,
+            metadata={"batch_id": "IMG-B1"},
+        )
+        self.image2 = Asset.objects.create(
+            team=self.team,
+            name="free image 2",
+            asset_type=Asset.Type.IMAGE,
+            source=Asset.Source.AI_GENERATED,
+            category=Asset.Category.FREE_CREATE,
+            metadata={"batch_id": "IMG-B1"},
+        )
+        self.video = Asset.objects.create(
+            team=self.team,
+            name="free video",
+            asset_type=Asset.Type.VIDEO,
+            source=Asset.Source.AI_GENERATED,
+            category=Asset.Category.FREE_CREATE,
+            metadata={"feature": "free_video"},
+        )
+
+    def test_image_creations_batches_exclude_videos(self):
+        data = self.client.get("/api/assets/batches/?tab=image_creations&page_size=10").json()
+        self.assertEqual(data["count"], 1)
+        self.assertEqual(data["results"][0]["count"], 2)
+        self.assertEqual({a["asset_type"] for a in data["results"][0]["items"]}, {"image"})
+
+    def test_video_creations_list_excludes_images(self):
+        data = self.client.get("/api/assets/?tab=video_creations&page_size=10").json()
+        self.assertEqual(data["count"], 1)
+        self.assertEqual(data["results"][0]["id"], str(self.video.id))
+        self.assertEqual(data["results"][0]["asset_type"], "video")
+
+    def test_summary_exposes_split_counts_and_keeps_legacy_creations(self):
+        data = self.client.get("/api/assets/summary/").json()
+        self.assertEqual(data["image_creations"], 1)
+        self.assertEqual(data["video_creations"], 1)
+        self.assertEqual(data["creations"], 2)
+
+    def test_delete_and_restore_video_returns_to_video_creations_tab(self):
+        res = self.client.delete(f"/api/assets/{self.video.id}/")
+        self.assertEqual(res.status_code, 204)
+        self.video.refresh_from_db()
+        self.assertTrue(self.video.is_deleted)
+
+        video_tab = self.client.get("/api/assets/?tab=video_creations&page_size=10").json()
+        self.assertEqual(video_tab["count"], 0)
+        summary = self.client.get("/api/assets/summary/").json()
+        self.assertEqual(summary["video_creations"], 0)
+        self.assertEqual(summary["image_creations"], 1)
+        trash_ids = {a["id"] for a in self.client.get("/api/assets/trash/").json()["results"]}
+        self.assertIn(str(self.video.id), trash_ids)
+
+        restore = self.client.post(f"/api/assets/{self.video.id}/restore/")
+        self.assertEqual(restore.status_code, 200)
+        self.video.refresh_from_db()
+        self.assertFalse(self.video.is_deleted)
+        self.assertEqual(self.video.category, Asset.Category.FREE_CREATE)
+        self.assertEqual(self.video.asset_type, Asset.Type.VIDEO)
+
+        video_tab = self.client.get("/api/assets/?tab=video_creations&page_size=10").json()
+        self.assertEqual(video_tab["count"], 1)
+        self.assertEqual(video_tab["results"][0]["id"], str(self.video.id))
+        image_batches = self.client.get("/api/assets/batches/?tab=image_creations&page_size=10").json()
+        self.assertEqual({a["asset_type"] for a in image_batches["results"][0]["items"]}, {"image"})
+
+
 class FacetsProductsTests(TestCase):
     """facets 返回本 tab 下真实关联的商品(?product= 三路之一:metadata.product_id 等),
     并按 tab 隔离、team 隔离、只回 id/title。"""

core/backend/apps/assets/views.py

@@ -50,7 +50,11 @@ def _tab_q(tab: str) -> Q:
         return Q(category="model_tryon")
     if tab == "kits":  # 平台套图(图片趴)
         return Q(category="platform_kit")
-    if tab == "creations":  # 自由创作(图片趴)
+    if tab == "image_creations":  # 图片自由创作
+        return Q(category="free_create", asset_type="image")
+    if tab == "video_creations":  # 视频自由创作
+        return Q(category="free_create", asset_type="video")
+    if tab == "creations":  # 自由创作(兼容旧入口:图片+视频)
         return Q(category="free_create")
     if tab == "uploads":
         return Q(category="upload")
@@ -64,7 +68,7 @@ def _tab_q(tab: str) -> Q:
 
 
 # 图片趴三类按「生成批次」成组展示:同一次提交的 N 张图共享 metadata.batch_id。
-BATCH_TABS = {"tryon", "kits", "creations"}
+BATCH_TABS = {"tryon", "kits", "creations", "image_creations"}
 
 # 资产名形如「AI 生成 · 模特上身图 · 4」,去掉尾部的「· 序号」得到批次名「AI 生成 · 模特上身图」。
 _BATCH_NAME_RE = re.compile(r"\s*·\s*\d+\s*$")
@@ -180,6 +184,7 @@ class AssetViewSet(TeamScopedViewSetMixin, ModelViewSet):
             for row in rows:
                 seen.add(_batch_key((row["metadata"] or {}).get("batch_id"), row["origin_task_id"], row["id"]))
             out[t] = len(seen)
+        out["video_creations"] = base.filter(_tab_q("video_creations")).count()
         out["others"] = base.filter(_tab_q("others")).count()
         return Response(out)