ColdVerify_local/models/Task/TaskData.go

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(&params)
	version, _ := strconv.Atoi(params[0]["version()"].(string)[0:1])
	return version
}