github/new-api
1
0
Fork 0
mirror of https://github.com/QuantumNous/new-api.git synced 2026-03-30 02:25:00 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat(task): add adaptor billing interface and async settlement framework

Browse Source 
Add three billing lifecycle methods to the TaskAdaptor interface:
- EstimateBilling: compute OtherRatios from user request before pricing
- AdjustBillingOnSubmit: adjust ratios from upstream submit response
- AdjustBillingOnComplete: determine final quota at task terminal state

Introduce BaseBilling as embeddable no-op default for adaptors without
custom billing. Move Sora/Ali OtherRatios logic from shared validation
into per-adaptor EstimateBilling implementations.

Add TaskBillingContext to persist pricing params (model_price, group_ratio,
other_ratios) in task private data for async polling settlement.

Extract RecalculateTaskQuota as a general-purpose delta settlement
function and unify polling billing via settleTaskBillingOnComplete
(adaptor-first, then token-based fallback).
This commit is contained in:
CaIon
2026-02-10 21:15:09 +08:00
parent 9e3954428d
commit d6e11fd2e1
19 changed files with 321 additions and 116 deletions
Download Patch File Download Diff File Expand all files Collapse all files

30
relay/channel/adapter.go
View File

@@ -36,6 +36,32 @@ type TaskAdaptor interface {
ValidateRequestAndSetAction(c *gin.Context, info *relaycommon.RelayInfo) *dto.TaskError
// ── Billing ──────────────────────────────────────────────────────
// EstimateBilling returns OtherRatios for pre-charge based on user request.
// Called after ValidateRequestAndSetAction, before price calculation.
// Adaptors should extract duration, resolution, etc. from the parsed request
// and return them as ratio multipliers (e.g. {"seconds": 5, "size": 1.666}).
// Return nil to use the base model price without extra ratios.
EstimateBilling(c *gin.Context, info *relaycommon.RelayInfo) map[string]float64
// AdjustBillingOnSubmit returns adjusted OtherRatios from the upstream
// submit response. Called after a successful DoResponse.
// If the upstream returned actual parameters that differ from the estimate
// (e.g. actual seconds), return updated ratios so the caller can recalculate
// the quota and settle the delta with the pre-charge.
// Return nil if no adjustment is needed.
AdjustBillingOnSubmit(info *relaycommon.RelayInfo, taskData []byte) map[string]float64
// AdjustBillingOnComplete returns the actual quota when a task reaches a
// terminal state (success/failure) during polling.
// Called by the polling loop after ParseTaskResult.
// Return a positive value to trigger delta settlement (supplement / refund).
// Return 0 to keep the pre-charged amount unchanged.
AdjustBillingOnComplete(task *model.Task, taskResult *relaycommon.TaskInfo) int
// ── Request / Response ───────────────────────────────────────────
BuildRequestURL(info *relaycommon.RelayInfo) (string, error)
BuildRequestHeader(c *gin.Context, req *http.Request, info *relaycommon.RelayInfo) error
BuildRequestBody(c *gin.Context, info *relaycommon.RelayInfo) (io.Reader, error)
@@ -46,9 +72,9 @@ type TaskAdaptor interface {
GetModelList() []string
GetChannelName() string
// FetchTask
FetchTask(baseUrl, key string, body map[string]any, proxy string) (*http.Response, error)
// ── Polling ──────────────────────────────────────────────────────
FetchTask(baseUrl, key string, body map[string]any, proxy string) (*http.Response, error)
ParseTaskResult(respBody []byte) (*relaycommon.TaskInfo, error)
}

57
relay/channel/task/ali/adaptor.go
View File

@@ -13,6 +13,7 @@ import (
"github.com/QuantumNous/new-api/logger"
"github.com/QuantumNous/new-api/model"
"github.com/QuantumNous/new-api/relay/channel"
"github.com/QuantumNous/new-api/relay/channel/task/taskcommon"
relaycommon "github.com/QuantumNous/new-api/relay/common"
"github.com/QuantumNous/new-api/service"
"github.com/samber/lo"
@@ -108,10 +109,10 @@ type AliMetadata struct {
// ============================
type TaskAdaptor struct {
taskcommon.BaseBilling
ChannelType int
apiKey string
baseURL string
aliReq *AliVideoRequest
}
func (a *TaskAdaptor) Init(info *relaycommon.RelayInfo) {
@@ -121,17 +122,7 @@ func (a *TaskAdaptor) Init(info *relaycommon.RelayInfo) {
}
func (a *TaskAdaptor) ValidateRequestAndSetAction(c *gin.Context, info *relaycommon.RelayInfo) (taskErr *dto.TaskError) {
// 阿里通义万相支持 JSON 格式，不使用 multipart
var taskReq relaycommon.TaskSubmitReq
if err := common.UnmarshalBodyReusable(c, &taskReq); err != nil {
return service.TaskErrorWrapper(err, "unmarshal_task_request_failed", http.StatusBadRequest)
}
aliReq, err := a.convertToAliRequest(info, taskReq)
if err != nil {
return service.TaskErrorWrapper(err, "convert_to_ali_request_failed", http.StatusInternalServerError)
}
a.aliReq = aliReq
logger.LogJson(c, "ali video request body", aliReq)
// ValidateMultipartDirect 负责解析并将原始 TaskSubmitReq 存入 context
return relaycommon.ValidateMultipartDirect(c, info)
}
@@ -148,11 +139,21 @@ func (a *TaskAdaptor) BuildRequestHeader(c *gin.Context, req *http.Request, info
}
func (a *TaskAdaptor) BuildRequestBody(c *gin.Context, info *relaycommon.RelayInfo) (io.Reader, error) {
bodyBytes, err := common.Marshal(a.aliReq)
taskReq, err := relaycommon.GetTaskRequest(c)
if err != nil {
return nil, errors.Wrap(err, "get_task_request_failed")
}
aliReq, err := a.convertToAliRequest(info, taskReq)
if err != nil {
return nil, errors.Wrap(err, "convert_to_ali_request_failed")
}
logger.LogJson(c, "ali video request body", aliReq)
bodyBytes, err := common.Marshal(aliReq)
if err != nil {
return nil, errors.Wrap(err, "marshal_ali_request_failed")
}
return bytes.NewReader(bodyBytes), nil
}
@@ -335,19 +336,33 @@ func (a *TaskAdaptor) convertToAliRequest(info *relaycommon.RelayInfo, req relay
return nil, errors.New("can't change model with metadata")
}
info.PriceData.OtherRatios = map[string]float64{
return aliReq, nil
}
// EstimateBilling 根据用户请求参数计算 OtherRatios（时长、分辨率等）。
// 在 ValidateRequestAndSetAction 之后、价格计算之前调用。
func (a *TaskAdaptor) EstimateBilling(c *gin.Context, info *relaycommon.RelayInfo) map[string]float64 {
taskReq, err := relaycommon.GetTaskRequest(c)
if err != nil {
return nil
}
aliReq, err := a.convertToAliRequest(info, taskReq)
if err != nil {
return nil
}
otherRatios := map[string]float64{
"seconds": float64(aliReq.Parameters.Duration),
}
ratios, err := ProcessAliOtherRatios(aliReq)
if err != nil {
return nil, err
return otherRatios
}
for s, f := range ratios {
info.PriceData.OtherRatios[s] = f
for k, v := range ratios {
otherRatios[k] = v
}
return aliReq, nil
return otherRatios
}
// DoRequest delegates to common helper

1
relay/channel/task/doubao/adaptor.go
View File

@@ -89,6 +89,7 @@ type responseTask struct {
// ============================
type TaskAdaptor struct {
taskcommon.BaseBilling
ChannelType int
apiKey string
baseURL string

1
relay/channel/task/gemini/adaptor.go
View File

@@ -85,6 +85,7 @@ type operationResponse struct {
// ============================
type TaskAdaptor struct {
taskcommon.BaseBilling
ChannelType int
apiKey string
baseURL string

2
relay/channel/task/hailuo/adaptor.go
View File

@@ -17,12 +17,14 @@ import (
"github.com/QuantumNous/new-api/constant"
"github.com/QuantumNous/new-api/dto"
"github.com/QuantumNous/new-api/relay/channel"
taskcommon "github.com/QuantumNous/new-api/relay/channel/task/taskcommon"
relaycommon "github.com/QuantumNous/new-api/relay/common"
"github.com/QuantumNous/new-api/service"
)
// https://platform.minimaxi.com/docs/api-reference/video-generation-intro
type TaskAdaptor struct {
taskcommon.BaseBilling
ChannelType int
apiKey string
baseURL string

1
relay/channel/task/jimeng/adaptor.go
View File

@@ -77,6 +77,7 @@ const (
// ============================
type TaskAdaptor struct {
taskcommon.BaseBilling
ChannelType int
accessKey string
secretKey string

1
relay/channel/task/kling/adaptor.go
View File

@@ -97,6 +97,7 @@ type responsePayload struct {
// ============================
type TaskAdaptor struct {
taskcommon.BaseBilling
ChannelType int
apiKey string
baseURL string

44
relay/channel/task/sora/adaptor.go
View File

@@ -4,6 +4,7 @@ import (
"fmt"
"io"
"net/http"
"strconv"
"strings"
"github.com/QuantumNous/new-api/common"
@@ -11,6 +12,7 @@ import (
"github.com/QuantumNous/new-api/dto"
"github.com/QuantumNous/new-api/model"
"github.com/QuantumNous/new-api/relay/channel"
taskcommon "github.com/QuantumNous/new-api/relay/channel/task/taskcommon"
relaycommon "github.com/QuantumNous/new-api/relay/common"
"github.com/QuantumNous/new-api/service"
@@ -56,6 +58,7 @@ type responseTask struct {
// ============================
type TaskAdaptor struct {
taskcommon.BaseBilling
ChannelType int
apiKey string
baseURL string
@@ -68,15 +71,15 @@ func (a *TaskAdaptor) Init(info *relaycommon.RelayInfo) {
}
func validateRemixRequest(c *gin.Context) *dto.TaskError {
var req struct {
Prompt string `json:"prompt"`
}
var req relaycommon.TaskSubmitReq
if err := common.UnmarshalBodyReusable(c, &req); err != nil {
return service.TaskErrorWrapperLocal(err, "invalid_request", http.StatusBadRequest)
}
if strings.TrimSpace(req.Prompt) == "" {
return service.TaskErrorWrapperLocal(fmt.Errorf("field prompt is required"), "invalid_request", http.StatusBadRequest)
}
// 存储原始请求到 context，与 ValidateMultipartDirect 路径保持一致
c.Set("task_request", req)
return nil
}
@@ -87,6 +90,41 @@ func (a *TaskAdaptor) ValidateRequestAndSetAction(c *gin.Context, info *relaycom
return relaycommon.ValidateMultipartDirect(c, info)
}
// EstimateBilling 根据用户请求的 seconds 和 size 计算 OtherRatios。
func (a *TaskAdaptor) EstimateBilling(c *gin.Context, info *relaycommon.RelayInfo) map[string]float64 {
// remix 路径的 OtherRatios 已在 ResolveOriginTask 中设置
if info.Action == constant.TaskActionRemix {
return nil
}
req, err := relaycommon.GetTaskRequest(c)
if err != nil {
return nil
}
seconds, _ := strconv.Atoi(req.Seconds)
if seconds == 0 {
seconds = req.Duration
}
if seconds <= 0 {
seconds = 4
}
size := req.Size
if size == "" {
size = "720x1280"
}
ratios := map[string]float64{
"seconds": float64(seconds),
"size": 1,
}
if size == "1792x1024" || size == "1024x1792" {
ratios["size"] = 1.666667
}
return ratios
}
func (a *TaskAdaptor) BuildRequestURL(info *relaycommon.RelayInfo) (string, error) {
if info.Action == constant.TaskActionRemix {
return fmt.Sprintf("%s/v1/videos/%s/remix", a.baseURL, info.OriginTaskID), nil

7
relay/channel/task/suno/adaptor.go
View File

@@ -13,6 +13,7 @@ import (
"github.com/QuantumNous/new-api/constant"
"github.com/QuantumNous/new-api/dto"
"github.com/QuantumNous/new-api/relay/channel"
taskcommon "github.com/QuantumNous/new-api/relay/channel/task/taskcommon"
relaycommon "github.com/QuantumNous/new-api/relay/common"
"github.com/QuantumNous/new-api/service"
@@ -20,6 +21,7 @@ import (
)
type TaskAdaptor struct {
taskcommon.BaseBilling
ChannelType int
}
@@ -79,10 +81,7 @@ func (a *TaskAdaptor) BuildRequestHeader(c *gin.Context, req *http.Request, info
func (a *TaskAdaptor) BuildRequestBody(c *gin.Context, info *relaycommon.RelayInfo) (io.Reader, error) {
sunoRequest, ok := c.Get("task_request")
if !ok {
err := common.UnmarshalBodyReusable(c, &sunoRequest)
if err != nil {
return nil, err
}
return nil, fmt.Errorf("task_request not found in context")
}
data, err := common.Marshal(sunoRequest)
if err != nil {

25
relay/channel/task/taskcommon/helpers.go
View File

@@ -5,7 +5,10 @@ import (
"fmt"
"github.com/QuantumNous/new-api/common"
"github.com/QuantumNous/new-api/model"
relaycommon "github.com/QuantumNous/new-api/relay/common"
"github.com/QuantumNous/new-api/setting/system_setting"
"github.com/gin-gonic/gin"
)
// UnmarshalMetadata converts a map[string]any metadata to a typed struct via JSON round-trip.
@@ -68,3 +71,25 @@ const (
ProgressInProgress = "30%"
ProgressComplete = "100%"
)
// ---------------------------------------------------------------------------
// BaseBilling — embeddable no-op implementations for TaskAdaptor billing methods.
// Adaptors that do not need custom billing can embed this struct directly.
// ---------------------------------------------------------------------------
type BaseBilling struct{}
// EstimateBilling returns nil (no extra ratios; use base model price).
func (BaseBilling) EstimateBilling(_ *gin.Context, _ *relaycommon.RelayInfo) map[string]float64 {
return nil
}
// AdjustBillingOnSubmit returns nil (no submit-time adjustment).
func (BaseBilling) AdjustBillingOnSubmit(_ *relaycommon.RelayInfo, _ []byte) map[string]float64 {
return nil
}
// AdjustBillingOnComplete returns 0 (keep pre-charged amount).
func (BaseBilling) AdjustBillingOnComplete(_ *model.Task, _ *relaycommon.TaskInfo) int {
return 0
}

41
relay/channel/task/vertex/adaptor.go
View File

@@ -62,6 +62,7 @@ type operationResponse struct {
// ============================
type TaskAdaptor struct {
taskcommon.BaseBilling
ChannelType int
apiKey string
baseURL string
@@ -133,6 +134,28 @@ func (a *TaskAdaptor) BuildRequestHeader(c *gin.Context, req *http.Request, info
return nil
}
// EstimateBilling 根据用户请求中的 sampleCount 计算 OtherRatios。
func (a *TaskAdaptor) EstimateBilling(c *gin.Context, _ *relaycommon.RelayInfo) map[string]float64 {
sampleCount := 1
v, ok := c.Get("task_request")
if ok {
req := v.(relaycommon.TaskSubmitReq)
if req.Metadata != nil {
if sc, exists := req.Metadata["sampleCount"]; exists {
if i, ok := sc.(int); ok && i > 0 {
sampleCount = i
}
if f, ok := sc.(float64); ok && int(f) > 0 {
sampleCount = int(f)
}
}
}
}
return map[string]float64{
"sampleCount": float64(sampleCount),
}
}
// BuildRequestBody converts request into Vertex specific format.
func (a *TaskAdaptor) BuildRequestBody(c *gin.Context, info *relaycommon.RelayInfo) (io.Reader, error) {
v, ok := c.Get("task_request")
@@ -166,24 +189,6 @@ func (a *TaskAdaptor) BuildRequestBody(c *gin.Context, info *relaycommon.RelayIn
return nil, fmt.Errorf("sampleCount must be greater than 0")
}
// if req.Duration > 0 {
// body.Parameters["durationSeconds"] = req.Duration
// } else if req.Seconds != "" {
// seconds, err := strconv.Atoi(req.Seconds)
// if err != nil {
// return nil, errors.Wrap(err, "convert seconds to int failed")
// }
// body.Parameters["durationSeconds"] = seconds
// }
info.PriceData.OtherRatios = map[string]float64{
"sampleCount": float64(body.Parameters["sampleCount"].(int)),
}
// if v, ok := body.Parameters["durationSeconds"]; ok {
// info.PriceData.OtherRatios["durationSeconds"] = float64(v.(int))
// }
data, err := common.Marshal(body)
if err != nil {
return nil, err

1
relay/channel/task/vidu/adaptor.go
View File

@@ -73,6 +73,7 @@ type creation struct {
// ============================
type TaskAdaptor struct {
taskcommon.BaseBilling
ChannelType int
baseURL string
}

10
relay/common/relay_utils.go
View File

@@ -173,16 +173,10 @@ func ValidateMultipartDirect(c *gin.Context, info *RelayInfo) *dto.TaskError {
if model == "sora-2-pro" && !lo.Contains([]string{"720x1280", "1280x720", "1792x1024", "1024x1792"}, size) {
return createTaskError(fmt.Errorf("sora-2 size is invalid"), "invalid_size", http.StatusBadRequest, true)
}
info.PriceData.OtherRatios = map[string]float64{
"seconds": float64(seconds),
"size": 1,
}
if lo.Contains([]string{"1792x1024", "1024x1792"}, size) {
info.PriceData.OtherRatios["size"] = 1.666667
}
// OtherRatios 已移到 Sora adaptor 的 EstimateBilling 中设置
}
info.Action = action
storeTaskRequest(c, info, action, req)
return nil
}

64
relay/relay_task.go
View File

@@ -128,8 +128,9 @@ func ResolveOriginTask(c *gin.Context, info *relaycommon.RelayInfo) *dto.TaskErr
}
// RelayTaskSubmit 完成 task 提交的全部流程（每次尝试调用一次）：
// 刷新渠道元数据 → 确定 platform/adaptor → 验证请求 → 计算价格 →
// 预扣费（仅首次，通过 info.Billing==nil 守卫）→ 构建/发送/解析上游请求。
// 刷新渠道元数据 → 确定 platform/adaptor → 验证请求 →
// 估算计费(EstimateBilling) → 计算价格 → 预扣费（仅首次）→
// 构建/发送/解析上游请求 → 提交后计费调整(AdjustBillingOnSubmit)。
// 控制器负责 defer Refund 和成功后 Settle。
func RelayTaskSubmit(c *gin.Context, info *relaycommon.RelayInfo) (*TaskSubmitResult, *dto.TaskError) {
info.InitChannelMeta(c)
@@ -159,10 +160,20 @@ func RelayTaskSubmit(c *gin.Context, info *relaycommon.RelayInfo) (*TaskSubmitRe
info.PublicTaskID = model.GenerateTaskID()
}
// 4. 价格计算
// 4. 价格计算：基础模型价格
info.OriginModelName = modelName
info.PriceData = helper.ModelPriceHelperPerCall(c, info)
// 5. 计费估算：让适配器根据用户请求提供 OtherRatios（时长、分辨率等）
// 必须在 ModelPriceHelperPerCall 之后调用（它会重建 PriceData）。
// ResolveOriginTask 可能已在 remix 路径中预设了 OtherRatios，此处合并。
if estimatedRatios := adaptor.EstimateBilling(c, info); len(estimatedRatios) > 0 {
for k, v := range estimatedRatios {
info.PriceData.AddOtherRatio(k, v)
}
}
// 6. 将 OtherRatios 应用到基础额度
if !common.StringsContains(constant.TaskPricePatches, modelName) {
for _, ra := range info.PriceData.OtherRatios {
if ra != 1.0 {
@@ -171,7 +182,7 @@ func RelayTaskSubmit(c *gin.Context, info *relaycommon.RelayInfo) (*TaskSubmitRe
}
}
// 5. 预扣费（仅首次 — 重试时 info.Billing 已存在，跳过）
// 7. 预扣费（仅首次 — 重试时 info.Billing 已存在，跳过）
if info.Billing == nil && !info.PriceData.FreeModel {
info.ForcePreConsume = true
if apiErr := service.PreConsumeBilling(c, info.PriceData.Quota, info); apiErr != nil {
@@ -179,13 +190,13 @@ func RelayTaskSubmit(c *gin.Context, info *relaycommon.RelayInfo) (*TaskSubmitRe
}
}
// 6. 构建请求体
// 8. 构建请求体
requestBody, err := adaptor.BuildRequestBody(c, info)
if err != nil {
return nil, service.TaskErrorWrapper(err, "build_request_failed", http.StatusInternalServerError)
}
// 7. 发送请求
// 9. 发送请求
resp, err := adaptor.DoRequest(c, info, requestBody)
if err != nil {
return nil, service.TaskErrorWrapper(err, "do_request_failed", http.StatusInternalServerError)
@@ -195,20 +206,59 @@ func RelayTaskSubmit(c *gin.Context, info *relaycommon.RelayInfo) (*TaskSubmitRe
return nil, service.TaskErrorWrapper(fmt.Errorf("%s", string(responseBody)), "fail_to_fetch_task", resp.StatusCode)
}
// 8. 解析响应
// 10. 返回 OtherRatios 给下游（header 必须在 DoResponse 写 body 之前设置）
otherRatios := info.PriceData.OtherRatios
if otherRatios == nil {
otherRatios = map[string]float64{}
}
ratiosJSON, _ := common.Marshal(otherRatios)
c.Header("X-New-Api-Other-Ratios", string(ratiosJSON))
// 11. 解析响应
upstreamTaskID, taskData, taskErr := adaptor.DoResponse(c, resp, info)
if taskErr != nil {
return nil, taskErr
}
// 11. 提交后计费调整：让适配器根据上游实际返回调整 OtherRatios
finalQuota := info.PriceData.Quota
if adjustedRatios := adaptor.AdjustBillingOnSubmit(info, taskData); len(adjustedRatios) > 0 {
// 基于调整后的 ratios 重新计算 quota
finalQuota = recalcQuotaFromRatios(info, adjustedRatios)
info.PriceData.OtherRatios = adjustedRatios
info.PriceData.Quota = finalQuota
}
return &TaskSubmitResult{
UpstreamTaskID: upstreamTaskID,
TaskData: taskData,
Platform: platform,
ModelName: modelName,
Quota: finalQuota,
}, nil
}
// recalcQuotaFromRatios 根据 adjustedRatios 重新计算 quota。
// 公式: baseQuota × ∏(ratio) — 其中 baseQuota 是不含 OtherRatios 的基础额度。
func recalcQuotaFromRatios(info *relaycommon.RelayInfo, ratios map[string]float64) int {
// 从 PriceData 获取不含 OtherRatios 的基础价格
baseQuota := info.PriceData.Quota
// 先除掉原有的 OtherRatios 恢复基础额度
for _, ra := range info.PriceData.OtherRatios {
if ra != 1.0 && ra > 0 {
baseQuota = int(float64(baseQuota) / ra)
}
}
// 应用新的 ratios
result := float64(baseQuota)
for _, ra := range ratios {
if ra != 1.0 {
result *= ra
}
}
return int(result)
}
var fetchRespBuilders = map[int]func(c *gin.Context) (respBody []byte, taskResp *dto.TaskError){
relayconstant.RelayModeSunoFetchByID: sunoFetchByIDRespBodyBuilder,
relayconstant.RelayModeSunoFetch: sunoFetchRespBodyBuilder,