Merge pull request #1659 from Sh1n3zZ/feat-vertex-veo

feat: vertex veo (#1450)

common/database.go

@@ -12,4 +12,4 @@ var LogSqlType = DatabaseTypeSQLite // Default to SQLite for logging SQL queries
 var UsingMySQL = false
 var UsingClickHouse = false
 
-var SQLitePath = "one-api.db?_busy_timeout=30000"
+var SQLitePath = "one-api.db?_busy_timeout=30000"
\ No newline at end of file

controller/setup.go

@@ -178,4 +178,4 @@ func boolToString(b bool) string {
 		return "true"
 	}
 	return "false"
-}
+}
\ No newline at end of file

controller/task_video.go

@@ -94,7 +94,7 @@ func updateVideoSingleTask(ctx context.Context, adaptor channel.TaskAdaptor, cha
 	} else if taskResult, err = adaptor.ParseTaskResult(responseBody); err != nil {
 		return fmt.Errorf("parseTaskResult failed for task %s: %w", taskId, err)
 	} else {
-		task.Data = responseBody
+		task.Data = redactVideoResponseBody(responseBody)
 	}
 
 	now := time.Now().Unix()
@@ -117,7 +117,9 @@ func updateVideoSingleTask(ctx context.Context, adaptor channel.TaskAdaptor, cha
 		if task.FinishTime == 0 {
 			task.FinishTime = now
 		}
-		task.FailReason = taskResult.Url
+		if !(len(taskResult.Url) > 5 && taskResult.Url[:5] == "data:") {
+			task.FailReason = taskResult.Url
+		}
 	case model.TaskStatusFailure:
 		task.Status = model.TaskStatusFailure
 		task.Progress = "100%"
@@ -146,3 +148,37 @@ func updateVideoSingleTask(ctx context.Context, adaptor channel.TaskAdaptor, cha
 
 	return nil
 }
+
+func redactVideoResponseBody(body []byte) []byte {
+	var m map[string]any
+	if err := json.Unmarshal(body, &m); err != nil {
+		return body
+	}
+	resp, _ := m["response"].(map[string]any)
+	if resp != nil {
+		delete(resp, "bytesBase64Encoded")
+		if v, ok := resp["video"].(string); ok {
+			resp["video"] = truncateBase64(v)
+		}
+		if vs, ok := resp["videos"].([]any); ok {
+			for i := range vs {
+				if vm, ok := vs[i].(map[string]any); ok {
+					delete(vm, "bytesBase64Encoded")
+				}
+			}
+		}
+	}
+	b, err := json.Marshal(m)
+	if err != nil {
+		return body
+	}
+	return b
+}
+
+func truncateBase64(s string) string {
+	const maxKeep = 256
+	if len(s) <= maxKeep {
+		return s
+	}
+	return s[:maxKeep] + "..."
+}

main.go

@@ -204,4 +204,4 @@ func InitResources() error {
 		return err
 	}
 	return nil
-}
+}
\ No newline at end of file

middleware/distributor.go

@@ -166,9 +166,9 @@ 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") {
-		err = common.UnmarshalBodyReusable(c, &modelRequest)
 		relayMode := relayconstant.RelayModeUnknown
 		if c.Request.Method == http.MethodPost {
+			err = common.UnmarshalBodyReusable(c, &modelRequest)
 			relayMode = relayconstant.RelayModeVideoSubmit
 		} else if c.Request.Method == http.MethodGet {
 			relayMode = relayconstant.RelayModeVideoFetchByID

relay/channel/vertex/adaptor.go

@@ -174,6 +174,7 @@ func (a *Adaptor) SetupRequestHeader(c *gin.Context, req *http.Header, info *rel
 		return err
 	}
 	req.Set("Authorization", "Bearer "+accessToken)
+	req.Set("x-goog-user-project", a.AccountCredentials.ProjectID)
 	return nil
 }
 

relay/channel/vertex/relay-vertex.go

@@ -12,7 +12,10 @@ func GetModelRegion(other string, localModelName string) string {
 		if m[localModelName] != nil {
 			return m[localModelName].(string)
 		} else {
-			return m["default"].(string)
+			if v, ok := m["default"]; ok {
+				return v.(string)
+			}
+			return "global"
 		}
 	}
 	return other

relay/channel/vertex/service_account.go

@@ -6,14 +6,15 @@ import (
 	"encoding/json"
 	"encoding/pem"
 	"errors"
-	"github.com/bytedance/gopkg/cache/asynccache"
-	"github.com/golang-jwt/jwt"
 	"net/http"
 	"net/url"
 	relaycommon "one-api/relay/common"
 	"one-api/service"
 	"strings"
 
+	"github.com/bytedance/gopkg/cache/asynccache"
+	"github.com/golang-jwt/jwt"
+
 	"fmt"
 	"time"
 )
@@ -137,3 +138,45 @@ func exchangeJwtForAccessToken(signedJWT string, info *relaycommon.RelayInfo) (s
 
 	return "", fmt.Errorf("failed to get access token: %v", result)
 }
+
+func AcquireAccessToken(creds Credentials, proxy string) (string, error) {
+	signedJWT, err := createSignedJWT(creds.ClientEmail, creds.PrivateKey)
+	if err != nil {
+		return "", fmt.Errorf("failed to create signed JWT: %w", err)
+	}
+	return exchangeJwtForAccessTokenWithProxy(signedJWT, proxy)
+}
+
+func exchangeJwtForAccessTokenWithProxy(signedJWT string, proxy string) (string, error) {
+	authURL := "https://www.googleapis.com/oauth2/v4/token"
+	data := url.Values{}
+	data.Set("grant_type", "urn:ietf:params:oauth:grant-type:jwt-bearer")
+	data.Set("assertion", signedJWT)
+
+	var client *http.Client
+	var err error
+	if proxy != "" {
+		client, err = service.NewProxyHttpClient(proxy)
+		if err != nil {
+			return "", fmt.Errorf("new proxy http client failed: %w", err)
+		}
+	} else {
+		client = service.GetHttpClient()
+	}
+
+	resp, err := client.PostForm(authURL, data)
+	if err != nil {
+		return "", err
+	}
+	defer resp.Body.Close()
+
+	var result map[string]interface{}
+	if err := json.NewDecoder(resp.Body).Decode(&result); err != nil {
+		return "", err
+	}
+
+	if accessToken, ok := result["access_token"].(string); ok {
+		return accessToken, nil
+	}
+	return "", fmt.Errorf("failed to get access token: %v", result)
+}

relay/relay_adaptor.go

 package relay
 
 import (
-	"github.com/gin-gonic/gin"
 	"one-api/constant"
 	"one-api/relay/channel"
 	"one-api/relay/channel/ali"
@@ -28,6 +27,7 @@ import (
 	taskjimeng "one-api/relay/channel/task/jimeng"
 	"one-api/relay/channel/task/kling"
 	"one-api/relay/channel/task/suno"
+	taskvertex "one-api/relay/channel/task/vertex"
 	taskVidu "one-api/relay/channel/task/vidu"
 	"one-api/relay/channel/tencent"
 	"one-api/relay/channel/vertex"
@@ -37,6 +37,8 @@ import (
 	"one-api/relay/channel/zhipu"
 	"one-api/relay/channel/zhipu_4v"
 	"strconv"
+
+	"github.com/gin-gonic/gin"
 )
 
 func GetAdaptor(apiType int) channel.Adaptor {
@@ -126,6 +128,8 @@ func GetTaskAdaptor(platform constant.TaskPlatform) channel.TaskAdaptor {
 			return &kling.TaskAdaptor{}
 		case constant.ChannelTypeJimeng:
 			return &taskjimeng.TaskAdaptor{}
+		case constant.ChannelTypeVertexAi:
+			return &taskvertex.TaskAdaptor{}
 		case constant.ChannelTypeVidu:
 			return &taskVidu.TaskAdaptor{}
 		}

relay/relay_task.go

@@ -15,6 +15,8 @@ import (
 	relayconstant "one-api/relay/constant"
 	"one-api/service"
 	"one-api/setting/ratio_setting"
+	"strconv"
+	"strings"
 
 	"github.com/gin-gonic/gin"
 )
@@ -33,6 +35,7 @@ func RelayTaskSubmit(c *gin.Context, info *relaycommon.RelayInfo) (taskErr *dto.
 		platform = GetTaskPlatform(c)
 	}
 
+	info.InitChannelMeta(c)
 	adaptor := GetTaskAdaptor(platform)
 	if adaptor == nil {
 		return service.TaskErrorWrapperLocal(fmt.Errorf("invalid api platform: %s", platform), "invalid_api_platform", http.StatusBadRequest)
@@ -197,6 +200,9 @@ func RelayTaskFetch(c *gin.Context, relayMode int) (taskResp *dto.TaskError) {
 	if taskErr != nil {
 		return taskErr
 	}
+	if len(respBody) == 0 {
+		respBody = []byte("{\"code\":\"success\",\"data\":null}")
+	}
 
 	c.Writer.Header().Set("Content-Type", "application/json")
 	_, err := io.Copy(c.Writer, bytes.NewBuffer(respBody))
@@ -276,10 +282,92 @@ func videoFetchByIDRespBodyBuilder(c *gin.Context) (respBody []byte, taskResp *d
 		return
 	}
 
-	respBody, err = json.Marshal(dto.TaskResponse[any]{
-		Code: "success",
-		Data: TaskModel2Dto(originTask),
-	})
+	func() {
+		channelModel, err2 := model.GetChannelById(originTask.ChannelId, true)
+		if err2 != nil {
+			return
+		}
+		if channelModel.Type != constant.ChannelTypeVertexAi {
+			return
+		}
+		baseURL := constant.ChannelBaseURLs[channelModel.Type]
+		if channelModel.GetBaseURL() != "" {
+			baseURL = channelModel.GetBaseURL()
+		}
+		adaptor := GetTaskAdaptor(constant.TaskPlatform(strconv.Itoa(channelModel.Type)))
+		if adaptor == nil {
+			return
+		}
+		resp, err2 := adaptor.FetchTask(baseURL, channelModel.Key, map[string]any{
+			"task_id": originTask.TaskID,
+			"action":  originTask.Action,
+		})
+		if err2 != nil || resp == nil {
+			return
+		}
+		defer resp.Body.Close()
+		body, err2 := io.ReadAll(resp.Body)
+		if err2 != nil {
+			return
+		}
+		ti, err2 := adaptor.ParseTaskResult(body)
+		if err2 == nil && ti != nil {
+			if ti.Status != "" {
+				originTask.Status = model.TaskStatus(ti.Status)
+			}
+			if ti.Progress != "" {
+				originTask.Progress = ti.Progress
+			}
+			if ti.Url != "" {
+				originTask.FailReason = ti.Url
+			}
+			_ = originTask.Update()
+			var raw map[string]any
+			_ = json.Unmarshal(body, &raw)
+			format := "mp4"
+			if respObj, ok := raw["response"].(map[string]any); ok {
+				if vids, ok := respObj["videos"].([]any); ok && len(vids) > 0 {
+					if v0, ok := vids[0].(map[string]any); ok {
+						if mt, ok := v0["mimeType"].(string); ok && mt != "" {
+							if strings.Contains(mt, "mp4") {
+								format = "mp4"
+							} else {
+								format = mt
+							}
+						}
+					}
+				}
+			}
+			status := "processing"
+			switch originTask.Status {
+			case model.TaskStatusSuccess:
+				status = "succeeded"
+			case model.TaskStatusFailure:
+				status = "failed"
+			case model.TaskStatusQueued, model.TaskStatusSubmitted:
+				status = "queued"
+			}
+			out := map[string]any{
+				"error":    nil,
+				"format":   format,
+				"metadata": nil,
+				"status":   status,
+				"task_id":  originTask.TaskID,
+				"url":      originTask.FailReason,
+			}
+			respBody, _ = json.Marshal(dto.TaskResponse[any]{
+				Code: "success",
+				Data: out,
+			})
+		}
+	}()
+
+	if len(respBody) == 0 {
+		respBody, err = json.Marshal(dto.TaskResponse[any]{
+			Code: "success",
+			Data: TaskModel2Dto(originTask),
+		})
+	}
 	return
 }