TaskData.go 30 KB

1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283848586878889909192939495969798991001011021031041051061071081091101111121131141151161171181191201211221231241251261271281291301311321331341351361371381391401411421431441451461471481491501511521531541551561571581591601611621631641651661671681691701711721731741751761771781791801811821831841851861871881891901911921931941951961971981992002012022032042052062072082092102112122132142152162172182192202212222232242252262272282292302312322332342352362372382392402412422432442452462472482492502512522532542552562572582592602612622632642652662672682692702712722732742752762772782792802812822832842852862872882892902912922932942952962972982993003013023033043053063073083093103113123133143153163173183193203213223233243253263273283293303313323333343353363373383393403413423433443453463473483493503513523533543553563573583593603613623633643653663673683693703713723733743753763773783793803813823833843853863873883893903913923933943953963973983994004014024034044054064074084094104114124134144154164174184194204214224234244254264274284294304314324334344354364374384394404414424434444454464474484494504514524534544554564574584594604614624634644654664674684694704714724734744754764774784794804814824834844854864874884894904914924934944954964974984995005015025035045055065075085095105115125135145155165175185195205215225235245255265275285295305315325335345355365375385395405415425435445455465475485495505515525535545555565575585595605615625635645655665675685695705715725735745755765775785795805815825835845855865875885895905915925935945955965975985996006016026036046056066076086096106116126136146156166176186196206216226236246256266276286296306316326336346356366376386396406416426436446456466476486496506516526536546556566576586596606616626636646656666676686696706716726736746756766776786796806816826836846856866876886896906916926936946956966976986997007017027037047057067077087097107117127137147157167177187197207217227237247257267277287297307317327337347357367377387397407417427437447457467477487497507517527537547557567577587597607617627637647657667677687697707717727737747757767777787797807817827837847857867877887897907917927937947957967977987998008018028038048058068078088098108118128138148158168178188198208218228238248258268278288298308318328338348358368378388398408418428438448458468478488498508518528538548558568578588598608618628638648658668678688698708718728738748758768778788798808818828838848858868878888898908918928938948958968978988999009019029039049059069079089099109119129139149159169179189199209219229239249259269279289299309319329339349359369379389399409419429439449459469479489499509519529539549559569579589599609619629639649659669679689699709719729739749759769779789799809819829839849859869879889899909919929939949959969979989991000
  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) (int64, error) {
  819. localOrm := orm2.NewOrmUsingDB(conf.Local_AliasName)
  820. tb_name := "z_task_data_" + T_task_id
  821. var cnt int64
  822. // 创建临时表
  823. sqlCreate := "CREATE TABLE `tmp_table` AS (SELECT MIN(`ID`) AS `min_id` FROM " + tb_name + " GROUP BY `t_sn`,`t_id`,`t_time`);"
  824. _, err := localOrm.Raw(sqlCreate).Exec()
  825. if err != nil {
  826. logs.Error(lib.FuncName(), err)
  827. return cnt, err
  828. }
  829. sqlDelete := "DELETE FROM " + tb_name + " WHERE `ID` NOT IN (SELECT `min_id` FROM `tmp_table`);"
  830. res, err := localOrm.Raw(sqlDelete).Exec()
  831. if err != nil {
  832. logs.Error(lib.FuncName(), err)
  833. return cnt, err
  834. }
  835. cnt, _ = res.RowsAffected()
  836. sqlDrop := "DROP TABLE `tmp_table`;"
  837. _, err = localOrm.Raw(sqlDrop).Exec()
  838. if err != nil {
  839. logs.Error(lib.FuncName(), err)
  840. return cnt, err
  841. }
  842. return cnt, nil
  843. }