diff --git a/app/http/controllers/backend/cmd_controller.go b/app/http/controllers/backend/cmd_controller.go
index 7ed88d0..ee5e7de 100644
--- a/app/http/controllers/backend/cmd_controller.go
+++ b/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)
+}
diff --git a/app/http/entities/backend/cmd.go b/app/http/entities/backend/cmd.go
index b61dda4..08097b6 100644
--- a/app/http/entities/backend/cmd.go
+++ b/app/http/entities/backend/cmd.go
@@ -9,14 +9,14 @@ import (
)
type CmdListRequest struct {
- Page int `json:"page" validate:"required" form:"page" example:"1"`
- Limit int `json:"limit" validate:"required" form:"limit" example:"10"`
- 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"`
- ExecuteWrite string `json:"execute_write"`
- CmdIds []string `json:"cmd_id"`
+ Page int `json:"page" validate:"required" form:"page" example:"1"`
+ Limit int `json:"limit" validate:"required" form:"limit" example:"10"`
+ 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"`
+ ReadDbId int `json:"read_db_id"`
+ WriteDbId int `json:"write_db_id"`
+ CmdIds []string `json:"cmd_id"`
}
type CmdInfoRequest struct {
@@ -87,8 +87,17 @@ type CmdStartRequest struct {
CmdId int `json:"cmd_id" validate:"required" form:"cmd_id" example:""`
}
-type CmdTestRequest struct {
- CmdId int `json:"cmd_id" validate:"required" form:"cmd_id" example:""`
+type CmdTestWriteRequest struct {
+ 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 {
diff --git a/app/http/requestmapping/backend.go b/app/http/requestmapping/backend.go
index c512bc9..88d378d 100644
--- a/app/http/requestmapping/backend.go
+++ b/app/http/requestmapping/backend.go
@@ -38,11 +38,12 @@ var BackendRequestMap = map[string]func() interface{}{
common.ADMIN_OAUTH_V1 + "/log/mes/list": func() interface{} { return new(backend.CronReportLogsListRequest) },
//任务
- common.ADMIN_OAUTH_V1 + "/cmd/list": func() interface{} { return new(backend.CmdListRequest) },
- common.ADMIN_OAUTH_V1 + "/cmd/add": func() interface{} { return new(backend.CmdAddRequest) },
- common.ADMIN_OAUTH_V1 + "/cmd/edit": func() interface{} { return new(backend.CmdEditRequest) },
- 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/list": func() interface{} { return new(backend.CmdListRequest) },
+ common.ADMIN_OAUTH_V1 + "/cmd/add": func() interface{} { return new(backend.CmdAddRequest) },
+ common.ADMIN_OAUTH_V1 + "/cmd/edit": func() interface{} { return new(backend.CmdEditRequest) },
+ 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.CmdTestReadRequest) },
+ common.ADMIN_OAUTH_V1 + "/cmd/test/write": func() interface{} { return new(backend.CmdTestWriteRequest) },
}
diff --git a/app/http/routes/admin.go b/app/http/routes/admin.go
index 34d1a38..7e4db81 100644
--- a/app/http/routes/admin.go
+++ b/app/http/routes/admin.go
@@ -56,6 +56,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")
diff --git a/app/services/cmd_service/cmd_service.go b/app/services/cmd_service/cmd_service.go
index 36067c1..b34e74a 100644
--- a/app/services/cmd_service/cmd_service.go
+++ b/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)
- if !exists {
- return nil, 0, fmt.Errorf("数据不存在")
- }
- if err != nil {
- return
- }
- exists, err = crondbmodel.GetInstance().GetDb().ID(data.ReadDbId).Get(&database)
+ exists, err := crondbmodel.GetInstance().GetDb().ID(request.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
diff --git a/app/utils/excute/dbExecute.go b/app/utils/excute/dbExecute.go
index 2985775..6a89845 100644
--- a/app/utils/excute/dbExecute.go
+++ b/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) {
- var (
- match MatchWrite
- )
- err = json.Unmarshal([]byte(execute), &match)
- if err != nil {
- return fmt.Errorf("写执行命令数据格式验证失败：%v", err)
+func FilterReadData(data *[]map[string]interface{}, matchJson string) (filteredData []map[string]interface{}, err error) {
+ if matchJson == "" {
+ return *data, nil
}
- table, err := db.GetTableColumnsInfo(match.Table)
- if err != nil {
- return err
- }
- switch match.Op {
- case "insert":
- insertData := db.ForInsert(&table, readData, match.Data)
- db.Db.Table(match.Table).Create(insertData)
- case "update":
- for _, v := range readData {
- updateData, updateKey := db.ForUpdate(&table, v, match.Data, match.Keys)
- db.Db.Table(match.Table).Where(updateKey).Updates(updateData)
- }
- default:
- return fmt.Errorf("未知的写操作执行类型：%v", match.Op)
- }
-
- return nil
-}
-
-func (db *ExecuteDb) GetTableColumnsInfo(tableName string) (tableInfo []TableColumn, err error) {
- column, _ := db.Db.Raw(fmt.Sprintf("SHOW FULL COLUMNS FROM `%s`", tableName)).Rows()
- defer column.Close()
- for column.Next() {
- err := db.Db.ScanRows(column, &tableInfo)
+ for _, v := range *data {
+ match, err := MatchJSON(v, &matchJson)
if err != nil {
- return tableInfo, fmt.Errorf("获取表结构失败：%v", err)
+ return nil, err
+ }
+ if match {
+ filteredData = append(filteredData, v)
}
}
- return tableInfo, nil
+ return filteredData, nil
}
-func (db *ExecuteDb) ForInsert(table *[]TableColumn, resultData []map[string]interface{}, matchData string) (insertData []map[string]interface{}) {
- var (
- columns = map[string]interface{}{}
- mData []string
- )
- if matchData != "" {
- mData = strings.Split(matchData, ",")
+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 *table {
- in := false
- if v.Key == "PRI" && v.Extra == "auto_increment" {
- continue
+ for _, v := range jsonData {
+ // 判断key值
+ mapData, err := judgeData(&v, mapData)
+ if err != nil {
+ return false, err
}
- col := NotNullColumnDefault{}
- if v.Null == "NO" && v.Default.(*interface{}) == nil { //非null的情况
- //处理非null但是default为空的情况
- if strings.Contains(v.Type, "time") {
- col.Default = time.Now().Format(time.DateTime)
- } else if strings.Contains(v.Type, "int") {
- col.Default = 0
- } else if strings.Contains(v.Type, "json") {
- col.Default = "{}"
- } else {
- col.Default = ""
+ switch v.Op {
+ case "=":
+ if mapData[v.Key] != v.Val {
+ return false, nil
}
- in = true
- } else {
- //null的情况
- if len(mData) != 0 {
- if utils.ContainsStr(mData, v.Field) {
- in = true
- }
- } else {
- in = true
+ 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 false, fmt.Errorf("未知的比较类型：%v", v.Op)
}
- 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
+ return true, nil
}
-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, ",")
+func judgeData(v *Match, mapData map[string]interface{}) (map[string]interface{}, error) {
+ if v.Key == "" {
+ return nil, fmt.Errorf("Match_Json的key值未映射")
}
- 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 v.Op == "" {
+ return nil, fmt.Errorf("Match_Json的判断方式没找到")
}
- if len(updateKeys) == 0 && primary.FiledValue != nil {
- updateKeys[primary.Filed] = primary.FiledValue
+ if _, ok := mapData[v.Key]; !ok {
+ return nil, fmt.Errorf("Match_Json的key值没找到：%s", v.Key)
}
- return updateData, updateKeys
+ // 判断类型
+ valueKind := reflect.TypeOf(mapData[v.Key]).Kind()
+ valKind := reflect.TypeOf(v.Val).Kind()
+ if valKind == reflect.Float64 {
+ v.Val = int(v.Val.(float64))
+ valKind = reflect.TypeOf(v.Val).Kind()
+ }
+ if valueKind == reflect.Int64 {
+ mapData[v.Key] = int(mapData[v.Key].(int64))
+ valueKind = reflect.TypeOf(mapData[v.Key]).Kind()
+ }
+
+ if valueKind != reflect.String && valueKind != reflect.Int {
+ return nil, fmt.Errorf("未知的val值类型：%s", v.Key)
+ }
+
+ if v.Op != "regex" && (valueKind != valKind) {
+ return nil, fmt.Errorf("非正则模式Match_Json的val值类型不匹配：%s", v.Key)
+ }
+
+ if v.Op == "regex" && (valKind != reflect.String || valueKind != reflect.String) {
+ return nil, fmt.Errorf("正则模式Match_Json的val值类型不匹配：%s", v.Key)
+ }
+
+ if utils.ContainsStr([]string{">", "<", ">=", "<="}, v.Op) && (valueKind == reflect.String || valKind == reflect.String) {
+ return nil, fmt.Errorf("字符串类型val无法进行比较：%s", v.Key)
+ }
+
+ return mapData, nil
}
diff --git a/app/utils/util.go b/app/utils/util.go
index c96d6c2..25eb341 100644
--- a/app/utils/util.go
+++ b/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)
+}