From 2ce06a02744b030f8b3e630bf0d52f1f64a8406a Mon Sep 17 00:00:00 2001 From: qian Date: Wed, 20 May 2026 08:35:17 +0800 Subject: [PATCH] =?UTF-8?q?fix(model):=E8=A7=86=E9=A2=91=E6=A8=A1=E5=9E=8B?= =?UTF-8?q?=E8=AE=A1=E8=B4=B9=E7=9B=B8=E5=85=B3?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- internal/logic/billing/task_billing.go | 189 ++++++++++++++++++++++--- internal/logic/task/async_polling.go | 100 +++++++++++-- internal/logic/task/async_provider.go | 57 ++++++++ relay/common/task_billing.go | 4 +- relay/common/task_provider.go | 3 + 5 files changed, 324 insertions(+), 29 deletions(-) diff --git a/internal/logic/billing/task_billing.go b/internal/logic/billing/task_billing.go index ff071d2..b19ecc9 100644 --- a/internal/logic/billing/task_billing.go +++ b/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 } diff --git a/internal/logic/task/async_polling.go b/internal/logic/task/async_polling.go index 9082288..b2c4aee 100644 --- a/internal/logic/task/async_polling.go +++ b/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 更新提交阶段写入的审计记录，补充异步任务最终结果 diff --git a/internal/logic/task/async_provider.go b/internal/logic/task/async_provider.go index a40e42a..655872b 100644 --- a/internal/logic/task/async_provider.go +++ b/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 { diff --git a/relay/common/task_billing.go b/relay/common/task_billing.go index f48cd12..30efb6c 100644 --- a/relay/common/task_billing.go +++ b/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 diff --git a/relay/common/task_provider.go b/relay/common/task_provider.go index 2ed2a17..a88186a 100644 --- a/relay/common/task_provider.go +++ b/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) }