fix(ai): 平台套图改走商品主图参考 image_edit · 模特上身图锁商品+模特双图

- 平台套图(cover)以前纯文生图,不参考真实商品主图 → 出图与商品对不上;
  现以商品主图为参考图1走 image_edit,锁包装一致性,有模特则参考图2=模特
- 新增 build_platform_cover_prompt_refs;model_url 解析对 cover 模式同样生效
- 补回归测试:cover 传商品主图、model 传商品主图+模特图、无主图回落文生图

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

core/backend/apps/ai/services.py

@@ -576,6 +576,17 @@ def _find_entity_group(project, kind: str, label: str, group_id: str | None):
     return None
 
 
+def _ratio_to_image_size(ratio: str) -> str:
+    """前端比例(1:1 / 3:4 / 9:16)→ gpt-image 支持的尺寸。竖屏统一 1024x1536,横屏 1536x1024。"""
+    return {
+        "1:1": "1024x1024",
+        "3:4": "1024x1536",
+        "9:16": "1024x1536",
+        "4:3": "1536x1024",
+        "16:9": "1536x1024",
+    }.get((ratio or "").strip(), "1024x1024")
+
+
 def _product_cover_url(product) -> str:
     """商品主图 URL:优先 cover_asset,其次标记为主图的商品图,再次首张商品图。无图返回 ''。"""
     if product is None:
@@ -604,6 +615,50 @@ def build_product_triview_prompt_refs(product, base_prompt: str = "") -> str:
     return " ".join(lines)
 
 
+def build_model_tryon_prompt_refs(product, has_model: bool, base_prompt: str = "") -> str:
+    """模特上身图 image_edit 提示词(refs 版):
+    参考图1=商品真实主图(锁商品外形/品牌/配色),参考图2=选中模特(锁人脸/身形/气质)。
+    生成「该模特自然展示/使用该商品」的电商效果图。"""
+    name = (getattr(product, "title", "") or "商品").strip()
+    lines = [f"参考图1是「{name}」的真实商品。"]
+    if has_model:
+        lines += [
+            "参考图2是出镜模特。请生成参考图2中的这位模特自然地展示/佩戴/使用参考图1中商品的电商效果图。",
+            "模特的五官、发型、肤色、身形与气质必须与参考图2高度一致,不要换人;",
+            "商品的外形、品牌文字、配色、Logo 必须与参考图1高度一致,不要改动或重新设计。",
+        ]
+    else:
+        lines += [
+            "请生成一位真人模特自然地展示/佩戴/使用参考图1中商品的电商效果图。",
+            "商品的外形、品牌文字、配色、Logo 必须与参考图1高度一致,不要改动或重新设计。",
+        ]
+    lines.append("自然光、真实质感、干净背景、电商主图构图,人物与商品比例真实协调。")
+    if base_prompt and base_prompt.strip():
+        lines.append(base_prompt.strip())
+    return " ".join(lines)
+
+
+def build_platform_cover_prompt_refs(product, has_model: bool, base_prompt: str = "") -> str:
+    """平台套图 image_edit 提示词(refs 版):参考图1=商品真实主图(锁外形/品牌/配色/Logo),
+    有模特时参考图2=出镜模特(锁人脸/身形)。生成电商平台主图 / 封面套图,商品须还原真实包装。"""
+    name = (getattr(product, "title", "") or "商品").strip()
+    lines = [f"参考图1是「{name}」的真实商品主图。"]
+    if has_model:
+        lines += [
+            "参考图2是出镜模特。请生成参考图2中的这位模特展示参考图1中商品的电商平台套图(主图 / 封面 / 详情);",
+            "模特的五官、发型、肤色、身形必须与参考图2高度一致,不要换人;",
+        ]
+    else:
+        lines.append("请基于该商品生成电商平台套图(主图 / 封面 / 详情排版),统一视觉风格;")
+    lines.append(
+        "商品的外形、品牌文字、配色、Logo、材质必须与参考图1高度一致,严禁改动或重新设计包装;"
+        "干净背景、电商主图构图、真实质感。"
+    )
+    if base_prompt and base_prompt.strip():
+        lines.append(base_prompt.strip())
+    return " ".join(lines)
+
+
 def generate_base_asset(*, project, user, kind: str, prompt: str, label: str = "", group_id: str | None = None) -> AITask:
     """提交基础资产生成(**异步**):Web 请求只建 RESERVED 任务 + 预留额度(秒级),
     慢出图(文生图 / 商品 image_edit)交给 Celery worker(run_base_asset_task)跑。
@@ -1349,7 +1404,7 @@ def _reap_stale_standalone_image_tasks(*, team) -> None:
             continue
 
 
-def enqueue_standalone_images(*, team, user, prompt: str, mode: str = "image", count: int = 1, product_id: str | None = None, reference_product: bool = False) -> list[AITask]:
+def enqueue_standalone_images(*, team, user, prompt: str, mode: str = "image", count: int = 1, product_id: str | None = None, reference_product: bool = False, model_id: str | None = None, ratio: str | None = None) -> list[AITask]:
     """独立生图(图片创作 / 模特上身图 / 平台套图)改为**异步**:本函数在 Web 请求里只做「建任务 +
     预留额度」这种秒级的活,真正 ~30s 的 ARK 出图交给 Celery worker(generate_standalone_image_task)。
 
@@ -1375,7 +1430,7 @@ def enqueue_standalone_images(*, team, user, prompt: str, mode: str = "image", c
             status=AITask.Status.CREATED,
             model_config=model_config,
             idempotency_key=f"standalone-image:{team.id}:{uuid.uuid4()}",
-            request_payload={"model": model_config.name, "endpoint": model_config.endpoint, "prompt": prompt, "mode": mode, "index": index, "product_id": str(product_id) if product_id else None, "reference_product": bool(reference_product)},
+            request_payload={"model": model_config.name, "endpoint": model_config.endpoint, "prompt": prompt, "mode": mode, "index": index, "product_id": str(product_id) if product_id else None, "reference_product": bool(reference_product), "model_id": str(model_id) if model_id else None, "ratio": str(ratio) if ratio else None},
             estimated_cost=cost,
         )
         # 预留额度若余额不足会抛 ValueError,在同步的 Web 请求里立刻反馈给前端(不会先建半套任务)
@@ -1402,25 +1457,54 @@ def run_standalone_image_task(*, task_id: str) -> None:
     mode = str(payload.get("mode") or "image")
     index = int(payload.get("index") or 0)
     product_id = payload.get("product_id") or None
-    category = _STANDALONE_CATEGORY.get(mode, Asset.Category.UNCATEGORIZED)
+    # 模特上身图(mode=model 且绑了商品)= 该商品的商品图,归到对应商品的 AI 资产,不进人物库;
+    # 「生成演员」同样走 mode=model 但无 product_id,仍归人物库(PERSON)。
+    if mode == "model" and product_id:
+        category = Asset.Category.PRODUCT_IMAGE
+    else:
+        category = _STANDALONE_CATEGORY.get(mode, Asset.Category.UNCATEGORIZED)
     model_config = task.model_config
     provider = get_image_provider(model_config)
     reservation = task.credit_reservation
-    # 商品三视图(商品详情页按 reference_product 提交):有真实商品主图 + 模型支持 image_edit →
-    # 以主图为参考锁包装一致性,否则回落纯文生图(仅凭商品名脑补,不保证还原真实包装)。
-    ref_url = ""
+    # 出图策略(都优先 image_edit 锁真实素材,模型不支持/无素材才回落纯文生图):
+    #  · 模特上身图(mode=model):参考图1=商品真实主图 + 参考图2=选中模特 → 生成「该模特用该商品」效果图;
+    #  · 平台套图(mode=cover):参考图1=商品真实主图(+ 有模特则参考图2=模特)→ 锁包装一致性出套图;
+    #  · 商品三视图(reference_product):参考图1=商品真实主图 → 锁包装一致性;
+    #  · 其余(图片创作):纯文生图。
     product = None
-    if bool(payload.get("reference_product")) and product_id:
+    if product_id:
         from apps.products.models import Product
 
         product = Product.objects.filter(id=product_id).first()
-        if product is not None:
-            ref_url = _product_cover_url(product)
-    use_edit = bool(ref_url) and hasattr(provider, "image_edit")
+    can_edit = hasattr(provider, "image_edit")
+    model_url = ""
+    if payload.get("model_id"):
+        model_asset = Asset.objects.filter(id=payload.get("model_id")).first()
+        if model_asset is not None:
+            model_url = _asset_preview_url(model_asset)
+    product_url = _product_cover_url(product) if product is not None else ""
+
+    edit_images: list[str] = []
+    edit_prompt = ""
+    if mode == "model" and can_edit and product_url:
+        # 模特上身图:商品图必有,模特图可缺(缺则让模型自取真人模特)
+        edit_images = [product_url] + ([model_url] if model_url else [])
+        edit_prompt = build_model_tryon_prompt_refs(product, has_model=bool(model_url), base_prompt=prompt)
+    elif mode == "cover" and can_edit and product_url:
+        # 平台套图:参考图1=商品真实主图(锁包装一致性),有模特则参考图2=模特(锁人脸/身形)
+        edit_images = [product_url] + ([model_url] if model_url else [])
+        edit_prompt = build_platform_cover_prompt_refs(product, has_model=bool(model_url), base_prompt=prompt)
+    elif bool(payload.get("reference_product")) and can_edit and product_url:
+        edit_images = [product_url]
+        edit_prompt = build_product_triview_prompt_refs(product, "")
+    use_edit = bool(edit_images)
     try:
         if use_edit:
-            edit_prompt = build_product_triview_prompt_refs(product, "")
-            response = provider.image_edit(model=model_config.name, prompt=edit_prompt, images=[ref_url], size="1536x1024")
+            if payload.get("reference_product"):
+                size = "1536x1024"  # 三视图固定横向
+            else:
+                size = _ratio_to_image_size(str(payload.get("ratio") or ""))  # 模特图按选中比例
+            response = provider.image_edit(model=model_config.name, prompt=edit_prompt, images=edit_images, size=size)
         else:
             response = provider.image_generation(model=model_config.name, endpoint=model_config.endpoint, prompt=prompt)
         media = provider.extract_first_media_url(response)
@@ -1436,8 +1520,9 @@ def run_standalone_image_task(*, task_id: str) -> None:
             asset_id = uuid.uuid4()
             object_key = f"teams/{team.id}/standalone/{asset_id}{suffix}"
             stored = TosStorage().upload_fileobj(fileobj=fileobj, object_key=object_key, content_type=content_type)
+            asset_label = {"model": "模特上身图", "cover": "平台套图", "image": "图片创作"}.get(mode, mode)
             asset = Asset.objects.create(
-                id=asset_id, team=team, created_by=user, name=f"AI 生成 · {mode} · {index + 1}",
+                id=asset_id, team=team, created_by=user, name=f"AI 生成 · {asset_label} · {index + 1}",
                 asset_type=Asset.Type.IMAGE, source=Asset.Source.AI_GENERATED, category=category, origin_task=task,
                 # 记下生图时选中的商品,商品详情页据此只展示「该商品」的 AI 素材(而非全团队)
                 metadata={"product_id": str(product_id)} if product_id else {},

core/backend/apps/ai/tests.py

@@ -93,3 +93,78 @@ class NormalizeDraftTests(SimpleTestCase):
         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.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()

core/backend/apps/ai/views.py

@@ -31,9 +31,11 @@ class GenerateImageView(APIView):
             count = 1
         product_id = str(request.data.get("product_id") or "").strip() or None
         reference_product = bool(request.data.get("reference_product"))
+        model_id = str(request.data.get("model_id") or "").strip() or None
+        ratio = str(request.data.get("ratio") or "").strip() or None
         team = get_current_team(request.user)
         try:
-            tasks = enqueue_standalone_images(team=team, user=request.user, prompt=prompt, mode=mode, count=count, product_id=product_id, reference_product=reference_product)
+            tasks = enqueue_standalone_images(team=team, user=request.user, prompt=prompt, mode=mode, count=count, product_id=product_id, reference_product=reference_product, model_id=model_id, ratio=ratio)
         except ValueError as exc:  # 无可用模型 / 余额不足等,立即反馈
             return Response({"detail": str(exc)}, status=status.HTTP_400_BAD_REQUEST)
         return Response(

core/backend/apps/ops/tasks.py

@@ -0,0 +1,21 @@
+from django.contrib.auth import get_user_model
+
+from airshelf.celery import app
+from apps.accounts.models import Team
+
+
+@app.task
+def ensure_team_notifications_task(team_id: str, user_id: str | None = None) -> None:
+    """后台生成/补齐团队站内通知。
+
+    原先在 NotificationViewSet.list() 里同步执行:每次刷新都要读计费账本 +
+    对项目/资产/扣费流水逐条 get_or_create,几十次 DB 往返压在请求线程上,拖慢通知接口。
+    现改由 Celery 异步执行,请求只负责读现有通知,生成搬到后台 worker。
+    """
+    from apps.ops.views import ensure_team_notifications  # 延迟导入避免循环依赖
+
+    team = Team.objects.filter(id=team_id).first()
+    if team is None:
+        return
+    user = get_user_model().objects.filter(id=user_id).first() if user_id else None
+    ensure_team_notifications(team, user)

core/backend/apps/ops/tests.py

@@ -0,0 +1,86 @@
+from decimal import Decimal
+
+from django.test import TestCase
+
+from apps.accounts.models import Team, TeamMember, User
+from apps.ai.models import AITask, ModelConfig, ModelProvider
+from apps.assets.models import Asset
+from apps.billing.models import CreditAccount, CreditLedger
+from apps.ops.models import Notification
+from apps.ops.views import ensure_team_notifications
+from apps.products.models import Product
+from apps.projects.models import Project
+
+
+class BillingNotificationTests(TestCase):
+    def setUp(self):
+        self.user = User.objects.create_user(username="owner", password="pass")
+        self.team = Team.objects.create(name="T", owner=self.user)
+        TeamMember.objects.create(team=self.team, user=self.user, role=TeamMember.Role.OWNER)
+        CreditAccount.objects.create(team=self.team, balance="500.0000")
+        self.provider = ModelProvider.objects.create(name="ops-test-prov", display_name="TSR")
+        self.model = ModelConfig.objects.create(
+            provider=self.provider, name="ops-test-image", capability=ModelConfig.Capability.IMAGE
+        )
+        self.product = Product.objects.create(team=self.team, created_by=self.user, title="补水面膜")
+        self.project = Project.objects.create(
+            team=self.team, name="补水面膜·短视频", product=self.product, created_by=self.user
+        )
+
+    def _charge(self, task_type, amount, balance_after):
+        task = AITask.objects.create(
+            team=self.team,
+            project=self.project,
+            task_type=task_type,
+            model_config=self.model,
+            idempotency_key=f"k-{task_type}-{amount}",
+            actual_cost=Decimal(amount),
+        )
+        CreditLedger.objects.create(
+            team=self.team,
+            user=self.user,
+            project=self.project,
+            task=task,
+            ledger_type=CreditLedger.Type.CHARGE,
+            amount=Decimal(amount),
+            balance_after=Decimal(balance_after),
+            reason="AI 任务扣费",
+        )
+        return task
+
+    def test_charge_becomes_billing_notification_with_real_amount(self):
+        self._charge("product_image", "0.32", "499.68")
+        self._charge("video_segment", "18.40", "481.28")
+
+        ensure_team_notifications(self.team, self.user)
+
+        billing = Notification.objects.filter(team=self.team, notification_type=Notification.Type.BILLING)
+        # 每笔扣费一条计费消息,且费用不为空、为真实金额
+        costs = sorted(n.cost_label for n in billing)
+        self.assertIn("¥0.32", costs)
+        self.assertIn("¥18.40", costs)
+        self.assertFalse(billing.filter(cost_label__in=["", "-"]).exists())
+        self.assertTrue(billing.filter(title__contains="图生成").exists())
+        self.assertTrue(billing.filter(title__contains="视频生成").exists())
+
+    def test_project_notification_shows_aggregated_spend(self):
+        self._charge("product_image", "0.32", "499.68")
+        self._charge("video_segment", "18.40", "481.28")
+
+        ensure_team_notifications(self.team, self.user)
+
+        proj_note = Notification.objects.get(
+            team=self.team, notification_type=Notification.Type.TASK, title__contains="补水面膜"
+        )
+        self.assertEqual(proj_note.cost_label, "¥18.72")  # 0.32 + 18.40
+
+    def test_existing_dash_cost_is_backfilled(self):
+        # 早期建的项目通知费用是占位「-」;补上扣费后再跑 ensure 应回填真实金额
+        ensure_team_notifications(self.team, self.user)
+        note = Notification.objects.get(team=self.team, title__contains="补水面膜")
+        self.assertEqual(note.cost_label, "-")
+
+        self._charge("product_image", "0.32", "499.68")
+        ensure_team_notifications(self.team, self.user)
+        note.refresh_from_db()
+        self.assertEqual(note.cost_label, "¥0.32")

core/backend/apps/ops/views.py

@@ -1,4 +1,7 @@
-from django.db.models import Q
+from datetime import timedelta
+
+from django.core.cache import cache
+from django.db.models import Q, Sum
 from django.utils import timezone
 from rest_framework import status
 from rest_framework.decorators import action
@@ -14,7 +17,7 @@ class NotificationPagination(PageNumberPagination):
     max_page_size = 100
 
 from apps.assets.models import Asset
-from apps.billing.models import CreditAccount
+from apps.billing.models import CreditAccount, CreditLedger
 from apps.common.api import TeamScopedViewSetMixin
 from apps.projects.models import Project
 
@@ -32,6 +35,20 @@ def project_stage_label(project):
     }.get(project.current_stage, "Stage 1 · 脚本")
 
 
+# AI 任务类型 → 计费消息里的中文阶段标签(脚本/图片/视频/配音都会扣费)
+TASK_TYPE_LABEL = {
+    "script_generation": "脚本生成",
+    "script_optimization": "脚本优化",
+    "product_image": "商品图生成",
+    "person_image": "模特图生成",
+    "scene_image": "场景图生成",
+    "storyboard": "故事板生成",
+    "video_segment": "视频生成",
+    "voiceover": "配音生成",
+    "export": "成片导出",
+}
+
+
 def project_priority(project):
     if project.status == Project.Status.COMPLETED:
         return Notification.Priority.OK
@@ -57,11 +74,23 @@ def ensure_team_notifications(team, user):
             dedupe_key=dedupe_key,
             defaults=payload,
         )
-        # 已存在的旧通知补齐「处理记录」timeline(早期建的没存),让历史消息也能显示处理记录
+        if created:
+            return
+        # 已存在的旧通知:补齐早期没存的字段(历史消息也能显示),不重建
+        updates = []
+        # 1) 处理记录 timeline
         timeline = (payload.get("metadata") or {}).get("timeline")
-        if not created and timeline and not (obj.metadata or {}).get("timeline"):
+        if timeline and not (obj.metadata or {}).get("timeline"):
             obj.metadata = {**(obj.metadata or {}), "timeline": timeline}
-            obj.save(update_fields=["metadata", "updated_at"])
+            updates.append("metadata")
+        # 2) 费用:早期建的是占位「-」,现在能从计费账本算出真实金额了就回填
+        new_cost = payload.get("cost_label")
+        if new_cost and new_cost != "-" and obj.cost_label in ("", "-"):
+            obj.cost_label = new_cost
+            updates.append("cost_label")
+        if updates:
+            updates.append("updated_at")
+            obj.save(update_fields=updates)
 
     create_once(
         "system:welcome",
@@ -80,6 +109,11 @@ def ensure_team_notifications(team, user):
 
     for project in Project.objects.filter(team=team).select_related("product", "created_by").order_by("-updated_at")[:5]:
         product_title = project.product.title if project.product_id else "未绑定商品"
+        # 项目累计花费 = 该项目所有 AI 扣费流水之和(脚本/图片/视频/导出),无扣费则显示「-」
+        project_spend = project.credit_ledgers.filter(
+            ledger_type=CreditLedger.Type.CHARGE
+        ).aggregate(total=Sum("amount"))["total"]
+        project_cost = f"¥{project_spend:.2f}" if project_spend else "-"
         create_once(
             f"project:{project.id}:status:{project.status}:{project.current_stage}",
             notification_type=Notification.Type.TASK,
@@ -91,7 +125,7 @@ def ensure_team_notifications(team, user):
             project=project,
             stage=project_stage_label(project),
             owner_label=project.created_by.username if project.created_by_id else "成员",
-            cost_label="-",
+            cost_label=project_cost,
             related_url=f"pipeline.html?project_id={project.id}",
             metadata={
                 "status": project.status,
@@ -104,7 +138,10 @@ def ensure_team_notifications(team, user):
             },
         )
 
-    for asset in Asset.objects.filter(team=team).select_related("created_by").order_by("-updated_at")[:3]:
+    for asset in Asset.objects.filter(team=team).select_related("created_by", "origin_task").order_by("-updated_at")[:3]:
+        # AI 生成的资产带 origin_task,可取其实际扣费;手动上传的资产无扣费 → 「-」
+        origin = asset.origin_task
+        asset_cost = f"¥{origin.actual_cost:.2f}" if origin and origin.actual_cost else "-"
         create_once(
             f"asset:{asset.id}:created",
             notification_type=Notification.Type.TASK,
@@ -115,7 +152,7 @@ def ensure_team_notifications(team, user):
             source="资产库",
             stage="资产入库",
             owner_label=asset.created_by.username if asset.created_by_id else "成员",
-            cost_label="-",
+            cost_label=asset_cost,
             related_url="library.html",
             metadata={
                 "asset_id": str(asset.id),
@@ -128,6 +165,53 @@ def ensure_team_notifications(team, user):
             },
         )
 
+    # 计费消息:把真实扣费流水(图片/视频/脚本/配音生成)写成「计费」类通知,带真实金额。
+    # 原先 ensure_team_notifications 完全没读账本,所以计费消息的费用都是空的。
+    # 计费通知覆盖「消息保留期(90 天)」内的全部扣费流水,而不是固定最近 10 条 ——
+    # 否则计费 tab 计数会被卡在 10,与实际扣费条数严重不符。上限 500 仅作暴走兜底。
+    charge_window_start = timezone.now() - timedelta(days=90)
+    recent_charges = (
+        CreditLedger.objects.filter(
+            team=team,
+            ledger_type=CreditLedger.Type.CHARGE,
+            created_at__gte=charge_window_start,
+        )
+        .select_related("task", "project", "user")
+        .order_by("-created_at")[:500]
+    )
+    for ledger in recent_charges:
+        task = ledger.task
+        type_label = TASK_TYPE_LABEL.get(task.task_type, "AI 生成") if task else "AI 生成"
+        project_name = ledger.project.name if ledger.project_id else ""
+        amount_text = f"¥{ledger.amount:.2f}"
+        scope = f" · {project_name}" if project_name else ""
+        create_once(
+            f"billing:charge:{ledger.id}",
+            notification_type=Notification.Type.BILLING,
+            priority=Notification.Priority.INFO,
+            title=f"{type_label}已扣费 {amount_text}",
+            brief=f"{ledger.reason or 'AI 任务扣费'}{scope} · 扣费后余额 ¥{ledger.balance_after:.2f}",
+            body=(
+                f"本次{type_label}消耗 {amount_text},来自真实计费账本(CreditLedger)。"
+                f"扣费后团队余额为 ¥{ledger.balance_after:.2f}。明细可在消费页查看完整账本。"
+            ),
+            source="计费中心",
+            project=ledger.project,
+            stage=type_label,
+            owner_label=ledger.user.username if ledger.user_id else "系统",
+            cost_label=amount_text,
+            related_url="account.html",
+            metadata={
+                "ledger_id": str(ledger.id),
+                "amount": str(ledger.amount),
+                "timeline": [
+                    [_step_time(ledger.created_at), f"{type_label}提交并预留额度"],
+                    [_step_time(ledger.created_at), f"实际扣费 {amount_text}"],
+                    [_step_time(ledger.created_at), f"扣费后余额 ¥{ledger.balance_after:.2f}"],
+                ],
+            },
+        )
+
     account, _ = CreditAccount.objects.get_or_create(team=team)
     if account.balance <= 100:
         create_once(
@@ -174,8 +258,23 @@ class NotificationViewSet(TeamScopedViewSetMixin, ModelViewSet):
             queryset = queryset.filter(is_read=False)
         return queryset
 
+    def _refresh_notifications(self, request):
+        """生成/补齐通知:搬到 Celery 后台执行,不再阻塞请求。
+        - 首次(团队还没有任何通知)同步跑一次,避免初次打开面板是空的;
+        - 之后每团队最多每 60s 触发一次后台刷新,其余请求只读现有数据 → 接口秒回。
+        """
+        team = self.get_team()
+        if not Notification.objects.filter(team=team).exists():
+            ensure_team_notifications(team, request.user)
+            return
+        if cache.add(f"ops:notif-refresh:{team.id}", 1, timeout=60):
+            from .tasks import ensure_team_notifications_task
+
+            user_id = str(request.user.id) if getattr(request.user, "id", None) else None
+            ensure_team_notifications_task.delay(str(team.id), user_id)
+
     def list(self, request, *args, **kwargs):
-        ensure_team_notifications(self.get_team(), request.user)
+        self._refresh_notifications(request)
         response = super().list(request, *args, **kwargs)
         data = response.data
         if isinstance(data, dict):