package Task import ( "bzd_server/conf" "bzd_server/lib" "bzd_server/logs" "fmt" _ "github.com/astaxie/beego/cache/redis" "github.com/beego/beego/v2/adapter/orm" orm2 "github.com/beego/beego/v2/client/orm" _ "github.com/go-sql-driver/mysql" "strconv" "strings" "sync" "time" ) // 模板 type TaskData struct { Id int `orm:"column(ID);size(11);auto;pk"` T_id string `orm:"size(256);null"` // 标题 T_t float32 `orm:"size(10);null"` // 温度 T_rh float32 `orm:"size(10);null"` // 湿度 T_time time.Time `orm:"type(timestamp);null;"` // 采集时间 CreateTime time.Time `orm:"column(create_time);type(timestamp);null;auto_now_add"` //auto_now_add 第一次保存时才设置时间 UpdateTime time.Time `orm:"column(update_time);type(timestamp);null;auto_now"` //auto_now 每次 model 保存时都会对时间自动更新 } func (t *TaskData) TableName() string { return "TaskData" // 数据库名称 // ************** 替换 FormulaList ************** } func init() { } //创建数据库 Device.CREATE_TaskData("") func CREATE_TaskData(alias_name, T_task_id string) bool { o := orm2.NewOrmUsingDB(alias_name) sql := "DROP TABLE IF EXISTS `Z_TaskData_" + T_task_id + "`" o.Raw(sql).Exec() sql = "CREATE TABLE IF NOT EXISTS `Z_TaskData_" + T_task_id + "` ( " + " `ID` int(11) NOT NULL AUTO_INCREMENT," + " `t_sn` varchar(256) CHARACTER SET utf8mb4 COLLATE utf8mb4_general_ci NULL DEFAULT NULL," + " `t_id` int(11) NULL DEFAULT NULL," + " `t_t` float(6, 1) NULL DEFAULT NULL," + " `t_rh` float(6, 1) NULL DEFAULT NULL," + " `t_time` datetime(0) NULL DEFAULT NULL," + " PRIMARY KEY (`ID`) USING BTREE" + ") ENGINE = InnoDB AUTO_INCREMENT = 5 CHARACTER SET = utf8mb4 COLLATE = utf8mb4_general_ci ROW_FORMAT = Dynamic;" _, err := o.Raw(sql).Exec() if err != nil { return false } return true } // ---------------- 特殊方法 ------------------- // 获取 ById //func Read_List_ById(id int) (r TaskData) { // o := orm.NewOrm() // // return r //} // 清空 func Truncate_TaskData(alias_name, T_task_id string) bool { o := orm2.NewOrmUsingDB(alias_name) sql := "truncate table Z_TaskData_" + T_task_id logs.Println(sql) o.Raw(sql).Exec() return true } // 添加 func Import_TaskData(Sn string, T_id int, T_task_id string, Time_start string, Time_end string) bool { //if conf.Server_test { // return true //} o := orm.NewOrm() // 开始插入数据 sql := "insert into Z_TaskData_" + T_task_id + "(t_id,t_t,t_rh,t_time) select t_id,t_t,t_rh,t_time from Z_DeviceData_" + Sn sql = sql + " WHERE t_time >= '" + Time_start + "' AND t_time <= '" + Time_end + "'" // 这里有时间优化 用于一次 prepare 多次 exec,以提高批量执行的速度 logs.Println(sql) res, err := o.Raw(sql).Exec() if err != nil { fmt.Println(err) return false } // 强行替换 ID sql = "UPDATE Z_TaskData_" + T_task_id + " SET `t_sn` = '" + Sn + "',`t_id` = " + strconv.Itoa(T_id) + " WHERE `t_sn` IS NULL" // 设备 ID //sql = "UPDATE Z_TaskData_"+T_task_id+" SET `t_sn` = '"+Sn+"' WHERE `t_sn` IS NULL" logs.Println(sql) o.Raw(sql).Exec() res.RowsAffected() return true } type TaskData_ struct { ID int `orm:"column(ID);size(100);null"` // ID T_sn string `orm:"column(t_sn);size(256);null"` // 标题 T_id int `orm:"column(t_id);size(10);null"` // ID T_t float32 `orm:"column(t_t);size(10);null"` // 温度 T_rh float32 `orm:"column(t_rh);size(10);null"` // 湿度 T_time string `orm:"column(t_times);null;"` // 采集时间 } type TaskDataClass_ struct { T_sn string `orm:"column(t_sn);size(256);null"` // 标题 T_id int `orm:"column(t_id);size(10);null"` // ID } // func Read_TaskData_ById_List(T_task_id string, SN string, T_id int, Time_start_ string, Time_end_ string, page int, page_z int) ([]TaskData_, int64) { o := orm.NewOrm() var maps []TaskData_ var maps_z []orm2.ParamsList pagez := page_z var offset int if page <= 1 { offset = 0 } else { page -= 1 offset = page * pagez } sql_condition := "" if len(Time_start_) > 1 { sql_condition += " t_time >= '" + Time_start_ + "' AND " } if len(Time_end_) > 1 { sql_condition += " t_time <= '" + Time_end_ + "' AND " } if T_id != -1 { sql_condition += " t_id = " + strconv.Itoa(T_id) + " AND " } sql_condition += " t_sn = '" + SN + "' " sql := "SELECT COUNT(ID) FROM Z_TaskData_" + T_task_id + " WHERE " + sql_condition fmt.Println(sql) _, err := o.Raw(sql).ValuesList(&maps_z) if err != nil { return maps, 0 } if len(maps_z) == 0 { return maps, 0 } //fmt.Println("maps_z;",maps_z[0][0]) sql = "SELECT ID,t_sn,t_id,t_t,t_rh,DATE_FORMAT(t_time,'%Y-%c-%d %H:%i:%s') AS t_times,t_time FROM Z_TaskData_" + T_task_id + " WHERE " + sql_condition + " ORDER BY t_time DESC " if page_z != 9999 { sql = sql + " LIMIT " + strconv.Itoa(offset) + "," + strconv.Itoa(pagez) } fmt.Println(sql) _, err = o.Raw(sql).QueryRows(&maps) //value, _ := strconv.ParseFloat(fmt.Sprintf("%.2f", cnt), 64) key, _ := strconv.Atoi(maps_z[0][0].(string)) return maps, int64(key) } // func Read_TaskData_ById_List_(T_task_id string, SN string, T_id int) []TaskData_ { o := orm.NewOrm() var maps []TaskData_ sql_condition := "" sql_condition += " t_id = " + strconv.Itoa(T_id) + " AND " sql_condition += " t_sn LIKE '%" + SN + "%' " //fmt.Println("maps_z;",maps_z[0][0]) sql := "SELECT ID,t_sn,t_id,t_t,t_rh,DATE_FORMAT(t_time,'%Y-%c-%d %H:%i') AS t_times,t_time FROM Z_TaskData_" + T_task_id + " WHERE " + sql_condition + " ORDER BY t_time " fmt.Println(sql) o.Raw(sql).QueryRows(&maps) return maps } // func Read_TaskData_ById_ClassList(T_task_id string) []TaskDataClass_ { o := orm.NewOrm() var maps []TaskDataClass_ //fmt.Println("maps_z;",maps_z[0][0]) sql := "SELECT DISTINCT t_sn,t_id FROM Z_TaskData_" + T_task_id + " ORDER BY t_id " fmt.Println(sql) o.Raw(sql).QueryRows(&maps) return maps } // 添加 func Add_TaskData(T_task_id string, T_sn string, T_id string, T_t string, T_rh string, T_time string) bool { o := orm.NewOrm() // 开始插入数据 // //sql := "INSERT INTO Z_TaskData_" + T_task_id + " (`t_sn`, `t_id`, `t_time`) " + // "VALUES ('" + T_sn + "', " + T_id + ", '" + T_time + "') " + // "ON DUPLICATE KEY UPDATE t_t=" + T_t + ", t_rh="+ T_rh +";" // 去重复数据 //sql := "DELETE FROM Z_TaskData_" + T_task_id + " WHERE " + " t_id = " + T_id + " AND "+ " t_sn = '"+T_sn+"' " + "AND t_time = '" + T_time + "' " // //// 这里有时间优化 用于一次 prepare 多次 exec,以提高批量执行的速度 //fmt.Println(sql) //res, err := o.Raw(sql).Exec() //if err != nil { // fmt.Println(err) // return false //} //res.RowsAffected() sql := "INSERT INTO Z_TaskData_" + T_task_id + " (`t_sn`, `t_id`, `t_t`, `t_rh`, `t_time`) " + "VALUES ('" + T_sn + "', " + T_id + ", " + T_t + "," + T_rh + ", '" + T_time + "')" fmt.Println(sql) res, err := o.Raw(sql).Exec() if err != nil { fmt.Println(err) return false } res.RowsAffected() //fmt.Println("mysql row affected nums: ", num) return true } // 修改 func Up_TaskData(T_task_id string, Id string, T_t string, T_rh string, T_time string) bool { o := orm.NewOrm() SET_str := " SET " if len(T_t) > 0 { SET_str = SET_str + " t_t = " + T_t + " ," } if len(T_rh) > 0 { SET_str = SET_str + " t_rh = " + T_rh + " ," } if len(T_time) > 0 { SET_str = SET_str + " t_time = '" + T_time + "' ," } if len(SET_str) > 7 { SET_str = SET_str[:(len(SET_str) - 1)] } // 开始插入数据 UPDATE `cold_verify`.`Z_TaskData_d8qMyeXLzIxn` SET `t_t` = 20.2 WHERE `ID` = 69 sql := "UPDATE Z_TaskData_" + T_task_id + " " + SET_str + " WHERE ID = " + Id // 这里有时间优化 用于一次 prepare 多次 exec,以提高批量执行的速度 //fmt.Println(sql) res, err := o.Raw(sql).Exec() if err != nil { fmt.Println(err) return false } res.RowsAffected() //fmt.Println("mysql row affected nums: ", num) return true } // 添加 func Del_TaskData(T_task_id string, Id string) bool { o := orm.NewOrm() // 开始插入数据 UPDATE `cold_verify`.`Z_TaskData_d8qMyeXLzIxn` SET `t_t` = 20.2 WHERE `ID` = 69 sql := "DELETE FROM Z_TaskData_" + T_task_id + " WHERE ID = " + Id // 这里有时间优化 用于一次 prepare 多次 exec,以提高批量执行的速度 logs.Println(sql) res, err := o.Raw(sql).Exec() if err != nil { fmt.Println(err) return false } res.RowsAffected() //fmt.Println("mysql row affected nums: ", num) return true } // 添加 func Del_TaskData_t_id(T_task_id string, Id string) bool { o := orm.NewOrm() // 开始插入数据 UPDATE `cold_verify`.`Z_TaskData_d8qMyeXLzIxn` SET `t_t` = 20.2 WHERE `ID` = 69 sql := "DELETE FROM Z_TaskData_" + T_task_id + " WHERE t_id = " + Id // 这里有时间优化 用于一次 prepare 多次 exec,以提高批量执行的速度 logs.Println(sql) res, err := o.Raw(sql).Exec() if err != nil { fmt.Println(err) return false } res.RowsAffected() //fmt.Println("mysql row affected nums: ", num) return true } // 检查导出表总数 func Check_TaskData_Num(T_task_id string) bool { o1 := orm2.NewOrmUsingDB(conf.Local_AliasName) o2 := orm2.NewOrmUsingDB(conf.Server_AliasName) var maps_z1, maps_z2 []orm2.ParamsList sql := "SELECT COUNT(ID) FROM Z_TaskData_" + T_task_id fmt.Println(sql) _, err := o1.Raw(sql).ValuesList(&maps_z1) if err != nil { return false } _, err = o2.Raw(sql).ValuesList(&maps_z2) if err != nil { return false } if len(maps_z1) != len(maps_z2) { return false } return true } func Read_TaskData_List(alias_name, T_task_id string, page int, page_z int) []*TaskData_ { o := orm2.NewOrmUsingDB(alias_name) var maps []*TaskData_ pagez := page_z var offset int if page <= 1 { offset = 0 } else { page -= 1 offset = page * pagez } sql := "SELECT ID,t_sn,t_id,t_t,t_rh,DATE_FORMAT(t_time,'%Y-%c-%d %H:%i:%s') AS t_times FROM Z_TaskData_" + T_task_id + " LIMIT " + strconv.Itoa(offset) + "," + strconv.Itoa(pagez) fmt.Println(sql) _, err := o.Raw(sql).QueryRows(&maps) if err != nil { logs.Println("查询失败 Z_TaskData_" + T_task_id) } return maps } func Add_TaskData_List(alias_name, T_task_id string, TaskList []*TaskData_) (int, int) { var wg sync.WaitGroup limiter := make(chan bool, 80) succeeds := 0 fails := 0 o := orm2.NewOrmUsingDB(alias_name) for _, val := range TaskList { wg.Add(1) limiter <- true go func(TaskData *TaskData_) { defer wg.Done() //if Add_TaskData(T_task_id, TaskData.T_sn, strconv.Itoa(TaskData.T_id), fmt.Sprintf("%.2f", TaskData.T_t), // fmt.Sprintf("%.2f", TaskData.T_rh), TaskData.T_time) { // succeeds++ //} else { // fails++ //} sql := "INSERT INTO Z_TaskData_" + T_task_id + " (`t_sn`, `t_id`, `t_t`, `t_rh`, `t_time`) " + "VALUES ('" + TaskData.T_sn + "', " + strconv.Itoa(TaskData.T_id) + ", " + fmt.Sprintf("%.2f", TaskData.T_t) + "," + fmt.Sprintf("%.2f", TaskData.T_rh) + ", '" + TaskData.T_time + "')" //fmt.Println(sql) _, err := o.Raw(sql).Exec() if err != nil { fails++ } else { succeeds++ } <-limiter }(val) } wg.Wait() return succeeds, fails } func Dump_TaskData(T_task_id, root, password, url_port, database, sql_file string) (string, error) { // url_port 127.0.0.1:3306 // mysql8.0 以上加 --column-statistics=0 // mysqldump --column-statistics=0 -uroot -proot -h127.0.0.1 -P3306 cold_verify Z_TaskData_ixEfo5zk2Oeb> /Data/Z_TaskData_ixEfo5zk2Oeb.sql v := Read_Local_Mysql_Version() host_port := strings.Split(url_port, ":") table_name := "Z_TaskData_" + T_task_id org := "mysqldump " if v >= 8 { org += "--column-statistics=0 " } //--no-create-info 只导出数据,而不添加 CREATE TABLE 语句。 org = org + fmt.Sprintf("-u%s -p%s -h%s -P%s %s %s> %s --no-create-info", root, password, host_port[0], host_port[1], database, table_name, sql_file) logs.Println(org) _, err := lib.Command(org) return org, err } func Insert_TaskData(root, password, url_port, database, sql_file string) (string, error) { // url_port 127.0.0.1:3306 // mysql -uroot -proot -h127.0.0.1 -P3306 cold_verify_local < /data/Z_TaskData_ixEfo5zk2Oeb.sql host_port := strings.Split(url_port, ":") org := fmt.Sprintf("mysql -u%s -p%s -h%s -P%s %s < %s", root, password, host_port[0], host_port[1], database, sql_file) logs.Println(org) _, err := lib.Command(org) return org, err } func Read_Local_Mysql_Version() int { o := orm2.NewOrmUsingDB(conf.Local_AliasName) var params []orm2.Params o.Raw("select version();").Values(¶ms) version, _ := strconv.Atoi(params[0]["version()"].(string)[0:1]) return version }