diff --git a/internal/logic/billing/task_billing.go b/internal/logic/billing/task_billing.go index ae99aac..ff071d2 100644 --- a/internal/logic/billing/task_billing.go +++ b/internal/logic/billing/task_billing.go @@ -16,28 +16,47 @@ func NewTaskBillingProvider() common.TaskBillingProvider { } // EstimateTaskCost 估算任务费用 -// ratios 包含计费比率(如 duration_ratio, resolution_ratio) +// ratios 包含计费比率(如 video_input 折扣)和预估参数(如 duration 秒数、resolution 乘数) func (b *TaskBillingProviderImpl) EstimateTaskCost(ctx context.Context, tenantID int64, modelName string, ratios map[string]float64) (float64, error) { - // 基础价格从模型定价表获取 pricing, err := GetModelPrice(ctx, tenantID, modelName) if err != nil { return 0.01, nil } - baseCost := pricing.PerRequestPrice - if err != nil { - return 0, fmt.Errorf("estimate task cost: %w", err) + + var cost float64 + + // 如果配了 token 单价且有 duration 预估,用 token 模式估算 + if pricing.OutputPrice > 0 { + duration := 5.0 // 默认 5 秒 + if d, ok := ratios["duration"]; ok && d > 0 { + duration = d + } + resolutionMul := 2.25 // 默认 720p + if r, ok := ratios["resolution"]; ok && r > 0 { + resolutionMul = r + } + + // 预估 tokens ≈ base_tokens_per_second × duration × resolution_multiplier + // base_tokens_per_second 根据模型定价反推: + // 火山方舟 completion_tokens 约 100 tokens/s (480p),用于预扣估算 + estimatedTokens := 100.0 * duration * resolutionMul + cost = estimatedTokens / 1_000_000.0 * pricing.OutputPrice * pricing.TenantMultiplier + } else { + cost = pricing.PerRequestPrice } - // 应用比率乘数 - for _, ratio := range ratios { - baseCost *= ratio + // 应用附加比率(video_input 折扣等) + for k, ratio := range ratios { + if k == "duration" || k == "resolution" { + continue + } + cost *= ratio } - // 最低消费保障 - if baseCost < 0.01 { - baseCost = 0.01 + if cost < 0.01 { + cost = 0.01 } - return baseCost, nil + return cost, nil } // PreDeductTask 预扣任务费用 @@ -102,3 +121,40 @@ func (b *TaskBillingProviderImpl) AdjustTaskBilling(ctx context.Context, tenantI } return newEstimatedCost, nil } + +// RecalculateByTokens 根据上游返回的 total_tokens 重算费用。 +// 公式:totalTokens / 1M × output_price × ratios_product +// 如果模型没有配置 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 { + return 0, nil + } + + pricing, err := GetModelPrice(ctx, tenantID, modelName) + if err != nil { + return 0, nil + } + + // 需要 output_price > 0 才能做 token 重算 + if pricing.OutputPrice <= 0 { + return 0, nil + } + + // 基础费用 = tokens × 单价 + cost := float64(totalTokens) / 1_000_000.0 * pricing.OutputPrice + + // 应用租户倍率 + cost *= pricing.TenantMultiplier + + // 应用附加比率(视频输入折扣等) + for _, ratio := range ratios { + cost *= ratio + } + + // 最低消费 + if cost < 0.01 { + cost = 0.01 + } + + return cost, nil +} diff --git a/internal/logic/task/async_polling.go b/internal/logic/task/async_polling.go index 43d6045..0ea6ed9 100644 --- a/internal/logic/task/async_polling.go +++ b/internal/logic/task/async_polling.go @@ -114,12 +114,12 @@ func processPlatformTasks(ctx context.Context, platform string, tasks []*common. } for channelID, channelTasks := range channelGroups { - processChannelTasks(ctx, platform, channelID, channelTasks) + processChannelTasks(ctx, channelID, channelTasks) } } // processChannelTasks 处理同一渠道的任务 -func processChannelTasks(ctx context.Context, platform string, channelID int64, tasks []*common.AsyncTask) { +func processChannelTasks(ctx context.Context, channelID int64, tasks []*common.AsyncTask) { // 获取渠道信息 channel, err := DefaultAsyncProvider.GetChannelByID(ctx, channelID) if err != nil || channel == nil { @@ -139,20 +139,29 @@ func processChannelTasks(ctx context.Context, platform string, channelID int64, } } +// privateData PrivateData 反序列化结构 +type privateData struct { + UpstreamTaskID string `json:"upstream_task_id"` + TaskType string `json:"task_type"` + BillingContext struct { + Ratios map[string]float64 `json:"ratios"` + ModelName string `json:"model_name"` + PreDeduct float64 `json:"pre_deduct"` + } `json:"billing_context"` +} + // pollSingleTask 轮询单个任务 func pollSingleTask(ctx context.Context, adaptor common.TaskAdaptor, channel *common.ChannelBasicInfo, task *common.AsyncTask) { - // 从 PrivateData 提取上游任务 ID - var private struct { - UpstreamTaskID string `json:"upstream_task_id"` - } - if err := json.Unmarshal(task.PrivateData, &private); err != nil || private.UpstreamTaskID == "" { + // 从 PrivateData 提取上游任务 ID 和计费上下文 + var pd privateData + if err := json.Unmarshal(task.PrivateData, &pd); err != nil || pd.UpstreamTaskID == "" { g.Log().Warningf(ctx, "poll: invalid private data for task %s", task.PublicTaskID) return } // 查询上游状态 taskData, _ := json.Marshal(map[string]any{ - "task_id": private.UpstreamTaskID, + "task_id": pd.UpstreamTaskID, }) resp, err := adaptor.FetchTask(channel.BaseURL, channel.ApiKey, taskData) @@ -214,9 +223,17 @@ func pollSingleTask(ctx context.Context, adaptor common.TaskAdaptor, channel *co if task.PreDeductAmount > 0 && !task.BillingSettled { taskBilling := billing.NewTaskBillingProvider() actualCost := task.PreDeductAmount + + // 优先用上游返回的 ActualCost if taskInfo.ActualCost > 0 { actualCost = taskInfo.ActualCost + } else if taskInfo.TotalTokens > 0 && pd.BillingContext.Ratios != nil { + // 用上游 total_tokens + 保存的 ratios 重算 + if tokenCost, err := taskBilling.RecalculateByTokens(ctx, task.TenantID, task.ModelName, taskInfo.TotalTokens, pd.BillingContext.Ratios); err == nil && tokenCost > 0 { + actualCost = tokenCost + } } + 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 { g.Log().Warningf(ctx, "poll: settle task %s: %v", task.PublicTaskID, err) diff --git a/relay/common/task_adaptor.go b/relay/common/task_adaptor.go index d958cc5..9b2f76a 100644 --- a/relay/common/task_adaptor.go +++ b/relay/common/task_adaptor.go @@ -59,6 +59,10 @@ type TaskInfo struct { Data []byte // 上游原始响应 SubTasks []SubTask // 子任务结果(如多图/多视频场景) ActualCost float64 // 上游返回的实际费用(0 表示未提供,使用预扣金额) + + // 上游 usage 信息,用于按 token 重算费用 + CompletionTokens int // 上游返回的 completion tokens + TotalTokens int // 上游返回的 total tokens } // TaskStatusEnum 任务状态枚举 diff --git a/relay/common/task_billing.go b/relay/common/task_billing.go index 35a2bca..f48cd12 100644 --- a/relay/common/task_billing.go +++ b/relay/common/task_billing.go @@ -19,4 +19,9 @@ type TaskBillingProvider interface { // AdjustTaskBilling 调整预扣金额(提交后上游确认了新参数) AdjustTaskBilling(ctx context.Context, tenantID int64, requestID string, preDeductAmount, newEstimatedCost float64) (float64, error) + + // RecalculateByTokens 根据上游返回的 total_tokens 重算费用 + // totalTokens: 上游返回的 token 计费单位 + // ratios: 提交时保存的计费比率(如 video_input 折扣) + RecalculateByTokens(ctx context.Context, tenantID int64, modelName string, totalTokens int, ratios map[string]float64) (float64, error) } diff --git a/relay/handler/task_handler.go b/relay/handler/task_handler.go index e2f8811..f5cb200 100644 --- a/relay/handler/task_handler.go +++ b/relay/handler/task_handler.go @@ -144,6 +144,23 @@ func HandleTaskSubmit( preDeductAmount, _ = billingProvider.AdjustTaskBilling(ctx, rc.TenantID, rc.RequestID, preDeductAmount, newCost) } + // 9.5 合并最终 ratios 用于结算时还原 + finalRatios := ratios + if adjustedRatios != nil { + if finalRatios == nil { + finalRatios = adjustedRatios + } else { + merged := make(map[string]float64, len(finalRatios)+len(adjustedRatios)) + for k, v := range finalRatios { + merged[k] = v + } + for k, v := range adjustedRatios { + merged[k] = v + } + finalRatios = merged + } + } + // 10. 生成公开任务 ID 并创建记录 publicTaskID := generatePublicTaskID() now := time.Now() @@ -151,6 +168,11 @@ func HandleTaskSubmit( privateData, _ := json.Marshal(map[string]any{ "upstream_task_id": upstreamTaskID, "task_type": platform, + "billing_context": map[string]any{ + "ratios": finalRatios, + "model_name": modelName, + "pre_deduct": preDeductAmount, + }, }) task := &common.AsyncTask{ diff --git a/relay/taskchannel/volcengine/adaptor.go b/relay/taskchannel/volcengine/adaptor.go index dd693f5..b506073 100644 --- a/relay/taskchannel/volcengine/adaptor.go +++ b/relay/taskchannel/volcengine/adaptor.go @@ -61,10 +61,16 @@ type taskResponse struct { Content struct { VideoURL string `json:"video_url"` } `json:"content"` - Seed int `json:"seed"` - Duration int `json:"duration"` - Ratio string `json:"ratio"` - Error struct { + Seed int `json:"seed"` + Duration int `json:"duration"` + Ratio string `json:"ratio"` + Resolution string `json:"resolution"` + FramesPerSecond int `json:"framespersecond"` + Usage struct { + CompletionTokens int `json:"completion_tokens"` + TotalTokens int `json:"total_tokens"` + } `json:"usage"` + Error struct { Code string `json:"code"` Message string `json:"message"` } `json:"error"` @@ -94,18 +100,31 @@ func (a *VolcengineVideoAdaptor) ValidateRequest(_ context.Context, _ *common.Re } func (a *VolcengineVideoAdaptor) EstimateBilling(_ context.Context, info *common.RelayInfo, body []byte) map[string]float64 { - ratios := map[string]float64{"base": 1.0} - - // 检测是否含视频输入,应用折扣比率 var req map[string]any if json.Unmarshal(body, &req) != nil { - return ratios + return nil } + + ratios := make(map[string]float64) + + // 提取 duration 用于 token 预估(默认 5s) + duration := extractDuration(req) + if duration > 0 { + ratios["duration"] = float64(duration) + } + + // 提取 resolution 用于 token 预估 + if resolution := extractResolution(req); resolution != "" { + ratios["resolution"] = resolutionMultiplier(resolution) + } + + // 检测是否含视频输入,应用折扣比率 if hasVideoInput(req) { if ratio, ok := getVideoInputRatio(info.ChannelMeta.UpstreamModelName); ok { ratios["video_input"] = ratio } } + return ratios } @@ -335,6 +354,9 @@ func (a *VolcengineVideoAdaptor) ParseTaskResult(body []byte) (*common.TaskInfo, info.Status = common.TaskStatusSuccess info.Progress = "100%" info.ResultURL = resp.Content.VideoURL + // 提取上游 usage 用于按 token 计费 + info.CompletionTokens = resp.Usage.CompletionTokens + info.TotalTokens = resp.Usage.TotalTokens case "failed": info.Status = common.TaskStatusFailure info.FailReason = resp.Error.Message @@ -382,3 +404,42 @@ func parseInt(s string) (int, error) { _, err := fmt.Sscanf(s, "%d", &n) return n, err } + +// extractDuration 从请求中提取 duration 秒数 +func extractDuration(req map[string]any) int { + if metadata, ok := req["metadata"].(map[string]any); ok { + if d, ok := metadata["duration"].(float64); ok && d > 0 { + return int(d) + } + } + if s, ok := req["seconds"].(string); ok { + if d, err := parseInt(s); err == nil && d > 0 { + return d + } + } + return 5 +} + +// extractResolution 从请求中提取 resolution +func extractResolution(req map[string]any) string { + if metadata, ok := req["metadata"].(map[string]any); ok { + if r, ok := metadata["resolution"].(string); ok && r != "" { + return r + } + } + return "720p" +} + +// resolutionMultiplier 分辨率乘数(相对于 480p) +func resolutionMultiplier(resolution string) float64 { + switch resolution { + case "480p": + return 1.0 + case "720p": + return 2.25 + case "1080p": + return 5.06 + default: + return 2.25 + } +}