From e93ffbd4f80c5cf622ba769489c226df88cd630e Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=BC=A0=E6=96=8C?= <259278618@qq.com> Date: Sat, 18 Jul 2026 14:11:26 +0800 Subject: [PATCH] 1 --- short_drama.db | Bin 462848 -> 462848 bytes shortdrama/dao/model_config_dao.go | 21 ++ shortdrama/model/dto/config_dto.go | 15 +- shortdrama/model/entity/model_config.go | 17 +- shortdrama/service/drama_service.go | 406 ++++++++++++++++++------ 5 files changed, 348 insertions(+), 111 deletions(-) diff --git a/short_drama.db b/short_drama.db index ee600f4d49178029586f2184fb5c79e1e64c84cf..6f07d104f078f6b06b91c11241441312651fddcc 100644 GIT binary patch delta 1905 zcmdT_du$X%7~h%Q9k{mFUa3%9O1Z8SXu0?GdhL>?n(#;seITKdptjrFJ-YU8_t@LJ zwwy>yqY!+6UgJne3}RYZpipgOi6&YDi2+||Ofbgqh>6667^+4RFu~bUu1GZgq5qid zB;V{e-#7Dp^ZWg#?_f#a!IFIo^G7I(+Db+<8LKIZ9eTy#2P1E{4}EB%`zi8qdAHJA zZA&J6lg)JADczPwSuBe`tysL%Qs0=~H@R}6;mkyK+Wt{7wzo4PL((jN=RCMxWSOMq zvaOW$8*8VPVO!_a=IqS@*$Jj)!fU?<4Rji`z^s9GeokRs9e92Nww(qg>CS6K#>=08 z1SoLWL#h%=Xqu#idfP&3LebOxUl&7fg;D!4*agW(;9Y#?EZ9vfcnY_i1A9!X5??$A zwq}`LGx2lX(}|7AvtTbUv<~n)Jb50V7Ulv)T}XTFpOwB*KH#uX7BF|QwXxCKUj*JO z0Je}CjYT9~64NWE9Jy7zr-t|WU4Cz{wyMT(m%;B!={0$`NF|$?+(DVnrV7hOnn97@ z5Aa(<=;V@gIVAs_Ll1$X`g)L*MM>=z6rZcwsri!5a72iAri`v6yvZ8EI@k_@13#01 zCHT@1T7+|kQ5l|UfCX6Xfi~&R8B6LI~4%a7PMP^kl{_+GWajo!p zc#p^D@l`uLL8mup_XhYtEg$gQanL(gfNW}h`9K-)l4IdpP4L^*VocK7qH-i6t4f=$ zc1cS7Zxj5bgWUKs<90Iywj3+5AzMl1e#KUj=9IGkqeeXXuNz_Zo&b-cjQ&3MI9i(l zk$YoY(&FxIor1EZLss0Yqyi{Xti?#~1Vk%p8N8vdkL<>~4zPP44emwq`Jq0r#JP95jKP;yP5Z zm*Z2Mti%!ooGC6UiLyWp+;s#k#NUsgaD}W((YWc0X)3{*E=$uhO$tjUxuv$GDu*O9 z=Ef0p7_T2g1^7f0oRc5dHCfrhZFg{bZ)|$O#Ze^UL!+p!@XjA@c1c2u%xv2_%ZhoK zSWfMDS_4cY$~ul zz2|l@UY~-?@t!f%goGZB_XhBR6m(iiWzm@4CiUo&BI3zya2c*Yj-JZYVk>rVnxM<7 z*@jsvd#m|c4l84*1YbRd*5O0P&I;{<9U0`o-**5UppxCmc5jy7k`Ht)aDp2~mHo@r#!O%&bq z>gNj#@>bl0kL~XmaGKg{DeA9pY}{xC<4b@oEXWZ_T=Pi7y49owVvPE92qaHM7c#f4=puD69XA(|eZ z@|YyEPy94hK^wnrVG4V|CY*Db*#Wxninw%IHLYRR2tiVEy9k}ynnO~AQ!0MJ3DsR^={3Ahij?kn&hbPR+{o3eB%{Hm8Fe}ktW>-ro%Mdv0V ziQ_2%rV2rIvgSK&z~=%Iw7ccjR(Y=*{`@v{<*lBK)mvEd&{rKa4a00O0zp089162~ zFS@lr*yen~S6fE3!?nl(6o1ixZo}xwY_YcYrm~?xqdy<&@wa!Rc5J1)j|d)mHGvcC zas=JB(d%QxFMr{L4)9LhfY+`V2Az(gDY2raH^+##e3HaY~!n+(T>+w%-JWeHD~LJBnRp05%Mnmqez_ehY|8I<$ElowIjq$mzT(VdSHaO z=#63GlpJd1+vB6Z3=_})ikcH-8!fFw`^$>?I2hLTvaGJ6=$lnM2U^~0xAInYv$d%^ z-?2k8)!ok-Rn;<`skF7JeS5w$wX;lZn58(mIv!c2`Cjh4(uhZ2X{?B~#$VQ~kF1Dm zvqL_6F|TZ1c&l=+lKb`dI$I(LvSb9^7vVA%U4uSF^x7B+yzz%E{JI~V>lbj2;hea^ zChKNHlc|RSMss6hh;=*?1`~re%)aG85SUVj9yoY-9>F6?Q4}5n;W5?!5foON6K@I& zj-dg+lgCl=mg6YH({0d|wz5_}wKJ1V(~iy9SFM{ylNmMwA)_T2ZXEA^I4rQg(^%OB zm(w}ZxDPI(2d42i(n=l}53ibi>K)9~JL276@lhZi 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。 // 用于串行模式中让下一段能提取上一段的尾帧作为首帧。