feat: 新增视频段级结果表及 DAO（flow_segment_result）

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
2026-08-25 10:01:48 +08:00
parent ac7936829f
commit 035c467a62
5 changed files with 196 additions and 0 deletions
@@ -847,3 +847,43 @@ COMMENT ON COLUMN black_deacon_exec_workflow_result.flow_id IS '工作流ID';
 COMMENT ON COLUMN black_deacon_exec_workflow_result.exec_id IS '执行ID';
 COMMENT ON COLUMN black_deacon_exec_workflow_result.result_file_url IS '结果文件路径（OSS）';
 --------------------pgsql创建black_deacon_exec_workflow_result表语句---------------------------
+--------------------pgsql创建black_deacon_flow_segment_result表语句---------------------------
+-- 视频节点段级生成结果表（只存成功段，断点续跑复用；失败段不落库视为需重新生成）
+CREATE TABLE IF NOT EXISTS black_deacon_flow_segment_result (
+    -- 基础字段（完全对齐项目规范）
+    id BIGINT PRIMARY KEY,
+    tenant_id BIGINT NOT NULL DEFAULT 0,
+    creator VARCHAR(64) NOT NULL,
+    created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
+    updater VARCHAR(64) NOT NULL,
+    updated_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
+    deleted_at timestamp(6),
+
+    -- 业务字段
+    execution_id BIGINT NOT NULL DEFAULT 0, -- 执行ID（exec_workflow；reExecute 复用同一条）
+    node_id VARCHAR(64) NOT NULL DEFAULT '', -- 视频生成节点ID
+    segment_index INT NOT NULL DEFAULT 0, -- 段序号
+    video_key VARCHAR(128) NOT NULL DEFAULT '', -- 视频输出字段key（重建输出记录保 key 一致，避免下游引用失配）
+    video_url VARCHAR(512) NOT NULL DEFAULT '' -- 已生成成功的视频地址
+);
+
+-- 唯一键 + 高频查询索引（ListByNode 命中 (execution_id, node_id) 前缀）
+CREATE UNIQUE INDEX IF NOT EXISTS uk_segment_result_exec_node_idx ON black_deacon_flow_segment_result(execution_id, node_id, segment_index);
+CREATE INDEX IF NOT EXISTS idx_segment_result_tenant_id ON black_deacon_flow_segment_result(tenant_id);
+CREATE INDEX IF NOT EXISTS idx_segment_result_deleted_at ON black_deacon_flow_segment_result(deleted_at);
+
+-- 表和字段注释
+COMMENT ON TABLE black_deacon_flow_segment_result IS '视频节点段级生成结果表';
+COMMENT ON COLUMN black_deacon_flow_segment_result.id IS '主键ID';
+COMMENT ON COLUMN black_deacon_flow_segment_result.tenant_id IS '租户ID';
+COMMENT ON COLUMN black_deacon_flow_segment_result.creator IS '创建人';
+COMMENT ON COLUMN black_deacon_flow_segment_result.created_at IS '创建时间';
+COMMENT ON COLUMN black_deacon_flow_segment_result.updater IS '更新人';
+COMMENT ON COLUMN black_deacon_flow_segment_result.updated_at IS '更新时间';
+COMMENT ON COLUMN black_deacon_flow_segment_result.deleted_at IS '删除时间';
+COMMENT ON COLUMN black_deacon_flow_segment_result.execution_id IS '执行ID';
+COMMENT ON COLUMN black_deacon_flow_segment_result.node_id IS '视频生成节点ID';
+COMMENT ON COLUMN black_deacon_flow_segment_result.segment_index IS '段序号';
+COMMENT ON COLUMN black_deacon_flow_segment_result.video_key IS '视频输出字段key';
+COMMENT ON COLUMN black_deacon_flow_segment_result.video_url IS '已生成成功的视频地址';
+--------------------pgsql创建black_deacon_flow_segment_result表语句---------------------------
@@ -22,4 +22,5 @@ const (
 	TableNameExecChat           = "exec_chat"
 	TableNameExecWorkflow       = "exec_workflow"
 	TableNameExecWorkflowResult = "exec_workflow_result"
+	TableNameFlowSegmentResult  = "black_deacon_flow_segment_result"
 )
@@ -0,0 +1,57 @@
+package flow
+
+import (
+	"ai-agent/workflow/consts/public"
+	"ai-agent/workflow/model/entity"
+	"context"
+
+	"gitea.redpowerfuture.com/red-future/common/db/gfdb"
+)
+
+var FlowSegmentResultDao = &flowSegmentResultDao{}
+
+type flowSegmentResultDao struct{}
+
+// Save 段成功后落库：唯一键 (execution_id, node_id, segment_index)，冲突则更新
+func (d *flowSegmentResultDao) Save(ctx context.Context, execId int64, nodeId string, segmentIndex int, videoKey, videoURL string) error {
+	rec := &entity.FlowSegmentResult{
+		ExecutionId:  execId,
+		NodeId:       nodeId,
+		SegmentIndex: segmentIndex,
+		VideoKey:     videoKey,
+		VideoURL:     videoURL,
+	}
+	_, err := gfdb.DB(ctx, public.DbNameBlackDeacon).
+		Model(ctx, public.TableNameFlowSegmentResult).
+		Data(rec).
+		OnConflict(entity.FlowSegmentResultCol.ExecutionId, entity.FlowSegmentResultCol.NodeId, entity.FlowSegmentResultCol.SegmentIndex).
+		Save()
+	return err
+}
+
+// ListByNode 返回该节点已成功段（段序号 → 视频引用）
+func (d *flowSegmentResultDao) ListByNode(ctx context.Context, execId int64, nodeId string) (map[int]entity.SegmentRef, error) {
+	var list []*entity.FlowSegmentResult
+	err := gfdb.DB(ctx, public.DbNameBlackDeacon).
+		Model(ctx, public.TableNameFlowSegmentResult).
+		Where(entity.FlowSegmentResultCol.ExecutionId, execId).
+		Where(entity.FlowSegmentResultCol.NodeId, nodeId).
+		Scan(&list)
+	if err != nil {
+		return nil, err
+	}
+	m := make(map[int]entity.SegmentRef, len(list))
+	for _, r := range list {
+		m[r.SegmentIndex] = entity.SegmentRef{Key: r.VideoKey, URL: r.VideoURL}
+	}
+	return m, nil
+}
+
+// DeleteByExecution 清理指定执行的段结果（全新执行前清旧段 / 工作流执行成功后清理）
+func (d *flowSegmentResultDao) DeleteByExecution(ctx context.Context, execId int64) error {
+	_, err := gfdb.DB(ctx, public.DbNameBlackDeacon).
+		Model(ctx, public.TableNameFlowSegmentResult).
+		Where(entity.FlowSegmentResultCol.ExecutionId, execId).
+		Delete()
+	return err
+}
@@ -0,0 +1,60 @@
+package flow
+
+import (
+	"context"
+	"os"
+	"testing"
+	"time"
+
+	"gitea.redpowerfuture.com/red-future/common/beans"
+)
+
+// TestFlowSegmentResultDao 段结果 DAO 集成测试：Save / ListByNode / DeleteByExecution。
+// 依赖本地 black_deacon + 用户信息注入；默认跳过，设置 AI_AGENT_TEST_DB=1 时运行。
+func TestFlowSegmentResultDao(t *testing.T) {
+	if os.Getenv("AI_AGENT_TEST_DB") == "" {
+		t.Skip("设置 AI_AGENT_TEST_DB=1 且本地 black_deacon 可用时运行")
+	}
+	// gfdb Insert/Select 钩子从 ctx 读用户信息补 tenant_id/creator 并追加租户过滤
+	ctx := context.WithValue(context.Background(), "user", &beans.User{UserName: "unit_test", TenantId: 999999})
+	execId := time.Now().UnixMilli()
+	nodeId := "seg_test_node"
+
+	dao := FlowSegmentResultDao
+	if err := dao.Save(ctx, execId, nodeId, 1, "video_url", "http://host/u1.mp4"); err != nil {
+		t.Fatalf("Save 段1失败: %v", err)
+	}
+	if err := dao.Save(ctx, execId, nodeId, 2, "video_url", "http://host/u2.mp4"); err != nil {
+		t.Fatalf("Save 段2失败: %v", err)
+	}
+
+	// 唯一键冲突 → 更新不报错
+	if err := dao.Save(ctx, execId, nodeId, 2, "video_url", "http://host/u2-new.mp4"); err != nil {
+		t.Fatalf("Save 段2(冲突更新)失败: %v", err)
+	}
+
+	m, err := dao.ListByNode(ctx, execId, nodeId)
+	if err != nil {
+		t.Fatalf("ListByNode失败: %v", err)
+	}
+	if len(m) != 2 {
+		t.Fatalf("期望 2 段，实际 %d", len(m))
+	}
+	if m[1].URL != "http://host/u1.mp4" || m[2].URL != "http://host/u2-new.mp4" {
+		t.Fatalf("段内容不符: %+v", m)
+	}
+	if m[2].Key != "video_url" {
+		t.Fatalf("段2 key 不符: %s", m[2].Key)
+	}
+
+	if err := dao.DeleteByExecution(ctx, execId); err != nil {
+		t.Fatalf("DeleteByExecution失败: %v", err)
+	}
+	m, err = dao.ListByNode(ctx, execId, nodeId)
+	if err != nil {
+		t.Fatalf("删除后 ListByNode失败: %v", err)
+	}
+	if len(m) != 0 {
+		t.Fatalf("删除后应无段结果，实际 %d", len(m))
+	}
+}
@@ -0,0 +1,38 @@
+package entity
+
+import "gitea.redpowerfuture.com/red-future/common/beans"
+
+// FlowSegmentResult 视频节点段级生成结果（只存成功段），供断点续跑复用
+type FlowSegmentResult struct {
+	beans.SQLBaseDO `orm:",inherit"` // 嵌入基础字段：Id, TenantId, Creator, CreatedAt, Updater, UpdatedAt, DeletedAt
+	// 业务字段
+	ExecutionId  int64  `orm:"execution_id" json:"executionId" description:"执行ID"`
+	NodeId       string `orm:"node_id" json:"nodeId" description:"视频生成节点ID"`
+	SegmentIndex int    `orm:"segment_index" json:"segmentIndex" description:"段序号"`
+	VideoKey     string `orm:"video_key" json:"videoKey" description:"视频输出字段key"`
+	VideoURL     string `orm:"video_url" json:"videoURL" description:"已生成成功的视频地址"`
+}
+
+// SegmentRef 一段已成功生成的视频引用（Key 保持模型原输出字段名，重建输出记录保证下游引用一致）
+type SegmentRef struct {
+	Key string
+	URL string
+}
+
+type flowSegmentResultCol struct {
+	beans.SQLBaseCol
+	ExecutionId  string
+	NodeId       string
+	SegmentIndex string
+	VideoKey     string
+	VideoURL     string
+}
+
+var FlowSegmentResultCol = flowSegmentResultCol{
+	SQLBaseCol:   beans.DefSQLBaseCol,
+	ExecutionId:  "execution_id",
+	NodeId:       "node_id",
+	SegmentIndex: "segment_index",
+	VideoKey:     "video_key",
+	VideoURL:     "video_url",
+}