edb_data_ths.go 4.3 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159
  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/hongze_chart_lib/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. o.Begin()
  68. defer func() {
  69. if err != nil {
  70. o.Rollback()
  71. } else {
  72. o.Commit()
  73. }
  74. }()
  75. thsUrl := utils.Hz_Data_Url + `edbInfo/ths?EdbCode=%s&StartDate=%s&EndDate=%s`
  76. thsUrl = fmt.Sprintf(thsUrl, edbCode, startDate, endDate)
  77. utils.FileLog.Info("thsUrl:%s", thsUrl)
  78. body, err := http.Get(thsUrl)
  79. fmt.Println("GetEdbDataByThs body:")
  80. fmt.Println(string(body))
  81. if err != nil {
  82. return
  83. }
  84. item := new(EdbDataFromThs)
  85. err = json.Unmarshal(body, &item)
  86. if err != nil {
  87. return
  88. }
  89. if item.Errorcode != 0 {
  90. err = errors.New(string(body))
  91. return
  92. }
  93. //获取指标所有数据
  94. dataList := make([]*EdbDataBase, 0)
  95. dataTableName := GetEdbDataTableName(source)
  96. sql := `SELECT * FROM %s WHERE edb_info_id=? `
  97. sql = fmt.Sprintf(sql, dataTableName)
  98. _, err = o.Raw(sql, edbInfoId).QueryRows(&dataList)
  99. if err != nil {
  100. return err
  101. }
  102. dataMap := make(map[string]string)
  103. for _, v := range dataList {
  104. dataMap[v.DataTime] = v.Value
  105. }
  106. edbInfoIdStr := strconv.Itoa(edbInfoId)
  107. if len(item.Tables) > 0 {
  108. table := item.Tables[0]
  109. dataLen := len(table.Time)
  110. addSql := ` INSERT INTO edb_data_ths(edb_info_id,edb_code,data_time,value,create_time,modify_time,status,data_timestamp) values `
  111. var isAdd bool
  112. for i := 0; i < dataLen; i++ {
  113. eDate := table.Time[i]
  114. sValue := table.Value[i]
  115. saveValue := utils.SubFloatToString(sValue, 30)
  116. if existVal, ok := dataMap[eDate]; !ok {
  117. dataTime, err := time.Parse(utils.FormatDate, eDate)
  118. if err != nil {
  119. }
  120. timestamp := dataTime.UnixNano() / 1e6
  121. timeStr := fmt.Sprintf("%d", timestamp)
  122. addSql += GetAddSql(edbInfoIdStr, edbCode, eDate, timeStr, saveValue)
  123. isAdd = true
  124. } else {
  125. if existVal != saveValue {
  126. sql := ` UPDATE %s SET value=?,modify_time=NOW() WHERE edb_info_id=? AND data_time=? `
  127. sql = fmt.Sprintf(sql, dataTableName)
  128. _, err = o.Raw(sql, sValue, edbInfoId, eDate).Exec()
  129. if err != nil {
  130. return err
  131. }
  132. }
  133. }
  134. }
  135. if isAdd {
  136. addSql = strings.TrimRight(addSql, ",")
  137. _, err = o.Raw(addSql).Exec()
  138. if err != nil {
  139. fmt.Println("RefreshAllEdbDataByThs add Err", err.Error())
  140. return
  141. }
  142. }
  143. }
  144. return
  145. }