github
/
allinssl
mirror of https://github.com/allinssl/allinssl
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

修改工作流上下文

pull/79/head^2
zhangchenhao 2025-05-09 18:01:41 +08:00
parent f8f222e740
commit 6e2fe8cf52
1 changed files with 18 additions and 17 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

35
backend/internal/workflow/workflow.go
View File

@ -4,7 +4,6 @@ import (
"ALLinSSL/backend/public" "ALLinSSL/backend/public"
"encoding/json" "encoding/json"
"fmt" "fmt"
"strings"
"sync" "sync"
"time" "time"
) )
@ -27,7 +26,7 @@ func GetList(search string, p, limit int64) ([]map[string]any, int, error) {
return data, 0, err return data, 0, err
} }
defer s.Close() defer s.Close()
var limits []int64 var limits []int64
if p >= 0 && limit >= 0 { if p >= 0 && limit >= 0 {
limits = []int64{0, limit} limits = []int64{0, limit}
@ -36,7 +35,7 @@ func GetList(search string, p, limit int64) ([]map[string]any, int, error) {
limits[1] = p * limit limits[1] = p * limit
} }
} }
if search != "" { if search != "" {
count, err = s.Where("name like ?", []interface{}{"%" + search + "%"}).Count() count, err = s.Where("name like ?", []interface{}{"%" + search + "%"}).Count()
data, err = s.Where("name like ?", []interface{}{"%" + search + "%"}).Order("update_time", "desc").Limit(limits).Select() data, err = s.Where("name like ?", []interface{}{"%" + search + "%"}).Order("update_time", "desc").Limit(limits).Select()
@ -56,7 +55,7 @@ func AddWorkflow(name, content, execType, active, execTime string) error {
if err != nil { if err != nil {
return fmt.Errorf("检测到工作流配置有问题：%v", err) return fmt.Errorf("检测到工作流配置有问题：%v", err)
} }
s, err := GetSqlite() s, err := GetSqlite()
if err != nil { if err != nil {
return err return err
@ -161,7 +160,7 @@ func ExecuteWorkflow(id string) error {
return fmt.Errorf("工作流正在执行中") return fmt.Errorf("工作流正在执行中")
} }
content := data[0]["content"].(string) content := data[0]["content"].(string)
go func(id, c string) { go func(id, c string) {
// defer wg.Done() // defer wg.Done()
// WorkflowID := strconv.FormatInt(id, 10) // WorkflowID := strconv.FormatInt(id, 10)
@ -192,13 +191,15 @@ func resolveInputs(inputs []WorkflowNodeParams, ctx *ExecutionContext) map[strin
for _, input := range inputs { for _, input := range inputs {
if input.FromNodeID != "" { if input.FromNodeID != "" {
if val, ok := ctx.GetOutput(input.FromNodeID); ok { if val, ok := ctx.GetOutput(input.FromNodeID); ok {
switch strings.Split(strings.TrimPrefix(input.FromNodeID, "-"), "-")[0] { // 暂时没有新的类型可以先写死
case "apply": // switch strings.Split(strings.TrimPrefix(input.FromNodeID, "-"), "-")[0] {
input.Name = "certificate" // case "apply":
case "upload": // input.Name = "certificate"
input.Name = "certificate" // case "upload":
} // input.Name = "certificate"
resolved[input.Name] = val // }
// resolved[input.Name] = val
resolved["certificate"] = val
} }
} }
} }
@ -217,10 +218,10 @@ func RunNode(node *WorkflowNode, ctx *ExecutionContext) error {
} }
node.Config["_runId"] = ctx.RunID node.Config["_runId"] = ctx.RunID
node.Config["logger"] = ctx.Logger node.Config["logger"] = ctx.Logger
// 执行当前节点 // 执行当前节点
result, err := Executors(node.Type, node.Config) result, err := Executors(node.Type, node.Config)
var status ExecutionStatus var status ExecutionStatus
if err != nil { if err != nil {
status = StatusFailed status = StatusFailed
@ -230,9 +231,9 @@ func RunNode(node *WorkflowNode, ctx *ExecutionContext) error {
} else { } else {
status = StatusSuccess status = StatusSuccess
} }
ctx.SetOutput(node.Id, result, status) ctx.SetOutput(node.Id, result, status)
// 普通的并行 // 普通的并行
if node.Type == "branch" { if node.Type == "branch" {
if len(node.ConditionNodes) > 0 { if len(node.ConditionNodes) > 0 {
@ -268,7 +269,7 @@ func RunNode(node *WorkflowNode, ctx *ExecutionContext) error {
} }
} }
} }
if node.ChildNode != nil { if node.ChildNode != nil {
return RunNode(node.ChildNode, ctx) return RunNode(node.ChildNode, ctx)
} }