diff --git a/.env.example b/.env.example index 6965c88..f551733 100644 --- a/.env.example +++ b/.env.example @@ -49,6 +49,13 @@ AI_GATEWAY_API_RUNTIME_IMAGE=alpine:3.22 AI_GATEWAY_NODE_BUILD_IMAGE=node:22-alpine AI_GATEWAY_WEB_RUNTIME_IMAGE=nginx:1.27-alpine +# Opt-in, billable China Kling V1 integration tests. Keep real AK/SK only in +# .env.local (gitignored); never commit them. +KELING_LIVE_TEST=0 +KELING_TEST_BASE_URL=https://api-beijing.klingai.com/v1 +KELING_TEST_ACCESS_KEY= +KELING_TEST_SECRET_KEY= + # Used when the gateway delegates OpenAPI sk-* validation, user/group sync, file upload, and settlement callbacks. SERVER_MAIN_BASE_URL=http://localhost:3000 SERVER_MAIN_INTERNAL_TOKEN=change-me diff --git a/README.md b/README.md index 0ba901e..ef73671 100644 --- a/README.md +++ b/README.md @@ -125,6 +125,8 @@ AI_GATEWAY_COMPOSE_DATABASE_URL='postgresql://easyai:easyai2025@postgres:5432/ea pnpm openapi ``` +中国区可灵 O1 / 3.0 Omni 的 V1 AK/SK 与 API 2.0 兼容接入方式见 [可灵兼容接口说明](docs/kling-compatible-api.md)。 + 默认 EasyAI 部署里，`easyai-pgvector` 在容器网络内的连接串是： diff --git a/apps/api/docs/swagger.json b/apps/api/docs/swagger.json index bc5222c..4540ec7 100644 --- a/apps/api/docs/swagger.json +++ b/apps/api/docs/swagger.json @@ -8359,6 +8359,278 @@ } } }, + "/kling/omni-video/{model}": { + "post": { + "security": [ + { + "BearerAuth": [] + } + ], + "description": "兼容可灵 API 2.0 的模型路径；调用方使用网关 API Key，网关转换并使用中国区 V1 AK/SK 上游。", + "consumes": [ + "application/json" + ], + "produces": [ + "application/json" + ], + "tags": [ + "kling-compatible" + ], + "summary": "创建可灵 API 2.0 Omni 视频任务", + "parameters": [ + { + "type": "string", + "description": "模型路径（kling-o1 或 kling-v3-omni）", + "name": "model", + "in": "path", + "required": true + }, + { + "description": "可灵 API 2.0 请求", + "name": "input", + "in": "body", + "required": true, + "schema": { + "type": "object", + "additionalProperties": true + } + } + ], + "responses": { + "200": { + "description": "OK", + "schema": { + "type": "object", + "additionalProperties": true + } + }, + "400": { + "description": "Bad Request", + "schema": { + "type": "object", + "additionalProperties": true + } + }, + "401": { + "description": "Unauthorized", + "schema": { + "$ref": "#/definitions/httpapi.ErrorEnvelope" + } + } + } + } + }, + "/kling/tasks": { + "get": { + "security": [ + { + "BearerAuth": [] + } + ], + "produces": [ + "application/json" + ], + "tags": [ + "kling-compatible" + ], + "summary": "按 ID 查询可灵 API 2.0 任务", + "parameters": [ + { + "type": "string", + "description": "逗号分隔的任务 ID", + "name": "task_ids", + "in": "query" + }, + { + "type": "string", + "description": "逗号分隔的外部任务 ID", + "name": "external_task_ids", + "in": "query" + } + ], + "responses": { + "200": { + "description": "OK", + "schema": { + "type": "object", + "additionalProperties": true + } + } + } + }, + "post": { + "security": [ + { + "BearerAuth": [] + } + ], + "consumes": [ + "application/json" + ], + "produces": [ + "application/json" + ], + "tags": [ + "kling-compatible" + ], + "summary": "分页查询可灵 API 2.0 任务", + "parameters": [ + { + "description": "游标、数量、时间范围和筛选条件", + "name": "input", + "in": "body", + "required": true, + "schema": { + "type": "object", + "additionalProperties": true + } + } + ], + "responses": { + "200": { + "description": "OK", + "schema": { + "type": "object", + "additionalProperties": true + } + } + } + } + }, + "/kling/v1/videos/omni-video": { + "get": { + "security": [ + { + "BearerAuth": [] + } + ], + "produces": [ + "application/json" + ], + "tags": [ + "kling-compatible" + ], + "summary": "分页查询可灵 V1 Omni 视频任务", + "parameters": [ + { + "type": "integer", + "default": 1, + "description": "页码", + "name": "pageNum", + "in": "query" + }, + { + "type": "integer", + "default": 30, + "description": "每页数量", + "name": "pageSize", + "in": "query" + } + ], + "responses": { + "200": { + "description": "OK", + "schema": { + "type": "object", + "additionalProperties": true + } + } + } + }, + "post": { + "security": [ + { + "BearerAuth": [] + } + ], + "description": "兼容中国区可灵 V1 /v1/videos/omni-video；用户使用网关 API Key，网关在服务端使用 AK/SK 调用上游。", + "consumes": [ + "application/json" + ], + "produces": [ + "application/json" + ], + "tags": [ + "kling-compatible" + ], + "summary": "创建可灵 V1 Omni 视频任务", + "parameters": [ + { + "description": "可灵 V1 Omni 请求", + "name": "input", + "in": "body", + "required": true, + "schema": { + "type": "object", + "additionalProperties": true + } + } + ], + "responses": { + "200": { + "description": "OK", + "schema": { + "type": "object", + "additionalProperties": true + } + }, + "400": { + "description": "Bad Request", + "schema": { + "type": "object", + "additionalProperties": true + } + }, + "401": { + "description": "Unauthorized", + "schema": { + "$ref": "#/definitions/httpapi.ErrorEnvelope" + } + } + } + } + }, + "/kling/v1/videos/omni-video/{taskID}": { + "get": { + "security": [ + { + "BearerAuth": [] + } + ], + "produces": [ + "application/json" + ], + "tags": [ + "kling-compatible" + ], + "summary": "查询可灵 V1 Omni 视频任务", + "parameters": [ + { + "type": "string", + "description": "任务 ID", + "name": "taskID", + "in": "path", + "required": true + } + ], + "responses": { + "200": { + "description": "OK", + "schema": { + "type": "object", + "additionalProperties": true + } + }, + "404": { + "description": "Not Found", + "schema": { + "type": "object", + "additionalProperties": true + } + } + } + } + }, "/music/generations": { "post": { "security": [ @@ -13538,6 +13810,9 @@ "executionLeaseExpiresAt": { "type": "string" }, + "externalTaskId": { + "type": "string" + }, "finalChargeAmount": { "type": "number" }, diff --git a/apps/api/docs/swagger.yaml b/apps/api/docs/swagger.yaml index b7271b8..5580469 100644 --- a/apps/api/docs/swagger.yaml +++ b/apps/api/docs/swagger.yaml @@ -1984,6 +1984,8 @@ definitions: type: string executionLeaseExpiresAt: type: string + externalTaskId: + type: string finalChargeAmount: type: number finishedAt: @@ -8491,6 +8493,181 @@ paths: summary: 创建或执行 AI 任务 tags: - tasks + /kling/omni-video/{model}: + post: + consumes: + - application/json + description: 兼容可灵 API 2.0 的模型路径；调用方使用网关 API Key，网关转换并使用中国区 V1 AK/SK 上游。 + parameters: + - description: 模型路径（kling-o1 或 kling-v3-omni） + in: path + name: model + required: true + type: string + - description: 可灵 API 2.0 请求 + in: body + name: input + required: true + schema: + additionalProperties: true + type: object + produces: + - application/json + responses: + "200": + description: OK + schema: + additionalProperties: true + type: object + "400": + description: Bad Request + schema: + additionalProperties: true + type: object + "401": + description: Unauthorized + schema: + $ref: '#/definitions/httpapi.ErrorEnvelope' + security: + - BearerAuth: [] + summary: 创建可灵 API 2.0 Omni 视频任务 + tags: + - kling-compatible + /kling/tasks: + get: + parameters: + - description: 逗号分隔的任务 ID + in: query + name: task_ids + type: string + - description: 逗号分隔的外部任务 ID + in: query + name: external_task_ids + type: string + produces: + - application/json + responses: + "200": + description: OK + schema: + additionalProperties: true + type: object + security: + - BearerAuth: [] + summary: 按 ID 查询可灵 API 2.0 任务 + tags: + - kling-compatible + post: + consumes: + - application/json + parameters: + - description: 游标、数量、时间范围和筛选条件 + in: body + name: input + required: true + schema: + additionalProperties: true + type: object + produces: + - application/json + responses: + "200": + description: OK + schema: + additionalProperties: true + type: object + security: + - BearerAuth: [] + summary: 分页查询可灵 API 2.0 任务 + tags: + - kling-compatible + /kling/v1/videos/omni-video: + get: + parameters: + - default: 1 + description: 页码 + in: query + name: pageNum + type: integer + - default: 30 + description: 每页数量 + in: query + name: pageSize + type: integer + produces: + - application/json + responses: + "200": + description: OK + schema: + additionalProperties: true + type: object + security: + - BearerAuth: [] + summary: 分页查询可灵 V1 Omni 视频任务 + tags: + - kling-compatible + post: + consumes: + - application/json + description: 兼容中国区可灵 V1 /v1/videos/omni-video；用户使用网关 API Key，网关在服务端使用 AK/SK + 调用上游。 + parameters: + - description: 可灵 V1 Omni 请求 + in: body + name: input + required: true + schema: + additionalProperties: true + type: object + produces: + - application/json + responses: + "200": + description: OK + schema: + additionalProperties: true + type: object + "400": + description: Bad Request + schema: + additionalProperties: true + type: object + "401": + description: Unauthorized + schema: + $ref: '#/definitions/httpapi.ErrorEnvelope' + security: + - BearerAuth: [] + summary: 创建可灵 V1 Omni 视频任务 + tags: + - kling-compatible + /kling/v1/videos/omni-video/{taskID}: + get: + parameters: + - description: 任务 ID + in: path + name: taskID + required: true + type: string + produces: + - application/json + responses: + "200": + description: OK + schema: + additionalProperties: true + type: object + "404": + description: Not Found + schema: + additionalProperties: true + type: object + security: + - BearerAuth: [] + summary: 查询可灵 V1 Omni 视频任务 + tags: + - kling-compatible /music/generations: post: consumes: diff --git a/apps/api/internal/clients/clients_test.go b/apps/api/internal/clients/clients_test.go index 0a4421e..f941145 100644 --- a/apps/api/internal/clients/clients_test.go +++ b/apps/api/internal/clients/clients_test.go @@ -2612,6 +2612,47 @@ func TestKelingOmniPayloadConvertsGatewayContent(t *testing.T) { } } +func TestKelingOmniPayloadPreservesLegacyV1Options(t *testing.T) { + payload, cleanupIDs, err := (KelingClient{}).kelingOmniPayload(context.Background(), Request{ + Kind: "videos.generations", + ModelType: "omni_video", + Body: map[string]any{ + "multi_shot": true, + "shot_type": "customize", + "multi_prompt": []any{ + map[string]any{"index": 1, "prompt": "镜头一", "duration": 7}, + map[string]any{"index": 2, "prompt": "镜头二", "duration": 8}, + }, + "resolution": "1080p", + "callback_url": "https://example.com/callback", + "external_task_id": "client-task-1", + "watermark_info": map[string]any{"enabled": true}, + "voice_list": []any{map[string]any{"voice_id": "voice-1"}}, + }, + Candidate: store.RuntimeModelCandidate{ + Provider: "keling", + ProviderModelName: "kling-v3-omni", + Capabilities: map[string]any{"omni_video": map[string]any{}}, + }, + }, "token") + if err != nil { + t.Fatalf("build keling legacy V1 payload: %v", err) + } + if len(cleanupIDs) != 0 { + t.Fatalf("unexpected cleanup ids: %+v", cleanupIDs) + } + if payload["multi_shot"] != true || payload["shot_type"] != "customize" || payload["duration"] != "15" { + t.Fatalf("unexpected multi-shot payload: %+v", payload) + } + if payload["callback_url"] != "https://example.com/callback" || payload["external_task_id"] != "client-task-1" { + t.Fatalf("legacy task options were not preserved: %+v", payload) + } + watermark := mapFromAny(payload["watermark_info"]) + if watermark["enabled"] != true || len(mapListFromAny(payload["voice_list"])) != 1 { + t.Fatalf("watermark or voice options were not preserved: %+v", payload) + } +} + func TestKelingClientVideoResumePollsWithoutSubmitting(t *testing.T) { var submitCalled bool var pollPath string diff --git a/apps/api/internal/clients/keling.go b/apps/api/internal/clients/keling.go index 310595b..55d5071 100644 --- a/apps/api/internal/clients/keling.go +++ b/apps/api/internal/clients/keling.go @@ -420,15 +420,21 @@ func (c KelingClient) kelingOmniPayload(ctx context.Context, request Request, to } uploadedElementIDs = append(uploadedElementIDs, createdIDs...) shots := kelingShotPrompts(content) - hasMultiPrompt := len(shots) > 0 + rawMultiPrompt := mapListFromAny(body["multi_prompt"]) + hasMultiPrompt := len(shots) > 0 || len(rawMultiPrompt) > 0 + multiShot := boolValue(body, "multi_shot") || hasMultiPrompt hasVideo := len(videos) > 0 hasVideoEdit := kelingHasBaseVideo(videos) hasFirstFrame := kelingHasFirstFrame(images) + watermarkEnabled := boolValue(body, "watermark") + if watermarkInfo := mapFromAny(body["watermark_info"]); watermarkInfo != nil { + watermarkEnabled = boolValue(watermarkInfo, "enabled") + } payload := map[string]any{ "model_name": upstreamModelName(request.Candidate), "mode": kelingModeByResolution(firstNonEmptyStringValue(body, "resolution", "size")), - "watermark_info": map[string]any{"enabled": false}, + "watermark_info": map[string]any{"enabled": watermarkEnabled}, "negative_prompt": strings.TrimSpace(stringFromAny(body["negative_prompt"])), } if !hasMultiPrompt { @@ -449,29 +455,62 @@ func (c KelingClient) kelingOmniPayload(ctx context.Context, request Request, to if len(elements) > 0 { payload["element_list"] = elements } + if voices := mapListFromAny(body["voice_list"]); len(voices) > 0 { + payload["voice_list"] = voices + } if (boolValue(body, "audio") || boolValue(body, "output_audio")) && !hasVideo { payload["sound"] = "on" } - if hasMultiPrompt { + if multiShot { payload["multi_shot"] = true - payload["shot_type"] = "customize" - total := 0.0 - multiPrompt := make([]any, 0, len(shots)) - for index, shot := range shots { - duration := shot.duration - if duration <= 0 { - duration = 5 + shotType := strings.TrimSpace(firstNonEmptyStringValue(body, "shot_type", "shotType")) + if shotType == "" { + if hasMultiPrompt { + shotType = "customize" + } else { + shotType = "intelligence" } - total += duration - multiPrompt = append(multiPrompt, map[string]any{ - "index": index + 1, - "prompt": shot.text, - "duration": fmtDuration(duration, 5), - }) } - delete(payload, "prompt") - payload["multi_prompt"] = multiPrompt - payload["duration"] = fmtDuration(total, 0) + payload["shot_type"] = shotType + if shotType == "customize" { + total := 0.0 + multiPrompt := make([]any, 0, len(rawMultiPrompt)+len(shots)) + if len(rawMultiPrompt) > 0 { + for index, item := range rawMultiPrompt { + duration := numericValue(item["duration"], 0) + total += duration + multiPrompt = append(multiPrompt, map[string]any{ + "index": int(math.Round(numericValue(item["index"], float64(index+1)))), + "prompt": strings.TrimSpace(stringFromAny(item["prompt"])), + "duration": fmtDuration(duration, 0), + }) + } + } else { + for index, shot := range shots { + duration := shot.duration + if duration <= 0 { + duration = 5 + } + total += duration + multiPrompt = append(multiPrompt, map[string]any{ + "index": index + 1, + "prompt": shot.text, + "duration": fmtDuration(duration, 5), + }) + } + } + delete(payload, "prompt") + payload["multi_prompt"] = multiPrompt + if total > 0 { + payload["duration"] = fmtDuration(total, 0) + } + } + } + if callbackURL := strings.TrimSpace(firstNonEmptyStringValue(body, "callback_url", "callbackUrl")); callbackURL != "" { + payload["callback_url"] = callbackURL + } + if externalTaskID := strings.TrimSpace(firstNonEmptyStringValue(body, "external_task_id", "externalTaskId")); externalTaskID != "" { + payload["external_task_id"] = externalTaskID } deleteEmptyStringFields(payload) if hasVideoEdit { @@ -1140,9 +1179,15 @@ func kelingVideoSuccessResult(request Request, upstreamTaskID string, raw map[st continue } item := map[string]any{"url": url, "video_url": url, "type": "video"} + if id := strings.TrimSpace(stringFromAny(video["id"])); id != "" { + item["id"] = id + } if duration := intFromAny(video["duration"]); duration > 0 { item["duration"] = duration } + if watermarkURL := strings.TrimSpace(stringFromAny(video["watermark_url"])); watermarkURL != "" { + item["watermark_url"] = watermarkURL + } items = append(items, item) } created := intFromAny(data["created_at"]) diff --git a/apps/api/internal/clients/keling_live_test.go b/apps/api/internal/clients/keling_live_test.go new file mode 100644 index 0000000..fd5ebb3 --- /dev/null +++ b/apps/api/internal/clients/keling_live_test.go @@ -0,0 +1,83 @@ +package clients + +import ( + "context" + "os" + "strings" + "testing" + "time" + + "github.com/easyai/easyai-ai-gateway/apps/api/internal/store" +) + +// TestKelingLegacyV1LiveOmni is opt-in because it creates billable upstream +// video tasks. Credentials must only be supplied through local environment +// variables; the test never prints them. +func TestKelingLegacyV1LiveOmni(t *testing.T) { + if strings.TrimSpace(os.Getenv("KELING_LIVE_TEST")) != "1" { + t.Skip("set KELING_LIVE_TEST=1 to run billable Kling V1 integration tests") + } + baseURL := strings.TrimRight(strings.TrimSpace(os.Getenv("KELING_TEST_BASE_URL")), "/") + accessKey := strings.TrimSpace(os.Getenv("KELING_TEST_ACCESS_KEY")) + secretKey := strings.TrimSpace(os.Getenv("KELING_TEST_SECRET_KEY")) + if baseURL == "" || accessKey == "" || secretKey == "" { + t.Fatal("KELING_TEST_BASE_URL, KELING_TEST_ACCESS_KEY, and KELING_TEST_SECRET_KEY are required") + } + + models := []string{"kling-video-o1", "kling-v3-omni"} + if selected := strings.TrimSpace(os.Getenv("KELING_LIVE_TEST_MODELS")); selected != "" { + models = strings.Split(selected, ",") + } + for _, model := range models { + model = strings.TrimSpace(model) + t.Run(model, func(t *testing.T) { + duration := 3 + if model == "kling-video-o1" { + duration = 5 + } + ctx, cancel := context.WithTimeout(context.Background(), 15*time.Minute) + defer cancel() + response, err := (KelingClient{}).Run(ctx, Request{ + Kind: "videos.generations", + ModelType: "omni_video", + Model: model, + Body: map[string]any{ + "prompt": "清晨的湖面上，一只白色纸鹤缓慢飞过，镜头平稳推进", + "duration": duration, + "aspect_ratio": "16:9", + "resolution": "720p", + "sound": "off", + }, + Candidate: store.RuntimeModelCandidate{ + BaseURL: baseURL, + Provider: "keling", + AuthType: "AccessKey-SecretKey", + ProviderModelName: model, + Credentials: map[string]any{ + "accessKey": accessKey, + "secretKey": secretKey, + }, + Capabilities: map[string]any{"omni_video": map[string]any{}}, + PlatformConfig: map[string]any{ + "kelingPollIntervalMs": 5000, + "kelingPollTimeoutSeconds": 840, + }, + }, + }) + if err != nil { + t.Fatalf("Kling V1 %s live task failed: %v", model, err) + } + items, _ := response.Result["data"].([]any) + if len(items) == 0 { + t.Fatalf("Kling V1 %s returned no video", model) + } + item, _ := items[0].(map[string]any) + if strings.TrimSpace(stringFromAny(item["url"])) == "" { + t.Fatalf("Kling V1 %s returned an empty video URL", model) + } + if strings.TrimSpace(response.RequestID) == "" { + t.Fatalf("Kling V1 %s returned an empty request id", model) + } + }) + } +} diff --git a/apps/api/internal/httpapi/kling_compat.go b/apps/api/internal/httpapi/kling_compat.go new file mode 100644 index 0000000..79636a3 --- /dev/null +++ b/apps/api/internal/httpapi/kling_compat.go @@ -0,0 +1,908 @@ +package httpapi + +import ( + "encoding/base64" + "encoding/json" + "errors" + "fmt" + "io" + "net/http" + "strconv" + "strings" + "time" + "unicode/utf8" + + "github.com/easyai/easyai-ai-gateway/apps/api/internal/auth" + "github.com/easyai/easyai-ai-gateway/apps/api/internal/clients" + "github.com/easyai/easyai-ai-gateway/apps/api/internal/store" +) + +const ( + klingCompatProvider = "kling" + klingO1Model = "kling-video-o1" + klingV3OmniModel = "kling-v3-omni" +) + +func (s *Server) registerKlingCompatibilityRoutes(mux *http.ServeMux) { + handler := func(next http.HandlerFunc) http.Handler { + return s.requireUser(auth.PermissionBasic, http.HandlerFunc(next)) + } + mux.Handle("POST /kling/v1/videos/omni-video", handler(s.klingV1CreateOmniVideo)) + mux.Handle("GET /kling/v1/videos/omni-video", handler(s.klingV1ListOmniVideos)) + mux.Handle("GET /kling/v1/videos/omni-video/{taskID}", handler(s.klingV1GetOmniVideo)) + + // Kling API 2.0 uses model-specific paths and a shared /tasks resource. + mux.Handle("POST /kling/omni-video/{model}", handler(s.klingV2CreateOmniVideo)) + mux.Handle("GET /kling/tasks", handler(s.klingV2GetTasks)) + mux.Handle("POST /kling/tasks", handler(s.klingV2ListTasks)) + // Versioned aliases help clients that keep the protocol version in their base path. + mux.Handle("POST /kling/v2/omni-video/{model}", handler(s.klingV2CreateOmniVideo)) + mux.Handle("GET /kling/v2/tasks", handler(s.klingV2GetTasks)) + mux.Handle("POST /kling/v2/tasks", handler(s.klingV2ListTasks)) +} + +// klingV1CreateOmniVideo godoc +// @Summary 创建可灵 V1 Omni 视频任务 +// @Description 兼容中国区可灵 V1 /v1/videos/omni-video；用户使用网关 API Key，网关在服务端使用 AK/SK 调用上游。 +// @Tags kling-compatible +// @Accept json +// @Produce json +// @Security BearerAuth +// @Param input body map[string]interface{} true "可灵 V1 Omni 请求" +// @Success 200 {object} map[string]interface{} +// @Failure 400 {object} map[string]interface{} +// @Failure 401 {object} ErrorEnvelope +// @Router /kling/v1/videos/omni-video [post] +func (s *Server) klingV1CreateOmniVideo(w http.ResponseWriter, r *http.Request) { + var native map[string]any + if err := decodeKlingJSON(r, &native); err != nil { + writeKlingCompatError(w, http.StatusBadRequest, "invalid json body", "invalid_json_body") + return + } + model := strings.TrimSpace(stringFromRequestAny(native["model_name"])) + if model == "" { + model = klingO1Model + } + s.createKlingCompatTask(w, r, "v1", model, native) +} + +// klingV2CreateOmniVideo godoc +// @Summary 创建可灵 API 2.0 Omni 视频任务 +// @Description 兼容可灵 API 2.0 的模型路径；调用方使用网关 API Key，网关转换并使用中国区 V1 AK/SK 上游。 +// @Tags kling-compatible +// @Accept json +// @Produce json +// @Security BearerAuth +// @Param model path string true "模型路径（kling-o1 或 kling-v3-omni）" +// @Param input body map[string]interface{} true "可灵 API 2.0 请求" +// @Success 200 {object} map[string]interface{} +// @Failure 400 {object} map[string]interface{} +// @Failure 401 {object} ErrorEnvelope +// @Router /kling/omni-video/{model} [post] +func (s *Server) klingV2CreateOmniVideo(w http.ResponseWriter, r *http.Request) { + model, ok := klingV2ProviderModel(r.PathValue("model")) + if !ok { + writeKlingCompatError(w, http.StatusNotFound, "unsupported Kling Omni model", "model_not_found") + return + } + var native map[string]any + if err := decodeKlingJSON(r, &native); err != nil { + writeKlingCompatError(w, http.StatusBadRequest, "invalid json body", "invalid_json_body") + return + } + s.createKlingCompatTask(w, r, "v2", model, native) +} + +func (s *Server) createKlingCompatTask(w http.ResponseWriter, r *http.Request, version string, model string, native map[string]any) { + user, ok := auth.UserFromContext(r.Context()) + if !ok { + writeKlingCompatError(w, http.StatusUnauthorized, "unauthorized", "unauthorized") + return + } + if !apiKeyScopeAllowed(user, "videos.generations") { + writeKlingCompatError(w, http.StatusForbidden, "api key scope does not allow video generation", "permission_denied") + return + } + body, externalTaskID, err := klingCompatTaskBody(version, model, native) + if err != nil { + writeKlingCompatError(w, http.StatusBadRequest, err.Error(), clients.ErrorCode(err)) + return + } + prepared, err := s.prepareTaskRequest(r.Context(), r, user, body) + if err != nil { + writeKlingCompatError(w, http.StatusBadRequest, err.Error(), clients.ErrorCode(err)) + return + } + createInput := store.CreateTaskInput{ + Kind: "videos.generations", + Model: model, + ExternalTaskID: externalTaskID, + RunMode: runModeFromRequest(prepared.Body), + Async: true, + Request: prepared.Body, + } + if idempotencyKey, hasKey, keyErr := optionalTaskIdempotencyKey(r); keyErr != nil { + writeKlingCompatError(w, http.StatusBadRequest, "invalid Idempotency-Key", "invalid_idempotency_key") + return + } else if hasKey { + createInput.IdempotencyKeyHash = taskIdempotencyKeyHash(idempotencyKey) + createInput.IdempotencyRequestHash = taskIdempotencyRequestHash(createInput.Kind, true, false, prepared.Body) + } + created, err := s.store.CreateTaskIdempotent(r.Context(), createInput, user) + if err != nil { + switch { + case errors.Is(err, store.ErrIdempotencyKeyReused): + writeKlingCompatError(w, http.StatusConflict, err.Error(), "idempotency_key_reused") + case errors.Is(err, store.ErrExternalTaskIDReused): + writeKlingCompatError(w, http.StatusConflict, err.Error(), "external_task_id_reused") + default: + s.logger.Error("create Kling compatibility task failed", "version", version, "model", model, "error", err) + writeKlingCompatError(w, http.StatusInternalServerError, "create task failed", "task_create_failed") + } + return + } + task := created.Task + if !created.Replayed { + if err := s.runner.EnqueueAsyncTask(r.Context(), task); err != nil { + writeKlingCompatError(w, http.StatusInternalServerError, err.Error(), "enqueue_failed") + return + } + } else { + w.Header().Set("Idempotent-Replayed", "true") + } + if version == "v2" { + writeJSON(w, http.StatusOK, klingV2Envelope(task)) + return + } + writeJSON(w, http.StatusOK, klingV1Envelope(task)) +} + +func klingCompatTaskBody(version string, model string, native map[string]any) (map[string]any, string, error) { + if model != klingO1Model && model != klingV3OmniModel { + return nil, "", &clients.ClientError{Code: "invalid_parameter", Message: "model_name must be kling-video-o1 or kling-v3-omni", StatusCode: 400, Retryable: false} + } + if native == nil { + native = map[string]any{} + } + body := cloneMap(native) + if version == "v2" { + body = klingV2ToLegacyBody(native) + } + body["model"] = model + body["modelType"] = "omni_video" + body["_compat_provider"] = klingCompatProvider + body["_kling_compat_version"] = version + body["content"] = klingLegacyContent(body) + mode := strings.TrimSpace(stringFromRequestAny(body["mode"])) + if mode == "" && version == "v1" { + // The legacy Omni API defaults to professional (1080p) mode. + mode = "pro" + body["mode"] = mode + } + if mode != "" { + resolution, ok := klingResolutionFromMode(mode) + if !ok { + return nil, "", &clients.ClientError{Code: "invalid_parameter", Message: "mode must be std, pro, or 4k", StatusCode: 400, Retryable: false} + } + body["resolution"] = resolution + } + if strings.EqualFold(strings.TrimSpace(stringFromRequestAny(body["sound"])), "on") { + body["audio"] = true + } + externalTaskID := strings.TrimSpace(stringFromRequestAny(body["external_task_id"])) + if err := validateKlingCompatBody(model, body); err != nil { + return nil, "", err + } + return body, externalTaskID, nil +} + +func klingV2ToLegacyBody(native map[string]any) map[string]any { + body := map[string]any{} + for _, key := range []string{"runMode", "simulation", "simulationDurationMs", "simulationProfile"} { + if value, ok := native[key]; ok { + body[key] = value + } + } + settings, _ := native["settings"].(map[string]any) + options, _ := native["options"].(map[string]any) + if settings == nil { + settings = map[string]any{} + } + if options == nil { + options = map[string]any{} + } + if resolution := strings.TrimSpace(stringFromRequestAny(settings["resolution"])); resolution != "" { + switch strings.ToLower(resolution) { + case "720p": + body["mode"] = "std" + case "1080p": + body["mode"] = "pro" + case "4k", "2160p": + body["mode"] = "4k" + default: + body["mode"] = resolution + } + } + for _, key := range []string{"aspect_ratio", "duration", "multi_shot", "shot_type", "multi_prompt"} { + if value, ok := settings[key]; ok { + body[key] = value + } + } + audio := strings.ToLower(strings.TrimSpace(stringFromRequestAny(settings["audio"]))) + if audio == "native" || audio == "on" { + body["sound"] = "on" + } else { + body["sound"] = "off" + } + for _, key := range []string{"callback_url", "external_task_id", "watermark_info"} { + if value, ok := options[key]; ok { + body[key] = value + } + } + contents, _ := native["contents"].([]any) + imageList := make([]any, 0) + videoList := make([]any, 0) + elementList := make([]any, 0) + for _, raw := range contents { + item, _ := raw.(map[string]any) + kind := strings.ToLower(strings.TrimSpace(stringFromRequestAny(item["type"]))) + switch kind { + case "prompt": + body["prompt"] = stringFromRequestAny(item["text"]) + case "first_frame", "last_frame", "refer_image", "reference_image": + image := map[string]any{"image_url": firstNonEmptyRequestString(item, "url", "image_url")} + if kind == "first_frame" { + image["type"] = "first_frame" + } else if kind == "last_frame" { + image["type"] = "end_frame" + } + imageList = append(imageList, image) + case "feature_video", "base_video", "refer_video", "reference_video": + referType := "feature" + if kind == "base_video" { + referType = "base" + } + video := map[string]any{ + "video_url": firstNonEmptyRequestString(item, "url", "video_url"), + "refer_type": referType, + "keep_original_sound": firstNonEmptyRequestString(item, "keep_original_sound", "keepOriginalSound"), + } + if audio == "original" && video["keep_original_sound"] == "" { + video["keep_original_sound"] = "yes" + } + videoList = append(videoList, video) + case "element": + elementList = append(elementList, map[string]any{"element_id": firstPresentRequest(item["element_id"], item["id"])}) + } + } + if len(imageList) > 0 { + body["image_list"] = imageList + } + if len(videoList) > 0 { + body["video_list"] = videoList + } + if len(elementList) > 0 { + body["element_list"] = elementList + } + return body +} + +func klingLegacyContent(body map[string]any) []any { + content := make([]any, 0) + if prompt := strings.TrimSpace(stringFromRequestAny(body["prompt"])); prompt != "" { + content = append(content, map[string]any{"type": "text", "text": prompt}) + } + for _, raw := range mapListFromRequest(body["image_list"]) { + role := "reference_image" + switch strings.TrimSpace(stringFromRequestAny(raw["type"])) { + case "first_frame": + role = "first_frame" + case "end_frame", "last_frame": + role = "last_frame" + } + content = append(content, map[string]any{ + "type": "image_url", "role": role, + "image_url": map[string]any{"url": firstNonEmptyRequestString(raw, "image_url", "url")}, + }) + } + for _, raw := range mapListFromRequest(body["video_list"]) { + referType := firstNonEmptyRequestString(raw, "refer_type", "referType") + role := "video_feature" + if referType == "base" { + role = "video_base" + } + content = append(content, map[string]any{ + "type": "video_url", "role": role, + "video_url": map[string]any{ + "url": firstNonEmptyRequestString(raw, "video_url", "url"), + "refer_type": referType, + "keep_original_sound": firstNonEmptyRequestString(raw, "keep_original_sound", "keepOriginalSound"), + }, + }) + } + for _, raw := range mapListFromRequest(body["element_list"]) { + content = append(content, map[string]any{ + "type": "element", + "element": map[string]any{"element_id": firstPresentRequest(raw["element_id"], raw["id"])}, + }) + } + return content +} + +func validateKlingCompatBody(model string, body map[string]any) error { + maxDuration := 10 + if model == klingV3OmniModel { + maxDuration = 15 + } + if duration, ok := klingCompatInt(body["duration"]); ok && (duration < 3 || duration > maxDuration) { + return &clients.ClientError{Code: "invalid_parameter", Message: fmt.Sprintf("duration must be between 3 and %d seconds for %s", maxDuration, model), StatusCode: 400, Retryable: false} + } + if model == klingO1Model && boolFromRequestAny(body["multi_shot"]) { + return &clients.ClientError{Code: "invalid_parameter", Message: "kling-video-o1 does not support multi_shot", StatusCode: 400, Retryable: false} + } + if model == klingO1Model && strings.EqualFold(strings.TrimSpace(stringFromRequestAny(body["mode"])), "4k") { + return &clients.ClientError{Code: "invalid_parameter", Message: "kling-video-o1 does not support 4k mode", StatusCode: 400, Retryable: false} + } + if ratio := strings.TrimSpace(stringFromRequestAny(body["aspect_ratio"])); ratio != "" && ratio != "16:9" && ratio != "9:16" && ratio != "1:1" { + return &clients.ClientError{Code: "invalid_parameter", Message: "aspect_ratio must be 16:9, 9:16, or 1:1", StatusCode: 400, Retryable: false} + } + if sound := strings.ToLower(strings.TrimSpace(stringFromRequestAny(body["sound"]))); sound != "" && sound != "on" && sound != "off" { + return &clients.ClientError{Code: "invalid_parameter", Message: "sound must be on or off", StatusCode: 400, Retryable: false} + } + prompt := strings.TrimSpace(stringFromRequestAny(body["prompt"])) + if utf8.RuneCountInString(prompt) > 2500 { + return &clients.ClientError{Code: "invalid_parameter", Message: "prompt must not exceed 2500 characters", StatusCode: 400, Retryable: false} + } + images := mapListFromRequest(body["image_list"]) + videos := mapListFromRequest(body["video_list"]) + elements := mapListFromRequest(body["element_list"]) + for _, image := range images { + if firstNonEmptyRequestString(image, "image_url", "url") == "" { + return &clients.ClientError{Code: "invalid_parameter", Message: "every image_list item requires image_url", StatusCode: 400, Retryable: false} + } + } + for _, video := range videos { + if firstNonEmptyRequestString(video, "video_url", "url") == "" { + return &clients.ClientError{Code: "invalid_parameter", Message: "every video_list item requires video_url", StatusCode: 400, Retryable: false} + } + if referType := strings.TrimSpace(firstNonEmptyRequestString(video, "refer_type", "referType")); referType != "" && referType != "base" && referType != "feature" { + return &clients.ClientError{Code: "invalid_parameter", Message: "video refer_type must be base or feature", StatusCode: 400, Retryable: false} + } + } + for _, element := range elements { + if klingStringAny(firstPresentRequest(element["element_id"], element["id"])) == "" { + return &clients.ClientError{Code: "invalid_parameter", Message: "every element_list item requires element_id", StatusCode: 400, Retryable: false} + } + } + if model == klingO1Model && len(images) == 0 && len(videos) == 0 && len(elements) == 0 { + if duration, ok := klingCompatInt(body["duration"]); ok && duration != 5 && duration != 10 { + return &clients.ClientError{Code: "invalid_parameter", Message: "kling-video-o1 text-only generation supports duration 5 or 10 seconds", StatusCode: 400, Retryable: false} + } + } + if len(videos) > 1 { + return &clients.ClientError{Code: "invalid_parameter", Message: "video_list supports at most one video", StatusCode: 400, Retryable: false} + } + if len(images)+len(elements) > 7 { + return &clients.ClientError{Code: "invalid_parameter", Message: "image_list and element_list support at most seven combined references", StatusCode: 400, Retryable: false} + } + if len(videos) > 0 && len(images)+len(elements) > 4 { + return &clients.ClientError{Code: "invalid_parameter", Message: "requests with video input support at most four image and element references", StatusCode: 400, Retryable: false} + } + if len(videos) > 0 && strings.EqualFold(strings.TrimSpace(stringFromRequestAny(body["sound"])), "on") { + return &clients.ClientError{Code: "invalid_parameter", Message: "sound must be off when video_list is provided", StatusCode: 400, Retryable: false} + } + if model == klingV3OmniModel && len(videos) > 0 { + if duration, ok := klingCompatInt(body["duration"]); ok && duration > 10 { + return &clients.ClientError{Code: "invalid_parameter", Message: "kling-v3-omni video-reference generation supports at most 10 seconds", StatusCode: 400, Retryable: false} + } + } + multiShot := boolFromRequestAny(body["multi_shot"]) + shotType := strings.ToLower(strings.TrimSpace(stringFromRequestAny(body["shot_type"]))) + if shotType != "" && shotType != "customize" && shotType != "intelligence" { + return &clients.ClientError{Code: "invalid_parameter", Message: "shot_type must be customize or intelligence", StatusCode: 400, Retryable: false} + } + if multiShot && shotType == "customize" { + multiPrompt := mapListFromRequest(body["multi_prompt"]) + if len(multiPrompt) == 0 || len(multiPrompt) > 6 { + return &clients.ClientError{Code: "invalid_parameter", Message: "customize multi-shot requires between one and six multi_prompt items", StatusCode: 400, Retryable: false} + } + totalDuration := 0 + for _, shot := range multiPrompt { + shotPrompt := strings.TrimSpace(stringFromRequestAny(shot["prompt"])) + shotDuration, ok := klingCompatInt(shot["duration"]) + if shotPrompt == "" || utf8.RuneCountInString(shotPrompt) > 2500 || !ok || shotDuration <= 0 { + return &clients.ClientError{Code: "invalid_parameter", Message: "every multi_prompt item requires prompt and a positive integer duration", StatusCode: 400, Retryable: false} + } + totalDuration += shotDuration + } + if totalDuration < 3 || totalDuration > maxDuration { + return &clients.ClientError{Code: "invalid_parameter", Message: fmt.Sprintf("multi_prompt duration must total between 3 and %d seconds", maxDuration), StatusCode: 400, Retryable: false} + } + } + if (!multiShot || shotType == "intelligence" || shotType == "") && prompt == "" { + return &clients.ClientError{Code: "invalid_parameter", Message: "prompt is required for single-shot and intelligence multi-shot generation", StatusCode: 400, Retryable: false} + } + return nil +} + +// klingV1GetOmniVideo godoc +// @Summary 查询可灵 V1 Omni 视频任务 +// @Tags kling-compatible +// @Produce json +// @Security BearerAuth +// @Param taskID path string true "任务 ID" +// @Success 200 {object} map[string]interface{} +// @Failure 404 {object} map[string]interface{} +// @Router /kling/v1/videos/omni-video/{taskID} [get] +func (s *Server) klingV1GetOmniVideo(w http.ResponseWriter, r *http.Request) { + user, _ := auth.UserFromContext(r.Context()) + task, err := s.store.GetCompatTask(r.Context(), user, klingCompatProvider, "v1", r.PathValue("taskID")) + if err != nil { + if store.IsNotFound(err) { + writeKlingCompatError(w, http.StatusNotFound, "task not found", "task_not_found") + return + } + writeKlingCompatError(w, http.StatusInternalServerError, "get task failed", "task_query_failed") + return + } + writeJSON(w, http.StatusOK, klingV1Envelope(task)) +} + +// klingV1ListOmniVideos godoc +// @Summary 分页查询可灵 V1 Omni 视频任务 +// @Tags kling-compatible +// @Produce json +// @Security BearerAuth +// @Param pageNum query int false "页码" default(1) +// @Param pageSize query int false "每页数量" default(30) +// @Success 200 {object} map[string]interface{} +// @Router /kling/v1/videos/omni-video [get] +func (s *Server) klingV1ListOmniVideos(w http.ResponseWriter, r *http.Request) { + user, _ := auth.UserFromContext(r.Context()) + page, err := positiveQueryInt(r.URL.Query().Get("pageNum"), 1) + if err != nil || page > 1000 { + writeKlingCompatError(w, http.StatusBadRequest, "invalid pageNum", "invalid_parameter") + return + } + pageSize, err := positiveQueryInt(r.URL.Query().Get("pageSize"), 30) + if err != nil || pageSize > 500 { + writeKlingCompatError(w, http.StatusBadRequest, "invalid pageSize", "invalid_parameter") + return + } + result, err := s.store.ListCompatTasks(r.Context(), user, store.CompatTaskListFilter{Provider: klingCompatProvider, Version: "v1", Page: page, PageSize: pageSize}) + if err != nil { + writeKlingCompatError(w, http.StatusInternalServerError, "list tasks failed", "task_query_failed") + return + } + data := make([]any, 0, len(result.Items)) + for _, task := range result.Items { + data = append(data, klingV1TaskData(task)) + } + writeJSON(w, http.StatusOK, map[string]any{"code": 0, "message": "success", "request_id": klingRequestID(result.Items), "data": data}) +} + +// klingV2GetTasks godoc +// @Summary 按 ID 查询可灵 API 2.0 任务 +// @Tags kling-compatible +// @Produce json +// @Security BearerAuth +// @Param task_ids query string false "逗号分隔的任务 ID" +// @Param external_task_ids query string false "逗号分隔的外部任务 ID" +// @Success 200 {object} map[string]interface{} +// @Router /kling/tasks [get] +func (s *Server) klingV2GetTasks(w http.ResponseWriter, r *http.Request) { + user, _ := auth.UserFromContext(r.Context()) + taskIDs := splitKlingIDs(r.URL.Query().Get("task_ids")) + externalIDs := splitKlingIDs(r.URL.Query().Get("external_task_ids")) + if (len(taskIDs) == 0) == (len(externalIDs) == 0) { + writeKlingCompatError(w, http.StatusBadRequest, "choose exactly one of task_ids or external_task_ids", "invalid_parameter") + return + } + identifiers := taskIDs + if len(externalIDs) > 0 { + identifiers = externalIDs + } + data := make([]any, 0, len(identifiers)) + for _, identifier := range identifiers { + task, err := s.store.GetCompatTask(r.Context(), user, klingCompatProvider, "v2", identifier) + if err != nil { + if store.IsNotFound(err) { + continue + } + writeKlingCompatError(w, http.StatusInternalServerError, "get task failed", "task_query_failed") + return + } + data = append(data, klingV2TaskData(task)) + } + writeJSON(w, http.StatusOK, map[string]any{"code": 0, "message": "success", "request_id": klingRequestIDFromAny(data), "data": data}) +} + +// klingV2ListTasks godoc +// @Summary 分页查询可灵 API 2.0 任务 +// @Tags kling-compatible +// @Accept json +// @Produce json +// @Security BearerAuth +// @Param input body map[string]interface{} true "游标、数量、时间范围和筛选条件" +// @Success 200 {object} map[string]interface{} +// @Router /kling/tasks [post] +func (s *Server) klingV2ListTasks(w http.ResponseWriter, r *http.Request) { + user, _ := auth.UserFromContext(r.Context()) + var body map[string]any + if err := decodeKlingJSON(r, &body); err != nil { + writeKlingCompatError(w, http.StatusBadRequest, "invalid json body", "invalid_json_body") + return + } + page, err := klingCursorPage(stringFromRequestAny(body["cursor"])) + if err != nil { + writeKlingCompatError(w, http.StatusBadRequest, "invalid cursor", "invalid_parameter") + return + } + limit, ok := klingCompatInt(body["limit"]) + if !ok || limit <= 0 { + limit = 100 + } + if limit > 500 { + writeKlingCompatError(w, http.StatusBadRequest, "limit must not exceed 500", "invalid_parameter") + return + } + createdFrom, err := klingMillisTime(body["start_time"]) + if err != nil { + writeKlingCompatError(w, http.StatusBadRequest, "invalid start_time", "invalid_parameter") + return + } + createdTo, err := klingMillisTime(body["end_time"]) + if err != nil { + writeKlingCompatError(w, http.StatusBadRequest, "invalid end_time", "invalid_parameter") + return + } + statuses := klingInternalStatuses(body["filters"]) + result, err := s.store.ListCompatTasks(r.Context(), user, store.CompatTaskListFilter{ + Provider: klingCompatProvider, Version: "v2", Statuses: statuses, + CreatedFrom: createdFrom, CreatedTo: createdTo, Page: page, PageSize: limit, + }) + if err != nil { + writeKlingCompatError(w, http.StatusInternalServerError, "list tasks failed", "task_query_failed") + return + } + items := make([]any, 0, len(result.Items)) + for _, task := range result.Items { + items = append(items, klingV2TaskData(task)) + } + hasMore := page*limit < result.Total + nextCursor := "" + if hasMore { + nextCursor = base64.RawURLEncoding.EncodeToString([]byte(strconv.Itoa(page + 1))) + } + writeJSON(w, http.StatusOK, map[string]any{ + "code": 0, "message": "success", "request_id": klingRequestID(result.Items), + "data": map[string]any{"result": items, "count": len(items), "next_cursor": nextCursor, "has_more": hasMore}, + }) +} + +func klingV1Envelope(task store.GatewayTask) map[string]any { + return map[string]any{"code": 0, "message": "success", "request_id": firstNonEmpty(task.RequestID, task.ID), "data": klingV1TaskData(task)} +} + +func klingV1TaskData(task store.GatewayTask) map[string]any { + data := map[string]any{ + "task_id": task.ID, "task_status": klingV1Status(task.Status), + "task_info": map[string]any{"external_task_id": task.ExternalTaskID}, + "created_at": task.CreatedAt.UnixMilli(), "updated_at": task.UpdatedAt.UnixMilli(), + } + if task.ErrorMessage != "" || task.Error != "" { + data["task_status_msg"] = firstNonEmpty(task.ErrorMessage, task.Error) + } + if watermarkInfo, ok := task.Request["watermark_info"].(map[string]any); ok { + data["watermark_info"] = watermarkInfo + } + videos := klingTaskVideos(task) + if len(videos) > 0 { + data["task_result"] = map[string]any{"videos": videos} + } + return data +} + +func klingV2Envelope(task store.GatewayTask) map[string]any { + return map[string]any{"code": 0, "message": "success", "request_id": firstNonEmpty(task.RequestID, task.ID), "data": klingV2TaskData(task)} +} + +func klingV2TaskData(task store.GatewayTask) map[string]any { + data := map[string]any{ + "id": task.ID, "status": klingV2Status(task.Status), + "create_time": task.CreatedAt.UnixMilli(), "update_time": task.UpdatedAt.UnixMilli(), + "external_id": task.ExternalTaskID, + } + if message := firstNonEmpty(task.ErrorMessage, task.Error); message != "" { + data["message"] = message + } + if outputs := klingV2Outputs(task); len(outputs) > 0 { + data["outputs"] = outputs + } + return data +} + +func klingTaskVideos(task store.GatewayTask) []any { + items, _ := task.Result["data"].([]any) + videos := make([]any, 0, len(items)) + for index, raw := range items { + item, _ := raw.(map[string]any) + url := firstNonEmptyRequestString(item, "url", "video_url") + if url == "" { + continue + } + video := map[string]any{"id": firstNonEmptyRequestString(item, "id")} + if video["id"] == "" { + video["id"] = fmt.Sprintf("%s-%d", task.ID, index+1) + } + video["url"] = url + if watermarkURL := firstNonEmptyRequestString(item, "watermark_url"); watermarkURL != "" { + video["watermark_url"] = watermarkURL + } + if duration := klingStringAny(item["duration"]); duration != "" { + video["duration"] = duration + } + videos = append(videos, video) + } + return videos +} + +func klingV2Outputs(task store.GatewayTask) []any { + videos := klingTaskVideos(task) + outputs := make([]any, 0, len(videos)) + for _, raw := range videos { + video, _ := raw.(map[string]any) + output := cloneMap(video) + output["type"] = "video" + outputs = append(outputs, output) + } + return outputs +} + +func klingV1Status(status string) string { + switch status { + case "succeeded": + return "succeed" + case "failed", "cancelled": + return "failed" + case "running": + return "processing" + default: + return "submitted" + } +} + +func klingV2Status(status string) string { + switch status { + case "succeeded": + return "succeeded" + case "failed", "cancelled": + return "failed" + case "running": + return "processing" + default: + return "submitted" + } +} + +func klingV2ProviderModel(pathModel string) (string, bool) { + switch strings.ToLower(strings.TrimSpace(pathModel)) { + case "kling-o1", "kling-video-o1": + return klingO1Model, true + case "kling-v3-omni", "kling-3.0-omni", "kling-3-omni": + return klingV3OmniModel, true + default: + return "", false + } +} + +func klingResolutionFromMode(mode string) (string, bool) { + switch strings.ToLower(strings.TrimSpace(mode)) { + case "std": + return "720p", true + case "pro": + return "1080p", true + case "4k": + return "2160p", true + default: + return "", false + } +} + +func decodeKlingJSON(r *http.Request, target any) error { + decoder := json.NewDecoder(r.Body) + decoder.UseNumber() + if err := decoder.Decode(target); err != nil { + return err + } + if err := decoder.Decode(&struct{}{}); !errors.Is(err, io.EOF) { + if err != nil { + return err + } + return errors.New("multiple json values") + } + return nil +} + +func writeKlingCompatError(w http.ResponseWriter, status int, message string, code string) { + if strings.TrimSpace(code) == "" { + code = "invalid_request" + } + writeJSON(w, status, map[string]any{ + "code": klingCompatErrorCode(status), + "message": message, + "request_id": "", + "error": code, + }) +} + +func klingCompatErrorCode(status int) int { + switch status { + case http.StatusBadRequest: + return 1001 + case http.StatusUnauthorized: + return 1100 + case http.StatusForbidden: + return 1302 + case http.StatusNotFound: + return 1201 + case http.StatusConflict: + return 1200 + case http.StatusTooManyRequests: + return 1400 + default: + return 5000 + } +} + +func mapListFromRequest(value any) []map[string]any { + items, _ := value.([]any) + if len(items) == 0 { + if typed, ok := value.([]map[string]any); ok { + return typed + } + } + out := make([]map[string]any, 0, len(items)) + for _, item := range items { + if mapped, ok := item.(map[string]any); ok { + out = append(out, mapped) + } + } + return out +} + +func firstPresentRequest(values ...any) any { + for _, value := range values { + if value == nil { + continue + } + if text, ok := value.(string); ok { + if strings.TrimSpace(text) != "" { + return value + } + continue + } + return value + } + return nil +} + +func boolFromRequestAny(value any) bool { + switch typed := value.(type) { + case bool: + return typed + case string: + parsed, _ := strconv.ParseBool(strings.TrimSpace(typed)) + return parsed + default: + return false + } +} + +func klingCompatInt(value any) (int, bool) { + text := klingStringAny(value) + if text == "" { + return 0, false + } + number, err := strconv.Atoi(text) + return number, err == nil +} + +func splitKlingIDs(raw string) []string { + parts := strings.Split(raw, ",") + out := make([]string, 0, len(parts)) + for _, part := range parts { + if part = strings.TrimSpace(part); part != "" { + out = append(out, part) + } + } + return out +} + +func klingCursorPage(cursor string) (int, error) { + cursor = strings.TrimSpace(cursor) + if cursor == "" { + return 1, nil + } + decoded, err := base64.RawURLEncoding.DecodeString(cursor) + if err != nil { + return 0, err + } + page, err := strconv.Atoi(string(decoded)) + if err != nil || page <= 0 { + return 0, errors.New("invalid cursor") + } + return page, nil +} + +func klingMillisTime(value any) (*time.Time, error) { + text := klingStringAny(value) + if text == "" { + return nil, nil + } + millis, err := strconv.ParseInt(text, 10, 64) + if err != nil || millis < 0 { + return nil, errors.New("invalid millisecond timestamp") + } + parsed := time.UnixMilli(millis) + return &parsed, nil +} + +func klingInternalStatuses(filters any) []string { + statuses := make([]string, 0) + for _, filter := range mapListFromRequest(filters) { + if stringFromRequestAny(filter["key"]) != "status" { + continue + } + values, _ := filter["values"].([]any) + for _, value := range values { + switch strings.ToLower(strings.TrimSpace(stringFromRequestAny(value))) { + case "submitted": + statuses = append(statuses, "queued") + case "processing": + statuses = append(statuses, "running") + case "succeeded": + statuses = append(statuses, "succeeded") + case "failed": + statuses = append(statuses, "failed", "cancelled") + } + } + } + return statuses +} + +func klingRequestID(tasks []store.GatewayTask) string { + if len(tasks) == 0 { + return "" + } + return firstNonEmpty(tasks[0].RequestID, tasks[0].ID) +} + +func klingRequestIDFromAny(items []any) string { + if len(items) == 0 { + return "" + } + item, _ := items[0].(map[string]any) + return stringFromRequestAny(item["id"]) +} + +func klingStringAny(value any) string { + switch typed := value.(type) { + case string: + return strings.TrimSpace(typed) + case json.Number: + return typed.String() + case float64: + return strconv.FormatFloat(typed, 'f', -1, 64) + case float32: + return strconv.FormatFloat(float64(typed), 'f', -1, 32) + case int: + return strconv.Itoa(typed) + case int64: + return strconv.FormatInt(typed, 10) + case int32: + return strconv.FormatInt(int64(typed), 10) + default: + return "" + } +} diff --git a/apps/api/internal/httpapi/kling_compat_integration_test.go b/apps/api/internal/httpapi/kling_compat_integration_test.go new file mode 100644 index 0000000..657439b --- /dev/null +++ b/apps/api/internal/httpapi/kling_compat_integration_test.go @@ -0,0 +1,193 @@ +package httpapi + +import ( + "context" + "io" + "log/slog" + "net/http" + "net/http/httptest" + "os" + "strconv" + "strings" + "testing" + "time" + + "github.com/easyai/easyai-ai-gateway/apps/api/internal/config" + "github.com/easyai/easyai-ai-gateway/apps/api/internal/store" +) + +func TestKlingCompatibilitySimulationFlow(t *testing.T) { + databaseURL := strings.TrimSpace(os.Getenv("AI_GATEWAY_TEST_DATABASE_URL")) + if databaseURL == "" { + t.Skip("set AI_GATEWAY_TEST_DATABASE_URL to run the Kling compatibility integration flow") + } + ctx := context.Background() + applyMigration(t, ctx, databaseURL) + db, err := store.Connect(ctx, databaseURL) + if err != nil { + t.Fatalf("connect store: %v", err) + } + defer db.Close() + + serverCtx, cancelServer := context.WithCancel(ctx) + defer cancelServer() + server := httptest.NewServer(NewServerWithContext(serverCtx, config.Config{ + AppEnv: "test", + HTTPAddr: ":0", + DatabaseURL: databaseURL, + IdentityMode: "hybrid", + JWTSecret: "test-secret", + CORSAllowedOrigin: "*", + }, db, slog.New(slog.NewTextHandler(io.Discard, nil)))) + defer server.Close() + + suffix := strconv.FormatInt(time.Now().UnixNano(), 10) + username := "kling_compat_" + suffix + password := "password123" + var registerResponse struct { + AccessToken string `json:"accessToken"` + } + doJSON(t, server.URL, http.MethodPost, "/api/v1/auth/register", "", map[string]any{ + "username": username, + "email": username + "@example.com", + "password": password, + }, http.StatusCreated, ®isterResponse) + var apiKeyResponse struct { + Secret string `json:"secret"` + } + doJSON(t, server.URL, http.MethodPost, "/api/v1/api-keys", registerResponse.AccessToken, map[string]any{ + "name": "Kling compatibility key", + }, http.StatusCreated, &apiKeyResponse) + if _, err := db.Pool().Exec(ctx, `UPDATE gateway_users SET roles = '["admin"]'::jsonb WHERE username = $1`, username); err != nil { + t.Fatalf("promote compatibility user: %v", err) + } + var loginResponse struct { + AccessToken string `json:"accessToken"` + } + doJSON(t, server.URL, http.MethodPost, "/api/v1/auth/login", "", map[string]any{ + "account": username, "password": password, + }, http.StatusOK, &loginResponse) + + var platform struct { + ID string `json:"id"` + } + doJSON(t, server.URL, http.MethodPost, "/api/admin/platforms", loginResponse.AccessToken, map[string]any{ + "provider": "keling", + "platformKey": "kling-compat-" + suffix, + "name": "Kling Compatibility Simulation", + "baseUrl": "https://api-beijing.klingai.com/v1", + "authType": "AccessKey-SecretKey", + "credentials": map[string]any{"accessKey": "test-ak", "secretKey": "test-sk"}, + }, http.StatusCreated, &platform) + for _, model := range []string{klingO1Model, klingV3OmniModel} { + doJSON(t, server.URL, http.MethodPost, "/api/admin/platforms/"+platform.ID+"/models", loginResponse.AccessToken, map[string]any{ + "canonicalModelKey": "keling:" + model, + "modelName": model, + "providerModelName": model, + "modelAlias": model, + "modelType": []string{"omni_video"}, + "displayName": model, + }, http.StatusCreated, nil) + } + + createV1 := func(model string, duration int, externalID string) string { + t.Helper() + var response map[string]any + doJSON(t, server.URL, http.MethodPost, "/kling/v1/videos/omni-video", apiKeyResponse.Secret, map[string]any{ + "model_name": model, + "prompt": "兼容接口模拟任务", + "duration": duration, + "mode": "std", + "external_task_id": externalID, + "runMode": "simulation", + "simulation": true, + "simulationDurationMs": 5, + }, http.StatusOK, &response) + if response["code"] != float64(0) { + t.Fatalf("unexpected V1 response: %#v", response) + } + data, _ := response["data"].(map[string]any) + taskID, _ := data["task_id"].(string) + if taskID == "" { + t.Fatalf("V1 response missing task id: %#v", response) + } + return taskID + } + + o1TaskID := createV1(klingO1Model, 5, "compat-o1-"+suffix) + v3TaskID := createV1(klingV3OmniModel, 15, "compat-v3-"+suffix) + for _, taskID := range []string{o1TaskID, v3TaskID} { + waitKlingV1SimulationTask(t, server.URL, apiKeyResponse.Secret, taskID) + } + var listResponse map[string]any + doJSON(t, server.URL, http.MethodGet, "/kling/v1/videos/omni-video?pageNum=1&pageSize=10", apiKeyResponse.Secret, nil, http.StatusOK, &listResponse) + items, _ := listResponse["data"].([]any) + if len(items) < 2 { + t.Fatalf("V1 task list did not return compatibility tasks: %#v", listResponse) + } + + var duplicateResponse map[string]any + doJSON(t, server.URL, http.MethodPost, "/kling/v1/videos/omni-video", apiKeyResponse.Secret, map[string]any{ + "model_name": klingO1Model, "prompt": "duplicate", "duration": 5, + "external_task_id": "compat-o1-" + suffix, + "runMode": "simulation", "simulation": true, + }, http.StatusConflict, &duplicateResponse) + if duplicateResponse["code"] != float64(1200) || duplicateResponse["error"] != "external_task_id_reused" { + t.Fatalf("unexpected duplicate external id response: %#v", duplicateResponse) + } + + var v2Response map[string]any + doJSON(t, server.URL, http.MethodPost, "/kling/v2/omni-video/kling-v3-omni", apiKeyResponse.Secret, map[string]any{ + "contents": []any{map[string]any{"type": "prompt", "text": "API 2.0 模拟任务"}}, + "settings": map[string]any{"duration": 3, "resolution": "720p", "aspect_ratio": "16:9", "audio": "off"}, + "options": map[string]any{"external_task_id": "compat-v2-" + suffix}, + "runMode": "simulation", + "simulation": true, + "simulationDurationMs": 5, + }, http.StatusOK, &v2Response) + v2Data, _ := v2Response["data"].(map[string]any) + v2TaskID, _ := v2Data["id"].(string) + if v2TaskID == "" { + t.Fatalf("V2 response missing task id: %#v", v2Response) + } + waitKlingV2SimulationTask(t, server.URL, apiKeyResponse.Secret, v2TaskID) +} + +func waitKlingV1SimulationTask(t *testing.T, baseURL string, apiKey string, taskID string) { + t.Helper() + deadline := time.Now().Add(5 * time.Second) + for time.Now().Before(deadline) { + var response map[string]any + doJSON(t, baseURL, http.MethodGet, "/kling/v1/videos/omni-video/"+taskID, apiKey, nil, http.StatusOK, &response) + data, _ := response["data"].(map[string]any) + switch data["task_status"] { + case "succeed": + return + case "failed": + t.Fatalf("V1 simulation task failed: %#v", response) + } + time.Sleep(20 * time.Millisecond) + } + t.Fatalf("V1 simulation task %s timed out", taskID) +} + +func waitKlingV2SimulationTask(t *testing.T, baseURL string, apiKey string, taskID string) { + t.Helper() + deadline := time.Now().Add(5 * time.Second) + for time.Now().Before(deadline) { + var response map[string]any + doJSON(t, baseURL, http.MethodGet, "/kling/v2/tasks?task_ids="+taskID, apiKey, nil, http.StatusOK, &response) + items, _ := response["data"].([]any) + if len(items) == 1 { + data, _ := items[0].(map[string]any) + switch data["status"] { + case "succeeded": + return + case "failed": + t.Fatalf("V2 simulation task failed: %#v", response) + } + } + time.Sleep(20 * time.Millisecond) + } + t.Fatalf("V2 simulation task %s timed out", taskID) +} diff --git a/apps/api/internal/httpapi/kling_compat_test.go b/apps/api/internal/httpapi/kling_compat_test.go new file mode 100644 index 0000000..07766de --- /dev/null +++ b/apps/api/internal/httpapi/kling_compat_test.go @@ -0,0 +1,137 @@ +package httpapi + +import ( + "encoding/json" + "net/http" + "net/http/httptest" + "strings" + "testing" +) + +func TestKlingV1O1CompatibilityBody(t *testing.T) { + body, externalID, err := klingCompatTaskBody("v1", klingO1Model, map[string]any{ + "prompt": "一只纸鹤飞过湖面", + "duration": json.Number("10"), + "aspect_ratio": "16:9", + "sound": "on", + "external_task_id": "client-o1-1", + "watermark_info": map[string]any{"enabled": true}, + }) + if err != nil { + t.Fatalf("build O1 compatibility body: %v", err) + } + if externalID != "client-o1-1" || body["model"] != klingO1Model || body["modelType"] != "omni_video" { + t.Fatalf("unexpected identity fields: %#v", body) + } + if body["mode"] != "pro" || body["resolution"] != "1080p" || body["audio"] != true { + t.Fatalf("unexpected V1 defaults: %#v", body) + } + content := mapListFromRequest(body["content"]) + if len(content) != 1 || content[0]["type"] != "text" || content[0]["text"] != "一只纸鹤飞过湖面" { + t.Fatalf("unexpected canonical content: %#v", content) + } +} + +func TestKlingV1V3OmniCompatibilityBody(t *testing.T) { + body, _, err := klingCompatTaskBody("v1", klingV3OmniModel, map[string]any{ + "multi_shot": true, + "shot_type": "customize", + "mode": "4k", + "multi_prompt": []any{ + map[string]any{"index": json.Number("1"), "prompt": "推近人物", "duration": json.Number("7")}, + map[string]any{"index": json.Number("2"), "prompt": "切到城市远景", "duration": json.Number("8")}, + }, + "image_list": []any{ + map[string]any{"image_url": "https://example.com/first.png", "type": "first_frame"}, + }, + }) + if err != nil { + t.Fatalf("build 3.0 Omni compatibility body: %v", err) + } + if body["resolution"] != "2160p" { + t.Fatalf("4k mode was not normalized: %#v", body) + } + content := mapListFromRequest(body["content"]) + if len(content) != 1 || content[0]["role"] != "first_frame" { + t.Fatalf("image input was not normalized: %#v", content) + } +} + +func TestKlingV2CompatibilityBody(t *testing.T) { + body, externalID, err := klingCompatTaskBody("v2", klingV3OmniModel, map[string]any{ + "contents": []any{ + map[string]any{"type": "prompt", "text": "让角色向镜头挥手"}, + map[string]any{"type": "first_frame", "url": "https://example.com/first.png"}, + map[string]any{"type": "element", "id": json.Number("42")}, + }, + "settings": map[string]any{ + "resolution": "1080p", + "duration": json.Number("15"), + "aspect_ratio": "9:16", + "audio": "native", + }, + "options": map[string]any{ + "external_task_id": "client-v2-1", + "callback_url": "https://example.com/callback", + }, + }) + if err != nil { + t.Fatalf("build V2 compatibility body: %v", err) + } + if externalID != "client-v2-1" || body["mode"] != "pro" || body["audio"] != true { + t.Fatalf("unexpected V2 settings: %#v", body) + } + if len(mapListFromRequest(body["image_list"])) != 1 || len(mapListFromRequest(body["element_list"])) != 1 { + t.Fatalf("unexpected V2 references: %#v", body) + } +} + +func TestKlingCompatibilityValidation(t *testing.T) { + tests := []struct { + name string + model string + body map[string]any + }{ + {name: "O1 duration", model: klingO1Model, body: map[string]any{"prompt": "test", "duration": 11}}, + {name: "O1 text-only flexible duration", model: klingO1Model, body: map[string]any{"prompt": "test", "duration": 3}}, + {name: "O1 4k", model: klingO1Model, body: map[string]any{"prompt": "test", "duration": 5, "mode": "4k"}}, + {name: "O1 multi-shot", model: klingO1Model, body: map[string]any{"multi_shot": true, "multi_prompt": []any{map[string]any{"prompt": "test", "duration": 3}}}}, + {name: "custom multi-shot without prompts", model: klingV3OmniModel, body: map[string]any{"multi_shot": true, "shot_type": "customize"}}, + {name: "intelligence multi-shot without prompt", model: klingV3OmniModel, body: map[string]any{"multi_shot": true, "shot_type": "intelligence"}}, + {name: "video with native audio", model: klingV3OmniModel, body: map[string]any{"prompt": "test", "sound": "on", "video_list": []any{map[string]any{"video_url": "https://example.com/input.mp4"}}}}, + {name: "video duration too long", model: klingV3OmniModel, body: map[string]any{"prompt": "test", "duration": 15, "video_list": []any{map[string]any{"video_url": "https://example.com/input.mp4"}}}}, + {name: "too many references", model: klingV3OmniModel, body: map[string]any{"prompt": "test", "image_list": []any{ + map[string]any{}, map[string]any{}, map[string]any{}, map[string]any{}, map[string]any{}, map[string]any{}, map[string]any{}, map[string]any{}, + }}}, + } + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + if _, _, err := klingCompatTaskBody("v1", test.model, test.body); err == nil { + t.Fatal("expected validation error") + } + }) + } + if _, _, err := klingCompatTaskBody("v1", klingV3OmniModel, map[string]any{"prompt": "test", "duration": 15}); err != nil { + t.Fatalf("3.0 Omni should allow a 15-second duration: %v", err) + } +} + +func TestDecodeKlingJSONRejectsTrailingData(t *testing.T) { + request := httptest.NewRequest(http.MethodPost, "/kling/v1/videos/omni-video", strings.NewReader(`{"prompt":"ok"} trailing`)) + var body map[string]any + if err := decodeKlingJSON(request, &body); err == nil { + t.Fatal("expected trailing JSON error") + } +} + +func TestWriteKlingCompatErrorUsesOfficialNumericEnvelope(t *testing.T) { + recorder := httptest.NewRecorder() + writeKlingCompatError(recorder, http.StatusBadRequest, "bad request", "invalid_parameter") + var body map[string]any + if err := json.Unmarshal(recorder.Body.Bytes(), &body); err != nil { + t.Fatalf("decode error response: %v", err) + } + if body["code"] != float64(1001) || body["error"] != "invalid_parameter" { + t.Fatalf("unexpected error envelope: %#v", body) + } +} diff --git a/apps/api/internal/httpapi/server.go b/apps/api/internal/httpapi/server.go index 4dc9020..7834444 100644 --- a/apps/api/internal/httpapi/server.go +++ b/apps/api/internal/httpapi/server.go @@ -265,6 +265,7 @@ func NewServerWithContext(ctx context.Context, cfg config.Config, db *store.Stor mux.Handle("DELETE /api/v1/voice_clone/voices/{voiceID}", server.requireUser(auth.PermissionBasic, http.HandlerFunc(server.deleteClonedVoice))) mux.Handle("POST /api/v1/files/upload", server.requireUser(auth.PermissionBasic, http.HandlerFunc(server.uploadFile))) server.registerGeminiGenerateContentRoutes(mux) + server.registerKlingCompatibilityRoutes(mux) mux.Handle("POST /upload/{version}/files", server.requireUser(auth.PermissionBasic, http.HandlerFunc(server.geminiFilesUpload))) mux.Handle("POST /upload/{version}/files/{uploadID}", server.requireUser(auth.PermissionBasic, http.HandlerFunc(server.geminiFilesUploadFinalize))) mux.Handle("GET /api/v1/tasks", server.requireUser(auth.PermissionBasic, http.HandlerFunc(server.listTasks))) diff --git a/apps/api/internal/store/postgres.go b/apps/api/internal/store/postgres.go index ee17719..036793f 100644 --- a/apps/api/internal/store/postgres.go +++ b/apps/api/internal/store/postgres.go @@ -63,6 +63,7 @@ var ( ErrBalanceBelowFrozen = errors.New("wallet balance cannot be below frozen balance") ErrInvalidWalletAmount = errors.New("wallet amount must be a decimal with at most nine fractional digits") ErrIdempotencyKeyReused = errors.New("idempotency key was reused for a different request") + ErrExternalTaskIDReused = errors.New("external task id was reused") ErrTaskExecutionLeaseUnavailable = errors.New("task execution lease is unavailable") ErrTaskExecutionLeaseLost = errors.New("task execution lease was lost") ErrTaskExecutionFinished = errors.New("task execution already finished") @@ -465,6 +466,7 @@ type RateLimitWindow struct { type CreateTaskInput struct { Kind string `json:"kind"` Model string `json:"model"` + ExternalTaskID string `json:"externalTaskId,omitempty"` RunMode string `json:"runMode"` Async bool `json:"async"` Request map[string]any `json:"request"` @@ -482,6 +484,7 @@ type CreateTaskResult struct { type GatewayTask struct { ID string `json:"id"` + ExternalTaskID string `json:"externalTaskId,omitempty"` Kind string `json:"kind"` RunMode string `json:"runMode"` UserID string `json:"userId"` @@ -541,7 +544,7 @@ type GatewayTask struct { } const gatewayTaskColumns = ` -id::text, kind, run_mode, user_id, COALESCE(gateway_user_id::text, ''), user_source, +id::text, COALESCE(external_task_id, ''), kind, run_mode, user_id, COALESCE(gateway_user_id::text, ''), user_source, COALESCE(gateway_tenant_id::text, ''), COALESCE(tenant_id, ''), COALESCE(tenant_key, ''), COALESCE(api_key_id, ''), COALESCE(api_key_name, ''), COALESCE(api_key_prefix, ''), COALESCE(user_group_id::text, ''), COALESCE(user_group_key, ''), model, @@ -1948,15 +1951,15 @@ func (s *Store) CreateTaskIdempotent(ctx context.Context, input CreateTaskInput, task, err := scanGatewayTask(tx.QueryRow(ctx, ` INSERT INTO gateway_tasks ( - kind, run_mode, user_id, gateway_user_id, user_source, gateway_tenant_id, tenant_id, tenant_key, + external_task_id, kind, run_mode, user_id, gateway_user_id, user_source, gateway_tenant_id, tenant_id, tenant_key, api_key_id, api_key_name, api_key_prefix, user_group_id, user_group_key, model, requested_model, request, async_mode, status, result, billings, conversation_id, new_message_count, idempotency_key_hash, idempotency_request_hash, finished_at ) - VALUES ($1, $2, $3, NULLIF($4, '')::uuid, COALESCE(NULLIF($5, ''), 'gateway'), NULLIF($6, '')::uuid, NULLIF($7, ''), NULLIF($8, ''), NULLIF($9, ''), NULLIF($10, ''), NULLIF($11, ''), NULLIF($12, '')::uuid, NULLIF($13, ''), $14, $14, $15, $16, $17, $18::jsonb, $19::jsonb, NULLIF($20, '')::uuid, $21, NULLIF($22, ''), NULLIF($23, ''), NULL) + VALUES (NULLIF($1, ''), $2, $3, $4, NULLIF($5, '')::uuid, COALESCE(NULLIF($6, ''), 'gateway'), NULLIF($7, '')::uuid, NULLIF($8, ''), NULLIF($9, ''), NULLIF($10, ''), NULLIF($11, ''), NULLIF($12, ''), NULLIF($13, '')::uuid, NULLIF($14, ''), $15, $15, $16, $17, $18, $19::jsonb, $20::jsonb, NULLIF($21, '')::uuid, $22, NULLIF($23, ''), NULLIF($24, ''), NULL) ON CONFLICT (user_id, idempotency_key_hash) WHERE idempotency_key_hash IS NOT NULL DO NOTHING RETURNING `+gatewayTaskColumns, - input.Kind, runMode, user.ID, user.GatewayUserID, user.Source, user.GatewayTenantID, user.TenantID, user.TenantKey, user.APIKeyID, user.APIKeyName, user.APIKeyPrefix, user.UserGroupID, user.UserGroupKey, input.Model, requestBody, input.Async, status, resultBody, billingsBody, input.ConversationID, input.NewMessageCount, strings.TrimSpace(input.IdempotencyKeyHash), strings.TrimSpace(input.IdempotencyRequestHash), + strings.TrimSpace(input.ExternalTaskID), input.Kind, runMode, user.ID, user.GatewayUserID, user.Source, user.GatewayTenantID, user.TenantID, user.TenantKey, user.APIKeyID, user.APIKeyName, user.APIKeyPrefix, user.UserGroupID, user.UserGroupKey, input.Model, requestBody, input.Async, status, resultBody, billingsBody, input.ConversationID, input.NewMessageCount, strings.TrimSpace(input.IdempotencyKeyHash), strings.TrimSpace(input.IdempotencyRequestHash), )) replayed := false if errors.Is(err, pgx.ErrNoRows) && strings.TrimSpace(input.IdempotencyKeyHash) != "" { @@ -1977,6 +1980,9 @@ FROM gateway_tasks WHERE user_id = $1 AND idempotency_key_hash = $2`, user.ID, strings.TrimSpace(input.IdempotencyKeyHash))) replayed = true } + if isUniqueViolation(err) && strings.TrimSpace(input.ExternalTaskID) != "" { + return CreateTaskResult{}, ErrExternalTaskIDReused + } if err != nil { return CreateTaskResult{}, err } @@ -2041,6 +2047,7 @@ func scanGatewayTask(scanner taskScanner) (GatewayTask, error) { var remoteTaskPayloadBytes []byte if err := scanner.Scan( &task.ID, + &task.ExternalTaskID, &task.Kind, &task.RunMode, &task.UserID, diff --git a/apps/api/internal/store/tasks_runtime.go b/apps/api/internal/store/tasks_runtime.go index fff569a..3d6d804 100644 --- a/apps/api/internal/store/tasks_runtime.go +++ b/apps/api/internal/store/tasks_runtime.go @@ -29,6 +29,16 @@ type TaskListResult struct { PageSize int } +type CompatTaskListFilter struct { + Provider string + Version string + Statuses []string + CreatedFrom *time.Time + CreatedTo *time.Time + Page int + PageSize int +} + func (s *Store) ListTasks(ctx context.Context, user *auth.User, filter TaskListFilter) (TaskListResult, error) { page := filter.Page if page <= 0 { @@ -146,6 +156,122 @@ LIMIT $8 OFFSET $9`, queryArgs...) }, nil } +func (s *Store) GetCompatTask(ctx context.Context, user *auth.User, provider string, version string, identifier string) (GatewayTask, error) { + gatewayUserID := localGatewayUserID(user) + apiKeyID := "" + userID := "" + if user != nil { + apiKeyID = strings.TrimSpace(user.APIKeyID) + userID = strings.TrimSpace(user.ID) + } + if gatewayUserID == "" && userID == "" { + return GatewayTask{}, ErrLocalUserRequired + } + task, err := scanGatewayTask(s.pool.QueryRow(ctx, ` +SELECT `+gatewayTaskColumns+` +FROM gateway_tasks +WHERE ( + (NULLIF($1, '')::uuid IS NOT NULL AND gateway_user_id = NULLIF($1, '')::uuid) + OR (NULLIF($1, '')::uuid IS NULL AND NULLIF($2, '') IS NOT NULL AND user_id = $2) + ) + AND (NULLIF($3, '') IS NULL OR api_key_id = $3) + AND request->>'_compat_provider' = $4 + AND request->>'_kling_compat_version' = $5 + AND (id::text = $6 OR external_task_id = $6) +ORDER BY created_at DESC +LIMIT 1`, gatewayUserID, userID, apiKeyID, strings.TrimSpace(provider), strings.TrimSpace(version), strings.TrimSpace(identifier))) + if err != nil { + return GatewayTask{}, err + } + attempts, err := s.ListTaskAttempts(ctx, task.ID) + if err != nil { + return GatewayTask{}, err + } + task.Attempts = attempts + return task, nil +} + +func (s *Store) ListCompatTasks(ctx context.Context, user *auth.User, filter CompatTaskListFilter) (TaskListResult, error) { + page := filter.Page + if page <= 0 { + page = 1 + } + pageSize := filter.PageSize + if pageSize <= 0 { + pageSize = 30 + } + if pageSize > 500 { + pageSize = 500 + } + offset := (page - 1) * pageSize + gatewayUserID := localGatewayUserID(user) + apiKeyID := "" + userID := "" + if user != nil { + apiKeyID = strings.TrimSpace(user.APIKeyID) + userID = strings.TrimSpace(user.ID) + } + if gatewayUserID == "" && userID == "" { + return TaskListResult{}, ErrLocalUserRequired + } + statuses := filter.Statuses + if len(statuses) == 0 { + statuses = nil + } + args := []any{ + gatewayUserID, + userID, + apiKeyID, + strings.TrimSpace(filter.Provider), + strings.TrimSpace(filter.Version), + nullableTaskListTime(filter.CreatedFrom), + nullableTaskListTime(filter.CreatedTo), + statuses, + } + whereSQL := ` +WHERE ( + (NULLIF($1, '')::uuid IS NOT NULL AND gateway_user_id = NULLIF($1, '')::uuid) + OR (NULLIF($1, '')::uuid IS NULL AND NULLIF($2, '') IS NOT NULL AND user_id = $2) + ) + AND (NULLIF($3, '') IS NULL OR api_key_id = $3) + AND request->>'_compat_provider' = $4 + AND request->>'_kling_compat_version' = $5 + AND ($6::timestamptz IS NULL OR created_at >= $6::timestamptz) + AND ($7::timestamptz IS NULL OR created_at <= $7::timestamptz) + AND ($8::text[] IS NULL OR status = ANY($8::text[]))` + var total int + if err := s.pool.QueryRow(ctx, `SELECT count(*) FROM gateway_tasks `+whereSQL, args...).Scan(&total); err != nil { + return TaskListResult{}, err + } + queryArgs := append(args, pageSize, offset) + rows, err := s.pool.Query(ctx, ` +SELECT `+gatewayTaskColumns+` +FROM gateway_tasks +`+whereSQL+` +ORDER BY created_at DESC +LIMIT $9 OFFSET $10`, queryArgs...) + if err != nil { + return TaskListResult{}, err + } + defer rows.Close() + items := make([]GatewayTask, 0) + for rows.Next() { + task, err := scanGatewayTask(rows) + if err != nil { + return TaskListResult{}, err + } + items = append(items, task) + } + if err := rows.Err(); err != nil { + return TaskListResult{}, err + } + items, err = s.attachTaskAttempts(ctx, items) + if err != nil { + return TaskListResult{}, err + } + return TaskListResult{Items: items, Total: total, Page: page, PageSize: pageSize}, nil +} + func nullableTaskListTime(value *time.Time) any { if value == nil { return nil diff --git a/apps/api/migrations/0071_kling_compat_external_task_id.sql b/apps/api/migrations/0071_kling_compat_external_task_id.sql new file mode 100644 index 0000000..e7f9e21 --- /dev/null +++ b/apps/api/migrations/0071_kling_compat_external_task_id.sql @@ -0,0 +1,9 @@ +CREATE UNIQUE INDEX IF NOT EXISTS uniq_gateway_tasks_kling_compat_external + ON gateway_tasks ( + COALESCE(gateway_user_id::text, user_id), + external_task_id + ) + WHERE external_task_id IS NOT NULL + AND external_task_id <> '' + AND request->>'_compat_provider' = 'kling' + AND request->>'_kling_compat_version' IN ('v1', 'v2'); diff --git a/docs/kling-compatible-api.md b/docs/kling-compatible-api.md new file mode 100644 index 0000000..e42c557 --- /dev/null +++ b/docs/kling-compatible-api.md @@ -0,0 +1,79 @@ +# 可灵 O1 / 3.0 Omni 兼容接口 + +网关兼容中国区可灵 V1 AK/SK Omni API 和 API 2.0 的调用路径。调用方只需要把可灵客户端的 `baseURL` 改为网关地址，并把认证改为 EasyAI Gateway API Key；中国区可灵 AK/SK 仅保存在网关平台配置中，不下发给调用方。 + +生产环境统一配置： + +```text +baseURL = https://ai.51easyai.com/gateway-api/kling +Authorization = Bearer +``` + +本地环境使用 `baseURL = http://localhost:8088/kling`。 + +## V1（AK/SK 旧版协议兼容） + +创建任务： + +```http +POST /v1/videos/omni-video +Authorization: Bearer +Content-Type: application/json + +{ + "model_name": "kling-video-o1", + "prompt": "一只白色纸鹤飞过清晨的湖面", + "mode": "pro", + "duration": "5", + "aspect_ratio": "16:9", + "sound": "off", + "external_task_id": "client-task-001" +} +``` + +`model_name` 支持： + +- `kling-video-o1`：O1，时长 3–10 秒；不带参考素材的纯文本任务只支持 5 或 10 秒。 +- `kling-v3-omni`：3.0 Omni，时长 3–15 秒，支持多镜头。 + +查询和列表接口： + +```http +GET /v1/videos/omni-video/{task_id} +GET /v1/videos/omni-video?pageNum=1&pageSize=30 +``` + +V1 支持并透传 `prompt`、`multi_shot`、`shot_type`、`multi_prompt`、`image_list`、`element_list`、`video_list`、`voice_list`、`sound`、`mode`、`aspect_ratio`、`duration`、`watermark_info`、`callback_url` 和 `external_task_id`。网关会校验模型时长、引用素材数量、参考视频数量以及参考视频与原生音频的互斥约束。 + +## API 2.0 + +创建任务： + +```http +POST /omni-video/kling-o1 +POST /omni-video/kling-v3-omni +``` + +同时提供带显式版本前缀的等价别名： + +```http +POST /v2/omni-video/kling-o1 +POST /v2/omni-video/kling-v3-omni +``` + +任务查询和列表： + +```http +GET /tasks?task_ids= +GET /tasks?external_task_ids= +POST /tasks +``` + +上述任务接口也提供 `/v2/tasks` 别名。API 2.0 请求使用 `contents`、`settings` 和 `options`；网关会转换到相同的 V1 AK/SK 上游链路，因此 O1 和 3.0 Omni 使用同一套中国区平台凭据。 + +## 安全约束 + +- 调用方只使用 EasyAI Gateway API Key，不接触上游 AK/SK。 +- 任务查询按用户和当前 API Key 隔离，不能读取其他 Key 创建的任务。 +- `external_task_id` 在同一用户的可灵兼容任务范围内唯一。 +- 真实联调测试是显式启用且会产生可灵费用；测试凭据只放在 `.env.local`。