fix(model):视频模型计费相关

internal/logic/billing/task_billing.go

@@ -3,7 +3,13 @@ package billing
 import (
 	"context"
 	"fmt"
+	"time"
 
+	"github.com/gogf/gf/v2/frame/g"
+	"github.com/gogf/gf/v2/os/gtime"
+
+	"github.com/qianfree/team-api/internal/dao"
+	do "github.com/qianfree/team-api/internal/model/do"
 	"github.com/qianfree/team-api/relay/common"
 )
 
@@ -25,8 +31,10 @@ func (b *TaskBillingProviderImpl) EstimateTaskCost(ctx context.Context, tenantID
 
 	var cost float64
 
-	// 如果配了 token 单价且有 duration 预估，用 token 模式估算
-	if pricing.OutputPrice > 0 {
+	// 按次计费：直接用单价
+	if pricing.BillingMode == "per_request" {
+		cost = pricing.PerRequestPrice
+	} else if pricing.OutputPrice > 0 {
 		duration := 5.0 // 默认 5 秒
 		if d, ok := ratios["duration"]; ok && d > 0 {
 			duration = d
@@ -37,9 +45,8 @@ func (b *TaskBillingProviderImpl) EstimateTaskCost(ctx context.Context, tenantID
 		}
 
 		// 预估 tokens ≈ base_tokens_per_second × duration × resolution_multiplier
-		// base_tokens_per_second 根据模型定价反推：
-		// 火山方舟 completion_tokens 约 100 tokens/s (480p)，用于预扣估算
-		estimatedTokens := 100.0 * duration * resolutionMul
+		// 火山方舟视频生成约 10000 tokens/s (480p 基准)，用于预扣估算
+		estimatedTokens := 10000.0 * duration * resolutionMul
 		cost = estimatedTokens / 1_000_000.0 * pricing.OutputPrice * pricing.TenantMultiplier
 	} else {
 		cost = pricing.PerRequestPrice
@@ -68,30 +75,170 @@ func (b *TaskBillingProviderImpl) PreDeductTask(ctx context.Context, tenantID in
 	return estimatedCost, nil
 }
 
-// SettleTaskSuccess 任务成功结算
-func (b *TaskBillingProviderImpl) SettleTaskSuccess(ctx context.Context, tenantID, userID, apiKeyID, channelID int64, modelName, requestID string, actualCost, preDeductAmount float64) error {
+// SettleTaskSuccess 任务成功结算（含计费快照）
+// totalTokens/completionTokens: 上游返回的 token 用量
+// ratios: 提交时保存的计费比率（如 video_input 折扣）
+func (b *TaskBillingProviderImpl) SettleTaskSuccess(ctx context.Context, tenantID, userID, apiKeyID, channelID int64, modelName, requestID string, actualCost, preDeductAmount float64, totalTokens, completionTokens int, ratios map[string]float64) (*common.SettlementResult, error) {
 	diff := actualCost - preDeductAmount
 
-	// actualCost > preDeductAmount: 补扣差额
-	// actualCost < preDeductAmount: 退还差额
-	// 这里复用 Settle 方法，以 token 等价方式记录
-	// 用实际费用/预扣费用差值来结算
-	_, err := Settle(ctx, tenantID, userID, apiKeyID, channelID, modelName, requestID, "task", 0, 0, preDeductAmount)
+	// 1. 钱包操作
+	wallet, err := GetWallet(ctx, tenantID)
 	if err != nil {
-		return fmt.Errorf("settle task success: %w", err)
+		return nil, fmt.Errorf("settle task: get wallet: %w", err)
 	}
 
-	// 如果有差额需要补扣或退还
+	now := time.Now()
+	_, err = g.DB().Exec(ctx,
+		"UPDATE bil_wallets SET frozen_balance = GREATEST(frozen_balance - ?, 0), balance = balance - ?, updated_at = ? WHERE id = ?",
+		preDeductAmount, actualCost, now, wallet.ID)
+	if err != nil {
+		return nil, fmt.Errorf("settle task: update wallet: %w", err)
+	}
+
+	walletCache.Delete(ctx, fmt.Sprintf("%d", tenantID))
+	InvalidateWalletRedis(ctx, tenantID)
+
+	// 2. 获取定价 + 构建 CostBreakdown
+	pricing, _ := GetModelPrice(ctx, tenantID, modelName)
+	breakdown := buildTaskCostBreakdown(pricing, actualCost, totalTokens, completionTokens)
+
+	// 3. 创建计费记录（含快照字段）
+	var billingMode string
+	var discountRatio, effectiveOutputPrice float64
+	if pricing != nil {
+		billingMode = pricing.BillingMode
+		discountRatio = pricing.DiscountRatio
+		effectiveOutputPrice = pricing.OutputPrice
+	}
+
+	_, _ = dao.BilRecords.Ctx(ctx).Insert(do.BilRecords{
+		TenantId:     tenantID,
+		UserId:       userID,
+		ApiKeyId:     apiKeyID,
+		ChannelId:    channelID,
+		ModelName:    modelName,
+		RequestId:    requestID,
+		RelayMode:    "task",
+		InputTokens:  0,
+		OutputTokens: totalTokens,
+		InputPrice:   0,
+		OutputPrice:  effectiveOutputPrice,
+		TotalCost:    actualCost,
+		Currency:     "USD",
+		Status:       "settled",
+		SettledAt:    gtime.NewFromTime(now),
+		BillingMode:  billingMode,
+		DiscountRatio: func() float64 {
+			if discountRatio > 0 {
+				return discountRatio
+			}
+			return 0
+		}(),
+		EffectiveInputPrice:     0,
+		EffectiveOutputPrice:    effectiveOutputPrice,
+		BillingInputMultiplier:  0,
+		BillingOutputMultiplier: 0,
+		CacheCreationTokens:     0,
+		CacheReadTokens:         0,
+		CacheCreationCost:       0,
+		CacheReadCost:           0,
+	})
+
+	// 4. 生成计费快照 + 摘要
+	result := &common.SettlementResult{
+		PreDeductAmount: preDeductAmount,
+		ActualCost:      actualCost,
+		BaseCost:        breakdown.BaseCost,
+		TotalCost:       actualCost,
+		OutputCost:      breakdown.OutputCost,
+	}
+
+	if diff > 0.001 {
+		result.SupplementAmount = diff
+	} else if diff < -0.001 {
+		result.RefundAmount = -diff
+	}
+
+	if pricing != nil {
+		internalSettlement := &SettlementResult{
+			PreDeductAmount:  preDeductAmount,
+			ActualCost:       actualCost,
+			BaseCost:         breakdown.BaseCost,
+			RefundAmount:     result.RefundAmount,
+			SupplementAmount: result.SupplementAmount,
+		}
+		snapshot := GenerateBillingSnapshot(pricing, breakdown, nil, internalSettlement, nil)
+		snapshot.RequestMeta.RequestedModel = modelName
+		result.BillingSnapshot = SnapshotToJSON(snapshot)
+		result.BillingSummary = GenerateBillingSummary(snapshot)
+		result.BillingMode = pricing.BillingMode
+		result.BillingSource = pricing.BillingSource
+		result.RateMultiplier = pricing.DiscountRatio
+	}
+
+	// 5. 差额处理
 	if diff > 0.001 {
 		if ok, err := PreDeduct(ctx, tenantID, diff, requestID+"_adjust"); !ok {
-			return fmt.Errorf("settle task adjust pre-deduct: %w", err)
+			return result, fmt.Errorf("settle task adjust pre-deduct: %w", err)
 		}
 	} else if diff < -0.001 {
 		if err := SettleFailed(ctx, tenantID, requestID+"_adjust", -diff); err != nil {
-			return fmt.Errorf("settle task adjust refund: %w", err)
+			return result, fmt.Errorf("settle task adjust refund: %w", err)
 		}
 	}
-	return nil
+
+	// 6. 记录流水
+	recordTransaction(ctx, wallet.ID, tenantID, "settle", -actualCost,
+		fmt.Sprintf("settle: %s model=%s", requestID, modelName))
+	if result.RefundAmount > 0 {
+		recordTransaction(ctx, wallet.ID, tenantID, "refund", result.RefundAmount,
+			fmt.Sprintf("refund: %s", requestID))
+	}
+	if result.SupplementAmount > 0 {
+		recordTransaction(ctx, wallet.ID, tenantID, "pre_deduct", -result.SupplementAmount,
+			fmt.Sprintf("supplement: %s", requestID))
+	}
+
+	return result, nil
+}
+
+// buildTaskCostBreakdown 构建任务计费的 CostBreakdown
+func buildTaskCostBreakdown(pricing *PricingResult, actualCost float64, totalTokens, _ int) *CostBreakdown {
+	if pricing == nil {
+		return &CostBreakdown{
+			TotalCost: actualCost,
+			BaseCost:  actualCost,
+			Currency:  "USD",
+		}
+	}
+
+	bd := &CostBreakdown{
+		BillingMode:      pricing.BillingMode,
+		PerRequestPrice:  pricing.PerRequestPrice,
+		DiscountRatio:    pricing.DiscountRatio,
+		TenantMultiplier: pricing.TenantMultiplier,
+		Currency:         pricing.Currency,
+	}
+
+	if pricing.BillingMode == "per_request" {
+		bd.BaseCost = pricing.PerRequestPrice
+		bd.TotalCost = actualCost
+		return bd
+	}
+
+	// token 模式
+	bd.OutputTokens = totalTokens
+	bd.OutputCost = actualCost
+	bd.TotalCost = actualCost
+
+	tenantMul := pricing.TenantMultiplier
+	if tenantMul > 0 {
+		bd.BaseCost = actualCost / tenantMul
+	} else {
+		bd.BaseCost = actualCost
+	}
+
+	return bd
 }
 
 // SettleTaskFailed 任务失败退还预扣
@@ -123,7 +270,7 @@ func (b *TaskBillingProviderImpl) AdjustTaskBilling(ctx context.Context, tenantI
 }
 
 // RecalculateByTokens 根据上游返回的 total_tokens 重算费用。
-// 公式：totalTokens / 1M × output_price × ratios_product
+// 公式：totalTokens / 1M × output_price × tenant_multiplier × 附加比率
 // 如果模型没有配置 token 单价（纯按次计费），返回 0 表示不做 token 重算。
 func (b *TaskBillingProviderImpl) RecalculateByTokens(ctx context.Context, tenantID int64, modelName string, totalTokens int, ratios map[string]float64) (float64, error) {
 	if totalTokens <= 0 {
@@ -147,7 +294,11 @@ func (b *TaskBillingProviderImpl) RecalculateByTokens(ctx context.Context, tenan
 	cost *= pricing.TenantMultiplier
 
 	// 应用附加比率（视频输入折扣等）
-	for _, ratio := range ratios {
+	// 注意：跳过 duration/resolution，它们已体现在上游返回的 token 数中，不应再乘
+	for k, ratio := range ratios {
+		if k == "duration" || k == "resolution" {
+			continue
+		}
 		cost *= ratio
 	}
 

internal/logic/task/async_polling.go

@@ -51,7 +51,10 @@ func pollOnce(ctx context.Context) {
 	// 1. 处理超时任务
 	handleTimedOutTasks(ctx)
 
-	// 2. 轮询非终态任务
+	// 2. 重试未结算的终态任务
+	handleUnsettledTasks(ctx)
+
+	// 3. 轮询非终态任务
 	tasks, err := DefaultAsyncProvider.GetNonTerminalTasks(ctx, pollBatchSize)
 	if err != nil {
 		g.Log().Warningf(ctx, "poll: get non-terminal tasks: %v", err)
@@ -110,6 +113,61 @@ func handleTimedOutTasks(ctx context.Context) {
 	}
 }
 
+// handleUnsettledTasks 重试终态但未结算的任务
+func handleUnsettledTasks(ctx context.Context) {
+	tasks, err := DefaultAsyncProvider.GetUnsettledTasks(ctx, timeoutBatchSize)
+	if err != nil {
+		g.Log().Warningf(ctx, "poll: get unsettled tasks: %v", err)
+		return
+	}
+
+	for _, t := range tasks {
+		var pd privateData
+		if err := json.Unmarshal(t.PrivateData, &pd); err != nil || pd.UpstreamTaskID == "" {
+			// 无法恢复，直接退还预扣
+			g.Log().Warningf(ctx, "poll: unsettled task %s has invalid private_data, refunding", t.PublicTaskID)
+			taskBilling := billing.NewTaskBillingProvider()
+			if err := taskBilling.SettleTaskFailed(ctx, t.TenantID, t.PublicTaskID, t.PreDeductAmount); err != nil {
+				g.Log().Errorf(ctx, "poll: refund unsettled task %s: %v", t.PublicTaskID, err)
+			} else {
+				t.BillingSettled = true
+				DefaultAsyncProvider.UpdateTask(ctx, t)
+			}
+			continue
+		}
+
+		if t.Status == "SUCCESS" {
+			// 成功任务：用 ActualCost（已在上次轮询中计算）结算
+			actualCost := t.ActualCost
+			if actualCost <= 0 {
+				actualCost = t.PreDeductAmount
+			}
+			taskBilling := billing.NewTaskBillingProvider()
+			_, err := taskBilling.SettleTaskSuccess(ctx, t.TenantID, t.UserID, t.ApiKeyID, t.ChannelID,
+				t.ModelName, t.PublicTaskID, actualCost, t.PreDeductAmount,
+				0, 0, pd.BillingContext.Ratios)
+			if err != nil {
+				g.Log().Warningf(ctx, "poll: retry settle task %s: %v", t.PublicTaskID, err)
+			} else {
+				t.BillingSettled = true
+				t.ActualCost = actualCost
+				DefaultAsyncProvider.UpdateTask(ctx, t)
+				g.Log().Infof(ctx, "poll: retried settlement for task %s", t.PublicTaskID)
+			}
+		} else {
+			// 失败任务：退还预扣
+			taskBilling := billing.NewTaskBillingProvider()
+			if err := taskBilling.SettleTaskFailed(ctx, t.TenantID, t.PublicTaskID, t.PreDeductAmount); err != nil {
+				g.Log().Warningf(ctx, "poll: retry refund task %s: %v", t.PublicTaskID, err)
+			} else {
+				t.BillingSettled = true
+				DefaultAsyncProvider.UpdateTask(ctx, t)
+				g.Log().Infof(ctx, "poll: retried refund for task %s", t.PublicTaskID)
+			}
+		}
+	}
+}
+
 // processPlatformTasks 处理同一平台的任务
 func processPlatformTasks(ctx context.Context, platform string, tasks []*common.AsyncTask) {
 	// 按 channel_id 分组
@@ -231,6 +289,7 @@ func pollSingleTask(ctx context.Context, adaptor common.TaskAdaptor, channel *co
 		DefaultAsyncProvider.UpdateTask(ctx, task) // 更新 result_url
 
 		// 结算计费
+		var settleResult *common.SettlementResult
 		if task.PreDeductAmount > 0 && !task.BillingSettled {
 			taskBilling := billing.NewTaskBillingProvider()
 			actualCost := task.PreDeductAmount
@@ -246,7 +305,8 @@ func pollSingleTask(ctx context.Context, adaptor common.TaskAdaptor, channel *co
 			}
 
 			task.ActualCost = actualCost
-			if err := taskBilling.SettleTaskSuccess(ctx, task.TenantID, task.UserID, task.ApiKeyID, task.ChannelID, task.ModelName, task.PublicTaskID, actualCost, task.PreDeductAmount); err != nil {
+			settleResult, err = taskBilling.SettleTaskSuccess(ctx, task.TenantID, task.UserID, task.ApiKeyID, task.ChannelID, task.ModelName, task.PublicTaskID, actualCost, task.PreDeductAmount, taskInfo.TotalTokens, taskInfo.CompletionTokens, pd.BillingContext.Ratios)
+			if err != nil {
 				g.Log().Warningf(ctx, "poll: settle task %s: %v", task.PublicTaskID, err)
 			} else {
 				task.BillingSettled = true
@@ -256,7 +316,7 @@ func pollSingleTask(ctx context.Context, adaptor common.TaskAdaptor, channel *co
 		g.Log().Infof(ctx, "poll: task %s completed", task.PublicTaskID)
 
 		// 记录用量日志
-		recordTaskUsage(task, true, "")
+		recordTaskUsage(task, true, "", settleResult)
 
 		// 更新审计记录
 		recordTaskCompletionAudit(task, "SUCCESS", string(body), upstreamRespHeaders(resp))
@@ -275,7 +335,7 @@ func pollSingleTask(ctx context.Context, adaptor common.TaskAdaptor, channel *co
 		g.Log().Infof(ctx, "poll: task %s failed: %s", task.PublicTaskID, task.FailReason)
 
 		// 记录用量日志
-		recordTaskUsage(task, false, task.FailReason)
+		recordTaskUsage(task, false, task.FailReason, nil)
 
 		// 更新审计记录
 		recordTaskCompletionAudit(task, "FAILURE", string(body), upstreamRespHeaders(resp))
@@ -283,7 +343,7 @@ func pollSingleTask(ctx context.Context, adaptor common.TaskAdaptor, channel *co
 }
 
 // recordTaskUsage 异步记录视频/音乐等异步任务的用量日志
-func recordTaskUsage(task *common.AsyncTask, success bool, errMsg string) {
+func recordTaskUsage(task *common.AsyncTask, success bool, errMsg string, settleResult *common.SettlementResult) {
 	latencyMs := 0
 	if task.SubmitTime != nil && task.FinishTime != nil {
 		latencyMs = int(task.FinishTime.Sub(*task.SubmitTime).Milliseconds())
@@ -294,7 +354,7 @@ func recordTaskUsage(task *common.AsyncTask, success bool, errMsg string) {
 		status = "error"
 	}
 
-	relay.NewDataProvider().RecordUsage(context.Background(), &common.UsageRecord{
+	record := &common.UsageRecord{
 		TenantID:         task.TenantID,
 		UserID:           task.UserID,
 		ApiKeyID:         task.ApiKeyID,
@@ -314,10 +374,34 @@ func recordTaskUsage(task *common.AsyncTask, success bool, errMsg string) {
 		TotalCost:        task.ActualCost,
 		ActualCost:       task.ActualCost,
 		PreDeductAmount:  task.PreDeductAmount,
-		BillingMode:      "per_request",
 		BillingSource:    "task",
 		TaskID:           task.PublicTaskID,
-	})
+	}
+
+	// 从结算结果填充计费快照
+	if settleResult != nil {
+		record.BillingSnapshot = settleResult.BillingSnapshot
+		record.BillingSummary = settleResult.BillingSummary
+		record.BillingMode = settleResult.BillingMode
+		record.BillingSource = settleResult.BillingSource
+		record.RateMultiplier = settleResult.RateMultiplier
+		record.InputCost = settleResult.InputCost
+		record.OutputCost = settleResult.OutputCost
+		record.RefundAmount = settleResult.RefundAmount
+		record.SupplementAmount = settleResult.SupplementAmount
+		record.Currency = "USD"
+	} else {
+		// 失败/超时任务：从定价中获取计费模式
+		billingMode := "per_request"
+		if pricing, err := billing.GetModelPrice(context.Background(), task.TenantID, task.ModelName); err == nil {
+			if pricing.BillingMode != "" {
+				billingMode = pricing.BillingMode
+			}
+		}
+		record.BillingMode = billingMode
+	}
+
+	relay.NewDataProvider().RecordUsage(context.Background(), record)
 }
 
 // recordTaskCompletionAudit 更新提交阶段写入的审计记录，补充异步任务最终结果

internal/logic/task/async_provider.go

@@ -344,6 +344,63 @@ func (p *AsyncProvider) GetTimedOutTasks(ctx context.Context, cutoffUnix int64,
 	return tasks, nil
 }
 
+// GetUnsettledTasks 获取终态但未结算的任务（用于结算重试）
+func (p *AsyncProvider) GetUnsettledTasks(ctx context.Context, limit int) ([]*common.AsyncTask, error) {
+	var rows []struct {
+		ID              int64           `json:"id"`
+		PublicTaskID    string          `json:"public_task_id"`
+		RequestId       string          `json:"request_id"`
+		Platform        string          `json:"platform"`
+		Status          string          `json:"status"`
+		TenantID        int64           `json:"tenant_id"`
+		UserID          int64           `json:"user_id"`
+		ApiKeyID        int64           `json:"api_key_id"`
+		ChannelID       int64           `json:"channel_id"`
+		ModelName       string          `json:"model_name"`
+		PreDeductAmount float64         `json:"pre_deduct_amount"`
+		ActualCost      float64         `json:"actual_cost"`
+		BillingSettled  bool            `json:"billing_settled"`
+		Data            json.RawMessage `json:"data"`
+		PrivateData     json.RawMessage `json:"private_data"`
+		SubmitTime      *time.Time      `json:"submit_time"`
+		CreatedAt       time.Time       `json:"created_at"`
+	}
+	err := dao.TskModelTasks.Ctx(ctx).
+		Where("status IN (?, ?)", "SUCCESS", "FAILURE").
+		Where("billing_settled = ?", false).
+		Where("pre_deduct_amount > 0").
+		Order("submit_time ASC").
+		Limit(limit).
+		Scan(&rows)
+	if err != nil {
+		return nil, gerror.Wrapf(err, "query unsettled tasks failed")
+	}
+
+	tasks := make([]*common.AsyncTask, 0, len(rows))
+	for _, r := range rows {
+		tasks = append(tasks, &common.AsyncTask{
+			ID:              r.ID,
+			PublicTaskID:    r.PublicTaskID,
+			RequestID:       r.RequestId,
+			Platform:        r.Platform,
+			Status:          r.Status,
+			TenantID:        r.TenantID,
+			UserID:          r.UserID,
+			ApiKeyID:        r.ApiKeyID,
+			ChannelID:       r.ChannelID,
+			ModelName:       r.ModelName,
+			PreDeductAmount: r.PreDeductAmount,
+			ActualCost:      r.ActualCost,
+			BillingSettled:  r.BillingSettled,
+			Data:            r.Data,
+			PrivateData:     r.PrivateData,
+			SubmitTime:      r.SubmitTime,
+			CreatedAt:       r.CreatedAt,
+		})
+	}
+	return tasks, nil
+}
+
 // GetChannelByID 获取渠道基本信息（含从 chn_channel_keys 解密的 API Key）
 func (p *AsyncProvider) GetChannelByID(ctx context.Context, channelID int64) (*common.ChannelBasicInfo, error) {
 	var row struct {

relay/common/task_billing.go

@@ -11,8 +11,8 @@ type TaskBillingProvider interface {
 	// PreDeductTask 预扣任务费用
 	PreDeductTask(ctx context.Context, tenantID int64, requestID string, estimatedCost float64) (float64, error)
 
-	// SettleTaskSuccess 任务成功结算
-	SettleTaskSuccess(ctx context.Context, tenantID, userID, apiKeyID, channelID int64, modelName, requestID string, actualCost, preDeductAmount float64) error
+	// SettleTaskSuccess 任务成功结算（含计费快照）
+	SettleTaskSuccess(ctx context.Context, tenantID, userID, apiKeyID, channelID int64, modelName, requestID string, actualCost, preDeductAmount float64, totalTokens, completionTokens int, ratios map[string]float64) (*SettlementResult, error)
 
 	// SettleTaskFailed 任务失败退还预扣
 	SettleTaskFailed(ctx context.Context, tenantID int64, requestID string, preDeductAmount float64) error

relay/common/task_provider.go

@@ -29,6 +29,9 @@ type TaskDataProvider interface {
 	// GetTimedOutTasks 获取超时未完成任务
 	GetTimedOutTasks(ctx context.Context, cutoffUnix int64, limit int) ([]*AsyncTask, error)
 
+	// GetUnsettledTasks 获取终态但未结算的任务（结算重试）
+	GetUnsettledTasks(ctx context.Context, limit int) ([]*AsyncTask, error)
+
 	// GetChannelByID 获取渠道基本信息
 	GetChannelByID(ctx context.Context, channelID int64) (*ChannelBasicInfo, error)
 }