diff --git a/short_drama.db b/short_drama.db index ee600f4..6f07d10 100644 Binary files a/short_drama.db and b/short_drama.db differ diff --git a/shortdrama/dao/model_config_dao.go b/shortdrama/dao/model_config_dao.go index 31a307d..2dc992c 100644 --- a/shortdrama/dao/model_config_dao.go +++ b/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 { diff --git a/shortdrama/model/dto/config_dto.go b/shortdrama/model/dto/config_dto.go index b2acb57..ef52cc6 100644 --- a/shortdrama/model/dto/config_dto.go +++ b/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 { diff --git a/shortdrama/model/entity/model_config.go b/shortdrama/model/entity/model_config.go index 5bef39c..10c404d 100644 --- a/shortdrama/model/entity/model_config.go +++ b/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:"更新时间"` } diff --git a/shortdrama/service/drama_service.go b/shortdrama/service/drama_service.go index b4a04a4..bc680b9 100644 --- a/shortdrama/service/drama_service.go +++ b/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。 // 用于串行模式中让下一段能提取上一段的尾帧作为首帧。