feat(billing): 计费商业化重构(积分制)+ 团队差异化调价

统一计价引擎 apps/billing/pricing.py:1积分=¥0.1,平台成本(¥)与用户价(积分)双记账,
视频按火山真实 usage.total_tokens 结算(true-up,终结¥1/段倒贴)+ 首发×1.5毛利系数。
Team.price_multiplier 差异化调价(jimeng同款):挂牌价×系数两步HALF_UP取整,视频类
按下单时的价格/汇率快照结算(中途改配不影响在途任务)。

BillingConfig 单例(汇率/毛利系数/预留buffer,admin可调即刻生效)。存量数据 ×10 rescale
迁移(RunPython+atomic,MySQL 迁移不可中断重跑)。开户赠送归零(DEFAULT_TRIAL_CREDITS=0,
商业决策)。audit_billing 加 I9(卖亏审计)。271 条测试 + tsc/build 全绿。

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>

core/backend/apps/ai/services.py

@@ -120,8 +120,8 @@ def get_video_provider(model_config: ModelConfig):
     return build_provider(model_config)
 
 
-def estimate_cost(model_config: ModelConfig) -> Decimal:
-    return model_config.unit_price if model_config.unit_price > 0 else Decimal("1.0000")
+# estimate_cost() 已退役:全平台定价统一走 apps/billing/pricing.py 计价引擎(积分制)。
+# flat 类型默认价由 create_ai_task 内 quote_flat 提供;视频/配音各入口自带 quote。
 
 
 def parse_segment_fields(block: str) -> tuple[str, str]:
@@ -629,8 +629,28 @@ def split_script_into_segments(content: str, count: int = 4) -> list[str]:
 
 
 @transaction.atomic
-def create_ai_task(*, project, user, task_type: str, model_config: ModelConfig, request_payload: dict) -> AITask:
-    cost = estimate_cost(model_config)
+def create_ai_task(
+    *,
+    project,
+    user,
+    task_type: str,
+    model_config: ModelConfig,
+    request_payload: dict,
+    quote: "Quote | None" = None,
+    reserve_amount: Decimal | None = None,
+) -> AITask:
+    """建任务 + 预留积分(统一计价枢纽)。
+
+    quote 不传 = flat 计价(unit_price 积分/次,文本/图像走这里);视频/配音入口自带 quote。
+    reserve_amount 仅视频类传(= 积分×buffer,应对真实 tokens 超预估;ledger 禁超预留扣费)。
+    estimated_cost 记用户价(积分),base_cost 记平台成本(¥,未配置=0)。
+    """
+    from apps.billing.pricing import quote_flat
+
+    quote = quote or quote_flat(model_config, team=project.team)
+    # 汇率快照:margin_yuan 报表用「计价当时」的 points_per_yuan,汇率调整不追溯历史任务(review 确认)
+    if quote.meta.get("rate"):
+        request_payload = {**request_payload, "points_per_yuan_snapshot": quote.meta["rate"]}
     task = AITask.objects.create(
         team=project.team,
         created_by=user,
@@ -640,9 +660,10 @@ def create_ai_task(*, project, user, task_type: str, model_config: ModelConfig,
         model_config=model_config,
         idempotency_key=f"{task_type}:{project.id}:{uuid.uuid4()}",
         request_payload=request_payload,
-        estimated_cost=cost,
+        estimated_cost=quote.points,
+        base_cost=quote.base_cost_yuan,
     )
-    reserve_credit(team=project.team, user=user, task=task, amount=cost)
+    reserve_credit(team=project.team, user=user, task=task, amount=reserve_amount or quote.points)
     task.status = AITask.Status.RESERVED
     task.save(update_fields=["status", "updated_at"])
     return task
@@ -2216,17 +2237,36 @@ def submit_video_segment(*, video_segment: VideoSegment, user, prompt: str) -> V
     reference_images = [r["url"] for r in refs]
     final_prompt = build_video_segment_prompt(project, video_segment, scene, refs, prompt)
 
+    # 视频段 token 计量计价(与自由创作同一成本表+同一毛利):按 9:16/720p/目标时长预估,
+    # 预留=积分×buffer,终态按火山真实 usage.total_tokens 结算(poll_video_segment true-up)。
+    # 这里终结了「视频 ¥1/段、成本 ¥15」的倒贴定价。
+    from apps.billing.pricing import quote_video_estimate, video_reserve_amount
+
+    est_tokens, quote = quote_video_estimate(
+        model_config,
+        aspect_ratio="9:16",
+        resolution="720p",
+        duration=video_segment.target_duration_seconds,
+        references=[],
+        team=project.team,
+    )
     task = create_ai_task(
         project=project,
         user=user,
         task_type=AITask.Type.VIDEO_SEGMENT,
         model_config=model_config,
+        quote=quote,
+        reserve_amount=video_reserve_amount(quote.points),
         request_payload={
             "model": model_config.name,
             "endpoint": model_config.endpoint,
             "prompt": final_prompt,
             "duration": video_segment.target_duration_seconds,
             "ratio": "9:16",
+            "resolution": "720p",
+            "estimated_tokens": est_tokens,
+            # 团队价格系数快照:按实结算用它,中途改价不影响在途任务(jimeng 同款纪律)
+            "price_multiplier": quote.meta.get("price_multiplier", "1"),
             "video_segment_id": str(video_segment.id),
             "reference_images": reference_images,
         },
@@ -2343,12 +2383,44 @@ def poll_video_segment(*, video_segment: VideoSegment, user) -> VideoSegmentVers
             existing = video_segment.versions.filter(task=locked_task).order_by("-created_at").first()
             if existing is not None:
                 return existing
+        # 按火山真实 usage.total_tokens 结算(true-up,与自由创作同口径):
+        # 多退(charge 差额自动 RELEASE)/超预留 clamp(ledger 禁超扣,差额平台承担并告警)。
+        # usage 缺失(异常响应)回落预估价,不阻断出片。
+        from apps.billing.pricing import quote_video_actual
+
+        reservation = locked_task.credit_reservation
+        payload = locked_task.request_payload or {}
+        try:
+            usage_tokens = int((response.get("usage") or {}).get("total_tokens") or 0)
+        except (TypeError, ValueError):
+            usage_tokens = 0
+        if usage_tokens > 0:
+            settle = quote_video_actual(
+                locked_task.model_config,
+                tokens=usage_tokens,
+                with_video_ref=False,
+                resolution=str(payload.get("resolution") or "720p"),
+                multiplier=Decimal(str(payload.get("price_multiplier") or "1")),
+            )
+            actual_points, base_cost = settle.points, settle.base_cost_yuan
+            if actual_points > reservation.amount:
+                logger.warning(
+                    "video segment task %s actual %s exceeds reserved %s, clamped",
+                    locked_task.id, actual_points, reservation.amount,
+                )
+                actual_points = reservation.amount
+        else:
+            actual_points, base_cost = locked_task.estimated_cost, locked_task.base_cost
+        if usage_tokens > 0 and settle.meta.get("rate"):
+            payload["points_per_yuan_snapshot"] = settle.meta["rate"]
+            locked_task.request_payload = payload
         locked_task.status = AITask.Status.SUCCEEDED
         locked_task.response_payload = response
-        locked_task.actual_cost = locked_task.estimated_cost
+        locked_task.actual_cost = actual_points
+        locked_task.base_cost = base_cost
         locked_task.completed_at = timezone.now()
-        locked_task.save(update_fields=["status", "response_payload", "actual_cost", "completed_at", "updated_at"])
-        charge_reserved_credit(reservation=locked_task.credit_reservation, actual_amount=locked_task.actual_cost)
+        locked_task.save(update_fields=["status", "request_payload", "response_payload", "actual_cost", "base_cost", "completed_at", "updated_at"])
+        charge_reserved_credit(reservation=reservation, actual_amount=actual_points)
         version = VideoSegmentVersion.objects.create(
             video_segment=video_segment,
             task=locked_task,
@@ -2452,13 +2524,17 @@ def enqueue_standalone_images(*, team, user, prompt: str, mode: str = "image", c
     # 平台套图:规范化平台 id(前端 dy/tb… → canonical),用于注入平台版式块(优化版);非 cover 模式忽略。
     platform_key = str(platform_id or "").strip() if mode == "cover" else ""
     platform_name = _PLATFORM_NAMES.get(platform_key, "")
+    from apps.billing.pricing import quote_flat
+
     tasks: list[AITask] = []
     for index in range(count):
-        cost = estimate_cost(model_config)
+        quote = quote_flat(model_config, team=team)
         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, "model_entity_id": str(model_entity_id) if model_entity_id else None, "batch_id": batch_id, "ratio": str(ratio) if ratio else None, "reference_image_ids": ref_ids, "platform_id": platform_key or None, "platform_name": platform_name or None}
         # 只在重跑/补图时落键(不落 False):workbench 用 KeyTextTransform 抽文本,"false" 字符串也是真值,会误判
         if is_append:
             request_payload["batch_append"] = True
+        if quote.meta.get("rate"):
+            request_payload["points_per_yuan_snapshot"] = quote.meta["rate"]
         task = AITask.objects.create(
             team=team,
             created_by=user,
@@ -2469,10 +2545,11 @@ def enqueue_standalone_images(*, team, user, prompt: str, mode: str = "image", c
             model_config=model_config,
             idempotency_key=f"standalone-image:{team.id}:{uuid.uuid4()}",
             request_payload=request_payload,
-            estimated_cost=cost,
+            estimated_cost=quote.points,
+            base_cost=quote.base_cost_yuan,
         )
         # 预留额度若余额不足会抛 ValueError,在同步的 Web 请求里立刻反馈给前端(不会先建半套任务)
-        reserve_credit(team=team, user=user, task=task, amount=cost)
+        reserve_credit(team=team, user=user, task=task, amount=quote.points)
         task.status = AITask.Status.RESERVED
         task.save(update_fields=["status", "updated_at"])
         tasks.append(task)
@@ -2678,14 +2755,21 @@ def synthesize_project_voiceover(*, project, user, items: list[dict], voice_type
     model_config = get_default_model(ModelConfig.Capability.AUDIO)
     if model_config is None:
         raise ValueError("no active audio model configured")
+    # 配音按字符数阶梯计价(默认每 500 字 10 积分,不足按 500):长短脚本不再同价
+    from apps.billing.pricing import quote_voiceover
+
+    char_count = sum(len(text) for _, _, text in texts)
+    quote = quote_voiceover(model_config, char_count=char_count, team=project.team)
     task = create_ai_task(
         project=project,
         user=user,
         task_type=AITask.Type.VOICEOVER,
         model_config=model_config,
+        quote=quote,
         request_payload={
             "voice_type": voice_type,
             "speed_ratio": float(speed_ratio or 1.0),
+            "char_count": char_count,
             "items": [{"index": idx, "cue": j, "text": text} for idx, j, text in texts],
         },
     )