package workflow import ( "fmt" "strings" jsoniter "github.com/json-iterator/go" "github.com/yaoapp/gou" gshare "github.com/yaoapp/gou/query/share" "github.com/yaoapp/kun/any" "github.com/yaoapp/kun/exception" "github.com/yaoapp/kun/maps" "github.com/yaoapp/kun/str" "github.com/yaoapp/xiang/config" "github.com/yaoapp/xiang/helper" "github.com/yaoapp/xiang/share" "github.com/yaoapp/xiang/xlog" ) // Process // 读取工作流 xiang.workflow.Find(name, workflow_id) // 读取工作流 xiang.workflow.Open(name, uid, data_id) // 保存工作流 xiang.workflow.Save(name, uid, node_name, data_id, input, ...output) // 进入下一个节点 xiang.workflow.Next(name, uid, workflow_id, output) // 跳转到指定节点 xiang.workflow.Goto(name, uid, workflow_id, node_name, output) // 更新工作流状态 xiang.workflow.Status(name, uid, workflow_id, status_name, output) // 标记结束流程 xiang.workflow.Done(name, uid, workflow_id, output) // 标记关闭流程 xiang.workflow.Close(name, uid, workflow_id, output) // 标记重置流程 xiang.workflow.Reset(name, uid, workflow_id, output) // API: // 读取工作流 GET /api/xiang/workflow/<工作流名称>/find/:id // 读取工作流 GET /api/xiang/workflow/<工作流名称>/open // 读取工作流配置 GET /api/xiang/workflow/<工作流名称>/setting // 调用自定义API POST /api/xiang/workflow/<工作流名称>/<自定义API路由> // WorkFlows 工作流列表 var WorkFlows = map[string]*WorkFlow{} // Load 加载数据表格 func Load(cfg config.Config) { LoadFrom(cfg.RootWorkFlow, "") } // LoadFrom 从特定目录加载 func LoadFrom(dir string, prefix string) { if share.DirNotExists(dir) { return } share.Walk(dir, ".json", func(root, filename string) { name := prefix + share.SpecName(root, filename) content := share.ReadFile(filename) _, err := LoadWorkFlow(content, name) if err != nil { exception.New("%s 工作流格式错误", 400, name).Ctx(filename).Throw() } }) } // LoadWorkFlow 载入工作流 func LoadWorkFlow(source []byte, name string) (*WorkFlow, error) { workflow := WorkFlow{Name: name, Source: source} err := jsoniter.Unmarshal(source, &workflow) if err != nil { xlog.Println(name) xlog.Println(err.Error()) xlog.Println(string(source)) return nil, err } workflow.SetupAPIs() WorkFlows[workflow.Name] = &workflow return WorkFlows[workflow.Name], nil } // Select 读取已加载图表 func Select(name string) *WorkFlow { workflow, has := WorkFlows[name] if !has { exception.New( fmt.Sprintf("工作流:%s; 尚未加载", name), 400, ).Throw() } return workflow } // Reload 重新载入工作流 func (workflow *WorkFlow) Reload() *WorkFlow { new, err := LoadWorkFlow(workflow.Source, workflow.Name) if err != nil { exception.New( fmt.Sprintf("工作流:%s; 加载失败", workflow.Name), 400, ).Throw() } WorkFlows[workflow.Name] = new return new } // Find 读取给定ID的工作流 // uid 当前处理人ID, id 数据ID func (workflow *WorkFlow) Find(id int) map[string]interface{} { wflow := gou.Select("xiang.workflow") res := wflow.MustFind(id, gou.QueryParam{ Select: []interface{}{ "data_id", "id", "input", "output", "name", "user_id", "users", "node_name", "node_status", "status", "updated_at", "created_at", }}) return res } // Open 读取当前工作流(未完成的) // uid 当前处理人ID, id 数据ID func (workflow *WorkFlow) Open(uid int, id interface{}) map[string]interface{} { wflow := gou.Select("xiang.workflow") params := gou.QueryParam{ Select: []interface{}{ "data_id", "id", "input", "output", "name", "user_id", "users", "node_name", "node_status", "status", "updated_at", "created_at", }, Wheres: []gou.QueryWhere{ {Column: "name", Value: workflow.Name}, {Column: "data_id", Value: id}, {Column: "user_ids", OP: "like", Value: fmt.Sprintf("%%|%d|%%", uid)}, {Column: "status", Value: "进行中"}, }, Limit: 1, } rows := wflow.MustGet(params) if len(rows) > 0 { return rows[0] } if workflow.Len() == 0 { exception.New("工作流没有任何节点", 400).Throw() } users := map[string]interface{}{} users[workflow.Nodes[0].Name] = uid return map[string]interface{}{ "name": workflow.Name, "data_id": id, "node_name": workflow.Nodes[0].Name, "user_id": uid, "users": users, "status": "进行中", "node_status": "进行中", "input": map[string]interface{}{}, "output": map[string]interface{}{}, } } // Setting 返回配置信息 // uid 当前处理人ID, id 数据ID func (workflow *WorkFlow) Setting(uid int, id interface{}) map[string]interface{} { wflow := workflow.Open(1, 1) data := maps.MapStr{ "$in": wflow["input"], "$input": wflow["input"], "$out": wflow["output"], "$outupt": wflow["output"], "$data": wflow["output"], } nodes := workflow.FlowNodes(data.Dot()) _, idx := workflow.pickNode(string(str.Of(wflow["node_name"]))) return map[string]interface{}{ "nodes": nodes, "node": nodes[idx], "node_index": idx, "read": true, "write": any.Of(wflow["user_id"]).CInt() == uid, "actions": workflow.Actions, "name": workflow.Name, "version": workflow.Version, "label": workflow.Label, "decription": workflow.Decription, } } // Save 保存工作流节点数据 此版本使用Like实现 // uid 当前处理人ID, id 数据ID func (workflow *WorkFlow) Save(uid int, name string, id interface{}, input Input, outputs ...map[string]interface{}) map[string]interface{} { wflow := gou.Select("xiang.workflow") params := gou.QueryParam{ Select: []interface{}{"id", "input", "output", "users"}, Wheres: []gou.QueryWhere{ {Column: "name", Value: workflow.Name}, {Column: "data_id", Value: id}, {Column: "user_ids", OP: "like", Value: fmt.Sprintf("%%|%d|%%", uid)}, {Column: "status", Value: "进行中"}, }, Limit: 1, } rows := wflow.MustGet(params) data := map[string]interface{}{ "name": workflow.Name, "data_id": id, "node_name": name, "user_id": uid, } users := map[string]interface{}{} users[name] = uid output := map[string]interface{}{} nodeInput := map[string]interface{}{} nodeInput[name] = input if len(outputs) > 0 { output = outputs[0] } if len(rows) > 0 { data["id"] = rows[0].Get("id") nodeInput = workflow.MergeData(rows[0].Get("input"), nodeInput) users = workflow.MergeUsers(rows[0].Get("users"), users) output = workflow.MergeData(rows[0].Get("output"), output) } else { data["status"] = "进行中" data["node_status"] = "进行中" } data["users"] = users data["user_ids"] = workflow.UserIDs(users) data["input"] = nodeInput data["output"] = output id = wflow.MustSave(data) return wflow.MustFind(id, gou.QueryParam{}) } // Done 标记完成 // uid 当前处理人ID, id 工作流ID func (workflow *WorkFlow) Done(uid int, id int, output map[string]interface{}) map[string]interface{} { return workflow.Status(uid, id, "已完成", output) } // Close 标记关闭 // uid 当前处理人ID, id 工作流ID func (workflow *WorkFlow) Close(uid int, id int, output map[string]interface{}) map[string]interface{} { return workflow.Status(uid, id, "已关闭", output) } // Status 设定状态 // uid 当前处理人ID, id 工作流ID func (workflow *WorkFlow) Status(uid int, id int, status string, output map[string]interface{}) map[string]interface{} { wflow := workflow.Find(id) mod := gou.Select("xiang.workflow") output = workflow.MergeData(wflow["output"], output) mod.Save(map[string]interface{}{ "id": wflow["id"], "output": output, "status": status, }) return workflow.Find(id) } // Reset 重新开始 // uid 当前处理人ID, id 工作流ID func (workflow *WorkFlow) Reset(uid int, id int, output map[string]interface{}) map[string]interface{} { return workflow.Goto(uid, id, workflow.Nodes[0].Name, output) } // Goto 工作流跳转 func (workflow *WorkFlow) Goto(uid int, id int, name string, output map[string]interface{}) map[string]interface{} { wflow := workflow.Find(id) users := map[string]interface{}{} curr, _ := workflow.pickNode(wflow["node_name"].(string)) users[curr.Name] = uid node, _ := workflow.pickNode(name) output = workflow.MergeData(wflow["output"], output) users = workflow.MergeUsers(wflow["users"], users) nodeUID := node.GetUID(users) mod := gou.Select("xiang.workflow") mod.Save(map[string]interface{}{ "id": wflow["id"], "output": output, "node_name": name, "node_status": "进行中", "user_id": nodeUID, "users": users, "user_ids": workflow.UserIDs(users), }) return workflow.Find(id) } // Next 下一个工作流 // uid 当前处理人ID, id 工作流ID func (workflow *WorkFlow) Next(uid int, id int, output map[string]interface{}) map[string]interface{} { wflow := workflow.Find(id) currNode, ok := wflow["node_name"].(string) if !ok { exception.New("流程数据异常: 当前节点信息错误", 500).Ctx(currNode).Throw() } users := map[string]interface{}{} users[currNode] = uid output = workflow.MergeData(wflow["output"], output) // 读取下一个节点 data := map[string]interface{}{ "$in": wflow["input"], "$input": wflow["input"], "$out": output, "$outupt": output, "$data": output, } nextNode := workflow.nextNode(currNode, data) nextUID := nextNode.MakeUID() users[nextNode.Name] = nextUID // 更新数据 users = workflow.MergeUsers(wflow["users"], users) mod := gou.Select("xiang.workflow") mod.Save(map[string]interface{}{ "id": wflow["id"], "output": output, "node_name": nextNode.Name, "node_status": "进行中", "user_id": nextUID, "users": users, "user_ids": workflow.UserIDs(users), }) return workflow.Find(id) } // GetUID 读取节点相关人 func (node *Node) GetUID(users interface{}) int { if users, ok := users.(map[string]interface{}); ok { if uid, has := users[node.Name]; has { return any.Of(uid).CInt() } } return node.MakeUID() } // MakeUID 根据条件选择节点处理人 func (node *Node) MakeUID() int { res := gou.NewProcess(node.User.Process, node.User.Args...).Run() return any.Of(res).CInt() } func (workflow *WorkFlow) pickNode(name string) (*Node, int) { for i, node := range workflow.Nodes { if node.Name == name { return &node, i } } exception.New("流程数据异常: 未找到节点 %s", 500, name).Throw() return nil, 0 } // nextNode 查找下一个节点 func (workflow *WorkFlow) nextNode(currentNode string, data map[string]interface{}) *Node { curr, index := workflow.pickNode(currentNode) nextIndex := index + 1 if nextIndex == workflow.Len() { exception.New("流程数据异常: 当前节点为最后一个节点", 500).Ctx(currentNode).Throw() } // 未声明 Next 节点, 转到下一个节点 if curr.Next == nil { return &workflow.Nodes[nextIndex] } // 声明 Next 节点, 按条件到指定节点 data = maps.Of(data).Dot() for _, next := range curr.Next { node := workflow.GetNodeWhen(next, data) if node != nil { return node } } exception.New("流程数据异常: 未找到符合条件的工作流节点", 500).Ctx(map[string]interface{}{"current": currentNode, "data": data}).Throw() return nil } // GetNodeWhen 读取节点 func (workflow *WorkFlow) GetNodeWhen(next Next, data map[string]interface{}) *Node { nextNode := "" conditions := workflow.Conditions(next.Conditions, data) if helper.When(conditions) { nextNode = next.Goto for i := 0; i < workflow.Len(); i++ { node := workflow.Nodes[i] if node.Name == nextNode { return &node } } } return nil } // Conditions 处理绑定参数 func (workflow *WorkFlow) Conditions(conds []helper.Condition, data map[string]interface{}) []helper.Condition { conditions := []helper.Condition{} for _, cond := range conds { if left, ok := cond.Left.(string); ok { cond.Left = gshare.Bind(left, data) } if right, ok := cond.Right.(string); ok { cond.Right = gshare.Bind(right, data) } conditions = append(conditions, cond) } return conditions } // UserIDs 读取用户ID func (workflow *WorkFlow) UserIDs(users map[string]interface{}) string { userIDs := []string{} for _, u := range users { userIDs = append(userIDs, fmt.Sprintf("|%d|", u)) } userIDs = helper.ArrayStringUnique(userIDs) return strings.Join(userIDs, ",") } // MergeUsers 合并数据 func (workflow *WorkFlow) MergeUsers(data interface{}, new interface{}) map[string]interface{} { res, ok := data.(map[string]interface{}) if !ok { return map[string]interface{}{} } if new, ok := new.(map[string]interface{}); ok { for name, value := range new { res[name] = value } } return res } // MergeData 合并数据 func (workflow *WorkFlow) MergeData(data interface{}, new interface{}) map[string]interface{} { res, ok := data.(map[string]interface{}) if !ok { return map[string]interface{}{} } if new, ok := new.(map[string]interface{}); ok { for key, value := range new { res[key] = value } } return res } // IsLastNode 检查是否为最后一个节点 func (workflow *WorkFlow) IsLastNode(name string) bool { length := workflow.Len() return workflow.Nodes[length-1].Name == name } // FlowNodes 转换为 flow func (workflow *WorkFlow) FlowNodes(data map[string]interface{}) []map[string]interface{} { res := []map[string]interface{}{} nodes, err := jsoniter.Marshal(workflow.Nodes) if err != nil { exception.New("JSON解析错误 %s", 500, nodes).Throw() } err = jsoniter.Unmarshal(nodes, &res) if err != nil { exception.New("JSON解析错误 %s", 500, nodes).Throw() } nameMaps := map[string]int{} for i, node := range workflow.Nodes { nameMaps[node.Name] = i } for i, node := range res { v := gshare.Bind(node, data) if v, ok := v.(map[string]interface{}); ok { res[i] = v } res[i]["id"] = i + 1 res[i]["label"] = res[i]["name"] delete(res[i], "user") delete(res[i], "next") if workflow.Nodes[i].Next != nil { for _, next := range workflow.Nodes[i].Next { name := next.Goto if id, has := nameMaps[name]; has { res[id]["source"] = i + 1 } } } } return res } // Len 节点数量 func (workflow *WorkFlow) Len() int { return len(workflow.Nodes) } // InputOf 映射表转换为Input func InputOf(in map[string]interface{}) Input { input := Input{Data: map[string]interface{}{}, Form: map[string]interface{}{}} if data, ok := in["data"].(map[string]interface{}); ok { input.Data = data } if form, ok := in["form"].(map[string]interface{}); ok { input.Form = form } return input }