forked from QuantumNous/new-api
-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathtask_billing.go
More file actions
327 lines (295 loc) · 11.2 KB
/
Copy pathtask_billing.go
File metadata and controls
327 lines (295 loc) · 11.2 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
package service
import (
"context"
"fmt"
"strings"
"github.com/QuantumNous/new-api/common"
"github.com/QuantumNous/new-api/constant"
"github.com/QuantumNous/new-api/logger"
"github.com/QuantumNous/new-api/model"
relaycommon "github.com/QuantumNous/new-api/relay/common"
"github.com/QuantumNous/new-api/setting/ratio_setting"
"github.com/QuantumNous/new-api/types"
"github.com/gin-gonic/gin"
)
// LogTaskConsumption 记录任务消费日志和统计信息(仅记录,不涉及实际扣费)。
// 实际扣费已由 BillingSession(PreConsumeBilling + SettleBilling)完成。
func LogTaskConsumption(c *gin.Context, info *relaycommon.RelayInfo) {
tokenName := c.GetString("token_name")
logContent := fmt.Sprintf("操作 %s", info.Action)
// 支持任务仅按次计费
if common.StringsContains(constant.TaskPricePatches, info.OriginModelName) {
logContent = fmt.Sprintf("%s,按次计费", logContent)
} else {
if otherRatios := info.PriceData.OtherRatios(); len(otherRatios) > 0 {
var contents []string
for key, ra := range otherRatios {
if 1.0 != ra {
contents = append(contents, fmt.Sprintf("%s: %.2f", key, ra))
}
}
if len(contents) > 0 {
logContent = fmt.Sprintf("%s, 计算参数:%s", logContent, strings.Join(contents, ", "))
}
}
}
other := make(map[string]interface{})
other["is_task"] = true
other["request_path"] = c.Request.URL.Path
other["model_price"] = info.PriceData.ModelPrice
if info.PriceData.ModelRatio > 0 {
other["model_ratio"] = info.PriceData.ModelRatio
}
other["group_ratio"] = info.PriceData.GroupRatioInfo.GroupRatio
if info.PriceData.GroupRatioInfo.HasSpecialRatio {
other["user_group_ratio"] = info.PriceData.GroupRatioInfo.GroupSpecialRatio
}
if info.IsModelMapped {
other["is_model_mapped"] = true
other["upstream_model_name"] = info.UpstreamModelName
}
attachQuotaSaturation(c, info, other)
model.RecordConsumeLog(c, info.UserId, model.RecordConsumeLogParams{
ChannelId: info.ChannelId,
ModelName: info.OriginModelName,
TokenName: tokenName,
Quota: info.PriceData.Quota,
Content: logContent,
TokenId: info.TokenId,
Group: info.UsingGroup,
Other: other,
})
model.UpdateUserUsedQuotaAndRequestCount(info.UserId, info.PriceData.Quota)
model.UpdateChannelUsedQuota(info.ChannelId, info.PriceData.Quota)
}
// ---------------------------------------------------------------------------
// 异步任务计费辅助函数
// ---------------------------------------------------------------------------
// resolveTokenKey 通过 TokenId 运行时获取令牌 Key(用于 Redis 缓存操作)。
// 如果令牌已被删除或查询失败,返回空字符串。
func resolveTokenKey(ctx context.Context, tokenId int, taskID string) string {
token, err := model.GetTokenById(tokenId)
if err != nil {
logger.LogWarn(ctx, fmt.Sprintf("获取令牌 key 失败 (tokenId=%d, task=%s): %s", tokenId, taskID, err.Error()))
return ""
}
return token.Key
}
// taskIsSubscription 判断任务是否通过订阅计费。
func taskIsSubscription(task *model.Task) bool {
return task.PrivateData.BillingSource == BillingSourceSubscription && task.PrivateData.SubscriptionId > 0
}
// taskAdjustFunding 调整任务的资金来源(钱包或订阅),delta > 0 表示扣费,delta < 0 表示退还。
func taskAdjustFunding(task *model.Task, delta int) error {
if taskIsSubscription(task) {
return model.PostConsumeUserSubscriptionDelta(task.PrivateData.SubscriptionId, int64(delta))
}
if delta > 0 {
return model.DecreaseUserQuota(task.UserId, delta, false)
}
return model.IncreaseUserQuota(task.UserId, -delta, false)
}
// taskAdjustTokenQuota 调整任务的令牌额度,delta > 0 表示扣费,delta < 0 表示退还。
// 需要通过 resolveTokenKey 运行时获取 key(不从 PrivateData 中读取)。
func taskAdjustTokenQuota(ctx context.Context, task *model.Task, delta int) {
if task.PrivateData.TokenId <= 0 || delta == 0 {
return
}
tokenKey := resolveTokenKey(ctx, task.PrivateData.TokenId, task.TaskID)
if tokenKey == "" {
return
}
var err error
if delta > 0 {
err = model.DecreaseTokenQuota(task.PrivateData.TokenId, tokenKey, delta)
} else {
err = model.IncreaseTokenQuota(task.PrivateData.TokenId, tokenKey, -delta)
}
if err != nil {
logger.LogWarn(ctx, fmt.Sprintf("调整令牌额度失败 (delta=%d, task=%s): %s", delta, task.TaskID, err.Error()))
}
}
// taskBillingOther 从 task 的 BillingContext 构建日志 Other 字段。
func taskBillingOther(task *model.Task) map[string]interface{} {
other := make(map[string]interface{})
if bc := task.PrivateData.BillingContext; bc != nil {
other["model_price"] = bc.ModelPrice
if bc.ModelRatio > 0 {
other["model_ratio"] = bc.ModelRatio
}
other["group_ratio"] = bc.GroupRatio
if priceData := taskBillingContextPriceData(bc); priceData != nil {
for k, v := range priceData.OtherRatios() {
other[k] = v
}
}
}
props := task.Properties
if props.UpstreamModelName != "" && props.UpstreamModelName != props.OriginModelName {
other["is_model_mapped"] = true
other["upstream_model_name"] = props.UpstreamModelName
}
return other
}
func taskBillingContextPriceData(bc *model.TaskBillingContext) *types.PriceData {
if bc == nil || len(bc.OtherRatios) == 0 {
return nil
}
priceData := &types.PriceData{}
if !priceData.ReplaceOtherRatios(bc.OtherRatios) {
return nil
}
return priceData
}
// taskModelName 从 BillingContext 或 Properties 中获取模型名称。
func taskModelName(task *model.Task) string {
if bc := task.PrivateData.BillingContext; bc != nil && bc.OriginModelName != "" {
return bc.OriginModelName
}
return task.Properties.OriginModelName
}
// RefundTaskQuota 统一的任务失败退款逻辑。
// 当异步任务失败时,将预扣的 quota 退还给用户(支持钱包和订阅),并退还令牌额度。
// 返回资金来源是否已成功退还;失败时保留 quota,供显式重试或人工对账。
func RefundTaskQuota(ctx context.Context, task *model.Task, reason string) bool {
quota := task.Quota
if quota == 0 {
return true
}
// 1. 退还资金来源(钱包或订阅)
if err := taskAdjustFunding(task, -quota); err != nil {
logger.LogWarn(ctx, fmt.Sprintf("退还资金来源失败 task %s: %s", task.TaskID, err.Error()))
return false
}
// 2. 退还令牌额度
taskAdjustTokenQuota(ctx, task, -quota)
// 3. 记录日志
other := taskBillingOther(task)
other["task_id"] = task.TaskID
other["reason"] = reason
model.RecordTaskBillingLog(model.RecordTaskBillingLogParams{
UserId: task.UserId,
LogType: model.LogTypeRefund,
Content: "",
ChannelId: task.ChannelId,
ModelName: taskModelName(task),
Quota: quota,
TokenId: task.PrivateData.TokenId,
Group: task.Group,
Other: other,
})
// 4. 资金退款完成后再清除持久化标记。
// 回写失败必须显式告警,避免漏掉潜在的重复退款风险。
task.Quota = 0
if err := task.UpdateQuota(); err != nil {
logger.LogError(ctx, fmt.Sprintf("退款成功但清除 task quota 失败 task %s: %s", task.TaskID, err.Error()))
}
return true
}
// RecalculateTaskQuota 通用的异步差额结算。
// actualQuota 是任务完成后的实际应扣额度,与预扣额度 (task.Quota) 做差额结算。
// reason 用于日志记录(例如 "token重算" 或 "adaptor调整")。
// clamps 可选:若计算 actualQuota 时发生额度饱和,将其记入日志 admin_info(仅管理员可见)。
func RecalculateTaskQuota(ctx context.Context, task *model.Task, actualQuota int, reason string, clamps ...*common.QuotaClamp) {
if actualQuota <= 0 {
return
}
preConsumedQuota := task.Quota
quotaDelta := actualQuota - preConsumedQuota
if quotaDelta == 0 {
logger.LogInfo(ctx, fmt.Sprintf("任务 %s 预扣费准确(%s,%s)",
task.TaskID, logger.LogQuota(actualQuota), reason))
return
}
logger.LogInfo(ctx, fmt.Sprintf("任务 %s 差额结算:delta=%s(实际:%s,预扣:%s,%s)",
task.TaskID,
logger.LogQuota(quotaDelta),
logger.LogQuota(actualQuota),
logger.LogQuota(preConsumedQuota),
reason,
))
// 调整资金来源
if err := taskAdjustFunding(task, quotaDelta); err != nil {
logger.LogError(ctx, fmt.Sprintf("差额结算资金调整失败 task %s: %s", task.TaskID, err.Error()))
return
}
// 调整令牌额度
taskAdjustTokenQuota(ctx, task, quotaDelta)
task.Quota = actualQuota
if err := task.UpdateQuota(); err != nil {
logger.LogError(ctx, fmt.Sprintf("差额结算回写 quota 失败 task %s: %s", task.TaskID, err.Error()))
}
var logType int
var logQuota int
if quotaDelta > 0 {
logType = model.LogTypeConsume
logQuota = quotaDelta
model.UpdateUserUsedQuotaAndRequestCount(task.UserId, quotaDelta)
model.UpdateChannelUsedQuota(task.ChannelId, quotaDelta)
} else {
logType = model.LogTypeRefund
logQuota = -quotaDelta
}
other := taskBillingOther(task)
other["task_id"] = task.TaskID
other["pre_consumed_quota"] = preConsumedQuota
other["actual_quota"] = actualQuota
for _, clamp := range clamps {
attachQuotaSaturationToOther(other, clamp)
}
model.RecordTaskBillingLog(model.RecordTaskBillingLogParams{
UserId: task.UserId,
LogType: logType,
Content: reason,
ChannelId: task.ChannelId,
ModelName: taskModelName(task),
Quota: logQuota,
TokenId: task.PrivateData.TokenId,
Group: task.Group,
Other: other,
NodeName: task.PrivateData.NodeName,
})
}
// RecalculateTaskQuotaByTokens 根据实际 token 消耗重新计费(异步差额结算)。
// 当任务成功且返回了 totalTokens 时,根据模型倍率和分组倍率重新计算实际扣费额度,
// 与预扣费的差额进行补扣或退还。支持钱包和订阅计费来源。
func RecalculateTaskQuotaByTokens(ctx context.Context, task *model.Task, totalTokens int) {
if totalTokens <= 0 {
return
}
modelName := taskModelName(task)
// 获取模型价格和倍率
modelRatio, hasRatioSetting, _ := ratio_setting.GetModelRatio(modelName)
// 只有配置了倍率(非固定价格)时才按 token 重新计费
if !hasRatioSetting || modelRatio <= 0 {
return
}
// 获取用户和组的倍率信息
group := task.Group
if group == "" {
user, err := model.GetUserById(task.UserId, false)
if err == nil {
group = user.Group
}
}
if group == "" {
return
}
groupRatio := ratio_setting.GetGroupRatio(group)
userGroupRatio, hasUserGroupRatio := ratio_setting.GetGroupGroupRatio(group, group)
var finalGroupRatio float64
if hasUserGroupRatio {
finalGroupRatio = userGroupRatio
} else {
finalGroupRatio = groupRatio
}
// 计算 OtherRatios 乘积(视频折扣、时长等)
otherMultiplier := 1.0
if priceData := taskBillingContextPriceData(task.PrivateData.BillingContext); priceData != nil {
otherMultiplier = priceData.OtherRatioMultiplier()
}
// 计算实际应扣费额度: totalTokens * modelRatio * groupRatio * otherMultiplier(饱和转换,防止溢出成负数)
actualQuota, clamp := common.QuotaFromFloatChecked(float64(totalTokens) * modelRatio * finalGroupRatio * otherMultiplier)
reason := fmt.Sprintf("token重算:tokens=%d, modelRatio=%.2f, groupRatio=%.2f, otherMultiplier=%.4f", totalTokens, modelRatio, finalGroupRatio, otherMultiplier)
RecalculateTaskQuota(ctx, task, actualQuota, reason, clamp)
}