feat: 视频段级续跑纯逻辑（复用决策+按序合并）及单测

2026-08-25 10:16:54 +08:00
parent 0de29988b4
commit 2d7f6888a6
3 changed files with 167 additions and 0 deletions
@@ -0,0 +1,55 @@
+package flow
+
+import (
+	"sort"
+
+	"ai-agent/workflow/model/entity"
+
+	"github.com/gogf/gf/v2/util/gconv"
+)
+
+// segmentGenerateMaxAttempts 视频段生成最大尝试次数（失败自动重试 1 次，共 2 次尝试），参数化可调
+const segmentGenerateMaxAttempts = 2
+
+// planSegmentResume 段级续跑决策：把 paramsList 各段映射到"是否需重新生成"。
+// savedMap 为该节点已成功段（段序号 → {key,url}）；段在表中缺失或地址为空则需重新生成。
+// 返回值与 paramsList 对齐。全新执行（savedMap 为 nil/空）时全部需生成。
+func planSegmentResume(paramsList []map[string]any, savedMap map[int]entity.SegmentRef) (idxList []int, needGen []bool) {
+	idxList = make([]int, len(paramsList))
+	needGen = make([]bool, len(paramsList))
+	for i, params := range paramsList {
+		idx := gconv.Int(params["segment_index"])
+		idxList[i] = idx
+		ref, ok := savedMap[idx]
+		needGen[i] = !ok || ref.URL == ""
+	}
+	return
+}
+
+// mergeSegmentOutputs 把复用段与新生段按段序号升序合并为 concat 输入列表（列表顺序即拼接顺序）。
+// 复用段重建 {key:url} 记录（key 保持模型原输出字段，避免下游引用失配）；新生段沿用模型原输出。
+// 任一段既无复用又无生成结果（生成空）则跳过——与现有"空段贡献空"行为一致，最终由 concat 校验兜底。
+func mergeSegmentOutputs(idxList []int, needGen []bool, newRes [][]map[string]any, savedMap map[int]entity.SegmentRef) []map[string]any {
+	type segOutput struct {
+		idx  int
+		recs []map[string]any
+	}
+	out := make([]segOutput, 0, len(idxList))
+	for i, idx := range idxList {
+		if needGen[i] {
+			if len(newRes[i]) > 0 {
+				out = append(out, segOutput{idx: idx, recs: newRes[i]})
+			}
+			continue
+		}
+		if ref, ok := savedMap[idx]; ok && ref.URL != "" {
+			out = append(out, segOutput{idx: idx, recs: []map[string]any{{ref.Key: ref.URL}}})
+		}
+	}
+	sort.SliceStable(out, func(a, b int) bool { return out[a].idx < out[b].idx })
+	var merged []map[string]any
+	for _, o := range out {
+		merged = append(merged, o.recs...)
+	}
+	return merged
+}
@@ -0,0 +1,102 @@
+package flow
+
+import (
+	"testing"
+
+	"ai-agent/workflow/model/entity"
+)
+
+// planSegmentResume：段序号从 params 的 segment_index 读出；已成功段复用，缺失/空地址需生成
+func TestPlanSegmentResume(t *testing.T) {
+	paramsList := []map[string]any{
+		{"segment_index": 1}, {"segment_index": 2}, {"segment_index": 3}, {"segment_index": 4}, {"segment_index": 5},
+	}
+	saved := map[int]entity.SegmentRef{
+		1: {Key: "video_url", URL: "u1"}, 2: {Key: "video_url", URL: "u2"},
+		3: {Key: "video_url", URL: "u3"}, 5: {Key: "video_url", URL: "u5"},
+	}
+	idxList, needGen := planSegmentResume(paramsList, saved)
+	if len(idxList) != 5 || len(needGen) != 5 {
+		t.Fatalf("长度不符: idxList=%v needGen=%v", idxList, needGen)
+	}
+	want := []bool{false, false, false, true, false} // 仅段4需重新生成
+	for i := range want {
+		if needGen[i] != want[i] {
+			t.Fatalf("needGen[%d]=%v 期望 %v", i, needGen[i], want[i])
+		}
+		if idxList[i] != i+1 {
+			t.Fatalf("idxList[%d]=%d 期望 %d", i, idxList[i], i+1)
+		}
+	}
+}
+
+// planSegmentResume：无已成功段（全新执行）→ 全部需生成
+func TestPlanSegmentResumeEmptySaved(t *testing.T) {
+	paramsList := []map[string]any{{"segment_index": 0}, {"segment_index": 1}}
+	_, needGen := planSegmentResume(paramsList, nil)
+	for i := range needGen {
+		if !needGen[i] {
+			t.Fatalf("needGen[%d] 期望 true（无复用）", i)
+		}
+	}
+}
+
+// mergeSegmentOutputs：复用段 + 新生段按段序号升序，顺序即拼接顺序
+func TestMergeSegmentOutputs(t *testing.T) {
+	idxList := []int{1, 2, 3, 4, 5}
+	needGen := []bool{false, false, false, true, false}
+	newRes := make([][]map[string]any, 5)
+	newRes[3] = []map[string]any{{"video_url": "new4"}}
+	saved := map[int]entity.SegmentRef{
+		1: {Key: "video_url", URL: "u1"}, 2: {Key: "video_url", URL: "u2"},
+		3: {Key: "video_url", URL: "u3"}, 5: {Key: "video_url", URL: "u5"},
+	}
+	got := mergeSegmentOutputs(idxList, needGen, newRes, saved)
+	if len(got) != 5 {
+		t.Fatalf("合并后应有 5 段，实际 %d: %v", len(got), got)
+	}
+	wantURLs := []string{"u1", "u2", "u3", "new4", "u5"}
+	for i, rec := range got {
+		if rec["video_url"] != wantURLs[i] {
+			t.Fatalf("第 %d 段=%v 期望 %s", i, rec, wantURLs[i])
+		}
+	}
+}
+
+// mergeSegmentOutputs：复用段 key 保持模型原字段，重建记录与新生记录 key 不一致也不影响顺序
+func TestMergeSegmentOutputsMixedKeys(t *testing.T) {
+	idxList := []int{1, 2, 3}
+	needGen := []bool{false, true, false}
+	newRes := make([][]map[string]any, 3)
+	newRes[1] = []map[string]any{{"video_oss_url": "new2"}}
+	saved := map[int]entity.SegmentRef{
+		1: {Key: "video_url", URL: "u1"}, 3: {Key: "file_url", URL: "u3"},
+	}
+	got := mergeSegmentOutputs(idxList, needGen, newRes, saved)
+	if len(got) != 3 {
+		t.Fatalf("合并后应有 3 段，实际 %d: %v", len(got), got)
+	}
+	if got[0]["video_url"] != "u1" || got[1]["video_oss_url"] != "new2" || got[2]["file_url"] != "u3" {
+		t.Fatalf("合并结果不符: %v", got)
+	}
+}
+
+// mergeSegmentOutputs：paramsList 乱序时仍按段序号升序输出（顺序保证）
+func TestMergeSegmentOutputsUnorderedParams(t *testing.T) {
+	idxList := []int{3, 1, 2}
+	needGen := []bool{true, false, true}
+	newRes := make([][]map[string]any, 3)
+	newRes[0] = []map[string]any{{"video_url": "new3"}}
+	newRes[2] = []map[string]any{{"video_url": "new2"}}
+	saved := map[int]entity.SegmentRef{1: {Key: "video_url", URL: "u1"}}
+	got := mergeSegmentOutputs(idxList, needGen, newRes, saved)
+	if len(got) != 3 {
+		t.Fatalf("合并后应有 3 段，实际 %d: %v", len(got), got)
+	}
+	wantURLs := []string{"u1", "new2", "new3"}
+	for i, rec := range got {
+		if rec["video_url"] != wantURLs[i] {
+			t.Fatalf("第 %d 段=%v 期望 %s（应段序升序）", i, rec, wantURLs[i])
+		}
+	}
+}
@@ -301,3 +301,13 @@ func normalizeVideoURL(ctx context.Context, url string) string {
 	}
 	return prefix + url
 }
+
+// FindVideoKey 返回视频 URL 所在字段的 key（规则同 findVideoKey），供工作流段级续跑重建输出记录保持 key 一致
+func FindVideoKey(params map[string]any) string {
+	return findVideoKey(params)
+}
+
+// FindVideoURL 从模型返回参数中提取视频 URL（规则同 findVideoURL），供段级续跑落库
+func FindVideoURL(ctx context.Context, params map[string]any) string {
+	return findVideoURL(ctx, params)
+}