edb_data_ths.go 4.3 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162
  1. package data_manage
  2. import (
  3. "encoding/json"
  4. "errors"
  5. "fmt"
  6. "github.com/beego/beego/v2/client/orm"
  7. "github.com/rdlucklib/rdluck_tools/http"
  8. "hongze/hz_crm_api/utils"
  9. "strconv"
  10. "strings"
  11. "time"
  12. )
  13. type EdbDataThs struct {
  14. EdbDataId int `orm:"column(edb_data_id);pk"`
  15. EdbInfoId int
  16. EdbCode string
  17. DataTime string
  18. Value float64
  19. Status int
  20. CreateTime time.Time
  21. ModifyTime time.Time
  22. DataTimestamp int64
  23. }
  24. func AddEdbDataThsBySql(sqlStr string) (err error) {
  25. o := orm.NewOrmUsingDB("data")
  26. _, err = o.Raw(sqlStr).Exec()
  27. return
  28. }
  29. func ModifyEdbDataThs(edbInfoId int64, dataTime string, value float64) (err error) {
  30. o := orm.NewOrmUsingDB("data")
  31. sql := ` UPDATE edb_data_ths SET value=?,modify_time=NOW() WHERE edb_info_id=? AND data_time=? `
  32. _, err = o.Raw(sql, value, edbInfoId, dataTime).Exec()
  33. return
  34. }
  35. func GetEdbDataThsMaxOrMinDate(edbCode string) (min_date, max_date string, err error) {
  36. o := orm.NewOrmUsingDB("data")
  37. sql := ` SELECT MIN(data_time) AS min_date,MAX(data_time) AS max_date FROM edb_data_ths WHERE edb_code=? `
  38. err = o.Raw(sql, edbCode).QueryRow(&min_date, &max_date)
  39. return
  40. }
  41. func GetEdbDataThsByCodeAndDate(edbCode string, startDate string) (count int, err error) {
  42. o := orm.NewOrmUsingDB("data")
  43. sql := ` SELECT COUNT(1) AS count FROM edb_data_ths WHERE edb_code=? AND data_time=? `
  44. err = o.Raw(sql, edbCode, startDate).QueryRow(&count)
  45. return
  46. }
  47. func GetEdbDataThsByCode(edbCode string, size int) (items []*EdbInfoSearchData, err error) {
  48. o := orm.NewOrmUsingDB("data")
  49. sql := ` SELECT * FROM edb_data_ths WHERE edb_code=? ORDER BY data_time DESC LIMIT ? `
  50. _, err = o.Raw(sql, edbCode, size).QueryRows(&items)
  51. return
  52. }
  53. type EdbDataFromThs struct {
  54. DataVol int64 `json:"dataVol"`
  55. Errmsg string `json:"errmsg"`
  56. Errorcode int64 `json:"errorcode"`
  57. Perf interface{} `json:"perf"`
  58. Tables []struct {
  59. ID []string `json:"id"`
  60. Time []string `json:"time"`
  61. Value []float64 `json:"value"`
  62. } `json:"tables"`
  63. }
  64. // 刷新所有数据
  65. func RefreshAllEdbDataByThs(edbInfoId, source int, edbCode, startDate, endDate string) (err error) {
  66. o := orm.NewOrmUsingDB("data")
  67. to, err := o.Begin()
  68. if err != nil {
  69. return
  70. }
  71. defer func() {
  72. if err != nil {
  73. _ = to.Rollback()
  74. } else {
  75. _ = to.Commit()
  76. }
  77. }()
  78. thsUrl := utils.Hz_Data_Url + `edbInfo/ths?EdbCode=%s&StartDate=%s&EndDate=%s`
  79. thsUrl = fmt.Sprintf(thsUrl, edbCode, startDate, endDate)
  80. utils.FileLog.Info("thsUrl:%s", thsUrl)
  81. body, err := http.Get(thsUrl)
  82. fmt.Println("GetEdbDataByThs body:")
  83. fmt.Println(string(body))
  84. if err != nil {
  85. return
  86. }
  87. item := new(EdbDataFromThs)
  88. err = json.Unmarshal(body, &item)
  89. if err != nil {
  90. return
  91. }
  92. if item.Errorcode != 0 {
  93. err = errors.New(string(body))
  94. return
  95. }
  96. //获取指标所有数据
  97. dataList := make([]*EdbDataBase, 0)
  98. dataTableName := GetEdbDataTableName(source)
  99. sql := `SELECT * FROM %s WHERE edb_info_id=? `
  100. sql = fmt.Sprintf(sql, dataTableName)
  101. _, err = to.Raw(sql, edbInfoId).QueryRows(&dataList)
  102. if err != nil {
  103. return err
  104. }
  105. dataMap := make(map[string]string)
  106. for _, v := range dataList {
  107. dataMap[v.DataTime] = v.Value
  108. }
  109. edbInfoIdStr := strconv.Itoa(edbInfoId)
  110. if len(item.Tables) > 0 {
  111. table := item.Tables[0]
  112. dataLen := len(table.Time)
  113. addSql := ` INSERT INTO edb_data_ths(edb_info_id,edb_code,data_time,value,create_time,modify_time,status,data_timestamp) values `
  114. var isAdd bool
  115. for i := 0; i < dataLen; i++ {
  116. eDate := table.Time[i]
  117. sValue := table.Value[i]
  118. saveValue := utils.SubFloatToString(sValue, 30)
  119. if existVal, ok := dataMap[eDate]; !ok {
  120. dataTime, err := time.Parse(utils.FormatDate, eDate)
  121. if err != nil {
  122. }
  123. timestamp := dataTime.UnixNano() / 1e6
  124. timeStr := fmt.Sprintf("%d", timestamp)
  125. addSql += GetAddSql(edbInfoIdStr, edbCode, eDate, timeStr, saveValue)
  126. isAdd = true
  127. } else {
  128. if existVal != saveValue {
  129. sql := ` UPDATE %s SET value=?,modify_time=NOW() WHERE edb_info_id=? AND data_time=? `
  130. sql = fmt.Sprintf(sql, dataTableName)
  131. _, err = to.Raw(sql, sValue, edbInfoId, eDate).Exec()
  132. if err != nil {
  133. return err
  134. }
  135. }
  136. }
  137. }
  138. if isAdd {
  139. addSql = strings.TrimRight(addSql, ",")
  140. _, err = to.Raw(addSql).Exec()
  141. if err != nil {
  142. fmt.Println("RefreshAllEdbDataByThs add Err", err.Error())
  143. return
  144. }
  145. }
  146. }
  147. return
  148. }