1

shortdrama/dao/model_config_dao.go

@@ -24,6 +24,7 @@ func init() {
 			"schema TEXT NOT NULL DEFAULT '',"+
 			"price INTEGER NOT NULL DEFAULT 0,"+
 			"price_unit TEXT NOT NULL DEFAULT 'second',"+
+			"concurrency_count INTEGER NOT NULL DEFAULT 1,"+
 			"created_at DATETIME DEFAULT (datetime('now','localtime')),"+
 			"updated_at DATETIME DEFAULT (datetime('now','localtime'))"+
 			")")
@@ -54,6 +55,7 @@ func init() {
 				"schema TEXT NOT NULL DEFAULT '',"+
 				"price INTEGER NOT NULL DEFAULT 0,"+
 				"price_unit TEXT NOT NULL DEFAULT 'second',"+
+				"concurrency_count INTEGER NOT NULL DEFAULT 1,"+
 				"created_at DATETIME DEFAULT (datetime('now','localtime')),"+
 				"updated_at DATETIME DEFAULT (datetime('now','localtime'))"+
 				")")
@@ -105,6 +107,25 @@ func init() {
 		g.Log().Info(ctx, "model_config 表结构迁移完成")
 	}
 
+	// 检测缺少 concurrency_count 列的情况（已有新表结构但字段不全）
+	if r, _ := g.DB().GetAll(ctx, "PRAGMA table_info("+public.TableNameModelConfig+")"); r != nil {
+		hasCol := false
+		for _, col := range r {
+			if col["name"].String() == "concurrency_count" {
+				hasCol = true
+				break
+			}
+		}
+		if !hasCol {
+			g.Log().Info(ctx, "检测到 model_config 缺少 concurrency_count 列，正在补充...")
+			if _, err := g.DB().Exec(ctx, "ALTER TABLE "+public.TableNameModelConfig+" ADD COLUMN concurrency_count INTEGER NOT NULL DEFAULT 1"); err != nil {
+				g.Log().Warningf(ctx, "补充 concurrency_count 列失败: %v", err)
+			} else {
+				g.Log().Info(ctx, "concurrency_count 列补充完成")
+			}
+		}
+	}
+
 	// 初始化默认模型配置（仅当表为空时）
 	count, _ := g.DB().Model(public.TableNameModelConfig).Ctx(ctx).Count()
 	if count == 0 {

shortdrama/model/dto/config_dto.go

@@ -23,13 +23,14 @@ type GetModelConfigListRes struct {
 
 // SaveModelConfigReq 保存单个模型配置
 type SaveModelConfigReq struct {
-	g.Meta    `path:"/model" method:"post" tags:"模型配置" summary:"保存模型配置"`
-	Id        int64       `json:"id"`
-	ModelType string      `json:"modelType" v:"required" dc:"chat=对话模型 video=视频模型"`
-	ModelName string      `json:"modelName" v:"required" dc:"模型名称"`
-	Schema    *gjson.Json `json:"schema" dc:"请求体JSON Schema"`
-	Price     int         `json:"price" dc:"价格（分）"`
-	PriceUnit string      `json:"priceUnit" dc:"价格单位(second=每秒/video=每次视频)"`
+	g.Meta           `path:"/model" method:"post" tags:"模型配置" summary:"保存模型配置"`
+	Id               int64       `json:"id"`
+	ModelType        string      `json:"modelType" v:"required" dc:"chat=对话模型 video=视频模型"`
+	ModelName        string      `json:"modelName" v:"required" dc:"模型名称"`
+	Schema           *gjson.Json `json:"schema" dc:"请求体JSON Schema"`
+	Price            int         `json:"price" dc:"价格（分）"`
+	PriceUnit        string      `json:"priceUnit" dc:"价格单位(second=每秒/video=每次视频)"`
+	ConcurrencyCount int         `json:"concurrencyCount" dc:"并发处理数(同一模型最多同时处理的请求数量)"`
 }
 
 type GetUserModelConfigReq struct {

shortdrama/model/entity/model_config.go

@@ -6,12 +6,13 @@ import (
 
 // ModelConfig 模型配置（每行一个模型，chat/video 分开存储）
 type ModelConfig struct {
-	Id        int64       `orm:"id" json:"id" dc:"配置ID"`
-	ModelType string      `orm:"model_type" json:"modelType" dc:"chat=对话模型 video=视频模型"`
-	ModelName string      `orm:"model_name" json:"modelName" dc:"模型名称"`
-	Schema    string      `orm:"schema" json:"schema" dc:"请求体JSON Schema"`
-	Price     int         `orm:"price" json:"price" dc:"价格（分）"`
-	PriceUnit string      `orm:"price_unit" json:"priceUnit" dc:"价格单位(second=每秒/video=每次视频)"`
-	CreatedAt *gtime.Time `orm:"created_at" json:"createdAt" dc:"创建时间"`
-	UpdatedAt *gtime.Time `orm:"updated_at" json:"updatedAt" dc:"更新时间"`
+	Id               int64       `orm:"id" json:"id" dc:"配置ID"`
+	ModelType        string      `orm:"model_type" json:"modelType" dc:"chat=对话模型 video=视频模型"`
+	ModelName        string      `orm:"model_name" json:"modelName" dc:"模型名称"`
+	Schema           string      `orm:"schema" json:"schema" dc:"请求体JSON Schema"`
+	Price            int         `orm:"price" json:"price" dc:"价格（分）"`
+	PriceUnit        string      `orm:"price_unit" json:"priceUnit" dc:"价格单位(second=每秒/video=每次视频)"`
+	ConcurrencyCount int         `orm:"concurrency_count" json:"concurrencyCount" dc:"并发处理数(同一视频模型最多同时处理的剧集数量)"`
+	CreatedAt        *gtime.Time `orm:"created_at" json:"createdAt" dc:"创建时间"`
+	UpdatedAt        *gtime.Time `orm:"updated_at" json:"updatedAt" dc:"更新时间"`
 }

shortdrama/service/drama_service.go

@@ -29,6 +29,7 @@ import (
 	"github.com/gogf/gf/v2/database/gdb"
 	"github.com/gogf/gf/v2/frame/g"
 	"github.com/gogf/gf/v2/os/gcache"
+	"github.com/gogf/gf/v2/os/grpool"
 )
 
 // ==================== Drama CRUD ====================
@@ -37,6 +38,9 @@ type dramaService struct{}
 
 var DramaService = new(dramaService)
 
+// videoPollPool 视频轮询 goroutine 池，在 StartVideoPoller 中初始化
+var videoPollPool *grpool.Pool
+
 const (
 	DefaultFirstFramePath = "default_first_frame.png" // 图生视频默认首帧图片路径
 )
@@ -976,6 +980,15 @@ func (s *dramaService) GetEpisodePollStatus(ctx context.Context, epId int64) (*d
 
 // StartVideoPoller 启动后台视频轮询器
 func (s *dramaService) StartVideoPoller(ctx context.Context) {
+	// 初始化视频轮询 goroutine 池（启动时初始化一次）
+	videoCfg := ConfigService.GetActiveModel(ctx, "video")
+	poolSize := 1
+	if videoCfg != nil && videoCfg.ConcurrencyCount > 0 {
+		poolSize = videoCfg.ConcurrencyCount
+	}
+	videoPollPool = grpool.New(poolSize)
+	g.Log().Infof(ctx, "视频轮询池已初始化，并发数=%d", poolSize)
+
 	go func() {
 		ticker := time.NewTicker(15 * time.Second)
 		defer ticker.Stop()
@@ -991,122 +1004,202 @@ func (s *dramaService) StartVideoPoller(ctx context.Context) {
 	}()
 }
 
-// pollPendingVideos 扫描所有 generating 任务，轮询或重试视频合成
+// pollPendingVideos 扫描所有 generating 任务，按剧集分组串行推进
 func (s *dramaService) pollPendingVideos(ctx context.Context) {
-	modelCfg := ConfigService.GetActiveModel(ctx, "video")
-	// 注意：各任务所属用户不同，轮询时使用 per-task 的 merged config
+	if videoPollPool == nil {
+		return
+	}
+
 	allTasks, err := dao.GenerationTask.ListByStatuses(ctx, []string{consts.TaskStatusGenerating, consts.TaskStatusReview})
 	if err != nil || len(allTasks) == 0 {
 		return
 	}
 
-	// 只处理 generating 任务，review/completed 任务由主流程推进时已更新 episode 状态
-	var tasks []*entity.GenerationTask
+	// 按 (DramaId, EpisodeId) 分组
+	groups := make(map[int64][]*entity.GenerationTask)
 	for _, t := range allTasks {
-		if t.Status == consts.TaskStatusGenerating {
-			tasks = append(tasks, t)
+		if t.Status != consts.TaskStatusGenerating {
+			continue
 		}
+		groups[t.EpisodeId] = append(groups[t.EpisodeId], t)
 	}
-
-	if len(tasks) == 0 {
+	if len(groups) == 0 {
 		return
 	}
 
-	// 按 episode 分组，后续统一更新 poll cache
-	epUpdates := make(map[int64]bool)
+	var wg sync.WaitGroup
 
-	for _, task := range tasks {
-		if task.VideoUrl != "" {
-			continue
-		}
-
-		if task.VideoTaskId != "" {
-			// 已有提交任务 — 轮询状态（不消耗 API 额度）
-			// 加载该任务所属用户的合并配置
-			var taskModel *MergedModelConfig
-			if task.DramaId > 0 {
-				if td, _ := dao.Drama.GetOne(ctx, task.DramaId); td != nil && td.UserId > 0 {
-					taskModel = ConfigService.GetMergedConfig(ctx, td.UserId, "video")
-				}
-			}
-			if taskModel == nil {
-				taskModel = &MergedModelConfig{ModelConfig: modelCfg}
-			}
-			videoURL, err := s.pollVideoTaskOnce(ctx, taskModel, task.VideoTaskId)
-			if err != nil {
-				if strings.Contains(err.Error(), "FAILED") || strings.Contains(err.Error(), "failed") {
-					// duration 不支持：清除 video_task_id 让下一轮重新提交（不带 duration）
-					if strings.Contains(strings.ToLower(err.Error()), "duration") {
-						g.Log().Warningf(ctx, "轮询器: 任务 %d 第%d段视频不支持自定义时长，清除 task_id 准备不携带 duration 重试", task.Id, task.SegmentIdx+1)
-						_ = dao.GenerationTask.UpdateFields(ctx, task.Id, g.Map{"video_task_id": ""})
-						// 持久化探测结果，让后续 calcSegDurs 用保守值切段
-						ConfigService.ClearModelListCache(ctx)
-						epUpdates[task.EpisodeId] = true
-					} else {
-						g.Log().Warningf(ctx, "轮询器: 任务 %d 第%d段视频任务失败: %v", task.Id, task.SegmentIdx+1, err)
-						_ = dao.GenerationTask.UpdateFailed(ctx, task.Id, err.Error())
-						epUpdates[task.EpisodeId] = true
-					}
-				} else if strings.Contains(err.Error(), "RUNNING") {
-					g.Log().Debugf(ctx, "轮询器: 任务 %d 第%d段视频正在生成中，继续等待...", task.Id, task.SegmentIdx+1)
-				} else {
-					g.Log().Warningf(ctx, "轮询器: 任务 %d 第%d段视频查询异常: %v", task.Id, task.SegmentIdx+1, err)
-				}
-				continue
-			}
-
-			// 视频就绪 — 下载到本地
-			dramaTitle := fmt.Sprintf("drama_%d", task.DramaId)
-			if d, _ := dao.Drama.GetOne(ctx, task.DramaId); d != nil {
-				dramaTitle = d.Title
-			}
-			epTitle := ""
-			if ep, _ := dao.Episode.GetOne(ctx, task.EpisodeId); ep != nil {
-				epTitle = ep.Title
-			}
-			safeEp := sanitizeDirName(epTitle)
-			localPath, dlErr := s.downloadToLocal(ctx, videoURL, dramaTitle, "产出视频",
-				fmt.Sprintf("%s_seg_%d.mp4", safeEp, task.SegmentIdx))
-			if dlErr != nil {
-				g.Log().Warningf(ctx, "轮询器: 任务 %d 第%d段视频下载失败，使用远程地址: %v", task.Id, task.SegmentIdx+1, dlErr)
-				_ = dao.GenerationTask.UpdateFields(ctx, task.Id, g.Map{"video_url": videoURL, "video_task_id": ""})
-			} else {
-				g.Log().Infof(ctx, "轮询器: 任务 %d 第%d段视频已下载: %s", task.Id, task.SegmentIdx+1, localPath)
-				_ = dao.GenerationTask.UpdateFields(ctx, task.Id, g.Map{"video_url": localPath, "video_task_id": ""})
-			}
-			_ = dao.GenerationTask.UpdateStatus(ctx, task.Id, consts.TaskStatusReview)
-			g.Log().Infof(ctx, "轮询器: 任务 %d 第%d段视频生成完成，进入审核", task.Id, task.SegmentIdx+1)
-			epUpdates[task.EpisodeId] = true
-			// 更新剧集状态：如果全部任务都到了 review/completed，剧集标记为 review
-			if allTasksDone, _ := dao.GenerationTask.ListByEpisode(ctx, task.EpisodeId); len(allTasksDone) > 0 {
-				allReview := true
-				for _, t := range allTasksDone {
-					if t.Status != consts.TaskStatusReview && t.Status != consts.TaskStatusCompleted {
-						allReview = false
-						break
-					}
-				}
-				if allReview {
-					_ = dao.Episode.UpdateStatus(ctx, task.EpisodeId, consts.EpisodeStatusReview, "")
-				}
-			}
-			continue
-		}
-	}
-	// 刷新所有 generating 任务的 episode 缓存（不管状态有无变化，确保前端轮询不走 DB）
-	for _, t := range tasks {
-		epUpdates[t.EpisodeId] = true
+	for _, tasks := range groups {
+		wg.Add(1)
+		tasks := tasks
+		videoPollPool.Add(ctx, func(ctx context.Context) {
+			defer wg.Done()
+			s.processOneEpisode(ctx, tasks)
+		})
 	}
 
-	// 更新 episode poll cache
-	for epId := range epUpdates {
+	wg.Wait()
+
+	// 更新所有涉及剧集的 poll cache
+	epDone := make(map[int64]bool)
+	for _, t := range allTasks {
+		if t.Status == consts.TaskStatusGenerating {
+			epDone[t.EpisodeId] = true
+		}
+	}
+	for epId := range epDone {
 		if ts, e := dao.GenerationTask.ListByEpisode(ctx, epId); e == nil {
 			setPollCache(ctx, epId, ts)
 		}
 	}
 }
 
-// pollVideoTaskOnce 单次查询视频生成任务状态
+// processOneEpisode 处理单个剧集的片段：串行推进，一次只处理一个未完成的段
+func (s *dramaService) processOneEpisode(ctx context.Context, tasks []*entity.GenerationTask) {
+	// 按 SegmentIdx 升序排列
+	sort.Slice(tasks, func(i, j int) bool {
+		return tasks[i].SegmentIdx < tasks[j].SegmentIdx
+	})
+
+	// 找到第一个未完成的任务（VideoUrl == ""）
+	var current *entity.GenerationTask
+	for _, t := range tasks {
+		if t.VideoUrl == "" {
+			current = t
+			break
+		}
+	}
+	if current == nil {
+		return
+	}
+
+	// 加载短剧和 merged config
+	drama, _ := dao.Drama.GetOne(ctx, current.DramaId)
+	if drama == nil {
+		g.Log().Warningf(ctx, "轮询器: 短剧 %d 不存在，跳过", current.DramaId)
+		return
+	}
+	modelCfg := ConfigService.GetMergedConfig(ctx, drama.UserId, "video")
+	if modelCfg.ApiKey == "" || modelCfg.BaseUrl == "" {
+		g.Log().Warningf(ctx, "轮询器: 短剧 %d 用户 %d 视频模型未配置", current.DramaId, drama.UserId)
+		return
+	}
+
+	// ====== 分支 A：已有 video_task_id → 轮询 ======
+	if current.VideoTaskId != "" {
+		videoURL, err := s.pollVideoTaskOnce(ctx, modelCfg, current.VideoTaskId)
+		if err != nil {
+			if strings.Contains(err.Error(), "FAILED") || strings.Contains(err.Error(), "failed") {
+				g.Log().Warningf(ctx, "轮询器: 任务 %d 第%d段视频任务失败: %v", current.Id, current.SegmentIdx+1, err)
+				_ = dao.GenerationTask.UpdateFailed(ctx, current.Id, err.Error())
+				clearPollCache(ctx, current.EpisodeId)
+			} else if strings.Contains(err.Error(), "RUNNING") {
+				g.Log().Debugf(ctx, "轮询器: 任务 %d 第%d段视频正在生成中，继续等待...", current.Id, current.SegmentIdx+1)
+			} else {
+				g.Log().Warningf(ctx, "轮询器: 任务 %d 第%d段视频查询异常: %v", current.Id, current.SegmentIdx+1, err)
+			}
+			return
+		}
+
+		// 视频就绪 — 下载到本地
+		dramaTitle := fmt.Sprintf("drama_%d", current.DramaId)
+		if drama.Title != "" {
+			dramaTitle = drama.Title
+		}
+		epTitle := ""
+		if ep, _ := dao.Episode.GetOne(ctx, current.EpisodeId); ep != nil {
+			epTitle = ep.Title
+		}
+		safeEp := sanitizeDirName(epTitle)
+		localPath, dlErr := s.downloadToLocal(ctx, videoURL, dramaTitle, "产出视频",
+			fmt.Sprintf("%s_seg_%d.mp4", safeEp, current.SegmentIdx))
+		if dlErr != nil {
+			g.Log().Warningf(ctx, "轮询器: 任务 %d 第%d段视频下载失败，使用远程地址: %v", current.Id, current.SegmentIdx+1, dlErr)
+			_ = dao.GenerationTask.UpdateFields(ctx, current.Id, g.Map{"video_url": videoURL, "video_task_id": ""})
+		} else {
+			g.Log().Infof(ctx, "轮询器: 任务 %d 第%d段视频已下载: %s", current.Id, current.SegmentIdx+1, localPath)
+			_ = dao.GenerationTask.UpdateFields(ctx, current.Id, g.Map{"video_url": localPath, "video_task_id": ""})
+		}
+		_ = dao.GenerationTask.UpdateStatus(ctx, current.Id, consts.TaskStatusReview)
+		g.Log().Infof(ctx, "轮询器: 任务 %d 第%d段视频生成完成，进入审核", current.Id, current.SegmentIdx+1)
+
+		// 更新剧集状态：如果全部任务都到了 review/completed，剧集标记为 review
+		if allTasksDone, _ := dao.GenerationTask.ListByEpisode(ctx, current.EpisodeId); len(allTasksDone) > 0 {
+			allReview := true
+			for _, t := range allTasksDone {
+				if t.Status != consts.TaskStatusReview && t.Status != consts.TaskStatusCompleted {
+					allReview = false
+					break
+				}
+			}
+			if allReview {
+				_ = dao.Episode.UpdateStatus(ctx, current.EpisodeId, consts.EpisodeStatusReview, "")
+			}
+		}
+		return
+	}
+
+	// ====== 分支 B：无 video_task_id → 需要提交（含首次提交和崩溃恢复）======
+
+	// 从 DB 查前一段是否已完成，并获取首帧参考 URL
+	firstFrameURL := ""
+	if current.SegmentIdx > 0 {
+		prev := findTaskBySegmentIdx(ctx, current.EpisodeId, current.SegmentIdx-1)
+		if prev == nil || prev.VideoUrl == "" {
+			return
+		}
+		firstFrameURL, _ = extractLastFrame(ctx, prev.VideoUrl)
+	}
+
+	// 使用 task.script（bodyJSON）直接提交，不走 Agent
+	var bodyMap map[string]any
+	if err := json.Unmarshal([]byte(current.Script), &bodyMap); err != nil {
+		g.Log().Errorf(ctx, "轮询器: 任务 %d script 不是合法 JSON: %v", current.Id, err)
+		_ = dao.GenerationTask.UpdateFailed(ctx, current.Id, "script 不是合法 JSON")
+		return
+	}
+
+	// 从 bodyJSON 的 model 字段查找当前任务实际使用的模型 schema
+	bodyModelSchema := modelCfg.Schema
+	var bodyModel struct {
+		Model string `json:"model"`
+	}
+	if err := json.Unmarshal([]byte(current.Script), &bodyModel); err == nil && bodyModel.Model != "" {
+		models, _ := dao.ModelConfig.GetAll(ctx)
+		for _, m := range models {
+			if m.ModelType == "video" && m.ModelName != "" && strings.HasPrefix(bodyModel.Model, m.ModelName) {
+				bodyModelSchema = m.Schema
+				break
+			}
+		}
+	}
+	// 按正确格式注入首帧（仅用于本次提交，不存回 script）
+	savedScript := current.Script
+	if firstFrameURL != "" {
+		injectFirstFrame(bodyMap, bodyModelSchema, firstFrameURL)
+	}
+
+	bodyJSON, _ := json.Marshal(bodyMap)
+
+	taskId, _, err := resubmitVideoTask(ctx, modelCfg.ApiKey, modelCfg.BaseUrl, modelCfg.Schema, bodyJSON)
+	if err != nil {
+		g.Log().Errorf(ctx, "轮询器: 第%d段视频提交失败: %v", current.SegmentIdx+1, err)
+		return
+	}
+
+	_ = dao.GenerationTask.UpdateFields(ctx, current.Id, g.Map{
+		"video_task_id": taskId,
+		"script":        savedScript,
+	})
+	g.Log().Infof(ctx, "轮询器: 第%d段视频已提交(taskId=%s)", current.SegmentIdx+1, taskId)
+}
+
+// findTaskBySegmentIdx 从数据库按 EpisodeId+SegmentIdx 查找前一段任务
+// 不查询内存 tasks 是因为前一段可能已是 review 状态，不在当前轮询列表中
+func findTaskBySegmentIdx(ctx context.Context, episodeId int64, segIdx int) *entity.GenerationTask {
+	t, _ := dao.GenerationTask.GetByEpisodeAndSegment(ctx, episodeId, segIdx)
+	return t
+}
 func (s *dramaService) pollVideoTaskOnce(ctx context.Context, modelCfg *MergedModelConfig, taskId string) (string, error) {
 	queryURL := strings.ReplaceAll(modelCfg.TaskCallbackUrl, "{task_id}", taskId)
 
@@ -1471,8 +1564,9 @@ func resubmitVideoTask(ctx context.Context, apiKey, baseURL, schema string, body
 		return "", nil, fmt.Errorf("解析 bodyJSON 失败: %w", err)
 	}
 
-	// 解析 reference_urls：http/https/data: 开头的保持原样，文件路径转为 base64
+	// 解析 reference_urls/media 中的文件路径：http/https/data: 保持原样，否则转 base64
 	if input, ok := bodyMap["input"].(map[string]any); ok {
+		// reference_urls 字符串数组
 		if refs, ok := input["reference_urls"].([]any); ok {
 			resolved := make([]any, len(refs))
 			for i, r := range refs {
@@ -1488,6 +1582,27 @@ func resubmitVideoTask(ctx context.Context, apiKey, baseURL, schema string, body
 			}
 			input["reference_urls"] = resolved
 		}
+		// media 对象数组，转 url 字段
+		if media, ok := input["media"].([]any); ok {
+			for _, item := range media {
+				m, ok := item.(map[string]any)
+				if !ok {
+					continue
+				}
+				u, _ := m["url"].(string)
+				if u == "" {
+					continue
+				}
+				if strings.HasPrefix(u, "http://") || strings.HasPrefix(u, "https://") || strings.HasPrefix(u, "data:") {
+					continue
+				}
+				if b64, err := imageFileToBase64(u); err == nil {
+					m["url"] = b64
+				} else {
+					g.Log().Warningf(ctx, "media url 无法解析: %s", u)
+				}
+			}
+		}
 	}
 
 	payload, _ := json.Marshal(bodyMap)
@@ -1590,6 +1705,79 @@ func schemaHeaders(schemaStr string) map[string]string {
 	return headers
 }
 
+// injectFirstFrame 从 model_config.schema 解析首帧字段，按正确格式注入
+//  1. items.properties.type.enum 含 first_frame → 追加 {type:"first_frame", url:...}
+//  2. reference_urls 字符串数组 → 前置插入 URL
+func injectFirstFrame(body map[string]any, schemaJSON, firstFrameURL string) {
+	if firstFrameURL == "" || schemaJSON == "" {
+		return
+	}
+	var vs map[string]any
+	if err := json.Unmarshal([]byte(schemaJSON), &vs); err != nil {
+		return
+	}
+	bodyDef, ok := nested(vs, "body").(map[string]any)
+	if !ok {
+		return
+	}
+	for _, sectionVal := range bodyDef {
+		section, ok := sectionVal.(map[string]any)
+		if !ok {
+			continue
+		}
+		for fieldKey, fieldDef := range section {
+			fd, ok := fieldDef.(map[string]any)
+			if !ok || fd["type"] != "array" {
+				continue
+			}
+			items, _ := fd["items"].(map[string]any)
+			if items == nil {
+				continue
+			}
+			if props, _ := items["properties"].(map[string]any); props != nil {
+				if typeProp, _ := props["type"].(map[string]any); typeProp != nil {
+					if enum, _ := typeProp["enum"].([]any); enum != nil {
+						for _, e := range enum {
+							if e == "first_frame" {
+								for _, v := range body {
+									if sm, ok := v.(map[string]any); ok {
+										if existing, ok := sm[fieldKey].([]any); ok {
+											// 已有 first_frame 则更新 URL，避免重复添加
+											found := false
+											for _, item := range existing {
+												if m, ok := item.(map[string]any); ok {
+													if m["type"] == "first_frame" {
+														m["url"] = firstFrameURL
+														found = true
+														break
+													}
+												}
+											}
+											if !found {
+												existing = append(existing, map[string]any{"type": "first_frame", "url": firstFrameURL})
+											}
+											return
+										}
+									}
+								}
+							}
+						}
+					}
+				}
+			}
+		}
+	}
+	// 回退：reference_urls 简单字符串数组
+	for _, v := range body {
+		if sm, ok := v.(map[string]any); ok {
+			if existing, ok := sm["reference_urls"].([]any); ok {
+				sm["reference_urls"] = append([]any{firstFrameURL}, existing...)
+				return
+			}
+		}
+	}
+}
+
 // setSchemaField 在 body 中按字段名查找并设置值（不关心在哪个 section 下）
 func setSchemaField(body map[string]any, name string, value any) {
 	for _, v := range body {
@@ -1811,6 +1999,32 @@ func (s *dramaService) overlayBackgroundMusic(ctx context.Context, dramaId int64
 	return videoPath, nil
 }
 
+// extractLastFrame 使用 ffmpeg 截取视频尾帧并返回 base64 data URL（不落盘）
+func extractLastFrame(ctx context.Context, videoPath string) (string, error) {
+	if _, err := os.Stat(videoPath); os.IsNotExist(err) {
+		return "", fmt.Errorf("视频文件不存在: %s", videoPath)
+	}
+	if _, err := exec.LookPath("ffmpeg"); err != nil {
+		return "", fmt.Errorf("ffmpeg 不可用: %w", err)
+	}
+	cmd := exec.Command("ffmpeg",
+		"-sseof", "-0.1",
+		"-i", videoPath,
+		"-q:v", "2",
+		"-vframes", "1",
+		"-f", "image2pipe",
+		"-",
+	)
+	out, err := cmd.Output()
+	if err != nil {
+		if ee, ok := err.(*exec.ExitError); ok {
+			return "", fmt.Errorf("ffmpeg 截取尾帧失败: %w, stderr: %s", err, string(ee.Stderr))
+		}
+		return "", fmt.Errorf("ffmpeg 截取尾帧失败: %w", err)
+	}
+	return "data:image/jpeg;base64," + base64.StdEncoding.EncodeToString(out), nil
+}
+
 // waitForSegmentVideo 等待指定任务的视频生成完成。
 // 轮询视频 API 直到视频就绪，下载到本地，更新 DB 的 video_url。
 // 用于串行模式中让下一段能提取上一段的尾帧作为首帧。