package httpapi import ( "encoding/json" "errors" "net/http" "strconv" "strings" "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/runner" "github.com/easyai/easyai-ai-gateway/apps/api/internal/store" ) const volcesContentsCompatibilityMarker = "volces_contents_generations_v3" // createVolcesContentsGenerationTask godoc // @Summary 创建火山内容生成任务 // @Description 统一公开入口兼容火山方舟内容生成任务。网关 task id 是查询与取消用的公开 id;上游 id 另以 upstream_task_id 保留。 // @Tags volces-compatible // @Accept json // @Produce json // @Security BearerAuth // @Success 200 {object} map[string]any // @Router /api/v1/contents/generations/tasks [post] func (s *Server) createVolcesContentsGenerationTask(w http.ResponseWriter, r *http.Request) { user, ok := auth.UserFromContext(r.Context()) if !ok || user == nil { writeError(w, http.StatusUnauthorized, "unauthorized") return } body, err := s.decodeTaskRequestBody(r.Context(), w, r, "videos.generations") if err != nil { writeError(w, http.StatusBadRequest, err.Error(), clients.ErrorCode(err)) return } task, err := s.createVolcesCompatibleTask(r, user, body) if err != nil { writeVolcesCompatibleTaskError(w, err) return } writeJSON(w, http.StatusOK, volcesCompatibleTask(task)) } // getVolcesContentsGenerationTask godoc // @Summary 查询火山内容生成任务 // @Tags volces-compatible // @Produce json // @Security BearerAuth // @Success 200 {object} map[string]any // @Router /api/v1/contents/generations/tasks/{taskID} [get] func (s *Server) getVolcesContentsGenerationTask(w http.ResponseWriter, r *http.Request) { task, ok := s.volcesCompatibleTaskForUser(w, r) if !ok { return } writeJSON(w, http.StatusOK, volcesCompatibleTask(task)) } // listVolcesContentsGenerationTasks godoc // @Summary 列出火山内容生成任务 // @Tags volces-compatible // @Produce json // @Security BearerAuth // @Success 200 {object} map[string]any // @Router /api/v1/contents/generations/tasks [get] func (s *Server) listVolcesContentsGenerationTasks(w http.ResponseWriter, r *http.Request) { user, ok := auth.UserFromContext(r.Context()) if !ok || user == nil { writeError(w, http.StatusUnauthorized, "unauthorized") return } page := portraitAssetQueryInt(r, "page_num", "pageNumber", "page") pageSize := portraitAssetQueryInt(r, "page_size", "pageSize") tasks, err := s.store.ListVolcesCompatibleTasks(r.Context(), user, store.VolcesCompatibleTaskListFilter{ CompatibilityMarker: volcesContentsCompatibilityMarker, Status: r.URL.Query().Get("filter.status"), Model: r.URL.Query().Get("filter.model"), TaskIDs: r.URL.Query()["filter.task_ids"], Page: page, PageSize: pageSize, }) if err != nil { s.logger.Error("list Volces-compatible tasks failed", "error", err) writeError(w, http.StatusInternalServerError, "list tasks failed") return } items := make([]any, 0) for _, task := range tasks.Items { items = append(items, volcesCompatibleTask(task)) } writeJSON(w, http.StatusOK, map[string]any{ "items": items, "total": tasks.Total, "page_num": tasks.Page, "page_size": tasks.PageSize, // data/page are retained as additive gateway fields for existing callers. "data": items, "page": tasks.Page, }) } // deleteVolcesContentsGenerationTask godoc // @Summary 取消火山内容生成任务 // @Description 取消网关任务;对于已提交且保存了上游任务标识的 Volces 视频任务,同时调用火山 DELETE 接口并持久化取消状态。 // @Tags volces-compatible // @Produce json // @Security BearerAuth // @Success 200 {object} map[string]any // @Router /api/v1/contents/generations/tasks/{taskID} [delete] func (s *Server) deleteVolcesContentsGenerationTask(w http.ResponseWriter, r *http.Request) { task, ok := s.volcesCompatibleTaskForUser(w, r) if !ok { return } user, _ := auth.UserFromContext(r.Context()) result, err := s.runner.CancelVolcesVideoTask(r.Context(), task, user) if err != nil { if errors.Is(err, runner.ErrTaskAccessDenied) { writeError(w, http.StatusNotFound, "task not found") return } s.logger.Error("cancel Volces-compatible task failed", "error", err) writeError(w, http.StatusInternalServerError, "cancel task failed") return } updated, err := s.store.GetTask(r.Context(), task.ID) if err != nil { writeError(w, http.StatusInternalServerError, "get cancelled task failed") return } response := volcesCompatibleTask(updated) response["cancelled"] = result.Cancelled response["cancellable"] = result.Cancellable response["message"] = result.Message writeJSON(w, http.StatusOK, response) } // createLegacyVolcesVideoGeneration godoc // @Summary 创建 server-main 兼容视频任务 // @Description 兼容 server-main 的 /api/v1/video/generations,返回 submitted 和 task_id;额外保留火山任务字段。 // @Tags volces-compatible // @Accept json // @Produce json // @Security BearerAuth // @Success 200 {object} map[string]any // @Router /api/v1/video/generations [post] func (s *Server) createLegacyVolcesVideoGeneration(w http.ResponseWriter, r *http.Request) { user, ok := auth.UserFromContext(r.Context()) if !ok || user == nil { writeError(w, http.StatusUnauthorized, "unauthorized") return } body, err := s.decodeTaskRequestBody(r.Context(), w, r, "videos.generations") if err != nil { writeError(w, http.StatusBadRequest, err.Error(), clients.ErrorCode(err)) return } task, err := s.createVolcesCompatibleTask(r, user, body) if err != nil { writeVolcesCompatibleTaskError(w, err) return } response := volcesCompatibleTask(task) response["status"] = "submitted" response["task_id"] = task.ID writeJSON(w, http.StatusOK, response) } // getLegacyVolcesVideoResult godoc // @Summary 查询 server-main 兼容视频结果 // @Tags volces-compatible // @Produce json // @Security BearerAuth // @Success 200 {object} map[string]any // @Router /api/v1/ai/result/{taskID} [get] func (s *Server) getLegacyVolcesVideoResult(w http.ResponseWriter, r *http.Request) { task, ok := s.volcesCompatibleTaskForUser(w, r) if !ok { return } compat := volcesCompatibleTask(task) legacyStatus := "process" switch compat["status"] { case "succeeded": legacyStatus = "success" case "failed", "cancelled": legacyStatus = "failed" } writeJSON(w, http.StatusOK, map[string]any{ "status": legacyStatus, "task_id": task.ID, "data": compat["content"], "result": compat, }) } func (s *Server) createVolcesCompatibleTask(r *http.Request, user *auth.User, body map[string]any) (store.GatewayTask, error) { model := strings.TrimSpace(volcesCompatString(body["model"])) if model == "" { return store.GatewayTask{}, &clients.ClientError{Code: "invalid_parameter", Message: "model is required", StatusCode: http.StatusBadRequest, Retryable: false} } if !apiKeyScopeAllowed(user, "videos.generations") { return store.GatewayTask{}, &clients.ClientError{Code: "forbidden", Message: "api key scope does not allow video generation", StatusCode: http.StatusForbidden, Retryable: false} } body["_gateway_compatibility"] = volcesContentsCompatibilityMarker task, err := s.prepareAndCreateGatewayTask(r.Context(), r, user, "videos.generations", model, body, true) if err != nil { return store.GatewayTask{}, err } if err := s.runner.EnqueueAsyncTask(r.Context(), task); err != nil { return store.GatewayTask{}, &clients.ClientError{Code: "enqueue_failed", Message: err.Error(), StatusCode: http.StatusServiceUnavailable, Retryable: true} } return task, nil } func (s *Server) volcesCompatibleTaskForUser(w http.ResponseWriter, r *http.Request) (store.GatewayTask, bool) { user, ok := auth.UserFromContext(r.Context()) if !ok || user == nil { writeError(w, http.StatusUnauthorized, "unauthorized") return store.GatewayTask{}, false } task, err := s.store.GetTask(r.Context(), strings.TrimSpace(r.PathValue("taskID"))) if err != nil { if store.IsNotFound(err) { writeError(w, http.StatusNotFound, "task not found") return store.GatewayTask{}, false } s.logger.Error("get Volces-compatible task failed", "error", err) writeError(w, http.StatusInternalServerError, "get task failed") return store.GatewayTask{}, false } if !isVolcesCompatibleTask(task) || !kelingCompatTaskOwnedBy(task, user) { writeError(w, http.StatusNotFound, "task not found") return store.GatewayTask{}, false } return task, true } func isVolcesCompatibleTask(task store.GatewayTask) bool { return task.Kind == "videos.generations" && strings.TrimSpace(volcesCompatString(task.Request["_gateway_compatibility"])) == volcesContentsCompatibilityMarker } func volcesCompatibleTask(task store.GatewayTask) map[string]any { response := cloneVolcesCompatibleMap(task.Result) if len(response) == 0 { response = cloneVolcesCompatibleMap(task.RemoteTaskPayload) } if response == nil { response = map[string]any{} } response["id"] = task.ID response["model"] = firstNonEmpty(volcesCompatString(response["model"]), task.Model) response["status"] = volcesCompatibleTaskStatus(task.Status) response["created_at"] = task.CreatedAt.Unix() response["updated_at"] = task.UpdatedAt.Unix() if task.RemoteTaskID != "" { response["upstream_task_id"] = task.RemoteTaskID } for _, key := range []string{"content", "seed", "resolution", "ratio", "duration", "frames", "framespersecond"} { if response[key] == nil && task.Request[key] != nil { response[key] = task.Request[key] } } if len(task.Usage) > 0 && response["usage"] == nil { response["usage"] = task.Usage } if task.Status == "failed" || task.Status == "cancelled" { response["error"] = map[string]any{"code": firstNonEmpty(task.ErrorCode, strings.ToUpper(task.Status)), "message": firstNonEmpty(task.ErrorMessage, task.Error, task.Message)} } response["gateway_task_id"] = task.ID response["gateway_status"] = task.Status response["billings"] = task.Billings response["billing_summary"] = task.BillingSummary response["final_charge_amount"] = task.FinalChargeAmount return response } func volcesCompatibleTaskStatus(status string) string { switch strings.ToLower(strings.TrimSpace(status)) { case "succeeded", "success", "completed": return "succeeded" case "failed": return "failed" case "cancelled", "canceled": return "cancelled" case "running", "processing": return "running" default: return "queued" } } func cloneVolcesCompatibleMap(source map[string]any) map[string]any { if len(source) == 0 { return nil } raw, err := json.Marshal(source) if err != nil { return map[string]any{} } var out map[string]any if err := json.Unmarshal(raw, &out); err != nil { return map[string]any{} } return out } func writeVolcesCompatibleTaskError(w http.ResponseWriter, err error) { status := http.StatusInternalServerError var staged *gatewayTaskCreationError if errors.As(err, &staged) { err = staged.Err } var clientErr *clients.ClientError if errors.As(err, &clientErr) && clientErr.StatusCode > 0 { status = clientErr.StatusCode } else if errors.As(err, &clientErr) { status = http.StatusBadRequest } writeError(w, status, err.Error(), clients.ErrorCode(err)) } func volcesCompatString(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) default: return "" } }