TaskData.go 30 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945946947948949950951952953954955956957958959960961962963964965966967968969970971972973974975976977978979980981982983984985986987988989990991992993994995996997998999
  1. package Task
  2. import (
  3. "ColdVerify_local/conf"
  4. "ColdVerify_local/lib"
  5. "ColdVerify_local/logs"
  6. "encoding/json"
  7. "fmt"
  8. "github.com/astaxie/beego/cache"
  9. _ "github.com/astaxie/beego/cache/redis"
  10. "github.com/beego/beego/v2/adapter/orm"
  11. orm2 "github.com/beego/beego/v2/client/orm"
  12. _ "github.com/go-sql-driver/mysql"
  13. "strconv"
  14. "strings"
  15. "time"
  16. )
  17. // 模板
  18. type TaskData struct {
  19. Id int `orm:"column(ID);size(11);auto;pk"`
  20. T_id string `orm:"size(256);null"` // 标题
  21. T_t float32 `orm:"size(10);null"` // 温度
  22. T_rh float32 `orm:"size(10);null"` // 湿度
  23. T_time time.Time `orm:"type(timestamp);null;"` // 采集时间
  24. 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 保存时都会对时间自动更新
  25. }
  26. func (t *TaskData) TableName() string {
  27. return "task_data" // 数据库名称 // ************** 替换 FormulaList **************
  28. }
  29. var redisCache_TaskData cache.Cache
  30. func init() {
  31. //orm2.Debug = true
  32. config := fmt.Sprintf(`{"key":"%s","conn":"%s","dbNum":"%s","password":"%s"}`,
  33. "redis_"+"TaskData", conf.Redis_address, conf.Redis_dbNum, conf.Redis_password)
  34. logs.Println(config)
  35. var err error
  36. redisCache_TaskData, err = cache.NewCache("redis", config)
  37. if err != nil || redisCache_TaskData == nil {
  38. errMsg := "failed to init redis"
  39. logs.Println(errMsg, err)
  40. }
  41. }
  42. type TaskDataJPG struct {
  43. State int `json:"state"` //1:生成中 2:已完成 3:失败
  44. Msg string `json:"msg"`
  45. Url string `json:"url"` //url
  46. }
  47. // ---------------- Redis -------------------
  48. // Redis_Set(m.T_sn,m) // Redis 更新缓存
  49. func Redis_TaskDataJPG_Set(key string, r TaskDataJPG) (err error) {
  50. //json序列化
  51. str, err := json.Marshal(r)
  52. if err != nil {
  53. logs.Error(lib.FuncName(), err)
  54. return
  55. }
  56. err = redisCache_TaskData.Put(key, str, 5*time.Minute)
  57. if err != nil {
  58. logs.Println("set key:", key, ",value:", str, err)
  59. }
  60. return
  61. }
  62. // if r,is :=Redis_Get(T_sn);is{
  63. // return r,nil
  64. // }
  65. func Redis_TaskDataJPG_Get(key string) (r TaskDataJPG, is bool) {
  66. if redisCache_TaskData.IsExist(key) {
  67. logs.Println("找到key:", key)
  68. v := redisCache_TaskData.Get(key)
  69. json.Unmarshal(v.([]byte), &r)
  70. return r, true
  71. }
  72. logs.Println("没有 找到key:", key)
  73. return TaskDataJPG{}, false
  74. }
  75. func Redis_TaskDataJPG_Del(key string) (err error) {
  76. err = redisCache_TaskData.Delete(key)
  77. return
  78. }
  79. // 创建数据库 Device.CREATE_TaskData("")
  80. func CREATE_TaskData(alias_name, T_task_id string) bool {
  81. o := orm2.NewOrmUsingDB(alias_name)
  82. sql := "DROP TABLE IF EXISTS `z_task_data_" + T_task_id + "`"
  83. o.Raw(sql).Exec()
  84. sql = "CREATE TABLE IF NOT EXISTS `z_task_data_" + T_task_id + "` ( " +
  85. " `ID` int(11) NOT NULL AUTO_INCREMENT," +
  86. " `t_sn` varchar(256) CHARACTER SET utf8mb4 COLLATE utf8mb4_general_ci NULL DEFAULT NULL," +
  87. " `t_id` varchar(256) DEFAULT NULL," +
  88. " `t_t` float(6, 1) NULL DEFAULT NULL," +
  89. " `t_rh` float(6, 1) NULL DEFAULT NULL," +
  90. " `t_time` datetime(0) NULL DEFAULT NULL," +
  91. " PRIMARY KEY (`ID`) USING BTREE," +
  92. " KEY `t_sn` (`t_sn`)," +
  93. " KEY `t_id` (`t_id`)," +
  94. " KEY `t_time` (`t_time`)" +
  95. ") ENGINE = InnoDB AUTO_INCREMENT = 5 CHARACTER SET = utf8mb4 COLLATE = utf8mb4_general_ci ROW_FORMAT = Dynamic;"
  96. _, err := o.Raw(sql).Exec()
  97. if err != nil {
  98. return false
  99. }
  100. //sqlIndex := "ALTER TABLE `z_task_data_" + T_task_id + "` add unique index(`t_sn`,`t_id`,`t_time`);"
  101. //logs.Debug(sqlIndex)
  102. //o.Raw(sqlIndex).Exec()
  103. return true
  104. }
  105. // ---------------- 特殊方法 -------------------
  106. // 清空
  107. func Truncate_TaskData(alias_name, T_task_id string) bool {
  108. o := orm2.NewOrmUsingDB(alias_name)
  109. sql := "truncate table z_task_data_" + T_task_id
  110. logs.Println(sql)
  111. _, err := o.Raw(sql).Exec()
  112. if err != nil {
  113. logs.Error(lib.FuncName(), err)
  114. return false
  115. }
  116. return true
  117. }
  118. type TaskData_ struct {
  119. ID int `orm:"column(ID);size(100);null"` // ID
  120. T_sn string `orm:"column(t_sn);size(256);null"` // sn
  121. T_id string `orm:"column(t_id);size(256);null"` // 标题
  122. T_t float32 `orm:"column(t_t);size(10);null"` // 温度
  123. T_rh float32 `orm:"column(t_rh);size(10);null"` // 湿度
  124. T_time string `orm:"column(t_times);null;"` // 采集时间
  125. T_Certificate_sn string `orm:"size(256);null"` // 证书编号
  126. }
  127. type TaskDataClass_ struct {
  128. T_sn string `orm:"column(t_sn);size(256);null"` // 标题
  129. T_id string `orm:"column(t_id);size(256);null"` // 名称
  130. }
  131. func Read_TaskData_ById_List(T_task_id string, SN string, T_id string, Time_start_ string, Time_end_ string, page int, page_z int) ([]TaskData_, int64) {
  132. o := orm.NewOrm()
  133. var maps []TaskData_
  134. var maps_z []orm2.ParamsList
  135. pagez := page_z
  136. var offset int
  137. if page <= 1 {
  138. offset = 0
  139. } else {
  140. page -= 1
  141. offset = page * pagez
  142. }
  143. sql_condition := ""
  144. if len(Time_start_) > 1 {
  145. sql_condition += " AND t_time >= '" + Time_start_ + "'"
  146. }
  147. if len(Time_end_) > 1 {
  148. sql_condition += " AND t_time <= '" + Time_end_ + "'"
  149. }
  150. if len(T_id) > 0 {
  151. sql_condition += " AND t_id = '" + T_id + "'"
  152. }
  153. if len(SN) > 0 {
  154. sql_condition += " AND t_sn = '" + SN + "'"
  155. }
  156. if len(sql_condition) > 0 {
  157. sql_condition = " WHERE " + strings.TrimLeft(sql_condition, " AND ")
  158. }
  159. sql := "SELECT COUNT(ID) FROM z_task_data_" + T_task_id + sql_condition
  160. fmt.Println(sql)
  161. _, err := o.Raw(sql).ValuesList(&maps_z)
  162. if err != nil {
  163. return maps, 0
  164. }
  165. if len(maps_z) == 0 {
  166. return maps, 0
  167. }
  168. //fmt.Println("maps_z;",maps_z[0][0])
  169. sql = "SELECT ID,t_sn,t_id,t_t,t_rh,DATE_FORMAT(t_time,'%Y-%m-%d %H:%i:%s') AS t_times,t_time FROM z_task_data_" + T_task_id + sql_condition + " ORDER BY t_time DESC"
  170. if page_z != 9999 {
  171. sql = sql + " LIMIT " + strconv.Itoa(offset) + "," + strconv.Itoa(pagez)
  172. }
  173. fmt.Println(sql)
  174. _, err = o.Raw(sql).QueryRows(&maps)
  175. if err != nil {
  176. logs.Error(lib.FuncName(), err)
  177. }
  178. //value, _ := strconv.ParseFloat(fmt.Sprintf("%.2f", cnt), 64)
  179. key, _ := strconv.Atoi(maps_z[0][0].(string))
  180. return maps, int64(key)
  181. }
  182. func Read_TaskData_ById_List_AES(T_task_id string, SN string, T_id string, Time_start_ string, Time_end_ string, page int, page_z int) ([]TaskData_, int64) {
  183. o := orm.NewOrm()
  184. var maps []TaskData_
  185. var maps_z []orm2.ParamsList
  186. pagez := page_z
  187. var offset int
  188. if page <= 1 {
  189. offset = 0
  190. } else {
  191. page -= 1
  192. offset = page * pagez
  193. }
  194. sql_condition := ""
  195. if len(Time_start_) > 1 {
  196. sql_condition += " AND t_time >= '" + Time_start_ + "'"
  197. }
  198. if len(Time_end_) > 1 {
  199. sql_condition += " AND t_time <= '" + Time_end_ + "'"
  200. }
  201. if len(T_id) > 0 {
  202. sql_condition += " AND t_id = '" + T_id + "'"
  203. }
  204. if len(SN) > 0 {
  205. sql_condition += " AND t_sn = '" + SN + "'"
  206. }
  207. if len(sql_condition) > 0 {
  208. sql_condition = " WHERE " + strings.TrimLeft(sql_condition, " AND ")
  209. }
  210. sql := "SELECT COUNT(ID) FROM z_task_data_" + T_task_id + sql_condition
  211. fmt.Println(sql)
  212. _, err := o.Raw(sql).ValuesList(&maps_z)
  213. if err != nil {
  214. return maps, 0
  215. }
  216. if len(maps_z) == 0 {
  217. return maps, 0
  218. }
  219. //fmt.Println("maps_z;",maps_z[0][0])
  220. sql = "SELECT ID,t_sn,t_id,t_t,t_rh,DATE_FORMAT(t_time,'%Y-%m-%d %H:%i:%s') AS t_times,t_time FROM z_task_data_" + T_task_id + sql_condition + " ORDER BY t_time"
  221. if page_z != 9999 {
  222. sql = sql + " LIMIT " + strconv.Itoa(offset) + "," + strconv.Itoa(pagez)
  223. }
  224. fmt.Println(sql)
  225. _, err = o.Raw(sql).QueryRows(&maps)
  226. if err != nil {
  227. logs.Error(lib.FuncName(), err)
  228. }
  229. //value, _ := strconv.ParseFloat(fmt.Sprintf("%.2f", cnt), 64)
  230. key, _ := strconv.Atoi(maps_z[0][0].(string))
  231. return maps, int64(key)
  232. }
  233. func Read_TaskData_ById_List_(T_task_id string, SN string) []TaskData_ {
  234. o := orm.NewOrm()
  235. var maps []TaskData_
  236. sql_condition := ""
  237. sql_condition += " t_sn ='" + SN + "'"
  238. //fmt.Println("maps_z;",maps_z[0][0])
  239. sql := "SELECT ID,t_sn,t_id,t_t,t_rh,DATE_FORMAT(t_time,'%Y-%m-%d %H:%i') AS t_times,t_time FROM z_task_data_" + T_task_id + " WHERE " + sql_condition + " ORDER BY t_time "
  240. fmt.Println(sql)
  241. _, err := o.Raw(sql).QueryRows(&maps)
  242. if err != nil {
  243. logs.Error(lib.FuncName(), err)
  244. return maps
  245. }
  246. return maps
  247. }
  248. func Read_TaskData_ById_ClassList(T_task_id string) []TaskDataClass_ {
  249. o := orm.NewOrm()
  250. var maps []TaskDataClass_
  251. //fmt.Println("maps_z;",maps_z[0][0])
  252. //sql := "SELECT DISTINCT t_sn,t_id FROM z_task_data_" + T_task_id + " ORDER BY t_id "
  253. //sql := "SELECT DISTINCT t_sn FROM z_task_data_" + T_task_id + " ORDER BY t_id "
  254. sql := "SELECT t_sn,t_id FROM z_task_data_" + T_task_id + " GROUP BY t_sn,t_id ORDER BY t_id+0 "
  255. fmt.Println(sql)
  256. _, err := o.Raw(sql).QueryRows(&maps)
  257. if err != nil {
  258. logs.Error(lib.FuncName(), err)
  259. return maps
  260. }
  261. return maps
  262. }
  263. func Read_TaskData_sn(T_task_id string) []TaskDataClass_ {
  264. o := orm.NewOrm()
  265. var maps []TaskDataClass_
  266. sql := "SELECT DISTINCT t_sn FROM z_task_data_" + T_task_id + " ORDER BY t_sn "
  267. fmt.Println(sql)
  268. _, err := o.Raw(sql).QueryRows(&maps)
  269. if err != nil {
  270. logs.Error(lib.FuncName(), err)
  271. return maps
  272. }
  273. return maps
  274. }
  275. func Update_TaskData_ByT_sn(T_task_id, T_sn, T_id string) error {
  276. o := orm.NewOrm()
  277. // 修改id
  278. sql := "UPDATE z_task_data_" + T_task_id + " SET `t_id` = '" + T_id + "' WHERE `t_sn` = '" + T_sn + "'"
  279. logs.Println(sql)
  280. _, err := o.Raw(sql).Exec()
  281. if err != nil {
  282. logs.Error(lib.FuncName(), err)
  283. return err
  284. }
  285. return nil
  286. }
  287. func Delete_TaskData_ByT_sn(T_task_id, T_sn string) error {
  288. o := orm.NewOrm()
  289. // 修改id
  290. sql := "DELETE FROM z_task_data_" + T_task_id + " WHERE `t_sn` = '" + T_sn + "'"
  291. logs.Println(sql)
  292. _, err := o.Raw(sql).Exec()
  293. if err != nil {
  294. logs.Error(lib.FuncName(), err)
  295. return err
  296. }
  297. return nil
  298. }
  299. // 添加
  300. func Add_TaskData(T_task_id string, T_sn string, T_id string, T_t string, T_rh string, T_time string) bool {
  301. o := orm2.NewOrm()
  302. tx, err := o.Begin()
  303. if err != nil {
  304. logs.Error("start the transaction failed")
  305. return false
  306. }
  307. // 开始插入数据
  308. //
  309. //sql := "INSERT INTO z_task_data_" + T_task_id + " (`t_sn`, `t_id`, `t_time`) " +
  310. // "VALUES ('" + T_sn + "', " + T_id + ", '" + T_time + "') " +
  311. // "ON DUPLICATE KEY UPDATE t_t=" + T_t + ", t_rh="+ T_rh +";"
  312. // 去重复数据
  313. sql := "DELETE FROM z_task_data_" + T_task_id + " WHERE " + " t_id = '" + T_id + "' AND " + " t_sn = '" + T_sn + "' " + "AND t_time = '" + T_time + "' "
  314. // 这里有时间优化 用于一次 prepare 多次 exec,以提高批量执行的速度
  315. //fmt.Println(sql)
  316. res, err := tx.Raw(sql).Exec()
  317. if err != nil {
  318. tx.Rollback()
  319. logs.Error(lib.FuncName(), err)
  320. return false
  321. }
  322. res.RowsAffected()
  323. sql = "INSERT INTO z_task_data_" + T_task_id + " (`t_sn`, `t_id`, `t_t`, `t_rh`, `t_time`) " +
  324. "VALUES ('" + T_sn + "', '" + T_id + "', " + T_t + "," + T_rh + ", '" + T_time + "')"
  325. // 更新数据
  326. //sql := "INSERT INTO z_task_data_" + T_task_id + " (`t_sn`, `t_id`, `t_t`, `t_rh`, `t_time`) " +
  327. // "VALUES ('" + T_sn + "', " + T_id + ", " + T_t + "," + T_rh + ", '" + T_time + "')" +
  328. // "on duplicate key update `t_t`=" + T_t + ",`t_rh`=" + T_rh
  329. //fmt.Println(sql)
  330. res, err = tx.Raw(sql).Exec()
  331. if err != nil {
  332. tx.Rollback()
  333. logs.Error(lib.FuncName(), err)
  334. return false
  335. }
  336. tx.Commit()
  337. //fmt.Println("mysql row affected nums: ", num)
  338. return true
  339. }
  340. // 创建唯一索引
  341. func Create_Unique_Index(T_task_id string) error {
  342. // 创建索引
  343. o := orm.NewOrm()
  344. createIndexSql := "ALTER TABLE z_task_data_" + T_task_id + " ADD CONSTRAINT unique_index_t_sn_t_id_t_time UNIQUE (t_sn, t_id, t_time);"
  345. //fmt.Println(sql)
  346. _, err := o.Raw(createIndexSql).Exec()
  347. if err != nil && !strings.Contains(err.Error(), "Duplicate key name 'unique_index_t_sn_t_id_t_time'") {
  348. logs.Error(lib.FuncName(), err)
  349. return err
  350. }
  351. return nil
  352. }
  353. // 删除唯一索引
  354. func Delete_Unique_Index(T_task_id string) error {
  355. // 创建索引
  356. o := orm.NewOrm()
  357. // 删除索引
  358. deleteIndexSql := "ALTER TABLE z_task_data_" + T_task_id + " DROP INDEX unique_index_t_sn_t_id_t_time;"
  359. //fmt.Println(sql)
  360. _, err := o.Raw(deleteIndexSql).Exec()
  361. if err != nil {
  362. logs.Error(lib.FuncName(), err)
  363. return err
  364. }
  365. return nil
  366. }
  367. func Adds_TaskData(T_task_id string, valueStrings []string) error {
  368. o := orm.NewOrm()
  369. // 插入数据
  370. sql := "INSERT INTO z_task_data_" + T_task_id + " (`t_sn`, `t_id`, `t_t`, `t_rh`, `t_time`) VALUES"
  371. sql += strings.Join(valueStrings, ",")
  372. sql += " ON DUPLICATE KEY UPDATE t_t = VALUES(t_t), t_rh = VALUES(t_rh)"
  373. //fmt.Println(sql)
  374. _, err := o.Raw(sql).Exec()
  375. if err != nil {
  376. logs.Error(lib.FuncName(), err)
  377. return err
  378. }
  379. //fmt.Println("mysql row affected nums: ", num)
  380. return nil
  381. }
  382. // 修改
  383. func Up_TaskData(T_task_id string, Id string, T_t string, T_rh string, T_time string) bool {
  384. o := orm.NewOrm()
  385. SET_str := " SET "
  386. if len(T_t) > 0 {
  387. SET_str = SET_str + " t_t = " + T_t + " ,"
  388. }
  389. if len(T_rh) > 0 {
  390. SET_str = SET_str + " t_rh = " + T_rh + " ,"
  391. }
  392. if len(T_time) > 0 {
  393. SET_str = SET_str + " t_time = '" + T_time + "' ,"
  394. }
  395. if len(SET_str) > 7 {
  396. SET_str = SET_str[:(len(SET_str) - 1)]
  397. }
  398. // 开始插入数据 UPDATE `cold_verify`.`Z_TaskData_d8qMyeXLzIxn` SET `t_t` = 20.2 WHERE `ID` = 69
  399. sql := "UPDATE z_task_data_" + T_task_id + " " + SET_str + " WHERE ID = " + Id
  400. // 这里有时间优化 用于一次 prepare 多次 exec,以提高批量执行的速度
  401. //fmt.Println(sql)
  402. res, err := o.Raw(sql).Exec()
  403. if err != nil {
  404. logs.Error(lib.FuncName(), err)
  405. return false
  406. }
  407. res.RowsAffected()
  408. //fmt.Println("mysql row affected nums: ", num)
  409. return true
  410. }
  411. // 添加
  412. func Del_TaskData(T_task_id string, Id string) bool {
  413. o := orm.NewOrm()
  414. // 开始插入数据 UPDATE `cold_verify`.`Z_TaskData_d8qMyeXLzIxn` SET `t_t` = 20.2 WHERE `ID` = 69
  415. sql := "DELETE FROM z_task_data_" + T_task_id + " WHERE ID = " + Id
  416. // 这里有时间优化 用于一次 prepare 多次 exec,以提高批量执行的速度
  417. logs.Println(sql)
  418. res, err := o.Raw(sql).Exec()
  419. if err != nil {
  420. logs.Error(lib.FuncName(), err)
  421. return false
  422. }
  423. res.RowsAffected()
  424. //fmt.Println("mysql row affected nums: ", num)
  425. return true
  426. }
  427. func Read_TaskData_ByT_id(T_task_id string, Id string) (t_sn string, err error) {
  428. o := orm.NewOrm()
  429. // 开始插入数据 UPDATE `cold_verify`.`Z_TaskData_d8qMyeXLzIxn` SET `t_t` = 20.2 WHERE `ID` = 69
  430. sql := "SELECT t_sn FROM z_task_data_" + T_task_id + " WHERE t_id = '" + Id + "'"
  431. // 这里有时间优化 用于一次 prepare 多次 exec,以提高批量执行的速度
  432. logs.Println(sql)
  433. err = o.Raw(sql).QueryRow(&t_sn)
  434. if err != nil {
  435. logs.Error(lib.FuncName(), err)
  436. return t_sn, err
  437. }
  438. //fmt.Println("mysql row affected nums: ", num)
  439. return t_sn, err
  440. }
  441. func Read_TaskData_ByT_sn(T_task_id string, sn string) (t_id string, err error) {
  442. o := orm.NewOrm()
  443. // 开始插入数据 UPDATE `cold_verify`.`Z_TaskData_d8qMyeXLzIxn` SET `t_t` = 20.2 WHERE `ID` = 69
  444. sql := "SELECT t_id FROM z_task_data_" + T_task_id + " WHERE t_sn = '" + sn + "'"
  445. // 这里有时间优化 用于一次 prepare 多次 exec,以提高批量执行的速度
  446. logs.Println(sql)
  447. err = o.Raw(sql).QueryRow(&t_id)
  448. if err != nil {
  449. logs.Error(lib.FuncName(), err)
  450. return t_id, err
  451. }
  452. //fmt.Println("mysql row affected nums: ", num)
  453. return t_id, err
  454. }
  455. // 添加
  456. func Del_TaskData_t_id(T_task_id string, Id string) bool {
  457. o := orm.NewOrm()
  458. // 开始插入数据 UPDATE `cold_verify`.`Z_TaskData_d8qMyeXLzIxn` SET `t_t` = 20.2 WHERE `ID` = 69
  459. sql := "DELETE FROM z_task_data_" + T_task_id + " WHERE t_id = '" + Id + "'"
  460. // 这里有时间优化 用于一次 prepare 多次 exec,以提高批量执行的速度
  461. logs.Println(sql)
  462. res, err := o.Raw(sql).Exec()
  463. if err != nil {
  464. logs.Error(lib.FuncName(), err)
  465. return false
  466. }
  467. res.RowsAffected()
  468. //fmt.Println("mysql row affected nums: ", num)
  469. return true
  470. }
  471. // 检查导出表总数
  472. func Check_TaskData_Num(T_task_id string) bool {
  473. o1 := orm2.NewOrmUsingDB(conf.Local_AliasName)
  474. o2 := orm2.NewOrmUsingDB(conf.Server_AliasName)
  475. var maps_z1, maps_z2 []orm2.ParamsList
  476. sql := "SELECT COUNT(ID) FROM z_task_data_" + T_task_id
  477. fmt.Println(sql)
  478. _, err := o1.Raw(sql).ValuesList(&maps_z1)
  479. if err != nil {
  480. return false
  481. }
  482. _, err = o2.Raw(sql).ValuesList(&maps_z2)
  483. if err != nil {
  484. return false
  485. }
  486. if len(maps_z1) != len(maps_z2) {
  487. return false
  488. }
  489. return true
  490. }
  491. // 检查导出表总数
  492. func Check_TaskData_Num_Verify1(T_task_id string) bool {
  493. o1 := orm2.NewOrmUsingDB(conf.Local_AliasName)
  494. o2 := orm2.NewOrmUsingDB(conf.Verify1_AliasName)
  495. var maps_z1, maps_z2 []orm2.ParamsList
  496. sql := "SELECT COUNT(ID) FROM z_task_data_" + strings.ToLower(T_task_id)
  497. fmt.Println(sql)
  498. _, err := o1.Raw(sql).ValuesList(&maps_z1)
  499. if err != nil {
  500. return false
  501. }
  502. sql2 := "SELECT COUNT(ID) FROM Z_TaskData_" + T_task_id
  503. _, err = o2.Raw(sql2).ValuesList(&maps_z2)
  504. if err != nil {
  505. return false
  506. }
  507. if len(maps_z1) != len(maps_z2) {
  508. return false
  509. }
  510. return true
  511. }
  512. func Read_TaskData_Num(T_task_id string) int {
  513. o1 := orm2.NewOrmUsingDB(conf.Local_AliasName)
  514. var maps_z []orm2.ParamsList
  515. sql := "SELECT COUNT(ID) FROM z_task_data_" + T_task_id
  516. fmt.Println(sql)
  517. _, err := o1.Raw(sql).ValuesList(&maps_z)
  518. if err != nil {
  519. logs.Error(lib.FuncName(), err)
  520. return 0
  521. }
  522. key, _ := strconv.Atoi(maps_z[0][0].(string))
  523. return key
  524. }
  525. func Dump_TaskData(T_task_id, root, password, url_port, database, sql_file string) (string, error) {
  526. // url_port 127.0.0.1:3306
  527. // mysql8.0 以上加 --column-statistics=0
  528. // mysqldump --column-statistics=0 -uroot -proot -h127.0.0.1 -P3306 cold_verify Z_TaskData_ixEfo5zk2Oeb > /Data/Z_TaskData_ixEfo5zk2Oeb.sql
  529. v := Read_Local_Mysql_Version()
  530. host_port := strings.Split(url_port, ":")
  531. table_name := "z_task_data_" + T_task_id
  532. org := "mysqldump "
  533. if v >= 8 {
  534. org += "--column-statistics=0 "
  535. }
  536. //--no-create-info 只导出数据,而不添加 CREATE TABLE 语句。
  537. //--single-transaction 在备份库的时候并不锁定数据表
  538. //--add-locks:在每个表导出之前增加LOCK TABLES并且之后UNLOCK TABLE。(默认为打开状态,使用--skip-add-locks取消选项)
  539. //--compact:压缩模式,产生更少的输出;
  540. org = org + fmt.Sprintf("-u%s -p%s -h%s -P%s --no-create-info --set-gtid-purged=OFF --skip-add-locks --compact --quick %s %s > %s ",
  541. root, password, host_port[0], host_port[1], database, table_name, sql_file)
  542. logs.Println(org)
  543. _, err := lib.Command(org)
  544. if err != nil {
  545. logs.Error(lib.FuncName(), err)
  546. }
  547. return org, err
  548. }
  549. func Dump_TaskData_Verify(T_task_id, root, password, url_port, database, sql_file string) (string, error) {
  550. // url_port 127.0.0.1:3306
  551. // mysql8.0 以上加 --column-statistics=0
  552. // mysqldump --column-statistics=0 -uroot -proot -h127.0.0.1 -P3306 cold_verify Z_TaskData_ixEfo5zk2Oeb > /Data/Z_TaskData_ixEfo5zk2Oeb.sql
  553. v := Read_Local_Mysql_Version()
  554. host_port := strings.Split(url_port, ":")
  555. table_name := "Z_TaskData_" + T_task_id
  556. org := "mysqldump "
  557. if v >= 8 {
  558. org += "--column-statistics=0 "
  559. }
  560. //--no-create-info 只导出数据,而不添加 CREATE TABLE 语句。
  561. //--single-transaction 在备份库的时候并不锁定数据表
  562. //--add-locks:在每个表导出之前增加LOCK TABLES并且之后UNLOCK TABLE。(默认为打开状态,使用--skip-add-locks取消选项)
  563. //--compact:压缩模式,产生更少的输出;
  564. org = org + fmt.Sprintf("-u%s -p%s -h%s -P%s --no-create-info --set-gtid-purged=OFF --skip-add-locks --compact --quick %s %s > %s ",
  565. root, password, host_port[0], host_port[1], database, table_name, sql_file)
  566. logs.Println(org)
  567. _, err := lib.Command(org)
  568. if err != nil {
  569. logs.Error(lib.FuncName(), err)
  570. }
  571. return org, err
  572. }
  573. func Insert_TaskData(root, password, url_port, database, sql_file string) (string, error) {
  574. // url_port 127.0.0.1:3306
  575. // mysql -u root -p root -h127.0.0.1 -P3306 cold_verify_local < /data/Z_TaskData_ixEfo5zk2Oeb.sql
  576. host_port := strings.Split(url_port, ":")
  577. org := fmt.Sprintf("mysql -u%s -p%s -h%s -P%s %s < %s",
  578. root, password, host_port[0], host_port[1], database, sql_file)
  579. logs.Println(org)
  580. _, err := lib.Command(org)
  581. if err != nil {
  582. logs.Error(lib.FuncName(), err)
  583. }
  584. return org, err
  585. }
  586. func Read_Local_Mysql_Version() int {
  587. o := orm2.NewOrmUsingDB(conf.Local_AliasName)
  588. var params []orm2.Params
  589. o.Raw("select version();").Values(&params)
  590. version, _ := strconv.Atoi(params[0]["version()"].(string)[0:1])
  591. return version
  592. }
  593. func Import_TaskData(T_task_id string, offset, pagez int) bool {
  594. var maps []TaskData_
  595. serverOrm := orm2.NewOrmUsingDB(conf.Server_AliasName)
  596. localOrm := orm2.NewOrmUsingDB(conf.Local_AliasName)
  597. tb_name := "z_task_data_" + T_task_id
  598. // 开从先上版查询数据
  599. sql := "select `ID`, `t_sn`, `t_id`, `t_t`, `t_rh`, DATE_FORMAT(t_time,'%Y-%m-%d %H:%i:%s') AS t_times from " + tb_name + " LIMIT " + strconv.Itoa(offset) + "," + strconv.Itoa(pagez)
  600. logs.Println(sql)
  601. _, err := serverOrm.Raw(sql).QueryRows(&maps)
  602. if err != nil {
  603. logs.Error(lib.FuncName(), err)
  604. return false
  605. }
  606. values := []string{}
  607. for _, row := range maps {
  608. // 处理数据
  609. values = append(values, fmt.Sprintf("('%s',%d,%v,%v,'%s')", row.T_sn, row.T_id, row.T_t, row.T_rh, row.T_time))
  610. }
  611. // 向本地版插入数据
  612. sql = "INSERT IGNORE INTO " + tb_name + "( `t_sn`, `t_id`, `t_t`, `t_rh`, `t_time`) VALUES " + strings.Join(values, ",")
  613. //logs.Println(sql)
  614. _, err = localOrm.Raw(sql).Exec()
  615. if err != nil {
  616. logs.Error(lib.FuncName(), err)
  617. return false
  618. }
  619. return true
  620. }
  621. // InsertTaskData 添加设备数据
  622. func InsertTaskData(T_task_id string, d TaskData_) {
  623. o := orm.NewOrm()
  624. var maps TaskData_
  625. sql := fmt.Sprintf("SELECT * FROM z_task_data_%s where t_time='%s' and t_id = '%s' and t_sn = '%s'", T_task_id, d.T_time, d.T_id, d.T_sn)
  626. err := o.Raw(sql).QueryRow(&maps)
  627. if err != nil && err.Error() != orm.ErrNoRows.Error() {
  628. logs.Error(lib.FuncName(), err)
  629. }
  630. // 添加
  631. if maps.ID == 0 {
  632. sqlStatement := fmt.Sprintf("insert into z_task_data_%s(t_sn, t_id, t_t, t_rh, t_time) value('%s','%s',%f,%f,'%s')", T_task_id, d.T_sn, d.T_id, d.T_t, d.T_rh, d.T_time)
  633. _, err = o.Raw(sqlStatement).Exec()
  634. if err != nil {
  635. logs.Error(lib.FuncName(), err)
  636. }
  637. return
  638. }
  639. //已经存在该条目,更新
  640. sqlStatement := fmt.Sprintf("update z_task_data_%s set t_t = %f,t_rh = %f where t_time= '%s' and t_id = '%s' and t_sn = '%s'", T_task_id, d.T_t, d.T_rh, d.T_time, d.T_id, d.T_sn)
  641. _, err = o.Raw(sqlStatement).Exec()
  642. if err != nil {
  643. logs.Error(lib.FuncName(), err)
  644. }
  645. }
  646. // DeleteTaskDataByTimeRange 删除时间范围内的数据
  647. func DeleteTaskDataByTimeRange(T_task_id, sn string, id string, start, end string) {
  648. sqlStatement := fmt.Sprintf("delete from z_task_data_%s where t_time between '%s' and '%s' and t_id = '%s' and t_sn = '%s'", T_task_id, start, end, id, sn)
  649. o := orm.NewOrm()
  650. exec, err := o.Raw(sqlStatement).Exec()
  651. if err != nil {
  652. fmt.Println(err.Error())
  653. }
  654. affected, _ := exec.RowsAffected()
  655. fmt.Printf("从%s~%s时间段删除了%d条数据", start, end, affected)
  656. }
  657. // UpdateTaskDataTemperatureAndHumidityRandom 随机更新
  658. func UpdateTaskDataTemperatureAndHumidityRandom(T_task_id, sn, id, startTime, endTime string, ttMax, ttMin, trhMax, trhMin int) {
  659. //sql语句
  660. sqlStatement := fmt.Sprintf("update z_task_data_%s set t_t = t_t + FLOOR(%d + RAND() * (%d - %d + 1)) /100.0, t_rh = t_rh + FLOOR(%d + RAND() * (%d - %d + 1)) /100.0 where t_time BETWEEN '%s' AND '%s' and t_id = '%s' and t_sn = '%s'", T_task_id, ttMin, ttMax, ttMin, trhMin, trhMax, trhMin, startTime, endTime, id, sn)
  661. o := orm.NewOrm()
  662. exec, err := o.Raw(sqlStatement).Exec()
  663. if err != nil {
  664. fmt.Println("执行错误:", sqlStatement)
  665. }
  666. affected, _ := exec.RowsAffected()
  667. fmt.Println("执行更新行数:", affected)
  668. }
  669. // UpdateTaskDataTemperatureAndHumidity 更新设备探头数据温湿度
  670. func UpdateTaskDataTemperatureAndHumidity(T_task_id, sn, id, startTime, endTime string, temperature, humidity float64) {
  671. sqlStatement := fmt.Sprintf("update z_task_data_%s set t_t = t_t + %f , t_rh = t_rh + %f where t_id = '%s' and t_sn = '%s' and t_time BETWEEN '%s' and '%s'", T_task_id, temperature, humidity, id, sn, startTime, endTime)
  672. o := orm.NewOrm()
  673. exec, err := o.Raw(sqlStatement).Exec()
  674. if err != nil {
  675. fmt.Println(err.Error())
  676. }
  677. affected, err := exec.RowsAffected()
  678. if err != nil {
  679. fmt.Println(err.Error())
  680. }
  681. fmt.Printf("影响了%d行\n", affected)
  682. }
  683. // UpdateTaskData 更新设备探头数据
  684. func UpdateTaskData(T_task_id, sn, id string, old, newO TaskData_) {
  685. sqlStatement := fmt.Sprintf("update z_task_data_%s set t_t = %f , t_rh = %f where t_sn = '%s' and t_id = '%s' and t_t = %f and t_rh = %f and t_time= '%s'", T_task_id, newO.T_t, newO.T_rh, sn, id, old.T_t, old.T_rh, old.T_time)
  686. fmt.Println("执行SQL:", sqlStatement)
  687. o := orm.NewOrm()
  688. exec, err := o.Raw(sqlStatement).Exec()
  689. if err != nil {
  690. fmt.Println("执行错误:", sqlStatement, err.Error())
  691. }
  692. affected, _ := exec.RowsAffected()
  693. fmt.Println("执行更新行数:", affected)
  694. }
  695. // 查询温度最小值 最大值
  696. func Read_TaskData_T_Min_Max(T_task_id string, SN string, T_id string, Time_start_ string, Time_end_ string) (minT, maxT float64) {
  697. o := orm.NewOrm()
  698. sql_condition := ""
  699. if len(Time_start_) > 1 {
  700. sql_condition += " AND t_time >= '" + Time_start_ + "'"
  701. }
  702. if len(Time_end_) > 1 {
  703. sql_condition += " AND t_time <= '" + Time_end_ + "'"
  704. }
  705. if len(T_id) > 0 {
  706. sql_condition += " AND t_id = '" + T_id + "'"
  707. }
  708. if len(SN) > 0 {
  709. sql_condition += " AND t_sn = '" + SN + "'"
  710. }
  711. if len(sql_condition) > 0 {
  712. sql_condition = " WHERE " + strings.TrimLeft(sql_condition, " AND ")
  713. }
  714. //fmt.Println("maps_z;",maps_z[0][0])
  715. sql := "SELECT MIN(t_t) AS min_t, MAX(t_t) AS max_t FROM z_task_data_" + T_task_id + sql_condition
  716. fmt.Println(sql)
  717. err := o.Raw(sql).QueryRow(&minT, &maxT)
  718. if err != nil {
  719. logs.Error(lib.FuncName(), err)
  720. }
  721. return
  722. }
  723. // UpdateTaskDataTemperatureAndHumidityByGeometric 更新设备探头数据温湿度 等比缩放
  724. func UpdateTaskDataTemperatureAndHumidityByGeometric(T_task_id, sn, id, startTime, endTime string, temperature, humidity float64) {
  725. sqlStatement := fmt.Sprintf("update z_task_data_%s set t_t = t_t * %f , t_rh = t_rh * %f where t_id = '%s' and t_sn = '%s' and t_time BETWEEN '%s' and '%s'", T_task_id, temperature, humidity, id, sn, startTime, endTime)
  726. o := orm.NewOrm()
  727. exec, err := o.Raw(sqlStatement).Exec()
  728. if err != nil {
  729. fmt.Println(err.Error())
  730. }
  731. affected, err := exec.RowsAffected()
  732. if err != nil {
  733. fmt.Println(err.Error())
  734. }
  735. fmt.Printf("影响了%d行\n", affected)
  736. }
  737. func Read_TaskData_T_Min_Max_Time_Min_Max(T_task_id string, SN []string, T_id []string, Time_start_ string, Time_end_ string) (minT, maxT float64, minTime, maxTime time.Time) {
  738. o := orm.NewOrm()
  739. sql_condition := ""
  740. if len(Time_start_) > 1 {
  741. sql_condition += " AND t_time >= '" + Time_start_ + "'"
  742. }
  743. if len(Time_end_) > 1 {
  744. sql_condition += " AND t_time <= '" + Time_end_ + "'"
  745. }
  746. if len(T_id) > 0 || len(SN) > 0 {
  747. sql_condition += " AND t_id in (" + strings.Join(T_id, ",") + ") OR t_sn in (" + strings.Join(SN, ",") + ")"
  748. }
  749. if len(sql_condition) > 0 {
  750. sql_condition = " WHERE " + strings.TrimLeft(sql_condition, " AND ")
  751. }
  752. //fmt.Println("maps_z;",maps_z[0][0])
  753. sql := "SELECT MIN(t_t) AS min_t, MAX(t_t) AS max_t,MIN(t_time) AS min_time, MAX(t_time) AS max_time FROM z_task_data_" + T_task_id + sql_condition
  754. fmt.Println(sql)
  755. err := o.Raw(sql).QueryRow(&minT, &maxT, &minTime, &maxTime)
  756. if err != nil {
  757. logs.Error(lib.FuncName(), err)
  758. }
  759. return
  760. }
  761. // 获取线上设备数据
  762. func Read_DeviceData_T_Min_Max_Time_Min_Max(T_sn, T_id string, Time_start_ string, Time_end_ string) (minT, maxT float64, minTime, maxTime time.Time) {
  763. o := orm2.NewOrmUsingDB(conf.Server_AliasName)
  764. sql_condition := ""
  765. if len(Time_start_) > 1 {
  766. sql_condition += " AND t_time >= '" + Time_start_ + "'"
  767. }
  768. if len(Time_end_) > 1 {
  769. sql_condition += " AND t_time <= '" + Time_end_ + "'"
  770. }
  771. sql_condition += " AND t_id = '" + T_id + "'"
  772. if len(sql_condition) > 0 {
  773. sql_condition = " WHERE " + strings.TrimLeft(sql_condition, " AND ")
  774. }
  775. //fmt.Println("maps_z;",maps_z[0][0])
  776. sql := "SELECT MIN(t_t) AS min_t, MAX(t_t) AS max_t,MIN(t_time) AS min_time, MAX(t_time) AS max_time FROM z_devicedata_" + T_sn + sql_condition
  777. fmt.Println(sql)
  778. err := o.Raw(sql).QueryRow(&minT, &maxT, &minTime, &maxTime)
  779. if err != nil {
  780. logs.Error(lib.FuncName(), err)
  781. }
  782. return
  783. }
  784. func Read_TaskData_ByIds_List(T_task_id string, SN []string, T_id []string, Time_start_ string, Time_end_ string) []TaskData {
  785. o := orm.NewOrm()
  786. var maps []TaskData
  787. sql_condition := ""
  788. if len(Time_start_) > 1 {
  789. sql_condition += " AND t_time >= '" + Time_start_ + "'"
  790. }
  791. if len(Time_end_) > 1 {
  792. sql_condition += " AND t_time <= '" + Time_end_ + "'"
  793. }
  794. if len(T_id) > 0 || len(SN) > 0 {
  795. sql_condition += " AND (t_id in (" + strings.Join(T_id, ",") + ") OR t_sn in (" + strings.Join(SN, ",") + "))"
  796. }
  797. if len(sql_condition) > 0 {
  798. sql_condition = " WHERE " + strings.TrimLeft(sql_condition, " AND ")
  799. }
  800. //fmt.Println("maps_z;",maps_z[0][0])
  801. sql := "SELECT ID,t_sn,t_id,t_t,t_rh,t_time FROM z_task_data_" + T_task_id + sql_condition + " ORDER BY t_time"
  802. fmt.Println(sql)
  803. _, err := o.Raw(sql).QueryRows(&maps)
  804. if err != nil {
  805. logs.Error(lib.FuncName(), err)
  806. }
  807. return maps
  808. }
  809. // 按照 t_id 分组数据
  810. func GroupDataByTID(data []TaskData) map[string][]TaskData {
  811. groupedData := make(map[string][]TaskData)
  812. for _, d := range data {
  813. groupedData[d.T_id] = append(groupedData[d.T_id], d)
  814. }
  815. return groupedData
  816. }
  817. // 删除的重复数据
  818. func DeleteDeduplicate(T_task_id string) error {
  819. localOrm := orm2.NewOrmUsingDB(conf.Local_AliasName)
  820. tb_name := "z_task_data_" + T_task_id
  821. // 创建临时表
  822. sqlCreate := "CREATE TABLE `tmp_table` AS (SELECT MAX(`ID`) AS `max_id` FROM " + tb_name + " GROUP BY `t_sn`,`t_id`,`t_time`);"
  823. _, err := localOrm.Raw(sqlCreate).Exec()
  824. if err != nil {
  825. logs.Error(lib.FuncName(), err)
  826. return err
  827. }
  828. sqlDelete := "DELETE FROM " + tb_name + " WHERE `ID` NOT IN (SELECT `max_id` FROM `tmp_table`);"
  829. _, err = localOrm.Raw(sqlDelete).Exec()
  830. if err != nil {
  831. logs.Error(lib.FuncName(), err)
  832. return err
  833. }
  834. sqlDrop := "DROP TABLE `tmp_table`;"
  835. _, err = localOrm.Raw(sqlDrop).Exec()
  836. if err != nil {
  837. logs.Error(lib.FuncName(), err)
  838. return err
  839. }
  840. return nil
  841. }