diff --git a/.env.example b/.env.example index bece06dbc827..d317e1f3c550 100644 --- a/.env.example +++ b/.env.example @@ -59,7 +59,7 @@ # 设置 Dify 渠道是否输出工作流和节点信息到客户端 # DIFY_DEBUG=true # 设置流式一次回复的超时时间 -# STREAMING_TIMEOUT=90 +# STREAMING_TIMEOUT=120 # 节点类型 diff --git a/README.en.md b/README.en.md index 10a3cdb058ef..b4ae921aeb90 100644 --- a/README.en.md +++ b/README.en.md @@ -100,7 +100,7 @@ This version supports multiple models, please refer to [API Documentation-Relay For detailed configuration instructions, please refer to [Installation Guide-Environment Variables Configuration](https://docs.newapi.pro/installation/environment-variables): - `GENERATE_DEFAULT_TOKEN`: Whether to generate initial tokens for newly registered users, default is `false` -- `STREAMING_TIMEOUT`: Streaming response timeout, default is 60 seconds +- `STREAMING_TIMEOUT`: Streaming response timeout, default is 120 seconds - `DIFY_DEBUG`: Whether to output workflow and node information for Dify channels, default is `true` - `FORCE_STREAM_OPTION`: Whether to override client stream_options parameter, default is `true` - `GET_MEDIA_TOKEN`: Whether to count image tokens, default is `true` diff --git a/README.md b/README.md index 6ba3574cdf86..498d7c6c80d5 100644 --- a/README.md +++ b/README.md @@ -103,7 +103,7 @@ New API提供了丰富的功能,详细特性请参考[特性说明](https://do 详细配置说明请参考[安装指南-环境变量配置](https://docs.newapi.pro/installation/environment-variables): - `GENERATE_DEFAULT_TOKEN`:是否为新注册用户生成初始令牌,默认为 `false` -- `STREAMING_TIMEOUT`:流式回复超时时间,默认60秒 +- `STREAMING_TIMEOUT`:流式回复超时时间,默认120秒 - `DIFY_DEBUG`:Dify渠道是否输出工作流和节点信息,默认 `true` - `FORCE_STREAM_OPTION`:是否覆盖客户端stream_options参数,默认 `true` - `GET_MEDIA_TOKEN`:是否统计图片token,默认 `true` diff --git a/common/constants.go b/common/constants.go index ac8031487b04..67625439c295 100644 --- a/common/constants.go +++ b/common/constants.go @@ -242,6 +242,7 @@ const ( ChannelTypeXai = 48 ChannelTypeCoze = 49 ChannelTypeKling = 50 + ChannelTypeJimeng = 51 ChannelTypeDummy // this one is only for count, do not add any channel after this ) @@ -298,4 +299,5 @@ var ChannelBaseURLs = []string{ "https://api.x.ai", //48 "https://api.coze.cn", //49 "https://api.klingai.com", //50 + "https://visual.volcengineapi.com", //51 } diff --git a/common/http.go b/common/http.go new file mode 100644 index 000000000000..315b19af7b3f --- /dev/null +++ b/common/http.go @@ -0,0 +1,56 @@ +package common + +import ( + "bytes" + "fmt" + "github.com/gin-gonic/gin" + "io" + "net/http" +) + +func CloseResponseBodyGracefully(httpResponse *http.Response) { + if httpResponse == nil || httpResponse.Body == nil { + return + } + err := httpResponse.Body.Close() + if err != nil { + SysError("failed to close response body: " + err.Error()) + } +} + +func IOCopyBytesGracefully(c *gin.Context, src *http.Response, data []byte) { + if src == nil || src.Body == nil { + return + } + + defer CloseResponseBodyGracefully(src) + + if c.Writer == nil { + return + } + + src.Body = io.NopCloser(bytes.NewBuffer(data)) + + // We shouldn't set the header before we parse the response body, because the parse part may fail. + // And then we will have to send an error response, but in this case, the header has already been set. + // So the httpClient will be confused by the response. + // For example, Postman will report error, and we cannot check the response at all. + for k, v := range src.Header { + // avoid setting Content-Length + if k == "Content-Length" { + continue + } + c.Writer.Header().Set(k, v[0]) + } + + // set Content-Length header manually + c.Writer.Header().Set("Content-Length", fmt.Sprintf("%d", len(data))) + + c.Writer.WriteHeader(src.StatusCode) + c.Writer.WriteHeaderNow() + + _, err := io.Copy(c.Writer, src.Body) + if err != nil { + LogError(c, fmt.Sprintf("failed to copy response body: %s", err.Error())) + } +} diff --git a/constant/context_key.go b/constant/context_key.go index 4b4d5cae0f04..895b0fcbb589 100644 --- a/constant/context_key.go +++ b/constant/context_key.go @@ -7,4 +7,5 @@ const ( ContextKeyUserStatus = "user_status" ContextKeyUserEmail = "user_email" ContextKeyUserGroup = "user_group" + ContextKeyUsingGroup = "group" ) diff --git a/constant/env.go b/constant/env.go index 612f3e8be0d6..f33c67ff7595 100644 --- a/constant/env.go +++ b/constant/env.go @@ -23,7 +23,7 @@ var ErrorLogEnabled bool //} func InitEnv() { - StreamingTimeout = common.GetEnvOrDefault("STREAMING_TIMEOUT", 60) + StreamingTimeout = common.GetEnvOrDefault("STREAMING_TIMEOUT", 120) DifyDebug = common.GetEnvOrDefaultBool("DIFY_DEBUG", true) MaxFileDownloadMB = common.GetEnvOrDefault("MAX_FILE_DOWNLOAD_MB", 20) // ForceStreamOption 覆盖请求参数,强制返回usage信息 diff --git a/constant/task.go b/constant/task.go index d466fc8a123b..e7af39a6e1d9 100644 --- a/constant/task.go +++ b/constant/task.go @@ -6,11 +6,15 @@ const ( TaskPlatformSuno TaskPlatform = "suno" TaskPlatformMidjourney = "mj" TaskPlatformKling TaskPlatform = "kling" + TaskPlatformJimeng TaskPlatform = "jimeng" ) const ( SunoActionMusic = "MUSIC" SunoActionLyrics = "LYRICS" + + TaskActionGenerate = "generate" + TaskActionTextGenerate = "textGenerate" ) var SunoModel2Action = map[string]string{ diff --git a/controller/channel-test.go b/controller/channel-test.go index d54ccf0dd2fd..b3badf3504ec 100644 --- a/controller/channel-test.go +++ b/controller/channel-test.go @@ -43,6 +43,9 @@ func testChannel(channel *model.Channel, testModel string) (err error, openAIErr if channel.Type == common.ChannelTypeKling { return errors.New("kling channel test is not supported"), nil } + if channel.Type == common.ChannelTypeJimeng { + return errors.New("jimeng channel test is not supported"), nil + } w := httptest.NewRecorder() c, _ := gin.CreateTestContext(w) @@ -171,7 +174,7 @@ func testChannel(channel *model.Channel, testModel string) (err error, openAIErr other := service.GenerateTextOtherInfo(c, info, priceData.ModelRatio, priceData.GroupRatioInfo.GroupRatio, priceData.CompletionRatio, usage.PromptTokensDetails.CachedTokens, priceData.CacheRatio, priceData.ModelPrice, priceData.GroupRatioInfo.GroupSpecialRatio) model.RecordConsumeLog(c, 1, channel.Id, usage.PromptTokens, usage.CompletionTokens, info.OriginModelName, "模型测试", - quota, "模型测试", 0, quota, int(consumedTime), false, info.Group, other) + quota, "模型测试", 0, quota, int(consumedTime), false, info.UsingGroup, other) common.SysLog(fmt.Sprintf("testing channel #%d, response: \n%s", channel.Id, string(respBody))) return nil, nil } diff --git a/controller/channel.go b/controller/channel.go index 13ed72b3dea0..e46b38f50a63 100644 --- a/controller/channel.go +++ b/controller/channel.go @@ -40,6 +40,17 @@ type OpenAIModelsResponse struct { Success bool `json:"success"` } +func parseStatusFilter(statusParam string) int { + switch strings.ToLower(statusParam) { + case "enabled", "1": + return common.ChannelStatusEnabled + case "disabled", "0": + return 0 + default: + return -1 + } +} + func GetAllChannels(c *gin.Context) { p, _ := strconv.Atoi(c.Query("p")) pageSize, _ := strconv.Atoi(c.Query("page_size")) @@ -52,6 +63,9 @@ func GetAllChannels(c *gin.Context) { channelData := make([]*model.Channel, 0) idSort, _ := strconv.ParseBool(c.Query("id_sort")) enableTagMode, _ := strconv.ParseBool(c.Query("tag_mode")) + statusParam := c.Query("status") + // statusFilter: -1 all, 1 enabled, 0 disabled (include auto & manual) + statusFilter := parseStatusFilter(statusParam) // type filter typeStr := c.Query("type") typeFilter := -1 @@ -64,42 +78,75 @@ func GetAllChannels(c *gin.Context) { var total int64 if enableTagMode { - // tag 分页:先分页 tag,再取各 tag 下 channels tags, err := model.GetPaginatedTags((p-1)*pageSize, pageSize) if err != nil { c.JSON(http.StatusOK, gin.H{"success": false, "message": err.Error()}) return } for _, tag := range tags { - if tag != nil && *tag != "" { - tagChannel, err := model.GetChannelsByTag(*tag, idSort) - if err == nil { - channelData = append(channelData, tagChannel...) + if tag == nil || *tag == "" { + continue + } + tagChannels, err := model.GetChannelsByTag(*tag, idSort) + if err != nil { + continue + } + filtered := make([]*model.Channel, 0) + for _, ch := range tagChannels { + if statusFilter == common.ChannelStatusEnabled && ch.Status != common.ChannelStatusEnabled { + continue + } + if statusFilter == 0 && ch.Status == common.ChannelStatusEnabled { + continue } + if typeFilter >= 0 && ch.Type != typeFilter { + continue + } + filtered = append(filtered, ch) } + channelData = append(channelData, filtered...) } - // 计算 tag 总数用于分页 total, _ = model.CountAllTags() - } else if typeFilter >= 0 { - channels, err := model.GetChannelsByType((p-1)*pageSize, pageSize, idSort, typeFilter) - if err != nil { - c.JSON(http.StatusOK, gin.H{"success": false, "message": err.Error()}) - return - } - channelData = channels - total, _ = model.CountChannelsByType(typeFilter) } else { - channels, err := model.GetAllChannels((p-1)*pageSize, pageSize, false, idSort) + baseQuery := model.DB.Model(&model.Channel{}) + if typeFilter >= 0 { + baseQuery = baseQuery.Where("type = ?", typeFilter) + } + if statusFilter == common.ChannelStatusEnabled { + baseQuery = baseQuery.Where("status = ?", common.ChannelStatusEnabled) + } else if statusFilter == 0 { + baseQuery = baseQuery.Where("status != ?", common.ChannelStatusEnabled) + } + + baseQuery.Count(&total) + + order := "priority desc" + if idSort { + order = "id desc" + } + + err := baseQuery.Order(order).Limit(pageSize).Offset((p-1)*pageSize).Omit("key").Find(&channelData).Error if err != nil { c.JSON(http.StatusOK, gin.H{"success": false, "message": err.Error()}) return } - channelData = channels - total, _ = model.CountAllChannels() } - // calculate type counts - typeCounts, _ := model.CountChannelsGroupByType() + countQuery := model.DB.Model(&model.Channel{}) + if statusFilter == common.ChannelStatusEnabled { + countQuery = countQuery.Where("status = ?", common.ChannelStatusEnabled) + } else if statusFilter == 0 { + countQuery = countQuery.Where("status != ?", common.ChannelStatusEnabled) + } + var results []struct { + Type int64 + Count int64 + } + _ = countQuery.Select("type, count(*) as count").Group("type").Find(&results).Error + typeCounts := make(map[int64]int64) + for _, r := range results { + typeCounts[r.Type] = r.Count + } c.JSON(http.StatusOK, gin.H{ "success": true, @@ -134,13 +181,6 @@ func FetchUpstreamModels(c *gin.Context) { return } - //if channel.Type != common.ChannelTypeOpenAI { - // c.JSON(http.StatusOK, gin.H{ - // "success": false, - // "message": "仅支持 OpenAI 类型渠道", - // }) - // return - //} baseURL := common.ChannelBaseURLs[channel.Type] if channel.GetBaseURL() != "" { baseURL = channel.GetBaseURL() @@ -206,6 +246,8 @@ func SearchChannels(c *gin.Context) { keyword := c.Query("keyword") group := c.Query("group") modelKeyword := c.Query("model") + statusParam := c.Query("status") + statusFilter := parseStatusFilter(statusParam) idSort, _ := strconv.ParseBool(c.Query("id_sort")) enableTagMode, _ := strconv.ParseBool(c.Query("tag_mode")) channelData := make([]*model.Channel, 0) @@ -238,17 +280,71 @@ func SearchChannels(c *gin.Context) { channelData = channels } + if statusFilter == common.ChannelStatusEnabled || statusFilter == 0 { + filtered := make([]*model.Channel, 0, len(channelData)) + for _, ch := range channelData { + if statusFilter == common.ChannelStatusEnabled && ch.Status != common.ChannelStatusEnabled { + continue + } + if statusFilter == 0 && ch.Status == common.ChannelStatusEnabled { + continue + } + filtered = append(filtered, ch) + } + channelData = filtered + } + // calculate type counts for search results typeCounts := make(map[int64]int64) for _, channel := range channelData { typeCounts[int64(channel.Type)]++ } + typeParam := c.Query("type") + typeFilter := -1 + if typeParam != "" { + if tp, err := strconv.Atoi(typeParam); err == nil { + typeFilter = tp + } + } + + if typeFilter >= 0 { + filtered := make([]*model.Channel, 0, len(channelData)) + for _, ch := range channelData { + if ch.Type == typeFilter { + filtered = append(filtered, ch) + } + } + channelData = filtered + } + + page, _ := strconv.Atoi(c.DefaultQuery("p", "1")) + pageSize, _ := strconv.Atoi(c.DefaultQuery("page_size", "20")) + if page < 1 { + page = 1 + } + if pageSize <= 0 { + pageSize = 20 + } + + total := len(channelData) + startIdx := (page - 1) * pageSize + if startIdx > total { + startIdx = total + } + endIdx := startIdx + pageSize + if endIdx > total { + endIdx = total + } + + pagedData := channelData[startIdx:endIdx] + c.JSON(http.StatusOK, gin.H{ "success": true, "message": "", "data": gin.H{ - "items": channelData, + "items": pagedData, + "total": total, "type_counts": typeCounts, }, }) @@ -546,6 +642,7 @@ func UpdateChannel(c *gin.Context) { }) return } + channel.Key = "" c.JSON(http.StatusOK, gin.H{ "success": true, "message": "", diff --git a/controller/ratio_sync.go b/controller/ratio_sync.go index f749f3845e10..0453870d0070 100644 --- a/controller/ratio_sync.go +++ b/controller/ratio_sync.go @@ -3,6 +3,7 @@ package controller import ( "context" "encoding/json" + "fmt" "net/http" "strings" "sync" @@ -43,7 +44,17 @@ func FetchUpstreamRatios(c *gin.Context) { var upstreams []dto.UpstreamDTO - if len(req.ChannelIDs) > 0 { + if len(req.Upstreams) > 0 { + for _, u := range req.Upstreams { + if strings.HasPrefix(u.BaseURL, "http") { + if u.Endpoint == "" { + u.Endpoint = defaultEndpoint + } + u.BaseURL = strings.TrimRight(u.BaseURL, "/") + upstreams = append(upstreams, u) + } + } + } else if len(req.ChannelIDs) > 0 { intIds := make([]int, 0, len(req.ChannelIDs)) for _, id64 := range req.ChannelIDs { intIds = append(intIds, int(id64)) @@ -57,6 +68,7 @@ func FetchUpstreamRatios(c *gin.Context) { for _, ch := range dbChannels { if base := ch.GetBaseURL(); strings.HasPrefix(base, "http") { upstreams = append(upstreams, dto.UpstreamDTO{ + ID: ch.Id, Name: ch.Name, BaseURL: strings.TrimRight(base, "/"), Endpoint: "", @@ -93,43 +105,125 @@ func FetchUpstreamRatios(c *gin.Context) { } fullURL := chItem.BaseURL + endpoint + uniqueName := chItem.Name + if chItem.ID != 0 { + uniqueName = fmt.Sprintf("%s(%d)", chItem.Name, chItem.ID) + } + ctx, cancel := context.WithTimeout(c.Request.Context(), time.Duration(req.Timeout)*time.Second) defer cancel() httpReq, err := http.NewRequestWithContext(ctx, http.MethodGet, fullURL, nil) if err != nil { common.LogWarn(c.Request.Context(), "build request failed: "+err.Error()) - ch <- upstreamResult{Name: chItem.Name, Err: err.Error()} + ch <- upstreamResult{Name: uniqueName, Err: err.Error()} return } resp, err := client.Do(httpReq) if err != nil { common.LogWarn(c.Request.Context(), "http error on "+chItem.Name+": "+err.Error()) - ch <- upstreamResult{Name: chItem.Name, Err: err.Error()} + ch <- upstreamResult{Name: uniqueName, Err: err.Error()} return } defer resp.Body.Close() if resp.StatusCode != http.StatusOK { common.LogWarn(c.Request.Context(), "non-200 from "+chItem.Name+": "+resp.Status) - ch <- upstreamResult{Name: chItem.Name, Err: resp.Status} + ch <- upstreamResult{Name: uniqueName, Err: resp.Status} return } + // 兼容两种上游接口格式: + // type1: /api/ratio_config -> data 为 map[string]any,包含 model_ratio/completion_ratio/cache_ratio/model_price + // type2: /api/pricing -> data 为 []Pricing 列表,需要转换为与 type1 相同的 map 格式 var body struct { - Success bool `json:"success"` - Data map[string]any `json:"data"` - Message string `json:"message"` + Success bool `json:"success"` + Data json.RawMessage `json:"data"` + Message string `json:"message"` } + if err := json.NewDecoder(resp.Body).Decode(&body); err != nil { common.LogWarn(c.Request.Context(), "json decode failed from "+chItem.Name+": "+err.Error()) - ch <- upstreamResult{Name: chItem.Name, Err: err.Error()} + ch <- upstreamResult{Name: uniqueName, Err: err.Error()} return } + if !body.Success { - ch <- upstreamResult{Name: chItem.Name, Err: body.Message} + ch <- upstreamResult{Name: uniqueName, Err: body.Message} return } - ch <- upstreamResult{Name: chItem.Name, Data: body.Data} + + // 尝试按 type1 解析 + var type1Data map[string]any + if err := json.Unmarshal(body.Data, &type1Data); err == nil { + // 如果包含至少一个 ratioTypes 字段,则认为是 type1 + isType1 := false + for _, rt := range ratioTypes { + if _, ok := type1Data[rt]; ok { + isType1 = true + break + } + } + if isType1 { + ch <- upstreamResult{Name: uniqueName, Data: type1Data} + return + } + } + + // 如果不是 type1,则尝试按 type2 (/api/pricing) 解析 + var pricingItems []struct { + ModelName string `json:"model_name"` + QuotaType int `json:"quota_type"` + ModelRatio float64 `json:"model_ratio"` + ModelPrice float64 `json:"model_price"` + CompletionRatio float64 `json:"completion_ratio"` + } + if err := json.Unmarshal(body.Data, &pricingItems); err != nil { + common.LogWarn(c.Request.Context(), "unrecognized data format from "+chItem.Name+": "+err.Error()) + ch <- upstreamResult{Name: uniqueName, Err: "无法解析上游返回数据"} + return + } + + modelRatioMap := make(map[string]float64) + completionRatioMap := make(map[string]float64) + modelPriceMap := make(map[string]float64) + + for _, item := range pricingItems { + if item.QuotaType == 1 { + modelPriceMap[item.ModelName] = item.ModelPrice + } else { + modelRatioMap[item.ModelName] = item.ModelRatio + // completionRatio 可能为 0,此时也直接赋值,保持与上游一致 + completionRatioMap[item.ModelName] = item.CompletionRatio + } + } + + converted := make(map[string]any) + + if len(modelRatioMap) > 0 { + ratioAny := make(map[string]any, len(modelRatioMap)) + for k, v := range modelRatioMap { + ratioAny[k] = v + } + converted["model_ratio"] = ratioAny + } + + if len(completionRatioMap) > 0 { + compAny := make(map[string]any, len(completionRatioMap)) + for k, v := range completionRatioMap { + compAny[k] = v + } + converted["completion_ratio"] = compAny + } + + if len(modelPriceMap) > 0 { + priceAny := make(map[string]any, len(modelPriceMap)) + for k, v := range modelPriceMap { + priceAny[k] = v + } + converted["model_price"] = priceAny + } + + ch <- upstreamResult{Name: uniqueName, Data: converted} }(chn) } @@ -202,6 +296,43 @@ func buildDifferences(localData map[string]any, successfulChannels []struct { } } + confidenceMap := make(map[string]map[string]bool) + + // 预处理阶段:检查pricing接口的可信度 + for _, channel := range successfulChannels { + confidenceMap[channel.name] = make(map[string]bool) + + modelRatios, hasModelRatio := channel.data["model_ratio"].(map[string]any) + completionRatios, hasCompletionRatio := channel.data["completion_ratio"].(map[string]any) + + if hasModelRatio && hasCompletionRatio { + // 遍历所有模型,检查是否满足不可信条件 + for modelName := range allModels { + // 默认为可信 + confidenceMap[channel.name][modelName] = true + + // 检查是否满足不可信条件:model_ratio为37.5且completion_ratio为1 + if modelRatioVal, ok := modelRatios[modelName]; ok { + if completionRatioVal, ok := completionRatios[modelName]; ok { + // 转换为float64进行比较 + if modelRatioFloat, ok := modelRatioVal.(float64); ok { + if completionRatioFloat, ok := completionRatioVal.(float64); ok { + if modelRatioFloat == 37.5 && completionRatioFloat == 1.0 { + confidenceMap[channel.name][modelName] = false + } + } + } + } + } + } + } else { + // 如果不是从pricing接口获取的数据,则全部标记为可信 + for modelName := range allModels { + confidenceMap[channel.name][modelName] = true + } + } + } + for modelName := range allModels { for _, ratioType := range ratioTypes { var localValue interface{} = nil @@ -214,6 +345,7 @@ func buildDifferences(localData map[string]any, successfulChannels []struct { } upstreamValues := make(map[string]interface{}) + confidenceValues := make(map[string]bool) hasUpstreamValue := false hasDifference := false @@ -241,6 +373,8 @@ func buildDifferences(localData map[string]any, successfulChannels []struct { } upstreamValues[channel.name] = upstreamValue + + confidenceValues[channel.name] = confidenceMap[channel.name][modelName] } shouldInclude := false @@ -262,6 +396,7 @@ func buildDifferences(localData map[string]any, successfulChannels []struct { differences[modelName][ratioType] = dto.DifferenceItem{ Current: localValue, Upstreams: upstreamValues, + Confidence: confidenceValues, } } } @@ -283,9 +418,26 @@ func buildDifferences(localData map[string]any, successfulChannels []struct { for chName := range item.Upstreams { if !channelHasDiff[chName] { delete(item.Upstreams, chName) + delete(item.Confidence, chName) + } + } + + allSame := true + for _, v := range item.Upstreams { + if v != "same" { + allSame = false + break } } - differences[modelName][ratioType] = item + if len(item.Upstreams) == 0 || allSame { + delete(ratioMap, ratioType) + } else { + differences[modelName][ratioType] = item + } + } + + if len(ratioMap) == 0 { + delete(differences, modelName) } } diff --git a/controller/task.go b/controller/task.go index f7523e87f45d..5cfa728aa673 100644 --- a/controller/task.go +++ b/controller/task.go @@ -74,8 +74,8 @@ func UpdateTaskByPlatform(platform constant.TaskPlatform, taskChannelM map[int][ //_ = UpdateMidjourneyTaskAll(context.Background(), tasks) case constant.TaskPlatformSuno: _ = UpdateSunoTaskAll(context.Background(), taskChannelM, taskM) - case constant.TaskPlatformKling: - _ = UpdateVideoTaskAll(context.Background(), taskChannelM, taskM) + case constant.TaskPlatformKling, constant.TaskPlatformJimeng: + _ = UpdateVideoTaskAll(context.Background(), platform, taskChannelM, taskM) default: common.SysLog("未知平台") } diff --git a/controller/task_video.go b/controller/task_video.go index a2c2431dff40..a17351b550ae 100644 --- a/controller/task_video.go +++ b/controller/task_video.go @@ -2,27 +2,26 @@ package controller import ( "context" - "encoding/json" "fmt" "io" - "net/http" "one-api/common" "one-api/constant" "one-api/model" "one-api/relay" "one-api/relay/channel" + "time" ) -func UpdateVideoTaskAll(ctx context.Context, taskChannelM map[int][]string, taskM map[string]*model.Task) error { +func UpdateVideoTaskAll(ctx context.Context, platform constant.TaskPlatform, taskChannelM map[int][]string, taskM map[string]*model.Task) error { for channelId, taskIds := range taskChannelM { - if err := updateVideoTaskAll(ctx, channelId, taskIds, taskM); err != nil { + if err := updateVideoTaskAll(ctx, platform, channelId, taskIds, taskM); err != nil { common.LogError(ctx, fmt.Sprintf("Channel #%d failed to update video async tasks: %s", channelId, err.Error())) } } return nil } -func updateVideoTaskAll(ctx context.Context, channelId int, taskIds []string, taskM map[string]*model.Task) error { +func updateVideoTaskAll(ctx context.Context, platform constant.TaskPlatform, channelId int, taskIds []string, taskM map[string]*model.Task) error { common.LogInfo(ctx, fmt.Sprintf("Channel #%d pending video tasks: %d", channelId, len(taskIds))) if len(taskIds) == 0 { return nil @@ -39,7 +38,7 @@ func updateVideoTaskAll(ctx context.Context, channelId int, taskIds []string, ta } return fmt.Errorf("CacheGetChannel failed: %w", err) } - adaptor := relay.GetTaskAdaptor(constant.TaskPlatformKling) + adaptor := relay.GetTaskAdaptor(platform) if adaptor == nil { return fmt.Errorf("video adaptor not found") } @@ -56,70 +55,64 @@ func updateVideoSingleTask(ctx context.Context, adaptor channel.TaskAdaptor, cha if channel.GetBaseURL() != "" { baseURL = channel.GetBaseURL() } + + task := taskM[taskId] + if task == nil { + common.LogError(ctx, fmt.Sprintf("Task %s not found in taskM", taskId)) + return fmt.Errorf("task %s not found", taskId) + } resp, err := adaptor.FetchTask(baseURL, channel.Key, map[string]any{ "task_id": taskId, + "action": task.Action, }) if err != nil { - return fmt.Errorf("FetchTask failed for task %s: %w", taskId, err) - } - if resp.StatusCode != http.StatusOK { - return fmt.Errorf("Get Video Task status code: %d", resp.StatusCode) + return fmt.Errorf("fetchTask failed for task %s: %w", taskId, err) } + //if resp.StatusCode != http.StatusOK { + //return fmt.Errorf("get Video Task status code: %d", resp.StatusCode) + //} defer resp.Body.Close() responseBody, err := io.ReadAll(resp.Body) if err != nil { - return fmt.Errorf("ReadAll failed for task %s: %w", taskId, err) + return fmt.Errorf("readAll failed for task %s: %w", taskId, err) } - var responseItem map[string]interface{} - err = json.Unmarshal(responseBody, &responseItem) + taskResult, err := adaptor.ParseTaskResult(responseBody) if err != nil { - common.LogError(ctx, fmt.Sprintf("Failed to parse video task response body: %v, body: %s", err, string(responseBody))) - return fmt.Errorf("Unmarshal failed for task %s: %w", taskId, err) - } - - code, _ := responseItem["code"].(float64) - if code != 0 { - return fmt.Errorf("video task fetch failed for task %s", taskId) - } - - data, ok := responseItem["data"].(map[string]interface{}) - if !ok { - common.LogError(ctx, fmt.Sprintf("Video task data format error: %s", string(responseBody))) - return fmt.Errorf("video task data format error for task %s", taskId) - } - - task := taskM[taskId] - if task == nil { - common.LogError(ctx, fmt.Sprintf("Task %s not found in taskM", taskId)) - return fmt.Errorf("task %s not found", taskId) + return fmt.Errorf("parseTaskResult failed for task %s: %w", taskId, err) } + //if taskResult.Code != 0 { + // return fmt.Errorf("video task fetch failed for task %s", taskId) + //} - if status, ok := data["task_status"].(string); ok { - switch status { - case "submitted", "queued": - task.Status = model.TaskStatusSubmitted - case "processing": - task.Status = model.TaskStatusInProgress - case "succeed": - task.Status = model.TaskStatusSuccess - task.Progress = "100%" - if url, err := adaptor.ParseResultUrl(responseItem); err == nil { - task.FailReason = url - } else { - common.LogWarn(ctx, fmt.Sprintf("Failed to get url from body for task %s: %s", task.TaskID, err.Error())) - } - case "failed": - task.Status = model.TaskStatusFailure - task.Progress = "100%" - if reason, ok := data["fail_reason"].(string); ok { - task.FailReason = reason - } + now := time.Now().Unix() + if taskResult.Status == "" { + return fmt.Errorf("task %s status is empty", taskId) + } + task.Status = model.TaskStatus(taskResult.Status) + switch taskResult.Status { + case model.TaskStatusSubmitted: + task.Progress = "10%" + case model.TaskStatusQueued: + task.Progress = "20%" + case model.TaskStatusInProgress: + task.Progress = "30%" + if task.StartTime == 0 { + task.StartTime = now } - } - - // If task failed, refund quota - if task.Status == model.TaskStatusFailure { + case model.TaskStatusSuccess: + task.Progress = "100%" + if task.FinishTime == 0 { + task.FinishTime = now + } + task.FailReason = taskResult.Url + case model.TaskStatusFailure: + task.Status = model.TaskStatusFailure + task.Progress = "100%" + if task.FinishTime == 0 { + task.FinishTime = now + } + task.FailReason = taskResult.Reason common.LogInfo(ctx, fmt.Sprintf("Task %s failed: %s", task.TaskID, task.FailReason)) quota := task.Quota if quota != 0 { @@ -129,6 +122,11 @@ func updateVideoSingleTask(ctx context.Context, adaptor channel.TaskAdaptor, cha logContent := fmt.Sprintf("Video async task failed %s, refund %s", task.TaskID, common.LogQuota(quota)) model.RecordLog(task.UserId, model.LogTypeSystem, logContent) } + default: + return fmt.Errorf("unknown task status %s for task %s", taskResult.Status, taskId) + } + if taskResult.Progress != "" { + task.Progress = taskResult.Progress } task.Data = responseBody diff --git a/controller/token.go b/controller/token.go index c57552c0da39..173fc22e22b7 100644 --- a/controller/token.go +++ b/controller/token.go @@ -258,3 +258,32 @@ func UpdateToken(c *gin.Context) { }) return } + +type TokenBatch struct { + Ids []int `json:"ids"` +} + +func DeleteTokenBatch(c *gin.Context) { + tokenBatch := TokenBatch{} + if err := c.ShouldBindJSON(&tokenBatch); err != nil || len(tokenBatch.Ids) == 0 { + c.JSON(http.StatusOK, gin.H{ + "success": false, + "message": "参数错误", + }) + return + } + userId := c.GetInt("id") + count, err := model.BatchDeleteTokens(tokenBatch.Ids, userId) + if err != nil { + c.JSON(http.StatusOK, gin.H{ + "success": false, + "message": err.Error(), + }) + return + } + c.JSON(http.StatusOK, gin.H{ + "success": true, + "message": "", + "data": count, + }) +} diff --git a/dto/openai_request.go b/dto/openai_request.go index 42c290ca758d..0104f347090d 100644 --- a/dto/openai_request.go +++ b/dto/openai_request.go @@ -57,6 +57,7 @@ type GeneralOpenAIRequest struct { ExtraBody json.RawMessage `json:"extra_body,omitempty"` WebSearchOptions *WebSearchOptions `json:"web_search_options,omitempty"` // OpenRouter Params + Usage json.RawMessage `json:"usage,omitempty"` Reasoning json.RawMessage `json:"reasoning,omitempty"` // Ali Qwen Params VlHighResolutionImages json.RawMessage `json:"vl_high_resolution_images,omitempty"` @@ -645,4 +646,6 @@ type ResponsesToolsCall struct { Name string `json:"name,omitempty"` Description string `json:"description,omitempty"` Parameters json.RawMessage `json:"parameters,omitempty"` + Function json.RawMessage `json:"function,omitempty"` + Container json.RawMessage `json:"container,omitempty"` } diff --git a/dto/openai_response.go b/dto/openai_response.go index 790d4df81957..d95acd9eb194 100644 --- a/dto/openai_response.go +++ b/dto/openai_response.go @@ -26,7 +26,7 @@ type OpenAITextResponse struct { Id string `json:"id"` Model string `json:"model"` Object string `json:"object"` - Created int64 `json:"created"` + Created any `json:"created"` Choices []OpenAITextResponseChoice `json:"choices"` Error *OpenAIError `json:"error,omitempty"` Usage `json:"usage"` @@ -178,6 +178,8 @@ type Usage struct { InputTokens int `json:"input_tokens"` OutputTokens int `json:"output_tokens"` InputTokensDetails *InputTokenDetails `json:"input_tokens_details"` + // OpenRouter Params + Cost float64 `json:"cost,omitempty"` } type InputTokenDetails struct { diff --git a/dto/ratio_sync.go b/dto/ratio_sync.go index 55a89025b085..6315f31ae6f5 100644 --- a/dto/ratio_sync.go +++ b/dto/ratio_sync.go @@ -1,18 +1,7 @@ package dto -// UpstreamDTO 提交到后端同步倍率的上游渠道信息 -// Endpoint 可以为空,后端会默认使用 /api/ratio_config -// BaseURL 必须以 http/https 开头,不要以 / 结尾 -// 例如: https://api.example.com -// Endpoint: /api/ratio_config -// 提交示例: -// { -// "name": "openai", -// "base_url": "https://api.openai.com", -// "endpoint": "/ratio_config" -// } - type UpstreamDTO struct { + ID int `json:"id,omitempty"` Name string `json:"name" binding:"required"` BaseURL string `json:"base_url" binding:"required"` Endpoint string `json:"endpoint"` @@ -20,6 +9,7 @@ type UpstreamDTO struct { type UpstreamRequest struct { ChannelIDs []int64 `json:"channel_ids"` + Upstreams []UpstreamDTO `json:"upstreams"` Timeout int `json:"timeout"` } @@ -37,10 +27,9 @@ type TestResult struct { type DifferenceItem struct { Current interface{} `json:"current"` Upstreams map[string]interface{} `json:"upstreams"` + Confidence map[string]bool `json:"confidence"` } -// SyncableChannel 可同步的渠道信息(base_url 不为空) - type SyncableChannel struct { ID int `json:"id"` Name string `json:"name"` diff --git a/middleware/auth.go b/middleware/auth.go index f387029fa5dc..ecf4844bc9b4 100644 --- a/middleware/auth.go +++ b/middleware/auth.go @@ -184,7 +184,7 @@ func TokenAuth() func(c *gin.Context) { } } // gemini api 从query中获取key - if strings.HasPrefix(c.Request.URL.Path, "/v1beta/models/") { + if strings.HasPrefix(c.Request.URL.Path, "/v1beta/models/") || strings.HasPrefix(c.Request.URL.Path, "/v1/models/") { skKey := c.Query("key") if skKey != "" { c.Request.Header.Set("Authorization", "Bearer "+skKey) diff --git a/middleware/distributor.go b/middleware/distributor.go index 9d074ce8d068..0a6a9af476b7 100644 --- a/middleware/distributor.go +++ b/middleware/distributor.go @@ -57,7 +57,7 @@ func Distribute() func(c *gin.Context) { } userGroup = tokenGroup } - c.Set("group", userGroup) + c.Set(constant.ContextKeyUsingGroup, userGroup) if ok { id, err := strconv.Atoi(channelId.(string)) if err != nil { @@ -171,15 +171,25 @@ func getModelRequest(c *gin.Context) (*ModelRequest, bool, error) { c.Set("platform", string(constant.TaskPlatformSuno)) c.Set("relay_mode", relayMode) } else if strings.Contains(c.Request.URL.Path, "/v1/video/generations") { - relayMode := relayconstant.Path2RelayKling(c.Request.Method, c.Request.URL.Path) - if relayMode == relayconstant.RelayModeKlingFetchByID { - shouldSelectChannel = false + err = common.UnmarshalBodyReusable(c, &modelRequest) + var platform string + var relayMode int + if strings.HasPrefix(modelRequest.Model, "jimeng") { + platform = string(constant.TaskPlatformJimeng) + relayMode = relayconstant.Path2RelayJimeng(c.Request.Method, c.Request.URL.Path) + if relayMode == relayconstant.RelayModeJimengFetchByID { + shouldSelectChannel = false + } } else { - err = common.UnmarshalBodyReusable(c, &modelRequest) + platform = string(constant.TaskPlatformKling) + relayMode = relayconstant.Path2RelayKling(c.Request.Method, c.Request.URL.Path) + if relayMode == relayconstant.RelayModeKlingFetchByID { + shouldSelectChannel = false + } } - c.Set("platform", string(constant.TaskPlatformKling)) + c.Set("platform", platform) c.Set("relay_mode", relayMode) - } else if strings.HasPrefix(c.Request.URL.Path, "/v1beta/models/") { + } else if strings.HasPrefix(c.Request.URL.Path, "/v1beta/models/") || strings.HasPrefix(c.Request.URL.Path, "/v1/models/") { // Gemini API 路径处理: /v1beta/models/gemini-2.0-flash:generateContent relayMode := relayconstant.RelayModeGemini modelName := extractModelNameFromGeminiPath(c.Request.URL.Path) diff --git a/middleware/kling_adapter.go b/middleware/kling_adapter.go new file mode 100644 index 000000000000..8e2a35519d30 --- /dev/null +++ b/middleware/kling_adapter.go @@ -0,0 +1,47 @@ +package middleware + +import ( + "bytes" + "encoding/json" + "io" + "one-api/common" + "one-api/constant" + + "github.com/gin-gonic/gin" +) + +func KlingRequestConvert() func(c *gin.Context) { + return func(c *gin.Context) { + var originalReq map[string]interface{} + if err := common.UnmarshalBodyReusable(c, &originalReq); err != nil { + c.Next() + return + } + + model, _ := originalReq["model"].(string) + prompt, _ := originalReq["prompt"].(string) + + unifiedReq := map[string]interface{}{ + "model": model, + "prompt": prompt, + "metadata": originalReq, + } + + jsonData, err := json.Marshal(unifiedReq) + if err != nil { + c.Next() + return + } + + // Rewrite request body and path + c.Request.Body = io.NopCloser(bytes.NewBuffer(jsonData)) + c.Request.URL.Path = "/v1/video/generations" + if image := originalReq["image"]; image == "" { + c.Set("action", constant.TaskActionTextGenerate) + } + + // We have to reset the request body for the next handlers + c.Set(common.KeyRequestBody, jsonData) + c.Next() + } +} diff --git a/model/token.go b/model/token.go index 2ed2c09a912b..7e68f1857ef2 100644 --- a/model/token.go +++ b/model/token.go @@ -327,3 +327,37 @@ func CountUserTokens(userId int) (int64, error) { err := DB.Model(&Token{}).Where("user_id = ?", userId).Count(&total).Error return total, err } + +// BatchDeleteTokens 删除指定用户的一组令牌,返回成功删除数量 +func BatchDeleteTokens(ids []int, userId int) (int, error) { + if len(ids) == 0 { + return 0, errors.New("ids 不能为空!") + } + + tx := DB.Begin() + + var tokens []Token + if err := tx.Where("user_id = ? AND id IN (?)", userId, ids).Find(&tokens).Error; err != nil { + tx.Rollback() + return 0, err + } + + if err := tx.Where("user_id = ? AND id IN (?)", userId, ids).Delete(&Token{}).Error; err != nil { + tx.Rollback() + return 0, err + } + + if err := tx.Commit().Error; err != nil { + return 0, err + } + + if common.RedisEnabled { + gopool.Go(func() { + for _, t := range tokens { + _ = cacheDeleteToken(t.Key) + } + }) + } + + return len(tokens), nil +} diff --git a/relay/channel/adapter.go b/relay/channel/adapter.go index 873997f6d3e1..2ff34e0176f1 100644 --- a/relay/channel/adapter.go +++ b/relay/channel/adapter.go @@ -45,5 +45,5 @@ type TaskAdaptor interface { // FetchTask FetchTask(baseUrl, key string, body map[string]any) (*http.Response, error) - ParseResultUrl(resp map[string]any) (string, error) + ParseTaskResult(respBody []byte) (*relaycommon.TaskInfo, error) } diff --git a/relay/channel/ali/image.go b/relay/channel/ali/image.go index 4420358335f9..c84c7885758c 100644 --- a/relay/channel/ali/image.go +++ b/relay/channel/ali/image.go @@ -132,10 +132,7 @@ func aliImageHandler(c *gin.Context, resp *http.Response, info *relaycommon.Rela if err != nil { return service.OpenAIErrorWrapper(err, "read_response_body_failed", http.StatusInternalServerError), nil } - err = resp.Body.Close() - if err != nil { - return service.OpenAIErrorWrapper(err, "close_response_body_failed", http.StatusInternalServerError), nil - } + common.CloseResponseBodyGracefully(resp) err = json.Unmarshal(responseBody, &aliTaskResponse) if err != nil { return service.OpenAIErrorWrapper(err, "unmarshal_response_body_failed", http.StatusInternalServerError), nil diff --git a/relay/channel/ali/rerank.go b/relay/channel/ali/rerank.go index c9ae066a1941..ebfe26deb105 100644 --- a/relay/channel/ali/rerank.go +++ b/relay/channel/ali/rerank.go @@ -4,6 +4,7 @@ import ( "encoding/json" "io" "net/http" + "one-api/common" "one-api/dto" relaycommon "one-api/relay/common" "one-api/service" @@ -35,10 +36,7 @@ func RerankHandler(c *gin.Context, resp *http.Response, info *relaycommon.RelayI if err != nil { return service.OpenAIErrorWrapper(err, "read_response_body_failed", http.StatusInternalServerError), nil } - err = resp.Body.Close() - if err != nil { - return service.OpenAIErrorWrapper(err, "close_response_body_failed", http.StatusInternalServerError), nil - } + common.CloseResponseBodyGracefully(resp) var aliResponse AliRerankResponse err = json.Unmarshal(responseBody, &aliResponse) diff --git a/relay/channel/ali/text.go b/relay/channel/ali/text.go index 2f1387c5c21f..fda8a7f478b4 100644 --- a/relay/channel/ali/text.go +++ b/relay/channel/ali/text.go @@ -45,10 +45,7 @@ func aliEmbeddingHandler(c *gin.Context, resp *http.Response) (*dto.OpenAIErrorW return service.OpenAIErrorWrapper(err, "unmarshal_response_body_failed", http.StatusInternalServerError), nil } - err = resp.Body.Close() - if err != nil { - return service.OpenAIErrorWrapper(err, "close_response_body_failed", http.StatusInternalServerError), nil - } + common.CloseResponseBodyGracefully(resp) if aliResponse.Code != "" { return &dto.OpenAIErrorWithStatusCode{ @@ -186,10 +183,7 @@ func aliStreamHandler(c *gin.Context, resp *http.Response) (*dto.OpenAIErrorWith return false } }) - err := resp.Body.Close() - if err != nil { - return service.OpenAIErrorWrapper(err, "close_response_body_failed", http.StatusInternalServerError), nil - } + common.CloseResponseBodyGracefully(resp) return nil, &usage } @@ -199,10 +193,7 @@ func aliHandler(c *gin.Context, resp *http.Response) (*dto.OpenAIErrorWithStatus if err != nil { return service.OpenAIErrorWrapper(err, "read_response_body_failed", http.StatusInternalServerError), nil } - err = resp.Body.Close() - if err != nil { - return service.OpenAIErrorWrapper(err, "close_response_body_failed", http.StatusInternalServerError), nil - } + common.CloseResponseBodyGracefully(resp) err = json.Unmarshal(responseBody, &aliResponse) if err != nil { return service.OpenAIErrorWrapper(err, "unmarshal_response_body_failed", http.StatusInternalServerError), nil diff --git a/relay/channel/baidu/relay-baidu.go b/relay/channel/baidu/relay-baidu.go index 55b6c1379fdf..011af26293b0 100644 --- a/relay/channel/baidu/relay-baidu.go +++ b/relay/channel/baidu/relay-baidu.go @@ -166,10 +166,7 @@ func baiduStreamHandler(c *gin.Context, resp *http.Response) (*dto.OpenAIErrorWi return false } }) - err := resp.Body.Close() - if err != nil { - return service.OpenAIErrorWrapper(err, "close_response_body_failed", http.StatusInternalServerError), nil - } + common.CloseResponseBodyGracefully(resp) return nil, &usage } @@ -179,10 +176,7 @@ func baiduHandler(c *gin.Context, resp *http.Response) (*dto.OpenAIErrorWithStat if err != nil { return service.OpenAIErrorWrapper(err, "read_response_body_failed", http.StatusInternalServerError), nil } - err = resp.Body.Close() - if err != nil { - return service.OpenAIErrorWrapper(err, "close_response_body_failed", http.StatusInternalServerError), nil - } + common.CloseResponseBodyGracefully(resp) err = json.Unmarshal(responseBody, &baiduResponse) if err != nil { return service.OpenAIErrorWrapper(err, "unmarshal_response_body_failed", http.StatusInternalServerError), nil @@ -215,10 +209,7 @@ func baiduEmbeddingHandler(c *gin.Context, resp *http.Response) (*dto.OpenAIErro if err != nil { return service.OpenAIErrorWrapper(err, "read_response_body_failed", http.StatusInternalServerError), nil } - err = resp.Body.Close() - if err != nil { - return service.OpenAIErrorWrapper(err, "close_response_body_failed", http.StatusInternalServerError), nil - } + common.CloseResponseBodyGracefully(resp) err = json.Unmarshal(responseBody, &baiduResponse) if err != nil { return service.OpenAIErrorWrapper(err, "unmarshal_response_body_failed", http.StatusInternalServerError), nil diff --git a/relay/channel/claude/relay-claude.go b/relay/channel/claude/relay-claude.go index 406ebc8a4017..f164fd4ddb70 100644 --- a/relay/channel/claude/relay-claude.go +++ b/relay/channel/claude/relay-claude.go @@ -7,6 +7,7 @@ import ( "net/http" "one-api/common" "one-api/dto" + "one-api/relay/channel/openrouter" relaycommon "one-api/relay/common" "one-api/relay/helper" "one-api/service" @@ -122,6 +123,21 @@ func RequestOpenAI2ClaudeMessage(textRequest dto.GeneralOpenAIRequest) (*dto.Cla claudeRequest.Model = strings.TrimSuffix(textRequest.Model, "-thinking") } + if textRequest.Reasoning != nil { + var reasoning openrouter.RequestReasoning + if err := common.DecodeJson(textRequest.Reasoning, &reasoning); err != nil { + return nil, err + } + + budgetTokens := reasoning.MaxTokens + if budgetTokens > 0 { + claudeRequest.Thinking = &dto.Thinking{ + Type: "enabled", + BudgetTokens: &budgetTokens, + } + } + } + if textRequest.Stop != nil { // stop maybe string/array string, convert to array string switch textRequest.Stop.(type) { diff --git a/relay/channel/cloudflare/relay_cloudflare.go b/relay/channel/cloudflare/relay_cloudflare.go index 50d4928af68f..1c3a26f7d699 100644 --- a/relay/channel/cloudflare/relay_cloudflare.go +++ b/relay/channel/cloudflare/relay_cloudflare.go @@ -81,10 +81,7 @@ func cfStreamHandler(c *gin.Context, resp *http.Response, info *relaycommon.Rela } helper.Done(c) - err := resp.Body.Close() - if err != nil { - common.LogError(c, "close_response_body_failed: "+err.Error()) - } + common.CloseResponseBodyGracefully(resp) return nil, usage } @@ -94,10 +91,7 @@ func cfHandler(c *gin.Context, resp *http.Response, info *relaycommon.RelayInfo) if err != nil { return service.OpenAIErrorWrapper(err, "read_response_body_failed", http.StatusInternalServerError), nil } - err = resp.Body.Close() - if err != nil { - return service.OpenAIErrorWrapperLocal(err, "close_response_body_failed", http.StatusInternalServerError), nil - } + common.CloseResponseBodyGracefully(resp) var response dto.TextResponse err = json.Unmarshal(responseBody, &response) if err != nil { @@ -127,10 +121,7 @@ func cfSTTHandler(c *gin.Context, resp *http.Response, info *relaycommon.RelayIn if err != nil { return service.OpenAIErrorWrapper(err, "read_response_body_failed", http.StatusInternalServerError), nil } - err = resp.Body.Close() - if err != nil { - return service.OpenAIErrorWrapper(err, "close_response_body_failed", http.StatusInternalServerError), nil - } + common.CloseResponseBodyGracefully(resp) err = json.Unmarshal(responseBody, &cfResp) if err != nil { return service.OpenAIErrorWrapper(err, "unmarshal_response_body_failed", http.StatusInternalServerError), nil diff --git a/relay/channel/cohere/relay-cohere.go b/relay/channel/cohere/relay-cohere.go index 29064242d2a7..4637740d13ca 100644 --- a/relay/channel/cohere/relay-cohere.go +++ b/relay/channel/cohere/relay-cohere.go @@ -173,10 +173,7 @@ func cohereHandler(c *gin.Context, resp *http.Response, modelName string, prompt if err != nil { return service.OpenAIErrorWrapper(err, "read_response_body_failed", http.StatusInternalServerError), nil } - err = resp.Body.Close() - if err != nil { - return service.OpenAIErrorWrapper(err, "close_response_body_failed", http.StatusInternalServerError), nil - } + common.CloseResponseBodyGracefully(resp) var cohereResp CohereResponseResult err = json.Unmarshal(responseBody, &cohereResp) if err != nil { @@ -217,10 +214,7 @@ func cohereRerankHandler(c *gin.Context, resp *http.Response, info *relaycommon. if err != nil { return service.OpenAIErrorWrapper(err, "read_response_body_failed", http.StatusInternalServerError), nil } - err = resp.Body.Close() - if err != nil { - return service.OpenAIErrorWrapper(err, "close_response_body_failed", http.StatusInternalServerError), nil - } + common.CloseResponseBodyGracefully(resp) var cohereResp CohereRerankResponseResult err = json.Unmarshal(responseBody, &cohereResp) if err != nil { diff --git a/relay/channel/coze/relay-coze.go b/relay/channel/coze/relay-coze.go index ac76476f2023..6c08261b9844 100644 --- a/relay/channel/coze/relay-coze.go +++ b/relay/channel/coze/relay-coze.go @@ -48,10 +48,7 @@ func cozeChatHandler(c *gin.Context, resp *http.Response, info *relaycommon.Rela if err != nil { return service.OpenAIErrorWrapper(err, "read_response_body_failed", http.StatusInternalServerError), nil } - err = resp.Body.Close() - if err != nil { - return service.OpenAIErrorWrapperLocal(err, "close_response_body_failed", http.StatusInternalServerError), nil - } + common.CloseResponseBodyGracefully(resp) // convert coze response to openai response var response dto.TextResponse var cozeResponse CozeChatDetailResponse diff --git a/relay/channel/dify/relay-dify.go b/relay/channel/dify/relay-dify.go index 115aed1b00f0..c330c79118c2 100644 --- a/relay/channel/dify/relay-dify.go +++ b/relay/channel/dify/relay-dify.go @@ -257,10 +257,7 @@ func difyHandler(c *gin.Context, resp *http.Response, info *relaycommon.RelayInf if err != nil { return service.OpenAIErrorWrapper(err, "read_response_body_failed", http.StatusInternalServerError), nil } - err = resp.Body.Close() - if err != nil { - return service.OpenAIErrorWrapper(err, "close_response_body_failed", http.StatusInternalServerError), nil - } + common.CloseResponseBodyGracefully(resp) err = json.Unmarshal(responseBody, &difyResponse) if err != nil { return service.OpenAIErrorWrapper(err, "unmarshal_response_body_failed", http.StatusInternalServerError), nil diff --git a/relay/channel/gemini/relay-gemini-native.go b/relay/channel/gemini/relay-gemini-native.go index 39757ceffcb1..822d30975cf8 100644 --- a/relay/channel/gemini/relay-gemini-native.go +++ b/relay/channel/gemini/relay-gemini-native.go @@ -20,10 +20,7 @@ func GeminiTextGenerationHandler(c *gin.Context, resp *http.Response, info *rela if err != nil { return nil, service.OpenAIErrorWrapper(err, "read_response_body_failed", http.StatusInternalServerError) } - err = resp.Body.Close() - if err != nil { - return nil, service.OpenAIErrorWrapper(err, "close_response_body_failed", http.StatusInternalServerError) - } + common.CloseResponseBodyGracefully(resp) if common.DebugEnabled { println(string(responseBody)) diff --git a/relay/channel/gemini/relay-gemini.go b/relay/channel/gemini/relay-gemini.go index 18edfd04d229..b01d46e4e9c8 100644 --- a/relay/channel/gemini/relay-gemini.go +++ b/relay/channel/gemini/relay-gemini.go @@ -78,26 +78,7 @@ func clampThinkingBudget(modelName string, budget int) int { return budget } -// Setting safety to the lowest possible values since Gemini is already powerless enough -func CovertGemini2OpenAI(textRequest dto.GeneralOpenAIRequest, info *relaycommon.RelayInfo) (*GeminiChatRequest, error) { - - geminiRequest := GeminiChatRequest{ - Contents: make([]GeminiChatContent, 0, len(textRequest.Messages)), - GenerationConfig: GeminiChatGenerationConfig{ - Temperature: textRequest.Temperature, - TopP: textRequest.TopP, - MaxOutputTokens: textRequest.MaxTokens, - Seed: int64(textRequest.Seed), - }, - } - - if model_setting.IsGeminiModelSupportImagine(info.UpstreamModelName) { - geminiRequest.GenerationConfig.ResponseModalities = []string{ - "TEXT", - "IMAGE", - } - } - +func ThinkingAdaptor(geminiRequest *GeminiChatRequest, info *relaycommon.RelayInfo) { if model_setting.GetGeminiSettings().ThinkingAdapterEnabled { modelName := info.UpstreamModelName isNew25Pro := strings.HasPrefix(modelName, "gemini-2.5-pro") && @@ -150,6 +131,29 @@ func CovertGemini2OpenAI(textRequest dto.GeneralOpenAIRequest, info *relaycommon } } } +} + +// Setting safety to the lowest possible values since Gemini is already powerless enough +func CovertGemini2OpenAI(textRequest dto.GeneralOpenAIRequest, info *relaycommon.RelayInfo) (*GeminiChatRequest, error) { + + geminiRequest := GeminiChatRequest{ + Contents: make([]GeminiChatContent, 0, len(textRequest.Messages)), + GenerationConfig: GeminiChatGenerationConfig{ + Temperature: textRequest.Temperature, + TopP: textRequest.TopP, + MaxOutputTokens: textRequest.MaxTokens, + Seed: int64(textRequest.Seed), + }, + } + + if model_setting.IsGeminiModelSupportImagine(info.UpstreamModelName) { + geminiRequest.GenerationConfig.ResponseModalities = []string{ + "TEXT", + "IMAGE", + } + } + + ThinkingAdaptor(&geminiRequest, info) safetySettings := make([]GeminiChatSafetySettings, 0, len(SafetySettingList)) for _, category := range SafetySettingList { @@ -862,10 +866,7 @@ func GeminiChatHandler(c *gin.Context, resp *http.Response, info *relaycommon.Re if err != nil { return service.OpenAIErrorWrapper(err, "read_response_body_failed", http.StatusInternalServerError), nil } - err = resp.Body.Close() - if err != nil { - return service.OpenAIErrorWrapper(err, "close_response_body_failed", http.StatusInternalServerError), nil - } + common.CloseResponseBodyGracefully(resp) if common.DebugEnabled { println(string(responseBody)) } diff --git a/relay/channel/mokaai/relay-mokaai.go b/relay/channel/mokaai/relay-mokaai.go index d7580d7a9178..645475ddc959 100644 --- a/relay/channel/mokaai/relay-mokaai.go +++ b/relay/channel/mokaai/relay-mokaai.go @@ -5,6 +5,7 @@ import ( "github.com/gin-gonic/gin" "io" "net/http" + "one-api/common" "one-api/dto" "one-api/service" ) @@ -26,7 +27,7 @@ func embeddingRequestOpenAI2Moka(request dto.GeneralOpenAIRequest) *dto.Embeddin } return &dto.EmbeddingRequest{ Input: input, - Model: request.Model, + Model: request.Model, } } @@ -53,10 +54,7 @@ func mokaEmbeddingHandler(c *gin.Context, resp *http.Response) (*dto.OpenAIError if err != nil { return service.OpenAIErrorWrapper(err, "read_response_body_failed", http.StatusInternalServerError), nil } - err = resp.Body.Close() - if err != nil { - return service.OpenAIErrorWrapper(err, "close_response_body_failed", http.StatusInternalServerError), nil - } + common.CloseResponseBodyGracefully(resp) err = json.Unmarshal(responseBody, &baiduResponse) if err != nil { return service.OpenAIErrorWrapper(err, "unmarshal_response_body_failed", http.StatusInternalServerError), nil @@ -80,4 +78,3 @@ func mokaEmbeddingHandler(c *gin.Context, resp *http.Response) (*dto.OpenAIError _, err = c.Writer.Write(jsonResponse) return nil, &fullTextResponse.Usage } - diff --git a/relay/channel/ollama/relay-ollama.go b/relay/channel/ollama/relay-ollama.go index 89a046461fa1..aa1ec441ed16 100644 --- a/relay/channel/ollama/relay-ollama.go +++ b/relay/channel/ollama/relay-ollama.go @@ -7,6 +7,7 @@ import ( "github.com/gin-gonic/gin" "io" "net/http" + "one-api/common" "one-api/dto" "one-api/service" "strings" @@ -88,10 +89,7 @@ func ollamaEmbeddingHandler(c *gin.Context, resp *http.Response, promptTokens in if err != nil { return service.OpenAIErrorWrapper(err, "read_response_body_failed", http.StatusInternalServerError), nil } - err = resp.Body.Close() - if err != nil { - return service.OpenAIErrorWrapper(err, "close_response_body_failed", http.StatusInternalServerError), nil - } + common.CloseResponseBodyGracefully(resp) err = json.Unmarshal(responseBody, &ollamaEmbeddingResponse) if err != nil { return service.OpenAIErrorWrapper(err, "unmarshal_response_body_failed", http.StatusInternalServerError), nil @@ -141,10 +139,7 @@ func ollamaEmbeddingHandler(c *gin.Context, resp *http.Response, promptTokens in if err != nil { return service.OpenAIErrorWrapper(err, "copy_response_body_failed", http.StatusInternalServerError), nil } - err = resp.Body.Close() - if err != nil { - return service.OpenAIErrorWrapper(err, "close_response_body_failed", http.StatusInternalServerError), nil - } + common.CloseResponseBodyGracefully(resp) return nil, usage } diff --git a/relay/channel/openai/adaptor.go b/relay/channel/openai/adaptor.go index 8358f3e2e54a..424fd3dfaaf9 100644 --- a/relay/channel/openai/adaptor.go +++ b/relay/channel/openai/adaptor.go @@ -159,6 +159,11 @@ func (a *Adaptor) ConvertOpenAIRequest(c *gin.Context, info *relaycommon.RelayIn if info.ChannelType != common.ChannelTypeOpenAI && info.ChannelType != common.ChannelTypeAzure { request.StreamOptions = nil } + if info.ChannelType == common.ChannelTypeOpenRouter { + if len(request.Usage) == 0 { + request.Usage = json.RawMessage(`{"include":true}`) + } + } if strings.HasPrefix(request.Model, "o") { if request.MaxCompletionTokens == 0 && request.MaxTokens != 0 { request.MaxCompletionTokens = request.MaxTokens diff --git a/relay/channel/openai/relay-openai.go b/relay/channel/openai/relay-openai.go index 71590cd62451..c2def5d937ea 100644 --- a/relay/channel/openai/relay-openai.go +++ b/relay/channel/openai/relay-openai.go @@ -2,7 +2,6 @@ package openai import ( "bytes" - "encoding/json" "fmt" "io" "math" @@ -111,12 +110,13 @@ func OaiStreamHandler(c *gin.Context, resp *http.Response, info *relaycommon.Rel return service.OpenAIErrorWrapper(fmt.Errorf("invalid response"), "invalid_response", http.StatusInternalServerError), nil } - containStreamUsage := false + defer common.CloseResponseBodyGracefully(resp) + + model := info.UpstreamModelName var responseId string var createAt int64 = 0 var systemFingerprint string - model := info.UpstreamModelName - + var containStreamUsage bool var responseTextBuilder strings.Builder var toolCount int var usage = &dto.Usage{} @@ -148,31 +148,15 @@ func OaiStreamHandler(c *gin.Context, resp *http.Response, info *relaycommon.Rel return true }) + // 处理最后的响应 shouldSendLastResp := true - var lastStreamResponse dto.ChatCompletionsStreamResponse - err := common.DecodeJsonStr(lastStreamData, &lastStreamResponse) - if err == nil { - responseId = lastStreamResponse.Id - createAt = lastStreamResponse.Created - systemFingerprint = lastStreamResponse.GetSystemFingerprint() - model = lastStreamResponse.Model - if service.ValidUsage(lastStreamResponse.Usage) { - containStreamUsage = true - usage = lastStreamResponse.Usage - if !info.ShouldIncludeUsage { - shouldSendLastResp = false - } - } - for _, choice := range lastStreamResponse.Choices { - if choice.FinishReason != nil { - shouldSendLastResp = true - } - } + if err := handleLastResponse(lastStreamData, &responseId, &createAt, &systemFingerprint, &model, &usage, + &containStreamUsage, info, &shouldSendLastResp); err != nil { + common.SysError("error handling last response: " + err.Error()) } - if shouldSendLastResp { - sendStreamData(c, info, lastStreamData, forceFormat, thinkToContent) - //err = handleStreamFormat(c, info, lastStreamData, forceFormat, thinkToContent) + if shouldSendLastResp && info.RelayFormat == relaycommon.RelayFormatOpenAI { + _ = sendStreamData(c, info, lastStreamData, forceFormat, thinkToContent) } // 处理token计算 @@ -202,10 +186,7 @@ func OpenaiHandler(c *gin.Context, resp *http.Response, info *relaycommon.RelayI if err != nil { return service.OpenAIErrorWrapper(err, "read_response_body_failed", http.StatusInternalServerError), nil } - err = resp.Body.Close() - if err != nil { - return service.OpenAIErrorWrapper(err, "close_response_body_failed", http.StatusInternalServerError), nil - } + common.CloseResponseBodyGracefully(resp) err = common.DecodeJson(responseBody, &simpleResponse) if err != nil { return service.OpenAIErrorWrapper(err, "unmarshal_response_body_failed", http.StatusInternalServerError), nil @@ -238,7 +219,7 @@ func OpenaiHandler(c *gin.Context, resp *http.Response, info *relaycommon.RelayI switch info.RelayFormat { case relaycommon.RelayFormatOpenAI: if forceFormat { - responseBody, err = json.Marshal(simpleResponse) + responseBody, err = common.EncodeJson(simpleResponse) if err != nil { return service.OpenAIErrorWrapper(err, "marshal_response_body_failed", http.StatusInternalServerError), nil } @@ -247,29 +228,15 @@ func OpenaiHandler(c *gin.Context, resp *http.Response, info *relaycommon.RelayI } case relaycommon.RelayFormatClaude: claudeResp := service.ResponseOpenAI2Claude(&simpleResponse, info) - claudeRespStr, err := json.Marshal(claudeResp) + claudeRespStr, err := common.EncodeJson(claudeResp) if err != nil { return service.OpenAIErrorWrapper(err, "marshal_response_body_failed", http.StatusInternalServerError), nil } responseBody = claudeRespStr } - // Reset response body - resp.Body = io.NopCloser(bytes.NewBuffer(responseBody)) - // We shouldn't set the header before we parse the response body, because the parse part may fail. - // And then we will have to send an error response, but in this case, the header has already been set. - // So the httpClient will be confused by the response. - // For example, Postman will report error, and we cannot check the response at all. - for k, v := range resp.Header { - c.Writer.Header().Set(k, v[0]) - } - c.Writer.WriteHeader(resp.StatusCode) - _, err = io.Copy(c.Writer, resp.Body) - if err != nil { - //return service.OpenAIErrorWrapper(err, "copy_response_body_failed", http.StatusInternalServerError), nil - common.SysError("error copying response body: " + err.Error()) - } - resp.Body.Close() + common.IOCopyBytesGracefully(c, resp, responseBody) + return nil, &simpleResponse.Usage } @@ -280,7 +247,7 @@ func OpenaiTTSHandler(c *gin.Context, resp *http.Response, info *relaycommon.Rel // if the upstream returns a specific status code, once the upstream has already written the header, // the subsequent failure of the response body should be regarded as a non-recoverable error, // and can be terminated directly. - defer resp.Body.Close() + defer common.CloseResponseBodyGracefully(resp) usage := &dto.Usage{} usage.PromptTokens = info.PromptTokens usage.TotalTokens = info.PromptTokens @@ -306,25 +273,10 @@ func OpenaiSTTHandler(c *gin.Context, resp *http.Response, info *relaycommon.Rel if err != nil { return service.OpenAIErrorWrapper(err, "read_response_body_failed", http.StatusInternalServerError), nil } - err = resp.Body.Close() - if err != nil { - return service.OpenAIErrorWrapper(err, "close_response_body_failed", http.StatusInternalServerError), nil - } - // Reset response body - resp.Body = io.NopCloser(bytes.NewBuffer(responseBody)) - // We shouldn't set the header before we parse the response body, because the parse part may fail. - // And then we will have to send an error response, but in this case, the header has already been set. - // So the httpClient will be confused by the response. - // For example, Postman will report error, and we cannot check the response at all. - for k, v := range resp.Header { - c.Writer.Header().Set(k, v[0]) - } - c.Writer.WriteHeader(resp.StatusCode) - _, err = io.Copy(c.Writer, resp.Body) - if err != nil { - return service.OpenAIErrorWrapper(err, "copy_response_body_failed", http.StatusInternalServerError), nil - } - resp.Body.Close() + common.CloseResponseBodyGracefully(resp) + + // 写入新的 response body + common.IOCopyBytesGracefully(c, resp, responseBody) usage := &dto.Usage{} usage.PromptTokens = audioTokens @@ -415,7 +367,7 @@ func OpenaiRealtimeHandler(c *gin.Context, info *relaycommon.RelayInfo) (*dto.Op } realtimeEvent := &dto.RealtimeEvent{} - err = json.Unmarshal(message, realtimeEvent) + err = common.DecodeJson(message, realtimeEvent) if err != nil { errChan <- fmt.Errorf("error unmarshalling message: %v", err) return @@ -475,7 +427,7 @@ func OpenaiRealtimeHandler(c *gin.Context, info *relaycommon.RelayInfo) (*dto.Op } info.SetFirstResponseTime() realtimeEvent := &dto.RealtimeEvent{} - err = json.Unmarshal(message, realtimeEvent) + err = common.DecodeJson(message, realtimeEvent) if err != nil { errChan <- fmt.Errorf("error unmarshalling message: %v", err) return @@ -522,9 +474,9 @@ func OpenaiRealtimeHandler(c *gin.Context, info *relaycommon.RelayInfo) (*dto.Op localUsage = &dto.RealtimeUsage{} // print now usage } - //common.LogInfo(c, fmt.Sprintf("realtime streaming sumUsage: %v", sumUsage)) - //common.LogInfo(c, fmt.Sprintf("realtime streaming localUsage: %v", localUsage)) - //common.LogInfo(c, fmt.Sprintf("realtime streaming localUsage: %v", localUsage)) + common.LogInfo(c, fmt.Sprintf("realtime streaming sumUsage: %v", sumUsage)) + common.LogInfo(c, fmt.Sprintf("realtime streaming localUsage: %v", localUsage)) + common.LogInfo(c, fmt.Sprintf("realtime streaming localUsage: %v", localUsage)) } else if realtimeEvent.Type == dto.RealtimeEventTypeSessionUpdated || realtimeEvent.Type == dto.RealtimeEventTypeSessionCreated { realtimeSession := realtimeEvent.Session @@ -605,36 +557,22 @@ func OpenaiHandlerWithUsage(c *gin.Context, resp *http.Response, info *relaycomm if err != nil { return service.OpenAIErrorWrapper(err, "read_response_body_failed", http.StatusInternalServerError), nil } - err = resp.Body.Close() - if err != nil { - return service.OpenAIErrorWrapper(err, "close_response_body_failed", http.StatusInternalServerError), nil - } - // Reset response body - resp.Body = io.NopCloser(bytes.NewBuffer(responseBody)) - // We shouldn't set the header before we parse the response body, because the parse part may fail. - // And then we will have to send an error response, but in this case, the header has already been set. - // So the httpClient will be confused by the response. - // For example, Postman will report error, and we cannot check the response at all. - for k, v := range resp.Header { - c.Writer.Header().Set(k, v[0]) - } - // reset content length - c.Writer.Header().Set("Content-Length", fmt.Sprintf("%d", len(responseBody))) - c.Writer.WriteHeader(resp.StatusCode) - _, err = io.Copy(c.Writer, resp.Body) - if err != nil { - return service.OpenAIErrorWrapper(err, "copy_response_body_failed", http.StatusInternalServerError), nil - } - err = resp.Body.Close() - if err != nil { - return service.OpenAIErrorWrapper(err, "close_response_body_failed", http.StatusInternalServerError), nil - } var usageResp dto.SimpleResponse - err = json.Unmarshal(responseBody, &usageResp) + err = common.DecodeJson(responseBody, &usageResp) if err != nil { return service.OpenAIErrorWrapper(err, "parse_response_body_failed", http.StatusInternalServerError), nil } + + // 关闭旧的 response body(已被读取,再次读取会导致错误) + common.CloseResponseBodyGracefully(resp) + + // 写入新的 response body + common.IOCopyBytesGracefully(c, resp, responseBody) + + // Once we've written to the client, we should not return errors anymore + // because the upstream has already consumed resources and returned content + // We should still perform billing even if parsing fails // format if usageResp.InputTokens > 0 { usageResp.PromptTokens += usageResp.InputTokens diff --git a/relay/channel/openai/relay_responses.go b/relay/channel/openai/relay_responses.go index da9382c32d70..f7eae7d37547 100644 --- a/relay/channel/openai/relay_responses.go +++ b/relay/channel/openai/relay_responses.go @@ -22,10 +22,7 @@ func OaiResponsesHandler(c *gin.Context, resp *http.Response, info *relaycommon. if err != nil { return service.OpenAIErrorWrapper(err, "read_response_body_failed", http.StatusInternalServerError), nil } - err = resp.Body.Close() - if err != nil { - return service.OpenAIErrorWrapper(err, "close_response_body_failed", http.StatusInternalServerError), nil - } + common.CloseResponseBodyGracefully(resp) err = common.DecodeJson(responseBody, &responsesResponse) if err != nil { return service.OpenAIErrorWrapper(err, "unmarshal_response_body_failed", http.StatusInternalServerError), nil diff --git a/relay/channel/palm/relay-palm.go b/relay/channel/palm/relay-palm.go index 9d3dbd67aef8..44c60713c219 100644 --- a/relay/channel/palm/relay-palm.go +++ b/relay/channel/palm/relay-palm.go @@ -83,12 +83,7 @@ func palmStreamHandler(c *gin.Context, resp *http.Response) (*dto.OpenAIErrorWit stopChan <- true return } - err = resp.Body.Close() - if err != nil { - common.SysError("error closing stream response: " + err.Error()) - stopChan <- true - return - } + common.CloseResponseBodyGracefully(resp) var palmResponse PaLMChatResponse err = json.Unmarshal(responseBody, &palmResponse) if err != nil { @@ -122,10 +117,7 @@ func palmStreamHandler(c *gin.Context, resp *http.Response) (*dto.OpenAIErrorWit return false } }) - err := resp.Body.Close() - if err != nil { - return service.OpenAIErrorWrapper(err, "close_response_body_failed", http.StatusInternalServerError), "" - } + common.CloseResponseBodyGracefully(resp) return nil, responseText } @@ -134,10 +126,7 @@ func palmHandler(c *gin.Context, resp *http.Response, promptTokens int, model st if err != nil { return service.OpenAIErrorWrapper(err, "read_response_body_failed", http.StatusInternalServerError), nil } - err = resp.Body.Close() - if err != nil { - return service.OpenAIErrorWrapper(err, "close_response_body_failed", http.StatusInternalServerError), nil - } + common.CloseResponseBodyGracefully(resp) var palmResponse PaLMChatResponse err = json.Unmarshal(responseBody, &palmResponse) if err != nil { diff --git a/relay/channel/siliconflow/relay-siliconflow.go b/relay/channel/siliconflow/relay-siliconflow.go index a01e745caf93..a52ebfdaef44 100644 --- a/relay/channel/siliconflow/relay-siliconflow.go +++ b/relay/channel/siliconflow/relay-siliconflow.go @@ -5,6 +5,7 @@ import ( "github.com/gin-gonic/gin" "io" "net/http" + "one-api/common" "one-api/dto" "one-api/service" ) @@ -14,10 +15,7 @@ func siliconflowRerankHandler(c *gin.Context, resp *http.Response) (*dto.OpenAIE if err != nil { return service.OpenAIErrorWrapper(err, "read_response_body_failed", http.StatusInternalServerError), nil } - err = resp.Body.Close() - if err != nil { - return service.OpenAIErrorWrapper(err, "close_response_body_failed", http.StatusInternalServerError), nil - } + common.CloseResponseBodyGracefully(resp) var siliconflowResp SFRerankResponse err = json.Unmarshal(responseBody, &siliconflowResp) if err != nil { diff --git a/relay/channel/task/jimeng/adaptor.go b/relay/channel/task/jimeng/adaptor.go new file mode 100644 index 000000000000..b3610c3f37df --- /dev/null +++ b/relay/channel/task/jimeng/adaptor.go @@ -0,0 +1,380 @@ +package jimeng + +import ( + "bytes" + "crypto/hmac" + "crypto/sha256" + "encoding/hex" + "encoding/json" + "fmt" + "io" + "net/http" + "net/url" + "one-api/model" + "sort" + "strings" + "time" + + "github.com/gin-gonic/gin" + "github.com/pkg/errors" + + "one-api/common" + "one-api/constant" + "one-api/dto" + "one-api/relay/channel" + relaycommon "one-api/relay/common" + "one-api/service" +) + +// ============================ +// Request / Response structures +// ============================ + +type requestPayload struct { + ReqKey string `json:"req_key"` + BinaryDataBase64 []string `json:"binary_data_base64,omitempty"` + ImageUrls []string `json:"image_urls,omitempty"` + Prompt string `json:"prompt,omitempty"` + Seed int64 `json:"seed"` + AspectRatio string `json:"aspect_ratio"` +} + +type responsePayload struct { + Code int `json:"code"` + Message string `json:"message"` + RequestId string `json:"request_id"` + Data struct { + TaskID string `json:"task_id"` + } `json:"data"` +} + +type responseTask struct { + Code int `json:"code"` + Data struct { + BinaryDataBase64 []interface{} `json:"binary_data_base64"` + ImageUrls interface{} `json:"image_urls"` + RespData string `json:"resp_data"` + Status string `json:"status"` + VideoUrl string `json:"video_url"` + } `json:"data"` + Message string `json:"message"` + RequestId string `json:"request_id"` + Status int `json:"status"` + TimeElapsed string `json:"time_elapsed"` +} + +// ============================ +// Adaptor implementation +// ============================ + +type TaskAdaptor struct { + ChannelType int + accessKey string + secretKey string + baseURL string +} + +func (a *TaskAdaptor) Init(info *relaycommon.TaskRelayInfo) { + a.ChannelType = info.ChannelType + a.baseURL = info.BaseUrl + + // apiKey format: "access_key,secret_key" + keyParts := strings.Split(info.ApiKey, ",") + if len(keyParts) == 2 { + a.accessKey = strings.TrimSpace(keyParts[0]) + a.secretKey = strings.TrimSpace(keyParts[1]) + } +} + +// ValidateRequestAndSetAction parses body, validates fields and sets default action. +func (a *TaskAdaptor) ValidateRequestAndSetAction(c *gin.Context, info *relaycommon.TaskRelayInfo) (taskErr *dto.TaskError) { + // Accept only POST /v1/video/generations as "generate" action. + action := constant.TaskActionGenerate + info.Action = action + + req := relaycommon.TaskSubmitReq{} + if err := common.UnmarshalBodyReusable(c, &req); err != nil { + taskErr = service.TaskErrorWrapperLocal(err, "invalid_request", http.StatusBadRequest) + return + } + if strings.TrimSpace(req.Prompt) == "" { + taskErr = service.TaskErrorWrapperLocal(fmt.Errorf("prompt is required"), "invalid_request", http.StatusBadRequest) + return + } + + // Store into context for later usage + c.Set("task_request", req) + return nil +} + +// BuildRequestURL constructs the upstream URL. +func (a *TaskAdaptor) BuildRequestURL(info *relaycommon.TaskRelayInfo) (string, error) { + return fmt.Sprintf("%s/?Action=CVSync2AsyncSubmitTask&Version=2022-08-31", a.baseURL), nil +} + +// BuildRequestHeader sets required headers. +func (a *TaskAdaptor) BuildRequestHeader(c *gin.Context, req *http.Request, info *relaycommon.TaskRelayInfo) error { + req.Header.Set("Content-Type", "application/json") + req.Header.Set("Accept", "application/json") + return a.signRequest(req, a.accessKey, a.secretKey) +} + +// BuildRequestBody converts request into Jimeng specific format. +func (a *TaskAdaptor) BuildRequestBody(c *gin.Context, info *relaycommon.TaskRelayInfo) (io.Reader, error) { + v, exists := c.Get("task_request") + if !exists { + return nil, fmt.Errorf("request not found in context") + } + req := v.(relaycommon.TaskSubmitReq) + + body, err := a.convertToRequestPayload(&req) + if err != nil { + return nil, errors.Wrap(err, "convert request payload failed") + } + data, err := json.Marshal(body) + if err != nil { + return nil, err + } + return bytes.NewReader(data), nil +} + +// DoRequest delegates to common helper. +func (a *TaskAdaptor) DoRequest(c *gin.Context, info *relaycommon.TaskRelayInfo, requestBody io.Reader) (*http.Response, error) { + return channel.DoTaskApiRequest(a, c, info, requestBody) +} + +// DoResponse handles upstream response, returns taskID etc. +func (a *TaskAdaptor) DoResponse(c *gin.Context, resp *http.Response, info *relaycommon.TaskRelayInfo) (taskID string, taskData []byte, taskErr *dto.TaskError) { + responseBody, err := io.ReadAll(resp.Body) + if err != nil { + taskErr = service.TaskErrorWrapper(err, "read_response_body_failed", http.StatusInternalServerError) + return + } + _ = resp.Body.Close() + + // Parse Jimeng response + var jResp responsePayload + if err := json.Unmarshal(responseBody, &jResp); err != nil { + taskErr = service.TaskErrorWrapper(errors.Wrapf(err, "body: %s", responseBody), "unmarshal_response_body_failed", http.StatusInternalServerError) + return + } + + if jResp.Code != 10000 { + taskErr = service.TaskErrorWrapper(fmt.Errorf(jResp.Message), fmt.Sprintf("%d", jResp.Code), http.StatusInternalServerError) + return + } + + c.JSON(http.StatusOK, gin.H{"task_id": jResp.Data.TaskID}) + return jResp.Data.TaskID, responseBody, nil +} + +// FetchTask fetch task status +func (a *TaskAdaptor) FetchTask(baseUrl, key string, body map[string]any) (*http.Response, error) { + taskID, ok := body["task_id"].(string) + if !ok { + return nil, fmt.Errorf("invalid task_id") + } + + uri := fmt.Sprintf("%s/?Action=CVSync2AsyncGetResult&Version=2022-08-31", baseUrl) + payload := map[string]string{ + "req_key": "jimeng_vgfm_t2v_l20", // This is fixed value from doc: https://www.volcengine.com/docs/85621/1544774 + "task_id": taskID, + } + payloadBytes, err := json.Marshal(payload) + if err != nil { + return nil, errors.Wrap(err, "marshal fetch task payload failed") + } + + req, err := http.NewRequest(http.MethodPost, uri, bytes.NewBuffer(payloadBytes)) + if err != nil { + return nil, err + } + + req.Header.Set("Accept", "application/json") + req.Header.Set("Content-Type", "application/json") + + keyParts := strings.Split(key, ",") + if len(keyParts) != 2 { + return nil, fmt.Errorf("invalid api key format for jimeng: expected 'ak,sk'") + } + accessKey := strings.TrimSpace(keyParts[0]) + secretKey := strings.TrimSpace(keyParts[1]) + + if err := a.signRequest(req, accessKey, secretKey); err != nil { + return nil, errors.Wrap(err, "sign request failed") + } + + return service.GetHttpClient().Do(req) +} + +func (a *TaskAdaptor) GetModelList() []string { + return []string{"jimeng_vgfm_t2v_l20"} +} + +func (a *TaskAdaptor) GetChannelName() string { + return "jimeng" +} + +func (a *TaskAdaptor) signRequest(req *http.Request, accessKey, secretKey string) error { + var bodyBytes []byte + var err error + + if req.Body != nil { + bodyBytes, err = io.ReadAll(req.Body) + if err != nil { + return errors.Wrap(err, "read request body failed") + } + _ = req.Body.Close() + req.Body = io.NopCloser(bytes.NewBuffer(bodyBytes)) // Rewind + } else { + bodyBytes = []byte{} + } + + payloadHash := sha256.Sum256(bodyBytes) + hexPayloadHash := hex.EncodeToString(payloadHash[:]) + + t := time.Now().UTC() + xDate := t.Format("20060102T150405Z") + shortDate := t.Format("20060102") + + req.Header.Set("Host", req.URL.Host) + req.Header.Set("X-Date", xDate) + req.Header.Set("X-Content-Sha256", hexPayloadHash) + + // Sort and encode query parameters to create canonical query string + queryParams := req.URL.Query() + sortedKeys := make([]string, 0, len(queryParams)) + for k := range queryParams { + sortedKeys = append(sortedKeys, k) + } + sort.Strings(sortedKeys) + var queryParts []string + for _, k := range sortedKeys { + values := queryParams[k] + sort.Strings(values) + for _, v := range values { + queryParts = append(queryParts, fmt.Sprintf("%s=%s", url.QueryEscape(k), url.QueryEscape(v))) + } + } + canonicalQueryString := strings.Join(queryParts, "&") + + headersToSign := map[string]string{ + "host": req.URL.Host, + "x-date": xDate, + "x-content-sha256": hexPayloadHash, + } + if req.Header.Get("Content-Type") != "" { + headersToSign["content-type"] = req.Header.Get("Content-Type") + } + + var signedHeaderKeys []string + for k := range headersToSign { + signedHeaderKeys = append(signedHeaderKeys, k) + } + sort.Strings(signedHeaderKeys) + + var canonicalHeaders strings.Builder + for _, k := range signedHeaderKeys { + canonicalHeaders.WriteString(k) + canonicalHeaders.WriteString(":") + canonicalHeaders.WriteString(strings.TrimSpace(headersToSign[k])) + canonicalHeaders.WriteString("\n") + } + signedHeaders := strings.Join(signedHeaderKeys, ";") + + canonicalRequest := fmt.Sprintf("%s\n%s\n%s\n%s\n%s\n%s", + req.Method, + req.URL.Path, + canonicalQueryString, + canonicalHeaders.String(), + signedHeaders, + hexPayloadHash, + ) + + hashedCanonicalRequest := sha256.Sum256([]byte(canonicalRequest)) + hexHashedCanonicalRequest := hex.EncodeToString(hashedCanonicalRequest[:]) + + region := "cn-north-1" + serviceName := "cv" + credentialScope := fmt.Sprintf("%s/%s/%s/request", shortDate, region, serviceName) + stringToSign := fmt.Sprintf("HMAC-SHA256\n%s\n%s\n%s", + xDate, + credentialScope, + hexHashedCanonicalRequest, + ) + + kDate := hmacSHA256([]byte(secretKey), []byte(shortDate)) + kRegion := hmacSHA256(kDate, []byte(region)) + kService := hmacSHA256(kRegion, []byte(serviceName)) + kSigning := hmacSHA256(kService, []byte("request")) + signature := hex.EncodeToString(hmacSHA256(kSigning, []byte(stringToSign))) + + authorization := fmt.Sprintf("HMAC-SHA256 Credential=%s/%s, SignedHeaders=%s, Signature=%s", + accessKey, + credentialScope, + signedHeaders, + signature, + ) + req.Header.Set("Authorization", authorization) + return nil +} + +func hmacSHA256(key []byte, data []byte) []byte { + h := hmac.New(sha256.New, key) + h.Write(data) + return h.Sum(nil) +} + +func (a *TaskAdaptor) convertToRequestPayload(req *relaycommon.TaskSubmitReq) (*requestPayload, error) { + r := requestPayload{ + ReqKey: "jimeng_vgfm_i2v_l20", + Prompt: req.Prompt, + AspectRatio: "16:9", // Default aspect ratio + Seed: -1, // Default to random + } + + // Handle one-of image_urls or binary_data_base64 + if req.Image != "" { + if strings.HasPrefix(req.Image, "http") { + r.ImageUrls = []string{req.Image} + } else { + r.BinaryDataBase64 = []string{req.Image} + } + } + metadata := req.Metadata + medaBytes, err := json.Marshal(metadata) + if err != nil { + return nil, errors.Wrap(err, "metadata marshal metadata failed") + } + err = json.Unmarshal(medaBytes, &r) + if err != nil { + return nil, errors.Wrap(err, "unmarshal metadata failed") + } + return &r, nil +} + +func (a *TaskAdaptor) ParseTaskResult(respBody []byte) (*relaycommon.TaskInfo, error) { + resTask := responseTask{} + if err := json.Unmarshal(respBody, &resTask); err != nil { + return nil, errors.Wrap(err, "unmarshal task result failed") + } + taskResult := relaycommon.TaskInfo{} + if resTask.Code == 10000 { + taskResult.Code = 0 + } else { + taskResult.Code = resTask.Code // todo uni code + taskResult.Reason = resTask.Message + taskResult.Status = model.TaskStatusFailure + taskResult.Progress = "100%" + } + switch resTask.Data.Status { + case "in_queue": + taskResult.Status = model.TaskStatusQueued + taskResult.Progress = "10%" + case "done": + taskResult.Status = model.TaskStatusSuccess + taskResult.Progress = "100%" + } + taskResult.Url = resTask.Data.VideoUrl + return &taskResult, nil +} diff --git a/relay/channel/task/kling/adaptor.go b/relay/channel/task/kling/adaptor.go index 9ea58728234d..afa392016a68 100644 --- a/relay/channel/task/kling/adaptor.go +++ b/relay/channel/task/kling/adaptor.go @@ -2,11 +2,12 @@ package kling import ( "bytes" - "context" "encoding/json" "fmt" + "github.com/samber/lo" "io" "net/http" + "one-api/model" "strings" "time" @@ -15,6 +16,7 @@ import ( "github.com/pkg/errors" "one-api/common" + "one-api/constant" "one-api/dto" "one-api/relay/channel" relaycommon "one-api/relay/common" @@ -41,16 +43,27 @@ type requestPayload struct { Mode string `json:"mode,omitempty"` Duration string `json:"duration,omitempty"` AspectRatio string `json:"aspect_ratio,omitempty"` - Model string `json:"model,omitempty"` ModelName string `json:"model_name,omitempty"` CfgScale float64 `json:"cfg_scale,omitempty"` } type responsePayload struct { - Code int `json:"code"` - Message string `json:"message"` - Data struct { - TaskID string `json:"task_id"` + Code int `json:"code"` + Message string `json:"message"` + RequestId string `json:"request_id"` + Data struct { + TaskId string `json:"task_id"` + TaskStatus string `json:"task_status"` + TaskStatusMsg string `json:"task_status_msg"` + TaskResult struct { + Videos []struct { + Id string `json:"id"` + Url string `json:"url"` + Duration string `json:"duration"` + } `json:"videos"` + } `json:"task_result"` + CreatedAt int64 `json:"created_at"` + UpdatedAt int64 `json:"updated_at"` } `json:"data"` } @@ -69,8 +82,8 @@ func (a *TaskAdaptor) Init(info *relaycommon.TaskRelayInfo) { a.ChannelType = info.ChannelType a.baseURL = info.BaseUrl - // apiKey format: "access_key,secret_key" - keyParts := strings.Split(info.ApiKey, ",") + // apiKey format: "access_key|secret_key" + keyParts := strings.Split(info.ApiKey, "|") if len(keyParts) == 2 { a.accessKey = strings.TrimSpace(keyParts[0]) a.secretKey = strings.TrimSpace(keyParts[1]) @@ -80,7 +93,7 @@ func (a *TaskAdaptor) Init(info *relaycommon.TaskRelayInfo) { // ValidateRequestAndSetAction parses body, validates fields and sets default action. func (a *TaskAdaptor) ValidateRequestAndSetAction(c *gin.Context, info *relaycommon.TaskRelayInfo) (taskErr *dto.TaskError) { // Accept only POST /v1/video/generations as "generate" action. - action := "generate" + action := constant.TaskActionGenerate info.Action = action var req SubmitReq @@ -94,13 +107,14 @@ func (a *TaskAdaptor) ValidateRequestAndSetAction(c *gin.Context, info *relaycom } // Store into context for later usage - c.Set("kling_request", req) + c.Set("task_request", req) return nil } // BuildRequestURL constructs the upstream URL. func (a *TaskAdaptor) BuildRequestURL(info *relaycommon.TaskRelayInfo) (string, error) { - return fmt.Sprintf("%s/v1/videos/image2video", a.baseURL), nil + path := lo.Ternary(info.Action == constant.TaskActionGenerate, "/v1/videos/image2video", "/v1/videos/text2video") + return fmt.Sprintf("%s%s", a.baseURL, path), nil } // BuildRequestHeader sets required headers. @@ -119,13 +133,16 @@ func (a *TaskAdaptor) BuildRequestHeader(c *gin.Context, req *http.Request, info // BuildRequestBody converts request into Kling specific format. func (a *TaskAdaptor) BuildRequestBody(c *gin.Context, info *relaycommon.TaskRelayInfo) (io.Reader, error) { - v, exists := c.Get("kling_request") + v, exists := c.Get("task_request") if !exists { return nil, fmt.Errorf("request not found in context") } req := v.(SubmitReq) - body := a.convertToRequestPayload(&req) + body, err := a.convertToRequestPayload(&req) + if err != nil { + return nil, err + } data, err := json.Marshal(body) if err != nil { return nil, err @@ -135,6 +152,9 @@ func (a *TaskAdaptor) BuildRequestBody(c *gin.Context, info *relaycommon.TaskRel // DoRequest delegates to common helper. func (a *TaskAdaptor) DoRequest(c *gin.Context, info *relaycommon.TaskRelayInfo, requestBody io.Reader) (*http.Response, error) { + if action := c.GetString("action"); action != "" { + info.Action = action + } return channel.DoTaskApiRequest(a, c, info, requestBody) } @@ -149,8 +169,8 @@ func (a *TaskAdaptor) DoResponse(c *gin.Context, resp *http.Response, info *rela // Attempt Kling response parse first. var kResp responsePayload if err := json.Unmarshal(responseBody, &kResp); err == nil && kResp.Code == 0 { - c.JSON(http.StatusOK, gin.H{"task_id": kResp.Data.TaskID}) - return kResp.Data.TaskID, responseBody, nil + c.JSON(http.StatusOK, gin.H{"task_id": kResp.Data.TaskId}) + return kResp.Data.TaskId, responseBody, nil } // Fallback generic task response. @@ -175,7 +195,12 @@ func (a *TaskAdaptor) FetchTask(baseUrl, key string, body map[string]any) (*http if !ok { return nil, fmt.Errorf("invalid task_id") } - url := fmt.Sprintf("%s/v1/videos/image2video/%s", baseUrl, taskID) + action, ok := body["action"].(string) + if !ok { + return nil, fmt.Errorf("invalid action") + } + path := lo.Ternary(action == constant.TaskActionGenerate, "/v1/videos/image2video", "/v1/videos/text2video") + url := fmt.Sprintf("%s%s/%s", baseUrl, path, taskID) req, err := http.NewRequest(http.MethodGet, url, nil) if err != nil { @@ -187,10 +212,6 @@ func (a *TaskAdaptor) FetchTask(baseUrl, key string, body map[string]any) (*http token = key } - ctx, cancel := context.WithTimeout(context.Background(), 15*time.Second) - defer cancel() - - req = req.WithContext(ctx) req.Header.Set("Accept", "application/json") req.Header.Set("Authorization", "Bearer "+token) req.Header.Set("User-Agent", "kling-sdk/1.0") @@ -210,22 +231,29 @@ func (a *TaskAdaptor) GetChannelName() string { // helpers // ============================ -func (a *TaskAdaptor) convertToRequestPayload(req *SubmitReq) *requestPayload { - r := &requestPayload{ +func (a *TaskAdaptor) convertToRequestPayload(req *SubmitReq) (*requestPayload, error) { + r := requestPayload{ Prompt: req.Prompt, Image: req.Image, Mode: defaultString(req.Mode, "std"), Duration: fmt.Sprintf("%d", defaultInt(req.Duration, 5)), AspectRatio: a.getAspectRatio(req.Size), - Model: req.Model, ModelName: req.Model, CfgScale: 0.5, } - if r.Model == "" { - r.Model = "kling-v1" + if r.ModelName == "" { r.ModelName = "kling-v1" } - return r + metadata := req.Metadata + medaBytes, err := json.Marshal(metadata) + if err != nil { + return nil, errors.Wrap(err, "metadata marshal metadata failed") + } + err = json.Unmarshal(medaBytes, &r) + if err != nil { + return nil, errors.Wrap(err, "unmarshal metadata failed") + } + return &r, nil } func (a *TaskAdaptor) getAspectRatio(size string) string { @@ -264,7 +292,7 @@ func (a *TaskAdaptor) createJWTToken() (string, error) { } func (a *TaskAdaptor) createJWTTokenWithKey(apiKey string) (string, error) { - parts := strings.Split(apiKey, ",") + parts := strings.Split(apiKey, "|") if len(parts) != 2 { return "", fmt.Errorf("invalid API key format, expected 'access_key,secret_key'") } @@ -286,27 +314,33 @@ func (a *TaskAdaptor) createJWTTokenWithKeys(accessKey, secretKey string) (strin return token.SignedString([]byte(secretKey)) } -// ParseResultUrl 提取视频任务结果的 url -func (a *TaskAdaptor) ParseResultUrl(resp map[string]any) (string, error) { - data, ok := resp["data"].(map[string]any) - if !ok { - return "", fmt.Errorf("data field not found or invalid") - } - taskResult, ok := data["task_result"].(map[string]any) - if !ok { - return "", fmt.Errorf("task_result field not found or invalid") - } - videos, ok := taskResult["videos"].([]interface{}) - if !ok || len(videos) == 0 { - return "", fmt.Errorf("videos field not found or empty") +func (a *TaskAdaptor) ParseTaskResult(respBody []byte) (*relaycommon.TaskInfo, error) { + resPayload := responsePayload{} + err := json.Unmarshal(respBody, &resPayload) + if err != nil { + return nil, errors.Wrap(err, "failed to unmarshal response body") } - video, ok := videos[0].(map[string]interface{}) - if !ok { - return "", fmt.Errorf("video item invalid") + taskInfo := &relaycommon.TaskInfo{} + taskInfo.Code = resPayload.Code + taskInfo.TaskID = resPayload.Data.TaskId + taskInfo.Reason = resPayload.Message + //任务状态,枚举值:submitted(已提交)、processing(处理中)、succeed(成功)、failed(失败) + status := resPayload.Data.TaskStatus + switch status { + case "submitted": + taskInfo.Status = model.TaskStatusSubmitted + case "processing": + taskInfo.Status = model.TaskStatusInProgress + case "succeed": + taskInfo.Status = model.TaskStatusSuccess + case "failed": + taskInfo.Status = model.TaskStatusFailure + default: + return nil, fmt.Errorf("unknown task status: %s", status) } - url, ok := video["url"].(string) - if !ok || url == "" { - return "", fmt.Errorf("url field not found or invalid") + if videos := resPayload.Data.TaskResult.Videos; len(videos) > 0 { + video := videos[0] + taskInfo.Url = video.Url } - return url, nil + return taskInfo, nil } diff --git a/relay/channel/task/suno/adaptor.go b/relay/channel/task/suno/adaptor.go index f7042348ee96..9c04c7ad4d9a 100644 --- a/relay/channel/task/suno/adaptor.go +++ b/relay/channel/task/suno/adaptor.go @@ -22,8 +22,8 @@ type TaskAdaptor struct { ChannelType int } -func (a *TaskAdaptor) ParseResultUrl(resp map[string]any) (string, error) { - return "", nil // todo implement this method if needed +func (a *TaskAdaptor) ParseTaskResult([]byte) (*relaycommon.TaskInfo, error) { + return nil, fmt.Errorf("not implement") // todo implement this method if needed } func (a *TaskAdaptor) Init(info *relaycommon.TaskRelayInfo) { diff --git a/relay/channel/tencent/relay-tencent.go b/relay/channel/tencent/relay-tencent.go index 1446e06e2202..a7106a88c8db 100644 --- a/relay/channel/tencent/relay-tencent.go +++ b/relay/channel/tencent/relay-tencent.go @@ -124,10 +124,7 @@ func tencentStreamHandler(c *gin.Context, resp *http.Response) (*dto.OpenAIError helper.Done(c) - err := resp.Body.Close() - if err != nil { - return service.OpenAIErrorWrapper(err, "close_response_body_failed", http.StatusInternalServerError), "" - } + common.CloseResponseBodyGracefully(resp) return nil, responseText } @@ -138,10 +135,7 @@ func tencentHandler(c *gin.Context, resp *http.Response) (*dto.OpenAIErrorWithSt if err != nil { return service.OpenAIErrorWrapper(err, "read_response_body_failed", http.StatusInternalServerError), nil } - err = resp.Body.Close() - if err != nil { - return service.OpenAIErrorWrapper(err, "close_response_body_failed", http.StatusInternalServerError), nil - } + common.CloseResponseBodyGracefully(resp) err = json.Unmarshal(responseBody, &tencentSb) if err != nil { return service.OpenAIErrorWrapper(err, "unmarshal_response_body_failed", http.StatusInternalServerError), nil diff --git a/relay/channel/xai/text.go b/relay/channel/xai/text.go index 408160fb11d7..9a300356735c 100644 --- a/relay/channel/xai/text.go +++ b/relay/channel/xai/text.go @@ -73,11 +73,7 @@ func xAIStreamHandler(c *gin.Context, resp *http.Response, info *relaycommon.Rel } helper.Done(c) - err := resp.Body.Close() - if err != nil { - //return service.OpenAIErrorWrapper(err, "close_response_body_failed", http.StatusInternalServerError), nil - common.SysError("close_response_body_failed: " + err.Error()) - } + common.CloseResponseBodyGracefully(resp) return nil, usage } @@ -110,10 +106,7 @@ func xAIHandler(c *gin.Context, resp *http.Response, info *relaycommon.RelayInfo if err != nil { return service.OpenAIErrorWrapper(err, "copy_response_body_failed", http.StatusInternalServerError), nil } - err = resp.Body.Close() - if err != nil { - return service.OpenAIErrorWrapper(err, "close_response_body_failed", http.StatusInternalServerError), nil - } + common.CloseResponseBodyGracefully(resp) return nil, &response.Usage } diff --git a/relay/channel/xinference/dto.go b/relay/channel/xinference/dto.go index 2f12ad1047ee..35f339fe6e4b 100644 --- a/relay/channel/xinference/dto.go +++ b/relay/channel/xinference/dto.go @@ -1,7 +1,7 @@ package xinference type XinRerankResponseDocument struct { - Document string `json:"document,omitempty"` + Document any `json:"document,omitempty"` Index int `json:"index"` RelevanceScore float64 `json:"relevance_score"` } diff --git a/relay/channel/zhipu/relay-zhipu.go b/relay/channel/zhipu/relay-zhipu.go index 744538e3f5ab..91cd384b8b8d 100644 --- a/relay/channel/zhipu/relay-zhipu.go +++ b/relay/channel/zhipu/relay-zhipu.go @@ -210,10 +210,7 @@ func zhipuStreamHandler(c *gin.Context, resp *http.Response) (*dto.OpenAIErrorWi return false } }) - err := resp.Body.Close() - if err != nil { - return service.OpenAIErrorWrapper(err, "close_response_body_failed", http.StatusInternalServerError), nil - } + common.CloseResponseBodyGracefully(resp) return nil, usage } @@ -223,10 +220,7 @@ func zhipuHandler(c *gin.Context, resp *http.Response) (*dto.OpenAIErrorWithStat if err != nil { return service.OpenAIErrorWrapper(err, "read_response_body_failed", http.StatusInternalServerError), nil } - err = resp.Body.Close() - if err != nil { - return service.OpenAIErrorWrapper(err, "close_response_body_failed", http.StatusInternalServerError), nil - } + common.CloseResponseBodyGracefully(resp) err = json.Unmarshal(responseBody, &zhipuResponse) if err != nil { return service.OpenAIErrorWrapper(err, "unmarshal_response_body_failed", http.StatusInternalServerError), nil diff --git a/relay/common/relay_info.go b/relay/common/relay_info.go index 3759c3633221..5fd94788af0d 100644 --- a/relay/common/relay_info.go +++ b/relay/common/relay_info.go @@ -65,8 +65,8 @@ type RelayInfo struct { TokenId int TokenKey string UserId int - Group string - UserGroup string + UsingGroup string // 使用的分组 + UserGroup string // 用户所在分组 TokenUnlimited bool StartTime time.Time FirstResponseTime time.Time @@ -219,7 +219,6 @@ func GenRelayInfo(c *gin.Context) *RelayInfo { tokenId := c.GetInt("token_id") tokenKey := c.GetString("token_key") userId := c.GetInt("id") - group := c.GetString("group") tokenUnlimited := c.GetBool("token_unlimited_quota") startTime := c.GetTime(constant.ContextKeyRequestStartTime) // firstResponseTime = time.Now() - 1 second @@ -239,7 +238,7 @@ func GenRelayInfo(c *gin.Context) *RelayInfo { TokenId: tokenId, TokenKey: tokenKey, UserId: userId, - Group: group, + UsingGroup: c.GetString(constant.ContextKeyUsingGroup), UserGroup: c.GetString(constant.ContextKeyUserGroup), TokenUnlimited: tokenUnlimited, StartTime: startTime, @@ -314,3 +313,22 @@ func GenTaskRelayInfo(c *gin.Context) *TaskRelayInfo { } return info } + +type TaskSubmitReq struct { + Prompt string `json:"prompt"` + Model string `json:"model,omitempty"` + Mode string `json:"mode,omitempty"` + Image string `json:"image,omitempty"` + Size string `json:"size,omitempty"` + Duration int `json:"duration,omitempty"` + Metadata map[string]interface{} `json:"metadata,omitempty"` +} + +type TaskInfo struct { + Code int `json:"code"` + TaskID string `json:"task_id"` + Status string `json:"status"` + Reason string `json:"reason,omitempty"` + Url string `json:"url,omitempty"` + Progress string `json:"progress,omitempty"` +} diff --git a/relay/common_handler/rerank.go b/relay/common_handler/rerank.go index 496278b5c06e..63ab4769f77f 100644 --- a/relay/common_handler/rerank.go +++ b/relay/common_handler/rerank.go @@ -16,10 +16,7 @@ func RerankHandler(c *gin.Context, info *relaycommon.RelayInfo, resp *http.Respo if err != nil { return service.OpenAIErrorWrapper(err, "read_response_body_failed", http.StatusInternalServerError), nil } - err = resp.Body.Close() - if err != nil { - return service.OpenAIErrorWrapper(err, "close_response_body_failed", http.StatusInternalServerError), nil - } + common.CloseResponseBodyGracefully(resp) if common.DebugEnabled { println("reranker response body: ", string(responseBody)) } @@ -38,10 +35,16 @@ func RerankHandler(c *gin.Context, info *relaycommon.RelayInfo, resp *http.Respo } if info.ReturnDocuments { var document any - if result.Document == "" { - document = info.Documents[result.Index] - } else { - document = result.Document + if result.Document != nil { + if doc, ok := result.Document.(string); ok { + if doc == "" { + document = info.Documents[result.Index] + } else { + document = doc + } + } else { + document = result.Document + } } respResult.Document = document } diff --git a/relay/constant/relay_mode.go b/relay/constant/relay_mode.go index 02a286e2f1c5..cc8a14945180 100644 --- a/relay/constant/relay_mode.go +++ b/relay/constant/relay_mode.go @@ -41,6 +41,9 @@ const ( RelayModeKlingFetchByID RelayModeKlingSubmit + RelayModeJimengFetchByID + RelayModeJimengSubmit + RelayModeRerank RelayModeResponses @@ -80,7 +83,7 @@ func Path2RelayMode(path string) int { relayMode = RelayModeRerank } else if strings.HasPrefix(path, "/v1/realtime") { relayMode = RelayModeRealtime - } else if strings.HasPrefix(path, "/v1beta/models") { + } else if strings.HasPrefix(path, "/v1beta/models") || strings.HasPrefix(path, "/v1/models") { relayMode = RelayModeGemini } return relayMode @@ -146,3 +149,13 @@ func Path2RelayKling(method, path string) int { } return relayMode } + +func Path2RelayJimeng(method, path string) int { + relayMode := RelayModeUnknown + if method == http.MethodPost && strings.HasSuffix(path, "/video/generations") { + relayMode = RelayModeJimengSubmit + } else if method == http.MethodGet && strings.Contains(path, "/video/generations/") { + relayMode = RelayModeJimengFetchByID + } + return relayMode +} diff --git a/relay/gemini_handler.go b/relay/gemini_handler.go index 14d58cc581e9..9185ce624e81 100644 --- a/relay/gemini_handler.go +++ b/relay/gemini_handler.go @@ -13,6 +13,7 @@ import ( "one-api/relay/helper" "one-api/service" "one-api/setting" + "one-api/setting/model_setting" "strings" "github.com/gin-gonic/gin" @@ -76,6 +77,33 @@ func getGeminiInputTokens(req *gemini.GeminiChatRequest, info *relaycommon.Relay return inputTokens } +func isNoThinkingRequest(req *gemini.GeminiChatRequest) bool { + if req.GenerationConfig.ThinkingConfig != nil && req.GenerationConfig.ThinkingConfig.ThinkingBudget != nil { + return *req.GenerationConfig.ThinkingConfig.ThinkingBudget <= 0 + } + return false +} + +func trimModelThinking(modelName string) string { + // 去除模型名称中的 -nothinking 后缀 + if strings.HasSuffix(modelName, "-nothinking") { + return strings.TrimSuffix(modelName, "-nothinking") + } + // 去除模型名称中的 -thinking 后缀 + if strings.HasSuffix(modelName, "-thinking") { + return strings.TrimSuffix(modelName, "-thinking") + } + + // 去除模型名称中的 -thinking-number + if strings.Contains(modelName, "-thinking-") { + parts := strings.Split(modelName, "-thinking-") + if len(parts) > 1 { + return parts[0] + "-thinking" + } + } + return modelName +} + func GeminiHelper(c *gin.Context) (openaiErr *dto.OpenAIErrorWithStatusCode) { req, err := getAndValidateGeminiRequest(c) if err != nil { @@ -107,12 +135,27 @@ func GeminiHelper(c *gin.Context) (openaiErr *dto.OpenAIErrorWithStatusCode) { relayInfo.SetPromptTokens(promptTokens) } else { promptTokens := getGeminiInputTokens(req, relayInfo) - if err != nil { - return service.OpenAIErrorWrapperLocal(err, "count_input_tokens_error", http.StatusBadRequest) - } c.Set("prompt_tokens", promptTokens) } + if model_setting.GetGeminiSettings().ThinkingAdapterEnabled { + if isNoThinkingRequest(req) { + // check is thinking + if !strings.Contains(relayInfo.OriginModelName, "-nothinking") { + // try to get no thinking model price + noThinkingModelName := relayInfo.OriginModelName + "-nothinking" + containPrice := helper.ContainPriceOrRatio(noThinkingModelName) + if containPrice { + relayInfo.OriginModelName = noThinkingModelName + relayInfo.UpstreamModelName = noThinkingModelName + } + } + } + if req.GenerationConfig.ThinkingConfig == nil { + gemini.ThinkingAdaptor(req, relayInfo) + } + } + priceData, err := helper.ModelPriceHelper(c, relayInfo, relayInfo.PromptTokens, int(req.GenerationConfig.MaxOutputTokens)) if err != nil { return service.OpenAIErrorWrapperLocal(err, "model_price_error", http.StatusInternalServerError) diff --git a/relay/helper/price.go b/relay/helper/price.go index 1ee2767e1653..ab614cbd5767 100644 --- a/relay/helper/price.go +++ b/relay/helper/price.go @@ -13,6 +13,7 @@ import ( type GroupRatioInfo struct { GroupRatio float64 GroupSpecialRatio float64 + HasSpecialRatio bool } type PriceData struct { @@ -31,7 +32,7 @@ func (p PriceData) ToSetting() string { return fmt.Sprintf("ModelPrice: %f, ModelRatio: %f, CompletionRatio: %f, CacheRatio: %f, GroupRatio: %f, UsePrice: %t, CacheCreationRatio: %f, ShouldPreConsumedQuota: %d, ImageRatio: %f", p.ModelPrice, p.ModelRatio, p.CompletionRatio, p.CacheRatio, p.GroupRatioInfo.GroupRatio, p.UsePrice, p.CacheCreationRatio, p.ShouldPreConsumedQuota, p.ImageRatio) } -// HandleGroupRatio checks for "auto_group" in the context and updates the group ratio and relayInfo.Group if present +// HandleGroupRatio checks for "auto_group" in the context and updates the group ratio and relayInfo.UsingGroup if present func HandleGroupRatio(ctx *gin.Context, relayInfo *relaycommon.RelayInfo) GroupRatioInfo { groupRatioInfo := GroupRatioInfo{ GroupRatio: 1.0, // default ratio @@ -44,18 +45,19 @@ func HandleGroupRatio(ctx *gin.Context, relayInfo *relaycommon.RelayInfo) GroupR if common.DebugEnabled { println(fmt.Sprintf("final group: %s", autoGroup)) } - relayInfo.Group = autoGroup.(string) + relayInfo.UsingGroup = autoGroup.(string) } // check user group special ratio - userGroupRatio, ok := ratio_setting.GetGroupGroupRatio(relayInfo.UserGroup, relayInfo.Group) + userGroupRatio, ok := ratio_setting.GetGroupGroupRatio(relayInfo.UserGroup, relayInfo.UsingGroup) if ok { // user group special ratio groupRatioInfo.GroupSpecialRatio = userGroupRatio groupRatioInfo.GroupRatio = userGroupRatio + groupRatioInfo.HasSpecialRatio = true } else { // normal group ratio - groupRatioInfo.GroupRatio = ratio_setting.GetGroupRatio(relayInfo.Group) + groupRatioInfo.GroupRatio = ratio_setting.GetGroupRatio(relayInfo.UsingGroup) } return groupRatioInfo @@ -120,6 +122,35 @@ func ModelPriceHelper(c *gin.Context, info *relaycommon.RelayInfo, promptTokens return priceData, nil } +type PerCallPriceData struct { + ModelPrice float64 + Quota int + GroupRatioInfo GroupRatioInfo +} + +// ModelPriceHelperPerCall 按次计费的 PriceHelper (MJ、Task) +func ModelPriceHelperPerCall(c *gin.Context, info *relaycommon.RelayInfo) PerCallPriceData { + groupRatioInfo := HandleGroupRatio(c, info) + + modelPrice, success := ratio_setting.GetModelPrice(info.OriginModelName, true) + // 如果没有配置价格,则使用默认价格 + if !success { + defaultPrice, ok := ratio_setting.GetDefaultModelRatioMap()[info.OriginModelName] + if !ok { + modelPrice = 0.1 + } else { + modelPrice = defaultPrice + } + } + quota := int(modelPrice * common.QuotaPerUnit * groupRatioInfo.GroupRatio) + priceData := PerCallPriceData{ + ModelPrice: modelPrice, + Quota: quota, + GroupRatioInfo: groupRatioInfo, + } + return priceData +} + func ContainPriceOrRatio(modelName string) bool { _, ok := ratio_setting.GetModelPrice(modelName, false) if ok { diff --git a/relay/relay-mj.go b/relay/relay-mj.go index ce4346b63876..b44890c1713c 100644 --- a/relay/relay-mj.go +++ b/relay/relay-mj.go @@ -13,9 +13,9 @@ import ( "one-api/model" relaycommon "one-api/relay/common" relayconstant "one-api/relay/constant" + "one-api/relay/helper" "one-api/service" "one-api/setting" - "one-api/setting/ratio_setting" "strconv" "strings" "time" @@ -174,18 +174,9 @@ func RelaySwapFace(c *gin.Context) *dto.MidjourneyResponse { return service.MidjourneyErrorWrapper(constant.MjRequestError, "sour_base64_and_target_base64_is_required") } modelName := service.CoverActionToModelName(constant.MjActionSwapFace) - modelPrice, success := ratio_setting.GetModelPrice(modelName, true) - // 如果没有配置价格,则使用默认价格 - if !success { - defaultPrice, ok := ratio_setting.GetDefaultModelRatioMap()[modelName] - if !ok { - modelPrice = 0.1 - } else { - modelPrice = defaultPrice - } - } - groupRatio := ratio_setting.GetGroupRatio(group) - ratio := modelPrice * groupRatio + + priceData := helper.ModelPriceHelperPerCall(c, relayInfo) + userQuota, err := model.GetUserQuota(userId, false) if err != nil { return &dto.MidjourneyResponse{ @@ -193,9 +184,8 @@ func RelaySwapFace(c *gin.Context) *dto.MidjourneyResponse { Description: err.Error(), } } - quota := int(ratio * common.QuotaPerUnit) - if userQuota-quota < 0 { + if userQuota-priceData.Quota < 0 { return &dto.MidjourneyResponse{ Code: 4, Description: "quota_not_enough", @@ -210,26 +200,18 @@ func RelaySwapFace(c *gin.Context) *dto.MidjourneyResponse { } defer func() { if mjResp.StatusCode == 200 && mjResp.Response.Code == 1 { - err := service.PostConsumeQuota(relayInfo, quota, 0, true) + err := service.PostConsumeQuota(relayInfo, priceData.Quota, 0, true) if err != nil { common.SysError("error consuming token remain quota: " + err.Error()) } - //err = model.CacheUpdateUserQuota(userId) - if err != nil { - common.SysError("error update user quota cache: " + err.Error()) - } - if quota != 0 { - tokenName := c.GetString("token_name") - logContent := fmt.Sprintf("模型固定价格 %.2f,分组倍率 %.2f,操作 %s", modelPrice, groupRatio, constant.MjActionSwapFace) - other := make(map[string]interface{}) - other["model_price"] = modelPrice - other["group_ratio"] = groupRatio - model.RecordConsumeLog(c, userId, channelId, 0, 0, modelName, tokenName, - quota, logContent, tokenId, userQuota, 0, false, group, other) - model.UpdateUserUsedQuotaAndRequestCount(userId, quota) - channelId := c.GetInt("channel_id") - model.UpdateChannelUsedQuota(channelId, quota) - } + + tokenName := c.GetString("token_name") + logContent := fmt.Sprintf("模型固定价格 %.2f,分组倍率 %.2f,操作 %s", priceData.ModelPrice, priceData.GroupRatioInfo.GroupRatio, constant.MjActionSwapFace) + other := service.GenerateMjOtherInfo(priceData) + model.RecordConsumeLog(c, userId, channelId, 0, 0, modelName, tokenName, + priceData.Quota, logContent, tokenId, userQuota, 0, false, group, other) + model.UpdateUserUsedQuotaAndRequestCount(userId, priceData.Quota) + model.UpdateChannelUsedQuota(channelId, priceData.Quota) } }() midjResponse := &mjResp.Response @@ -250,7 +232,7 @@ func RelaySwapFace(c *gin.Context) *dto.MidjourneyResponse { Progress: "0%", FailReason: "", ChannelId: c.GetInt("channel_id"), - Quota: quota, + Quota: priceData.Quota, } err = midjourneyTask.Insert() if err != nil { @@ -480,18 +462,9 @@ func RelayMidjourneySubmit(c *gin.Context, relayMode int) *dto.MidjourneyRespons fullRequestURL := fmt.Sprintf("%s%s", baseURL, requestURL) modelName := service.CoverActionToModelName(midjRequest.Action) - modelPrice, success := ratio_setting.GetModelPrice(modelName, true) - // 如果没有配置价格,则使用默认价格 - if !success { - defaultPrice, ok := ratio_setting.GetDefaultModelRatioMap()[modelName] - if !ok { - modelPrice = 0.1 - } else { - modelPrice = defaultPrice - } - } - groupRatio := ratio_setting.GetGroupRatio(group) - ratio := modelPrice * groupRatio + + priceData := helper.ModelPriceHelperPerCall(c, relayInfo) + userQuota, err := model.GetUserQuota(userId, false) if err != nil { return &dto.MidjourneyResponse{ @@ -499,9 +472,8 @@ func RelayMidjourneySubmit(c *gin.Context, relayMode int) *dto.MidjourneyRespons Description: err.Error(), } } - quota := int(ratio * common.QuotaPerUnit) - if consumeQuota && userQuota-quota < 0 { + if consumeQuota && userQuota-priceData.Quota < 0 { return &dto.MidjourneyResponse{ Code: 4, Description: "quota_not_enough", @@ -516,22 +488,17 @@ func RelayMidjourneySubmit(c *gin.Context, relayMode int) *dto.MidjourneyRespons defer func() { if consumeQuota && midjResponseWithStatus.StatusCode == 200 { - err := service.PostConsumeQuota(relayInfo, quota, 0, true) + err := service.PostConsumeQuota(relayInfo, priceData.Quota, 0, true) if err != nil { common.SysError("error consuming token remain quota: " + err.Error()) } - if quota != 0 { - tokenName := c.GetString("token_name") - logContent := fmt.Sprintf("模型固定价格 %.2f,分组倍率 %.2f,操作 %s,ID %s", modelPrice, groupRatio, midjRequest.Action, midjResponse.Result) - other := make(map[string]interface{}) - other["model_price"] = modelPrice - other["group_ratio"] = groupRatio - model.RecordConsumeLog(c, userId, channelId, 0, 0, modelName, tokenName, - quota, logContent, tokenId, userQuota, 0, false, group, other) - model.UpdateUserUsedQuotaAndRequestCount(userId, quota) - channelId := c.GetInt("channel_id") - model.UpdateChannelUsedQuota(channelId, quota) - } + 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(priceData) + model.RecordConsumeLog(c, userId, channelId, 0, 0, modelName, tokenName, + priceData.Quota, logContent, tokenId, userQuota, 0, false, group, other) + model.UpdateUserUsedQuotaAndRequestCount(userId, priceData.Quota) + model.UpdateChannelUsedQuota(channelId, priceData.Quota) } }() @@ -559,7 +526,7 @@ func RelayMidjourneySubmit(c *gin.Context, relayMode int) *dto.MidjourneyRespons Progress: "0%", FailReason: "", ChannelId: c.GetInt("channel_id"), - Quota: quota, + Quota: priceData.Quota, } if midjResponse.Code == 3 { //无实例账号自动禁用渠道(No available account instance) diff --git a/relay/relay-text.go b/relay/relay-text.go index db8d0d3b927a..e0c8f047b297 100644 --- a/relay/relay-text.go +++ b/relay/relay-text.go @@ -541,5 +541,5 @@ func postConsumeQuota(ctx *gin.Context, relayInfo *relaycommon.RelayInfo, other["audio_input_price"] = audioInputPrice } model.RecordConsumeLog(ctx, relayInfo.UserId, relayInfo.ChannelId, promptTokens, completionTokens, logModel, - tokenName, quota, logContent, relayInfo.TokenId, userQuota, int(useTimeSeconds), relayInfo.IsStream, relayInfo.Group, other) + tokenName, quota, logContent, relayInfo.TokenId, userQuota, int(useTimeSeconds), relayInfo.IsStream, relayInfo.UsingGroup, other) } diff --git a/relay/relay_adaptor.go b/relay/relay_adaptor.go index 626bb7e4dd8b..f648b4d572d9 100644 --- a/relay/relay_adaptor.go +++ b/relay/relay_adaptor.go @@ -22,6 +22,7 @@ import ( "one-api/relay/channel/palm" "one-api/relay/channel/perplexity" "one-api/relay/channel/siliconflow" + "one-api/relay/channel/task/jimeng" "one-api/relay/channel/task/kling" "one-api/relay/channel/task/suno" "one-api/relay/channel/tencent" @@ -104,6 +105,8 @@ func GetTaskAdaptor(platform commonconstant.TaskPlatform) channel.TaskAdaptor { return &suno.TaskAdaptor{} case commonconstant.TaskPlatformKling: return &kling.TaskAdaptor{} + case commonconstant.TaskPlatformJimeng: + return &jimeng.TaskAdaptor{} } return nil } diff --git a/relay/relay_task.go b/relay/relay_task.go index 245fd68139fb..702cff4c66c7 100644 --- a/relay/relay_task.go +++ b/relay/relay_task.go @@ -5,7 +5,6 @@ import ( "encoding/json" "errors" "fmt" - "github.com/gin-gonic/gin" "io" "net/http" "one-api/common" @@ -16,6 +15,8 @@ import ( relayconstant "one-api/relay/constant" "one-api/service" "one-api/setting/ratio_setting" + + "github.com/gin-gonic/gin" ) /* @@ -51,8 +52,14 @@ func RelayTaskSubmit(c *gin.Context, relayMode int) (taskErr *dto.TaskError) { } // 预扣 - groupRatio := ratio_setting.GetGroupRatio(relayInfo.Group) - ratio := modelPrice * groupRatio + groupRatio := ratio_setting.GetGroupRatio(relayInfo.UsingGroup) + var ratio float64 + userGroupRatio, hasUserGroupRatio := ratio_setting.GetGroupGroupRatio(relayInfo.UserGroup, relayInfo.UsingGroup) + if hasUserGroupRatio { + ratio = modelPrice * userGroupRatio + } else { + ratio = modelPrice * groupRatio + } userQuota, err := model.GetUserQuota(relayInfo.UserId, false) if err != nil { taskErr = service.TaskErrorWrapper(err, "get_user_quota_failed", http.StatusInternalServerError) @@ -121,12 +128,19 @@ func RelayTaskSubmit(c *gin.Context, relayMode int) (taskErr *dto.TaskError) { } if quota != 0 { tokenName := c.GetString("token_name") - logContent := fmt.Sprintf("模型固定价格 %.2f,分组倍率 %.2f,操作 %s", modelPrice, groupRatio, relayInfo.Action) + gRatio := groupRatio + if hasUserGroupRatio { + gRatio = userGroupRatio + } + logContent := fmt.Sprintf("模型固定价格 %.2f,分组倍率 %.2f,操作 %s", modelPrice, gRatio, relayInfo.Action) other := make(map[string]interface{}) other["model_price"] = modelPrice other["group_ratio"] = groupRatio + if hasUserGroupRatio { + other["user_group_ratio"] = userGroupRatio + } model.RecordConsumeLog(c, relayInfo.UserId, relayInfo.ChannelId, 0, 0, - modelName, tokenName, quota, logContent, relayInfo.TokenId, userQuota, 0, false, relayInfo.Group, other) + modelName, tokenName, quota, logContent, relayInfo.TokenId, userQuota, 0, false, relayInfo.UsingGroup, other) model.UpdateUserUsedQuotaAndRequestCount(relayInfo.UserId, quota) model.UpdateChannelUsedQuota(relayInfo.ChannelId, quota) } @@ -231,7 +245,7 @@ func sunoFetchByIDRespBodyBuilder(c *gin.Context) (respBody []byte, taskResp *dt } func videoFetchByIDRespBodyBuilder(c *gin.Context) (respBody []byte, taskResp *dto.TaskError) { - taskId := c.Param("id") + taskId := c.Param("task_id") userId := c.GetInt("id") originTask, exist, err := model.GetByTaskId(userId, taskId) diff --git a/router/api-router.go b/router/api-router.go index badfa7bf4e28..db4c3898524b 100644 --- a/router/api-router.go +++ b/router/api-router.go @@ -125,6 +125,7 @@ func SetApiRouter(router *gin.Engine) { tokenRoute.POST("/", controller.AddToken) tokenRoute.PUT("/", controller.UpdateToken) tokenRoute.DELETE("/:id", controller.DeleteToken) + tokenRoute.POST("/batch", controller.DeleteTokenBatch) } redemptionRoute := apiRouter.Group("/redemption") redemptionRoute.Use(middleware.AdminAuth()) diff --git a/router/relay-router.go b/router/relay-router.go index aa7f27a88b3c..325ef1355dfb 100644 --- a/router/relay-router.go +++ b/router/relay-router.go @@ -63,6 +63,7 @@ func SetRelayRouter(router *gin.Engine) { httpRouter.DELETE("/models/:model", controller.RelayNotImplemented) httpRouter.POST("/moderations", controller.Relay) httpRouter.POST("/rerank", controller.Relay) + httpRouter.POST("/models/*path", controller.Relay) } relayMjRouter := router.Group("/mj") diff --git a/router/video-router.go b/router/video-router.go index 7201c34ab804..9e605d5411d6 100644 --- a/router/video-router.go +++ b/router/video-router.go @@ -14,4 +14,11 @@ func SetVideoRouter(router *gin.Engine) { videoV1Router.POST("/video/generations", controller.RelayTask) videoV1Router.GET("/video/generations/:task_id", controller.RelayTask) } + + klingV1Router := router.Group("/kling/v1") + klingV1Router.Use(middleware.KlingRequestConvert(), middleware.TokenAuth(), middleware.Distribute()) + { + klingV1Router.POST("/videos/text2video", controller.RelayTask) + klingV1Router.POST("/videos/image2video", controller.RelayTask) + } } diff --git a/service/convert.go b/service/convert.go index 7a9e8403456e..df7acf0d5967 100644 --- a/service/convert.go +++ b/service/convert.go @@ -276,12 +276,15 @@ func StreamResponseOpenAI2Claude(openAIResponse *dto.ChatCompletionsStreamRespon } if info.Done { claudeResponses = append(claudeResponses, generateStopBlock(info.ClaudeConvertInfo.Index)) - if info.ClaudeConvertInfo.Usage != nil { + oaiUsage := info.ClaudeConvertInfo.Usage + if oaiUsage != nil { claudeResponses = append(claudeResponses, &dto.ClaudeResponse{ Type: "message_delta", Usage: &dto.ClaudeUsage{ - InputTokens: info.ClaudeConvertInfo.Usage.PromptTokens, - OutputTokens: info.ClaudeConvertInfo.Usage.CompletionTokens, + InputTokens: oaiUsage.PromptTokens, + OutputTokens: oaiUsage.CompletionTokens, + CacheCreationInputTokens: oaiUsage.PromptTokensDetails.CachedCreationTokens, + CacheReadInputTokens: oaiUsage.PromptTokensDetails.CachedTokens, }, Delta: &dto.ClaudeMediaMessage{ StopReason: common.GetPointer[string](stopReasonOpenAI2Claude(info.FinishReason)), diff --git a/service/error.go b/service/error.go index f3d8a17decbf..21835f2a4fad 100644 --- a/service/error.go +++ b/service/error.go @@ -90,10 +90,7 @@ func RelayErrorHandler(resp *http.Response, showBodyWhenFail bool) (errWithStatu if err != nil { return } - err = resp.Body.Close() - if err != nil { - return - } + common.CloseResponseBodyGracefully(resp) var errResponse dto.GeneralErrorResponse err = json.Unmarshal(responseBody, &errResponse) if err != nil { diff --git a/service/log_info_generate.go b/service/log_info_generate.go index 1edc90736980..affae5fba85c 100644 --- a/service/log_info_generate.go +++ b/service/log_info_generate.go @@ -3,6 +3,7 @@ package service import ( "one-api/dto" relaycommon "one-api/relay/common" + "one-api/relay/helper" "github.com/gin-gonic/gin" ) @@ -63,3 +64,13 @@ func GenerateClaudeOtherInfo(ctx *gin.Context, relayInfo *relaycommon.RelayInfo, info["cache_creation_ratio"] = cacheCreationRatio return info } + +func GenerateMjOtherInfo(priceData helper.PerCallPriceData) map[string]interface{} { + other := make(map[string]interface{}) + other["model_price"] = priceData.ModelPrice + other["group_ratio"] = priceData.GroupRatioInfo.GroupRatio + if priceData.GroupRatioInfo.HasSpecialRatio { + other["user_group_ratio"] = priceData.GroupRatioInfo.GroupSpecialRatio + } + return other +} diff --git a/service/midjourney.go b/service/midjourney.go index 635c29ae7fbe..b4ef15a9b120 100644 --- a/service/midjourney.go +++ b/service/midjourney.go @@ -228,10 +228,7 @@ func DoMidjourneyHttpRequest(c *gin.Context, timeout time.Duration, fullRequestU if err != nil { return MidjourneyErrorWithStatusCodeWrapper(constant.MjErrorUnknown, "read_response_body_failed", statusCode), nullBytes, err } - err = resp.Body.Close() - if err != nil { - return MidjourneyErrorWithStatusCodeWrapper(constant.MjErrorUnknown, "close_response_body_failed", statusCode), responseBody, err - } + common.CloseResponseBodyGracefully(resp) respStr := string(responseBody) log.Printf("respStr: %s", respStr) if respStr == "" { diff --git a/service/quota.go b/service/quota.go index 973deba7c1fd..c17616a74f61 100644 --- a/service/quota.go +++ b/service/quota.go @@ -4,6 +4,7 @@ import ( "errors" "fmt" "log" + "math" "one-api/common" constant2 "one-api/constant" "one-api/dto" @@ -94,18 +95,18 @@ func PreWssConsumeQuota(ctx *gin.Context, relayInfo *relaycommon.RelayInfo, usag textOutTokens := usage.OutputTokenDetails.TextTokens audioInputTokens := usage.InputTokenDetails.AudioTokens audioOutTokens := usage.OutputTokenDetails.AudioTokens - groupRatio := ratio_setting.GetGroupRatio(relayInfo.Group) + groupRatio := ratio_setting.GetGroupRatio(relayInfo.UsingGroup) modelRatio, _ := ratio_setting.GetModelRatio(modelName) autoGroup, exists := ctx.Get("auto_group") if exists { groupRatio = ratio_setting.GetGroupRatio(autoGroup.(string)) log.Printf("final group ratio: %f", groupRatio) - relayInfo.Group = autoGroup.(string) + relayInfo.UsingGroup = autoGroup.(string) } actualGroupRatio := groupRatio - userGroupRatio, ok := ratio_setting.GetGroupGroupRatio(relayInfo.UserGroup, relayInfo.Group) + userGroupRatio, ok := ratio_setting.GetGroupGroupRatio(relayInfo.UserGroup, relayInfo.UsingGroup) if ok { actualGroupRatio = userGroupRatio } @@ -209,7 +210,7 @@ func PostWssConsumeQuota(ctx *gin.Context, relayInfo *relaycommon.RelayInfo, mod other := GenerateWssOtherInfo(ctx, relayInfo, usage, modelRatio, groupRatio, completionRatio.InexactFloat64(), audioRatio.InexactFloat64(), audioCompletionRatio.InexactFloat64(), modelPrice, priceData.GroupRatioInfo.GroupSpecialRatio) model.RecordConsumeLog(ctx, relayInfo.UserId, relayInfo.ChannelId, usage.InputTokens, usage.OutputTokens, logModel, - tokenName, quota, logContent, relayInfo.TokenId, userQuota, int(useTimeSeconds), relayInfo.IsStream, relayInfo.Group, other) + tokenName, quota, logContent, relayInfo.TokenId, userQuota, int(useTimeSeconds), relayInfo.IsStream, relayInfo.UsingGroup, other) } func PostClaudeConsumeQuota(ctx *gin.Context, relayInfo *relaycommon.RelayInfo, @@ -231,6 +232,17 @@ func PostClaudeConsumeQuota(ctx *gin.Context, relayInfo *relaycommon.RelayInfo, cacheCreationRatio := priceData.CacheCreationRatio cacheCreationTokens := usage.PromptTokensDetails.CachedCreationTokens + if relayInfo.ChannelType == common.ChannelTypeOpenRouter { + promptTokens -= cacheTokens + if cacheCreationTokens == 0 && priceData.CacheCreationRatio != 1 && usage.Cost != 0 { + maybeCacheCreationTokens := CalcOpenRouterCacheCreateTokens(*usage, priceData) + if promptTokens >= maybeCacheCreationTokens { + cacheCreationTokens = maybeCacheCreationTokens + } + } + promptTokens -= cacheCreationTokens + } + calculateQuota := 0.0 if !priceData.UsePrice { calculateQuota = float64(promptTokens) @@ -275,7 +287,28 @@ func PostClaudeConsumeQuota(ctx *gin.Context, relayInfo *relaycommon.RelayInfo, other := GenerateClaudeOtherInfo(ctx, relayInfo, modelRatio, groupRatio, completionRatio, cacheTokens, cacheRatio, cacheCreationTokens, cacheCreationRatio, modelPrice, priceData.GroupRatioInfo.GroupSpecialRatio) model.RecordConsumeLog(ctx, relayInfo.UserId, relayInfo.ChannelId, promptTokens, completionTokens, modelName, - tokenName, quota, logContent, relayInfo.TokenId, userQuota, int(useTimeSeconds), relayInfo.IsStream, relayInfo.Group, other) + tokenName, quota, logContent, relayInfo.TokenId, userQuota, int(useTimeSeconds), relayInfo.IsStream, relayInfo.UsingGroup, other) +} + +func CalcOpenRouterCacheCreateTokens(usage dto.Usage, priceData helper.PriceData) int { + if priceData.CacheCreationRatio == 1 { + return 0 + } + quotaPrice := priceData.ModelRatio / common.QuotaPerUnit + promptCacheCreatePrice := quotaPrice * priceData.CacheCreationRatio + promptCacheReadPrice := quotaPrice * priceData.CacheRatio + completionPrice := quotaPrice * priceData.CompletionRatio + + cost := usage.Cost + totalPromptTokens := float64(usage.PromptTokens) + completionTokens := float64(usage.CompletionTokens) + promptCacheReadTokens := float64(usage.PromptTokensDetails.CachedTokens) + + return int(math.Round((cost - + totalPromptTokens*quotaPrice + + promptCacheReadTokens*(quotaPrice-promptCacheReadPrice) - + completionTokens*completionPrice) / + (promptCacheCreatePrice - quotaPrice))) } func PostAudioConsumeQuota(ctx *gin.Context, relayInfo *relaycommon.RelayInfo, @@ -352,7 +385,7 @@ func PostAudioConsumeQuota(ctx *gin.Context, relayInfo *relaycommon.RelayInfo, other := GenerateAudioOtherInfo(ctx, relayInfo, usage, modelRatio, groupRatio, completionRatio.InexactFloat64(), audioRatio.InexactFloat64(), audioCompletionRatio.InexactFloat64(), modelPrice, priceData.GroupRatioInfo.GroupSpecialRatio) model.RecordConsumeLog(ctx, relayInfo.UserId, relayInfo.ChannelId, usage.PromptTokens, usage.CompletionTokens, logModel, - tokenName, quota, logContent, relayInfo.TokenId, userQuota, int(useTimeSeconds), relayInfo.IsStream, relayInfo.Group, other) + tokenName, quota, logContent, relayInfo.TokenId, userQuota, int(useTimeSeconds), relayInfo.IsStream, relayInfo.UsingGroup, other) } func PreConsumeTokenQuota(relayInfo *relaycommon.RelayInfo, quota int) error { diff --git a/setting/ratio_setting/group_ratio.go b/setting/ratio_setting/group_ratio.go index f600a7b59642..86f4a8d19c9e 100644 --- a/setting/ratio_setting/group_ratio.go +++ b/setting/ratio_setting/group_ratio.go @@ -73,15 +73,15 @@ func GetGroupRatio(name string) float64 { return ratio } -func GetGroupGroupRatio(group, name string) (float64, bool) { +func GetGroupGroupRatio(userGroup, usingGroup string) (float64, bool) { groupGroupRatioMutex.RLock() defer groupGroupRatioMutex.RUnlock() - gp, ok := GroupGroupRatio[group] + gp, ok := GroupGroupRatio[userGroup] if !ok { return -1, false } - ratio, ok := gp[name] + ratio, ok := gp[usingGroup] if !ok { return -1, false } diff --git a/setting/ratio_setting/model_ratio.go b/setting/ratio_setting/model_ratio.go index 879423eb04a8..8da144448176 100644 --- a/setting/ratio_setting/model_ratio.go +++ b/setting/ratio_setting/model_ratio.go @@ -501,16 +501,19 @@ func getHardcodedCompletionModelRatio(name string) (float64, bool) { } else if strings.HasPrefix(name, "gemini-2.0") { return 4, true } else if strings.HasPrefix(name, "gemini-2.5-pro") { // 移除preview来增加兼容性,这里假设正式版的倍率和preview一致 - return 8, true + return 8, false } else if strings.HasPrefix(name, "gemini-2.5-flash") { // 处理不同的flash模型倍率 if strings.HasPrefix(name, "gemini-2.5-flash-preview") { if strings.HasSuffix(name, "-nothinking") { - return 4, true + return 4, false } - return 3.5 / 0.15, true + return 3.5 / 0.15, false } - if strings.HasPrefix(name, "gemini-2.5-flash-lite-preview") { - return 4, true + if strings.HasPrefix(name, "gemini-2.5-flash-lite") { + if strings.HasPrefix(name, "gemini-2.5-flash-lite-preview") { + return 4, false + } + return 4, false } return 2.5 / 0.3, true } diff --git a/web/src/components/auth/LoginForm.js b/web/src/components/auth/LoginForm.js index c8847a33d720..85d22dfcb3cf 100644 --- a/web/src/components/auth/LoginForm.js +++ b/web/src/components/auth/LoginForm.js @@ -34,20 +34,20 @@ import LinuxDoIcon from '../common/logo/LinuxDoIcon.js'; import { useTranslation } from 'react-i18next'; const LoginForm = () => { + let navigate = useNavigate(); + const { t } = useTranslation(); const [inputs, setInputs] = useState({ username: '', password: '', wechat_verification_code: '', }); + const { username, password } = inputs; const [searchParams, setSearchParams] = useSearchParams(); const [submitted, setSubmitted] = useState(false); - const { username, password } = inputs; const [userState, userDispatch] = useContext(UserContext); const [turnstileEnabled, setTurnstileEnabled] = useState(false); const [turnstileSiteKey, setTurnstileSiteKey] = useState(''); const [turnstileToken, setTurnstileToken] = useState(''); - let navigate = useNavigate(); - const [status, setStatus] = useState({}); const [showWeChatLoginModal, setShowWeChatLoginModal] = useState(false); const [showEmailLogin, setShowEmailLogin] = useState(false); const [wechatLoading, setWechatLoading] = useState(false); @@ -59,7 +59,6 @@ const LoginForm = () => { const [resetPasswordLoading, setResetPasswordLoading] = useState(false); const [otherLoginOptionsLoading, setOtherLoginOptionsLoading] = useState(false); const [wechatCodeSubmitLoading, setWechatCodeSubmitLoading] = useState(false); - const { t } = useTranslation(); const logo = getLogo(); const systemName = getSystemName(); @@ -69,19 +68,22 @@ const LoginForm = () => { localStorage.setItem('aff', affCode); } + const [status] = useState(() => { + const savedStatus = localStorage.getItem('status'); + return savedStatus ? JSON.parse(savedStatus) : {}; + }); + + useEffect(() => { + if (status.turnstile_check) { + setTurnstileEnabled(true); + setTurnstileSiteKey(status.turnstile_site_key); + } + }, [status]); + useEffect(() => { if (searchParams.get('expired')) { showError(t('未登录或登录已过期,请重新登录')); } - let status = localStorage.getItem('status'); - if (status) { - status = JSON.parse(status); - setStatus(status); - if (status.turnstile_check) { - setTurnstileEnabled(true); - setTurnstileSiteKey(status.turnstile_site_key); - } - } }, []); const onWeChatLoginClicked = () => { @@ -356,9 +358,19 @@ const LoginForm = () => { -