feat: implement scenario-driven sales SOP platform
This commit is contained in:
Eric 1549169735@qq.com
129
codes/internal/run/engine.go
Normal file
129
codes/internal/run/engine.go
Normal file
@@ -0,0 +1,129 @@
|
||||
package run
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"reflect"
|
||||
"strings"
|
||||
)
|
||||
|
||||
func matchCondition(raw json.RawMessage, answers map[string]interface{}) (bool, error) {
|
||||
if len(raw) == 0 || string(raw) == "{}" || string(raw) == "null" {
|
||||
return true, nil
|
||||
}
|
||||
var rule map[string]interface{}
|
||||
if err := json.Unmarshal(raw, &rule); err != nil {
|
||||
return false, err
|
||||
}
|
||||
if all, ok := rule["all"].([]interface{}); ok {
|
||||
for _, item := range all {
|
||||
object, ok := item.(map[string]interface{})
|
||||
if !ok {
|
||||
return false, fmt.Errorf("invalid all condition")
|
||||
}
|
||||
matched, err := matchRule(object, answers)
|
||||
if err != nil || !matched {
|
||||
return false, err
|
||||
}
|
||||
}
|
||||
return true, nil
|
||||
}
|
||||
if any, ok := rule["any"].([]interface{}); ok {
|
||||
for _, item := range any {
|
||||
object, ok := item.(map[string]interface{})
|
||||
if !ok {
|
||||
continue
|
||||
}
|
||||
matched, err := matchRule(object, answers)
|
||||
if err != nil {
|
||||
return false, err
|
||||
}
|
||||
if matched {
|
||||
return true, nil
|
||||
}
|
||||
}
|
||||
return false, nil
|
||||
}
|
||||
return matchRule(rule, answers)
|
||||
}
|
||||
|
||||
func matchRule(rule map[string]interface{}, answers map[string]interface{}) (bool, error) {
|
||||
field, _ := rule["field"].(string)
|
||||
operator, _ := rule["operator"].(string)
|
||||
if field == "" || operator == "" {
|
||||
return false, fmt.Errorf("condition field and operator are required")
|
||||
}
|
||||
actual, exists := answers[field]
|
||||
expected := rule["value"]
|
||||
switch operator {
|
||||
case "exists":
|
||||
return exists && actual != nil && fmt.Sprint(actual) != "", nil
|
||||
case "not_exists":
|
||||
return !exists || actual == nil || fmt.Sprint(actual) == "", nil
|
||||
case "equals":
|
||||
return reflect.DeepEqual(normalizeValue(actual), normalizeValue(expected)), nil
|
||||
case "not_equals":
|
||||
return !reflect.DeepEqual(normalizeValue(actual), normalizeValue(expected)), nil
|
||||
case "contains":
|
||||
return strings.Contains(strings.ToLower(fmt.Sprint(actual)), strings.ToLower(fmt.Sprint(expected))), nil
|
||||
case "greater_than", "less_than":
|
||||
left, leftOK := toFloat(actual)
|
||||
right, rightOK := toFloat(expected)
|
||||
if !leftOK || !rightOK {
|
||||
return false, nil
|
||||
}
|
||||
if operator == "greater_than" {
|
||||
return left > right, nil
|
||||
}
|
||||
return left < right, nil
|
||||
case "in":
|
||||
values, ok := expected.([]interface{})
|
||||
if !ok {
|
||||
return false, nil
|
||||
}
|
||||
for _, value := range values {
|
||||
if reflect.DeepEqual(normalizeValue(actual), normalizeValue(value)) {
|
||||
return true, nil
|
||||
}
|
||||
}
|
||||
return false, nil
|
||||
default:
|
||||
return false, fmt.Errorf("unsupported operator: %s", operator)
|
||||
}
|
||||
}
|
||||
|
||||
func normalizeValue(value interface{}) interface{} {
|
||||
switch typed := value.(type) {
|
||||
case json.Number:
|
||||
if number, err := typed.Float64(); err == nil {
|
||||
return number
|
||||
}
|
||||
case int:
|
||||
return float64(typed)
|
||||
case int64:
|
||||
return float64(typed)
|
||||
case uint64:
|
||||
return float64(typed)
|
||||
}
|
||||
return value
|
||||
}
|
||||
|
||||
func toFloat(value interface{}) (float64, bool) {
|
||||
switch typed := value.(type) {
|
||||
case float64:
|
||||
return typed, true
|
||||
case float32:
|
||||
return float64(typed), true
|
||||
case int:
|
||||
return float64(typed), true
|
||||
case int64:
|
||||
return float64(typed), true
|
||||
case uint64:
|
||||
return float64(typed), true
|
||||
case json.Number:
|
||||
result, err := typed.Float64()
|
||||
return result, err == nil
|
||||
default:
|
||||
return 0, false
|
||||
}
|
||||
}
|
||||
33
codes/internal/run/engine_test.go
Normal file
33
codes/internal/run/engine_test.go
Normal file
@@ -0,0 +1,33 @@
|
||||
package run
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"testing"
|
||||
)
|
||||
|
||||
func TestMatchCondition(t *testing.T) {
|
||||
answers := map[string]interface{}{"weight": float64(6), "urgent": true, "symptom": "持续呕吐"}
|
||||
tests := []struct {
|
||||
name string
|
||||
rule string
|
||||
want bool
|
||||
}{
|
||||
{name: "default", rule: `{}`, want: true},
|
||||
{name: "equals", rule: `{"field":"urgent","operator":"equals","value":true}`, want: true},
|
||||
{name: "greater", rule: `{"field":"weight","operator":"greater_than","value":5}`, want: true},
|
||||
{name: "contains", rule: `{"field":"symptom","operator":"contains","value":"呕吐"}`, want: true},
|
||||
{name: "all", rule: `{"all":[{"field":"urgent","operator":"equals","value":true},{"field":"weight","operator":"greater_than","value":5}]}`, want: true},
|
||||
{name: "any false", rule: `{"any":[{"field":"urgent","operator":"equals","value":false},{"field":"weight","operator":"less_than","value":3}]}`, want: false},
|
||||
}
|
||||
for _, test := range tests {
|
||||
t.Run(test.name, func(t *testing.T) {
|
||||
got, err := matchCondition(json.RawMessage(test.rule), answers)
|
||||
if err != nil {
|
||||
t.Fatalf("matchCondition returned error: %v", err)
|
||||
}
|
||||
if got != test.want {
|
||||
t.Fatalf("matchCondition() = %v, want %v", got, test.want)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
247
codes/internal/run/handler.go
Normal file
247
codes/internal/run/handler.go
Normal file
@@ -0,0 +1,247 @@
|
||||
package run
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"net/http"
|
||||
"sort"
|
||||
"strconv"
|
||||
"time"
|
||||
|
||||
"git.iwork-ai.com/xdc/iqudo-top1/internal/audit"
|
||||
"git.iwork-ai.com/xdc/iqudo-top1/internal/auth"
|
||||
"git.iwork-ai.com/xdc/iqudo-top1/internal/model"
|
||||
"git.iwork-ai.com/xdc/iqudo-top1/internal/response"
|
||||
"github.com/gin-gonic/gin"
|
||||
"gorm.io/datatypes"
|
||||
"gorm.io/gorm"
|
||||
)
|
||||
|
||||
type Handler struct {
|
||||
db *gorm.DB
|
||||
}
|
||||
|
||||
func NewHandler(db *gorm.DB) *Handler {
|
||||
return &Handler{db: db}
|
||||
}
|
||||
|
||||
func (h *Handler) PublishedSOPs(c *gin.Context) {
|
||||
p, _ := auth.PrincipalFromContext(c)
|
||||
type item struct {
|
||||
ID uint64 `json:"id"`
|
||||
Name string `json:"name"`
|
||||
Description string `json:"description"`
|
||||
ScenarioID uint64 `json:"scenario_id"`
|
||||
ScenarioName string `json:"scenario_name"`
|
||||
Version int `json:"version"`
|
||||
}
|
||||
var items []item
|
||||
err := h.db.Table("sops s").Select("s.id, s.name, s.description, s.scenario_id, sc.name AS scenario_name, sv.version").Joins("JOIN scenarios sc ON sc.id = s.scenario_id").Joins("JOIN sop_versions sv ON sv.sop_id = s.id AND sv.status = ?", "published").Where("s.tenant_id = ? AND s.status = ?", p.TenantID, "published").Order("s.updated_at DESC").Scan(&items).Error
|
||||
if err != nil {
|
||||
response.Error(c, http.StatusInternalServerError, "QUERY_FAILED", "查询可执行 SOP 失败")
|
||||
return
|
||||
}
|
||||
response.OK(c, gin.H{"items": items, "total": len(items)})
|
||||
}
|
||||
|
||||
func (h *Handler) Start(c *gin.Context) {
|
||||
p, _ := auth.PrincipalFromContext(c)
|
||||
var input struct {
|
||||
SOPID uint64 `json:"sop_id" binding:"required"`
|
||||
}
|
||||
if err := c.ShouldBindJSON(&input); err != nil {
|
||||
response.Error(c, http.StatusBadRequest, "INVALID_ARGUMENT", "请选择要执行的 SOP")
|
||||
return
|
||||
}
|
||||
var version model.SOPVersion
|
||||
if err := h.db.Where("sop_id = ? AND tenant_id = ? AND status = ?", input.SOPID, p.TenantID, "published").Order("version DESC").First(&version).Error; err != nil {
|
||||
response.Error(c, http.StatusNotFound, "NOT_FOUND", "没有可执行的已发布版本")
|
||||
return
|
||||
}
|
||||
run := model.SOPRun{TenantID: p.TenantID, SOPID: input.SOPID, SOPVersionID: version.ID, OperatorID: p.UserID, CurrentNodeKey: version.StartNodeKey, Status: "running", Answers: datatypes.JSON([]byte(`{}`)), Result: "", StartedAt: time.Now()}
|
||||
err := h.db.Transaction(func(tx *gorm.DB) error {
|
||||
if err := tx.Create(&run).Error; err != nil {
|
||||
return err
|
||||
}
|
||||
return tx.Create(&model.SOPRunEvent{TenantID: p.TenantID, RunID: run.ID, NodeKey: run.CurrentNodeKey, Action: "start", Payload: datatypes.JSON([]byte(`{}`))}).Error
|
||||
})
|
||||
if err != nil {
|
||||
response.Error(c, http.StatusInternalServerError, "START_FAILED", "启动 SOP 失败")
|
||||
return
|
||||
}
|
||||
_ = audit.Record(h.db, p, "start", "sop_run", run.ID, gin.H{"sop_id": input.SOPID})
|
||||
h.respondRun(c, run)
|
||||
}
|
||||
|
||||
func (h *Handler) Get(c *gin.Context) {
|
||||
p, _ := auth.PrincipalFromContext(c)
|
||||
id, ok := runID(c)
|
||||
if !ok {
|
||||
return
|
||||
}
|
||||
var item model.SOPRun
|
||||
if err := h.db.Where("id = ? AND tenant_id = ?", id, p.TenantID).First(&item).Error; err != nil {
|
||||
response.Error(c, http.StatusNotFound, "NOT_FOUND", "执行记录不存在")
|
||||
return
|
||||
}
|
||||
h.respondRun(c, item)
|
||||
}
|
||||
|
||||
func (h *Handler) List(c *gin.Context) {
|
||||
p, _ := auth.PrincipalFromContext(c)
|
||||
type row struct {
|
||||
model.SOPRun
|
||||
SOPName string `json:"sop_name"`
|
||||
}
|
||||
var items []row
|
||||
if err := h.db.Table("sop_runs r").Select("r.*, s.name AS sop_name").Joins("JOIN sops s ON s.id = r.sop_id").Where("r.tenant_id = ?", p.TenantID).Order("r.created_at DESC").Limit(100).Scan(&items).Error; err != nil {
|
||||
response.Error(c, http.StatusInternalServerError, "QUERY_FAILED", "查询执行记录失败")
|
||||
return
|
||||
}
|
||||
response.OK(c, gin.H{"items": items, "total": len(items)})
|
||||
}
|
||||
|
||||
func (h *Handler) Answer(c *gin.Context) {
|
||||
p, _ := auth.PrincipalFromContext(c)
|
||||
id, ok := runID(c)
|
||||
if !ok {
|
||||
return
|
||||
}
|
||||
var input struct {
|
||||
Answers map[string]interface{} `json:"answers"`
|
||||
}
|
||||
if err := c.ShouldBindJSON(&input); err != nil {
|
||||
response.Error(c, http.StatusBadRequest, "INVALID_ARGUMENT", "回答格式不正确")
|
||||
return
|
||||
}
|
||||
var updated model.SOPRun
|
||||
err := h.db.Transaction(func(tx *gorm.DB) error {
|
||||
if err := tx.Clauses().Where("id = ? AND tenant_id = ?", id, p.TenantID).First(&updated).Error; err != nil {
|
||||
return err
|
||||
}
|
||||
if updated.Status != "running" {
|
||||
return errors.New("run is not active")
|
||||
}
|
||||
answers := map[string]interface{}{}
|
||||
if len(updated.Answers) > 0 {
|
||||
_ = json.Unmarshal(updated.Answers, &answers)
|
||||
}
|
||||
for key, value := range input.Answers {
|
||||
answers[key] = value
|
||||
}
|
||||
var edges []model.SOPEdge
|
||||
if err := tx.Where("sop_version_id = ? AND source_node_key = ?", updated.SOPVersionID, updated.CurrentNodeKey).Order("priority, id").Find(&edges).Error; err != nil {
|
||||
return err
|
||||
}
|
||||
sort.SliceStable(edges, func(i, j int) bool { return edges[i].Priority < edges[j].Priority })
|
||||
nextKey := ""
|
||||
for _, edge := range edges {
|
||||
matched, matchErr := matchCondition(json.RawMessage(edge.Condition), answers)
|
||||
if matchErr != nil {
|
||||
return matchErr
|
||||
}
|
||||
if matched {
|
||||
nextKey = edge.TargetNodeKey
|
||||
break
|
||||
}
|
||||
}
|
||||
if nextKey == "" {
|
||||
return errors.New("没有满足条件的下一节点")
|
||||
}
|
||||
answerBytes, _ := json.Marshal(answers)
|
||||
payload, _ := json.Marshal(input)
|
||||
if err := tx.Create(&model.SOPRunEvent{TenantID: p.TenantID, RunID: updated.ID, NodeKey: updated.CurrentNodeKey, Action: "answer", Payload: datatypes.JSON(payload)}).Error; err != nil {
|
||||
return err
|
||||
}
|
||||
var next model.SOPNode
|
||||
if err := tx.Where("sop_version_id = ? AND node_key = ?", updated.SOPVersionID, nextKey).First(&next).Error; err != nil {
|
||||
return err
|
||||
}
|
||||
updates := map[string]interface{}{"current_node_key": nextKey, "answers": datatypes.JSON(answerBytes)}
|
||||
if next.Type == "finish" || next.Type == "escalate" {
|
||||
now := time.Now()
|
||||
updates["status"] = "completed"
|
||||
updates["completed_at"] = &now
|
||||
updates["result"] = next.Type
|
||||
updated.Status = "completed"
|
||||
updated.CompletedAt = &now
|
||||
updated.Result = next.Type
|
||||
}
|
||||
if err := tx.Model(&updated).Updates(updates).Error; err != nil {
|
||||
return err
|
||||
}
|
||||
updated.CurrentNodeKey = nextKey
|
||||
updated.Answers = datatypes.JSON(answerBytes)
|
||||
return tx.Create(&model.SOPRunEvent{TenantID: p.TenantID, RunID: updated.ID, NodeKey: nextKey, Action: "enter", Payload: datatypes.JSON([]byte(`{}`))}).Error
|
||||
})
|
||||
if err != nil {
|
||||
response.Error(c, http.StatusUnprocessableEntity, "ADVANCE_FAILED", err.Error())
|
||||
return
|
||||
}
|
||||
h.respondRun(c, updated)
|
||||
}
|
||||
|
||||
func (h *Handler) Finish(c *gin.Context) {
|
||||
p, _ := auth.PrincipalFromContext(c)
|
||||
id, ok := runID(c)
|
||||
if !ok {
|
||||
return
|
||||
}
|
||||
var input struct {
|
||||
Result string `json:"result" binding:"required,max=64"`
|
||||
}
|
||||
if err := c.ShouldBindJSON(&input); err != nil {
|
||||
response.Error(c, http.StatusBadRequest, "INVALID_ARGUMENT", "请选择执行结果")
|
||||
return
|
||||
}
|
||||
now := time.Now()
|
||||
result := h.db.Model(&model.SOPRun{}).Where("id = ? AND tenant_id = ?", id, p.TenantID).Updates(map[string]interface{}{"status": "completed", "result": input.Result, "completed_at": &now})
|
||||
if result.Error != nil || result.RowsAffected == 0 {
|
||||
response.Error(c, http.StatusNotFound, "NOT_FOUND", "执行记录不存在")
|
||||
return
|
||||
}
|
||||
_ = audit.Record(h.db, p, "finish", "sop_run", id, input)
|
||||
h.Get(c)
|
||||
}
|
||||
|
||||
func (h *Handler) Feedback(c *gin.Context) {
|
||||
p, _ := auth.PrincipalFromContext(c)
|
||||
id, ok := runID(c)
|
||||
if !ok {
|
||||
return
|
||||
}
|
||||
var input struct {
|
||||
Score int `json:"score" binding:"required,min=1,max=5"`
|
||||
Comment string `json:"comment" binding:"max=2000"`
|
||||
}
|
||||
if err := c.ShouldBindJSON(&input); err != nil {
|
||||
response.Error(c, http.StatusBadRequest, "INVALID_ARGUMENT", "反馈内容不正确")
|
||||
return
|
||||
}
|
||||
item := model.SOPFeedback{TenantID: p.TenantID, RunID: id, UserID: p.UserID, Score: input.Score, Comment: input.Comment}
|
||||
if err := h.db.Create(&item).Error; err != nil {
|
||||
response.Error(c, http.StatusConflict, "FEEDBACK_EXISTS", "该执行记录已经提交反馈")
|
||||
return
|
||||
}
|
||||
response.Created(c, item)
|
||||
}
|
||||
|
||||
func (h *Handler) respondRun(c *gin.Context, item model.SOPRun) {
|
||||
var node model.SOPNode
|
||||
if err := h.db.Where("sop_version_id = ? AND node_key = ?", item.SOPVersionID, item.CurrentNodeKey).First(&node).Error; err != nil {
|
||||
response.Error(c, http.StatusInternalServerError, "NODE_NOT_FOUND", "当前流程节点不存在")
|
||||
return
|
||||
}
|
||||
var fields []model.ScenarioField
|
||||
h.db.Table("scenario_fields sf").Joins("JOIN sops s ON s.scenario_id = sf.scenario_id").Where("s.id = ? AND sf.tenant_id = ?", item.SOPID, item.TenantID).Order("sf.sort_order, sf.id").Find(&fields)
|
||||
response.OK(c, gin.H{"run": item, "node": node, "fields": fields})
|
||||
}
|
||||
|
||||
func runID(c *gin.Context) (uint64, bool) {
|
||||
id, err := strconv.ParseUint(c.Param("id"), 10, 64)
|
||||
if err != nil || id == 0 {
|
||||
response.Error(c, http.StatusBadRequest, "INVALID_ID", "执行记录 ID 不正确")
|
||||
return 0, false
|
||||
}
|
||||
return id, true
|
||||
}
|
||||
Reference in New Issue
Block a user