From 035c467a62d02869448b8617b19ec3a3134efc71 Mon Sep 17 00:00:00 2001 From: qhd <1766646056@qq.com> Date: Tue, 25 Aug 2026 10:01:48 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20=E6=96=B0=E5=A2=9E=E8=A7=86=E9=A2=91?= =?UTF-8?q?=E6=AE=B5=E7=BA=A7=E7=BB=93=E6=9E=9C=E8=A1=A8=E5=8F=8A=20DAO?= =?UTF-8?q?=EF=BC=88flow=5Fsegment=5Fresult=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-Authored-By: Claude Opus 4.7 --- update.sql | 40 +++++++++++++ workflow/consts/public/table_name.go | 1 + workflow/dao/flow/flow_segment_result_dao.go | 57 ++++++++++++++++++ .../dao/flow/flow_segment_result_dao_test.go | 60 +++++++++++++++++++ workflow/model/entity/flow_segment_result.go | 38 ++++++++++++ 5 files changed, 196 insertions(+) create mode 100644 workflow/dao/flow/flow_segment_result_dao.go create mode 100644 workflow/dao/flow/flow_segment_result_dao_test.go create mode 100644 workflow/model/entity/flow_segment_result.go diff --git a/update.sql b/update.sql index 690e849..46b2c08 100644 --- a/update.sql +++ b/update.sql @@ -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表语句--------------------------- diff --git a/workflow/consts/public/table_name.go b/workflow/consts/public/table_name.go index 2665069..15e6bff 100644 --- a/workflow/consts/public/table_name.go +++ b/workflow/consts/public/table_name.go @@ -22,4 +22,5 @@ const ( TableNameExecChat = "exec_chat" TableNameExecWorkflow = "exec_workflow" TableNameExecWorkflowResult = "exec_workflow_result" + TableNameFlowSegmentResult = "black_deacon_flow_segment_result" ) diff --git a/workflow/dao/flow/flow_segment_result_dao.go b/workflow/dao/flow/flow_segment_result_dao.go new file mode 100644 index 0000000..11fd2ed --- /dev/null +++ b/workflow/dao/flow/flow_segment_result_dao.go @@ -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 +} diff --git a/workflow/dao/flow/flow_segment_result_dao_test.go b/workflow/dao/flow/flow_segment_result_dao_test.go new file mode 100644 index 0000000..5320461 --- /dev/null +++ b/workflow/dao/flow/flow_segment_result_dao_test.go @@ -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)) + } +} diff --git a/workflow/model/entity/flow_segment_result.go b/workflow/model/entity/flow_segment_result.go new file mode 100644 index 0000000..490e132 --- /dev/null +++ b/workflow/model/entity/flow_segment_result.go @@ -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", +}