diff --git a/common/constants.go b/common/constants.go index 274c514f9146..547cce773c0e 100644 --- a/common/constants.go +++ b/common/constants.go @@ -50,6 +50,7 @@ var WeChatAuthEnabled = false var TelegramOAuthEnabled = false var TurnstileCheckEnabled = false var RegisterEnabled = true +var SingleDeviceLoginEnabled = true var EmailDomainRestrictionEnabled = false // 是否启用邮箱域名限制 var EmailAliasRestrictionEnabled = false // 是否启用邮箱别名限制 @@ -173,7 +174,7 @@ var ( CriticalRateLimitEnable bool CriticalRateLimitNum = 20 - CriticalRateLimitDuration int64 = 20 * 60 + CriticalRateLimitDuration int64 = 60 UploadRateLimitNum = 10 UploadRateLimitDuration int64 = 60 diff --git a/common/gin_log_writer_factory.go b/common/gin_log_writer_factory.go new file mode 100644 index 000000000000..559b592f95ae --- /dev/null +++ b/common/gin_log_writer_factory.go @@ -0,0 +1,14 @@ +package common + +import ( + "io" + "os" +) + +// GinLogWriterFactory, when set by the host application (e.g. LynxtonAPI), is used by +// logger.SetupLogger to build gin.DefaultWriter and gin.DefaultErrorWriter instead of the +// default io.MultiWriter(stdout, file) / io.MultiWriter(stderr, file). +// +// Implementations should write human-readable legacy lines to logFile as needed and emit +// structured logs separately (e.g. JSON on stdout) so container collectors see one format. +var GinLogWriterFactory func(stdout io.Writer, logFile *os.File) (out io.Writer, errOut io.Writer) diff --git a/common/init.go b/common/init.go index 35b4c6be17ee..107301832678 100644 --- a/common/init.go +++ b/common/init.go @@ -120,7 +120,7 @@ func InitEnv() { CriticalRateLimitEnable = GetEnvOrDefaultBool("CRITICAL_RATE_LIMIT_ENABLE", true) CriticalRateLimitNum = GetEnvOrDefault("CRITICAL_RATE_LIMIT", 20) - CriticalRateLimitDuration = int64(GetEnvOrDefault("CRITICAL_RATE_LIMIT_DURATION", 20*60)) + CriticalRateLimitDuration = int64(GetEnvOrDefault("CRITICAL_RATE_LIMIT_DURATION", 60)) SearchRateLimitEnable = GetEnvOrDefaultBool("SEARCH_RATE_LIMIT_ENABLE", true) SearchRateLimitNum = GetEnvOrDefault("SEARCH_RATE_LIMIT", 10) diff --git a/common/ops_hook.go b/common/ops_hook.go new file mode 100644 index 000000000000..06f13b80bff5 --- /dev/null +++ b/common/ops_hook.go @@ -0,0 +1,7 @@ +package common + +import "context" + +// EmitAsyncBillingOpsLog is an optional hook set by Lynxton (after applog.Init) to emit structured JSON +// for async task billing (worker has no *gin.Context). When nil, new-api stays silent on this channel. +var EmitAsyncBillingOpsLog func(ctx context.Context, msg string, kv map[string]any) diff --git a/common/verification.go b/common/verification.go index 41fd3c943e7e..8dac2c2b270b 100644 --- a/common/verification.go +++ b/common/verification.go @@ -1,10 +1,13 @@ package common import ( + "context" + "errors" "strings" "sync" "time" + "github.com/go-redis/redis/v8" "github.com/google/uuid" ) @@ -23,6 +26,8 @@ var verificationMap map[string]verificationValue var verificationMapMaxSize = 10 var VerificationValidMinutes = 10 +const verificationRedisKeyPrefix = "verification:" + func GenerateVerificationCode(length int) string { code := uuid.New().String() code = strings.Replace(code, "-", "", -1) @@ -33,6 +38,15 @@ func GenerateVerificationCode(length int) string { } func RegisterVerificationCodeWithKey(key string, code string, purpose string) { + key = normalizeVerificationKey(key) + if RedisEnabled && RDB != nil { + err := RDB.Set(context.Background(), verificationRedisKey(key, purpose), code, verificationTTL()).Err() + if err == nil { + return + } + SysLog("failed to save verification code to Redis, falling back to memory: " + err.Error()) + } + verificationMutex.Lock() defer verificationMutex.Unlock() verificationMap[purpose+key] = verificationValue{ @@ -45,6 +59,19 @@ func RegisterVerificationCodeWithKey(key string, code string, purpose string) { } func VerifyCodeWithKey(key string, code string, purpose string) bool { + key = normalizeVerificationKey(key) + code = strings.TrimSpace(code) + if RedisEnabled && RDB != nil { + value, err := RDB.Get(context.Background(), verificationRedisKey(key, purpose)).Result() + if err == nil { + return code == value + } + if errors.Is(err, redis.Nil) { + return false + } + SysLog("failed to read verification code from Redis, falling back to memory: " + err.Error()) + } + verificationMutex.Lock() defer verificationMutex.Unlock() value, okay := verificationMap[purpose+key] @@ -56,6 +83,13 @@ func VerifyCodeWithKey(key string, code string, purpose string) bool { } func DeleteKey(key string, purpose string) { + key = normalizeVerificationKey(key) + if RedisEnabled && RDB != nil { + if err := RDB.Del(context.Background(), verificationRedisKey(key, purpose)).Err(); err != nil { + SysLog("failed to delete verification code from Redis: " + err.Error()) + } + } + verificationMutex.Lock() defer verificationMutex.Unlock() delete(verificationMap, purpose+key) @@ -71,6 +105,18 @@ func removeExpiredPairs() { } } +func verificationRedisKey(key string, purpose string) string { + return verificationRedisKeyPrefix + purpose + ":" + key +} + +func verificationTTL() time.Duration { + return time.Duration(VerificationValidMinutes) * time.Minute +} + +func normalizeVerificationKey(key string) string { + return strings.ToLower(strings.TrimSpace(key)) +} + func init() { verificationMutex.Lock() defer verificationMutex.Unlock() diff --git a/constant/api_type.go b/constant/api_type.go index 536ebd2c7198..fc82752ba65c 100644 --- a/constant/api_type.go +++ b/constant/api_type.go @@ -12,6 +12,7 @@ const ( APITypeTencent APITypeGemini APITypeZhipuV4 + APITypeOllama APITypePerplexity APITypeAws diff --git a/constant/context_key.go b/constant/context_key.go index c28ad202514b..4c7d3137c00a 100644 --- a/constant/context_key.go +++ b/constant/context_key.go @@ -66,4 +66,15 @@ const ( // ContextKeyLanguage stores the user's language preference for i18n ContextKeyLanguage ContextKey = "language" ContextKeyIsStream ContextKey = "is_stream" + + /* enterprise discount related keys */ + ContextKeyOrgID ContextKey = "org_id" + ContextKeyOrgDiscountRate ContextKey = "org_discount_rate" + + // Ops billing snapshot (set by new-api/service before RecordConsumeLog; read by Lynxton relay_request_summary). + ContextKeyOpsBillingPromptTokens ContextKey = "ops_billing_prompt_tokens" + ContextKeyOpsBillingCompletionTokens ContextKey = "ops_billing_completion_tokens" + ContextKeyOpsBillingTotalTokens ContextKey = "ops_billing_total_tokens" + ContextKeyOpsBillingConsumeQuota ContextKey = "ops_billing_consume_quota" + ContextKeyOpsBillingSnapshotFrom ContextKey = "ops_billing_snapshot_from" ) diff --git a/controller/channel.go b/controller/channel.go index b0dd22861507..b3a87ddae2af 100644 --- a/controller/channel.go +++ b/controller/channel.go @@ -68,6 +68,14 @@ func clearChannelInfo(channel *model.Channel) { } } +func normalizeChannelBaseURL(channel *model.Channel) { + if channel == nil || channel.BaseURL == nil { + return + } + normalized := strings.TrimRight(strings.TrimSpace(*channel.BaseURL), "/") + channel.BaseURL = &normalized +} + func GetAllChannels(c *gin.Context) { pageInfo := common.GetPageQuery(c) channelData := make([]*model.Channel, 0) @@ -403,7 +411,7 @@ func GetChannelKey(c *gin.Context) { } // 记录操作日志 - model.RecordLog(userId, model.LogTypeSystem, fmt.Sprintf("查看渠道密钥信息 (渠道ID: %d)", channelId)) + model.RecordLog(userId, model.LogTypeSystem, fmt.Sprintf("查看渠道密钥信息 (渠道ID: %d)", channelId), 0) // 返回渠道密钥 c.JSON(http.StatusOK, gin.H{ @@ -570,6 +578,7 @@ func AddChannel(c *gin.Context) { common.ApiError(c, err) return } + normalizeChannelBaseURL(addChannelRequest.Channel) // 使用统一的校验函数 if err := validateChannel(addChannelRequest.Channel, true); err != nil { @@ -846,6 +855,7 @@ func UpdateChannel(c *gin.Context) { common.ApiError(c, err) return } + normalizeChannelBaseURL(&channel.Channel) // 使用统一的校验函数 if err := validateChannel(&channel.Channel, false); err != nil { diff --git a/controller/checkin.go b/controller/checkin.go index cc8bf4f96d7c..5f9d0937b371 100644 --- a/controller/checkin.go +++ b/controller/checkin.go @@ -61,7 +61,7 @@ func DoCheckin(c *gin.Context) { }) return } - model.RecordLog(userId, model.LogTypeSystem, fmt.Sprintf("用户签到，获得额度 %s", logger.LogQuota(checkin.QuotaAwarded))) + model.RecordLog(userId, model.LogTypeSystem, fmt.Sprintf("用户签到，获得额度 %s", logger.LogQuota(checkin.QuotaAwarded)), checkin.QuotaAwarded) c.JSON(http.StatusOK, gin.H{ "success": true, "message": "签到成功", diff --git a/controller/model.go b/controller/model.go index aa6c6e2b9db7..8de0bab87e71 100644 --- a/controller/model.go +++ b/controller/model.go @@ -141,6 +141,7 @@ func ListModels(c *gin.Context, modelType int) { } if oaiModel, ok := openAIModelsMap[allowModel]; ok { oaiModel.SupportedEndpointTypes = model.GetModelSupportEndpointTypes(allowModel) + oaiModel.Modalities = ratio_setting.GetModelModalities(allowModel) userOpenAiModels = append(userOpenAiModels, oaiModel) } else { userOpenAiModels = append(userOpenAiModels, dto.OpenAIModels{ @@ -149,6 +150,7 @@ func ListModels(c *gin.Context, modelType int) { Created: 1626777600, OwnedBy: "custom", SupportedEndpointTypes: model.GetModelSupportEndpointTypes(allowModel), + Modalities: ratio_setting.GetModelModalities(allowModel), }) } } @@ -189,6 +191,7 @@ func ListModels(c *gin.Context, modelType int) { } if oaiModel, ok := openAIModelsMap[modelName]; ok { oaiModel.SupportedEndpointTypes = model.GetModelSupportEndpointTypes(modelName) + oaiModel.Modalities = ratio_setting.GetModelModalities(modelName) userOpenAiModels = append(userOpenAiModels, oaiModel) } else { userOpenAiModels = append(userOpenAiModels, dto.OpenAIModels{ @@ -197,6 +200,7 @@ func ListModels(c *gin.Context, modelType int) { Created: 1626777600, OwnedBy: "custom", SupportedEndpointTypes: model.GetModelSupportEndpointTypes(modelName), + Modalities: ratio_setting.GetModelModalities(modelName), }) } } @@ -273,6 +277,8 @@ func RetrieveModel(c *gin.Context, modelType int) { Type: "model", }) default: + aiModel.SupportedEndpointTypes = model.GetModelSupportEndpointTypes(modelId) + aiModel.Modalities = ratio_setting.GetModelModalities(modelId) c.JSON(200, aiModel) } } else { @@ -286,4 +292,4 @@ func RetrieveModel(c *gin.Context, modelType int) { "error": openAIError, }) } -} +} \ No newline at end of file diff --git a/controller/option.go b/controller/option.go index 5a7c9f418972..5e09678d0b22 100644 --- a/controller/option.go +++ b/controller/option.go @@ -306,6 +306,30 @@ func UpdateOption(c *gin.Context) { }) return } + case "ModelDisplayName": + if err = ratio_setting.UpdateModelDisplayNameByJSONString(option.Value.(string)); err != nil { + c.JSON(http.StatusOK, gin.H{ + "success": false, + "message": "模型显示名称设置失败: " + err.Error(), + }) + return + } + case "ModelModalities": + if err = ratio_setting.UpdateModelModalitiesByJSONString(option.Value.(string)); err != nil { + c.JSON(http.StatusOK, gin.H{ + "success": false, + "message": "模型输出类型设置失败: " + err.Error(), + }) + return + } + case "ModelImg": + if err = ratio_setting.UpdateModelImgByJSONString(option.Value.(string)); err != nil { + c.JSON(http.StatusOK, gin.H{ + "success": false, + "message": "模型图片设置失败: " + err.Error(), + }) + return + } } err = model.UpdateOption(option.Key, option.Value.(string)) if err != nil { diff --git a/controller/redemption.go b/controller/redemption.go index 76c35bc32bcd..1f96420b062d 100644 --- a/controller/redemption.go +++ b/controller/redemption.go @@ -1,6 +1,7 @@ package controller import ( + "fmt" "net/http" "strconv" "unicode/utf8" @@ -84,9 +85,13 @@ func AddRedemption(c *gin.Context) { var keys []string for i := 0; i < redemption.Count; i++ { key := common.GetUUID() + name := redemption.Name + if redemption.Count > 1 { + name = fmt.Sprintf("%s_%d", redemption.Name, i+1) + } cleanRedemption := model.Redemption{ UserId: c.GetInt("id"), - Name: redemption.Name, + Name: name, Key: key, CreatedTime: common.GetTimestamp(), Quota: redemption.Quota, diff --git a/controller/relay.go b/controller/relay.go index c97ab45b4ac4..7f8652e908ba 100644 --- a/controller/relay.go +++ b/controller/relay.go @@ -568,6 +568,8 @@ func RelayTask(c *gin.Context) { if settleErr := service.SettleBilling(c, relayInfo, result.Quota); settleErr != nil { common.SysError("settle task billing error: " + settleErr.Error()) } + recordedQuota := service.NormalizeRecordedQuota(c, relayInfo, result.Quota) + relayInfo.PriceData.Quota = recordedQuota service.LogTaskConsumption(c, relayInfo) task := model.InitTask(result.Platform, relayInfo) @@ -583,7 +585,7 @@ func RelayTask(c *gin.Context) { OriginModelName: relayInfo.OriginModelName, PerCallBilling: common.StringsContains(constant.TaskPricePatches, relayInfo.OriginModelName) || relayInfo.PriceData.UsePrice, } - task.Quota = result.Quota + task.Quota = recordedQuota task.Data = result.TaskData task.Action = relayInfo.Action if insertErr := task.Insert(); insertErr != nil { diff --git a/controller/secure_verification.go b/controller/secure_verification.go index b229a66b1c26..1a3144505999 100644 --- a/controller/secure_verification.go +++ b/controller/secure_verification.go @@ -127,7 +127,7 @@ func UniversalVerify(c *gin.Context) { } // 记录日志 - model.RecordLog(userId, model.LogTypeSystem, fmt.Sprintf("通用安全验证成功 (验证方式: %s)", verifyMethod)) + model.RecordLog(userId, model.LogTypeSystem, fmt.Sprintf("通用安全验证成功 (验证方式: %s)", verifyMethod), 0) c.JSON(http.StatusOK, gin.H{ "success": true, diff --git a/controller/topup.go b/controller/topup.go index 86d361a349cb..36c548d9d2cc 100644 --- a/controller/topup.go +++ b/controller/topup.go @@ -189,6 +189,9 @@ func getMinTopup() int64 { } func RequestEpay(c *gin.Context) { + if blockEnterpriseTopupForNonOwner(c) { + return + } var req EpayRequest err := c.ShouldBindJSON(&req) if err != nil { diff --git a/controller/topup_creem.go b/controller/topup_creem.go index 139dd43fbed1..d2abb4638cff 100644 --- a/controller/topup_creem.go +++ b/controller/topup_creem.go @@ -141,6 +141,9 @@ func (*CreemAdaptor) RequestPay(c *gin.Context, req *CreemPayRequest) { } func RequestCreemPay(c *gin.Context) { + if blockEnterpriseTopupForNonOwner(c) { + return + } var req CreemPayRequest // 读取body内容用于打印，同时保留原始数据供后续使用 diff --git a/controller/topup_enterprise_guard.go b/controller/topup_enterprise_guard.go new file mode 100644 index 000000000000..9f08b0abe8f2 --- /dev/null +++ b/controller/topup_enterprise_guard.go @@ -0,0 +1,48 @@ +package controller + +import ( + "errors" + "net/http" + + "github.com/QuantumNous/new-api/common" + "github.com/QuantumNous/new-api/model" + "github.com/gin-gonic/gin" + "gorm.io/gorm" +) + +func blockEnterpriseTopupForNonOwner(c *gin.Context) bool { + userID := c.GetInt("id") + var ownerOrg struct { + ID uint + } + err := model.DB.Table("lc_organizations"). + Select("id"). + Where("owner_id = ? AND status = 1", userID). + First(&ownerOrg).Error + if err == nil { + return false + } + if err != nil && !errors.Is(err, gorm.ErrRecordNotFound) { + common.ApiError(c, err) + return true + } + + var count int64 + err = model.DB.Table("lc_org_members"). + Joins("JOIN lc_organizations ON lc_organizations.id = lc_org_members.org_id AND lc_organizations.status = 1"). + Where("lc_org_members.user_id = ? AND lc_org_members.status = 1", userID). + Count(&count).Error + if err != nil { + common.ApiError(c, err) + return true + } + if count > 0 { + c.JSON(http.StatusOK, gin.H{ + "success": false, + "message": "企业充值仅限企业拥有者操作", + "data": "企业充值仅限企业拥有者操作", + }) + return true + } + return false +} diff --git a/controller/topup_stripe.go b/controller/topup_stripe.go index 23ddb3b90e3e..fa2c41aa7d58 100644 --- a/controller/topup_stripe.go +++ b/controller/topup_stripe.go @@ -135,6 +135,9 @@ func RequestStripeAmount(c *gin.Context) { } func RequestStripePay(c *gin.Context) { + if blockEnterpriseTopupForNonOwner(c) { + return + } var req StripePayRequest err := c.ShouldBindJSON(&req) if err != nil { diff --git a/controller/topup_waffo.go b/controller/topup_waffo.go index c00680628196..843a46d85b17 100644 --- a/controller/topup_waffo.go +++ b/controller/topup_waffo.go @@ -131,6 +131,9 @@ func RequestWaffoAmount(c *gin.Context) { // RequestWaffoPay 创建 Waffo 支付订单 func RequestWaffoPay(c *gin.Context) { + if blockEnterpriseTopupForNonOwner(c) { + return + } if !setting.WaffoEnabled { c.JSON(http.StatusOK, gin.H{"message": "error", "data": "Waffo 支付未启用"}) return diff --git a/controller/topup_waffo_pancake.go b/controller/topup_waffo_pancake.go index 81515a56ed35..a0132e31f8b3 100644 --- a/controller/topup_waffo_pancake.go +++ b/controller/topup_waffo_pancake.go @@ -111,6 +111,9 @@ func getWaffoPancakeReturnURL() string { } func RequestWaffoPancakePay(c *gin.Context) { + if blockEnterpriseTopupForNonOwner(c) { + return + } if !setting.WaffoPancakeEnabled { c.JSON(http.StatusOK, gin.H{"message": "error", "data": "Waffo Pancake 支付未启用"}) return diff --git a/controller/twofa.go b/controller/twofa.go index 123c74e2cf44..8585babc608c 100644 --- a/controller/twofa.go +++ b/controller/twofa.go @@ -120,7 +120,7 @@ func Setup2FA(c *gin.Context) { } // 记录操作日志 - model.RecordLog(userId, model.LogTypeSystem, "开始设置两步验证") + model.RecordLog(userId, model.LogTypeSystem, "开始设置两步验证", 0) c.JSON(http.StatusOK, gin.H{ "success": true, @@ -192,7 +192,7 @@ func Enable2FA(c *gin.Context) { } // 记录操作日志 - model.RecordLog(userId, model.LogTypeSystem, "成功启用两步验证") + model.RecordLog(userId, model.LogTypeSystem, "成功启用两步验证", 0) c.JSON(http.StatusOK, gin.H{ "success": true, @@ -264,7 +264,7 @@ func Disable2FA(c *gin.Context) { } // 记录操作日志 - model.RecordLog(userId, model.LogTypeSystem, "禁用两步验证") + model.RecordLog(userId, model.LogTypeSystem, "禁用两步验证", 0) c.JSON(http.StatusOK, gin.H{ "success": true, @@ -383,7 +383,7 @@ func RegenerateBackupCodes(c *gin.Context) { } // 记录操作日志 - model.RecordLog(userId, model.LogTypeSystem, "重新生成两步验证备用码") + model.RecordLog(userId, model.LogTypeSystem, "重新生成两步验证备用码", 0) c.JSON(http.StatusOK, gin.H{ "success": true, @@ -549,7 +549,7 @@ func AdminDisable2FA(c *gin.Context) { "admin_username": adminName, } model.RecordLogWithAdminInfo(userId, model.LogTypeManage, - "管理员强制禁用了用户的两步验证", adminInfo) + "管理员强制禁用了用户的两步验证", 0, adminInfo) c.JSON(http.StatusOK, gin.H{ "success": true, diff --git a/controller/user.go b/controller/user.go index d6becdd8f0e9..4fdc07905271 100644 --- a/controller/user.go +++ b/controller/user.go @@ -91,13 +91,20 @@ func Login(c *gin.Context) { // setup session & cookies and then return user info func setupLogin(user *model.User, c *gin.Context) { + // 每次登录都轮换会话令牌，确保同账号仅保留最新设备会话 + sessionToken, err := model.IssueUserSessionToken(user.Id) + if err != nil { + common.ApiErrorI18n(c, i18n.MsgUserSessionSaveFailed) + return + } session := sessions.Default(c) session.Set("id", user.Id) session.Set("username", user.Username) session.Set("role", user.Role) session.Set("status", user.Status) session.Set("group", user.Group) - err := session.Save() + session.Set("session_token", sessionToken) + err = session.Save() if err != nil { common.ApiErrorI18n(c, i18n.MsgUserSessionSaveFailed) return @@ -152,6 +159,12 @@ func Register(c *gin.Context) { common.ApiErrorI18n(c, i18n.MsgUserInputInvalid, map[string]any{"Error": err.Error()}) return } + if setting.CheckSensitiveEnabled { + if ok, _ := service.CheckSensitiveText(user.Username); ok { + common.ApiErrorMsg(c, "用户名包含敏感词，请修改后重试") + return + } + } if common.EmailVerificationEnabled { if user.Email == "" || user.VerificationCode == "" { common.ApiErrorI18n(c, i18n.MsgUserEmailVerificationRequired) @@ -395,6 +408,7 @@ func GetSelf(c *gin.Context) { "role": user.Role, "status": user.Status, "email": user.Email, + "phone": user.Phone, "github_id": user.GitHubId, "discord_id": user.DiscordId, "oidc_id": user.OidcId, @@ -577,6 +591,13 @@ func UpdateUser(c *gin.Context) { common.ApiError(c, err) return } + // 管理员修改密码后，强制该账号所有旧设备会话失效 + if updatePassword { + if _, err := model.RotateUserSessionToken(updatedUser.Id); err != nil { + common.ApiError(c, err) + return + } + } c.JSON(http.StatusOK, gin.H{ "success": true, "message": "", @@ -614,7 +635,7 @@ func AdminClearUserBinding(c *gin.Context) { return } - model.RecordLog(user.Id, model.LogTypeManage, fmt.Sprintf("admin cleared %s binding for user %s", bindingType, user.Username)) + model.RecordLog(user.Id, model.LogTypeManage, fmt.Sprintf("admin cleared %s binding for user %s", bindingType, user.Username), 0) c.JSON(http.StatusOK, gin.H{ "success": true, @@ -726,6 +747,20 @@ func UpdateSelf(c *gin.Context) { common.ApiError(c, err) return } + // 用户自行修改密码后，轮换会话令牌并更新当前设备会话 + if updatePassword { + sessionToken, err := model.RotateUserSessionToken(cleanUser.Id) + if err != nil { + common.ApiError(c, err) + return + } + session := sessions.Default(c) + session.Set("session_token", sessionToken) + if err := session.Save(); err != nil { + common.ApiErrorI18n(c, i18n.MsgUserSessionSaveFailed) + return + } + } c.JSON(http.StatusOK, gin.H{ "success": true, @@ -934,7 +969,7 @@ func ManageUser(c *gin.Context) { return } model.RecordLogWithAdminInfo(user.Id, model.LogTypeManage, - fmt.Sprintf("管理员增加用户额度 %s", logger.LogQuota(req.Value)), adminInfo) + fmt.Sprintf("管理员增加用户额度 %s", logger.LogQuota(req.Value)), req.Value, adminInfo) case "subtract": if req.Value <= 0 { common.ApiErrorI18n(c, i18n.MsgUserQuotaChangeZero) @@ -945,7 +980,7 @@ func ManageUser(c *gin.Context) { return } model.RecordLogWithAdminInfo(user.Id, model.LogTypeManage, - fmt.Sprintf("管理员减少用户额度 %s", logger.LogQuota(req.Value)), adminInfo) + fmt.Sprintf("管理员减少用户额度 %s", logger.LogQuota(req.Value)), req.Value, adminInfo) case "override": oldQuota := user.Quota if err := model.DB.Model(&model.User{}).Where("id = ?", user.Id).Update("quota", req.Value).Error; err != nil { @@ -953,7 +988,7 @@ func ManageUser(c *gin.Context) { return } model.RecordLogWithAdminInfo(user.Id, model.LogTypeManage, - fmt.Sprintf("管理员覆盖用户额度从 %s 为 %s", logger.LogQuota(oldQuota), logger.LogQuota(req.Value)), adminInfo) + fmt.Sprintf("管理员覆盖用户额度从 %s 为 %s", logger.LogQuota(oldQuota), logger.LogQuota(req.Value)), req.Value, adminInfo) default: common.ApiErrorI18n(c, i18n.MsgInvalidParams) return @@ -1080,6 +1115,9 @@ func getTopUpLock(userID int) *topUpTryLock { } func TopUp(c *gin.Context) { + if blockEnterpriseTopupForNonOwner(c) { + return + } id := c.GetInt("id") lock := getTopUpLock(id) if !lock.TryLock() { diff --git a/dto/pricing.go b/dto/pricing.go index 1ed8dcd31c29..9597e51d1d65 100644 --- a/dto/pricing.go +++ b/dto/pricing.go @@ -9,6 +9,7 @@ type OpenAIModels struct { Created int `json:"created"` OwnedBy string `json:"owned_by"` SupportedEndpointTypes []constant.EndpointType `json:"supported_endpoint_types"` + Modalities string `json:"modalities,omitempty"` } type AnthropicModel struct { @@ -32,4 +33,4 @@ type GeminiModel struct { MaxTemperature interface{} `json:"maxTemperature"` TopP interface{} `json:"topP"` TopK interface{} `json:"topK"` -} +} \ No newline at end of file diff --git a/logger/logger.go b/logger/logger.go index 7b0c82de50d6..3ee3c7344e4f 100644 --- a/logger/logger.go +++ b/logger/logger.go @@ -64,8 +64,12 @@ func SetupLogger() { currentLogPathMu.Unlock() common.LogWriterMu.Lock() - gin.DefaultWriter = io.MultiWriter(os.Stdout, fd) - gin.DefaultErrorWriter = io.MultiWriter(os.Stderr, fd) + if common.GinLogWriterFactory != nil { + gin.DefaultWriter, gin.DefaultErrorWriter = common.GinLogWriterFactory(os.Stdout, fd) + } else { + gin.DefaultWriter = io.MultiWriter(os.Stdout, fd) + gin.DefaultErrorWriter = io.MultiWriter(os.Stderr, fd) + } if oldFile != nil { _ = oldFile.Close() } @@ -124,7 +128,7 @@ func LogQuota(quota int) string { case operation_setting.QuotaDisplayTypeCNY: usd := q / common.QuotaPerUnit cny := usd * operation_setting.USDExchangeRate - return fmt.Sprintf("¥%.6f 额度", cny) + return fmt.Sprintf("¥%.2f 额度", cny) case operation_setting.QuotaDisplayTypeCustom: usd := q / common.QuotaPerUnit rate := operation_setting.GetGeneralSetting().CustomCurrencyExchangeRate @@ -136,11 +140,11 @@ func LogQuota(quota int) string { rate = 1 } v := usd * rate - return fmt.Sprintf("%s%.6f 额度", symbol, v) + return fmt.Sprintf("%s%.2f 额度", symbol, v) case operation_setting.QuotaDisplayTypeTokens: return fmt.Sprintf("%d 点额度", quota) default: // USD - return fmt.Sprintf("＄%.6f 额度", q/common.QuotaPerUnit) + return fmt.Sprintf("¥%.2f 额度", q/common.QuotaPerUnit) } } @@ -150,7 +154,7 @@ func FormatQuota(quota int) string { case operation_setting.QuotaDisplayTypeCNY: usd := q / common.QuotaPerUnit cny := usd * operation_setting.USDExchangeRate - return fmt.Sprintf("¥%.6f", cny) + return fmt.Sprintf("¥%.2f", cny) case operation_setting.QuotaDisplayTypeCustom: usd := q / common.QuotaPerUnit rate := operation_setting.GetGeneralSetting().CustomCurrencyExchangeRate @@ -162,11 +166,11 @@ func FormatQuota(quota int) string { rate = 1 } v := usd * rate - return fmt.Sprintf("%s%.6f", symbol, v) + return fmt.Sprintf("%s%.2f", symbol, v) case operation_setting.QuotaDisplayTypeTokens: return fmt.Sprintf("%d", quota) default: - return fmt.Sprintf("＄%.6f", q/common.QuotaPerUnit) + return fmt.Sprintf("¥%.2f", q/common.QuotaPerUnit) } } diff --git a/main.go b/main.go index dbbf44a1826b..a4b59293d25f 100644 --- a/main.go +++ b/main.go @@ -172,7 +172,7 @@ func main() { store := cookie.NewStore([]byte(common.SessionSecret)) store.Options(sessions.Options{ Path: "/", - MaxAge: 2592000, // 30 days + MaxAge: 604800, // 7 days HttpOnly: true, Secure: false, SameSite: http.SameSiteStrictMode, diff --git a/middleware/auth.go b/middleware/auth.go index 23d933fbe0c1..e86611c719b2 100644 --- a/middleware/auth.go +++ b/middleware/auth.go @@ -33,6 +33,37 @@ func validUserInfo(username string, role int) bool { return true } +func sessionNumberToInt(v any) (int, bool) { + switch n := v.(type) { + case int: + return n, true + case int8: + return int(n), true + case int16: + return int(n), true + case int32: + return int(n), true + case int64: + return int(n), true + case uint: + return int(n), true + case uint8: + return int(n), true + case uint16: + return int(n), true + case uint32: + return int(n), true + case uint64: + return int(n), true + case float32: + return int(n), true + case float64: + return int(n), true + default: + return 0, false + } +} + func authHelper(c *gin.Context, minRole int) { session := sessions.Default(c) username := session.Get("username") @@ -92,6 +123,42 @@ func authHelper(c *gin.Context, minRole int) { return } } + // Session 登录态下，校验会话令牌是否仍为最新；关闭单设备登录时，多设备共享同一个令牌。 + if !useAccessToken { + sessionToken, _ := session.Get("session_token").(string) + idInt, ok := sessionNumberToInt(id) + if !ok || idInt <= 0 || sessionToken == "" { + session.Clear() + _ = session.Save() + c.JSON(http.StatusUnauthorized, gin.H{ + "success": false, + "message": common.TranslateMessage(c, i18n.MsgAuthNotLoggedIn), + }) + c.Abort() + return + } else { + valid, err := model.ValidateUserSessionToken(idInt, sessionToken) + if err != nil { + common.SysLog("ValidateUserSessionToken database error: " + err.Error()) + c.JSON(http.StatusInternalServerError, gin.H{ + "success": false, + "message": common.TranslateMessage(c, i18n.MsgDatabaseError), + }) + c.Abort() + return + } + if !valid { + session.Clear() + _ = session.Save() + c.JSON(http.StatusUnauthorized, gin.H{ + "success": false, + "message": "登录状态已失效：该账号已在其他设备登录，请重新登录", + }) + c.Abort() + return + } + } + } // get header New-Api-User apiUserIdStr := c.Request.Header.Get("New-Api-User") if apiUserIdStr == "" { @@ -112,7 +179,8 @@ func authHelper(c *gin.Context, minRole int) { return } - if id != apiUserId { + idInt, ok := sessionNumberToInt(id) + if !ok || idInt != apiUserId { c.JSON(http.StatusUnauthorized, gin.H{ "success": false, "message": common.TranslateMessage(c, i18n.MsgAuthUserIdMismatch), @@ -330,6 +398,19 @@ func TokenAuth() func(c *gin.Context) { key = parts[0] } token, err := model.ValidateUserToken(key) + if err != nil && shouldTryTokenQuotaFallback(token) { + fallbackToken, fallbackErr := model.GetNextAvailableTokenForUser(token.UserId, token.Id) + if fallbackErr == nil { + common.SysLog(fmt.Sprintf("TokenAuth switched exhausted token %d to fallback token %d for user %d", token.Id, fallbackToken.Id, token.UserId)) + token = fallbackToken + err = nil + } else if errors.Is(fallbackErr, model.ErrDatabase) { + common.SysLog("TokenAuth GetNextAvailableTokenForUser database error: " + fallbackErr.Error()) + abortWithOpenAiMessage(c, http.StatusInternalServerError, + common.TranslateMessage(c, i18n.MsgDatabaseError)) + return + } + } if token != nil { id := c.GetInt("id") if id == 0 { @@ -406,6 +487,19 @@ func TokenAuth() func(c *gin.Context) { } } +func shouldTryTokenQuotaFallback(token *model.Token) bool { + if token == nil || token.UserId == 0 || token.UnlimitedQuota { + return false + } + if token.Status != common.TokenStatusEnabled && token.Status != common.TokenStatusExhausted { + return false + } + if token.ExpiredTime != -1 && token.ExpiredTime < common.GetTimestamp() { + return false + } + return token.RemainQuota <= 0 +} + func SetupContextForToken(c *gin.Context, token *model.Token, parts ...string) error { if token == nil { return fmt.Errorf("token is nil") diff --git a/middleware/rate-limit.go b/middleware/rate-limit.go index d8dd15d9c5d7..9c6ab6728eab 100644 --- a/middleware/rate-limit.go +++ b/middleware/rate-limit.go @@ -18,6 +18,14 @@ var defNext = func(c *gin.Context) { c.Next() } +func abortRateLimited(c *gin.Context) { + c.JSON(http.StatusTooManyRequests, gin.H{ + "success": false, + "message": "请求过于频繁，请稍后再试", + }) + c.Abort() +} + func redisRateLimiter(c *gin.Context, maxRequestNum int, duration int64, mark string) { ctx := context.Background() rdb := common.RDB @@ -53,8 +61,7 @@ func redisRateLimiter(c *gin.Context, maxRequestNum int, duration int64, mark st // See: https://stackoverflow.com/questions/50970900/why-is-time-since-returning-negative-durations-on-windows if int64(nowTime.Sub(oldTime).Seconds()) < duration { rdb.Expire(ctx, key, common.RateLimitKeyExpirationDuration) - c.Status(http.StatusTooManyRequests) - c.Abort() + abortRateLimited(c) return } else { rdb.LPush(ctx, key, time.Now().Format(timeFormat)) @@ -67,8 +74,7 @@ func redisRateLimiter(c *gin.Context, maxRequestNum int, duration int64, mark st func memoryRateLimiter(c *gin.Context, maxRequestNum int, duration int64, mark string) { key := mark + c.ClientIP() if !inMemoryRateLimiter.Request(key, maxRequestNum, duration) { - c.Status(http.StatusTooManyRequests) - c.Abort() + abortRateLimited(c) return } } @@ -143,8 +149,7 @@ func userRateLimitFactory(maxRequestNum int, duration int64, mark string) func(c } key := fmt.Sprintf("%s:user:%d", mark, userId) if !inMemoryRateLimiter.Request(key, maxRequestNum, duration) { - c.Status(http.StatusTooManyRequests) - c.Abort() + abortRateLimited(c) return } } @@ -184,8 +189,7 @@ func userRedisRateLimiter(c *gin.Context, maxRequestNum int, duration int64, key } if int64(nowTime.Sub(oldTime).Seconds()) < duration { rdb.Expire(ctx, key, common.RateLimitKeyExpirationDuration) - c.Status(http.StatusTooManyRequests) - c.Abort() + abortRateLimited(c) return } else { rdb.LPush(ctx, key, time.Now().Format(timeFormat)) diff --git a/model/ability.go b/model/ability.go index 1d7c53fa5805..023a9fb274f3 100644 --- a/model/ability.go +++ b/model/ability.go @@ -144,7 +144,7 @@ func GetChannel(group string, model string, retry int) (*Channel, error) { } func (channel *Channel) AddAbilities(tx *gorm.DB) error { - models_ := strings.Split(channel.Models, ",") + models_ := channel.GetModels() groups_ := strings.Split(channel.Group, ",") abilitySet := make(map[string]struct{}) abilities := make([]Ability, 0, len(models_)) @@ -216,7 +216,7 @@ func (channel *Channel) UpdateAbilities(tx *gorm.DB) error { } // Then add new abilities - models_ := strings.Split(channel.Models, ",") + models_ := channel.GetModels() groups_ := strings.Split(channel.Group, ",") abilitySet := make(map[string]struct{}) abilities := make([]Ability, 0, len(models_)) diff --git a/model/channel.go b/model/channel.go index f256b54ce35b..1b580976b2a7 100644 --- a/model/channel.go +++ b/model/channel.go @@ -197,7 +197,15 @@ func (channel *Channel) GetModels() []string { if channel.Models == "" { return []string{} } - return strings.Split(strings.Trim(channel.Models, ","), ",") + models := strings.Split(strings.Trim(channel.Models, ","), ",") + out := models[:0] + for _, m := range models { + m = strings.TrimSpace(m) + if m != "" { + out = append(out, m) + } + } + return out } func (channel *Channel) GetGroups() []string { diff --git a/model/channel_cache.go b/model/channel_cache.go index c9c503576038..af203ae49f46 100644 --- a/model/channel_cache.go +++ b/model/channel_cache.go @@ -43,8 +43,8 @@ func InitChannelCache() { continue // skip disabled channels } groups := strings.Split(channel.Group, ",") + models := channel.GetModels() for _, group := range groups { - models := strings.Split(channel.Models, ",") for _, model := range models { if _, ok := newGroup2model2channels[group][model]; !ok { newGroup2model2channels[group][model] = make([]int, 0) diff --git a/model/log.go b/model/log.go index 9203ff28be13..06fbe40be6a9 100644 --- a/model/log.go +++ b/model/log.go @@ -37,6 +37,8 @@ type Log struct { Ip string `json:"ip" gorm:"index;default:''"` RequestId string `json:"request_id,omitempty" gorm:"type:varchar(64);index:idx_logs_request_id;default:''"` Other string `json:"other"` + Balance int `json:"balance" gorm:"default:0"` + BalanceValid bool `json:"-" gorm:"default:false"` } // don't use iota, avoid change log type value @@ -66,13 +68,77 @@ func formatUserLogs(logs []*Log, startIdx int) { } } +func logBalanceDelta(log *Log) int { + if log == nil { + return 0 + } + switch log.Type { + case LogTypeConsume: + return -log.Quota + case LogTypeRefund, LogTypeTopup, LogTypeManage: + return log.Quota + default: + return 0 + } +} + +func AttachLogBalances(logs []*Log) { + if len(logs) == 0 { + return + } + balances := make(map[int]int) + for _, log := range logs { + if log == nil || log.UserId == 0 || log.BalanceValid { + continue + } + if _, ok := balances[log.UserId]; ok { + continue + } + quota, err := GetUserQuota(log.UserId, false) + if err != nil { + continue + } + var newerDelta int + _ = LOG_DB.Model(&Log{}). + Select("COALESCE(SUM(CASE WHEN type = ? THEN -quota WHEN type IN (?, ?) THEN quota ELSE 0 END), 0)", LogTypeConsume, LogTypeTopup, LogTypeRefund). + Where("user_id = ? AND id > ?", log.UserId, log.Id). + Scan(&newerDelta).Error + balances[log.UserId] = quota - newerDelta + } + for _, log := range logs { + if log == nil || log.UserId == 0 || log.BalanceValid { + continue + } + balance, ok := balances[log.UserId] + if !ok { + continue + } + log.Balance = balance + balances[log.UserId] = balance - logBalanceDelta(log) + } +} + +func attachCurrentBalanceSnapshot(log *Log) { + if log == nil || log.UserId == 0 { + return + } + quota, err := GetUserQuota(log.UserId, true) + if err != nil { + common.SysLog(fmt.Sprintf("failed to attach log balance snapshot for user %d: %v", log.UserId, err)) + return + } + quota += pendingBatchUpdateValue(BatchUpdateTypeUserQuota, log.UserId) + log.Balance = quota + log.BalanceValid = true +} + func GetLogByTokenId(tokenId int) (logs []*Log, err error) { err = LOG_DB.Model(&Log{}).Where("token_id = ?", tokenId).Order("id desc").Limit(common.MaxRecentItems).Find(&logs).Error formatUserLogs(logs, 0) return logs, err } -func RecordLog(userId int, logType int, content string) { +func RecordLog(userId int, logType int, content string, quota int) { if logType == LogTypeConsume && !common.LogConsumeEnabled { return } @@ -83,7 +149,9 @@ func RecordLog(userId int, logType int, content string) { CreatedAt: common.GetTimestamp(), Type: logType, Content: content, + Quota: quota, } + attachCurrentBalanceSnapshot(log) err := LOG_DB.Create(log).Error if err != nil { common.SysLog("failed to record log: " + err.Error()) @@ -91,7 +159,7 @@ func RecordLog(userId int, logType int, content string) { } // RecordLogWithAdminInfo 记录操作日志，并将管理员相关信息存入 Other.admin_info， -func RecordLogWithAdminInfo(userId int, logType int, content string, adminInfo map[string]interface{}) { +func RecordLogWithAdminInfo(userId int, logType int, content string, quota int, adminInfo map[string]interface{}) { if logType == LogTypeConsume && !common.LogConsumeEnabled { return } @@ -102,7 +170,9 @@ func RecordLogWithAdminInfo(userId int, logType int, content string, adminInfo m CreatedAt: common.GetTimestamp(), Type: logType, Content: content, + Quota: quota, } + attachCurrentBalanceSnapshot(log) if len(adminInfo) > 0 { other := map[string]interface{}{ "admin_info": adminInfo, @@ -136,6 +206,7 @@ func RecordTopupLog(userId int, content string, callerIp string, paymentMethod s Ip: callerIp, Other: common.MapToJsonStr(other), } + attachCurrentBalanceSnapshot(log) err := LOG_DB.Create(log).Error if err != nil { common.SysLog("failed to record topup log: " + err.Error()) @@ -180,6 +251,7 @@ func RecordErrorLog(c *gin.Context, userId int, channelId int, modelName string, RequestId: requestId, Other: otherStr, } + attachCurrentBalanceSnapshot(log) err := LOG_DB.Create(log).Error if err != nil { logger.LogError(c, "failed to record log: "+err.Error()) @@ -241,13 +313,17 @@ func RecordConsumeLog(c *gin.Context, userId int, params RecordConsumeLogParams) RequestId: requestId, Other: otherStr, } + attachCurrentBalanceSnapshot(log) err := LOG_DB.Create(log).Error if err != nil { logger.LogError(c, "failed to record log: "+err.Error()) } if common.DataExportEnabled { gopool.Go(func() { - LogQuotaData(userId, username, params.ModelName, params.Quota, common.GetTimestamp(), params.PromptTokens+params.CompletionTokens) + now := common.GetTimestamp() + tokenUsed := params.PromptTokens + params.CompletionTokens + LogQuotaData(userId, username, params.ModelName, params.Quota, now, tokenUsed) + LogMinuteQuotaData(userId, username, params.ModelName, params.Quota, now, tokenUsed) }) } } @@ -289,6 +365,7 @@ func RecordTaskBillingLog(params RecordTaskBillingLogParams) { Group: params.Group, Other: common.MapToJsonStr(params.Other), } + attachCurrentBalanceSnapshot(log) err := LOG_DB.Create(log).Error if err != nil { common.SysLog("failed to record task billing log: " + err.Error()) @@ -376,6 +453,7 @@ func GetAllLogs(logType int, startTimestamp int64, endTimestamp int64, modelName } } + AttachLogBalances(logs) return logs, total, err } @@ -422,6 +500,7 @@ func GetUserLogs(userId int, logType int, startTimestamp int64, endTimestamp int return nil, 0, errors.New("查询日志失败") } + AttachLogBalances(logs) formatUserLogs(logs, startIdx) return logs, total, err } diff --git a/model/main.go b/model/main.go index f37cb667cd43..368fff8f46c1 100644 --- a/model/main.go +++ b/model/main.go @@ -267,6 +267,7 @@ func migrateDB() error { &Midjourney{}, &TopUp{}, &QuotaData{}, + &MinuteQuotaData{}, &Task{}, &Model{}, &Vendor{}, diff --git a/model/option.go b/model/option.go index 37fb6cf5bdc6..d07e417e7fff 100644 --- a/model/option.go +++ b/model/option.go @@ -44,6 +44,7 @@ func InitOptionMap() { common.OptionMap["WeChatAuthEnabled"] = strconv.FormatBool(common.WeChatAuthEnabled) common.OptionMap["TurnstileCheckEnabled"] = strconv.FormatBool(common.TurnstileCheckEnabled) common.OptionMap["RegisterEnabled"] = strconv.FormatBool(common.RegisterEnabled) + common.OptionMap["SingleDeviceLoginEnabled"] = strconv.FormatBool(common.SingleDeviceLoginEnabled) common.OptionMap["AutomaticDisableChannelEnabled"] = strconv.FormatBool(common.AutomaticDisableChannelEnabled) common.OptionMap["AutomaticEnableChannelEnabled"] = strconv.FormatBool(common.AutomaticEnableChannelEnabled) common.OptionMap["LogConsumeEnabled"] = strconv.FormatBool(common.LogConsumeEnabled) @@ -63,6 +64,13 @@ func InitOptionMap() { common.OptionMap["SMTPToken"] = "" common.OptionMap["SMTPSSLEnabled"] = strconv.FormatBool(common.SMTPSSLEnabled) common.OptionMap["SMTPForceAuthLogin"] = strconv.FormatBool(common.SMTPForceAuthLogin) + common.OptionMap["SmsEnabled"] = "false" + common.OptionMap["SmsProvider"] = "" + common.OptionMap["SmsSignName"] = "" + common.OptionMap["SmsTemplateCode"] = "" + common.OptionMap["SmsAccessKeyId"] = "" + common.OptionMap["SmsAccessKeySecret"] = "" + common.OptionMap["SmsRegisterTemplate"] = "您的验证码是 {{code}}，{{minutes}} 分钟内有效。如非本人操作，请忽略。" common.OptionMap["Notice"] = "" common.OptionMap["About"] = "" common.OptionMap["HomePageContent"] = "" @@ -152,6 +160,9 @@ func InitOptionMap() { common.OptionMap["ImageRatio"] = ratio_setting.ImageRatio2JSONString() common.OptionMap["AudioRatio"] = ratio_setting.AudioRatio2JSONString() common.OptionMap["AudioCompletionRatio"] = ratio_setting.AudioCompletionRatio2JSONString() + common.OptionMap["ModelDisplayName"] = ratio_setting.ModelDisplayName2JSONString() + common.OptionMap["ModelModalities"] = ratio_setting.ModelModalities2JSONString() + common.OptionMap["ModelImg"] = ratio_setting.ModelImg2JSONString() common.OptionMap["TopUpLink"] = common.TopUpLink //common.OptionMap["ChatLink"] = common.ChatLink //common.OptionMap["ChatLink2"] = common.ChatLink2 @@ -267,6 +278,8 @@ func updateOptionMap(key string, value string) (err error) { common.TurnstileCheckEnabled = boolValue case "RegisterEnabled": common.RegisterEnabled = boolValue + case "SingleDeviceLoginEnabled": + common.SingleDeviceLoginEnabled = boolValue case "EmailDomainRestrictionEnabled": common.EmailDomainRestrictionEnabled = boolValue case "EmailAliasRestrictionEnabled": @@ -521,6 +534,12 @@ func updateOptionMap(key string, value string) (err error) { err = ratio_setting.UpdateAudioRatioByJSONString(value) case "AudioCompletionRatio": err = ratio_setting.UpdateAudioCompletionRatioByJSONString(value) + case "ModelDisplayName": + err = ratio_setting.UpdateModelDisplayNameByJSONString(value) + case "ModelModalities": + err = ratio_setting.UpdateModelModalitiesByJSONString(value) + case "ModelImg": + err = ratio_setting.UpdateModelImgByJSONString(value) case "TopUpLink": common.TopUpLink = value //case "ChatLink": diff --git a/model/pricing.go b/model/pricing.go index 54ae98451337..02f41611f430 100644 --- a/model/pricing.go +++ b/model/pricing.go @@ -16,6 +16,9 @@ import ( type Pricing struct { ModelName string `json:"model_name"` + DisplayName string `json:"display_name,omitempty"` + Modalities string `json:"modalities,omitempty"` + ModelImg string `json:"model_img"` Description string `json:"description,omitempty"` Icon string `json:"icon,omitempty"` Tags string `json:"tags,omitempty"` @@ -292,6 +295,9 @@ func updatePricing() { pricing.Tags = meta.Tags pricing.VendorID = meta.VendorID } + pricing.DisplayName = ratio_setting.GetModelDisplayName(model) + pricing.Modalities = ratio_setting.GetModelModalities(model) + pricing.ModelImg = ratio_setting.GetModelImg(model) modelPrice, findPrice := ratio_setting.GetModelPrice(model, false) if findPrice { pricing.ModelPrice = modelPrice diff --git a/model/redemption.go b/model/redemption.go index b0ccb5df1403..72ca302ab076 100644 --- a/model/redemption.go +++ b/model/redemption.go @@ -151,7 +151,7 @@ func Redeem(key string, userId int) (quota int, err error) { common.SysError("redemption failed: " + err.Error()) return 0, ErrRedeemFailed } - RecordLog(userId, LogTypeTopup, fmt.Sprintf("通过兑换码充值 %s，兑换码ID %d", logger.LogQuota(redemption.Quota), redemption.Id)) + RecordLog(userId, LogTypeTopup, fmt.Sprintf("通过兑换码充值 %s，兑换码ID %d ", logger.LogQuota(redemption.Quota), redemption.Id), redemption.Quota) return redemption.Quota, nil } diff --git a/model/subscription.go b/model/subscription.go index 10e750c3f355..16d5d5146feb 100644 --- a/model/subscription.go +++ b/model/subscription.go @@ -569,7 +569,7 @@ func CompleteSubscriptionOrder(tradeNo string, providerPayload string, expectedP } if logUserId > 0 { msg := fmt.Sprintf("订阅购买成功，套餐: %s，支付金额: %.2f，支付方式: %s", logPlanTitle, logMoney, logPaymentMethod) - RecordLog(logUserId, LogTypeTopup, msg) + RecordLog(logUserId, LogTypeTopup, msg, 0) } return nil } diff --git a/model/token.go b/model/token.go index 0529e2c7cf14..ed7163853d6d 100644 --- a/model/token.go +++ b/model/token.go @@ -225,6 +225,48 @@ func ValidateUserToken(key string) (token *Token, err error) { return nil, fmt.Errorf("%w: %v", ErrDatabase, err) } +func GetNextAvailableTokenForUser(userId int, currentTokenId int) (*Token, error) { + if userId == 0 { + return nil, errors.New("userId 为空！") + } + + now := common.GetTimestamp() + findCandidates := func(afterCurrent bool) ([]Token, error) { + var tokens []Token + query := DB.Where("user_id = ? and id <> ? and status = ? and (expired_time = -1 or expired_time > ?)", + userId, currentTokenId, common.TokenStatusEnabled, now) + if currentTokenId > 0 { + if afterCurrent { + query = query.Where("id < ?", currentTokenId) + } else { + query = query.Where("id > ?", currentTokenId) + } + } else if !afterCurrent { + return tokens, nil + } + err := query.Order("id desc").Find(&tokens).Error + return tokens, err + } + + for _, afterCurrent := range []bool{true, false} { + candidates, err := findCandidates(afterCurrent) + if err != nil { + return nil, fmt.Errorf("%w: %v", ErrDatabase, err) + } + for i := range candidates { + token, err := ValidateUserToken(candidates[i].Key) + if err == nil { + return token, nil + } + if !errors.Is(err, ErrTokenInvalid) && !errors.Is(err, ErrTokenNotProvided) { + return nil, err + } + } + } + + return nil, ErrTokenInvalid +} + func GetTokenByIds(id int, userId int) (*Token, error) { if id == 0 || userId == 0 { return nil, errors.New("id 或 userId 为空！") diff --git a/model/topup.go b/model/topup.go index c1ac663f7595..f8bd78839219 100644 --- a/model/topup.go +++ b/model/topup.go @@ -571,7 +571,7 @@ func RechargeWaffoPancake(tradeNo string) (err error) { } if quotaToAdd > 0 { - RecordLog(topUp.UserId, LogTypeTopup, fmt.Sprintf("Waffo Pancake充值成功，充值额度: %v，支付金额: %.2f", logger.FormatQuota(quotaToAdd), topUp.Money)) + RecordLog(topUp.UserId, LogTypeTopup, fmt.Sprintf("Waffo Pancake充值成功，充值额度: %v，支付金额: %.2f", logger.FormatQuota(quotaToAdd), topUp.Money), quotaToAdd) } return nil diff --git a/model/usedata.go b/model/usedata.go index f0ea055ae395..53c9c71f8448 100644 --- a/model/usedata.go +++ b/model/usedata.go @@ -16,16 +16,36 @@ type QuotaData struct { Username string `json:"username" gorm:"index:idx_qdt_model_user_name,priority:2;size:64;default:''"` ModelName string `json:"model_name" gorm:"index:idx_qdt_model_user_name,priority:1;size:64;default:''"` CreatedAt int64 `json:"created_at" gorm:"bigint;index:idx_qdt_created_at,priority:2"` + EndAt int64 `json:"end_at" gorm:"bigint;index:idx_qdt_end_at,priority:2"` TokenUsed int `json:"token_used" gorm:"default:0"` Count int `json:"count" gorm:"default:0"` Quota int `json:"quota" gorm:"default:0"` } +// MinuteQuotaData 分钟级柱状图数据。 +// 聚合粒度：user_id + username + model_name + created_at(分钟)。 +type MinuteQuotaData struct { + Id int `json:"id"` + UserID int `json:"user_id" gorm:"index;uniqueIndex:uk_min_qdt_user_model_time,priority:1"` + Username string `json:"username" gorm:"index:idx_min_qdt_model_user_name,priority:2;uniqueIndex:uk_min_qdt_user_model_time,priority:2;size:64;default:''"` + ModelName string `json:"model_name" gorm:"index:idx_min_qdt_model_user_name,priority:1;uniqueIndex:uk_min_qdt_user_model_time,priority:3;size:64;default:''"` + CreatedAt int64 `json:"created_at" gorm:"bigint;index:idx_min_qdt_created_at,priority:2;uniqueIndex:uk_min_qdt_user_model_time,priority:4"` + EndAt int64 `json:"end_at" gorm:"bigint;index:idx_min_qdt_end_at,priority:2"` + TokenUsed int `json:"token_used" gorm:"default:0"` + Count int `json:"count" gorm:"default:0"` + Quota int `json:"quota" gorm:"default:0"` +} + +func (MinuteQuotaData) TableName() string { + return "quota_data_minute" +} + func UpdateQuotaData() { for { if common.DataExportEnabled { common.SysLog("正在更新数据看板数据...") SaveQuotaDataCache() + SaveMinuteQuotaDataCache() } time.Sleep(time.Duration(common.DataExportInterval) * time.Minute) } @@ -33,6 +53,8 @@ func UpdateQuotaData() { var CacheQuotaData = make(map[string]*QuotaData) var CacheQuotaDataLock = sync.Mutex{} +var CacheMinuteQuotaData = make(map[string]*MinuteQuotaData) +var CacheMinuteQuotaDataLock = sync.Mutex{} func logQuotaDataCache(userId int, username string, modelName string, quota int, createdAt int64, tokenUsed int) { key := fmt.Sprintf("%d-%s-%s-%d", userId, username, modelName, createdAt) @@ -47,6 +69,7 @@ func logQuotaDataCache(userId int, username string, modelName string, quota int, Username: username, ModelName: modelName, CreatedAt: createdAt, + EndAt: createdAt + 3600, Count: 1, Quota: quota, TokenUsed: tokenUsed, @@ -64,6 +87,37 @@ func LogQuotaData(userId int, username string, modelName string, quota int, crea logQuotaDataCache(userId, username, modelName, quota, createdAt, tokenUsed) } +func logMinuteQuotaDataCache(userId int, username string, modelName string, quota int, createdAt int64, tokenUsed int) { + key := fmt.Sprintf("%d-%s-%s-%d", userId, username, modelName, createdAt) + quotaData, ok := CacheMinuteQuotaData[key] + if ok { + quotaData.Count += 1 + quotaData.Quota += quota + quotaData.TokenUsed += tokenUsed + } else { + quotaData = &MinuteQuotaData{ + UserID: userId, + Username: username, + ModelName: modelName, + CreatedAt: createdAt, + EndAt: createdAt + 60, + Count: 1, + Quota: quota, + TokenUsed: tokenUsed, + } + } + CacheMinuteQuotaData[key] = quotaData +} + +func LogMinuteQuotaData(userId int, username string, modelName string, quota int, createdAt int64, tokenUsed int) { + // 只精确到分钟 + createdAt = createdAt - (createdAt % 60) + + CacheMinuteQuotaDataLock.Lock() + defer CacheMinuteQuotaDataLock.Unlock() + logMinuteQuotaDataCache(userId, username, modelName, quota, createdAt, tokenUsed) +} + func SaveQuotaDataCache() { CacheQuotaDataLock.Lock() defer CacheQuotaDataLock.Unlock() @@ -80,7 +134,7 @@ func SaveQuotaDataCache() { //quotaDataDB.Count += quotaData.Count //quotaDataDB.Quota += quotaData.Quota //DB.Table("quota_data").Save(quotaDataDB) - increaseQuotaData(quotaData.UserID, quotaData.Username, quotaData.ModelName, quotaData.Count, quotaData.Quota, quotaData.CreatedAt, quotaData.TokenUsed) + increaseQuotaData(quotaData.UserID, quotaData.Username, quotaData.ModelName, quotaData.Count, quotaData.Quota, quotaData.CreatedAt, quotaData.EndAt, quotaData.TokenUsed) } else { DB.Table("quota_data").Create(quotaData) } @@ -89,11 +143,30 @@ func SaveQuotaDataCache() { common.SysLog(fmt.Sprintf("保存数据看板数据成功，共保存%d条数据", size)) } -func increaseQuotaData(userId int, username string, modelName string, count int, quota int, createdAt int64, tokenUsed int) { +func SaveMinuteQuotaDataCache() { + CacheMinuteQuotaDataLock.Lock() + defer CacheMinuteQuotaDataLock.Unlock() + size := len(CacheMinuteQuotaData) + for _, quotaData := range CacheMinuteQuotaData { + quotaDataDB := &MinuteQuotaData{} + DB.Table("quota_data_minute").Where("user_id = ? and username = ? and model_name = ? and created_at = ?", + quotaData.UserID, quotaData.Username, quotaData.ModelName, quotaData.CreatedAt).First(quotaDataDB) + if quotaDataDB.Id > 0 { + increaseMinuteQuotaData(quotaData.UserID, quotaData.Username, quotaData.ModelName, quotaData.Count, quotaData.Quota, quotaData.CreatedAt, quotaData.EndAt, quotaData.TokenUsed) + } else { + DB.Table("quota_data_minute").Create(quotaData) + } + } + CacheMinuteQuotaData = make(map[string]*MinuteQuotaData) + common.SysLog(fmt.Sprintf("保存分钟级数据看板数据成功，共保存%d条数据", size)) +} + +func increaseQuotaData(userId int, username string, modelName string, count int, quota int, createdAt int64, endAt int64, tokenUsed int) { err := DB.Table("quota_data").Where("user_id = ? and username = ? and model_name = ? and created_at = ?", userId, username, modelName, createdAt).Updates(map[string]interface{}{ "count": gorm.Expr("count + ?", count), "quota": gorm.Expr("quota + ?", quota), + "end_at": endAt, "token_used": gorm.Expr("token_used + ?", tokenUsed), }).Error if err != nil { @@ -101,6 +174,19 @@ func increaseQuotaData(userId int, username string, modelName string, count int, } } +func increaseMinuteQuotaData(userId int, username string, modelName string, count int, quota int, createdAt int64, endAt int64, tokenUsed int) { + err := DB.Table("quota_data_minute").Where("user_id = ? and username = ? and model_name = ? and created_at = ?", + userId, username, modelName, createdAt).Updates(map[string]interface{}{ + "count": gorm.Expr("count + ?", count), + "quota": gorm.Expr("quota + ?", quota), + "end_at": endAt, + "token_used": gorm.Expr("token_used + ?", tokenUsed), + }).Error + if err != nil { + common.SysLog(fmt.Sprintf("increaseMinuteQuotaData error: %s", err)) + } +} + func GetQuotaDataByUsername(username string, startTime int64, endTime int64) (quotaData []*QuotaData, err error) { var quotaDatas []*QuotaData // 从quota_data表中查询数据 @@ -118,7 +204,7 @@ func GetQuotaDataByUserId(userId int, startTime int64, endTime int64) (quotaData func GetQuotaDataGroupByUser(startTime int64, endTime int64) (quotaData []*QuotaData, err error) { var quotaDatas []*QuotaData err = DB.Table("quota_data"). - Select("username, created_at, sum(count) as count, sum(quota) as quota, sum(token_used) as token_used"). + Select("username, created_at, max(end_at) as end_at, sum(count) as count, sum(quota) as quota, sum(token_used) as token_used"). Where("created_at >= ? and created_at <= ?", startTime, endTime). Group("username, created_at"). Find("aDatas).Error @@ -133,6 +219,6 @@ func GetAllQuotaDates(startTime int64, endTime int64, username string) (quotaDat // 从quota_data表中查询数据 // only select model_name, sum(count) as count, sum(quota) as quota, model_name, created_at from quota_data group by model_name, created_at; //err = DB.Table("quota_data").Where("created_at >= ? and created_at <= ?", startTime, endTime).Find("aDatas).Error - err = DB.Table("quota_data").Select("model_name, sum(count) as count, sum(quota) as quota, sum(token_used) as token_used, created_at").Where("created_at >= ? and created_at <= ?", startTime, endTime).Group("model_name, created_at").Find("aDatas).Error + err = DB.Table("quota_data").Select("model_name, created_at, max(end_at) as end_at, sum(count) as count, sum(quota) as quota, sum(token_used) as token_used").Where("created_at >= ? and created_at <= ?", startTime, endTime).Group("model_name, created_at").Find("aDatas).Error return quotaDatas, err } diff --git a/model/usedata_minute_test.go b/model/usedata_minute_test.go new file mode 100644 index 000000000000..9c96cc5240d9 --- /dev/null +++ b/model/usedata_minute_test.go @@ -0,0 +1,36 @@ +package model + +import "testing" + +func TestLogMinuteQuotaDataAggregatesByMinute(t *testing.T) { + CacheMinuteQuotaDataLock.Lock() + CacheMinuteQuotaData = make(map[string]*MinuteQuotaData) + CacheMinuteQuotaDataLock.Unlock() + + LogMinuteQuotaData(7, "alice", "gpt-test", 10, 1710000061, 12) + LogMinuteQuotaData(7, "alice", "gpt-test", 15, 1710000079, 18) + + CacheMinuteQuotaDataLock.Lock() + defer CacheMinuteQuotaDataLock.Unlock() + + if len(CacheMinuteQuotaData) != 1 { + t.Fatalf("expected 1 minute bucket, got %d", len(CacheMinuteQuotaData)) + } + + item := CacheMinuteQuotaData["7-alice-gpt-test-1710000060"] + if item == nil { + t.Fatalf("expected bucket at minute timestamp 1710000060") + } + if item.EndAt != 1710000120 { + t.Fatalf("expected end_at 1710000120, got %d", item.EndAt) + } + if item.Count != 2 { + t.Fatalf("expected count 2, got %d", item.Count) + } + if item.Quota != 25 { + t.Fatalf("expected quota 25, got %d", item.Quota) + } + if item.TokenUsed != 30 { + t.Fatalf("expected token_used 30, got %d", item.TokenUsed) + } +} diff --git a/model/user.go b/model/user.go index 79e63e8fd592..03610bc37938 100644 --- a/model/user.go +++ b/model/user.go @@ -29,6 +29,7 @@ type User struct { Role int `json:"role" gorm:"type:int;default:1"` // admin, common Status int `json:"status" gorm:"type:int;default:1"` // enabled, disabled Email string `json:"email" gorm:"index" validate:"max=50"` + Phone *string `json:"phone,omitempty" gorm:"type:varchar(20);uniqueIndex"` GitHubId string `json:"github_id" gorm:"column:github_id;index"` DiscordId string `json:"discord_id" gorm:"column:discord_id;index"` OidcId string `json:"oidc_id" gorm:"column:oidc_id;index"` @@ -36,6 +37,7 @@ type User struct { TelegramId string `json:"telegram_id" gorm:"column:telegram_id;index"` VerificationCode string `json:"verification_code" gorm:"-:all"` // this field is only for Email verification, don't save it to database! AccessToken *string `json:"access_token" gorm:"type:char(32);column:access_token;uniqueIndex"` // this token is for system management + SessionToken string `json:"-" gorm:"type:char(32);column:session_token;default:'';index"` Quota int `json:"quota" gorm:"type:int;default:0"` UsedQuota int `json:"used_quota" gorm:"type:int;default:0;column:used_quota"` // used quota RequestCount int `json:"request_count" gorm:"type:int;default:0;"` // request number @@ -76,6 +78,56 @@ func (user *User) SetAccessToken(token string) { user.AccessToken = &token } +func generateSessionToken() string { + // uuid 去掉连字符，保持紧凑定长 token + return strings.ReplaceAll(common.GetUUID(), "-", "") +} + +// RotateUserSessionToken rotates current valid dashboard session token for the user. +// After rotating, previous session cookies on other devices will fail auth checks. +func RotateUserSessionToken(userId int) (string, error) { + token := generateSessionToken() + if err := DB.Model(&User{}).Where("id = ?", userId).Update("session_token", token).Error; err != nil { + return "", err + } + return token, nil +} + +// IssueUserSessionToken returns the token that should be stored in a newly +// created dashboard session. When single-device login is enabled it rotates the +// token so older devices are invalidated; otherwise it reuses the existing token +// so multiple devices can stay signed in. Password changes should still call +// RotateUserSessionToken directly to invalidate all old sessions. +func IssueUserSessionToken(userId int) (string, error) { + if common.SingleDeviceLoginEnabled { + return RotateUserSessionToken(userId) + } + var current string + if err := DB.Model(&User{}).Where("id = ?", userId).Select("session_token").Scan(¤t).Error; err != nil { + return "", err + } + if current != "" { + return current, nil + } + return RotateUserSessionToken(userId) +} + +// ValidateUserSessionToken checks whether the session token matches the latest one in DB. +func ValidateUserSessionToken(userId int, sessionToken string) (bool, error) { + if userId <= 0 || sessionToken == "" { + return false, nil + } + var current string + err := DB.Model(&User{}).Where("id = ?", userId).Select("session_token").Scan(¤t).Error + if err != nil { + return false, err + } + if current == "" { + return false, nil + } + return current == sessionToken, nil +} + func (user *User) GetSetting() dto.UserSetting { setting := dto.UserSetting{} if user.Setting != "" { @@ -713,7 +765,15 @@ func ResetUserPasswordByEmail(email string, password string) error { if err != nil { return err } - err = DB.Model(&User{}).Where("email = ?", email).Update("password", hashedPassword).Error + var user User + if err = DB.Where("email = ?", email).Select("id").First(&user).Error; err != nil { + return err + } + if err = DB.Model(&User{}).Where("email = ?", email).Update("password", hashedPassword).Error; err != nil { + return err + } + // 重置密码后立即轮换会话令牌，强制所有旧设备会话失效 + _, err = RotateUserSessionToken(user.Id) return err } diff --git a/model/utils.go b/model/utils.go index adfd8e139a05..ca53f47f4ed6 100644 --- a/model/utils.go +++ b/model/utils.go @@ -49,6 +49,15 @@ func addNewRecord(type_ int, id int, value int) { } } +func pendingBatchUpdateValue(type_ int, id int) int { + if !common.BatchUpdateEnabled { + return 0 + } + batchUpdateLocks[type_].Lock() + defer batchUpdateLocks[type_].Unlock() + return batchUpdateStores[type_][id] +} + func batchUpdate() { // check if there's any data to update hasData := false diff --git a/relay/channel/openai/relay-openai.go b/relay/channel/openai/relay-openai.go index d33c5555f267..4a335533b453 100644 --- a/relay/channel/openai/relay-openai.go +++ b/relay/channel/openai/relay-openai.go @@ -181,6 +181,28 @@ func OaiStreamHandler(c *gin.Context, info *relaycommon.RelayInfo, resp *http.Re } if !containStreamUsage { + // 失败判定（顺序检查）：先看 containStreamUsage != true（上游没给 usage chunk）， + // 进入此分支后再判 EndReason != done，命中即视为失败：返回错误让 Relay defer + // 走 Billing.Refund，不再用估算的 prompt tokens 扣费、也不写 RecordConsumeLog。 + // 主要拦截上游 TPM/配额限制下中途断流或塞 error chunk 的情形。 + var endReason relaycommon.StreamEndReason + if info.StreamStatus != nil { + endReason = info.StreamStatus.EndReason + } + if endReason != relaycommon.StreamEndReasonDone { + logger.LogError(c, fmt.Sprintf( + "stream incomplete (channel=%d type=%d model=%s reason=%s sent=%d items=%d), suppressing billing; lastChunk=[%s]", + info.ChannelId, info.ChannelType, info.UpstreamModelName, + endReason, info.SendResponseCount, len(streamItems), lastStreamData, + )) + errMsg := fmt.Errorf("upstream stream ended without [DONE] or usage; reason=%s", endReason) + // 已经向客户端发过 chunk 时，重试会拼接脏数据 → 禁用重试 + if info.SendResponseCount > 0 { + return nil, types.NewOpenAIError(errMsg, types.ErrorCodeBadResponse, http.StatusInternalServerError, types.ErrOptionWithSkipRetry()) + } + return nil, types.NewOpenAIError(errMsg, types.ErrorCodeBadResponse, http.StatusInternalServerError) + } + // [DONE] 已收到但没有 usage chunk：合法的空响应 / refusal，沿用估算路径。 usage = service.ResponseText2Usage(c, responseTextBuilder.String(), info.UpstreamModelName, info.GetEstimatePromptTokens()) usage.CompletionTokens += toolCount * 7 } diff --git a/relay/channel/openai/responses_via_chat.go b/relay/channel/openai/responses_via_chat.go new file mode 100644 index 000000000000..a4d9b74a21f1 --- /dev/null +++ b/relay/channel/openai/responses_via_chat.go @@ -0,0 +1,480 @@ +package openai + +import ( + "fmt" + "io" + "net/http" + "strings" + "time" + + "github.com/QuantumNous/new-api/common" + "github.com/QuantumNous/new-api/dto" + "github.com/QuantumNous/new-api/logger" + relaycommon "github.com/QuantumNous/new-api/relay/common" + "github.com/QuantumNous/new-api/relay/helper" + "github.com/QuantumNous/new-api/service" + "github.com/QuantumNous/new-api/types" + + "github.com/gin-gonic/gin" +) + +// OaiChatToResponsesHandler reads a chat-completions response from upstream and +// reformats it as an OpenAI /v1/responses response for the client. +func OaiChatToResponsesHandler(c *gin.Context, info *relaycommon.RelayInfo, resp *http.Response) (*dto.Usage, *types.NewAPIError) { + if resp == nil || resp.Body == nil { + return nil, types.NewOpenAIError(fmt.Errorf("invalid response"), types.ErrorCodeBadResponse, http.StatusInternalServerError) + } + + defer service.CloseResponseBodyGracefully(resp) + + body, err := io.ReadAll(resp.Body) + if err != nil { + return nil, types.NewOpenAIError(err, types.ErrorCodeReadResponseBodyFailed, http.StatusInternalServerError) + } + + var chatResp dto.OpenAITextResponse + if err := common.Unmarshal(body, &chatResp); err != nil { + return nil, types.NewOpenAIError(err, types.ErrorCodeBadResponseBody, http.StatusInternalServerError) + } + + if oaiErr := chatResp.GetOpenAIError(); oaiErr != nil && oaiErr.Type != "" { + return nil, types.WithOpenAIError(*oaiErr, resp.StatusCode) + } + + usage := chatResp.Usage + if usage.PromptTokens == 0 && usage.CompletionTokens == 0 { + var combined strings.Builder + for _, choice := range chatResp.Choices { + combined.WriteString(choice.Message.StringContent()) + combined.WriteString(choice.Message.ReasoningContent) + combined.WriteString(choice.Message.Reasoning) + } + fallback := service.ResponseText2Usage(c, combined.String(), info.UpstreamModelName, info.GetEstimatePromptTokens()) + usage = *fallback + } + + respID := helper.GetResponseID(c) + createdAt := time.Now().Unix() + if createdRaw, ok := chatResp.Created.(int64); ok && createdRaw != 0 { + createdAt = createdRaw + } else if createdRaw, ok := chatResp.Created.(float64); ok && createdRaw != 0 { + createdAt = int64(createdRaw) + } + + out := buildResponsesResponseFromChat(&chatResp, &usage, info, respID, createdAt) + bodyOut, err := common.Marshal(out) + if err != nil { + return nil, types.NewOpenAIError(err, types.ErrorCodeJsonMarshalFailed, http.StatusInternalServerError) + } + + service.IOCopyBytesGracefully(c, resp, bodyOut) + return &usage, nil +} + +// OaiChatToResponsesStreamHandler consumes a chat-completions SSE stream from +// upstream and re-emits it as an OpenAI /v1/responses event stream. +func OaiChatToResponsesStreamHandler(c *gin.Context, info *relaycommon.RelayInfo, resp *http.Response) (*dto.Usage, *types.NewAPIError) { + if resp == nil || resp.Body == nil { + return nil, types.NewOpenAIError(fmt.Errorf("invalid response"), types.ErrorCodeBadResponse, http.StatusInternalServerError) + } + + defer service.CloseResponseBodyGracefully(resp) + + respID := helper.GetResponseID(c) + createdAt := time.Now().Unix() + model := info.UpstreamModelName + + var ( + usage = &dto.Usage{} + textBuilder strings.Builder + streamErr *types.NewAPIError + sentCreated bool + messageItemID = fmt.Sprintf("msg_%s", respID) + messageOpen bool + contentPartOpn bool + toolItemIDs = map[int]string{} + toolNames = map[int]string{} + toolArgs = map[int]string{} + toolCallIDs = map[int]string{} + finishReason string + ) + + emit := func(eventType string, payload any) bool { + data, err := common.Marshal(payload) + if err != nil { + streamErr = types.NewOpenAIError(err, types.ErrorCodeJsonMarshalFailed, http.StatusInternalServerError) + return false + } + c.Render(-1, common.CustomEvent{Data: fmt.Sprintf("event: %s\n", eventType)}) + c.Render(-1, common.CustomEvent{Data: fmt.Sprintf("data: %s\n", string(data))}) + _ = helper.FlushWriter(c) + return true + } + + emitCreated := func() bool { + if sentCreated { + return true + } + sentCreated = true + payload := map[string]any{ + "type": "response.created", + "response": map[string]any{ + "id": respID, + "object": "response", + "created_at": createdAt, + "status": "in_progress", + "model": model, + "output": []any{}, + }, + } + return emit("response.created", payload) + } + + openMessageItem := func() bool { + if messageOpen { + return true + } + if !emitCreated() { + return false + } + messageOpen = true + itemPayload := map[string]any{ + "type": "response.output_item.added", + "output_index": 0, + "item": map[string]any{ + "type": "message", + "id": messageItemID, + "status": "in_progress", + "role": "assistant", + "content": []any{}, + }, + } + if !emit("response.output_item.added", itemPayload) { + return false + } + partPayload := map[string]any{ + "type": "response.content_part.added", + "item_id": messageItemID, + "output_index": 0, + "content_index": 0, + "part": map[string]any{ + "type": "output_text", + "text": "", + }, + } + if !emit("response.content_part.added", partPayload) { + return false + } + contentPartOpn = true + return true + } + + closeMessageItem := func() bool { + if !messageOpen { + return true + } + fullText := textBuilder.String() + if contentPartOpn { + donePart := map[string]any{ + "type": "response.output_text.done", + "item_id": messageItemID, + "output_index": 0, + "content_index": 0, + "text": fullText, + } + if !emit("response.output_text.done", donePart) { + return false + } + doneContent := map[string]any{ + "type": "response.content_part.done", + "item_id": messageItemID, + "output_index": 0, + "content_index": 0, + "part": map[string]any{ + "type": "output_text", + "text": fullText, + }, + } + if !emit("response.content_part.done", doneContent) { + return false + } + contentPartOpn = false + } + itemDone := map[string]any{ + "type": "response.output_item.done", + "output_index": 0, + "item": map[string]any{ + "type": "message", + "id": messageItemID, + "status": "completed", + "role": "assistant", + "content": []map[string]any{ + {"type": "output_text", "text": fullText}, + }, + }, + } + if !emit("response.output_item.done", itemDone) { + return false + } + messageOpen = false + return true + } + + helper.StreamScannerHandler(c, resp, info, func(data string, sr *helper.StreamResult) { + if streamErr != nil { + sr.Stop(streamErr) + return + } + data = strings.TrimSpace(data) + if data == "" || data == "[DONE]" { + return + } + + var chunk dto.ChatCompletionsStreamResponse + if err := common.UnmarshalJsonStr(data, &chunk); err != nil { + logger.LogError(c, "failed to unmarshal chat stream chunk: "+err.Error()) + sr.Error(err) + return + } + if chunk.Model != "" { + model = chunk.Model + } + if chunk.Created != 0 { + createdAt = chunk.Created + } + if chunk.Usage != nil && chunk.Usage.TotalTokens != 0 { + *usage = *chunk.Usage + } + + for _, choice := range chunk.Choices { + if choice.Delta.Content != nil && *choice.Delta.Content != "" { + if !openMessageItem() { + sr.Stop(streamErr) + return + } + delta := *choice.Delta.Content + textBuilder.WriteString(delta) + payload := map[string]any{ + "type": "response.output_text.delta", + "item_id": messageItemID, + "output_index": 0, + "content_index": 0, + "delta": delta, + } + if !emit("response.output_text.delta", payload) { + sr.Stop(streamErr) + return + } + } + for _, tc := range choice.Delta.ToolCalls { + idx := 0 + if tc.Index != nil { + idx = *tc.Index + } + if _, ok := toolItemIDs[idx]; !ok { + if !emitCreated() { + sr.Stop(streamErr) + return + } + itemID := fmt.Sprintf("fc_%s_%d", respID, idx) + toolItemIDs[idx] = itemID + toolCallIDs[idx] = tc.ID + toolNames[idx] = tc.Function.Name + addPayload := map[string]any{ + "type": "response.output_item.added", + "output_index": idx + 1, + "item": map[string]any{ + "type": "function_call", + "id": itemID, + "status": "in_progress", + "call_id": tc.ID, + "name": tc.Function.Name, + "arguments": "", + }, + } + if !emit("response.output_item.added", addPayload) { + sr.Stop(streamErr) + return + } + } else { + if tc.ID != "" && toolCallIDs[idx] == "" { + toolCallIDs[idx] = tc.ID + } + if tc.Function.Name != "" && toolNames[idx] == "" { + toolNames[idx] = tc.Function.Name + } + } + if tc.Function.Arguments != "" { + toolArgs[idx] += tc.Function.Arguments + argPayload := map[string]any{ + "type": "response.function_call_arguments.delta", + "item_id": toolItemIDs[idx], + "output_index": idx + 1, + "delta": tc.Function.Arguments, + } + if !emit("response.function_call_arguments.delta", argPayload) { + sr.Stop(streamErr) + return + } + } + } + if choice.FinishReason != nil && *choice.FinishReason != "" { + finishReason = *choice.FinishReason + } + } + }) + + if streamErr != nil { + return nil, streamErr + } + + if !closeMessageItem() { + return nil, streamErr + } + + // Close any open function-call items. + for idx, itemID := range toolItemIDs { + doneArgs := map[string]any{ + "type": "response.function_call_arguments.done", + "item_id": itemID, + "output_index": idx + 1, + "arguments": toolArgs[idx], + } + if !emit("response.function_call_arguments.done", doneArgs) { + return nil, streamErr + } + itemDone := map[string]any{ + "type": "response.output_item.done", + "output_index": idx + 1, + "item": map[string]any{ + "type": "function_call", + "id": itemID, + "status": "completed", + "call_id": toolCallIDs[idx], + "name": toolNames[idx], + "arguments": toolArgs[idx], + }, + } + if !emit("response.output_item.done", itemDone) { + return nil, streamErr + } + } + + if usage.TotalTokens == 0 { + fallback := service.ResponseText2Usage(c, textBuilder.String(), info.UpstreamModelName, info.GetEstimatePromptTokens()) + *usage = *fallback + } + + if !sentCreated { + _ = emitCreated() + } + + // Build final response object for response.completed. + finalUsage := map[string]any{ + "input_tokens": usage.PromptTokens, + "output_tokens": usage.CompletionTokens, + "total_tokens": usage.TotalTokens, + } + + var output []map[string]any + fullText := textBuilder.String() + if fullText != "" || len(toolItemIDs) == 0 { + output = append(output, map[string]any{ + "type": "message", + "id": messageItemID, + "status": "completed", + "role": "assistant", + "content": []map[string]any{ + {"type": "output_text", "text": fullText}, + }, + }) + } + for idx := 0; idx < len(toolItemIDs); idx++ { + itemID, ok := toolItemIDs[idx] + if !ok { + continue + } + output = append(output, map[string]any{ + "type": "function_call", + "id": itemID, + "status": "completed", + "call_id": toolCallIDs[idx], + "name": toolNames[idx], + "arguments": toolArgs[idx], + }) + } + + status := "completed" + if finishReason == "length" { + status = "incomplete" + } + + completedPayload := map[string]any{ + "type": "response.completed", + "response": map[string]any{ + "id": respID, + "object": "response", + "created_at": createdAt, + "model": model, + "status": status, + "output": output, + "usage": finalUsage, + }, + } + if !emit("response.completed", completedPayload) { + return nil, streamErr + } + + return usage, nil +} + +func buildResponsesResponseFromChat(chatResp *dto.OpenAITextResponse, usage *dto.Usage, info *relaycommon.RelayInfo, respID string, createdAt int64) *dto.OpenAIResponsesResponse { + out := &dto.OpenAIResponsesResponse{ + ID: respID, + Object: "response", + CreatedAt: int(createdAt), + Model: chatResp.Model, + Usage: usage, + } + if info != nil && info.UpstreamModelName != "" && out.Model == "" { + out.Model = info.UpstreamModelName + } + + status := "completed" + for _, choice := range chatResp.Choices { + msgID := fmt.Sprintf("msg_%s_%d", respID, choice.Index) + text := choice.Message.StringContent() + if text != "" { + out.Output = append(out.Output, dto.ResponsesOutput{ + Type: "message", + ID: msgID, + Status: "completed", + Role: "assistant", + Content: []dto.ResponsesOutputContent{ + {Type: "output_text", Text: text}, + }, + }) + } + for tcIdx, tc := range choice.Message.ParseToolCalls() { + callID := tc.ID + if callID == "" { + callID = fmt.Sprintf("call_%s_%d", respID, tcIdx) + } + out.Output = append(out.Output, dto.ResponsesOutput{ + Type: "function_call", + ID: fmt.Sprintf("fc_%s_%d_%d", respID, choice.Index, tcIdx), + Status: "completed", + CallId: callID, + Name: tc.Function.Name, + Arguments: tc.Function.Arguments, + }) + } + if choice.FinishReason == "length" { + status = "incomplete" + } + } + + statusBytes, _ := common.Marshal(status) + out.Status = statusBytes + return out +} diff --git a/relay/channel/task/doubao/adaptor.go b/relay/channel/task/doubao/adaptor.go index a6dabb5f1086..1dacc7b2ec81 100644 --- a/relay/channel/task/doubao/adaptor.go +++ b/relay/channel/task/doubao/adaptor.go @@ -6,6 +6,7 @@ import ( "io" "net/http" "strconv" + "strings" "time" "github.com/QuantumNous/new-api/common" @@ -42,6 +43,7 @@ type MediaURL struct { type requestPayload struct { Model string `json:"model"` + TaskType string `json:"task_type,omitempty"` Content []ContentItem `json:"content,omitempty"` CallbackURL string `json:"callback_url,omitempty"` ReturnLastFrame *dto.BoolValue `json:"return_last_frame,omitempty"` @@ -267,6 +269,64 @@ func (a *TaskAdaptor) GetChannelName() string { return ChannelName } +// 火山方舟「视频生成任务」content 里 image_url 的 role 不是对话里的 user/assistant， +// 而是素材语义。reference_image 会走 r2v；部分 Seedance Pro（1.0 / 1.5 等）不支持 r2v， +// 仅支持 i2v（首帧/尾帧）。支持 r2v 的模型：默认 reference_image；若 metadata.content 已带 role 则保留（首尾帧/参考等多形态由调用方与上游约定）。 +const ( + doubaoImageRoleFirstFrame = "first_frame" + doubaoImageRoleLastFrame = "last_frame" + doubaoImageRoleReference = "reference_image" +) + +// doubaoSeedanceModelDisallowsR2V 为 true 时，带图请求不能走 r2v（多 reference），须显式 i2v + 首尾帧语义。 +func doubaoSeedanceModelDisallowsR2V(model string) bool { + m := strings.ToLower(strings.TrimSpace(model)) + return strings.Contains(m, "doubao-seedance-1-0-pro") || + strings.Contains(m, "doubao-seedance-1-5-pro") +} + +func normalizeDoubaoVideoContentRoles(model string, content *[]ContentItem) { + items := *content + if doubaoSeedanceModelDisallowsR2V(model) { + var out []ContentItem + imgIdx := 0 + for i := range items { + switch items[i].Type { + case "text": + items[i].Role = "" + out = append(out, items[i]) + case "image_url": + // i2v 最多保留 2 张（首帧+尾帧）；多图参考需 r2v，本系列模型不支持 + if imgIdx >= 2 { + imgIdx++ + continue + } + if imgIdx == 0 { + items[i].Role = doubaoImageRoleFirstFrame + } else { + items[i].Role = doubaoImageRoleLastFrame + } + imgIdx++ + out = append(out, items[i]) + default: + out = append(out, items[i]) + } + } + *content = out + return + } + for i := range items { + switch items[i].Type { + case "text": + items[i].Role = "" + case "image_url": + if strings.TrimSpace(items[i].Role) == "" { + items[i].Role = doubaoImageRoleReference + } + } + } +} + func (a *TaskAdaptor) convertToRequestPayload(req *relaycommon.TaskSubmitReq) (*requestPayload, error) { r := requestPayload{ Model: req.Model, @@ -290,6 +350,8 @@ func (a *TaskAdaptor) convertToRequestPayload(req *relaycommon.TaskSubmitReq) (* return nil, errors.Wrap(err, "unmarshal metadata failed") } + normalizeDoubaoContentImageURLs(&r.Content) + if sec, _ := strconv.Atoi(req.Seconds); sec > 0 { r.Duration = lo.ToPtr(dto.IntValue(sec)) } @@ -300,6 +362,11 @@ func (a *TaskAdaptor) convertToRequestPayload(req *relaycommon.TaskSubmitReq) (* Text: req.Prompt, }) + normalizeDoubaoVideoContentRoles(r.Model, &r.Content) + if req.HasImage() && doubaoSeedanceModelDisallowsR2V(r.Model) { + r.TaskType = "i2v" + } + return &r, nil } diff --git a/relay/channel/task/doubao/image_url_normalize.go b/relay/channel/task/doubao/image_url_normalize.go new file mode 100644 index 000000000000..9879c28f4dd8 --- /dev/null +++ b/relay/channel/task/doubao/image_url_normalize.go @@ -0,0 +1,81 @@ +package doubao + +import ( + "encoding/base64" + "strings" +) + +// normalizeDoubaoImageURLString 将裸 base64（无前缀）包装为 data URL，便于火山方舟 image_url.url 识别。 +// 已支持的格式原样返回：http(s)、data:、asset://、oss://。 +func normalizeDoubaoImageURLString(s string) string { + s = strings.TrimSpace(s) + if s == "" { + return s + } + low := strings.ToLower(s) + if strings.HasPrefix(low, "http://") || strings.HasPrefix(low, "https://") { + return s + } + if strings.HasPrefix(low, "data:") { + return s + } + if strings.HasPrefix(low, "asset://") || strings.HasPrefix(low, "oss://") { + return s + } + clean := stripDoubaoBase64Whitespace(s) + if len(clean) < 32 { + return s + } + decoded, err := base64.StdEncoding.DecodeString(clean) + if err != nil || len(decoded) < 8 { + return s + } + prefix := sniffDoubaoImageDataURLPrefix(decoded) + return prefix + clean +} + +func stripDoubaoBase64Whitespace(s string) string { + var b strings.Builder + b.Grow(len(s)) + for i := 0; i < len(s); i++ { + c := s[i] + if c == ' ' || c == '\n' || c == '\r' || c == '\t' { + continue + } + b.WriteByte(c) + } + return b.String() +} + +func sniffDoubaoImageDataURLPrefix(b []byte) string { + if len(b) >= 2 && b[0] == 0xFF && b[1] == 0xD8 { + return "data:image/jpeg;base64," + } + if len(b) >= 8 && b[0] == 0x89 && b[1] == 0x50 && b[2] == 0x4E && b[3] == 0x47 && b[4] == 0x0D && b[5] == 0x0A && b[6] == 0x1A && b[7] == 0x0A { + return "data:image/png;base64," + } + if len(b) >= 6 { + h := string(b[:6]) + if h == "GIF87a" || h == "GIF89a" { + return "data:image/gif;base64," + } + } + if len(b) >= 12 && string(b[0:4]) == "RIFF" && string(b[8:12]) == "WEBP" { + return "data:image/webp;base64," + } + return "data:image/jpeg;base64," +} + +func normalizeDoubaoContentImageURLs(content *[]ContentItem) { + if content == nil { + return + } + items := *content + for i := range items { + if items[i].Type != "image_url" || items[i].ImageURL == nil { + continue + } + items[i].ImageURL.URL = normalizeDoubaoImageURLString(items[i].ImageURL.URL) + } + *content = items +} diff --git a/relay/channel/task/doubao/image_url_normalize_test.go b/relay/channel/task/doubao/image_url_normalize_test.go new file mode 100644 index 000000000000..cc5e5924874d --- /dev/null +++ b/relay/channel/task/doubao/image_url_normalize_test.go @@ -0,0 +1,31 @@ +package doubao + +import ( + "encoding/base64" + "strings" + "testing" +) + +func TestNormalizeDoubaoImageURLString(t *testing.T) { + // 足够长的 JPEG 头 + 填充，使 base64 串长度 > 32 + jpegLike := make([]byte, 64) + jpegLike[0] = 0xff + jpegLike[1] = 0xd8 + for i := 2; i < len(jpegLike); i++ { + jpegLike[i] = byte(i) + } + b64 := base64.StdEncoding.EncodeToString(jpegLike) + out := normalizeDoubaoImageURLString(b64) + if !strings.HasPrefix(out, "data:image/jpeg;base64,") { + t.Fatalf("expected jpeg data URL prefix, got %.60q", out) + } + + httpsURL := "https://example.com/a.jpg" + if normalizeDoubaoImageURLString(httpsURL) != httpsURL { + t.Fatal("https URL should pass through") + } + data := "data:image/jpeg;base64,abc" + if normalizeDoubaoImageURLString(data) != data { + t.Fatal("data: URL should pass through") + } +} diff --git a/relay/mjproxy_handler.go b/relay/mjproxy_handler.go index ee48ca64b10b..409cbd991db5 100644 --- a/relay/mjproxy_handler.go +++ b/relay/mjproxy_handler.go @@ -232,6 +232,7 @@ func RelaySwapFace(c *gin.Context, info *relaycommon.RelayInfo) *dto.MidjourneyR tokenName := c.GetString("token_name") logContent := fmt.Sprintf("模型固定价格 %.2f，分组倍率 %.2f，操作 %s", priceData.ModelPrice, priceData.GroupRatioInfo.GroupRatio, constant.MjActionSwapFace) other := service.GenerateMjOtherInfo(info, priceData) + service.PublishBillingSnapshotForOpsLog(c, 0, 0, 0, priceData.Quota) model.RecordConsumeLog(c, info.UserId, model.RecordConsumeLogParams{ ChannelId: info.ChannelId, ModelName: modelName, @@ -538,6 +539,7 @@ func RelayMidjourneySubmit(c *gin.Context, relayInfo *relaycommon.RelayInfo) *dt tokenName := c.GetString("token_name") logContent := fmt.Sprintf("模型固定价格 %.2f，分组倍率 %.2f，操作 %s，ID %s", priceData.ModelPrice, priceData.GroupRatioInfo.GroupRatio, midjRequest.Action, midjResponse.Result) other := service.GenerateMjOtherInfo(relayInfo, priceData) + service.PublishBillingSnapshotForOpsLog(c, 0, 0, 0, priceData.Quota) model.RecordConsumeLog(c, relayInfo.UserId, model.RecordConsumeLogParams{ ChannelId: relayInfo.ChannelId, ModelName: modelName, diff --git a/relay/responses_handler.go b/relay/responses_handler.go index 58324aa7cec9..bdabec354f0a 100644 --- a/relay/responses_handler.go +++ b/relay/responses_handler.go @@ -70,6 +70,20 @@ func ResponsesHelper(c *gin.Context, info *relaycommon.RelayInfo) (newAPIError * return types.NewError(fmt.Errorf("invalid api type: %d", info.ApiType), types.ErrorCodeInvalidApiType, types.ErrOptionWithSkipRetry()) } adaptor.Init(info) + + passThroughGlobal := model_setting.GetGlobalSettings().PassThroughRequestEnabled + if info.RelayMode == relayconstant.RelayModeResponses && + !passThroughGlobal && + !info.ChannelSetting.PassThroughBodyEnabled && + service.ShouldResponsesUseChatCompletionsGlobal(info.ChannelId, info.ChannelType, info.OriginModelName) { + usage, newApiErr := responsesViaChatCompletions(c, info, adaptor, request) + if newApiErr != nil { + return newApiErr + } + service.PostTextConsumeQuota(c, info, usage, nil) + return nil + } + var requestBody io.Reader if model_setting.GetGlobalSettings().PassThroughRequestEnabled || info.ChannelSetting.PassThroughBodyEnabled { storage, err := common.GetBodyStorage(c) diff --git a/relay/responses_via_chat_completions.go b/relay/responses_via_chat_completions.go new file mode 100644 index 000000000000..3a97ed1a4761 --- /dev/null +++ b/relay/responses_via_chat_completions.go @@ -0,0 +1,96 @@ +package relay + +import ( + "bytes" + "net/http" + "strings" + + "github.com/QuantumNous/new-api/common" + "github.com/QuantumNous/new-api/dto" + "github.com/QuantumNous/new-api/relay/channel" + openaichannel "github.com/QuantumNous/new-api/relay/channel/openai" + relaycommon "github.com/QuantumNous/new-api/relay/common" + relayconstant "github.com/QuantumNous/new-api/relay/constant" + "github.com/QuantumNous/new-api/service" + "github.com/QuantumNous/new-api/types" + + "github.com/gin-gonic/gin" +) + +// responsesViaChatCompletions handles client /v1/responses calls by converting them +// into /v1/chat/completions requests for upstream, then re-emitting the upstream +// chat-completions reply as a Responses-format payload. +func responsesViaChatCompletions(c *gin.Context, info *relaycommon.RelayInfo, adaptor channel.Adaptor, request *dto.OpenAIResponsesRequest) (*dto.Usage, *types.NewAPIError) { + chatReq, err := service.ResponsesRequestToChatCompletionsRequest(request) + if err != nil { + return nil, types.NewErrorWithStatusCode(err, types.ErrorCodeInvalidRequest, http.StatusBadRequest, types.ErrOptionWithSkipRetry()) + } + info.AppendRequestConversion(types.RelayFormatOpenAI) + + savedRelayMode := info.RelayMode + savedRequestURLPath := info.RequestURLPath + defer func() { + info.RelayMode = savedRelayMode + info.RequestURLPath = savedRequestURLPath + }() + + info.RelayMode = relayconstant.RelayModeChatCompletions + info.RequestURLPath = "/v1/chat/completions" + + convertedRequest, err := adaptor.ConvertOpenAIRequest(c, info, chatReq) + if err != nil { + return nil, types.NewError(err, types.ErrorCodeConvertRequestFailed, types.ErrOptionWithSkipRetry()) + } + relaycommon.AppendRequestConversionFromRequest(info, convertedRequest) + + jsonData, err := common.Marshal(convertedRequest) + if err != nil { + return nil, types.NewError(err, types.ErrorCodeConvertRequestFailed, types.ErrOptionWithSkipRetry()) + } + + jsonData, err = relaycommon.RemoveDisabledFields(jsonData, info.ChannelOtherSettings, info.ChannelSetting.PassThroughBodyEnabled) + if err != nil { + return nil, types.NewError(err, types.ErrorCodeConvertRequestFailed, types.ErrOptionWithSkipRetry()) + } + + if len(info.ParamOverride) > 0 { + jsonData, err = relaycommon.ApplyParamOverrideWithRelayInfo(jsonData, info) + if err != nil { + return nil, newAPIErrorFromParamOverride(err) + } + } + + resp, err := adaptor.DoRequest(c, info, bytes.NewBuffer(jsonData)) + if err != nil { + return nil, types.NewOpenAIError(err, types.ErrorCodeDoRequestFailed, http.StatusInternalServerError) + } + if resp == nil { + return nil, types.NewOpenAIError(nil, types.ErrorCodeBadResponse, http.StatusInternalServerError) + } + + statusCodeMappingStr := c.GetString("status_code_mapping") + + httpResp := resp.(*http.Response) + info.IsStream = info.IsStream || strings.HasPrefix(httpResp.Header.Get("Content-Type"), "text/event-stream") + if httpResp.StatusCode != http.StatusOK { + newApiErr := service.RelayErrorHandler(c.Request.Context(), httpResp, false) + service.ResetStatusCode(newApiErr, statusCodeMappingStr) + return nil, newApiErr + } + + if info.IsStream { + usage, newApiErr := openaichannel.OaiChatToResponsesStreamHandler(c, info, httpResp) + if newApiErr != nil { + service.ResetStatusCode(newApiErr, statusCodeMappingStr) + return nil, newApiErr + } + return usage, nil + } + + usage, newApiErr := openaichannel.OaiChatToResponsesHandler(c, info, httpResp) + if newApiErr != nil { + service.ResetStatusCode(newApiErr, statusCodeMappingStr) + return nil, newApiErr + } + return usage, nil +} diff --git a/service/billing.go b/service/billing.go index 81daeed82c29..37626b576023 100644 --- a/service/billing.go +++ b/service/billing.go @@ -1,12 +1,16 @@ package service import ( + "context" "fmt" + "time" "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/types" "github.com/gin-gonic/gin" + "gorm.io/gorm" ) const ( @@ -14,10 +18,50 @@ const ( BillingSourceSubscription = "subscription" ) +// GetUserOrgDiscountRate 获取用户的企业折扣率 +// 如果用户不属于任何企业或没有折扣规则，返回1 +func GetUserOrgDiscountRate(userID int, modelName string) (float64, error) { + type userExt struct { + OrgID uint `json:"org_id"` + } + var ext userExt + err := model.DB.Table("lc_user_ext").Where("user_id = ?", userID).Select("org_id").Scan(&ext).Error + if err != nil { + return 1.0, err + } + if ext.OrgID == 0 { + return 1.0, nil + } + + type discountRule struct { + DiscountRate float64 `json:"discount_rate"` + } + var rule discountRule + now := time.Now().Unix() + err = model.DB.Table("lc_business_discount_rules"). + Where("org_id = ? AND model_name = ? AND effective_from <= ? AND (effective_to = 0 OR effective_to >= ?)", + ext.OrgID, modelName, now, now). + Order("effective_from DESC"). + Select("discount_rate"). + Scan(&rule).Error + if err != nil { + if err == gorm.ErrRecordNotFound { + return 1.0, nil + } + return 1.0, err + } + if rule.DiscountRate <= 0 { + return 1.0, nil + } + return rule.DiscountRate, nil +} + // PreConsumeBilling 根据用户计费偏好创建 BillingSession 并执行预扣费。 // 会话存储在 relayInfo.Billing 上，供后续 Settle / Refund 使用。 +// 企业折扣会在此处应用到 preConsumedQuota 上。 func PreConsumeBilling(c *gin.Context, preConsumedQuota int, relayInfo *relaycommon.RelayInfo) *types.NewAPIError { - session, apiErr := NewBillingSession(c, relayInfo, preConsumedQuota) + + session, apiErr := NewBillingSession(c, relayInfo, preConsumedQuota, 1.0) if apiErr != nil { return apiErr } @@ -32,38 +76,66 @@ func PreConsumeBilling(c *gin.Context, preConsumedQuota int, relayInfo *relaycom // SettleBilling 执行计费结算。如果 RelayInfo 上有 BillingSession 则通过 session 结算， // 否则回退到旧的 PostConsumeQuota 路径（兼容按次计费等场景）。 func SettleBilling(ctx *gin.Context, relayInfo *relaycommon.RelayInfo, actualQuota int) error { + return settleBilling(ctx, relayInfo, actualQuota, false) +} + +// SettleBillingDiscounted settles a quota value that has already had enterprise +// discount applied by the quota calculator. +func SettleBillingDiscounted(ctx *gin.Context, relayInfo *relaycommon.RelayInfo, actualQuota int) error { + return settleBilling(ctx, relayInfo, actualQuota, true) +} + +type discountedSettler interface { + SettleDiscounted(actualQuota int) error +} + +func settleBilling(ctx *gin.Context, relayInfo *relaycommon.RelayInfo, actualQuota int, alreadyDiscounted bool) error { if relayInfo.Billing != nil { + normalizedActual := actualQuota + if !alreadyDiscounted { + normalizedActual = NormalizeRecordedQuota(ctx, relayInfo, actualQuota) + } preConsumed := relayInfo.Billing.GetPreConsumedQuota() - delta := actualQuota - preConsumed + delta := normalizedActual - preConsumed if delta > 0 { logger.LogInfo(ctx, fmt.Sprintf("预扣费后补扣费：%s（实际消耗：%s，预扣费：%s）", logger.FormatQuota(delta), - logger.FormatQuota(actualQuota), + logger.FormatQuota(normalizedActual), logger.FormatQuota(preConsumed), )) } else if delta < 0 { logger.LogInfo(ctx, fmt.Sprintf("预扣费后返还扣费：%s（实际消耗：%s，预扣费：%s）", logger.FormatQuota(-delta), - logger.FormatQuota(actualQuota), + logger.FormatQuota(normalizedActual), logger.FormatQuota(preConsumed), )) } else { logger.LogInfo(ctx, fmt.Sprintf("预扣费与实际消耗一致，无需调整：%s（按次计费）", - logger.FormatQuota(actualQuota), + logger.FormatQuota(normalizedActual), )) } - if err := relayInfo.Billing.Settle(actualQuota); err != nil { - return err + if alreadyDiscounted { + settler, ok := relayInfo.Billing.(discountedSettler) + if !ok { + return fmt.Errorf("billing session does not support discounted settlement") + } + if err := settler.SettleDiscounted(actualQuota); err != nil { + return err + } + } else { + if err := relayInfo.Billing.Settle(actualQuota); err != nil { + return err + } } // 发送额度通知（订阅计费使用订阅剩余额度） - if actualQuota != 0 { + if normalizedActual != 0 { if relayInfo.BillingSource == BillingSourceSubscription { checkAndSendSubscriptionQuotaNotify(relayInfo) } else { - checkAndSendQuotaNotify(relayInfo, actualQuota-preConsumed, preConsumed) + checkAndSendQuotaNotify(relayInfo, delta, preConsumed) } } return nil @@ -76,3 +148,29 @@ func SettleBilling(ctx *gin.Context, relayInfo *relaycommon.RelayInfo, actualQuo } return nil } + +// NormalizeRecordedQuota 返回应当写入日志/任务记录的最终额度。 +// BillingSession 在结算时会对 actualQuota 应用企业折扣，但调用方保存任务/日志时 +// 仍可能持有未折扣的原始额度。这里统一转换为折后值，避免后续轮询以未折扣额度作为基线 +// 触发错误的补扣/退款。 +func NormalizeRecordedQuota(ctx context.Context, relayInfo *relaycommon.RelayInfo, actualQuota int) int { + if relayInfo == nil || actualQuota <= 0 { + return actualQuota + } + logger.LogInfo(ctx, fmt.Sprintf("NormalizeRecordedQuota: billing_type=%T, billing_nil=%v, actual=%d", + relayInfo.Billing, relayInfo.Billing == nil, actualQuota)) + session, ok := relayInfo.Billing.(*BillingSession) + if !ok || session == nil { + return actualQuota + } + if session.discountRate < 1.0 { + discountedActual := int(float64(actualQuota) * session.discountRate) + if discountedActual < 1 && actualQuota > 0 { + discountedActual = 1 + } + logger.LogInfo(ctx, fmt.Sprintf("用户 %d 应用企业折扣到记录额度：原始额度=%s，折扣率=%.4f，折后额度=%s", + relayInfo.UserId, logger.LogQuota(actualQuota), session.discountRate, logger.LogQuota(discountedActual))) + return discountedActual + } + return actualQuota +} diff --git a/service/billing_session.go b/service/billing_session.go index f24b68e55a80..94878470838f 100644 --- a/service/billing_session.go +++ b/service/billing_session.go @@ -25,11 +25,12 @@ import ( type BillingSession struct { relayInfo *relaycommon.RelayInfo funding FundingSource - preConsumedQuota int // 实际预扣额度（信任用户可能为 0） - tokenConsumed int // 令牌额度实际扣减量 - fundingSettled bool // funding.Settle 已成功，资金来源已提交 - settled bool // Settle 全部完成（资金 + 令牌） - refunded bool // Refund 已调用 + preConsumedQuota int // 实际预扣额度（信任用户可能为 0，已应用企业折扣） + tokenConsumed int // 令牌额度实际扣减量 + fundingSettled bool // funding.Settle 已成功，资金来源已提交 + settled bool // Settle 全部完成（资金 + 令牌） + refunded bool // Refund 已调用 + discountRate float64 // 企业折扣率（0-1之间，1表示无折扣） mu sync.Mutex } @@ -37,11 +38,28 @@ type BillingSession struct { // 资金来源和令牌额度分两步提交：若资金来源已提交但令牌调整失败， // 会标记 fundingSettled 防止 Refund 对已提交的资金来源执行退款。 func (s *BillingSession) Settle(actualQuota int) error { + return s.settle(actualQuota, false) +} + +// SettleDiscounted settles a quota value that already has the enterprise +// discount applied by the caller. +func (s *BillingSession) SettleDiscounted(actualQuota int) error { + return s.settle(actualQuota, true) +} + +func (s *BillingSession) settle(actualQuota int, alreadyDiscounted bool) error { s.mu.Lock() defer s.mu.Unlock() if s.settled { return nil } + if !alreadyDiscounted && s.discountRate < 1.0 { + discountedActual := int(float64(actualQuota) * s.discountRate) + if discountedActual < 1 && actualQuota > 0 { + discountedActual = 1 + } + actualQuota = discountedActual + } delta := actualQuota - s.preConsumedQuota if delta == 0 { s.settled = true @@ -86,10 +104,11 @@ func (s *BillingSession) Refund(c *gin.Context) { s.refunded = true s.mu.Unlock() - logger.LogInfo(c, fmt.Sprintf("用户 %d 请求失败, 返还预扣费（token_quota=%s, funding=%s）", + logger.LogInfo(c, fmt.Sprintf("用户 %d 请求失败, 返还预扣费（token_quota=%s, funding=%s, discount_rate=%.2f）", s.relayInfo.UserId, logger.FormatQuota(s.tokenConsumed), s.funding.Source(), + s.discountRate, )) // 复制需要的值到闭包中 @@ -110,6 +129,8 @@ func (s *BillingSession) Refund(c *gin.Context) { common.SysLog("error refunding token quota: " + err.Error()) } } + // 提交阶段失败不会写消费日志，也不会增加 users.used_quota。 + // 这里只回滚预扣的资金和令牌额度；异步任务已生成消费日志后的失败退款由 RefundTaskQuota 回退 used_quota。 }) } @@ -252,11 +273,36 @@ func (s *BillingSession) syncRelayInfo() { // --------------------------------------------------------------------------- // NewBillingSession 根据用户计费偏好创建 BillingSession，处理 subscription_first / wallet_first 的回退。 -func NewBillingSession(c *gin.Context, relayInfo *relaycommon.RelayInfo, preConsumedQuota int) (*BillingSession, *types.NewAPIError) { +// discountRate 是企业折扣率（0-1之间，1表示无折扣） +func NewBillingSession(c *gin.Context, relayInfo *relaycommon.RelayInfo, preConsumedQuota int, discountRate float64) (*BillingSession, *types.NewAPIError) { if relayInfo == nil { return nil, types.NewError(fmt.Errorf("relayInfo is nil"), types.ErrorCodeInvalidRequest, types.ErrOptionWithSkipRetry()) } + // 应用企业折扣 + modelName := relayInfo.OriginModelName + logger.LogInfo(c, fmt.Sprintf("开始应用企业折扣：用户ID=%d, 模型名称=%s", relayInfo.UserId, modelName)) + + if modelName != "" { + orgDiscountRate, err := getUserOrgDiscount(relayInfo.UserId, modelName) + logger.LogInfo(c, fmt.Sprintf("获取企业折扣结果：折扣率=%.4f, 错误=%v", orgDiscountRate, err)) + + if err == nil && orgDiscountRate != 1.0 { + originalQuota := preConsumedQuota + preConsumedQuota = int(float64(preConsumedQuota) * orgDiscountRate) + discountRate = orgDiscountRate + logger.LogInfo(c, fmt.Sprintf("用户 %d 模型 %s 应用企业折扣：原始预扣费 %s，折扣率 %.4f，折后预扣费 %s", + relayInfo.UserId, modelName, logger.FormatQuota(originalQuota), orgDiscountRate, logger.FormatQuota(preConsumedQuota))) + } else if err != nil { + logger.LogWarn(c, fmt.Sprintf("获取企业折扣失败：%v", err)) + } else { + discountRate = orgDiscountRate + logger.LogInfo(c, fmt.Sprintf("无企业折扣应用：折扣率=%.4f", orgDiscountRate)) + } + } else { + logger.LogInfo(c, fmt.Sprintf("用户 %d 模型名称为空，跳过企业折扣", relayInfo.UserId)) + } + pref := common.NormalizeBillingPreference(relayInfo.UserSetting.BillingPreference) // 钱包路径需要先检查用户额度 @@ -280,8 +326,9 @@ func NewBillingSession(c *gin.Context, relayInfo *relaycommon.RelayInfo, preCons relayInfo.UserQuota = userQuota session := &BillingSession{ - relayInfo: relayInfo, - funding: &WalletFunding{userId: relayInfo.UserId}, + relayInfo: relayInfo, + funding: &WalletFunding{userId: relayInfo.UserId}, + discountRate: discountRate, } if apiErr := session.preConsume(c, preConsumedQuota); apiErr != nil { return nil, apiErr @@ -295,7 +342,8 @@ func NewBillingSession(c *gin.Context, relayInfo *relaycommon.RelayInfo, preCons subConsume = 1 } session := &BillingSession{ - relayInfo: relayInfo, + relayInfo: relayInfo, + discountRate: discountRate, funding: &SubscriptionFunding{ requestId: relayInfo.RequestId, userId: relayInfo.UserId, diff --git a/service/openai_chat_responses_compat.go b/service/openai_chat_responses_compat.go index 2e887386339d..8780b1dd4bb2 100644 --- a/service/openai_chat_responses_compat.go +++ b/service/openai_chat_responses_compat.go @@ -9,6 +9,10 @@ func ChatCompletionsRequestToResponsesRequest(req *dto.GeneralOpenAIRequest) (*d return openaicompat.ChatCompletionsRequestToResponsesRequest(req) } +func ResponsesRequestToChatCompletionsRequest(req *dto.OpenAIResponsesRequest) (*dto.GeneralOpenAIRequest, error) { + return openaicompat.ResponsesRequestToChatCompletionsRequest(req) +} + func ResponsesResponseToChatCompletionsResponse(resp *dto.OpenAIResponsesResponse, id string) (*dto.OpenAITextResponse, *dto.Usage, error) { return openaicompat.ResponsesResponseToChatCompletionsResponse(resp, id) } diff --git a/service/openai_chat_responses_mode.go b/service/openai_chat_responses_mode.go index c66c33c9dc91..7910ee6e87c3 100644 --- a/service/openai_chat_responses_mode.go +++ b/service/openai_chat_responses_mode.go @@ -12,3 +12,11 @@ func ShouldChatCompletionsUseResponsesPolicy(policy model_setting.ChatCompletion func ShouldChatCompletionsUseResponsesGlobal(channelID int, channelType int, model string) bool { return openaicompat.ShouldChatCompletionsUseResponsesGlobal(channelID, channelType, model) } + +func ShouldResponsesUseChatCompletionsPolicy(policy model_setting.ResponsesToChatCompletionsPolicy, channelID int, channelType int, model string) bool { + return openaicompat.ShouldResponsesUseChatCompletionsPolicy(policy, channelID, channelType, model) +} + +func ShouldResponsesUseChatCompletionsGlobal(channelID int, channelType int, model string) bool { + return openaicompat.ShouldResponsesUseChatCompletionsGlobal(channelID, channelType, model) +} diff --git a/service/openaicompat/policy.go b/service/openaicompat/policy.go index b600b0fdc799..8408641c8a5b 100644 --- a/service/openaicompat/policy.go +++ b/service/openaicompat/policy.go @@ -17,3 +17,19 @@ func ShouldChatCompletionsUseResponsesGlobal(channelID int, channelType int, mod model, ) } + +func ShouldResponsesUseChatCompletionsPolicy(policy model_setting.ResponsesToChatCompletionsPolicy, channelID int, channelType int, model string) bool { + if !policy.IsChannelEnabled(channelID, channelType) { + return false + } + return matchAnyRegex(policy.ModelPatterns, model) +} + +func ShouldResponsesUseChatCompletionsGlobal(channelID int, channelType int, model string) bool { + return ShouldResponsesUseChatCompletionsPolicy( + model_setting.GetGlobalSettings().ResponsesToChatCompletionsPolicy, + channelID, + channelType, + model, + ) +} diff --git a/service/openaicompat/responses_request_to_chat.go b/service/openaicompat/responses_request_to_chat.go new file mode 100644 index 000000000000..357f06c1b061 --- /dev/null +++ b/service/openaicompat/responses_request_to_chat.go @@ -0,0 +1,372 @@ +package openaicompat + +import ( + "errors" + "fmt" + "strings" + + "github.com/QuantumNous/new-api/common" + "github.com/QuantumNous/new-api/dto" + "github.com/samber/lo" +) + +// ResponsesRequestToChatCompletionsRequest converts a Responses API request into a +// Chat Completions request. It is the inverse of ChatCompletionsRequestToResponsesRequest +// and is used when the upstream channel only supports /v1/chat/completions but the +// client called /v1/responses. +func ResponsesRequestToChatCompletionsRequest(req *dto.OpenAIResponsesRequest) (*dto.GeneralOpenAIRequest, error) { + if req == nil { + return nil, errors.New("request is nil") + } + if req.Model == "" { + return nil, errors.New("model is required") + } + + out := &dto.GeneralOpenAIRequest{ + Model: req.Model, + Stream: req.Stream, + Temperature: req.Temperature, + TopP: req.TopP, + User: req.User, + Store: req.Store, + Metadata: req.Metadata, + } + + if req.MaxOutputTokens != nil { + out.MaxTokens = lo.ToPtr(*req.MaxOutputTokens) + } + + // instructions => system/developer message at the very beginning. + if instructions, ok := extractStringFromRaw(req.Instructions); ok && strings.TrimSpace(instructions) != "" { + out.Messages = append(out.Messages, dto.Message{ + Role: out.GetSystemRoleName(), + Content: instructions, + }) + } + + // Convert input items into chat messages. + msgs, err := convertResponsesInputToMessages(req) + if err != nil { + return nil, err + } + out.Messages = append(out.Messages, msgs...) + + // tools + if len(req.Tools) > 0 { + var rawTools []map[string]any + if err := common.Unmarshal(req.Tools, &rawTools); err == nil { + tools := make([]dto.ToolCallRequest, 0, len(rawTools)) + for _, t := range rawTools { + typeStr, _ := t["type"].(string) + if typeStr != "function" { + // Only function tools have a direct chat-completions analogue. + continue + } + name, _ := t["name"].(string) + if name == "" { + if fn, ok := t["function"].(map[string]any); ok { + name, _ = fn["name"].(string) + } + } + if name == "" { + continue + } + desc, _ := t["description"].(string) + params := t["parameters"] + if params == nil { + if fn, ok := t["function"].(map[string]any); ok { + params = fn["parameters"] + if desc == "" { + desc, _ = fn["description"].(string) + } + } + } + tools = append(tools, dto.ToolCallRequest{ + Type: "function", + Function: dto.FunctionRequest{ + Name: name, + Description: desc, + Parameters: params, + }, + }) + } + if len(tools) > 0 { + out.Tools = tools + } + } + } + + // tool_choice: Responses {"type":"function","name":"x"} => Chat {"type":"function","function":{"name":"x"}} + if len(req.ToolChoice) > 0 { + if s, ok := extractStringFromRaw(req.ToolChoice); ok { + out.ToolChoice = s + } else { + var m map[string]any + if err := common.Unmarshal(req.ToolChoice, &m); err == nil && m != nil { + if t, _ := m["type"].(string); t == "function" { + if name, ok := m["name"].(string); ok && name != "" { + out.ToolChoice = map[string]any{ + "type": "function", + "function": map[string]any{"name": name}, + } + } else { + out.ToolChoice = m + } + } else { + out.ToolChoice = m + } + } + } + } + + if len(req.ParallelToolCalls) > 0 { + var b bool + if err := common.Unmarshal(req.ParallelToolCalls, &b); err == nil { + out.ParallelTooCalls = lo.ToPtr(b) + } + } + + // text.format => response_format + if rf := convertResponsesTextToChatResponseFormat(req.Text); rf != nil { + out.ResponseFormat = rf + } + + // reasoning => reasoning_effort (string form) + if req.Reasoning != nil && strings.TrimSpace(req.Reasoning.Effort) != "" { + out.ReasoningEffort = req.Reasoning.Effort + } + + if req.StreamOptions != nil { + out.StreamOptions = req.StreamOptions + } + + return out, nil +} + +func extractStringFromRaw(raw []byte) (string, bool) { + if len(raw) == 0 { + return "", false + } + if common.GetJsonType(raw) != "string" { + return "", false + } + var s string + if err := common.Unmarshal(raw, &s); err != nil { + return "", false + } + return s, true +} + +func convertResponsesTextToChatResponseFormat(raw []byte) *dto.ResponseFormat { + if len(raw) == 0 { + return nil + } + var wrapper map[string]any + if err := common.Unmarshal(raw, &wrapper); err != nil { + return nil + } + formatAny, ok := wrapper["format"] + if !ok { + return nil + } + format, ok := formatAny.(map[string]any) + if !ok { + return nil + } + typeStr, _ := format["type"].(string) + if typeStr == "" { + return nil + } + rf := &dto.ResponseFormat{Type: typeStr} + if typeStr == "json_schema" { + // Chat expects `json_schema` to be a sibling object containing schema/name/strict. + inner := map[string]any{} + for k, v := range format { + if k == "type" { + continue + } + inner[k] = v + } + if len(inner) > 0 { + if b, err := common.Marshal(inner); err == nil { + rf.JsonSchema = b + } + } + } + return rf +} + +func convertResponsesInputToMessages(req *dto.OpenAIResponsesRequest) ([]dto.Message, error) { + if req.Input == nil { + return nil, nil + } + + // A plain string input becomes a single user message. + if common.GetJsonType(req.Input) == "string" { + var s string + if err := common.Unmarshal(req.Input, &s); err != nil { + return nil, err + } + return []dto.Message{{Role: "user", Content: s}}, nil + } + + // Array of input items. + if common.GetJsonType(req.Input) != "array" { + return nil, fmt.Errorf("unsupported responses input type") + } + + var items []map[string]any + if err := common.Unmarshal(req.Input, &items); err != nil { + return nil, err + } + + var messages []dto.Message + // Buffer assistant tool_calls so they attach to the matching assistant turn. + for _, item := range items { + itemType, _ := item["type"].(string) + + switch itemType { + case "function_call": + callID, _ := item["call_id"].(string) + if callID == "" { + callID, _ = item["id"].(string) + } + name, _ := item["name"].(string) + args, _ := item["arguments"].(string) + tc := dto.ToolCallRequest{ + ID: callID, + Type: "function", + Function: dto.FunctionRequest{ + Name: name, + Arguments: args, + }, + } + // Attach to the previous assistant message when possible, otherwise create one. + if n := len(messages); n > 0 && messages[n-1].Role == "assistant" { + existing := messages[n-1].ParseToolCalls() + existing = append(existing, tc) + messages[n-1].SetToolCalls(existing) + } else { + m := dto.Message{Role: "assistant"} + m.SetNullContent() + m.SetToolCalls([]dto.ToolCallRequest{tc}) + messages = append(messages, m) + } + continue + + case "function_call_output": + callID, _ := item["call_id"].(string) + output := item["output"] + var contentStr string + switch v := output.(type) { + case string: + contentStr = v + case nil: + contentStr = "" + default: + if b, err := common.Marshal(v); err == nil { + contentStr = string(b) + } else { + contentStr = fmt.Sprintf("%v", v) + } + } + messages = append(messages, dto.Message{ + Role: "tool", + Content: contentStr, + ToolCallId: callID, + }) + continue + } + + // Default: a chat-style message with role + content. + role, _ := item["role"].(string) + if role == "" { + continue + } + msg := dto.Message{Role: role} + contentAny, hasContent := item["content"] + if !hasContent || contentAny == nil { + msg.SetStringContent("") + messages = append(messages, msg) + continue + } + + switch content := contentAny.(type) { + case string: + msg.SetStringContent(content) + case []any: + parts := convertResponsesContentParts(content, role) + if len(parts) == 1 && parts[0].Type == dto.ContentTypeText { + msg.SetStringContent(parts[0].Text) + } else if len(parts) > 0 { + msg.SetMediaContent(parts) + } else { + msg.SetStringContent("") + } + default: + if b, err := common.Marshal(content); err == nil { + msg.SetStringContent(string(b)) + } + } + messages = append(messages, msg) + } + + return messages, nil +} + +func convertResponsesContentParts(parts []any, role string) []dto.MediaContent { + out := make([]dto.MediaContent, 0, len(parts)) + for _, partAny := range parts { + part, ok := partAny.(map[string]any) + if !ok { + continue + } + partType, _ := part["type"].(string) + switch partType { + case "input_text", "output_text", "text": + text, _ := part["text"].(string) + out = append(out, dto.MediaContent{Type: dto.ContentTypeText, Text: text}) + case "input_image": + urlAny := part["image_url"] + var url string + switch v := urlAny.(type) { + case string: + url = v + case map[string]any: + url, _ = v["url"].(string) + } + if url != "" { + out = append(out, dto.MediaContent{ + Type: dto.ContentTypeImageURL, + ImageUrl: &dto.MessageImageUrl{Url: url}, + }) + } + case "input_audio": + out = append(out, dto.MediaContent{ + Type: dto.ContentTypeInputAudio, + InputAudio: part["input_audio"], + }) + case "input_file": + out = append(out, dto.MediaContent{ + Type: dto.ContentTypeFile, + File: part["file"], + }) + case "input_video": + urlAny := part["video_url"] + var url string + switch v := urlAny.(type) { + case string: + url = v + case map[string]any: + url, _ = v["url"].(string) + } + if url != "" { + out = append(out, dto.MediaContent{ + Type: dto.ContentTypeVideoUrl, + VideoUrl: &dto.MessageVideoUrl{Url: url}, + }) + } + } + } + return out +} diff --git a/service/ops_billing_context.go b/service/ops_billing_context.go new file mode 100644 index 000000000000..0a040b2dc248 --- /dev/null +++ b/service/ops_billing_context.go @@ -0,0 +1,23 @@ +package service + +import ( + "github.com/QuantumNous/new-api/common" + "github.com/QuantumNous/new-api/constant" + "github.com/gin-gonic/gin" +) + +const opsBillingSnapshotFrom = "new-api" + +// PublishBillingSnapshotForOpsLog stores reconciled token usage and consume quota on the Gin context +// so outer ops middleware (e.g. Lynxton relay_request_summary) can emit them after c.Next(). +// Safe with zeros when totals are unknown. Last call in a request wins. +func PublishBillingSnapshotForOpsLog(c *gin.Context, promptTokens, completionTokens, totalTokens, consumeQuota int) { + if c == nil { + return + } + common.SetContextKey(c, constant.ContextKeyOpsBillingPromptTokens, promptTokens) + common.SetContextKey(c, constant.ContextKeyOpsBillingCompletionTokens, completionTokens) + common.SetContextKey(c, constant.ContextKeyOpsBillingTotalTokens, totalTokens) + common.SetContextKey(c, constant.ContextKeyOpsBillingConsumeQuota, consumeQuota) + common.SetContextKey(c, constant.ContextKeyOpsBillingSnapshotFrom, opsBillingSnapshotFrom) +} diff --git a/service/quota.go b/service/quota.go index 4150c44434bb..355ae4d57862 100644 --- a/service/quota.go +++ b/service/quota.go @@ -37,6 +37,7 @@ type QuotaInfo struct { ModelPrice float64 ModelRatio float64 GroupRatio float64 + UserId int } func hasCustomModelRatio(modelName string, currentRatio float64) bool { @@ -54,6 +55,15 @@ func calculateAudioQuota(info QuotaInfo) int { groupRatio := decimal.NewFromFloat(info.GroupRatio) quota := modelPrice.Mul(quotaPerUnit).Mul(groupRatio) + + // 应用企业折扣 + orgDiscountRate := 1.0 + discountRate, err := getUserOrgDiscount(info.UserId, info.ModelName) + if err == nil { + orgDiscountRate = discountRate + quota = quota.Mul(decimal.NewFromFloat(orgDiscountRate)) + } + return int(quota.IntPart()) } @@ -78,6 +88,14 @@ func calculateAudioQuota(info QuotaInfo) int { quota = quota.Mul(ratio) + // 应用企业折扣 + orgDiscountRate := 1.0 + discountRate, err := getUserOrgDiscount(info.UserId, info.ModelName) + if err == nil { + orgDiscountRate = discountRate + quota = quota.Mul(decimal.NewFromFloat(orgDiscountRate)) + } + // If ratio is not zero and quota is less than or equal to zero, set quota to 1 if !ratio.IsZero() && quota.LessThanOrEqual(decimal.Zero) { quota = decimal.NewFromInt(1) @@ -134,6 +152,7 @@ func PreWssConsumeQuota(ctx *gin.Context, relayInfo *relaycommon.RelayInfo, usag UsePrice: relayInfo.UsePrice, ModelRatio: modelRatio, GroupRatio: actualGroupRatio, + UserId: relayInfo.UserId, } quota := calculateAudioQuota(quotaInfo) @@ -219,6 +238,14 @@ func PostWssConsumeQuota(ctx *gin.Context, relayInfo *relaycommon.RelayInfo, mod } other := GenerateWssOtherInfo(ctx, relayInfo, usage, modelRatio, groupRatio, completionRatio.InexactFloat64(), audioRatio.InexactFloat64(), audioCompletionRatio.InexactFloat64(), modelPrice, relayInfo.PriceData.GroupRatioInfo.GroupSpecialRatio) + totalTok := usage.TotalTokens + if totalTok <= 0 { + totalTok = usage.InputTokens + usage.OutputTokens + if totalTok < 0 { + totalTok = 0 + } + } + PublishBillingSnapshotForOpsLog(ctx, usage.InputTokens, usage.OutputTokens, totalTok, quota) model.RecordConsumeLog(ctx, relayInfo.UserId, model.RecordConsumeLogParams{ ChannelId: relayInfo.ChannelId, PromptTokens: usage.InputTokens, @@ -288,6 +315,7 @@ func PostAudioConsumeQuota(ctx *gin.Context, relayInfo *relaycommon.RelayInfo, u UsePrice: usePrice, ModelRatio: modelRatio, GroupRatio: groupRatio, + UserId: relayInfo.UserId, } quota := calculateAudioQuota(quotaInfo) @@ -314,7 +342,7 @@ func PostAudioConsumeQuota(ctx *gin.Context, relayInfo *relaycommon.RelayInfo, u model.UpdateChannelUsedQuota(relayInfo.ChannelId, quota) } - if err := SettleBilling(ctx, relayInfo, quota); err != nil { + if err := SettleBillingDiscounted(ctx, relayInfo, quota); err != nil { logger.LogError(ctx, "error settling billing: "+err.Error()) } @@ -324,6 +352,7 @@ func PostAudioConsumeQuota(ctx *gin.Context, relayInfo *relaycommon.RelayInfo, u } other := GenerateAudioOtherInfo(ctx, relayInfo, usage, modelRatio, groupRatio, completionRatio.InexactFloat64(), audioRatio.InexactFloat64(), audioCompletionRatio.InexactFloat64(), modelPrice, relayInfo.PriceData.GroupRatioInfo.GroupSpecialRatio) + PublishBillingSnapshotForOpsLog(ctx, usage.PromptTokens, usage.CompletionTokens, usage.TotalTokens, quota) model.RecordConsumeLog(ctx, relayInfo.UserId, model.RecordConsumeLogParams{ ChannelId: relayInfo.ChannelId, PromptTokens: usage.PromptTokens, diff --git a/service/task_billing.go b/service/task_billing.go index 6cf7a965c8eb..4d3c98162ba8 100644 --- a/service/task_billing.go +++ b/service/task_billing.go @@ -4,6 +4,7 @@ import ( "context" "fmt" "strings" + "time" "github.com/QuantumNous/new-api/common" "github.com/QuantumNous/new-api/constant" @@ -50,6 +51,8 @@ func LogTaskConsumption(c *gin.Context, info *relaycommon.RelayInfo) { other["is_model_mapped"] = true other["upstream_model_name"] = info.UpstreamModelName } + est := info.GetEstimatePromptTokens() + PublishBillingSnapshotForOpsLog(c, est, 0, est, info.PriceData.Quota) model.RecordConsumeLog(c, info.UserId, model.RecordConsumeLogParams{ ChannelId: info.ChannelId, ModelName: info.OriginModelName, @@ -147,6 +150,32 @@ func taskModelName(task *model.Task) string { return task.Properties.OriginModelName } +func emitAsyncBillingOps(ctx context.Context, task *model.Task, event string, preConsumed, actualQuota, quotaDelta int, reason string, billingTotalTokens int) { + if common.EmitAsyncBillingOpsLog == nil { + return + } + kv := map[string]any{ + "log_type": "model_relay", + "event": event, + "billing_snapshot_repo": "new-api", + "task_id": task.TaskID, + "upstream_task_id": task.GetUpstreamTaskID(), + "user_id": task.UserId, + "channel_id": task.ChannelId, + "platform": string(task.Platform), + "action": task.Action, + "model_name": taskModelName(task), + "billing_pre_consumed_quota": preConsumed, + "billing_consume_quota_actual": actualQuota, + "billing_quota_delta": quotaDelta, + "settle_reason": reason, + } + if billingTotalTokens >= 0 { + kv["billing_total_tokens"] = billingTotalTokens + } + common.EmitAsyncBillingOpsLog(ctx, "task_async_billing", kv) +} + // RefundTaskQuota 统一的任务失败退款逻辑。 // 当异步任务失败时，将预扣的 quota 退还给用户（支持钱包和订阅），并退还令牌额度。 func RefundTaskQuota(ctx context.Context, task *model.Task, reason string) { @@ -164,7 +193,11 @@ func RefundTaskQuota(ctx context.Context, task *model.Task, reason string) { // 2. 退还令牌额度 taskAdjustTokenQuota(ctx, task, -quota) - // 3. 记录日志 + // 3. 更新用户和渠道的使用额度 + model.UpdateUserUsedQuotaAndRequestCount(task.UserId, -quota) + model.UpdateChannelUsedQuota(task.ChannelId, -quota) + + // 4. 记录日志 other := taskBillingOther(task) other["task_id"] = task.TaskID other["reason"] = reason @@ -179,21 +212,48 @@ func RefundTaskQuota(ctx context.Context, task *model.Task, reason string) { Group: task.Group, Other: other, }) + emitAsyncBillingOps(ctx, task, "task_async_refund", quota, 0, -quota, reason, -1) } // RecalculateTaskQuota 通用的异步差额结算。 // actualQuota 是任务完成后的实际应扣额度，与预扣额度 (task.Quota) 做差额结算。 // reason 用于日志记录（例如 "token重算" 或 "adaptor调整"）。 -func RecalculateTaskQuota(ctx context.Context, task *model.Task, actualQuota int, reason string) { +// billingTotalTokens 为上游 total_tokens；无则传 -1 表示不在 ops 日志中输出该字段。 +func RecalculateTaskQuota(ctx context.Context, task *model.Task, actualQuota int, reason string, billingTotalTokens int) { if actualQuota <= 0 { return } + + // 应用企业折扣（token重算场景已在 RecalculateTaskQuotaByTokens 中应用） + orgDiscountRate := 1.0 + isTokenRecalc := strings.Contains(reason, "token重算") + if !isTokenRecalc { + discountRate, err := getUserOrgDiscount(task.UserId, taskModelName(task)) + if err != nil { + logger.LogInfo(ctx, fmt.Sprintf("任务 %s 查询企业折扣失败：%s，使用折扣率=1.0", task.TaskID, err.Error())) + } else { + orgDiscountRate = discountRate + logger.LogInfo(ctx, fmt.Sprintf("任务 %s 查询企业折扣成功：折扣率=%.4f", task.TaskID, orgDiscountRate)) + } + actualQuota = int(float64(actualQuota) * orgDiscountRate) + } + + if orgDiscountRate != 1.0 { + logger.LogInfo(ctx, fmt.Sprintf("任务 %s 应用企业折扣：原始额度=%s，折扣率=%.4f，折后额度=%s", + task.TaskID, + logger.LogQuota(actualQuota/int(orgDiscountRate)), + orgDiscountRate, + logger.LogQuota(actualQuota), + )) + } + preConsumedQuota := task.Quota quotaDelta := actualQuota - preConsumedQuota if quotaDelta == 0 { logger.LogInfo(ctx, fmt.Sprintf("任务 %s 预扣费准确（%s，%s）", task.TaskID, logger.LogQuota(actualQuota), reason)) + emitAsyncBillingOps(ctx, task, "task_async_billing_precise", preConsumedQuota, actualQuota, 0, reason, billingTotalTokens) return } @@ -226,11 +286,16 @@ func RecalculateTaskQuota(ctx context.Context, task *model.Task, actualQuota int } else { logType = model.LogTypeRefund logQuota = -quotaDelta + model.UpdateUserUsedQuotaAndRequestCount(task.UserId, quotaDelta) + model.UpdateChannelUsedQuota(task.ChannelId, quotaDelta) } other := taskBillingOther(task) other["task_id"] = task.TaskID other["pre_consumed_quota"] = preConsumedQuota other["actual_quota"] = actualQuota + if orgDiscountRate != 1.0 { + other["org_discount_rate"] = orgDiscountRate + } model.RecordTaskBillingLog(model.RecordTaskBillingLogParams{ UserId: task.UserId, LogType: logType, @@ -242,6 +307,7 @@ func RecalculateTaskQuota(ctx context.Context, task *model.Task, actualQuota int Group: task.Group, Other: other, }) + emitAsyncBillingOps(ctx, task, "task_async_billing_settle", preConsumedQuota, actualQuota, quotaDelta, reason, billingTotalTokens) } // RecalculateTaskQuotaByTokens 根据实际 token 消耗重新计费（异步差额结算）。 @@ -296,6 +362,63 @@ func RecalculateTaskQuotaByTokens(ctx context.Context, task *model.Task, totalTo // 计算实际应扣费额度: totalTokens * modelRatio * groupRatio * otherMultiplier actualQuota := int(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) + // 应用企业折扣 + orgDiscountRate := 1.0 + discountRate, err := getUserOrgDiscount(task.UserId, modelName) + if err == nil { + orgDiscountRate = discountRate + actualQuota = int(float64(actualQuota) * orgDiscountRate) + } + + reason := fmt.Sprintf("token重算：tokens=%d, modelRatio=%.2f, groupRatio=%.2f, otherMultiplier=%.4f, orgDiscountRate=%.4f", + totalTokens, modelRatio, finalGroupRatio, otherMultiplier, orgDiscountRate) + RecalculateTaskQuota(ctx, task, actualQuota, reason, totalTokens) +} + +// BusinessDiscountRule 企业折扣规则模型 +type BusinessDiscountRule struct { + ID uint `gorm:"primaryKey"` + OrgID int `gorm:"column:org_id;not null"` + ModelName string `gorm:"column:model_name;not null"` + DiscountRate float64 `gorm:"column:discount_rate;not null"` + EffectiveFrom int64 `gorm:"column:effective_from;not null"` + EffectiveTo int64 `gorm:"column:effective_to;default:0"` +} + +func (BusinessDiscountRule) TableName() string { return "lc_business_discount_rules" } + +// getUserOrgDiscount 获取用户的企业折扣率 +// 如果用户不是企业成员，返回1；如果是企业成员但没有对应模型的折扣，也返回1 +func getUserOrgDiscount(userID int, modelName string) (float64, error) { + // 查询用户的企业ID + var userExt struct { + OrgID uint `gorm:"column:org_id"` + } + err := model.DB.Table("lc_user_ext").Where("user_id = ?", userID).Select("org_id").Scan(&userExt).Error + if err != nil { + logger.LogWarn(context.Background(), fmt.Sprintf("查询用户企业ID失败 userID=%d: %s", userID, err.Error())) + return 1.0, err + } + + // 如果用户不属于任何企业，返回折扣率1 + if userExt.OrgID == 0 { + logger.LogInfo(context.Background(), fmt.Sprintf("用户 %d 不属于任何企业，跳过企业折扣", userID)) + return 1.0, nil + } + + // 查询该企业该模型的有效折扣规则 + now := time.Now().Unix() + var rule BusinessDiscountRule + err = model.DB.Where("org_id = ? AND model_name = ? AND effective_from <= ? AND (effective_to = 0 OR effective_to >= ?)", + userExt.OrgID, modelName, now, now).Order("effective_from DESC").First(&rule).Error + if err != nil { + logger.LogInfo(context.Background(), fmt.Sprintf("用户 %d 企业 %d 模型 %s 查询折扣规则失败: %s", + userID, userExt.OrgID, modelName, err.Error())) + return 1.0, nil + } + + logger.LogInfo(context.Background(), fmt.Sprintf("用户 %d 企业 %d 模型 %s 折扣率: %.4f", + userID, userExt.OrgID, modelName, rule.DiscountRate)) + + return rule.DiscountRate, nil } diff --git a/service/task_billing_test.go b/service/task_billing_test.go index 39cb8f1da1aa..9d88b6ea54b4 100644 --- a/service/task_billing_test.go +++ b/service/task_billing_test.go @@ -184,6 +184,31 @@ func countLogs(t *testing.T) int64 { return count } +func TestNormalizeRecordedQuota_NoBillingSession(t *testing.T) { + relayInfo := &relaycommon.RelayInfo{} + assert.Equal(t, 1000, NormalizeRecordedQuota(context.Background(), relayInfo, 1000)) +} + +func TestNormalizeRecordedQuota_WithDiscount(t *testing.T) { + relayInfo := &relaycommon.RelayInfo{ + UserId: 123, + Billing: &BillingSession{ + discountRate: 0.5, + }, + } + assert.Equal(t, 500, NormalizeRecordedQuota(context.Background(), relayInfo, 1000)) +} + +func TestNormalizeRecordedQuota_MinOne(t *testing.T) { + relayInfo := &relaycommon.RelayInfo{ + UserId: 123, + Billing: &BillingSession{ + discountRate: 0.1, + }, + } + assert.Equal(t, 1, NormalizeRecordedQuota(context.Background(), relayInfo, 1)) +} + // =========================================================================== // RefundTaskQuota tests // =========================================================================== @@ -308,7 +333,7 @@ func TestRecalculate_PositiveDelta(t *testing.T) { task := makeTask(userID, channelID, preConsumed, tokenID, BillingSourceWallet, 0) - RecalculateTaskQuota(ctx, task, actualQuota, "adaptor adjustment") + RecalculateTaskQuota(ctx, task, actualQuota, "adaptor adjustment", -1) // User quota should decrease by the delta (1000 additional charge) assert.Equal(t, initQuota-(actualQuota-preConsumed), getUserQuota(t, userID)) @@ -341,7 +366,7 @@ func TestRecalculate_NegativeDelta(t *testing.T) { task := makeTask(userID, channelID, preConsumed, tokenID, BillingSourceWallet, 0) - RecalculateTaskQuota(ctx, task, actualQuota, "adaptor adjustment") + RecalculateTaskQuota(ctx, task, actualQuota, "adaptor adjustment", -1) // User quota should increase by abs(delta) = 2000 (refund overpayment) assert.Equal(t, initQuota+(preConsumed-actualQuota), getUserQuota(t, userID)) @@ -370,7 +395,7 @@ func TestRecalculate_ZeroDelta(t *testing.T) { task := makeTask(userID, 0, preConsumed, 0, BillingSourceWallet, 0) - RecalculateTaskQuota(ctx, task, preConsumed, "exact match") + RecalculateTaskQuota(ctx, task, preConsumed, "exact match", -1) // No change to user quota assert.Equal(t, initQuota, getUserQuota(t, userID)) @@ -390,7 +415,7 @@ func TestRecalculate_ActualQuotaZero(t *testing.T) { task := makeTask(userID, 0, 5000, 0, BillingSourceWallet, 0) - RecalculateTaskQuota(ctx, task, 0, "zero actual") + RecalculateTaskQuota(ctx, task, 0, "zero actual", -1) // No change (early return) assert.Equal(t, initQuota, getUserQuota(t, userID)) @@ -414,7 +439,7 @@ func TestRecalculate_Subscription_NegativeDelta(t *testing.T) { task := makeTask(userID, channelID, preConsumed, tokenID, BillingSourceSubscription, subID) - RecalculateTaskQuota(ctx, task, actualQuota, "subscription over-charge") + RecalculateTaskQuota(ctx, task, actualQuota, "subscription over-charge", -1) // Subscription used should decrease by delta (refund 3000) assert.Equal(t, subUsed-int64(preConsumed-actualQuota), getSubscriptionUsed(t, subID)) @@ -476,7 +501,7 @@ func simulatePollBilling(ctx context.Context, task *model.Task, newStatus model. } if shouldSettle && actualQuota > 0 { - RecalculateTaskQuota(ctx, task, actualQuota, "test settle") + RecalculateTaskQuota(ctx, task, actualQuota, "test settle", -1) } if shouldRefund { RefundTaskQuota(ctx, task, task.FailReason) diff --git a/service/task_polling.go b/service/task_polling.go index dc85e579e8cc..07b5577f75e7 100644 --- a/service/task_polling.go +++ b/service/task_polling.go @@ -548,7 +548,7 @@ func settleTaskBillingOnComplete(ctx context.Context, adaptor TaskPollingAdaptor } // 1. 优先让 adaptor 决定最终额度 if actualQuota := adaptor.AdjustBillingOnComplete(task, taskResult); actualQuota > 0 { - RecalculateTaskQuota(ctx, task, actualQuota, "adaptor计费调整") + RecalculateTaskQuota(ctx, task, actualQuota, "adaptor计费调整", -1) return } // 2. 回退到 token 重算 diff --git a/service/text_quota.go b/service/text_quota.go index 8caee8f28799..a28b18b391ea 100644 --- a/service/text_quota.go +++ b/service/text_quota.go @@ -247,6 +247,14 @@ func calculateTextQuotaSummary(ctx *gin.Context, relayInfo *relaycommon.RelayInf quotaCalculateDecimal = quotaCalculateDecimal.Add(audioInputQuota) quotaCalculateDecimal = quotaCalculateDecimal.Add(dImageGenerationCallQuota) + // 应用企业折扣 + orgDiscountRate := 1.0 + discountRate, err := getUserOrgDiscount(relayInfo.UserId, relayInfo.OriginModelName) + if err == nil { + orgDiscountRate = discountRate + quotaCalculateDecimal = quotaCalculateDecimal.Mul(decimal.NewFromFloat(orgDiscountRate)) + } + if len(relayInfo.PriceData.OtherRatios) > 0 { for _, otherRatio := range relayInfo.PriceData.OtherRatios { quotaCalculateDecimal = quotaCalculateDecimal.Mul(decimal.NewFromFloat(otherRatio)) @@ -264,6 +272,15 @@ func calculateTextQuotaSummary(ctx *gin.Context, relayInfo *relaycommon.RelayInf quotaCalculateDecimal = quotaCalculateDecimal.Add(dFileSearchQuota) quotaCalculateDecimal = quotaCalculateDecimal.Add(audioInputQuota) quotaCalculateDecimal = quotaCalculateDecimal.Add(dImageGenerationCallQuota) + + // 应用企业折扣 + orgDiscountRate := 1.0 + discountRate, err := getUserOrgDiscount(relayInfo.UserId, relayInfo.OriginModelName) + if err == nil { + orgDiscountRate = discountRate + quotaCalculateDecimal = quotaCalculateDecimal.Mul(decimal.NewFromFloat(orgDiscountRate)) + } + if len(relayInfo.PriceData.OtherRatios) > 0 { for _, otherRatio := range relayInfo.PriceData.OtherRatios { quotaCalculateDecimal = quotaCalculateDecimal.Mul(decimal.NewFromFloat(otherRatio)) @@ -327,7 +344,7 @@ func PostTextConsumeQuota(ctx *gin.Context, relayInfo *relaycommon.RelayInfo, us model.UpdateChannelUsedQuota(relayInfo.ChannelId, summary.Quota) } - if err := SettleBilling(ctx, relayInfo, summary.Quota); err != nil { + if err := SettleBillingDiscounted(ctx, relayInfo, summary.Quota); err != nil { logger.LogError(ctx, "error settling billing: "+err.Error()) } @@ -413,6 +430,7 @@ func PostTextConsumeQuota(ctx *gin.Context, relayInfo *relaycommon.RelayInfo, us other["input_tokens_total"] = usage.InputTokens } + PublishBillingSnapshotForOpsLog(ctx, summary.PromptTokens, summary.CompletionTokens, summary.TotalTokens, summary.Quota) model.RecordConsumeLog(ctx, relayInfo.UserId, model.RecordConsumeLogParams{ ChannelId: relayInfo.ChannelId, PromptTokens: summary.PromptTokens, diff --git a/service/violation_fee.go b/service/violation_fee.go index 45508856135d..e60ad630fba7 100644 --- a/service/violation_fee.go +++ b/service/violation_fee.go @@ -147,6 +147,7 @@ func ChargeViolationFeeIfNeeded(ctx *gin.Context, relayInfo *relaycommon.RelayIn "violation_fee_marker": CSAMViolationMarker, } + PublishBillingSnapshotForOpsLog(ctx, 0, 0, 0, feeQuota) model.RecordConsumeLog(ctx, relayInfo.UserId, model.RecordConsumeLogParams{ ChannelId: relayInfo.ChannelId, ModelName: relayInfo.OriginModelName, diff --git a/setting/model_setting/global.go b/setting/model_setting/global.go index d0c4d312893c..b4ba7704f6d7 100644 --- a/setting/model_setting/global.go +++ b/setting/model_setting/global.go @@ -32,10 +32,39 @@ func (p ChatCompletionsToResponsesPolicy) IsChannelEnabled(channelID int, channe return false } +// ResponsesToChatCompletionsPolicy 用于配置客户端调用 /v1/responses +// 但上游通道只支持 /v1/chat/completions 时的转换策略。 +// 字段语义与 ChatCompletionsToResponsesPolicy 一致。 +type ResponsesToChatCompletionsPolicy struct { + Enabled bool `json:"enabled"` + AllChannels bool `json:"all_channels"` + ChannelIDs []int `json:"channel_ids,omitempty"` + ChannelTypes []int `json:"channel_types,omitempty"` + ModelPatterns []string `json:"model_patterns,omitempty"` +} + +func (p ResponsesToChatCompletionsPolicy) IsChannelEnabled(channelID int, channelType int) bool { + if !p.Enabled { + return false + } + if p.AllChannels { + return true + } + + if channelID > 0 && len(p.ChannelIDs) > 0 && slices.Contains(p.ChannelIDs, channelID) { + return true + } + if channelType > 0 && len(p.ChannelTypes) > 0 && slices.Contains(p.ChannelTypes, channelType) { + return true + } + return false +} + type GlobalSettings struct { PassThroughRequestEnabled bool `json:"pass_through_request_enabled"` ThinkingModelBlacklist []string `json:"thinking_model_blacklist"` ChatCompletionsToResponsesPolicy ChatCompletionsToResponsesPolicy `json:"chat_completions_to_responses_policy"` + ResponsesToChatCompletionsPolicy ResponsesToChatCompletionsPolicy `json:"responses_to_chat_completions_policy"` } // 默认配置 @@ -49,6 +78,11 @@ var defaultOpenaiSettings = GlobalSettings{ Enabled: false, AllChannels: true, }, + ResponsesToChatCompletionsPolicy: ResponsesToChatCompletionsPolicy{ + Enabled: true, + AllChannels: true, + ModelPatterns: []string{".*"}, + }, } // 全局实例 diff --git a/setting/ratio_setting/exposed_cache.go b/setting/ratio_setting/exposed_cache.go index c88216fcb015..bcf2eb687af4 100644 --- a/setting/ratio_setting/exposed_cache.go +++ b/setting/ratio_setting/exposed_cache.go @@ -47,6 +47,9 @@ func GetExposedData() gin.H { "cache_ratio": GetCacheRatioCopy(), "create_cache_ratio": GetCreateCacheRatioCopy(), "model_price": GetModelPriceCopy(), + "model_display_name": GetModelDisplayNameCopy(), + "model_modalities": GetModelModalitiesCopy(), + "model_img": GetModelImgCopy(), } exposedData.Store(&exposedCache{ data: newData, diff --git a/setting/ratio_setting/model_ratio.go b/setting/ratio_setting/model_ratio.go index 7556fd9482c7..1213b7b67f5e 100644 --- a/setting/ratio_setting/model_ratio.go +++ b/setting/ratio_setting/model_ratio.go @@ -331,6 +331,9 @@ var defaultAudioCompletionRatio = map[string]float64{ var modelPriceMap = types.NewRWMap[string, float64]() var modelRatioMap = types.NewRWMap[string, float64]() var completionRatioMap = types.NewRWMap[string, float64]() +var modelDisplayNameMap = types.NewRWMap[string, string]() +var modelModalitiesMap = types.NewRWMap[string, string]() +var modelImgMap = types.NewRWMap[string, string]() var defaultCompletionRatio = map[string]float64{ "gpt-4-gizmo-*": 2, @@ -694,6 +697,62 @@ func UpdateAudioCompletionRatioByJSONString(jsonStr string) error { return types.LoadFromJsonStringWithCallback(audioCompletionRatioMap, jsonStr, InvalidateExposedDataCache) } +func ModelDisplayName2JSONString() string { + return modelDisplayNameMap.MarshalJSONString() +} + +func UpdateModelDisplayNameByJSONString(jsonStr string) error { + return types.LoadFromJsonStringWithCallback(modelDisplayNameMap, jsonStr, InvalidateExposedDataCache) +} + +func GetModelDisplayName(name string) string { + name = FormatMatchingModelName(name) + if displayName, ok := modelDisplayNameMap.Get(name); ok { + return displayName + } + return "" +} + +func ModelModalities2JSONString() string { + return modelModalitiesMap.MarshalJSONString() +} + +func UpdateModelModalitiesByJSONString(jsonStr string) error { + return types.LoadFromJsonStringWithCallback(modelModalitiesMap, jsonStr, InvalidateExposedDataCache) +} + +func GetModelModalities(name string) string { + name = FormatMatchingModelName(name) + if modalities, ok := modelModalitiesMap.Get(name); ok { + return modalities + } + return "" +} + +func ModelImg2JSONString() string { + return modelImgMap.MarshalJSONString() +} + +func UpdateModelImgByJSONString(jsonStr string) error { + return types.LoadFromJsonStringWithCallback(modelImgMap, jsonStr, InvalidateExposedDataCache) +} + +func GetModelImg(name string) string { + name = FormatMatchingModelName(name) + if img, ok := modelImgMap.Get(name); ok { + return img + } + return "" +} + +func GetModelDisplayNameCopy() map[string]string { + return modelDisplayNameMap.ReadAll() +} + +func GetModelModalitiesCopy() map[string]string { + return modelModalitiesMap.ReadAll() +} + func GetModelRatioCopy() map[string]float64 { return modelRatioMap.ReadAll() } @@ -705,6 +764,9 @@ func GetModelPriceCopy() map[string]float64 { func GetCompletionRatioCopy() map[string]float64 { return completionRatioMap.ReadAll() } +func GetModelImgCopy() map[string]string { + return modelImgMap.ReadAll() +} // 转换模型名，减少渠道必须配置各种带参数模型 func FormatMatchingModelName(name string) string {