Merge remote-tracking branch 'origin/main' into main

app/http/controllers/backend/cmd_controller.go

@@ -50,7 +50,13 @@ func CmdStart(c *gin.Context) {
 }
 
 func CmdTestRead(c *gin.Context) {
-	request := controllers.GetRequest(c).(*backend.CmdTestRequest)
+	request := controllers.GetRequest(c).(*backend.CmdTestReadRequest)
 	result, time_duration, err := cmd_services.TestRead(request)
 	controllers.HandRes(c, gin.H{"result": result, "time_duration": time_duration}, err)
 }
+
+func CmdTestWrite(c *gin.Context) {
+	request := controllers.GetRequest(c).(*backend.CmdTestWriteRequest)
+	time_duration, err := cmd_services.TestWrite(request)
+	controllers.HandRes(c, gin.H{"time_duration": time_duration}, err)
+}

app/http/entities/backend/cmd.go

@@ -14,8 +14,8 @@ type CmdListRequest struct {
 	CmdName     string   `json:"cmd_name" form:"cmd_name" example:"155555555"`
 	Status      int      `json:"status" form:"status" example:"1"`
 	ExecuteType int      `json:"execute_type" form:"execute_type" example:"46516"`
-	ExecuteRead  string   `json:"execute_read"`
+	ReadDbId    int      `json:"read_db_id"`
-	ExecuteWrite string   `json:"execute_write"`
+	WriteDbId   int      `json:"write_db_id"`
 	CmdIds      []string `json:"cmd_id"`
 }
 
@@ -87,8 +87,17 @@ type CmdStartRequest struct {
 	CmdId int `json:"cmd_id" validate:"required" form:"cmd_id" example:""`
 }
 
-type CmdTestRequest struct {
+type CmdTestWriteRequest struct {
-	CmdId int `json:"cmd_id" validate:"required" form:"cmd_id" example:""`
+	ReadDbId     int    `json:"read_db_id" validate:"required"`
+	ExecuteRead  string `json:"execute_read" validate:"required"`
+	WriteDbId    int    `json:"write_db_id" validate:"required"`
+	ExecuteWrite string `json:"execute_write" validate:"required"`
+	MatchJson    string `json:"match_json" validate:"required"`
+}
+
+type CmdTestReadRequest struct {
+	ReadDbId    int    `json:"read_db_id" validate:"required" form:"read_db_id" example:""`
+	ExecuteRead string `json:"execute_read" validate:"required" form:"execute_read" example:""`
 }
 
 type CmdStopRequest struct {

app/http/requestmapping/backend.go

@@ -44,5 +44,6 @@ var BackendRequestMap = map[string]func() interface{}{
 	common.ADMIN_OAUTH_V1 + "/cmd/del":        func() interface{} { return new(backend.CmdDeleteRequest) },
 	common.ADMIN_OAUTH_V1 + "/cmd/start":      func() interface{} { return new(backend.CmdStartRequest) },
 	common.ADMIN_OAUTH_V1 + "/cmd/stop":       func() interface{} { return new(backend.CmdStopRequest) },
-	common.ADMIN_OAUTH_V1 + "/cmd/test/read": func() interface{} { return new(backend.CmdTestRequest) },
+	common.ADMIN_OAUTH_V1 + "/cmd/test/read":  func() interface{} { return new(backend.CmdTestReadRequest) },
+	common.ADMIN_OAUTH_V1 + "/cmd/test/write": func() interface{} { return new(backend.CmdTestWriteRequest) },
 }

app/http/routes/admin.go

@@ -59,6 +59,7 @@ func RegisterAdminRoute(router *gin.Engine) {
 					cmd.POST("/start", backend.CmdStart)
 					cmd.POST("/stop", backend.CmdStop)
 					cmd.POST("/test/read", backend.CmdTestRead)
+					cmd.POST("/test/write", backend.CmdTestWrite)
 				}
 				//消息管理
 				mes := v1.Group("/channel")

app/services/cmd_service/cmd_service.go

@@ -35,6 +35,16 @@ func GetListByWhere(request *backend.CmdListRequest, page int, limit int) (count
 		cond = cond.And(builder.In("cmd_id", request.CmdIds))
 	}
 
+	if request.ReadDbId != 0 {
+		// 使用IN查询
+		cond = cond.And(builder.In("read_db_id", request.ReadDbId))
+	}
+
+	if request.WriteDbId != 0 {
+		// 使用IN查询
+		cond = cond.And(builder.In("write_db_id", request.WriteDbId))
+	}
+
 	session := croncmdmodel.GetInstance().GetDb().Where(cond).OrderBy("update_time desc")
 
 	if page != 0 && limit != 0 {
@@ -108,19 +118,11 @@ func Stop(request *backend.CmdStopRequest) (err error) {
 	return
 }
 
-func TestRead(request *backend.CmdTestRequest) (result interface{}, time_duration float64, err error) {
+func TestRead(request *backend.CmdTestReadRequest) (result []map[string]interface{}, time_duration float64, err error) {
 	var (
-		data     croncmdmodel.CronCmd
 		database crondbmodel.CronDb
 	)
-	exists, err := croncmdmodel.GetInstance().GetDb().ID(request.CmdId).Get(&data)
+	exists, err := crondbmodel.GetInstance().GetDb().ID(request.ReadDbId).Get(&database)
-	if !exists {
-		return nil, 0, fmt.Errorf("数据不存在")
-	}
-	if err != nil {
-		return
-	}
-	exists, err = crondbmodel.GetInstance().GetDb().ID(data.ReadDbId).Get(&database)
 	if !exists {
 		return nil, 0, fmt.Errorf("数据库不存在")
 	}
@@ -130,22 +132,67 @@ func TestRead(request *backend.CmdTestRequest) (result interface{}, time_duratio
 	//执行
 	start := time.Now()
 	db, err := excute.NewExecuteDb(database.Source, database.DbName)
+	dbConn, _ := db.Db.DB()
+	defer dbConn.Close()
 	if err != nil {
 		return nil, 0, err
 	}
-	result, err = db.ExecuteRead(data.ExecuteRead)
+	result, err = db.ExecuteRead(request.ExecuteRead)
 	if err != nil {
 		return result, 0, err
 	}
 	// 记录结束时间
 	end := time.Now()
-
 	// 计算执行时间
 	duration := end.Sub(start)
 	time_duration = duration.Seconds()
 	return
 }
 
+func TestWrite(request *backend.CmdTestWriteRequest) (time_duration float64, err error) {
+	var (
+		finalData []map[string]interface{}
+		database  crondbmodel.CronDb
+	)
+	readRes, readDuration, err := TestRead(&backend.CmdTestReadRequest{
+		ReadDbId:    request.ReadDbId,
+		ExecuteRead: request.ExecuteRead,
+	})
+	if err != nil {
+		return 0, err
+	}
+	if len(readRes) == 0 {
+		return 0, err
+	}
+	//过滤数据
+	finalData, err = excute.FilterReadData(&readRes, request.MatchJson)
+	if err != nil {
+		return 0, err
+	}
+	//执行
+	start := time.Now()
+	exists, err := crondbmodel.GetInstance().GetDb().ID(request.WriteDbId).Get(&database)
+	if !exists {
+		return 0, fmt.Errorf("数据库不存在")
+	}
+	if err != nil {
+		return
+	}
+	db, err := excute.NewExecuteDb(database.Source, database.DbName)
+	dbConn, _ := db.Db.DB()
+	defer dbConn.Close()
+	_, err = db.ExecuteWriteV2(finalData, request.ExecuteWrite)
+	if err != nil {
+		return 0, err
+	}
+	// 记录结束时间
+	end := time.Now()
+	// 计算执行时间
+	duration := end.Sub(start)
+	time_duration = duration.Seconds() + readDuration
+	return
+}
+
 func ListToRep(list []croncmdmodel.CronCmd) (cmdList []backend.CmdListResponse, err error) {
 	var (
 		userIdList  []int

app/utils/excute/dbExecute.go

@@ -4,10 +4,9 @@ import (
 	"cron_admin/app/utils"
 	"encoding/json"
 	"fmt"
-	"github.com/jinzhu/copier"
 	"gorm.io/gorm"
+	"reflect"
 	"strings"
-	"time"
 )
 
 type ExecuteDb struct {
@@ -65,150 +64,112 @@ func (db *ExecuteDb) replaceExecuteString(execute string, rowData map[string]int
 	return executeString
 }
 
-func (db *ExecuteDb) ExecuteWriteV1(readData []map[string]interface{}, execute string) (err error) {
+func FilterReadData(data *[]map[string]interface{}, matchJson string) (filteredData []map[string]interface{}, err error) {
-	var (
+	if matchJson == "" {
-		match MatchWrite
+		return *data, nil
-	)
-	err = json.Unmarshal([]byte(execute), &match)
-	if err != nil {
-		return fmt.Errorf("写执行命令数据格式验证失败：%v", err)
 	}
-	table, err := db.GetTableColumnsInfo(match.Table)
+	for _, v := range *data {
+		match, err := MatchJSON(v, &matchJson)
 		if err != nil {
-		return err
+			return nil, err
 		}
-	switch match.Op {
+		if match {
-	case "insert":
+			filteredData = append(filteredData, v)
-		insertData := db.ForInsert(&table, readData, match.Data)
+		}
-		db.Db.Table(match.Table).Create(insertData)
+	}
-	case "update":
+	return filteredData, nil
-		for _, v := range readData {
+}
-			updateData, updateKey := db.ForUpdate(&table, v, match.Data, match.Keys)
+
-			db.Db.Table(match.Table).Where(updateKey).Updates(updateData)
+func MatchJSON(mapData map[string]interface{}, jsonString *string) (bool, error) {
+	// 解析JSON字符串
+	var jsonData []Match
+	err := json.Unmarshal([]byte(*jsonString), &jsonData)
+	if err != nil {
+		return false, fmt.Errorf("解析Match_Json失败：%v", err)
+	}
+	for _, v := range jsonData {
+		// 判断key值
+		mapData, err := judgeData(&v, mapData)
+		if err != nil {
+			return false, err
+		}
+		switch v.Op {
+		case "=":
+			if mapData[v.Key] != v.Val {
+				return false, nil
+			}
+		case "!=":
+			if mapData[v.Key] == v.Val {
+				return false, nil
+			}
+		case "<":
+			if mapData[v.Key].(int) >= v.Val.(int) {
+				return false, nil
+			}
+		case "<=":
+			if mapData[v.Key].(int) > v.Val.(int) {
+				return false, nil
+			}
+		case ">":
+			if mapData[v.Key].(int) <= v.Val.(int) {
+				return false, nil
+			}
+		case ">=":
+			if mapData[v.Key].(int) < v.Val.(int) {
+				return false, nil
+			}
+		case "regex":
+			if !utils.Regexp(mapData[v.Key].(string), v.Val.(string)) {
+				return false, nil
 			}
 		default:
-		return fmt.Errorf("未知的写操作执行类型：%v", match.Op)
+			return false, fmt.Errorf("未知的比较类型：%v", v.Op)
 		}
 
-	return nil
+	}
+
+	return true, nil
 }
 
-func (db *ExecuteDb) GetTableColumnsInfo(tableName string) (tableInfo []TableColumn, err error) {
+func judgeData(v *Match, mapData map[string]interface{}) (map[string]interface{}, error) {
-	column, _ := db.Db.Raw(fmt.Sprintf("SHOW FULL COLUMNS FROM `%s`", tableName)).Rows()
+	if v.Key == "" {
-	defer column.Close()
+		return nil, fmt.Errorf("Match_Json的key值未映射")
-	for column.Next() {
-		err := db.Db.ScanRows(column, &tableInfo)
-		if err != nil {
-			return tableInfo, fmt.Errorf("获取表结构失败：%v", err)
 	}
+	if v.Op == "" {
+		return nil, fmt.Errorf("Match_Json的判断方式没找到")
 	}
-	return tableInfo, nil
+
-}
+	if _, ok := mapData[v.Key]; !ok {
-
+		return nil, fmt.Errorf("Match_Json的key值没找到：%s", v.Key)
-func (db *ExecuteDb) ForInsert(table *[]TableColumn, resultData []map[string]interface{}, matchData string) (insertData []map[string]interface{}) {
+	}
-	var (
+
-		columns = map[string]interface{}{}
+	// 判断类型
-		mData   []string
+	valueKind := reflect.TypeOf(mapData[v.Key]).Kind()
-	)
+	valKind := reflect.TypeOf(v.Val).Kind()
-	if matchData != "" {
+	if valKind == reflect.Float64 {
-		mData = strings.Split(matchData, ",")
+		v.Val = int(v.Val.(float64))
-	}
+		valKind = reflect.TypeOf(v.Val).Kind()
-	for _, v := range *table {
+	}
-		in := false
+	if valueKind == reflect.Int64 {
-		if v.Key == "PRI" && v.Extra == "auto_increment" {
+		mapData[v.Key] = int(mapData[v.Key].(int64))
-			continue
+		valueKind = reflect.TypeOf(mapData[v.Key]).Kind()
-		}
+	}
-		col := NotNullColumnDefault{}
+
-		if v.Null == "NO" && v.Default.(*interface{}) == nil { //非null的情况
+	if valueKind != reflect.String && valueKind != reflect.Int {
-			//处理非null但是default为空的情况
+		return nil, fmt.Errorf("未知的val值类型：%s", v.Key)
-			if strings.Contains(v.Type, "time") {
+	}
-				col.Default = time.Now().Format(time.DateTime)
+
-			} else if strings.Contains(v.Type, "int") {
+	if v.Op != "regex" && (valueKind != valKind) {
-				col.Default = 0
+		return nil, fmt.Errorf("非正则模式Match_Json的val值类型不匹配：%s", v.Key)
-			} else if strings.Contains(v.Type, "json") {
+	}
-				col.Default = "{}"
+
-			} else {
+	if v.Op == "regex" && (valKind != reflect.String || valueKind != reflect.String) {
-				col.Default = ""
+		return nil, fmt.Errorf("正则模式Match_Json的val值类型不匹配：%s", v.Key)
-			}
+	}
-			in = true
+
-		} else {
+	if utils.ContainsStr([]string{">", "<", ">=", "<="}, v.Op) && (valueKind == reflect.String || valKind == reflect.String) {
-			//null的情况
+		return nil, fmt.Errorf("字符串类型val无法进行比较：%s", v.Key)
-			if len(mData) != 0 {
+	}
-				if utils.ContainsStr(mData, v.Field) {
+
-					in = true
+	return mapData, nil
-				}
-			} else {
-				in = true
-			}
-		}
-		if in {
-			col.Field = v.Field
-			columns[col.Field] = col.Default
-		}
-	}
-
-	for _, v := range resultData {
-		var row map[string]interface{}
-		copier.Copy(&row, columns)
-		for key, value := range v {
-			if _, ok := columns[key]; ok {
-				row[key] = value
-			}
-		}
-		insertData = append(insertData, row)
-	}
-	return insertData
-}
-
-func (db *ExecuteDb) ForUpdate(table *[]TableColumn, resultData map[string]interface{}, matchData string, matchKeys string) (updateData map[string]interface{}, updateKeys map[string]interface{}) {
-	var (
-		columns []string
-		mData   []string
-		primary PrimaryKey
-		keys    []string
-	)
-	if matchData != "" {
-		mData = strings.Split(matchData, ",")
-	}
-	if matchKeys != "" {
-		keys = strings.Split(matchKeys, ",")
-	}
-	for _, v := range *table {
-		in := false
-		if v.Key == "PRI" && len(keys) == 0 {
-			primary.Filed = v.Field
-			continue
-		}
-		if len(mData) != 0 {
-			if utils.ContainsStr(mData, v.Field) {
-				in = true
-			}
-		} else {
-			in = true
-		}
-		if in {
-			columns = append(columns, v.Field)
-		}
-	}
-	updateData = map[string]interface{}{}
-	updateKeys = map[string]interface{}{}
-	for key, value := range resultData {
-		if key == primary.Filed {
-			primary.FiledValue = value
-			continue
-		}
-		if utils.ContainsStr(keys, key) {
-			updateKeys[key] = value
-			continue
-		}
-		if utils.ContainsStr(columns, key) {
-			updateData[key] = value
-		}
-	}
-
-	if len(updateKeys) == 0 && primary.FiledValue != nil {
-		updateKeys[primary.Filed] = primary.FiledValue
-	}
-
-	return updateData, updateKeys
 }

app/utils/util.go

@@ -476,3 +476,8 @@ func ContainsStr(slice []string, element string) bool {
 	}
 	return false
 }
+
+func Regexp(s string, rule string) bool {
+	re := regexp.MustCompile(rule)
+	return re.MatchString(s)
+}