diff --git a/internal/cmd/cmd.go b/internal/cmd/cmd.go index 9a48fcd..5ca5c39 100644 --- a/internal/cmd/cmd.go +++ b/internal/cmd/cmd.go @@ -539,6 +539,11 @@ func registerRelayRoutes(server *ghttp.Server) { group.POST("/moderations", relay.HandleModerations) group.POST("/images/edits", relay.HandleImagesEdits) group.GET("/realtime", relay.HandleRealtime) + // OpenAI Videos 协议端点(官方 SDK 兼容,对接第三方视频模型) + group.POST("/videos", relay.HandleVideoCreate) + group.GET("/videos/{video_id}", relay.HandleVideoRetrieve) + group.GET("/videos/{video_id}/content", relay.HandleVideoContent) + group.DELETE("/videos/{video_id}", relay.HandleVideoDelete) }) // Gemini 兼容路由(/v1beta/models/{model}:generateContent) diff --git a/internal/dao/internal/sys_admin_role_perms.go b/internal/dao/internal/sys_admin_role_perms.go index eecd622..17d09d4 100644 --- a/internal/dao/internal/sys_admin_role_perms.go +++ b/internal/dao/internal/sys_admin_role_perms.go @@ -23,7 +23,7 @@ type SysAdminRolePermsDao struct { type SysAdminRolePermsColumns struct { Id string // 主键ID AdminUserId string // 关联的管理员用户ID - PermissionPoint string // 权限点标识(如 tenant:create、channel:edit) + PermissionPoint string // 特批的权限点标识(独立于角色,角色禁用不影响特批) CreatedAt string // 创建时间 UpdatedAt string // 更新时间 } diff --git a/internal/dao/internal/tsk_model_tasks.go b/internal/dao/internal/tsk_model_tasks.go index 53660ff..a725f94 100644 --- a/internal/dao/internal/tsk_model_tasks.go +++ b/internal/dao/internal/tsk_model_tasks.go @@ -46,6 +46,8 @@ type TskModelTasksColumns struct { CreatedAt string // 记录创建时间 UpdatedAt string // 记录更新时间 RequestId string // 任务提交时的原始请求 ID(req_xxxxx),关联 aud_request_logs.request_id + Deleted string // 软删除标记:true 表示已被客户端删除,查询/下载按不存在处理(计费与审计记录保留) + DeletedAt string // 软删除时间 } // tskModelTasksColumns holds the columns for the table tsk_model_tasks. @@ -75,6 +77,8 @@ var tskModelTasksColumns = TskModelTasksColumns{ CreatedAt: "created_at", UpdatedAt: "updated_at", RequestId: "request_id", + Deleted: "deleted", + DeletedAt: "deleted_at", } // NewTskModelTasksDao creates and returns a new DAO object for table data access. diff --git a/internal/handler/relay/videos.go b/internal/handler/relay/videos.go new file mode 100644 index 0000000..055ccad --- /dev/null +++ b/internal/handler/relay/videos.go @@ -0,0 +1,165 @@ +package relay + +import ( + "github.com/gogf/gf/v2/frame/g" + "github.com/gogf/gf/v2/net/ghttp" + + "github.com/qianfree/team-api/internal/logic/monitor" + "github.com/qianfree/team-api/internal/middleware" + relay_constant "github.com/qianfree/team-api/relay/constant" + relay_handler "github.com/qianfree/team-api/relay/handler" +) + +// OpenAI Videos 协议入站胶水(/v1/videos 端点): +// 创建请求在此完成 multipart/JSON 解析与通用任务体转换,复用任务提交管线 +// (selectTaskChannel 渠道调度 + relay_handler.HandleTaskSubmit 计费/提交/落库); +// 查询/删除/内容下载直连 relay_handler 的 Videos 处理器。 + +// HandleVideoCreate 处理 POST /v1/videos(创建视频生成任务) +func HandleVideoCreate(r *ghttp.Request) { + body := r.GetBody() + if len(body) == 0 { + r.Response.WriteStatus(400, g.Map{ + "error": g.Map{"type": "invalid_request_error", "message": "request body is empty"}, + }) + return + } + + // OpenAI Videos 请求(multipart/JSON)→ 归一结构(官方 SDK 一律 multipart,必须支持) + req, taskErr := relay_handler.ParseVideosCreateRequest(body, r.Header.Get("Content-Type")) + if taskErr != nil { + writeVideosError(r, taskErr.StatusCode, taskErr.Message, taskErr.ErrCode) + return + } + if taskErr := relay_handler.ValidateVideosCreateRequest(req); taskErr != nil { + writeVideosError(r, taskErr.StatusCode, taskErr.Message, taskErr.ErrCode) + return + } + + // 转成任务框架通用请求体(适配器消费 model/prompt/seconds/metadata) + taskBody, err := relay_handler.BuildVideosTaskBody(req) + if err != nil { + writeVideosError(r, 500, "build task body failed", "") + return + } + + rc := &relay_handler.TaskRelayContext{ + TenantID: middleware.GetTenantID(r.Context()), + UserID: middleware.GetUserID(r.Context()), + ApiKeyID: middleware.GetApiKeyID(r.Context()), + ProjectID: middleware.GetProjectID(r.Context()), + RequestID: r.GetCtxVar("RequestId").String(), + Writer: r.Response.Writer, + Scope: r.GetCtxVar("ApiKeyScope").String(), + ClientIP: r.GetClientIp(), + KeyRateLimitQps: r.GetCtxVar(middleware.CtxKeyApiKeyRateLimitQps).Int(), + KeyConcurrency: r.GetCtxVar(middleware.CtxKeyApiKeyRateLimitConcurrency).Int(), + KeyIpWhitelist: r.GetCtxVar(middleware.CtxKeyApiKeyIpWhitelist).String(), + KeyTotalQuota: r.GetCtxVar(middleware.CtxKeyApiKeyTotalQuota).Float64(), + KeyUsedQuota: r.GetCtxVar(middleware.CtxKeyApiKeyUsedQuota).Float64(), + // OpenAI Videos 协议:官方 Video 对象响应 + video_ 前缀 ID + 请求回显落库 + Protocol: "openai_videos", + RelayMode: int(relay_constant.RelayModeVideos), + RequestEcho: relay_handler.NewVideosRequestEcho(req), + } + + // 选择渠道(模型权限校验 + 调度引擎,与 /v1/video/generations 同一链路) + channelMeta, err := selectTaskChannel(r, taskBody) + if err != nil { + writeTaskChannelSelectionError(r, err) + return + } + + // 响应捕获用于审计(对齐 HandleTaskSubmit 模式) + capture := NewResponseCaptureWriter(r.Response.Writer) + rc.Writer = capture + + registerAsyncTask(rc.RequestID, rc.TenantID, rc.UserID, rc.ProjectID, req.Model, channelMeta, r.URL.Path) + + relay_handler.HandleTaskSubmit( + r.Context(), taskBody, r.URL.Path, r.Header, + rc, taskDataProvider, taskBillingProvider, channelMeta, + ) + + // 提交成功后切换为任务 ID 跟踪(任务生命周期远超 HTTP 请求) + if rc.TaskID != "" { + monitor.SwitchToTaskID(rc.RequestID, rc.TaskID) + } else { + monitor.UnregisterRequest(rc.RequestID) + } + + rc.ForwardingTrace = buildTaskForwardingTrace(r.URL.Path, taskBody, channelMeta, capture.StatusCode()) + go recordTaskSubmitAudit(r, rc, capture, taskBody) +} + +// HandleVideoRetrieve 处理 GET /v1/videos/{video_id}(查询视频任务) +func HandleVideoRetrieve(r *ghttp.Request) { + videoID := r.Get("video_id").String() + if videoID == "" { + writeVideosError(r, 400, "video id is required", "") + return + } + + rc := &relay_handler.TaskRelayContext{ + TenantID: middleware.GetTenantID(r.Context()), + UserID: middleware.GetUserID(r.Context()), + ApiKeyID: middleware.GetApiKeyID(r.Context()), + ProjectID: middleware.GetProjectID(r.Context()), + RequestID: r.GetCtxVar("RequestId").String(), + Writer: r.Response.Writer, + } + + relay_handler.HandleVideosRetrieve(r.Context(), videoID, rc, taskDataProvider) +} + +// HandleVideoDelete 处理 DELETE /v1/videos/{video_id}(删除视频任务) +func HandleVideoDelete(r *ghttp.Request) { + videoID := r.Get("video_id").String() + if videoID == "" { + writeVideosError(r, 400, "video id is required", "") + return + } + + rc := &relay_handler.TaskRelayContext{ + TenantID: middleware.GetTenantID(r.Context()), + UserID: middleware.GetUserID(r.Context()), + ApiKeyID: middleware.GetApiKeyID(r.Context()), + ProjectID: middleware.GetProjectID(r.Context()), + RequestID: r.GetCtxVar("RequestId").String(), + Writer: r.Response.Writer, + } + + relay_handler.HandleVideosDelete(r.Context(), videoID, rc, taskDataProvider) +} + +// HandleVideoContent 处理 GET /v1/videos/{video_id}/content(下载成品视频) +func HandleVideoContent(r *ghttp.Request) { + videoID := r.Get("video_id").String() + if videoID == "" { + writeVideosError(r, 400, "video id is required", "") + return + } + + rc := &relay_handler.TaskRelayContext{ + TenantID: middleware.GetTenantID(r.Context()), + UserID: middleware.GetUserID(r.Context()), + ApiKeyID: middleware.GetApiKeyID(r.Context()), + ProjectID: middleware.GetProjectID(r.Context()), + RequestID: r.GetCtxVar("RequestId").String(), + Writer: r.Response.Writer, + } + + relay_handler.HandleVideosContent(r.Context(), videoID, r.Get("variant", "video").String(), rc, taskDataProvider) +} + +// writeVideosError OpenAI 官方错误格式输出(与 writeTaskError 同构,供胶水层解析阶段使用) +func writeVideosError(r *ghttp.Request, statusCode int, message, code string) { + errObj := g.Map{ + "type": "invalid_request_error", + "message": message, + } + if code != "" { + errObj["code"] = code + } + r.Response.WriteStatus(statusCode, g.Map{"error": errObj}) +} diff --git a/internal/logic/docs/openapi.go b/internal/logic/docs/openapi.go index f8c92a8..aec9279 100644 --- a/internal/logic/docs/openapi.go +++ b/internal/logic/docs/openapi.go @@ -202,6 +202,115 @@ func buildOpenAPISpec(ctx context.Context) map[string]any { }, }, }, + "/v1/videos": map[string]any{ + "post": map[string]any{ + "summary": "Create Video", + "description": "创建视频生成任务(OpenAI Videos 协议,兼容 openai-python / openai-node SDK 的 client.videos.create)。请求体支持 multipart/form-data(官方 SDK 形态,input_reference 可为文件 part)与 application/json 两种编码。返回官方 Video 对象(status: queued),轮询 GET /v1/videos/{video_id} 直到 completed,再经 GET /v1/videos/{video_id}/content 获取视频二进制。", + "operationId": "createVideo", + "security": []map[string]any{{"BearerAuth": []any{}}}, + "requestBody": map[string]any{ + "required": true, + "content": map[string]any{ + "multipart/form-data": map[string]any{ + "schema": map[string]any{ + "type": "object", + "required": []string{"model", "prompt"}, + "properties": map[string]any{ + "model": map[string]any{"type": "string", "example": "kling-v2-master"}, + "prompt": map[string]any{"type": "string"}, + "seconds": map[string]any{"type": "string", "enum": []string{"4", "8", "12"}, "default": "4"}, + "size": map[string]any{"type": "string", "enum": []string{"720x1280", "1280x720", "1024x1792", "1792x1024"}, "default": "720x1280"}, + "input_reference": map[string]any{"type": "string", "format": "binary", "description": "可选参考图(图生视频),文件上传或 {\"image_url\": \"...\"} 引用对象;file_id 形态不支持"}, + }, + }, + }, + "application/json": map[string]any{ + "schema": map[string]any{ + "type": "object", + "required": []string{"model", "prompt"}, + "properties": map[string]any{ + "model": map[string]any{"type": "string"}, + "prompt": map[string]any{"type": "string"}, + "seconds": map[string]any{"type": "string"}, + "size": map[string]any{"type": "string"}, + "input_reference": map[string]any{"type": "object", "properties": map[string]any{"image_url": map[string]any{"type": "string"}}}, + }, + }, + }, + }, + }, + "responses": map[string]any{ + "200": map[string]any{"description": "Video 对象(object: video,status: queued)"}, + }, + }, + }, + "/v1/videos/{video_id}": map[string]any{ + "get": map[string]any{ + "summary": "Retrieve Video", + "description": "查询视频生成任务,返回官方 Video 对象。status: queued / in_progress / completed / failed;非终态响应携带 openai-poll-after-ms 轮询间隔建议头。生成失败以 status=failed + error 对象表达(HTTP 仍为 200)。", + "operationId": "retrieveVideo", + "security": []map[string]any{{"BearerAuth": []any{}}}, + "parameters": []map[string]any{ + { + "name": "video_id", + "in": "path", + "required": true, + "schema": map[string]any{"type": "string"}, + }, + }, + "responses": map[string]any{ + "200": map[string]any{"description": "Video 对象(含 progress 百分比、completed_at、失败时的 error)"}, + "404": map[string]any{"description": "视频不存在或已被删除"}, + }, + }, + "delete": map[string]any{ + "summary": "Delete Video", + "description": "删除已完成或已失败的视频任务(软删除,计费与审计记录保留),删除后查询与下载均返回 404。进行中的任务不可删除。", + "operationId": "deleteVideo", + "security": []map[string]any{{"BearerAuth": []any{}}}, + "parameters": []map[string]any{ + { + "name": "video_id", + "in": "path", + "required": true, + "schema": map[string]any{"type": "string"}, + }, + }, + "responses": map[string]any{ + "200": map[string]any{"description": "删除确认(object: video.deleted,deleted: true)"}, + "400": map[string]any{"description": "任务非终态或计费未结算,不可删除"}, + "404": map[string]any{"description": "视频不存在或已被删除"}, + }, + }, + }, + "/v1/videos/{video_id}/content": map[string]any{ + "get": map[string]any{ + "summary": "Download Video Content", + "description": "下载已生成完成的视频二进制流(mp4)。仅 status=completed 的任务可下载;variant 仅支持默认 video(thumbnail / spritesheet 暂不支持)。", + "operationId": "downloadVideoContent", + "security": []map[string]any{{"BearerAuth": []any{}}}, + "parameters": []map[string]any{ + { + "name": "video_id", + "in": "path", + "required": true, + "schema": map[string]any{"type": "string"}, + }, + { + "name": "variant", + "in": "query", + "required": false, + "schema": map[string]any{"type": "string", "enum": []string{"video"}, "default": "video"}, + }, + }, + "responses": map[string]any{ + "200": map[string]any{"description": "视频二进制流(video/mp4)"}, + "400": map[string]any{"description": "任务未完成或无可用内容"}, + "404": map[string]any{"description": "视频不存在或已被删除"}, + "502": map[string]any{"description": "上游内容拉取失败"}, + }, + }, + }, "/v1/messages": map[string]any{ "post": map[string]any{ "summary": "Claude Messages", diff --git a/internal/logic/task/async_provider.go b/internal/logic/task/async_provider.go index 1b584a9..7183dba 100644 --- a/internal/logic/task/async_provider.go +++ b/internal/logic/task/async_provider.go @@ -53,6 +53,8 @@ func asyncTaskFromEntity(row *entity.TskModelTasks) *common.AsyncTask { FinishTime: timePtrFromGTime(row.FinishTime), CreatedAt: timeFromGTime(row.CreatedAt), UpdatedAt: timeFromGTime(row.UpdatedAt), + Deleted: row.Deleted, + DeletedAt: timePtrFromGTime(row.DeletedAt), } } @@ -187,11 +189,12 @@ func (p *AsyncProvider) UpdateTaskCAS(ctx context.Context, task *common.AsyncTas return nil } -// GetTaskByPublicID 根据公开任务 ID 查询 +// GetTaskByPublicID 根据公开任务 ID 查询(软删除的任务视为不存在) func (p *AsyncProvider) GetTaskByPublicID(ctx context.Context, publicTaskID string) (*common.AsyncTask, error) { var row *entity.TskModelTasks err := dao.TskModelTasks.Ctx(ctx). Where("public_task_id", publicTaskID). + Where("deleted", false). Scan(&row) if err != nil { return nil, gerror.Wrapf(err, "query async task failed: public_id=%s", publicTaskID) @@ -199,13 +202,14 @@ func (p *AsyncProvider) GetTaskByPublicID(ctx context.Context, publicTaskID stri return asyncTaskFromEntity(row), nil } -// GetTaskByPublicIDAndUser 根据公开任务 ID + 用户 ID 查询 +// GetTaskByPublicIDAndUser 根据公开任务 ID + 用户 ID 查询(软删除的任务视为不存在) func (p *AsyncProvider) GetTaskByPublicIDAndUser(ctx context.Context, publicTaskID string, userID int64, tenantID int64) (*common.AsyncTask, error) { var row *entity.TskModelTasks err := dao.TskModelTasks.Ctx(ctx). Where("public_task_id", publicTaskID). Where("user_id", userID). Where("tenant_id", tenantID). + Where("deleted", false). Scan(&row) if err != nil { return nil, gerror.Wrapf(err, "query async task failed: public_id=%s user_id=%d", publicTaskID, userID) @@ -262,6 +266,30 @@ func (p *AsyncProvider) GetUnsettledTasks(ctx context.Context, limit int) ([]*co return asyncTasksFromEntities(rows), nil } +// SoftDeleteTask 软删除任务(DELETE /v1/videos/{id})。 +// 仅标记 deleted,不动 status/计费字段:调用方保证只有终态且已结算的任务可删, +// 轮询 cron 的 GetNonTerminalTasks/GetUnsettledTasks 查询不感知 deleted 列。 +func (p *AsyncProvider) SoftDeleteTask(ctx context.Context, task *common.AsyncTask) error { + result, err := dao.TskModelTasks.Ctx(ctx). + Where("id", task.ID). + Where("tenant_id", task.TenantID). + Update(map[string]any{ + "deleted": true, + "deleted_at": gtime.Now(), + "result_url": "", // 清掉成品直链,避免软删后内部路径仍可寻址 + "updated_at": time.Now(), + }) + if err != nil { + return gerror.Wrapf(err, "soft delete async task failed: id=%d", task.ID) + } + rows, _ := result.RowsAffected() + if rows == 0 { + return fmt.Errorf("soft delete no rows affected: id=%d", task.ID) + } + g.Log().Infof(ctx, "[AsyncProvider] SoftDeleteTask: public_id=%s, tenant=%d", task.PublicTaskID, task.TenantID) + return nil +} + // GetChannelByID 获取渠道基本信息(含从 chn_channel_keys 解密的 API Key) func (p *AsyncProvider) GetChannelByID(ctx context.Context, channelID int64) (*common.ChannelBasicInfo, error) { var row *struct { diff --git a/internal/model/do/sys_admin_role_perms.go b/internal/model/do/sys_admin_role_perms.go index 9fb1ad3..865d548 100644 --- a/internal/model/do/sys_admin_role_perms.go +++ b/internal/model/do/sys_admin_role_perms.go @@ -14,7 +14,7 @@ type SysAdminRolePerms struct { g.Meta `orm:"table:sys_admin_role_perms, do:true"` Id any // 主键ID AdminUserId any // 关联的管理员用户ID - PermissionPoint any // 权限点标识(如 tenant:create、channel:edit) + PermissionPoint any // 特批的权限点标识(独立于角色,角色禁用不影响特批) CreatedAt *gtime.Time // 创建时间 UpdatedAt *gtime.Time // 更新时间 } diff --git a/internal/model/do/tsk_model_tasks.go b/internal/model/do/tsk_model_tasks.go index 0766f75..3542625 100644 --- a/internal/model/do/tsk_model_tasks.go +++ b/internal/model/do/tsk_model_tasks.go @@ -37,4 +37,6 @@ type TskModelTasks struct { CreatedAt *gtime.Time // 记录创建时间 UpdatedAt *gtime.Time // 记录更新时间 RequestId any // 任务提交时的原始请求 ID(req_xxxxx),关联 aud_request_logs.request_id + Deleted any // 软删除标记:true 表示已被客户端删除,查询/下载按不存在处理(计费与审计记录保留) + DeletedAt *gtime.Time // 软删除时间 } diff --git a/internal/model/entity/sys_admin_role_perms.go b/internal/model/entity/sys_admin_role_perms.go index ed4cda1..ebbb197 100644 --- a/internal/model/entity/sys_admin_role_perms.go +++ b/internal/model/entity/sys_admin_role_perms.go @@ -10,9 +10,9 @@ import ( // SysAdminRolePerms is the golang structure for table sys_admin_role_perms. type SysAdminRolePerms struct { - Id int64 `json:"id" orm:"id" description:"主键ID"` // 主键ID - AdminUserId int64 `json:"admin_user_id" orm:"admin_user_id" description:"关联的管理员用户ID"` // 关联的管理员用户ID - PermissionPoint string `json:"permission_point" orm:"permission_point" description:"权限点标识(如 tenant:create、channel:edit)"` // 权限点标识(如 tenant:create、channel:edit) - CreatedAt *gtime.Time `json:"created_at" orm:"created_at" description:"创建时间"` // 创建时间 - UpdatedAt *gtime.Time `json:"updated_at" orm:"updated_at" description:"更新时间"` // 更新时间 + Id int64 `json:"id" orm:"id" description:"主键ID"` // 主键ID + AdminUserId int64 `json:"admin_user_id" orm:"admin_user_id" description:"关联的管理员用户ID"` // 关联的管理员用户ID + PermissionPoint string `json:"permission_point" orm:"permission_point" description:"特批的权限点标识(独立于角色,角色禁用不影响特批)"` // 特批的权限点标识(独立于角色,角色禁用不影响特批) + CreatedAt *gtime.Time `json:"created_at" orm:"created_at" description:"创建时间"` // 创建时间 + UpdatedAt *gtime.Time `json:"updated_at" orm:"updated_at" description:"更新时间"` // 更新时间 } diff --git a/internal/model/entity/tsk_model_tasks.go b/internal/model/entity/tsk_model_tasks.go index 2a51b6d..2a5c7f4 100644 --- a/internal/model/entity/tsk_model_tasks.go +++ b/internal/model/entity/tsk_model_tasks.go @@ -36,4 +36,6 @@ type TskModelTasks struct { CreatedAt *gtime.Time `json:"created_at" orm:"created_at" description:"记录创建时间"` // 记录创建时间 UpdatedAt *gtime.Time `json:"updated_at" orm:"updated_at" description:"记录更新时间"` // 记录更新时间 RequestId string `json:"request_id" orm:"request_id" description:"任务提交时的原始请求 ID(req_xxxxx),关联 aud_request_logs.request_id"` // 任务提交时的原始请求 ID(req_xxxxx),关联 aud_request_logs.request_id + Deleted bool `json:"deleted" orm:"deleted" description:"软删除标记:true 表示已被客户端删除,查询/下载按不存在处理(计费与审计记录保留)"` // 软删除标记:true 表示已被客户端删除,查询/下载按不存在处理(计费与审计记录保留) + DeletedAt *gtime.Time `json:"deleted_at" orm:"deleted_at" description:"软删除时间"` // 软删除时间 } diff --git a/relay/common/task_provider.go b/relay/common/task_provider.go index 29c7770..405277f 100644 --- a/relay/common/task_provider.go +++ b/relay/common/task_provider.go @@ -25,6 +25,10 @@ type TaskDataProvider interface { // GetTaskByPublicIDAndUser 根据公开任务 ID + 用户 ID 查询(客户端查询) GetTaskByPublicIDAndUser(ctx context.Context, publicTaskID string, userID int64, tenantID int64) (*AsyncTask, error) + // SoftDeleteTask 软删除任务(DELETE /v1/videos/{id}):置 deleted 标记, + // 计费与审计记录保留,后续客户端查询按不存在处理 + SoftDeleteTask(ctx context.Context, task *AsyncTask) error + // GetNonTerminalTasks 获取所有非终态任务(轮询用) GetNonTerminalTasks(ctx context.Context, limit int) ([]*AsyncTask, error) @@ -74,6 +78,10 @@ type AsyncTask struct { FinishTime *time.Time `json:"finish_time,omitempty"` CreatedAt time.Time `json:"created_at"` UpdatedAt time.Time `json:"updated_at"` + + // 软删除标记(DELETE /v1/videos/{id}):true 后客户端查询/下载按 404 处理 + Deleted bool `json:"deleted,omitempty"` + DeletedAt *time.Time `json:"deleted_at,omitempty"` } // ChannelBasicInfo 渠道基本信息(轮询时查询) diff --git a/relay/constant/relay_mode.go b/relay/constant/relay_mode.go index 46e1ee4..0a8493e 100644 --- a/relay/constant/relay_mode.go +++ b/relay/constant/relay_mode.go @@ -31,6 +31,7 @@ const ( RelayModeMjFetch // GET /mj/task/:id/fetch RelayModeMjImage // GET /mj/image/:id RelayModeResponsesCompact // POST /v1/responses/compact + RelayModeVideos // OpenAI Videos 协议 /v1/videos(创建/查询/删除/内容下载) ) // Path2RelayMode 根据请求路径判断 RelayMode @@ -101,6 +102,8 @@ func Path2RelayMode(path string) RelayMode { result = RelayModeVideoGenerations case strings.Contains(path, "/video/generations/"): result = RelayModeVideoFetch + case strings.HasSuffix(path, "/videos"), strings.Contains(path, "/videos/"): + result = RelayModeVideos case strings.HasSuffix(path, "/moderations"): result = RelayModeModerations default: @@ -151,6 +154,8 @@ func (m RelayMode) String() string { return "moderations" case RelayModeImagesEdits: return "images_edits" + case RelayModeVideos: + return "videos" case RelayModeMjSubmit: return "mj_submit" case RelayModeMjFetch: diff --git a/relay/handler/task_handler.go b/relay/handler/task_handler.go index ae180ff..ff24a66 100644 --- a/relay/handler/task_handler.go +++ b/relay/handler/task_handler.go @@ -44,6 +44,16 @@ type TaskRelayContext struct { KeyTotalQuota float64 KeyUsedQuota float64 ForwardingTrace *common.ForwardingTrace + + // Protocol 入站协议标识:"" 默认(legacy 任务响应格式,task_ 前缀 ID); + // videosProtocolOpenAI 切换为 OpenAI Videos 协议(官方 Video 对象响应,video_ 前缀 ID)。 + Protocol string + // RelayMode 任务 relay 模式:0 = 默认 RelayModeVideoGenerations(现有端点), + // OpenAI Videos 入站传 RelayModeVideos 以区分审计与监控统计。 + RelayMode int + // RequestEcho 协议层请求回显(OpenAI Videos 的 prompt/seconds/size), + // 提交时随 PrivateData 落库,retrieve 时回显给客户端;nil = 无回显。 + RequestEcho any } func checkTaskIPWhitelist(whitelist string, clientIP string) bool { @@ -154,13 +164,17 @@ func HandleTaskSubmit( g.Log().Debugf(ctx, "HandleTaskSubmit: modelName=%s, platform=%s, channelID=%d, channelType=%d, baseURL=%s, upstreamModel=%s", modelName, platform, channelMeta.ChannelID, channelMeta.ChannelType, channelMeta.BaseURL, channelMeta.UpstreamModelName) // 4. 构建 RelayInfo + relayMode := rc.RelayMode + if relayMode == 0 { + relayMode = int(constant.RelayModeVideoGenerations) + } info := &common.RelayInfo{ Context: ctx, TenantID: rc.TenantID, UserID: rc.UserID, ApiKeyID: rc.ApiKeyID, RequestID: rc.RequestID, - RelayMode: int(constant.RelayModeVideoGenerations), + RelayMode: relayMode, OriginModelName: modelName, RequestURLPath: path, RequestHeaders: headers, @@ -259,11 +273,14 @@ func HandleTaskSubmit( } } - // 10. 生成公开任务 ID 并创建记录 + // 10. 生成公开任务 ID 并创建记录(OpenAI Videos 协议用 video_ 前缀对齐官方形态) publicTaskID := generatePublicTaskID() + if rc.Protocol == videosProtocolOpenAI { + publicTaskID = generatePublicTaskIDWithPrefix("video") + } now := time.Now() - privateData, _ := json.Marshal(map[string]any{ + privateDataMap := map[string]any{ "upstream_task_id": upstreamTaskID, "task_type": platform, "billing_context": map[string]any{ @@ -274,7 +291,11 @@ func HandleTaskSubmit( // 进而 Ratios 也读不出、轮询判为「invalid private data」。故此处显式转 float64。 "pre_deduct": preDeductAmount.InexactFloat64(), }, - }) + } + if rc.RequestEcho != nil { + privateDataMap["request_echo"] = rc.RequestEcho + } + privateData, _ := json.Marshal(privateDataMap) task := &common.AsyncTask{ PublicTaskID: publicTaskID, @@ -307,7 +328,11 @@ func HandleTaskSubmit( // 设置 TaskID 供外层审计使用 rc.TaskID = publicTaskID - // 11. 返回响应 + // 11. 返回响应(OpenAI Videos 协议返回官方 Video 对象,legacy 保持原格式) + if rc.Protocol == videosProtocolOpenAI { + writeVideosSubmitResponse(rc.Writer, publicTaskID, modelName, now, rc.RequestEcho) + return + } respBody := map[string]any{ "id": publicTaskID, "status": "SUBMITTED", @@ -410,6 +435,11 @@ func generatePublicTaskID() string { return fmt.Sprintf("task_%s", randomHex(32)) } +// generatePublicTaskIDWithPrefix 生成带指定前缀的公开任务 ID(如 OpenAI Videos 协议的 video_ 前缀) +func generatePublicTaskIDWithPrefix(prefix string) string { + return fmt.Sprintf("%s_%s", prefix, randomHex(32)) +} + func randomHex(n int) string { b := make([]byte, (n+1)/2) if _, err := rand.Read(b); err != nil { diff --git a/relay/handler/videos.go b/relay/handler/videos.go new file mode 100644 index 0000000..a221022 --- /dev/null +++ b/relay/handler/videos.go @@ -0,0 +1,305 @@ +package handler + +import ( + "context" + "encoding/json" + "io" + "net/http" + "net/url" + "strconv" + "strings" + "time" + + "github.com/gogf/gf/v2/frame/g" + + "github.com/qianfree/team-api/relay/common" +) + +// OpenAI Videos 协议(/v1/videos)核心处理:Video 对象构建、retrieve / delete / content。 +// 协议规格见 docs/协议文档/OpenAI/OpenAI-Videos-API文档.md。 +// 注意状态机只有 queued/in_progress/completed/failed 四值(无 cancelled), +// 错误分两层:请求层用 HTTP 错误(writeTaskError,OpenAI 格式), +// 生成层失败用 200 + status:"failed" + 内嵌 error 对象表达。 + +// videosProtocolOpenAI TaskRelayContext.Protocol 的 OpenAI Videos 协议标识 +const videosProtocolOpenAI = "openai_videos" + +// videosPollAfterMs 非终态任务 retrieve 响应头 openai-poll-after-ms 的建议轮询间隔(毫秒)。 +// 官方 SDK poll() 优先读取该响应头,缺省时 SDK 自取 1000ms;视频任务耗时长,放慢到 2s。 +const videosPollAfterMs = "2000" + +// videosContentHTTPClient 成品内容回源拉流的 HTTP 客户端(大文件流式转发,整体超时兜底) +var videosContentHTTPClient = &http.Client{Timeout: 120 * time.Second} + +// videoErrorObject 官方 VideoCreateError 的简化形态(misalignment 为 Sora 内容安全特有,第三方上游无对应,不透出) +type videoErrorObject struct { + Code string `json:"code"` + Message string `json:"message"` +} + +// videoObject 官方 Video 对象。可空字段按官方 schema 输出 null(非省略)。 +type videoObject struct { + ID string `json:"id"` + Object string `json:"object"` // 恒为 "video" + CreatedAt int64 `json:"created_at"` + CompletedAt *int64 `json:"completed_at"` + Status string `json:"status"` // queued / in_progress / completed / failed + Progress int `json:"progress"` + Model string `json:"model"` + Prompt *string `json:"prompt"` + Seconds string `json:"seconds,omitempty"` + Size string `json:"size,omitempty"` + Error *videoErrorObject `json:"error"` + RemixedFromVideoID *string `json:"remixed_from_video_id"` + ExpiresAt *int64 `json:"expires_at"` +} + +// videosStatusFromTask 平台任务状态 → 官方四值状态机映射 +func videosStatusFromTask(status string) string { + switch status { + case "IN_PROGRESS": + return "in_progress" + case "SUCCESS": + return "completed" + case "FAILURE": + return "failed" + default: + // NOT_START / SUBMITTED / QUEUED 及一切未知中间态都归入 queued + return "queued" + } +} + +// videosProgressFromTask 平台进度("42%" 字符串)→ 官方整数百分比。 +// 解析失败时按状态给粗粒度值(第三方上游无进度信息时的兜底)。 +func videosProgressFromTask(task *common.AsyncTask) int { + if n, err := strconv.Atoi(strings.TrimSpace(strings.TrimSuffix(task.Progress, "%"))); err == nil { + if n < 0 { + return 0 + } + if n > 100 { + return 100 + } + return n + } + switch videosStatusFromTask(task.Status) { + case "completed": + return 100 + case "in_progress": + return 50 + default: + return 0 + } +} + +// videosEchoFromTask 从 PrivateData.request_echo 提取提交时的请求回显 +func videosEchoFromTask(task *common.AsyncTask) videosRequestEcho { + var echo videosRequestEcho + if len(task.PrivateData) == 0 { + return echo + } + var pd struct { + RequestEcho *videosRequestEcho `json:"request_echo"` + } + if err := json.Unmarshal(task.PrivateData, &pd); err == nil && pd.RequestEcho != nil { + echo = *pd.RequestEcho + } + return echo +} + +// buildVideoObject 平台任务 → 官方 Video 对象 +func buildVideoObject(task *common.AsyncTask) *videoObject { + obj := &videoObject{ + ID: task.PublicTaskID, + Object: "video", + CreatedAt: task.CreatedAt.Unix(), + Status: videosStatusFromTask(task.Status), + Progress: videosProgressFromTask(task), + Model: task.ModelName, + } + + echo := videosEchoFromTask(task) + prompt := echo.Prompt + obj.Prompt = &prompt + obj.Seconds = echo.Seconds + obj.Size = echo.Size + + if task.FinishTime != nil { + completedAt := task.FinishTime.Unix() + obj.CompletedAt = &completedAt + } + if obj.Status == "failed" { + message := task.FailReason + if message == "" { + message = "video generation failed" + } + obj.Error = &videoErrorObject{Code: "video_generation_failed", Message: message} + } + return obj +} + +// writeVideosSubmitResponse 提交成功响应:官方 Video 对象(初始 queued / progress 0) +func writeVideosSubmitResponse(w http.ResponseWriter, publicTaskID, modelName string, now time.Time, echo any) { + obj := &videoObject{ + ID: publicTaskID, + Object: "video", + CreatedAt: now.Unix(), + Status: "queued", + Progress: 0, + Model: modelName, + } + if echo != nil { + if e, ok := echo.(videosRequestEcho); ok { + prompt := e.Prompt + obj.Prompt = &prompt + obj.Seconds = e.Seconds + obj.Size = e.Size + } + } + writeJSON(w, http.StatusOK, obj) +} + +// HandleVideosRetrieve GET /v1/videos/{video_id} 查询管线。 +// 租户双键隔离(user_id + tenant_id)复用 GetTaskByPublicIDAndUser;软删除任务已被过滤,天然 404。 +func HandleVideosRetrieve( + ctx context.Context, + videoID string, + rc *TaskRelayContext, + dataProvider common.TaskDataProvider, +) { + task, err := dataProvider.GetTaskByPublicIDAndUser(ctx, videoID, rc.UserID, rc.TenantID) + if err != nil { + g.Log().Errorf(ctx, "HandleVideosRetrieve: query task failed, videoID=%s, err=%v", videoID, err) + writeTaskError(rc.Writer, http.StatusInternalServerError, "query task failed", "") + return + } + if task == nil { + writeTaskError(rc.Writer, http.StatusNotFound, "No video found with id '"+videoID+"'.", "") + return + } + + obj := buildVideoObject(task) + // 非终态附带官方轮询提示头(SDK poll() 优先读取) + if obj.Status == "queued" || obj.Status == "in_progress" { + rc.Writer.Header().Set("openai-poll-after-ms", videosPollAfterMs) + } + writeJSON(rc.Writer, http.StatusOK, obj) +} + +// HandleVideosDelete DELETE /v1/videos/{video_id} 删除管线。 +// 对齐官方语义「永久删除已完成/失败的视频及其资产」:仅终态可删; +// 网关侧软删除保留计费与审计记录,删除后查询/下载均按 404 处理(幂等)。 +func HandleVideosDelete( + ctx context.Context, + videoID string, + rc *TaskRelayContext, + dataProvider common.TaskDataProvider, +) { + task, err := dataProvider.GetTaskByPublicIDAndUser(ctx, videoID, rc.UserID, rc.TenantID) + if err != nil { + g.Log().Errorf(ctx, "HandleVideosDelete: query task failed, videoID=%s, err=%v", videoID, err) + writeTaskError(rc.Writer, http.StatusInternalServerError, "query task failed", "") + return + } + if task == nil { + writeTaskError(rc.Writer, http.StatusNotFound, "No video found with id '"+videoID+"'.", "") + return + } + + // 仅终态(completed/failed)可删,进行中任务官方亦不允许删除 + if videosStatusFromTask(task.Status) != "completed" && videosStatusFromTask(task.Status) != "failed" { + writeTaskError(rc.Writer, http.StatusBadRequest, "only completed or failed videos can be deleted", "video_not_deletable") + return + } + // 未结算的终态任务删除会造成退款悬空:结算由轮询 cron 兜底完成后即可删除 + if !task.BillingSettled { + writeTaskError(rc.Writer, http.StatusBadRequest, "video billing is not settled yet, retry later", "billing_pending") + return + } + + if err := dataProvider.SoftDeleteTask(ctx, task); err != nil { + g.Log().Errorf(ctx, "HandleVideosDelete: soft delete failed, videoID=%s, err=%v", videoID, err) + writeTaskError(rc.Writer, http.StatusInternalServerError, "delete video failed", "") + return + } + g.Log().Infof(ctx, "HandleVideosDelete: video deleted, videoID=%s, tenant=%d, user=%d", videoID, rc.TenantID, rc.UserID) + + writeJSON(rc.Writer, http.StatusOK, map[string]any{ + "id": task.PublicTaskID, + "object": "video.deleted", + "deleted": true, + }) +} + +// HandleVideosContent GET /v1/videos/{video_id}/content 成品下载管线。 +// 从任务 ResultURL 回源拉流并转发二进制(不透出上游 CDN 直链)。 +// 仅支持官方默认 variant=video;thumbnail/spritesheet 第三方上游无此资产。 +func HandleVideosContent( + ctx context.Context, + videoID string, + variant string, + rc *TaskRelayContext, + dataProvider common.TaskDataProvider, +) { + if variant != "" && variant != "video" { + writeTaskError(rc.Writer, http.StatusBadRequest, "unsupported content variant: "+variant, "unsupported_variant") + return + } + + task, err := dataProvider.GetTaskByPublicIDAndUser(ctx, videoID, rc.UserID, rc.TenantID) + if err != nil { + g.Log().Errorf(ctx, "HandleVideosContent: query task failed, videoID=%s, err=%v", videoID, err) + writeTaskError(rc.Writer, http.StatusInternalServerError, "query task failed", "") + return + } + if task == nil { + writeTaskError(rc.Writer, http.StatusNotFound, "No video found with id '"+videoID+"'.", "") + return + } + + // 官方语义:仅 completed 可下载(未完成/失败均报错) + if task.Status != "SUCCESS" { + writeTaskError(rc.Writer, http.StatusBadRequest, "video content is not available until the video is completed", "video_not_ready") + return + } + // 部分上游(如官方 sora)不提供成品直链,需要二跳 content 拉取——当前不支持,明确报错 + if task.ResultURL == "" { + writeTaskError(rc.Writer, http.StatusBadRequest, "video content is not available for this task", "content_not_available") + return + } + + // ResultURL 来自上游响应持久化数据,仍做 scheme 白名单防异构值 + resultURL, err := url.Parse(task.ResultURL) + if err != nil || (resultURL.Scheme != "http" && resultURL.Scheme != "https") { + g.Log().Errorf(ctx, "HandleVideosContent: invalid result url, videoID=%s, url=%q, err=%v", videoID, task.ResultURL, err) + writeTaskError(rc.Writer, http.StatusInternalServerError, "invalid video result url", "") + return + } + + resp, err := videosContentHTTPClient.Get(task.ResultURL) + if err != nil { + // 上游拉流失败属预期内运营事件,用 Warningf 避免自动打栈(对齐任务提交链路日志分级) + g.Log().Warningf(ctx, "HandleVideosContent: fetch upstream failed, videoID=%s, err=%v", videoID, err) + writeTaskError(rc.Writer, http.StatusBadGateway, "failed to fetch video content", "") + return + } + defer resp.Body.Close() + + if resp.StatusCode != http.StatusOK { + g.Log().Warningf(ctx, "HandleVideosContent: upstream status=%d, videoID=%s", resp.StatusCode, videoID) + writeTaskError(rc.Writer, http.StatusBadGateway, "failed to fetch video content", "") + return + } + + // 透传内容元信息后流式转发字节 + if ct := resp.Header.Get("Content-Type"); ct != "" { + rc.Writer.Header().Set("Content-Type", ct) + } + if cl := resp.Header.Get("Content-Length"); cl != "" { + rc.Writer.Header().Set("Content-Length", cl) + } + rc.Writer.WriteHeader(http.StatusOK) + if _, err := io.Copy(rc.Writer, resp.Body); err != nil { + // 响应头已发出,只能记录日志,无法再改写为错误响应 + g.Log().Warningf(ctx, "HandleVideosContent: stream copy failed, videoID=%s, err=%v", videoID, err) + } +} diff --git a/relay/handler/videos_convert.go b/relay/handler/videos_convert.go new file mode 100644 index 0000000..5200f83 --- /dev/null +++ b/relay/handler/videos_convert.go @@ -0,0 +1,290 @@ +package handler + +import ( + "bytes" + "encoding/base64" + "encoding/json" + "fmt" + "io" + "mime" + "mime/multipart" + "net/http" + "strconv" + "strings" + + "github.com/qianfree/team-api/relay/common" +) + +// OpenAI Videos 协议(/v1/videos)入站转换层。 +// 官方 SDK(openai-python / openai-node)对 POST /v1/videos 一律发送 multipart/form-data, +// 裸 HTTP 客户端常用 JSON——两种编码都要支持;解析后统一转成任务框架的通用请求体 +// {model, prompt, seconds, metadata{duration, aspect_ratio, resolution, image}}, +// 由各 taskchannel 适配器按需消费(kling 读顶层 seconds,volcengine 读 metadata.duration 等)。 + +// videosInputReferenceMaxBytes 参考图上传大小上限。 +// GoFrame server 默认 ClientMaxBodySize 8MB,base64 膨胀 4/3 后须留余量,取 6MB。 +const videosInputReferenceMaxBytes = 6 << 20 + +// videosCreateRequest OpenAI Videos 创建请求的归一形态(multipart/JSON 解析后的产物) +type videosCreateRequest struct { + Model string + Prompt string + Seconds string // 统一归一为字符串(官方协议为 "4"/"8"/"12");空 = 未指定 + Size string // 原样保留(如 "1280x720");空 = 未指定 + InputReference string // 参考图:data URL 或 http(s) URL;空 = 无 +} + +// videosRequestEcho 提交时落 PrivateData.request_echo 的请求回显, +// retrieve(GET /v1/videos/{id})时用于回显 prompt/seconds/size(上游响应不含这些字段)。 +type videosRequestEcho struct { + Prompt string `json:"prompt"` + Seconds string `json:"seconds,omitempty"` + Size string `json:"size,omitempty"` +} + +// jsonVideosCreateRequest JSON 编码的创建请求结构。 +// input_reference 保留 RawMessage 以区分 {file_id} / {image_url} 两种引用形态。 +type jsonVideosCreateRequest struct { + Model string `json:"model"` + Prompt string `json:"prompt"` + Seconds json.RawMessage `json:"seconds"` // string 或 number + Size string `json:"size"` + InputReference json.RawMessage `json:"input_reference"` +} + +// ParseVideosCreateRequest 解析 OpenAI Videos 创建请求(multipart 或 JSON)。 +// 返回的 TaskError 已携带面向客户端的 HTTP 状态码与错误信息。 +func ParseVideosCreateRequest(body []byte, contentType string) (*videosCreateRequest, *common.TaskError) { + mediaType, params, err := mime.ParseMediaType(contentType) + if err != nil { + mediaType = contentType // 非法 Content-Type 按 JSON 尝试解析 + } + + if strings.HasPrefix(mediaType, "multipart/") { + return parseVideosCreateMultipart(body, params["boundary"]) + } + return parseVideosCreateJSON(body) +} + +// parseVideosCreateJSON 解析 JSON 编码的创建请求 +func parseVideosCreateJSON(body []byte) (*videosCreateRequest, *common.TaskError) { + var req jsonVideosCreateRequest + if err := json.Unmarshal(body, &req); err != nil { + return nil, &common.TaskError{StatusCode: http.StatusBadRequest, Message: "invalid request body: " + err.Error()} + } + + out := &videosCreateRequest{ + Model: strings.TrimSpace(req.Model), + Prompt: req.Prompt, + Size: strings.TrimSpace(req.Size), + } + + seconds, terr := normalizeVideosSeconds(req.Seconds) + if terr != nil { + return nil, terr + } + out.Seconds = seconds + + ref, terr := parseVideosInputReferenceJSON(req.InputReference) + if terr != nil { + return nil, terr + } + out.InputReference = ref + + return out, nil +} + +// parseVideosCreateMultipart 解析 multipart/form-data 编码的创建请求(官方 SDK 形态)。 +// input_reference 可能是文件 part(上传),也可能是字符串 part(引用对象 JSON)。 +func parseVideosCreateMultipart(body []byte, boundary string) (*videosCreateRequest, *common.TaskError) { + if boundary == "" { + return nil, &common.TaskError{StatusCode: http.StatusBadRequest, Message: "missing multipart boundary"} + } + + form, err := multipart.NewReader(bytes.NewReader(body), boundary).ReadForm(videosInputReferenceMaxBytes) + if err != nil { + return nil, &common.TaskError{StatusCode: http.StatusBadRequest, Message: "invalid multipart form: " + err.Error()} + } + defer form.RemoveAll() + + formValue := func(name string) string { + if vals := form.Value[name]; len(vals) > 0 { + return vals[0] + } + return "" + } + + out := &videosCreateRequest{ + Model: strings.TrimSpace(formValue("model")), + Prompt: formValue("prompt"), + Size: strings.TrimSpace(formValue("size")), + } + + seconds, terr := normalizeVideosSeconds(json.RawMessage(formValue("seconds"))) + if terr != nil { + return nil, terr + } + out.Seconds = seconds + + // input_reference:优先文件 part,其次字符串 part(引用对象 JSON) + if files := form.File["input_reference"]; len(files) > 0 { + fh := files[0] + f, err := fh.Open() + if err != nil { + return nil, &common.TaskError{StatusCode: http.StatusBadRequest, Message: "failed to open input_reference file"} + } + defer f.Close() + + raw, err := io.ReadAll(io.LimitReader(f, videosInputReferenceMaxBytes+1)) + if err != nil { + return nil, &common.TaskError{StatusCode: http.StatusBadRequest, Message: "failed to read input_reference file"} + } + if len(raw) > videosInputReferenceMaxBytes { + return nil, &common.TaskError{StatusCode: http.StatusBadRequest, Message: "input_reference file too large, max 6MB"} + } + if len(raw) == 0 { + return nil, &common.TaskError{StatusCode: http.StatusBadRequest, Message: "input_reference file is empty"} + } + + // 内容类型探测优先于 part 声明的 Content-Type(客户端常误标 application/octet-stream) + ct := fh.Header.Get("Content-Type") + if ct == "" || ct == "application/octet-stream" { + ct = http.DetectContentType(raw) + } + out.InputReference = "data:" + ct + ";base64," + base64.StdEncoding.EncodeToString(raw) + } else if v := formValue("input_reference"); v != "" { + ref, terr := parseVideosInputReferenceJSON(json.RawMessage(v)) + if terr != nil { + return nil, terr + } + out.InputReference = ref + } + + return out, nil +} + +// normalizeVideosSeconds 归一 seconds 字段:接受字符串(官方形态)或数字(宽松兼容),统一转字符串。 +// 不做 4/8/12 白名单校验——第三方视频模型时长档位各异,透传给适配器自行对齐。 +func normalizeVideosSeconds(raw json.RawMessage) (string, *common.TaskError) { + if len(raw) == 0 { + return "", nil + } + var s string + if err := json.Unmarshal(raw, &s); err == nil { + return strings.TrimSpace(s), nil + } + var n json.Number + if err := json.Unmarshal(raw, &n); err == nil { + if _, convErr := strconv.ParseFloat(n.String(), 64); convErr == nil { + return n.String(), nil + } + } + return "", &common.TaskError{StatusCode: http.StatusBadRequest, Message: "invalid seconds field: must be a string or number"} +} + +// parseVideosInputReferenceJSON 解析引用对象形态的 input_reference: +// - {"image_url": "https://... 或 data:..."} → 原样透传 +// - {"file_id": "..."} → 不支持(平台无 Files API),返回 400 引导改用 image_url/文件上传 +func parseVideosInputReferenceJSON(raw json.RawMessage) (string, *common.TaskError) { + if len(raw) == 0 || string(raw) == "null" { + return "", nil + } + + var ref struct { + FileID string `json:"file_id"` + ImageURL string `json:"image_url"` + } + if err := json.Unmarshal(raw, &ref); err != nil { + return "", &common.TaskError{StatusCode: http.StatusBadRequest, Message: "invalid input_reference: expected file upload or {image_url}"} + } + switch { + case ref.ImageURL != "": + return ref.ImageURL, nil + case ref.FileID != "": + return "", &common.TaskError{ + StatusCode: http.StatusBadRequest, + Message: "input_reference.file_id is not supported; upload the file directly or use image_url instead", + ErrCode: "file_id_not_supported", + } + default: + return "", &common.TaskError{StatusCode: http.StatusBadRequest, Message: "invalid input_reference: expected file upload or {image_url}"} + } +} + +// videosSizeMapping 官方四档分辨率 → 第三方通用参数(aspect_ratio + resolution)。 +// 未命中四档的 size 不做映射(第三方分辨率自由,透传 metadata.size 原值即可)。 +var videosSizeMapping = map[string]struct { + AspectRatio string + Resolution string +}{ + "720x1280": {"9:16", "720p"}, + "1280x720": {"16:9", "720p"}, + "1024x1792": {"9:16", "1080p"}, + "1792x1024": {"16:9", "1080p"}, +} + +// BuildVideosTaskBody 把归一后的创建请求转成任务框架通用请求体(JSON 字节)。 +// 顶层保留 seconds + images,metadata 写入 duration/ratio/aspect_ratio/resolution/size/image, +// 未识别字段一律透传原值,不在此处做模型能力裁剪。 +// 参数采用「双写」策略覆盖两套适配器词汇(未知键互不干扰,json.Unmarshal 自动忽略): +// - kling:顶层 seconds、metadata.aspect_ratio、metadata.image +// - volcengine(seedance):顶层 seconds、images[] 数组(→content[].image_url 首帧)、 +// metadata.ratio、metadata.duration、metadata.resolution +func BuildVideosTaskBody(req *videosCreateRequest) ([]byte, error) { + body := map[string]any{ + "model": req.Model, + "prompt": req.Prompt, + } + metadata := map[string]any{} + + if req.Seconds != "" { + body["seconds"] = req.Seconds + if d, err := strconv.Atoi(req.Seconds); err == nil { + metadata["duration"] = d + } + } + if req.Size != "" { + metadata["size"] = req.Size + if m, ok := videosSizeMapping[req.Size]; ok { + // aspect_ratio(kling)与 ratio(volcengine)值域同构:"16:9"/"9:16" + metadata["aspect_ratio"] = m.AspectRatio + metadata["ratio"] = m.AspectRatio + metadata["resolution"] = m.Resolution + } + } + if req.InputReference != "" { + metadata["image"] = req.InputReference + // volcengine 图生视频读根级 images 数组(kling 亦有该兜底) + body["images"] = []string{req.InputReference} + } + if len(metadata) > 0 { + body["metadata"] = metadata + } + + data, err := json.Marshal(body) + if err != nil { + return nil, fmt.Errorf("marshal videos task body: %w", err) + } + return data, nil +} + +// ValidateVideosCreateRequest 请求必填校验(model/prompt 官方均为必填) +func ValidateVideosCreateRequest(req *videosCreateRequest) *common.TaskError { + if req.Model == "" { + return &common.TaskError{StatusCode: http.StatusBadRequest, Message: "model is required"} + } + if strings.TrimSpace(req.Prompt) == "" { + return &common.TaskError{StatusCode: http.StatusBadRequest, Message: "prompt is required"} + } + return nil +} + +// NewVideosRequestEcho 从归一请求构造提交回显(胶水层经 TaskRelayContext.RequestEcho 传入, +// 提交时随 PrivateData 落库,retrieve 时回显 prompt/seconds/size) +func NewVideosRequestEcho(req *videosCreateRequest) videosRequestEcho { + return videosRequestEcho{ + Prompt: req.Prompt, + Seconds: req.Seconds, + Size: req.Size, + } +} diff --git a/relay/handler/videos_convert_test.go b/relay/handler/videos_convert_test.go new file mode 100644 index 0000000..956abe1 --- /dev/null +++ b/relay/handler/videos_convert_test.go @@ -0,0 +1,221 @@ +package handler + +import ( + "bytes" + "encoding/json" + "fmt" + "mime/multipart" + "net/textproto" + "strings" + "testing" +) + +// buildMultipartVideosBody 构造 multipart/form-data 的创建请求体(模拟官方 SDK 形态)。 +// 文件 part 的 Content-Type 统一声明为 application/octet-stream,以覆盖内容类型嗅探路径。 +func buildMultipartVideosBody(t *testing.T, fields map[string]string, fileField, fileName string, fileBytes []byte) ([]byte, string) { + t.Helper() + var buf bytes.Buffer + w := multipart.NewWriter(&buf) + + for k, v := range fields { + if err := w.WriteField(k, v); err != nil { + t.Fatalf("write field %s: %v", k, err) + } + } + if fileField != "" { + h := textproto.MIMEHeader{} + h.Set("Content-Disposition", fmt.Sprintf(`form-data; name="%s"; filename="%s"`, fileField, fileName)) + h.Set("Content-Type", "application/octet-stream") + part, err := w.CreatePart(h) + if err != nil { + t.Fatalf("create file part: %v", err) + } + if _, err := part.Write(fileBytes); err != nil { + t.Fatalf("write file part: %v", err) + } + } + if err := w.Close(); err != nil { + t.Fatalf("close writer: %v", err) + } + return buf.Bytes(), w.FormDataContentType() +} + +func TestParseVideosCreateRequest_JSON(t *testing.T) { + body := `{"model":"kling-v2-master","prompt":"a cat","seconds":"8","size":"1280x720","input_reference":{"image_url":"https://example.com/ref.png"}}` + req, taskErr := ParseVideosCreateRequest([]byte(body), "application/json") + if taskErr != nil { + t.Fatalf("unexpected error: %+v", taskErr) + } + if req.Model != "kling-v2-master" || req.Prompt != "a cat" || req.Seconds != "8" || req.Size != "1280x720" { + t.Fatalf("unexpected request: %+v", req) + } + if req.InputReference != "https://example.com/ref.png" { + t.Fatalf("unexpected input_reference: %s", req.InputReference) + } +} + +func TestParseVideosCreateRequest_JSONSecondsNumber(t *testing.T) { + body := `{"model":"m","prompt":"p","seconds":8}` + req, taskErr := ParseVideosCreateRequest([]byte(body), "application/json; charset=utf-8") + if taskErr != nil { + t.Fatalf("unexpected error: %+v", taskErr) + } + if req.Seconds != "8" { + t.Fatalf("expected seconds normalized to \"8\", got %q", req.Seconds) + } +} + +func TestParseVideosCreateRequest_JSONInvalidSeconds(t *testing.T) { + body := `{"model":"m","prompt":"p","seconds":true}` + _, taskErr := ParseVideosCreateRequest([]byte(body), "application/json") + if taskErr == nil || taskErr.StatusCode != 400 { + t.Fatalf("expected 400 for bool seconds, got %+v", taskErr) + } +} + +func TestParseVideosCreateRequest_JSONFileIDRejected(t *testing.T) { + body := `{"model":"m","prompt":"p","input_reference":{"file_id":"file-abc"}}` + _, taskErr := ParseVideosCreateRequest([]byte(body), "application/json") + if taskErr == nil { + t.Fatal("expected file_id to be rejected") + } + if taskErr.StatusCode != 400 || taskErr.ErrCode != "file_id_not_supported" { + t.Fatalf("unexpected error: %+v", taskErr) + } +} + +func TestParseVideosCreateRequest_MultipartWithFile(t *testing.T) { + // PNG 魔数,DetectContentType 应识别为 image/png(part 声明为 octet-stream) + pngBytes := []byte{0x89, 0x50, 0x4E, 0x47, 0x0D, 0x0A, 0x1A, 0x0A, 0x00, 0x00, 0x00, 0x0D} + body, ct := buildMultipartVideosBody(t, + map[string]string{"model": "kling-v2-master", "prompt": "a cat", "seconds": "12", "size": "720x1280"}, + "input_reference", "ref.png", pngBytes) + + req, taskErr := ParseVideosCreateRequest(body, ct) + if taskErr != nil { + t.Fatalf("unexpected error: %+v", taskErr) + } + if req.Model != "kling-v2-master" || req.Seconds != "12" || req.Size != "720x1280" { + t.Fatalf("unexpected request: %+v", req) + } + if !strings.HasPrefix(req.InputReference, "data:image/png;base64,") { + t.Fatalf("expected data url with sniffed image/png, got %.40q", req.InputReference) + } +} + +func TestParseVideosCreateRequest_MultipartStringRef(t *testing.T) { + body, ct := buildMultipartVideosBody(t, + map[string]string{ + "model": "m", + "prompt": "p", + "input_reference": `{"image_url":"https://example.com/x.jpg"}`, + }, "", "", nil) + + req, taskErr := ParseVideosCreateRequest(body, ct) + if taskErr != nil { + t.Fatalf("unexpected error: %+v", taskErr) + } + if req.InputReference != "https://example.com/x.jpg" { + t.Fatalf("unexpected input_reference: %s", req.InputReference) + } +} + +func TestBuildVideosTaskBody_FullMapping(t *testing.T) { + req := &videosCreateRequest{ + Model: "kling-v2-master", + Prompt: "a cat", + Seconds: "8", + Size: "1280x720", + InputReference: "https://example.com/ref.png", + } + body, err := BuildVideosTaskBody(req) + if err != nil { + t.Fatalf("build: %v", err) + } + + var parsed map[string]any + if err := json.Unmarshal(body, &parsed); err != nil { + t.Fatalf("unmarshal: %v", err) + } + if parsed["model"] != "kling-v2-master" || parsed["prompt"] != "a cat" { + t.Fatalf("unexpected body: %s", body) + } + if parsed["seconds"] != "8" { + t.Fatalf("expected top-level seconds \"8\", got %v", parsed["seconds"]) + } + + meta, ok := parsed["metadata"].(map[string]any) + if !ok { + t.Fatalf("expected metadata object: %s", body) + } + if meta["duration"] != float64(8) { + t.Fatalf("expected metadata.duration 8, got %v", meta["duration"]) + } + if meta["aspect_ratio"] != "16:9" || meta["resolution"] != "720p" { + t.Fatalf("expected 16:9/720p, got %v/%v", meta["aspect_ratio"], meta["resolution"]) + } + // 双写:ratio 供 volcengine(seedance) 消费,images 数组供其图生视频读取 + if meta["ratio"] != "16:9" { + t.Fatalf("expected ratio 16:9 for volcengine, got %v", meta["ratio"]) + } + if meta["size"] != "1280x720" { + t.Fatalf("expected raw size passthrough, got %v", meta["size"]) + } + if meta["image"] != "https://example.com/ref.png" { + t.Fatalf("expected metadata.image, got %v", meta["image"]) + } + images, ok := parsed["images"].([]any) + if !ok || len(images) != 1 || images[0] != "https://example.com/ref.png" { + t.Fatalf("expected root images array with reference, got %v", parsed["images"]) + } +} + +func TestBuildVideosTaskBody_Minimal(t *testing.T) { + req := &videosCreateRequest{Model: "m", Prompt: "p"} + body, err := BuildVideosTaskBody(req) + if err != nil { + t.Fatalf("build: %v", err) + } + var parsed map[string]any + if err := json.Unmarshal(body, &parsed); err != nil { + t.Fatalf("unmarshal: %v", err) + } + if _, ok := parsed["seconds"]; ok { + t.Fatal("expected no seconds field") + } + if _, ok := parsed["images"]; ok { + t.Fatal("expected no images field without input_reference") + } + if _, ok := parsed["metadata"]; ok { + t.Fatal("expected no metadata field") + } +} + +func TestBuildVideosTaskBody_UnknownSizeNoRatio(t *testing.T) { + req := &videosCreateRequest{Model: "m", Prompt: "p", Size: "1920x1080"} + body, _ := BuildVideosTaskBody(req) + var parsed map[string]any + json.Unmarshal(body, &parsed) + meta := parsed["metadata"].(map[string]any) + if _, ok := meta["aspect_ratio"]; ok { + t.Fatal("unknown size must not map aspect_ratio") + } + if _, ok := meta["ratio"]; ok { + t.Fatal("unknown size must not map ratio") + } + if meta["size"] != "1920x1080" { + t.Fatalf("expected raw size passthrough, got %v", meta["size"]) + } +} + +func TestValidateVideosCreateRequest(t *testing.T) { + if taskErr := ValidateVideosCreateRequest(&videosCreateRequest{Model: "m", Prompt: "p"}); taskErr != nil { + t.Fatalf("unexpected: %+v", taskErr) + } + if taskErr := ValidateVideosCreateRequest(&videosCreateRequest{Prompt: "p"}); taskErr == nil || taskErr.Message != "model is required" { + t.Fatalf("expected model required, got %+v", taskErr) + } + if taskErr := ValidateVideosCreateRequest(&videosCreateRequest{Model: "m", Prompt: " "}); taskErr == nil || taskErr.Message != "prompt is required" { + t.Fatalf("expected prompt required, got %+v", taskErr) + } +} diff --git a/relay/handler/videos_pipeline_test.go b/relay/handler/videos_pipeline_test.go new file mode 100644 index 0000000..f6348cc --- /dev/null +++ b/relay/handler/videos_pipeline_test.go @@ -0,0 +1,336 @@ +package handler + +import ( + "context" + "encoding/json" + "io" + "net/http" + "net/http/httptest" + "strings" + "testing" + + "github.com/shopspring/decimal" + + "github.com/qianfree/team-api/relay/common" + "github.com/qianfree/team-api/relay/constant" +) + +// 端到端集成测试:OpenAI Videos 协议(multipart)→ 转换 → 任务提交管线(kling 适配器 + mock 上游) +// → video_ 前缀 ID + 官方 Video 对象响应 + 请求回显落 PrivateData + 上游收到映射后的 kling 参数。 + +// stubVideosBillingProvider 提交路径的 TaskBillingProvider 桩(计费全放行,金额固定) +type stubVideosBillingProvider struct { + common.TaskBillingProvider +} + +func (s *stubVideosBillingProvider) EstimateTaskCost(context.Context, int64, string, map[string]float64) (decimal.Decimal, error) { + return decimal.NewFromFloat(0.1), nil +} + +func (s *stubVideosBillingProvider) PreDeductTask(_ context.Context, _ int64, _ string, cost decimal.Decimal, _ string) (decimal.Decimal, error) { + return cost, nil +} + +func (s *stubVideosBillingProvider) CheckRateLimit(context.Context, int64, int64, int64, int) (bool, string, int, int, int64) { + return true, "", 0, 0, 0 +} + +func (s *stubVideosBillingProvider) AcquireApiKeyConcurrent(context.Context, int64, int) bool { + return true +} +func (s *stubVideosBillingProvider) ReleaseApiKeyConcurrent(context.Context, int64) {} +func (s *stubVideosBillingProvider) CheckApiKeyQuota(context.Context, int64, decimal.Decimal) error { + return nil +} + +func (s *stubVideosBillingProvider) AdjustTaskBilling(_ context.Context, _ int64, _ string, _, newCost decimal.Decimal) (decimal.Decimal, error) { + return newCost, nil +} + +// stubVideosSubmitProvider 提交路径的 TaskDataProvider 桩(记录创建的任务) +type stubVideosSubmitProvider struct { + common.TaskDataProvider + + created *common.AsyncTask +} + +func (s *stubVideosSubmitProvider) CreateTask(_ context.Context, task *common.AsyncTask) error { + s.created = task + return nil +} + +func (s *stubVideosSubmitProvider) GetTaskByPublicIDAndUser(_ context.Context, publicTaskID string, _ int64, _ int64) (*common.AsyncTask, error) { + if s.created != nil && s.created.PublicTaskID == publicTaskID { + return s.created, nil + } + return nil, nil +} + +func TestHandleTaskSubmit_OpenAIVideosProtocol_FullPipeline(t *testing.T) { + // 1. mock kling 上游:记录收到的请求体,返回标准提交成功响应 + var upstreamReceived map[string]any + upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if !strings.HasSuffix(r.URL.Path, "/v1/videos/text2video") { + t.Errorf("unexpected upstream path: %s", r.URL.Path) + } + body, _ := io.ReadAll(r.Body) + if err := json.Unmarshal(body, &upstreamReceived); err != nil { + t.Errorf("upstream body not json: %v", err) + } + w.Header().Set("Content-Type", "application/json") + _ = json.NewEncoder(w).Encode(map[string]any{ + "code": 0, + "message": "success", + "data": map[string]any{"task_id": "kling-up-1", "task_status": "submitted"}, + }) + })) + defer upstream.Close() + + // 2. multipart 创建请求 → 归一 → 通用任务体(与胶水层 HandleVideoCreate 相同链路) + mBody, ct := buildMultipartVideosBody(t, map[string]string{ + "model": "kling-v2-master", + "prompt": "a cat surfing", + "seconds": "8", + "size": "1280x720", + }, "", "", nil) + req, taskErr := ParseVideosCreateRequest(mBody, ct) + if taskErr != nil { + t.Fatalf("parse: %+v", taskErr) + } + taskBody, err := BuildVideosTaskBody(req) + if err != nil { + t.Fatalf("build task body: %v", err) + } + + // 3. 跑完整提交管线(kling 适配器 + mock 上游 + 计费/存储桩) + rec := httptest.NewRecorder() + rc := &TaskRelayContext{ + TenantID: 1, + UserID: 2, + ApiKeyID: 3, + RequestID: "req_videos_test", + Writer: rec, + Protocol: videosProtocolOpenAI, + RelayMode: int(constant.RelayModeVideos), + RequestEcho: NewVideosRequestEcho(req), + } + channelMeta := &common.ChannelMeta{ + ChannelID: 11, + ChannelType: int(constant.ProviderKling), + ChannelName: "mock-kling", + BaseURL: upstream.URL, + ApiKey: "sk-test", + UpstreamModelName: "kling-v2-master", + } + billing := &stubVideosBillingProvider{} + data := &stubVideosSubmitProvider{} + + HandleTaskSubmit(context.Background(), taskBody, "/v1/videos", http.Header{}, rc, data, billing, channelMeta) + + // 4. 断言客户端响应:官方 Video 对象 + if rec.Code != http.StatusOK { + t.Fatalf("expected 200, got %d, body=%s", rec.Code, rec.Body.String()) + } + resp := decodeVideosJSON(t, rec) + if resp["object"] != "video" || resp["status"] != "queued" || resp["progress"] != float64(0) { + t.Fatalf("unexpected video object: %s", rec.Body.String()) + } + videoID, _ := resp["id"].(string) + if !strings.HasPrefix(videoID, "video_") { + t.Fatalf("expected video_ prefixed id, got %q", videoID) + } + if resp["model"] != "kling-v2-master" || resp["seconds"] != "8" || resp["size"] != "1280x720" { + t.Fatalf("expected echo fields in submit response: %s", rec.Body.String()) + } + if prompt, _ := resp["prompt"].(string); prompt != "a cat surfing" { + t.Fatalf("expected prompt echo: %s", rec.Body.String()) + } + if rc.TaskID != videoID { + t.Fatalf("rc.TaskID=%q should match response id %q", rc.TaskID, videoID) + } + + // 5. 断言落库任务:video_ ID、kling 平台、PrivateData 含 request_echo + if data.created == nil { + t.Fatal("expected task record created") + } + if data.created.PublicTaskID != videoID { + t.Fatalf("created task id=%q", data.created.PublicTaskID) + } + if data.created.Platform != "kling" { + t.Fatalf("expected platform kling, got %q", data.created.Platform) + } + var pd struct { + UpstreamTaskID string `json:"upstream_task_id"` + RequestEcho *videosRequestEcho `json:"request_echo"` + } + if err := json.Unmarshal(data.created.PrivateData, &pd); err != nil { + t.Fatalf("private data unmarshal: %v", err) + } + if pd.UpstreamTaskID != "kling-up-1" { + t.Fatalf("expected upstream task id captured, got %q", pd.UpstreamTaskID) + } + if pd.RequestEcho == nil || pd.RequestEcho.Seconds != "8" || pd.RequestEcho.Size != "1280x720" { + t.Fatalf("expected request echo persisted: %+v", pd.RequestEcho) + } + + // 6. 断言上游收到映射后的 kling 参数(seconds→duration、size→aspect_ratio) + if upstreamReceived["model_name"] != "kling-v2-master" { + t.Fatalf("upstream model_name: %v", upstreamReceived["model_name"]) + } + if upstreamReceived["duration"] != "8" { + t.Fatalf("upstream duration: %v", upstreamReceived["duration"]) + } + if upstreamReceived["aspect_ratio"] != "16:9" { + t.Fatalf("upstream aspect_ratio: %v", upstreamReceived["aspect_ratio"]) + } + if upstreamReceived["prompt"] != "a cat surfing" { + t.Fatalf("upstream prompt: %v", upstreamReceived["prompt"]) + } +} + +func TestHandleTaskSubmit_OpenAIVideosProtocol_Seedance(t *testing.T) { + // mock 火山引擎(seedance)上游:提交端点返回任务 ID + var upstreamReceived map[string]any + upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if !strings.HasSuffix(r.URL.Path, "/api/v3/contents/generations/tasks") { + t.Errorf("unexpected upstream path: %s", r.URL.Path) + } + body, _ := io.ReadAll(r.Body) + if err := json.Unmarshal(body, &upstreamReceived); err != nil { + t.Errorf("upstream body not json: %v", err) + } + w.Header().Set("Content-Type", "application/json") + _ = json.NewEncoder(w).Encode(map[string]any{"id": "cgt-seedance-1"}) + })) + defer upstream.Close() + + // multipart 创建请求:图生视频(input_reference 文件)+ 高清横屏档 + pngBytes := []byte{0x89, 0x50, 0x4E, 0x47, 0x0D, 0x0A, 0x1A, 0x0A} + mBody, ct := buildMultipartVideosBody(t, map[string]string{ + "model": "doubao-seedance-2-0-260128", + "prompt": "小猫对着镜头打哈欠", + "seconds": "12", + "size": "1792x1024", + }, "input_reference", "ref.png", pngBytes) + req, taskErr := ParseVideosCreateRequest(mBody, ct) + if taskErr != nil { + t.Fatalf("parse: %+v", taskErr) + } + taskBody, err := BuildVideosTaskBody(req) + if err != nil { + t.Fatalf("build task body: %v", err) + } + + rec := httptest.NewRecorder() + rc := &TaskRelayContext{ + TenantID: 1, + UserID: 2, + ApiKeyID: 3, + RequestID: "req_seedance_test", + Writer: rec, + Protocol: videosProtocolOpenAI, + RelayMode: int(constant.RelayModeVideos), + RequestEcho: NewVideosRequestEcho(req), + } + channelMeta := &common.ChannelMeta{ + ChannelID: 12, + ChannelType: int(constant.ProviderVolcengine), + ChannelName: "mock-volcengine", + BaseURL: upstream.URL, + ApiKey: "sk-test", + UpstreamModelName: "doubao-seedance-2-0-260128", + } + data := &stubVideosSubmitProvider{} + + HandleTaskSubmit(context.Background(), taskBody, "/v1/videos", http.Header{}, rc, data, &stubVideosBillingProvider{}, channelMeta) + + // 客户端响应:官方 Video 对象 + if rec.Code != http.StatusOK { + t.Fatalf("expected 200, got %d, body=%s", rec.Code, rec.Body.String()) + } + resp := decodeVideosJSON(t, rec) + if resp["object"] != "video" || resp["status"] != "queued" { + t.Fatalf("unexpected video object: %s", rec.Body.String()) + } + if resp["seconds"] != "12" || resp["size"] != "1792x1024" { + t.Fatalf("expected echo fields: %s", rec.Body.String()) + } + if data.created == nil || data.created.Platform != "volcengine" { + t.Fatalf("expected volcengine task record, got %+v", data.created) + } + + // 上游收到的 seedance 参数:content 文本+首帧图、duration/ratio/resolution 全部对上 + if upstreamReceived["model"] != "doubao-seedance-2-0-260128" { + t.Fatalf("upstream model: %v", upstreamReceived["model"]) + } + if upstreamReceived["duration"] != float64(12) { + t.Fatalf("upstream duration: %v (%T)", upstreamReceived["duration"], upstreamReceived["duration"]) + } + if upstreamReceived["ratio"] != "16:9" { + t.Fatalf("upstream ratio: %v", upstreamReceived["ratio"]) + } + if upstreamReceived["resolution"] != "1080p" { + t.Fatalf("upstream resolution: %v", upstreamReceived["resolution"]) + } + content, ok := upstreamReceived["content"].([]any) + if !ok || len(content) != 2 { + t.Fatalf("expected text + first_frame image content items, got %v", upstreamReceived["content"]) + } + textItem, _ := content[0].(map[string]any) + if textItem["type"] != "text" || textItem["text"] != "小猫对着镜头打哈欠" { + t.Fatalf("unexpected text item: %v", content[0]) + } + imgItem, _ := content[1].(map[string]any) + if imgItem["type"] != "image_url" { + t.Fatalf("expected image_url item, got %v", content[1]) + } + imgURL, _ := imgItem["image_url"].(map[string]any) + urlStr, _ := imgURL["url"].(string) + if !strings.HasPrefix(urlStr, "data:image/png;base64,") { + t.Fatalf("expected base64 data url as first frame, got %.40q", urlStr) + } +} + +func TestHandleTaskSubmit_LegacyProtocolUnchanged(t *testing.T) { + // 回归:Protocol 为空(legacy)时提交响应保持 {id: task_*, status: SUBMITTED} 旧格式 + upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + _ = json.NewEncoder(w).Encode(map[string]any{ + "code": 0, + "message": "success", + "data": map[string]any{"task_id": "kling-up-2", "task_status": "submitted"}, + }) + })) + defer upstream.Close() + + rec := httptest.NewRecorder() + rc := &TaskRelayContext{ + TenantID: 1, + UserID: 2, + ApiKeyID: 3, + RequestID: "req_legacy_test", + Writer: rec, + } + channelMeta := &common.ChannelMeta{ + ChannelID: 11, + ChannelType: int(constant.ProviderKling), + BaseURL: upstream.URL, + ApiKey: "sk-test", + UpstreamModelName: "kling-v2-master", + } + body := []byte(`{"model":"kling-v2-master","prompt":"legacy","metadata":{"duration":5}}`) + HandleTaskSubmit(context.Background(), body, "/v1/video/generations", http.Header{}, rc, &stubVideosSubmitProvider{}, &stubVideosBillingProvider{}, channelMeta) + + if rec.Code != http.StatusOK { + t.Fatalf("expected 200, got %d, body=%s", rec.Code, rec.Body.String()) + } + resp := decodeVideosJSON(t, rec) + if resp["status"] != "SUBMITTED" { + t.Fatalf("legacy submit must keep SUBMITTED status: %s", rec.Body.String()) + } + if id, _ := resp["id"].(string); !strings.HasPrefix(id, "task_") { + t.Fatalf("legacy submit must keep task_ prefix: %s", rec.Body.String()) + } + if _, hasObject := resp["object"]; hasObject { + t.Fatalf("legacy submit must not carry video object field: %s", rec.Body.String()) + } +} diff --git a/relay/handler/videos_test.go b/relay/handler/videos_test.go new file mode 100644 index 0000000..abe3269 --- /dev/null +++ b/relay/handler/videos_test.go @@ -0,0 +1,336 @@ +package handler + +import ( + "context" + "encoding/json" + "net/http" + "net/http/httptest" + "strings" + "testing" + "time" + + "github.com/qianfree/team-api/relay/common" +) + +// stubVideosTaskProvider 查询/删除路径的 TaskDataProvider 桩: +// 嵌入接口零值,未覆盖的方法被调用时 panic(测试应只触达覆盖的方法)。 +type stubVideosTaskProvider struct { + common.TaskDataProvider + + task *common.AsyncTask + fetchErr error + deleteCalls int + deleteErr error +} + +func (s *stubVideosTaskProvider) GetTaskByPublicIDAndUser(_ context.Context, publicTaskID string, _ int64, _ int64) (*common.AsyncTask, error) { + if s.fetchErr != nil { + return nil, s.fetchErr + } + if s.task != nil && s.task.PublicTaskID == publicTaskID { + return s.task, nil + } + return nil, nil +} + +func (s *stubVideosTaskProvider) SoftDeleteTask(_ context.Context, task *common.AsyncTask) error { + s.deleteCalls++ + if s.deleteErr != nil { + return s.deleteErr + } + task.Deleted = true + return nil +} + +func newVideosTestTask() *common.AsyncTask { + finish := time.Unix(1760000432, 0) + privateData, _ := json.Marshal(map[string]any{ + "upstream_task_id": "up-1", + "task_type": "kling", + "request_echo": map[string]any{"prompt": "a cat", "seconds": "8", "size": "1280x720"}, + }) + return &common.AsyncTask{ + ID: 1, + PublicTaskID: "video_abc", + Status: "SUCCESS", + Progress: "100%", + ModelName: "kling-v2-master", + ResultURL: "https://cdn.example.com/v.mp4", + BillingSettled: true, + FinishTime: &finish, + CreatedAt: time.Unix(1760000000, 0), + PrivateData: privateData, + TenantID: 7, + UserID: 9, + } +} + +func videosTestRC(w http.ResponseWriter) *TaskRelayContext { + return &TaskRelayContext{TenantID: 7, UserID: 9, Writer: w} +} + +func decodeVideosJSON(t *testing.T, rec *httptest.ResponseRecorder) map[string]any { + t.Helper() + var m map[string]any + if err := json.Unmarshal(rec.Body.Bytes(), &m); err != nil { + t.Fatalf("decode json: %v, body=%s", err, rec.Body.String()) + } + return m +} + +func TestVideosStatusFromTask(t *testing.T) { + cases := map[string]string{ + "NOT_START": "queued", + "SUBMITTED": "queued", + "QUEUED": "queued", + "IN_PROGRESS": "in_progress", + "SUCCESS": "completed", + "FAILURE": "failed", + "WEIRD": "queued", + } + for in, want := range cases { + if got := videosStatusFromTask(in); got != want { + t.Errorf("videosStatusFromTask(%q) = %q, want %q", in, got, want) + } + } +} + +func TestVideosProgressFromTask(t *testing.T) { + makeTask := func(status, progress string) *common.AsyncTask { + return &common.AsyncTask{Status: status, Progress: progress} + } + if got := videosProgressFromTask(makeTask("IN_PROGRESS", "42%")); got != 42 { + t.Errorf("42%% → %d", got) + } + if got := videosProgressFromTask(makeTask("IN_PROGRESS", "150%")); got != 100 { + t.Errorf("150%% should clamp to 100, got %d", got) + } + if got := videosProgressFromTask(makeTask("IN_PROGRESS", "")); got != 50 { + t.Errorf("fallback in_progress → 50, got %d", got) + } + if got := videosProgressFromTask(makeTask("SUCCESS", "")); got != 100 { + t.Errorf("fallback success → 100, got %d", got) + } +} + +func TestBuildVideoObject_SuccessWithEcho(t *testing.T) { + obj := buildVideoObject(newVideosTestTask()) + if obj.ID != "video_abc" || obj.Object != "video" || obj.Status != "completed" || obj.Progress != 100 { + t.Fatalf("unexpected object: %+v", obj) + } + if obj.Model != "kling-v2-master" || obj.Prompt == nil || *obj.Prompt != "a cat" { + t.Fatalf("unexpected model/prompt: %+v", obj) + } + if obj.Seconds != "8" || obj.Size != "1280x720" { + t.Fatalf("expected echo seconds/size, got %q/%q", obj.Seconds, obj.Size) + } + if obj.CompletedAt == nil || *obj.CompletedAt != 1760000432 { + t.Fatalf("unexpected completed_at: %v", obj.CompletedAt) + } + if obj.Error != nil || obj.RemixedFromVideoID != nil || obj.ExpiresAt != nil { + t.Fatalf("nullables must be nil: %+v", obj) + } +} + +func TestBuildVideoObject_Failure(t *testing.T) { + task := newVideosTestTask() + task.Status = "FAILURE" + task.Progress = "30%" + task.FailReason = "upstream exploded" + obj := buildVideoObject(task) + if obj.Status != "failed" || obj.Error == nil { + t.Fatalf("expected failed with error: %+v", obj) + } + if obj.Error.Code != "video_generation_failed" || obj.Error.Message != "upstream exploded" { + t.Fatalf("unexpected error object: %+v", obj.Error) + } +} + +func TestHandleVideosRetrieve_FoundAndPollHeader(t *testing.T) { + task := newVideosTestTask() + task.Status = "IN_PROGRESS" + task.Progress = "42%" + task.FinishTime = nil + + rec := httptest.NewRecorder() + provider := &stubVideosTaskProvider{task: task} + HandleVideosRetrieve(context.Background(), "video_abc", videosTestRC(rec), provider) + + if rec.Code != http.StatusOK { + t.Fatalf("status = %d, body=%s", rec.Code, rec.Body.String()) + } + if got := rec.Header().Get("openai-poll-after-ms"); got != videosPollAfterMs { + t.Fatalf("expected poll header %q, got %q", videosPollAfterMs, got) + } + m := decodeVideosJSON(t, rec) + if m["status"] != "in_progress" || m["progress"] != float64(42) || m["object"] != "video" { + t.Fatalf("unexpected body: %s", rec.Body.String()) + } +} + +func TestHandleVideosRetrieve_TerminalNoPollHeader(t *testing.T) { + rec := httptest.NewRecorder() + provider := &stubVideosTaskProvider{task: newVideosTestTask()} + HandleVideosRetrieve(context.Background(), "video_abc", videosTestRC(rec), provider) + if got := rec.Header().Get("openai-poll-after-ms"); got != "" { + t.Fatalf("terminal task must not carry poll header, got %q", got) + } +} + +func TestHandleVideosRetrieve_NotFound(t *testing.T) { + rec := httptest.NewRecorder() + provider := &stubVideosTaskProvider{} + HandleVideosRetrieve(context.Background(), "video_missing", videosTestRC(rec), provider) + if rec.Code != http.StatusNotFound { + t.Fatalf("expected 404, got %d", rec.Code) + } +} + +func TestHandleVideosDelete_HappyPath(t *testing.T) { + rec := httptest.NewRecorder() + provider := &stubVideosTaskProvider{task: newVideosTestTask()} + HandleVideosDelete(context.Background(), "video_abc", videosTestRC(rec), provider) + + if rec.Code != http.StatusOK { + t.Fatalf("expected 200, got %d, body=%s", rec.Code, rec.Body.String()) + } + if provider.deleteCalls != 1 { + t.Fatalf("expected 1 soft delete call, got %d", provider.deleteCalls) + } + m := decodeVideosJSON(t, rec) + if m["object"] != "video.deleted" || m["deleted"] != true || m["id"] != "video_abc" { + t.Fatalf("unexpected body: %s", rec.Body.String()) + } +} + +func TestHandleVideosDelete_NonTerminalRejected(t *testing.T) { + task := newVideosTestTask() + task.Status = "IN_PROGRESS" + rec := httptest.NewRecorder() + provider := &stubVideosTaskProvider{task: task} + HandleVideosDelete(context.Background(), "video_abc", videosTestRC(rec), provider) + + if rec.Code != http.StatusBadRequest { + t.Fatalf("expected 400, got %d", rec.Code) + } + if provider.deleteCalls != 0 { + t.Fatal("must not delete non-terminal task") + } + if !strings.Contains(rec.Body.String(), "video_not_deletable") { + t.Fatalf("expected video_not_deletable code, body=%s", rec.Body.String()) + } +} + +func TestHandleVideosDelete_UnsettledRejected(t *testing.T) { + task := newVideosTestTask() + task.BillingSettled = false + rec := httptest.NewRecorder() + provider := &stubVideosTaskProvider{task: task} + HandleVideosDelete(context.Background(), "video_abc", videosTestRC(rec), provider) + + if rec.Code != http.StatusBadRequest || provider.deleteCalls != 0 { + t.Fatalf("expected 400 without delete, got %d calls=%d", rec.Code, provider.deleteCalls) + } + if !strings.Contains(rec.Body.String(), "billing_pending") { + t.Fatalf("expected billing_pending code, body=%s", rec.Body.String()) + } +} + +func TestHandleVideosDelete_NotFound(t *testing.T) { + rec := httptest.NewRecorder() + provider := &stubVideosTaskProvider{} + HandleVideosDelete(context.Background(), "video_gone", videosTestRC(rec), provider) + if rec.Code != http.StatusNotFound || provider.deleteCalls != 0 { + t.Fatalf("expected 404 without delete, got %d", rec.Code) + } +} + +func TestHandleVideosContent_StreamsBytes(t *testing.T) { + content := []byte("FAKE-MP4-BYTES") + upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "video/mp4") + w.Write(content) + })) + defer upstream.Close() + + task := newVideosTestTask() + task.ResultURL = upstream.URL + + rec := httptest.NewRecorder() + provider := &stubVideosTaskProvider{task: task} + HandleVideosContent(context.Background(), "video_abc", "video", videosTestRC(rec), provider) + + if rec.Code != http.StatusOK { + t.Fatalf("expected 200, got %d, body=%s", rec.Code, rec.Body.String()) + } + if ct := rec.Header().Get("Content-Type"); ct != "video/mp4" { + t.Fatalf("expected video/mp4 passthrough, got %q", ct) + } + if rec.Body.String() != string(content) { + t.Fatalf("body mismatch: %q", rec.Body.String()) + } +} + +func TestHandleVideosContent_VariantUnsupported(t *testing.T) { + rec := httptest.NewRecorder() + provider := &stubVideosTaskProvider{task: newVideosTestTask()} + HandleVideosContent(context.Background(), "video_abc", "thumbnail", videosTestRC(rec), provider) + if rec.Code != http.StatusBadRequest { + t.Fatalf("expected 400, got %d", rec.Code) + } +} + +func TestHandleVideosContent_NotCompleted(t *testing.T) { + task := newVideosTestTask() + task.Status = "IN_PROGRESS" + rec := httptest.NewRecorder() + provider := &stubVideosTaskProvider{task: task} + HandleVideosContent(context.Background(), "video_abc", "", videosTestRC(rec), provider) + if rec.Code != http.StatusBadRequest || !strings.Contains(rec.Body.String(), "video_not_ready") { + t.Fatalf("expected 400 video_not_ready, got %d body=%s", rec.Code, rec.Body.String()) + } +} + +func TestHandleVideosContent_NoResultURL(t *testing.T) { + task := newVideosTestTask() + task.ResultURL = "" + rec := httptest.NewRecorder() + provider := &stubVideosTaskProvider{task: task} + HandleVideosContent(context.Background(), "video_abc", "", videosTestRC(rec), provider) + if rec.Code != http.StatusBadRequest || !strings.Contains(rec.Body.String(), "content_not_available") { + t.Fatalf("expected 400 content_not_available, got %d body=%s", rec.Code, rec.Body.String()) + } +} + +func TestHandleVideosContent_UpstreamFailure(t *testing.T) { + upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + w.WriteHeader(http.StatusForbidden) + })) + defer upstream.Close() + + task := newVideosTestTask() + task.ResultURL = upstream.URL + + rec := httptest.NewRecorder() + provider := &stubVideosTaskProvider{task: task} + HandleVideosContent(context.Background(), "video_abc", "", videosTestRC(rec), provider) + if rec.Code != http.StatusBadGateway { + t.Fatalf("expected 502, got %d", rec.Code) + } +} + +func TestWriteVideosSubmitResponse(t *testing.T) { + rec := httptest.NewRecorder() + writeVideosSubmitResponse(rec, "video_new", "kling-v2-master", time.Unix(1760000000, 0), videosRequestEcho{Prompt: "p", Seconds: "8", Size: "720x1280"}) + + if rec.Code != http.StatusOK { + t.Fatalf("expected 200, got %d", rec.Code) + } + m := decodeVideosJSON(t, rec) + if m["id"] != "video_new" || m["object"] != "video" || m["status"] != "queued" || m["progress"] != float64(0) { + t.Fatalf("unexpected submit response: %s", rec.Body.String()) + } + if m["seconds"] != "8" || m["size"] != "720x1280" { + t.Fatalf("expected echo fields: %s", rec.Body.String()) + } +}