red-future/ai-agent
4
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: 修复执行结果列表依赖结果表查询

Browse Source
This commit is contained in:
19904408334
2026-08-22 09:15:05 +08:00
parent 670fe61033
commit c7091e3702
5 changed files with 125 additions and 111 deletions
Download Patch File Download Diff File Expand all files Collapse all files
workflow/dao/session/exec_workflow_dao.go
-41
View File
@@ -117,44 +117,3 @@ func (d *execWorkflowDao) ListBySession(ctx context.Context, sessionId string) (
err = r.Structs(&res)
return
}
// ListDates 按创建人查询去重后的创建日期（倒序，支持分页；page 为 nil 返回全部日期）
func (d *execWorkflowDao) ListDates(ctx context.Context, creator string, page *beans.Page) (dates []string, err error) {
fieldAlias := "create_date"
m := gfdb.DB(ctx, public.DbNameBlackDeacon).Model(ctx, public.TableNameExecWorkflow).
Fields("DATE("+entity.ExecWorkflowCol.CreatedAt+") AS "+fieldAlias).
Where(entity.ExecWorkflowCol.Creator, creator).
Group("DATE(" + entity.ExecWorkflowCol.CreatedAt + ")"). // 和select表达式保持一致，按自然日分组去重
OrderDesc(fieldAlias) // 按日期别名倒序
if page != nil {
m.Page(int(page.PageNum), int(page.PageSize))
}
r, err := m.All()
if err != nil {
return nil, err
}
for _, rec := range r {
dates = append(dates, rec[fieldAlias].String())
}
return dates, nil
}
// ListByDates 按创建人查询指定创建日期（DATE(created_at) 命中）内的执行记录，按创建时间倒序
func (d *execWorkflowDao) ListByDates(ctx context.Context, creator string, dates []string) (res []*entity.ExecWorkflow, err error) {
if len(dates) == 0 {
return
}
r, err := gfdb.DB(ctx, public.DbNameBlackDeacon).Model(ctx, public.TableNameExecWorkflow).
Where(entity.ExecWorkflowCol.Creator, creator).
WhereIn("DATE(created_at)", dates).
OrderDesc(entity.ExecWorkflowCol.CreatedAt).
All()
if err != nil {
return
}
err = r.Structs(&res)
return
}
workflow/dao/session/exec_workflow_result_dao.go
+30 -5
View File
@@ -87,14 +87,39 @@ func (d *execWorkflowResultDao) ListByExecId(ctx context.Context, execId int64)
return
}
// ListByExecIds 批量查询多个执行记录下的结果文件（按创建时间正序）
func (d *execWorkflowResultDao) ListByExecIds(ctx context.Context, execIds []int64) (res []*entity.ExecWorkflowResult, err error) {
if len(execIds) == 0 {
// ListDates 按创建人查询去重后的创建日期（倒序，支持分页；page 为 nil 返回全部日期）
func (d *execWorkflowResultDao) ListDates(ctx context.Context, creator string, page *beans.Page) (dates []string, err error) {
fieldAlias := "create_date"
m := gfdb.DB(ctx, public.DbNameBlackDeacon).Model(ctx, public.TableNameExecWorkflowResult).
Fields("DATE("+entity.ExecWorkflowResultCol.CreatedAt+") AS "+fieldAlias).
Where(entity.ExecWorkflowResultCol.Creator, creator).
Group("DATE(" + entity.ExecWorkflowResultCol.CreatedAt + ")").
OrderDesc(fieldAlias)
if page != nil {
m.Page(int(page.PageNum), int(page.PageSize))
}
r, err := m.All()
if err != nil {
return nil, err
}
for _, rec := range r {
dates = append(dates, rec[fieldAlias].String())
}
return dates, nil
}
// ListByDates 按创建人查询指定创建日期（DATE(created_at) 命中）内的结果记录，按创建时间倒序
func (d *execWorkflowResultDao) ListByDates(ctx context.Context, creator string, dates []string) (res []*entity.ExecWorkflowResult, err error) {
if len(dates) == 0 {
return
}
r, err := gfdb.DB(ctx, public.DbNameBlackDeacon).Model(ctx, public.TableNameExecWorkflowResult).
WhereIn(entity.ExecWorkflowResultCol.ExecId, execIds).
OrderAsc(entity.ExecWorkflowResultCol.CreatedAt).
Where(entity.ExecWorkflowResultCol.Creator, creator).
WhereIn("DATE(created_at)", dates).
OrderDesc(entity.ExecWorkflowResultCol.CreatedAt).
All()
if err != nil {
return
workflow/service/flow/flow_ws_exec.go
+3
View File
@@ -287,10 +287,13 @@ func executeOrResume(ctx context.Context, conn *wsCommon.WsConnection, req *sess
_ = writeJSON(conn, &wsCommon.WsPushMsg{Type: "round_start", Message: "运行开始", Data: map[string]interface{}{
"id": lastExec.Id,
}})
glog.Infof(ctx, "工作流断点续跑，execId: %v", lastExec.Id)
return reExecute(ctx, lastExec.Id)
}
glog.Infof(ctx, "工作流全新执行，lastExec: %v", lastExec)
return execute(ctx, conn, lastExec.Id, lastExec.Status, req)
}
glog.Infof(ctx, "工作流全新执行，无历史记录")
return execute(ctx, conn, 0, nil, req)
}
workflow/service/flow/lambda_node.go
+29 -7
View File
@@ -359,13 +359,7 @@ func injectSubFlowFields(global *flowDto.FlowExecutionInput, subFlowContent *ent
}
value := field["value"]
if vs, has := field["valueSource"]; has && vs != nil {
vsMap := gconv.Map(vs)
vsNodeId := gconv.String(vsMap["nodeId"])
vsField := gconv.String(vsMap["fieldName"])
if vsField == "" {
vsField = gconv.String(vsMap["field"])
}
if vsNodeId != "" && vsField != "" {
if vsNodeId, vsField := firstValueSource(vs); vsNodeId != "" && vsField != "" {
if v, _, ok := resolveValueSource(global, vsNodeId, vsField); ok {
value = v
}
@@ -380,6 +374,34 @@ func injectSubFlowFields(global *flowDto.FlowExecutionInput, subFlowContent *ent
}
}
// firstValueSource 从 valueSource 提取第一个引用源 (nodeId, field)。
// 前端统一发送数组 [{nodeId, field}]（见 serializeSubFlowConfig），旧 DSL 可能是单对象
// {nodeId, fieldName}，两种形态都兼容；子流程字段与引用源一一对应，只取第一个。
func firstValueSource(vs any) (nodeId, field string) {
if vs == nil {
return
}
switch v := vs.(type) {
case []any:
if len(v) > 0 {
return firstValueSource(v[0])
}
return
case []map[string]any:
if len(v) > 0 {
return firstValueSource(v[0])
}
return
}
m := gconv.Map(vs)
nodeId = gconv.String(m["nodeId"])
field = gconv.String(m["fieldName"])
if field == "" {
field = gconv.String(m["field"])
}
return
}
// subFlowStartNode 返回工作流开始节点
func subFlowStartNode(content *entity.FlowInfo) *entity.FlowNode {
if content == nil {
workflow/service/session/session_service.go
+63 -58
View File
@@ -216,14 +216,16 @@ func workflowExecVO(w *entity.ExecWorkflow, resultFileUrl string) *sessionDto.VO
}
}
// ResultList 工作流执行结果树：按创建人分页查询工作流执行记录，返回按天分组的树结构（日期→流程→结果文件）。
// ResultList 工作流执行结果树：按创建人分页查询工作流结果记录（exec_workflow_result），
// 返回按天分组的树结构（日期→流程→结果文件）。
// 只依赖结果表，不关联 exec_workflow 执行记录——即使执行记录被删除/清理，产出文件仍可查看。
// 分页单位为"天"：每页返回 pageSize 个完整日期，同一天内的流程与文件不会被拆到不同页；pageSize 为 0 时返回全部。
func (s *sessionService) ResultList(ctx context.Context, req *sessionDto.ListWorkflowResultReq) (res *flowDto.ListFlowExecutionTreeRes, err error) {
user, err := utils.GetUserInfo(ctx)
if err != nil {
return
}
dates, err := sessionDao.ExecWorkflowDao.ListDates(ctx, user.UserName, req.Page)
dates, err := sessionDao.ExecWorkflowResultDao.ListDates(ctx, user.UserName, req.Page)
if err != nil {
return
}
@@ -233,22 +235,42 @@ func (s *sessionService) ResultList(ctx context.Context, req *sessionDto.ListWor
return
}
execs, err := sessionDao.ExecWorkflowDao.ListByDates(ctx, user.UserName, dates)
// 仅查结果表：同一执行的多个结果按 exec_id 归为一个流程节点
results, err := sessionDao.ExecWorkflowResultDao.ListByDates(ctx, user.UserName, dates)
if err != nil {
return
}
// 汇总本页所有执行的 id 与去重后的流程 id，一次取回结果文件与流程名
var execIds []int64
// 按 exec_id 分组（ListByDates 按创建时间倒序，同执行内结果最后反转回正序，保持旧列表输出顺序）
type resultGroup struct {
ExecId int64
SessionId string
FlowId int64
Date string
Results []*entity.ExecWorkflowResult
}
groupByExec := make(map[int64]*resultGroup)
var groups []*resultGroup
flowIdSet := make(map[int64]struct{})
for _, e := range execs {
execIds = append(execIds, e.Id)
flowIdSet[e.FlowId] = struct{}{}
for _, r := range results {
g := groupByExec[r.ExecId]
if g == nil {
g = &resultGroup{ExecId: r.ExecId, SessionId: r.SessionId, FlowId: r.FlowId}
groupByExec[r.ExecId] = g
groups = append(groups, g)
if r.CreatedAt != nil {
g.Date = r.CreatedAt.Format("Y-m-d")
}
}
g.Results = append(g.Results, r)
flowIdSet[r.FlowId] = struct{}{}
}
results, err := sessionDao.ExecWorkflowResultDao.ListByExecIds(ctx, execIds)
if err != nil {
return
for _, g := range groups {
for i, j := 0, len(g.Results)-1; i < j; i, j = i+1, j-1 {
g.Results[i], g.Results[j] = g.Results[j], g.Results[i]
}
}
flowNameMap := make(map[int64]string)
for fid := range flowIdSet {
if fu, e := flowDao.FlowUserDao.Get(ctx, &flowDto.GetFlowUserReq{Id: fid}); e == nil && fu != nil && fu.FlowName != "" {
@@ -256,60 +278,43 @@ func (s *sessionService) ResultList(ctx context.Context, req *sessionDto.ListWor
}
}
resultsByExec := make(map[int64][]*entity.ExecWorkflowResult)
for _, r := range results {
resultsByExec[r.ExecId] = append(resultsByExec[r.ExecId], r)
}
execsByDate := make(map[string][]*entity.ExecWorkflow)
for _, e := range execs {
date := ""
if e.CreatedAt != nil {
date = e.CreatedAt.Format("Y-m-d")
// 按日期分组构建树（flow 顺序即各组首次出现的倒序，与旧行为一致）
flowsByDate := make(map[string][]flowDto.FlowNode)
for _, g := range groups {
flowName := flowNameMap[g.FlowId]
if flowName == "" {
flowName = "工作流"
}
execsByDate[date] = append(execsByDate[date], e)
}
for _, d := range dates {
execList := execsByDate[d]
if len(execList) == 0 {
continue
}
var flows []flowDto.FlowNode
for _, e := range execList {
flowName := flowNameMap[e.FlowId]
if flowName == "" {
flowName = "工作流"
}
var items []flowDto.OutputItem
suffixCount := make(map[string]int)
for _, rf := range resultsByExec[e.Id] {
if rf.ResultFileUrl == "" {
continue
}
content := rf.ResultFileUrl
ext := flowService.GetFileTypeByPath(content)
suffix := outputItemSuffix(ext)
suffixCount[suffix]++
items = append(items, flowDto.OutputItem{
Content: content,
Type: ext,
Label: fmt.Sprintf("%s_%d", suffix, suffixCount[suffix]),
})
}
if len(items) == 0 {
var items []flowDto.OutputItem
suffixCount := make(map[string]int)
for _, rf := range g.Results {
if rf.ResultFileUrl == "" {
continue
}
flows = append(flows, flowDto.FlowNode{
FlowName: flowName,
Id: e.Id,
SessionId: e.SessionId,
Items: items,
content := rf.ResultFileUrl
ext := flowService.GetFileTypeByPath(content)
suffix := outputItemSuffix(ext)
suffixCount[suffix]++
items = append(items, flowDto.OutputItem{
Content: content,
Type: ext,
Label: fmt.Sprintf("%s_%d", suffix, suffixCount[suffix]),
})
}
if len(flows) == 0 {
if len(items) == 0 {
continue
}
res.Tree = append(res.Tree, flowDto.DateNode{CreateDate: d, Flows: flows})
flowsByDate[g.Date] = append(flowsByDate[g.Date], flowDto.FlowNode{
FlowName: flowName,
Id: g.ExecId,
SessionId: g.SessionId,
Items: items,
})
}
for _, d := range dates {
if fs := flowsByDate[d]; len(fs) > 0 {
res.Tree = append(res.Tree, flowDto.DateNode{CreateDate: d, Flows: fs})
}
}
return
}