edb_data.go 6.7 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213
  1. package controllers
  2. import (
  3. "encoding/json"
  4. "eta/eta_forum_hub/models"
  5. "eta/eta_forum_hub/services"
  6. "eta/eta_forum_hub/utils"
  7. "fmt"
  8. "strconv"
  9. "time"
  10. )
  11. type EdbDataController struct {
  12. BaseAuthController
  13. }
  14. // 根据binlog监听记录,更新指标数据
  15. // SaveByBinlog 根据binlog监听记录,更新指标数据
  16. // @Title 根据binlog监听记录,更新指标数据
  17. // @Description 根据binlog监听记录,更新指标数据
  18. // @Param request body models.EdbDataBinlogReq true "type json string"
  19. // @Success 200 {object} models.BaseResponse
  20. // @router /save_by_binlog [post]
  21. func (this *EdbDataController) SaveByBinlog() {
  22. br := new(models.BaseResponse).Init()
  23. defer func() {
  24. this.Data["json"] = br
  25. this.ServeJSON()
  26. }()
  27. var req models.EdbDataBinlogReq
  28. err := json.Unmarshal(this.Ctx.Input.RequestBody, &req)
  29. if err != nil {
  30. br.Msg = "参数解析异常!"
  31. br.ErrMsg = "参数解析失败,Err:" + err.Error()
  32. return
  33. }
  34. if len(req.List) == 0 {
  35. br.Msg = "参数异常!"
  36. br.ErrMsg = "参数异常,请检查!"
  37. return
  38. }
  39. edbInfoIdMap := make(map[int]*models.AddEdbDataBinlogReq)
  40. reqList := make([]*models.EdbDataBinlogItem, 0)
  41. // 判断指标库里是否存在这些指标,如果不存在,则丢弃,无需处理
  42. edbInfoIds := make([]int, 0)
  43. idMap := make(map[int]bool)
  44. for _, item := range req.List {
  45. // 解析json
  46. var dataItem models.EdbDataBinlogItem
  47. err := json.Unmarshal([]byte(item.Item), &dataItem)
  48. if err != nil {
  49. utils.FileLog.Info("解析binlog数据失败,Err:" + err.Error(), "item", item)
  50. continue
  51. }
  52. dataItem.OpType = item.OpType
  53. reqList = append(reqList, &dataItem)
  54. if _, ok := idMap[dataItem.EdbInfoId]; !ok {
  55. idMap[dataItem.EdbInfoId] = true
  56. edbInfoIds = append(edbInfoIds, dataItem.EdbInfoId)
  57. }
  58. }
  59. // 查询指标信息
  60. edbInfoList, err := models.GetEdbInfoByIdList(edbInfoIds)
  61. if err != nil {
  62. br.Msg = "指标信息查询异常!"
  63. br.ErrMsg = "指标信息查询失败,Err:" + err.Error()
  64. return
  65. }
  66. existEdbInfoMap := make(map[int]*models.EdbInfo)
  67. for _, item := range edbInfoList {
  68. existEdbInfoMap[item.EdbInfoId] = item
  69. }
  70. // edbInfoList := []*models.EdbInfo{
  71. // {
  72. // EdbInfoId: 1,
  73. // EdbCode: "test_code_1",
  74. // EdbType: 1,
  75. // },
  76. // {
  77. // EdbInfoId: 2,
  78. // EdbCode: "test_code_2",
  79. // EdbType: 1,
  80. // },
  81. // }
  82. // existEdbInfoMap := make(map[int]*models.EdbInfo)
  83. // for _, item := range edbInfoList {
  84. // existEdbInfoMap[item.EdbInfoId] = item
  85. // }
  86. edbDataIdMap := make(map[string]*models.EdbDataBaseWithOpType)
  87. for _, item := range reqList {
  88. edbInfo, ok := existEdbInfoMap[item.EdbInfoId]
  89. if !ok {
  90. continue
  91. }
  92. if _, ok := edbInfoIdMap[item.EdbInfoId]; !ok {
  93. tmp := &models.AddEdbDataBinlogReq{
  94. EdbInfoId: item.EdbInfoId,
  95. EdbCode: item.EdbCode,
  96. EdbType: edbInfo.EdbType,
  97. }
  98. edbInfoIdMap[item.EdbInfoId] = tmp
  99. }
  100. value, ok := item.Value.(float64)
  101. if !ok {
  102. valueStr, ok:= item.Value.(string)
  103. if !ok {
  104. utils.FileLog.Info("binlog数据类型异常,Err:" + err.Error(), "item", item)
  105. continue
  106. }
  107. value, err = strconv.ParseFloat(valueStr, 64)
  108. if err != nil {
  109. utils.FileLog.Info("binlog数据类型转换异常,Err:" + err.Error(), "item", item)
  110. continue
  111. }
  112. }
  113. createTime, _ := time.ParseInLocation("2006-01-02 15:04:05", item.CreateTime, time.Local)
  114. modifyTime, _ := time.ParseInLocation("2006-01-02 15:04:05", item.ModifyTime, time.Local)
  115. dataItem := &models.EdbDataBaseWithOpType{
  116. EdbDataBase: &models.EdbDataBase{
  117. EdbInfoId: item.EdbInfoId,
  118. EdbCode: item.EdbCode,
  119. Value: strconv.FormatFloat(value, 'f', -1, 64),
  120. DataTimestamp: item.DataTimestamp,
  121. DataTime: item.DataTime,
  122. CreateTime: createTime,
  123. ModifyTime: modifyTime,
  124. },
  125. OpType: item.OpType,
  126. }
  127. if _, ok := edbDataIdMap[fmt.Sprintf("%d_%s", item.EdbInfoId, item.DataTime)]; ok {
  128. edbDataIdMap[fmt.Sprintf("%d_%s", item.EdbInfoId, item.DataTime)] = dataItem
  129. } else {
  130. edbDataIdMap[fmt.Sprintf("%d_%s", item.EdbInfoId, item.DataTime)] = dataItem
  131. }
  132. //edbInfoIdMap[item.EdbInfoId].DataList = append(edbInfoIdMap[item.EdbInfoId].DataList, dataItem)
  133. }
  134. edbDataMinDataTimeMap := make(map[int]string)
  135. edbDataMaxDataTimeMap := make(map[int]string)
  136. for _, item := range edbDataIdMap {
  137. edbInfoIdMap[item.EdbInfoId].DataList = append(edbInfoIdMap[item.EdbInfoId].DataList, item)
  138. if edbDataMinDataTimeMap[item.EdbInfoId] == "" || edbDataMinDataTimeMap[item.EdbInfoId] > item.DataTime {
  139. edbDataMinDataTimeMap[item.EdbInfoId] = item.DataTime
  140. }
  141. if edbDataMaxDataTimeMap[item.EdbInfoId] == "" || edbDataMaxDataTimeMap[item.EdbInfoId] < item.DataTime {
  142. edbDataMaxDataTimeMap[item.EdbInfoId] = item.DataTime
  143. }
  144. }
  145. for _, item := range edbInfoIdMap {
  146. minDataTime := edbDataMinDataTimeMap[item.EdbInfoId]
  147. maxDataTime := edbDataMaxDataTimeMap[item.EdbInfoId]
  148. if item.EdbType == 1 {
  149. err := services.AddOrUpdateEdbDataWithOpType(item.EdbInfoId, item.DataList, minDataTime, maxDataTime)
  150. if err != nil {
  151. br.Msg = "更新指标数据失败!"
  152. br.ErrMsg = "更新指标数据失败,Err:" + err.Error()
  153. return
  154. }
  155. } else {
  156. err := services.AddOrUpdateEdbDataCalculateWithOpType(item.EdbInfoId, item.DataList, minDataTime, maxDataTime)
  157. if err != nil {
  158. br.Msg = "更新指标数据失败!"
  159. br.ErrMsg = "更新指标数据失败,Err:" + err.Error()
  160. return
  161. }
  162. }
  163. }
  164. br.Ret = 200
  165. br.Success = true
  166. return
  167. }
  168. // TestSaveByBinlog 测试SaveByBinlog接口的测试数据
  169. // @Title 测试SaveByBinlog接口的测试数据
  170. // @Description 测试SaveByBinlog接口的测试数据
  171. // @Success 200 {object} models.BaseResponse
  172. // @router /test_save_by_binlog [get]
  173. func (this *EdbDataController) TestSaveByBinlog() {
  174. br := new(models.BaseResponse).Init()
  175. defer func() {
  176. this.Data["json"] = br
  177. this.ServeJSON()
  178. }()
  179. // 构造测试数据
  180. testData := models.EdbDataBinlogReq{
  181. List: []*models.EdbDataBinlogDataReq{
  182. {
  183. OpType: "INSERT",
  184. Item: `{"create_time":"2024-10-12 17:03:04","data_time":"2024-09-08","data_timestamp":1725724800000,"edb_code":"test_code_1","edb_data_id":1001,"edb_info_id":1,"modify_time":"2024-10-14 13:37:30","value":71988}`,
  185. },
  186. {
  187. OpType: "UPDATE",
  188. Item: `{"create_time":"2024-10-12 17:03:04","data_time":"2024-09-09","data_timestamp":1725811200000,"edb_code":"test_code_1","edb_data_id":1002,"edb_info_id":1,"modify_time":"2024-10-14 13:37:30","value":72500}`,
  189. },
  190. {
  191. OpType: "DELETE",
  192. Item: `{"create_time":"2024-10-12 17:03:04","data_time":"2024-09-10","data_timestamp":1725897600000,"edb_code":"test_code_2","edb_data_id":2001,"edb_info_id":2,"modify_time":"2024-10-14 13:37:30","value":85600}`,
  193. },
  194. },
  195. }
  196. br.Ret = 200
  197. br.Success = true
  198. br.Msg = "获取测试数据成功"
  199. br.Data = testData
  200. return
  201. }