base_from_wind.go 4.7 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160
  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 AddEdbDataFromWind(edbCode string, item *services.EdbDataFromWind) (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("wind数据获取失败:err:"+errMsg, 3)
  19. }
  20. }()
  21. var isAdd bool
  22. addSql := ` INSERT INTO edb_data_wind(edb_info_id,edb_code,data_time,value,create_time,modify_time,data_timestamp) values `
  23. for k, v := range item.Dt {
  24. timeStr := fmt.Sprintf("%d", v)
  25. v = v / 1000
  26. t := time.Unix(v, 0)
  27. dateTime := t.Format(utils.FormatDate)
  28. val := item.Close[k]
  29. saveVal := utils.SubFloatToString(val, 20)
  30. addSql += GetAddSql("0", edbCode, dateTime, timeStr, saveVal)
  31. isAdd = true
  32. }
  33. if isAdd {
  34. addSql = strings.TrimRight(addSql, ",")
  35. _, err = o.Raw(addSql).Exec()
  36. if err != nil {
  37. errMsg = " tx.Exec Err :" + err.Error()
  38. return
  39. }
  40. }
  41. return
  42. }
  43. // RefreshEdbDataFromWind 刷新wind指标数据
  44. func RefreshEdbDataFromWind(edbInfoId int, edbCode, startDate string, item *services.EdbDataFromWind) (err error) {
  45. o := orm.NewOrm()
  46. source := utils.DATA_SOURCE_WIND
  47. // 真实数据的最大日期 , 插入规则配置的日期
  48. var realDataMaxDate, edbDataInsertConfigDate time.Time
  49. var edbDataInsertConfig *EdbDataInsertConfig
  50. var isFindConfigDateRealData bool //是否找到配置日期的实际数据的值
  51. {
  52. edbDataInsertConfig, err = GetEdbDataInsertConfigByEdbId(edbInfoId)
  53. if err != nil && err.Error() != utils.ErrNoRow() {
  54. return
  55. }
  56. if edbDataInsertConfig != nil {
  57. edbDataInsertConfigDate = edbDataInsertConfig.Date
  58. }
  59. }
  60. var condition string
  61. var pars []interface{}
  62. condition += " AND edb_info_id=? "
  63. pars = append(pars, edbInfoId)
  64. var startDateTime time.Time
  65. if startDate != "" {
  66. condition += " AND data_time>=? "
  67. pars = append(pars, startDate)
  68. startDateTime, _ = time.ParseInLocation(utils.FormatDate, startDate, time.Local)
  69. }
  70. existList, err := GetEdbDataByCondition(source, condition, pars)
  71. existMap := make(map[string]*EdbInfoSearchData)
  72. for _, v := range existList {
  73. existMap[v.DataTime] = v
  74. }
  75. addSql := ` INSERT INTO edb_data_wind(edb_info_id,edb_code,data_time,value,create_time,modify_time,data_timestamp) values `
  76. var isAdd bool
  77. addMap := make(map[string]string)
  78. edbInfoIdStr := strconv.Itoa(edbInfoId)
  79. for k, v := range item.Dt {
  80. timeStr := fmt.Sprintf("%d", v)
  81. v = v / 1000
  82. t := time.Unix(v, 0)
  83. dateTime := t.Format(utils.FormatDate)
  84. t, _ = time.ParseInLocation(utils.FormatDate, dateTime, time.Local) // 这里的目的是为了处理成北京时间,所以多转一遍
  85. val := item.Close[k]
  86. saveVal := utils.SubFloatToString(val, 30)
  87. //如果传入的开始时间是空的,且当前数据日期早于传入的开始日期,那么需要判断下当前日期的数据是否存在
  88. if !startDateTime.IsZero() && t.Before(startDateTime) {
  89. tmpItem, tmpErr := GetEdbDataByDate(source, edbCode, dateTime)
  90. if tmpErr == nil && tmpItem != nil {
  91. existMap[tmpItem.DataTime] = tmpItem
  92. }
  93. }
  94. if findItem, ok := existMap[dateTime]; !ok {
  95. if _, existOk := addMap[dateTime]; !existOk {
  96. isAdd = true
  97. addSql += GetAddSql(edbInfoIdStr, edbCode, dateTime, timeStr, saveVal)
  98. addMap[dateTime] = saveVal
  99. }
  100. } else {
  101. if findItem != nil && utils.SubFloatToString(findItem.Value, 30) != saveVal {
  102. err = ModifyEdbDataById(source, findItem.EdbDataId, saveVal)
  103. if err != nil {
  104. return err
  105. }
  106. }
  107. }
  108. // 下面代码主要目的是处理掉手动插入的数据判断
  109. {
  110. if realDataMaxDate.IsZero() || t.After(realDataMaxDate) {
  111. realDataMaxDate = t
  112. }
  113. if edbDataInsertConfigDate.IsZero() || t.Equal(edbDataInsertConfigDate) {
  114. isFindConfigDateRealData = true
  115. }
  116. }
  117. }
  118. // 处理手工数据补充的配置
  119. HandleConfigInsertEdbData(realDataMaxDate, edbDataInsertConfig, edbInfoId, source, existMap, isFindConfigDateRealData)
  120. if isAdd {
  121. addSql = strings.TrimRight(addSql, ",")
  122. _, err = o.Raw(addSql).Exec()
  123. if err != nil {
  124. fmt.Println("RefreshEdbDataFromWind add Err", err.Error())
  125. return
  126. }
  127. }
  128. return
  129. }
  130. type ServerUrlCountGroup struct {
  131. ServerUrl string
  132. Total int
  133. }
  134. // GetServerUrlCountGroupByServerUrl 获取服务器地址分组总数
  135. func GetServerUrlCountGroupByServerUrl() (list []ServerUrlCountGroup, err error) {
  136. o := orm.NewOrm()
  137. sql := `select server_url,count(1) total from edb_info where source = 2 and server_url != "" group by server_url; `
  138. _, err = o.Raw(sql).QueryRows(&list)
  139. return
  140. }