base_from_pb.go 4.7 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156
  1. package models
  2. import (
  3. "fmt"
  4. "github.com/beego/beego/v2/client/orm"
  5. "hongze/hongze_edb_lib/services"
  6. "hongze/hongze_edb_lib/services/alarm_msg"
  7. "hongze/hongze_edb_lib/utils"
  8. "strconv"
  9. "strings"
  10. "time"
  11. )
  12. func AddEdbDataFromPb(edbCode string, item *services.EdbDataFromPb) (err error) {
  13. var errMsg string
  14. o := orm.NewOrm()
  15. defer func() {
  16. if err != nil {
  17. //go utils.SendEmail(utils.APP_NAME_CN+"【"+utils.RunMode+"】"+"失败提醒", " 彭博数据获取失败:err:"+errMsg, utils.EmailSendToUsers)
  18. go alarm_msg.SendAlarmMsg("彭博数据获取失败:err:"+errMsg, 3)
  19. }
  20. }()
  21. if len(item.Date) > 0 {
  22. var isAdd bool
  23. dateMap := item.Date
  24. addSql := ` INSERT INTO edb_data_pb(edb_info_id,edb_code,data_time,value,create_time,modify_time,ticker,field,data_timestamp) values `
  25. nowStr := time.Now().Format(utils.FormatDateTime)
  26. for k, v := range dateMap {
  27. timeStr := fmt.Sprintf("%d", v)
  28. v = v / 1000
  29. t := time.Unix(v, 0)
  30. dateTime := t.Format(utils.FormatDate)
  31. val := item.Value[k]
  32. field := item.Field[k]
  33. ticker := item.Ticker[k]
  34. if field == "PX_LAST" {
  35. addSql += "("
  36. addSql += "0," + "'" + edbCode + "'" + "," + "'" + dateTime + "'" + "," + utils.SubFloatToString(val, 20) + "," + "'" + nowStr + "'" +
  37. "," + "'" + nowStr + "'" + "," + "'" + ticker + "'" + "," + "'" + field + "'" + "," + "'" + timeStr + "'"
  38. addSql += "),"
  39. isAdd = true
  40. }
  41. }
  42. if isAdd {
  43. addSql = strings.TrimRight(addSql, ",")
  44. _, err = o.Raw(addSql).Exec()
  45. if err != nil {
  46. errMsg = " tx.Exec Err :" + err.Error()
  47. return
  48. }
  49. }
  50. }
  51. return
  52. }
  53. // 刷新彭博指标数据
  54. func RefreshEdbDataFromPb(edbInfoId int, edbCode, startDate string, item *services.EdbDataFromPb) (err error) {
  55. o := orm.NewOrm()
  56. source := utils.DATA_SOURCE_PB
  57. // 真实数据的最大日期 , 插入规则配置的日期
  58. var realDataMaxDate, edbDataInsertConfigDate time.Time
  59. var edbDataInsertConfig *EdbDataInsertConfig
  60. var isFindConfigDateRealData bool //是否找到配置日期的实际数据的值
  61. {
  62. edbDataInsertConfig, err = GetEdbDataInsertConfigByEdbId(edbInfoId)
  63. if err != nil && err.Error() != utils.ErrNoRow() {
  64. return
  65. }
  66. if edbDataInsertConfig != nil {
  67. edbDataInsertConfigDate = edbDataInsertConfig.Date
  68. }
  69. }
  70. var condition string
  71. var pars []interface{}
  72. condition += " AND edb_info_id=? "
  73. pars = append(pars, edbInfoId)
  74. condition += " AND data_time>=? "
  75. pars = append(pars, startDate)
  76. existList, err := GetEdbDataByCondition(source, condition, pars)
  77. existMap := make(map[string]*EdbInfoSearchData)
  78. for _, v := range existList {
  79. existMap[v.DataTime] = v
  80. }
  81. edbInfoIdStr := strconv.Itoa(edbInfoId)
  82. addMap := make(map[string]string)
  83. if len(item.Date) > 0 {
  84. dateMap := item.Date
  85. var isAdd bool
  86. addSql := ` INSERT INTO edb_data_pb(edb_info_id,edb_code,data_time,value,create_time,modify_time,ticker,field,data_timestamp) values `
  87. nowStr := time.Now().Format(utils.FormatDateTime)
  88. for k, v := range dateMap {
  89. timeStr := fmt.Sprintf("%d", v)
  90. v = v / 1000
  91. t := time.Unix(v, 0)
  92. dateTime := t.Format(utils.FormatDate)
  93. t, _ = time.ParseInLocation(utils.FormatDate, dateTime, time.Local) //这里的目的是为了北京时间,以防万一,还是重新转一下吧
  94. val := item.Value[k]
  95. field := item.Field[k]
  96. ticker := item.Ticker[k]
  97. saveValue := utils.SubFloatToString(val, 30)
  98. if field == "PX_LAST" {
  99. if findItem, ok := existMap[dateTime]; !ok {
  100. if _, addOk := addMap[dateTime]; !addOk {
  101. addSql += "("
  102. addSql += edbInfoIdStr + "," + "'" + edbCode + "'" + "," + "'" + dateTime + "'" + "," + saveValue + "," + "'" + nowStr + "'" +
  103. "," + "'" + nowStr + "'" + "," + "'" + ticker + "'" + "," + "'" + field + "'" + "," + "'" + timeStr + "'"
  104. addSql += "),"
  105. isAdd = true
  106. addMap[dateTime] = saveValue
  107. }
  108. } else {
  109. if findItem != nil && utils.SubFloatToString(findItem.Value, 30) != saveValue {
  110. err = ModifyEdbDataById(source, findItem.EdbDataId, saveValue)
  111. if err != nil {
  112. return err
  113. }
  114. }
  115. }
  116. // 下面代码主要目的是处理掉手动插入的数据判断
  117. {
  118. if realDataMaxDate.IsZero() || t.After(realDataMaxDate) {
  119. realDataMaxDate = t
  120. }
  121. if edbDataInsertConfigDate.IsZero() || t.Equal(edbDataInsertConfigDate) {
  122. isFindConfigDateRealData = true
  123. }
  124. }
  125. }
  126. }
  127. // 处理手工数据补充的配置
  128. HandleConfigInsertEdbData(realDataMaxDate, edbDataInsertConfig, edbInfoId, source, existMap, isFindConfigDateRealData)
  129. if isAdd {
  130. addSql = strings.TrimRight(addSql, ",")
  131. _, err = o.Raw(addSql).Exec()
  132. if err != nil {
  133. fmt.Println("RefreshAllEdbDataByPb add Err", err.Error())
  134. return
  135. }
  136. }
  137. }
  138. return
  139. }