From 5100a52bca6be4212716cfc0ad6f637431d00278 Mon Sep 17 00:00:00 2001 From: wonder Date: Mon, 25 May 2026 19:05:55 +0800 Subject: [PATCH 1/2] =?UTF-8?q?fix:=20=E4=B8=83=E7=89=9B=E4=BA=91=E7=A7=81?= =?UTF-8?q?=E6=9C=89=20bucket=20=E4=B8=8B=E8=BD=BD=20URL=20=E7=AD=BE?= =?UTF-8?q?=E5=90=8D=E6=94=AF=E6=8C=81?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 对私有读 bucket，通过 HMAC-SHA1 签名生成带 e（过期时间）和 token（下载凭证） 参数的临时下载 URL，前端获取的素材 URL 和下载接口均返回签名 URL。 新增配置项 GEN2D_QINIU_URL_EXPIRE，默认 3600 秒。 --- backend/internal/config/config.go | 3 ++ backend/internal/config/config.yml | 8 +++++ backend/internal/handler/generate.go | 2 +- backend/internal/service/storage.go | 49 +++++++++++++++++++++++++--- docs/api.md | 4 +-- docs/database.md | 11 ++++--- 6 files changed, 65 insertions(+), 12 deletions(-) diff --git a/backend/internal/config/config.go b/backend/internal/config/config.go index 000f62e..61be9e8 100755 --- a/backend/internal/config/config.go +++ b/backend/internal/config/config.go @@ -71,6 +71,7 @@ type QiniuConfig struct { Bucket string `mapstructure:"bucket"` CDNHost string `mapstructure:"cdn_host"` UseHTTPS bool `mapstructure:"use_https"` + URLExpire int64 `mapstructure:"url_expire"` // 私有下载签名 URL 有效期（秒），默认 3600 } // Load 从 YAML 配置文件和环境变量加载配置。 @@ -142,6 +143,7 @@ func setDefaults(v *viper.Viper) { v.SetDefault("qiniu.bucket", "") v.SetDefault("qiniu.cdn_host", "") v.SetDefault("qiniu.use_https", true) + v.SetDefault("qiniu.url_expire", int64(3600)) } func bindEnvVars(v *viper.Viper) { @@ -179,4 +181,5 @@ func bindEnvVars(v *viper.Viper) { v.BindEnv("qiniu.bucket", "GEN2D_QINIU_BUCKET") v.BindEnv("qiniu.cdn_host", "GEN2D_QINIU_CDN_HOST") v.BindEnv("qiniu.use_https", "GEN2D_QINIU_USE_HTTPS") + v.BindEnv("qiniu.url_expire", "GEN2D_QINIU_URL_EXPIRE") } diff --git a/backend/internal/config/config.yml b/backend/internal/config/config.yml index 74055ba..3e5e9a5 100755 --- a/backend/internal/config/config.yml +++ b/backend/internal/config/config.yml @@ -31,3 +31,11 @@ image_gen: width: 1024 height: 1024 quality: "low" + +qiniu: + access_key: "" + secret_key: "" + bucket: "" + cdn_host: "" + use_https: true + url_expire: 3600 # 私有下载签名 URL 有效期（秒） diff --git a/backend/internal/handler/generate.go b/backend/internal/handler/generate.go index b5fb767..7e877d4 100755 --- a/backend/internal/handler/generate.go +++ b/backend/internal/handler/generate.go @@ -207,7 +207,7 @@ func GetAssets(c *gin.Context) { for _, a := range assets { response = append(response, model.AssetResponse{ Key: a.Key, - URL: a.URL, + URL: storageSvc.GetSignedURL(a.Key), Format: a.Format, }) } diff --git a/backend/internal/service/storage.go b/backend/internal/service/storage.go index 8eac783..ac8a7d1 100644 --- a/backend/internal/service/storage.go +++ b/backend/internal/service/storage.go @@ -3,6 +3,9 @@ package service import ( "bytes" "context" + "crypto/hmac" + "crypto/sha1" + "encoding/base64" "fmt" "net/url" "path/filepath" @@ -65,15 +68,43 @@ func (s *StorageService) Upload(ctx context.Context, key string, data []byte) (s return "", fmt.Errorf("qiniu upload: %w", err) } - cdnURL := s.buildURL(key) + cdnURL := s.buildBaseURL(key) l.Info("upload succeeded", "key", key, "url", cdnURL) return cdnURL, nil } // GetDownloadURL 生成素材的下载 URL。 -// 公开 bucket 直接返回 CDN URL；私有 bucket 生成带签名的临时 URL。 +// 私有 bucket 生成带签名和过期时间的临时 URL。 func (s *StorageService) GetDownloadURL(_ context.Context, key string) (string, error) { - return s.buildURL(key), nil + return s.signDownloadURL(key), nil +} + +// GetSignedURL 获取私有资源签名 URL（公开方法，供 handler 层使用）。 +func (s *StorageService) GetSignedURL(key string) string { + return s.signDownloadURL(key) +} + +// signDownloadURL 生成七牛云私有资源下载签名 URL。 +// 格式：http:///?e=&token= +func (s *StorageService) signDownloadURL(key string) string { + baseURL := s.buildBaseURL(key) + + expire := s.cfg.URLExpire + if expire <= 0 { + expire = 3600 + } + deadline := time.Now().Unix() + expire + + signStr := fmt.Sprintf("%s?e=%d", baseURL, deadline) + + mac := hmac.New(sha1.New, []byte(s.cfg.SecretKey)) + mac.Write([]byte(signStr)) + signature := mac.Sum(nil) + + encodedSign := base64.URLEncoding.EncodeToString(signature) + token := fmt.Sprintf("%s:%s", s.cfg.AccessKey, encodedSign) + + return fmt.Sprintf("%s?e=%d&token=%s", baseURL, deadline, token) } // Delete 从七牛云删除指定对象。 @@ -91,9 +122,17 @@ func (s *StorageService) Delete(ctx context.Context, key string) error { return nil } -// buildURL 根据配置构建完整的 CDN URL。 -func (s *StorageService) buildURL(key string) string { +// buildBaseURL 根据配置构建完整的基础 CDN URL（不含签名参数）。 +func (s *StorageService) buildBaseURL(key string) string { host := strings.TrimRight(s.cfg.CDNHost, "/") + scheme := "http" + if s.cfg.UseHTTPS { + scheme = "https" + } + // 如果 host 已经包含 scheme，直接使用；否则添加 + if !strings.HasPrefix(host, "http://") && !strings.HasPrefix(host, "https://") { + host = scheme + "://" + host + } segments := strings.Split(key, "/") for i, seg := range segments { segments[i] = url.PathEscape(seg) diff --git a/docs/api.md b/docs/api.md index 9ebe4e2..c5352d9 100755 --- a/docs/api.md +++ b/docs/api.md @@ -516,14 +516,14 @@ POST /api/v1/prompt/optimize ### GET /api/v1/assets/download -公开接口（无需认证）。返回 302 重定向到七牛云 CDN URL。前端可通过此接口加载素材图片，作为 CDN 直接访问的替代方案。 +公开接口（无需认证）。返回 302 重定向到七牛云带签名的临时下载 URL（私有 bucket 签名有效时长默认为 3600 秒）。前端可通过此接口加载素材图片。 | 参数 | 类型 | 说明 | |------|------|------| | `key` | string | 对象存储 Key，查询参数 | 响应： -- HTTP 302，`Location` 头指向 CDN URL +- HTTP 302，`Location` 头指向带签名的临时下载 URL - 400：缺少 key 参数 - 500：生成下载链接失败 diff --git a/docs/database.md b/docs/database.md index c5b7589..2c0aa10 100755 --- a/docs/database.md +++ b/docs/database.md @@ -150,16 +150,18 @@ users/user_a1B2c3/projects/proj_abc/tasks/task_xyz/output/spritesheet.png ### 访问方式 -素材上传后返回 CDN URL，前端直接通过该 URL 访问图片。 +素材上传后 asset 表存储对象 Key 和基础 CDN URL。由于 bucket 为私有读，前端通过 API 获取的 `url` 字段为带签名参数的临时下载链接（有效期可配置，默认 3600 秒）。 -下载接口 `GET /api/v1/assets/download?key=...` 返回 302 重定向到 CDN URL。 +格式：`https://cdn.example.com/?e=&token=` + +下载接口 `GET /api/v1/assets/download?key=...` 返回 302 重定向到带签名的临时 URL。 ### asset 表的 url 字段 -存储七牛云 CDN 完整 URL，如： +存储七牛云基础 CDN URL（不含签名），如： `https://cdn.example.com/users/user_a1B2c3/projects/proj_abc/tasks/task_xyz/output/spritesheet.png` -前端直接使用该 URL 加载图片，无需拼接路径。 +注意：该字段仅在数据库中使用，API 返回给前端的 URL 会根据对象 Key 实时生成带签名的临时下载链接。 --- @@ -190,6 +192,7 @@ users/user_a1B2c3/projects/proj_abc/tasks/task_xyz/output/spritesheet.png | `GEN2D_QINIU_BUCKET` | — | 七牛云 Bucket 名称 | | `GEN2D_QINIU_CDN_HOST` | — | CDN 域名（如 `https://cdn.example.com`） | | `GEN2D_QINIU_USE_HTTPS` | `true` | 是否使用 HTTPS | +| `GEN2D_QINIU_URL_EXPIRE` | `3600` | 私有下载签名 URL 有效期（秒） | 在 `config.go` 中扩展字段即可，无需额外依赖。 From 96ac56cb4e04e9b10b2c856ee0693be2f38bbc15 Mon Sep 17 00:00:00 2001 From: Gmaker689 <1711322114@qq.com> Date: Mon, 25 May 2026 19:10:46 +0800 Subject: [PATCH 2/2] =?UTF-8?q?fix:=20=E4=BF=AE=E5=A4=8D=E7=AE=A1=E7=BA=BF?= =?UTF-8?q?context=E5=8F=96=E6=B6=88=E3=80=81=E6=B7=BB=E5=8A=A0FIFO?= =?UTF-8?q?=E4=BB=BB=E5=8A=A1=E9=98=9F=E5=88=97=E3=80=81=E7=A7=BB=E9=99=A4?= =?UTF-8?q?=E9=87=8D=E5=A4=8D=E6=8F=90=E7=A4=BA=E8=AF=8D=E4=BC=98=E5=8C=96?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 三个修复： 1. runPipelineBg 改用 context.Background()，避免 HTTP 响应返回后 Gin 取消 request context 导致后台管线静默失败 2. 新增 TaskQueue FIFO 串行队列，任务提交后进入 pending 状态排队， 按提交顺序逐个执行，前端轮询显示排队中 3. promptOptimizerNode 移除 RunPromptAgent 调用，提示词优化仅由 前端在提交前通过 /api/v1/prompt/optimize 执行一次，管线内只做 风格合并和技术参数追加 --- backend/cmd/main.go | 5 ++ backend/internal/handler/generate.go | 38 +++++++--- backend/internal/service/nodes.go | 27 ++----- backend/internal/service/queue.go | 104 +++++++++++++++++++++++++++ frontend/src/stores/generation.ts | 2 +- 5 files changed, 145 insertions(+), 31 deletions(-) create mode 100644 backend/internal/service/queue.go diff --git a/backend/cmd/main.go b/backend/cmd/main.go index 967a175..5d06703 100755 --- a/backend/cmd/main.go +++ b/backend/cmd/main.go @@ -44,6 +44,11 @@ func main() { // 初始化项目服务 service.InitProjectService(db.GetDB()) + // 初始化生成任务队列（FIFO 串行执行） + generateQueue := service.NewTaskQueue() + handler.SetGenerateQueue(generateQueue) + defer generateQueue.Stop() + r := gin.New() r.Use(mildware.Logger()) r.Use(mildware.Recovery()) diff --git a/backend/internal/handler/generate.go b/backend/internal/handler/generate.go index b5fb767..e9f928f 100755 --- a/backend/internal/handler/generate.go +++ b/backend/internal/handler/generate.go @@ -16,6 +16,14 @@ import ( "github.com/gin-gonic/gin" ) +// generateQueue 全局生成任务队列，由 main 通过 SetGenerateQueue 注入。 +var generateQueue *service.TaskQueue + +// SetGenerateQueue 设置生成任务队列。 +func SetGenerateQueue(q *service.TaskQueue) { + generateQueue = q +} + // GenerateRequest 素材生成请求。 type GenerateRequest struct { ProjectID string `json:"projectId"` @@ -43,7 +51,7 @@ type AssetsResponse struct { } // Generate 素材生成接口（异步）。 -// 立即返回 taskId，后台执行管线，前端通过 GET /tasks/:taskId 轮询进度。 +// 立即返回 taskId，任务进入 FIFO 队列串行执行，前端通过 GET /tasks/:taskId 轮询进度。 func Generate(c *gin.Context) { var req GenerateRequest if err := c.ShouldBindJSON(&req); err != nil { @@ -57,18 +65,29 @@ func Generate(c *gin.Context) { } taskID := fmt.Sprintf("task-%d", time.Now().UnixMilli()) - // 保存任务到数据库 + // 保存任务到数据库，初始状态为 pending if err := saveTaskToDB(c.Request.Context(), projectID, taskID, req); err != nil { logger.FromCtx(c.Request.Context()).Error("failed to save task", "error", err) c.JSON(http.StatusInternalServerError, model.Fail(http.StatusInternalServerError, "创建任务失败")) return } - // 返回 taskId - c.JSON(http.StatusOK, model.OK(GenerateResponse{TaskID: taskID})) + // 加入 FIFO 队列 + queuePos := 1 + if generateQueue != nil { + queuePos = generateQueue.Enqueue(&service.TaskJob{ + Ctx: context.Background(), + ProjectID: projectID, + TaskID: taskID, + Execute: func(ctx context.Context) error { + runPipelineBg(ctx, projectID, taskID, req) + return nil + }, + }) + } - // 后台执行管线 - go runPipelineBg(c.Request.Context(), projectID, taskID, req) + c.JSON(http.StatusOK, model.OK(GenerateResponse{TaskID: taskID})) + _ = queuePos } // runPipelineBg 后台执行生成管线，更新任务状态。 @@ -82,6 +101,7 @@ func runPipelineBg(ctx context.Context, projectID, taskID string, req GenerateRe updateTaskInDB(ctx, taskID, "running", stage, "", progress) }) + // 队列调度后才标记为 running，初始写入时是 pending updateTaskInDB(ctx, taskID, "running", "prompt_builder", "", 5) in := service.PipelineInput{ @@ -240,9 +260,9 @@ func saveTaskToDB(ctx context.Context, projectID, taskID string, req GenerateReq ProjectID: uint(projectIDUint), Prompt: req.Prompt, AssetType: req.AssetType, - Status: "running", - Progress: 5, - Stage: "prompt_builder", + Status: "pending", + Progress: 0, + Stage: "", RetryCount: 0, CreatedAt: time.Now(), UpdatedAt: time.Now(), diff --git a/backend/internal/service/nodes.go b/backend/internal/service/nodes.go index 56c35b3..ce7c847 100755 --- a/backend/internal/service/nodes.go +++ b/backend/internal/service/nodes.go @@ -14,10 +14,10 @@ import ( "github.com/cloudwego/eino/compose" ) -// promptOptimizerNode 节点：调用 PromptAgent 生成规范提示词，合并风格与重试信息。 -// 输入 PipelineInput，输出最终提示词字符串（直接供 AssetGenerator 消费）。 +// promptOptimizerNode 节点：合并风格描述、重试信息与技术参数，输出最终提示词。 +// 提示词优化已由前端在提交前完成，管线内不再重复调用 PromptAgent。 var promptOptimizerNode = compose.InvokableLambda(func(ctx context.Context, in PipelineInput) (string, error) { - // 合并风格描述，注入原始 Prompt 中 + // 合并风格描述 styleDesc := buildStyleDescription(in.ProjectStyle, in.TaskStyle) if styleDesc != "" { if in.Prompt != "" { @@ -36,26 +36,11 @@ var promptOptimizerNode = compose.InvokableLambda(func(ctx context.Context, in P } } - if len(in.Tags) == 0 && in.Prompt == "" { - return "", fmt.Errorf("pipeline: Prompt and Tags are both empty") + if in.Prompt == "" { + return "", fmt.Errorf("pipeline: prompt is empty") } - // 有标签时调用 PromptAgent 优化提示词 - if len(in.Tags) > 0 { - agentIn := PromptAgentInput{ - Tags: in.Tags, - AssetType: in.AssetType, - Prompt: in.Prompt, - UserNote: in.UserNote, - } - output, err := RunPromptAgent(ctx, agentIn) - if err != nil { - return "", fmt.Errorf("prompt agent: %w", err) - } - return output.Prompt, nil - } - - // 无标签时直接使用原始 Prompt，补上技术参数段 + // 追加技术参数段，不再调用 PromptAgent return appendTechNotes(in.Prompt, in.AssetType, in.Params), nil }) diff --git a/backend/internal/service/queue.go b/backend/internal/service/queue.go new file mode 100644 index 0000000..ea89ce0 --- /dev/null +++ b/backend/internal/service/queue.go @@ -0,0 +1,104 @@ +package service + +import ( + "context" + "sync" + + "gen2d/internal/logger" +) + +// TaskJob 队列中的任务。 +type TaskJob struct { + Ctx context.Context + ProjectID string + TaskID string + Execute func(ctx context.Context) error +} + +// TaskQueue 串行 FIFO 任务队列。 +type TaskQueue struct { + mu sync.Mutex + jobs []*TaskJob + ready chan struct{} + stop chan struct{} + stopped bool +} + +// NewTaskQueue 创建任务队列并启动调度协程。 +func NewTaskQueue() *TaskQueue { + q := &TaskQueue{ + jobs: make([]*TaskJob, 0), + ready: make(chan struct{}, 1), + stop: make(chan struct{}), + } + go q.run() + return q +} + +// Enqueue 将任务加入队尾，返回队列中的位置（1-based）。 +func (q *TaskQueue) Enqueue(job *TaskJob) int { + q.mu.Lock() + defer q.mu.Unlock() + q.jobs = append(q.jobs, job) + pos := len(q.jobs) + select { + case q.ready <- struct{}{}: + default: + } + return pos +} + +// QueueLen 返回当前队列长度。 +func (q *TaskQueue) QueueLen() int { + q.mu.Lock() + defer q.mu.Unlock() + return len(q.jobs) +} + +// Stop 优雅关闭队列。 +func (q *TaskQueue) Stop() { + q.mu.Lock() + defer q.mu.Unlock() + if !q.stopped { + q.stopped = true + close(q.stop) + } +} + +func (q *TaskQueue) run() { + for { + select { + case <-q.stop: + return + case <-q.ready: + q.processNext() + } + } +} + +func (q *TaskQueue) processNext() { + q.mu.Lock() + if len(q.jobs) == 0 { + q.mu.Unlock() + return + } + job := q.jobs[0] + q.jobs = q.jobs[1:] + // 如果队列还有任务，重新发信号 + if len(q.jobs) > 0 { + select { + case q.ready <- struct{}{}: + default: + } + } + q.mu.Unlock() + + l := logger.With("task_id", job.TaskID, "project_id", job.ProjectID) + l.Info("task queue executing job", "queue_remaining", len(q.jobs)) + + if err := job.Execute(job.Ctx); err != nil { + l.Error("task job failed", "error", err) + } else { + l.Info("task job completed") + } +} diff --git a/frontend/src/stores/generation.ts b/frontend/src/stores/generation.ts index 6686a2f..9f740ea 100755 --- a/frontend/src/stores/generation.ts +++ b/frontend/src/stores/generation.ts @@ -120,7 +120,7 @@ export const useGenerationStore = create((set) => ({ function stageLabel(stage?: string): string { switch (stage) { - case 'prompt_builder': return '优化提示词...' + case 'prompt_builder': return '构建提示词...' case 'asset_generator': return '生成素材中...' case 'quality_supervisor': return '质检中...' case 'format_adapter': return '格式转换中...'