diff --git a/workflow/dao/session/exec_workflow_dao.go b/workflow/dao/session/exec_workflow_dao.go index e2f516a..a81ffe1 100644 --- a/workflow/dao/session/exec_workflow_dao.go +++ b/workflow/dao/session/exec_workflow_dao.go @@ -36,6 +36,9 @@ func (d *execWorkflowDao) Delete(ctx context.Context, req *sessionDto.DeleteExec } func (d *execWorkflowDao) Update(ctx context.Context, req *sessionDto.UpdateWorkflowReq) (rows int64, err error) { + if req.Id <= 0 { + return + } r, err := gfdb.DB(ctx, public.DbNameBlackDeacon).Model(ctx, public.TableNameExecWorkflow).OmitEmpty().Data(&req).Where(entity.ExecWorkflowCol.Id, req.Id).Update() if err != nil { return diff --git a/workflow/service/flow/flow_ws_exec.go b/workflow/service/flow/flow_ws_exec.go index 7c5153b..a6f4e48 100644 --- a/workflow/service/flow/flow_ws_exec.go +++ b/workflow/service/flow/flow_ws_exec.go @@ -147,7 +147,10 @@ func handleExecute(ctx context.Context, conn *wsCommon.WsConnection, payload int _ = writeJSON(conn, &wsCommon.WsPushMsg{Type: "ack", Message: fmt.Sprintf("开始执行工作流(共 %d 个节点)", len(execPayload.FlowContent.Nodes))}) execId, err := executeOrResume(progressCtx, conn, execPayload) - recordWorkflow(saveCtx, execId, time.Since(start), err) + if !g.IsEmpty(execId) { + glog.Infof(saveCtx, "工作流执行完成，execId: %v", execId) + recordWorkflow(saveCtx, execId, time.Since(start), err) + } if err != nil { _ = writeJSON(conn, &wsCommon.WsPushMsg{Type: "error", Message: "工作流执行失败", Error: err.Error()}) return @@ -243,15 +246,18 @@ func executeOrResume(ctx context.Context, conn *wsCommon.WsConnection, req *sess lastExec, err := sessionDao.ExecWorkflowDao.GetLatestBySessionAndFlow(ctx, conn.SessionId, req.FlowId) if err != nil { glog.Errorf(ctx, "查询最近工作流执行记录失败: %v", err) - return execute(ctx, conn, req) + return 0, fmt.Errorf("查询最近工作流执行记录失败: %v", err) } - if lastExec != nil && *lastExec.Status == *flow.FlowExecutionStatusFailed.Code() && flowContentEqual(lastExec.RequestParams, req.FlowContent) { - _ = writeJSON(conn, &wsCommon.WsPushMsg{Type: "round_start", Message: "运行开始", Data: map[string]interface{}{ - "id": lastExec.Id, - }}) - return reExecute(ctx, lastExec.Id) + if lastExec != nil { + if *lastExec.Status == *flow.FlowExecutionStatusFailed.Code() && flowContentEqual(lastExec.RequestParams, req.FlowContent) { + _ = writeJSON(conn, &wsCommon.WsPushMsg{Type: "round_start", Message: "运行开始", Data: map[string]interface{}{ + "id": lastExec.Id, + }}) + return reExecute(ctx, lastExec.Id) + } + return execute(ctx, conn, lastExec.Id, lastExec.Status, req) } - return execute(ctx, conn, req) + return execute(ctx, conn, 0, nil, req) } // flowContentEqual 判断两次工作流参数是否一致（JSON 序列化后字节比对。 @@ -269,48 +275,42 @@ func flowContentEqual(a, b *entity.FlowInfo) bool { } // execute 执行工作流（首次执行；同会话+同工作流最近一次执行为失败状态时复用该记录重新执行，不新建数据） -func execute(ctx context.Context, conn *wsCommon.WsConnection, req *sessionDto.WebSocketExecWorkflowReq) (id int64, err error) { +func execute(ctx context.Context, conn *wsCommon.WsConnection, execId int64, status flow.FlowExecutionStatus, req *sessionDto.WebSocketExecWorkflowReq) (id int64, err error) { var nodeGroupId = uuid.NewString() - - // 复用失败记录：查询会话+工作流最近一次执行，若为失败状态则复用同一条记录（更新状态+节点组+本次请求参数）， - // 全新执行（forceNewRun，参数取本次请求），避免重跑新建数据；查询出错按无记录处理走新建 - lastExec, qErr := sessionDao.ExecWorkflowDao.GetLatestBySessionAndFlow(ctx, conn.SessionId, req.FlowId) - if qErr != nil { - glog.Errorf(ctx, "查询最近工作流执行记录失败: %v", qErr) - lastExec = nil - } - if lastExec != nil && *lastExec.Status == *flow.FlowExecutionStatusFailed.Code() { - id = lastExec.Id - _, err = sessionDao.ExecWorkflowDao.Update(ctx, &sessionDto.UpdateWorkflowReq{ - Id: id, - NodeGroupId: nodeGroupId, - Status: flow.FlowExecutionStatusRunning.Code(), - RequestParams: req.FlowContent, - }) - if err != nil { - return - } - } else { - id, err = sessionDao.ExecWorkflowDao.Insert(ctx, &sessionDto.CreateWorkflowReq{ + if g.IsEmpty(execId) { + execId, err = sessionDao.ExecWorkflowDao.Insert(ctx, &sessionDto.CreateWorkflowReq{ SessionId: conn.SessionId, FlowId: req.FlowId, NodeGroupId: nodeGroupId, Status: flow.FlowExecutionStatusRunning.Code(), RequestParams: req.FlowContent, }) - if err != nil { + if err != nil || g.IsEmpty(execId) { + glog.Errorf(ctx, "工作流执行记录创建失败: %v", err) return } + } else { + if status == flow.FlowExecutionStatusFailed.Code() { + _, err = sessionDao.ExecWorkflowDao.Update(ctx, &sessionDto.UpdateWorkflowReq{ + Id: execId, + NodeGroupId: nodeGroupId, + Status: flow.FlowExecutionStatusRunning.Code(), + RequestParams: req.FlowContent, + }) + if err != nil { + return + } + } } _ = writeJSON(conn, &wsCommon.WsPushMsg{Type: "round_start", Message: "运行开始", Data: map[string]interface{}{ - "id": id, + "id": execId, }}) - err = BuildExecution(ctx, true, req.FlowId, id, nodeGroupId, conn.SessionId, req.FlowContent) + err = BuildExecution(ctx, true, req.FlowId, execId, nodeGroupId, conn.SessionId, req.FlowContent) if err != nil { return } - return id, nil + return execId, nil } // reExecute 重新执行工作流 diff --git a/workflow/service/flow/lambda_node_util.go b/workflow/service/flow/lambda_node_util.go index aef3001..5310c09 100644 --- a/workflow/service/flow/lambda_node_util.go +++ b/workflow/service/flow/lambda_node_util.go @@ -10,6 +10,7 @@ import ( "regexp" "strings" "sync" + "unicode/utf8" commonHttp "gitea.redpowerfuture.com/red-future/common/http" "gitea.redpowerfuture.com/red-future/common/utils" @@ -297,55 +298,56 @@ func prependFilePathPrefix(prefix string, v any) any { } } +// punctRe 切分/剥离用的中文标点（含顿号、） +var punctRe = regexp.MustCompile(`[，。；！？、]`) + // BuildSubtitles 核心工具：单个sentence生成多条subtitle func BuildSubtitles(sents *[]flowDto.Sentence) ([]flowDto.Subtitle, error) { var subtitles []flowDto.Subtitle for _, sent := range *sents { - // 1. 先按标点把文本拆成多个片段（保留标点） + // 1. 先按标点把文本拆成多个片段 segList := splitTextByPunct(sent.Text) if len(segList) == 0 { continue } - wordIdx := 0 - allWords := sent.Words - // 2. 遍历每个文本片段，匹配对应的Words + // 去标点后得到纯净片段（纯空白/纯标点片段跳过） + var cleans []string for _, seg := range segList { - // 去除文本片段的标点，方便和Word.Word拼接内容匹配 - segClean := strings.ReplaceAll(seg, "，", "") - segClean = strings.ReplaceAll(segClean, "。", "") - segClean = strings.ReplaceAll(segClean, "；", "") - segClean = strings.ReplaceAll(segClean, "！", "") - segClean = strings.ReplaceAll(segClean, "？", "") - - var collectWords []flowDto.Word - var currentText strings.Builder - - // 收集Word直到拼接内容覆盖当前分段 - for wordIdx < len(allWords) { - word := allWords[wordIdx] - currentText.WriteString(word.Word) - collectWords = append(collectWords, word) - wordIdx++ - - // 当拼接的文本包含当前分段的纯文本时，停止收集 - if strings.Contains(currentText.String(), segClean) { - break - } + c := strings.TrimSpace(cleanPunct(seg)) + if c != "" { + cleans = append(cleans, c) } + } + if len(cleans) == 0 || len(sent.Words) == 0 { + continue + } - if len(collectWords) == 0 { + // 2. 词级文本与句子文本一致时，按词精确对齐取首尾词时间（最准） + if spans, ok := alignAllSegments(sent.Words, cleans); ok { + for i, span := range spans { + subtitles = append(subtitles, flowDto.Subtitle{ + Start: sent.Words[span[0]].StartTime, + End: sent.Words[span[1]].EndTime, + Text: cleans[i], + }) + } + continue + } + + // 3. ASR 词级转写与句子文本不一致时（如 血→谑、数字写法不一）， + // 整句回退为按片段字符占比分配时间，避免整句被吞成一条字幕 + segWords := allocWordsByProportion(sent.Words, cleans) + for i, ws := range segWords { + if len(ws) == 0 { continue } - - // 3. 生成字幕（时间戳取首尾Word的时间） - sub := flowDto.Subtitle{ - Start: collectWords[0].StartTime, - End: collectWords[len(collectWords)-1].EndTime, - Text: segClean, - } - subtitles = append(subtitles, sub) + subtitles = append(subtitles, flowDto.Subtitle{ + Start: ws[0].StartTime, + End: ws[len(ws)-1].EndTime, + Text: cleans[i], + }) } } @@ -357,9 +359,7 @@ func BuildSubtitles(sents *[]flowDto.Sentence) ([]flowDto.Subtitle, error) { // 会变成：["这个叫高血压调理方，", "注意是根源调理不是临时缓解，"] func splitTextByPunct(raw string) []string { // 匹配中文标点并保留在文本中，按标点位置切分 - re := regexp.MustCompile(`[，。；！？]`) - // 先找到所有标点的位置 - indexes := re.FindAllStringIndex(raw, -1) + indexes := punctRe.FindAllStringIndex(raw, -1) if len(indexes) == 0 { return []string{raw} } @@ -378,3 +378,80 @@ func splitTextByPunct(raw string) []string { } return res } + +// cleanPunct 去掉中文标点，得到纯净文本 +func cleanPunct(raw string) string { + return punctRe.ReplaceAllString(raw, "") +} + +// alignAllSegments 按顺序把各纯净片段与词级文本逐字符对齐（允许个别字符不一致）。 +// 全部片段对齐成功且词被完整覆盖时返回各片段对应的词区间，否则 ok=false， +// 由调用方回退到时间占比分配。 +func alignAllSegments(words []flowDto.Word, cleans []string) ([][2]int, bool) { + spans := make([][2]int, len(cleans)) + wordIdx := 0 + for i, seg := range cleans { + start := wordIdx + segRunes := []rune(seg) + s := 0 + for wordIdx < len(words) && s < len(segRunes) { + for _, r := range []rune(words[wordIdx].Word) { + if s < len(segRunes) && r == segRunes[s] { + s++ + } + } + wordIdx++ + } + // 片段文本没被完整匹配，或该片段没吃到任何词 → 无法精确对齐 + if s < len(segRunes) || start == wordIdx { + return nil, false + } + spans[i] = [2]int{start, wordIdx - 1} + } + // 有剩余词未被任何片段覆盖，说明对齐失败，避免吞掉剩余时间 + if wordIdx < len(words) { + return nil, false + } + return spans, true +} + +// allocWordsByProportion 按纯净片段字符占比把整句时间区间切成段，再按时间中点把 +// 每个 word 归属到所属片段（对词级转写与句子文本不一致的情况兜底）。 +func allocWordsByProportion(words []flowDto.Word, cleans []string) [][]flowDto.Word { + runes := make([]int, len(cleans)) + totalChars := 0 + for i, c := range cleans { + runes[i] = utf8.RuneCountInString(c) + totalChars += runes[i] + } + + sentStart := words[0].StartTime + sentEnd := words[len(words)-1].EndTime + duration := sentEnd - sentStart + if duration < 0 { + duration = 0 + } + + bounds := make([]float64, len(cleans)+1) + bounds[0] = sentStart + accum := 0.0 + for i := range cleans { + if totalChars > 0 { + accum += float64(runes[i]) / float64(totalChars) + } + bounds[i+1] = sentStart + accum*duration + } + + segWords := make([][]flowDto.Word, len(cleans)) + for _, w := range words { + mid := (w.StartTime + w.EndTime) / 2 + idx := 0 + for b := 0; b < len(bounds)-1; b++ { + if mid >= bounds[b+1] { + idx = b + 1 + } + } + segWords[idx] = append(segWords[idx], w) + } + return segWords +} diff --git a/workflow/service/flow/lambda_script_transcribe.go b/workflow/service/flow/lambda_script_transcribe.go index 08e0dfb..aee5ca4 100644 --- a/workflow/service/flow/lambda_script_transcribe.go +++ b/workflow/service/flow/lambda_script_transcribe.go @@ -108,6 +108,10 @@ func ScriptTranscribeLambda(ctx context.Context, input any) (any, error) { } else { systemPrompt += shotDurationConstraintPrompt(maxSeg) } + // 静音模式硬约束：从转写源头杜绝对白/旁白/开口说话，后续清洗只做兜底 + if noSpeech { + systemPrompt += noSpeechSystemPromptConstraint() + } info, err := gateway.GetModelInfoById(ctx, &gateway.GetModelInfoByIdReq{ModelId: nodeInput.Config.ModelConfig.ModelId}) if err != nil { @@ -148,6 +152,8 @@ func ScriptTranscribeLambda(ctx context.Context, input any) (any, error) { for i := range shots { shots[i].Dialogue = "" shots[i].Narration = "" + // event 里的说话动词仍会经 事件：%s 块写进分段 prompt，导致视频模型生成口型/字幕，需确定性清洗 + shots[i].Event = cleanSpeechVerbs(shots[i].Event) } } @@ -164,6 +170,7 @@ func ScriptTranscribeLambda(ctx context.Context, input any) (any, error) { FlatRefs: refsItem, Seed: nodeInput.Global.ExecutionId % 1000000, NegativePrompt: nodeInput.Config.NegativePrompt, + NoSpeech: noSpeech, } data, err := processor.Call(ctx, "split_shots_pipeline", gconv.Map(args)) if err != nil { @@ -320,6 +327,32 @@ func splitDialogueNarration(s string) (dialogue, narration string) { } } +// cleanSpeechVerbs 静音模式下清洗 event 中的说话动词：把常见说话/喊叫/对白表达替换为空串， +// 避免"事件：%s"块里残留的说话动词让视频模型生成口型/字幕。仅做机械兜底，硬约束在转写提示词。 +// NewReplacer 按最长匹配替换，故先列含"说/喊"的非开口语义词做保护（no-op，如"说明""呐喊"）， +// 再列开口表达；"叫"语义多变（呼叫/叫停/叫住），不做裸清洗以免误伤。 +func cleanSpeechVerbs(s string) string { + if s == "" { + return "" + } + repl := strings.NewReplacer( + "说明", "说明", "解说", "解说", "据说", "据说", "传说", "传说", "小说", "小说", + "学说", "学说", "说法", "说法", "说服", "说服", "呐喊", "呐喊", + "开口说话", "", "开口说", "", "开口", "", + "说道：", "", "说道:", "", "说道", "", + "说着", "", "说话", "", "讲话", "", "台词", "", "对白", "", + "喊道：", "", "喊道:", "", "喊道", "", "大喊", "", "喊叫", "", + "叫道：", "", "叫道:", "", "叫道", "", "叫到", "", "叫喊", "", + "回答", "", "答道", "", "回应", "", "回话", "", + "问道：", "", "问道:", "", "问道", "", + "念叨", "", "嘟囔", "", "嘀咕", "", "自言自语", "", + "呼唤", "", "呼叫", "", "叫唤", "", "惊叫", "", "惨叫", "", + "说：", "", "说:", "", "说", "", + "喊：", "", "喊:", "", "喊", "", + ) + return strings.TrimSpace(repl.Replace(s)) +} + // splitList 按常见分隔符拆分人名/道具列表（兼容中英文顿号、逗号、分号、"和""及"等）。 func splitList(s string) []string { repl := strings.NewReplacer("、", "|", "，", "|", ",", "|", "；", "|", ";", "|", "和", "|", "及", "|", "&", "|", "/", "|", " ", "|") @@ -339,3 +372,13 @@ func shotDurationConstraintPrompt(maxSeg int) string { } return fmt.Sprintf("\n\n单个镜头时长不超过 %d 秒：每镜的 startTime 与 endTime 之差必须 ≤ %d 秒。", maxSeg, maxSeg) } + +// noSpeechSystemPromptConstraint 静音模式的转写硬约束：要求模型从源头就不产出对白/旁白/说话动词， +// 后续 noSpeech 清洗（清空台词旁白 + cleanSpeechVerbs）只做机械兜底。 +func noSpeechSystemPromptConstraint() string { + return "\n\n本片为静音模式，镜头里禁止任何声音类内容：\n" + + "- 所有镜头禁止出现台词、旁白、画外音，narration 与 dialogue 一律留空、不要输出；\n" + + "- 禁止角色开口说话，事件描述只能写无声的动作、表情、神态、场景变化，不要出现“说”“喊”“叫”“对白”“讲话”“开口”“问”“回答”“念叨”等说话类动词；\n" + + "- characters 只是出镜角色名，不代表开口说话；\n" + + "- 视频不包含口型动作与字幕，据此调整分镜描写。" +} diff --git a/workflow/service/flow/processor/builtin/split_shots_pipeline/pipeline/pipeline.go b/workflow/service/flow/processor/builtin/split_shots_pipeline/pipeline/pipeline.go index d20c641..11f9ad3 100644 --- a/workflow/service/flow/processor/builtin/split_shots_pipeline/pipeline/pipeline.go +++ b/workflow/service/flow/processor/builtin/split_shots_pipeline/pipeline/pipeline.go @@ -18,6 +18,7 @@ type Input struct { Refs Refs // 参考素材（角色/场景/道具/产品，具名） Seed int64 // 随机种子基数，各段 = Seed + 段序号 NegativePrompt string // 全局负面 prompt（单段可覆盖，见 §7.5） + NoSpeech bool // 静音模式：段级 prompt 追加静音硬约束（视频模型不产生口播/字幕/口型） Cfg Config // 阈值/语速/容差等统一配置，零值取 DefaultConfig() TokenCfg TokenConfig // 实体名替换 token 的生成配置 } diff --git a/workflow/service/flow/processor/builtin/split_shots_pipeline/pipeline/prompt.go b/workflow/service/flow/processor/builtin/split_shots_pipeline/pipeline/prompt.go index e75cdf7..701a1d1 100644 --- a/workflow/service/flow/processor/builtin/split_shots_pipeline/pipeline/prompt.go +++ b/workflow/service/flow/processor/builtin/split_shots_pipeline/pipeline/prompt.go @@ -421,6 +421,11 @@ func BuildSegmentPrompt(segShots []Shot, reg *TokenRegistry, in Input) (string, } prompt = truncatePrompt(prompt, cfg) + // 静音模式段级硬约束：放在 truncate 之后追加，避免被截断丢弃； + // 明确告知视频模型本段是无声画面，从 prompt 层面杜绝口播/字幕/口型 + if in.NoSpeech { + prompt += "\n\n静音模式：本段为无声画面，禁止人物开口说话、禁止出现字幕与口型动作，只保留纯画面动作、表情、神态与场景变化。" + } return prompt, segRefs } diff --git a/workflow/service/flow/processor/builtin/split_shots_pipeline/split_shots_pipeline.go b/workflow/service/flow/processor/builtin/split_shots_pipeline/split_shots_pipeline.go index 42ad745..18a5d26 100644 --- a/workflow/service/flow/processor/builtin/split_shots_pipeline/split_shots_pipeline.go +++ b/workflow/service/flow/processor/builtin/split_shots_pipeline/split_shots_pipeline.go @@ -35,6 +35,7 @@ type SplitShotsInput struct { FlatRefs []pipeline.RefItem `json:"flat_refs"` // 参考素材（平铺形态，类别由 categorizeRefs 推断） Seed int64 `json:"seed"` // 随机种子基数，各段 = baseSeed + 段序号 NegativePrompt string `json:"negative_prompt,omitempty"` + NoSpeech bool `json:"no_speech,omitempty"` // 静音模式：透传 pipeline，段级 prompt 追加静音硬约束 } // SplitShotsPipelineProcessor 新拆段前置处理器。入参 args 即模型请求参数（SplitShotsInput 形状）， @@ -63,6 +64,7 @@ func SplitShotsPipelineProcessor() *processor.Processor { Refs: refs, Seed: input.Seed, NegativePrompt: input.NegativePrompt, + NoSpeech: input.NoSpeech, TokenCfg: pipeline.TokenConfig{}, })) if err != nil {