base_from_wind.go 6.7 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217
  1. package services
  2. import (
  3. "encoding/json"
  4. "errors"
  5. "eta/eta_index_lib/models"
  6. "eta/eta_index_lib/services/alarm_msg"
  7. "eta/eta_index_lib/utils"
  8. "fmt"
  9. "github.com/rdlucklib/rdluck_tools/http"
  10. "time"
  11. )
  12. // GetEdbDataFromWind 获取wind数据
  13. func GetEdbDataFromWind(edbCode, startDate, endDate string) (item *models.EdbDataFromWind, errorCode int, err error) {
  14. windUrl, err := GetWindUrl(edbCode)
  15. if err != nil {
  16. errorCode = 421
  17. go alarm_msg.SendAlarmMsg(fmt.Sprintf("获取wind服务器地址失败,err:%s", err.Error()), 3)
  18. return
  19. }
  20. thsUrl := windUrl + `edbInfo/wind?EdbCode=%s&StartDate=%s&EndDate=%s`
  21. thsUrl = fmt.Sprintf(thsUrl, edbCode, startDate, endDate)
  22. utils.FileLog.Info(fmt.Sprintf("windUrl:%s", thsUrl))
  23. body, err := http.Get(thsUrl)
  24. fmt.Println("GetEdbDataByWind body:")
  25. fmt.Println(string(body))
  26. utils.FileLog.Info(fmt.Sprint("指标编码:", edbCode, ";wind result:", string(body)))
  27. if err != nil {
  28. return
  29. }
  30. item = new(models.EdbDataFromWind)
  31. err = json.Unmarshal(body, &item)
  32. //异常的话,需要邮件通知
  33. if len(item.ErrorCode) > 0 {
  34. if item.ErrorCode["0"] != 0 {
  35. if item.ErrorCode["0"] == -40522017 {
  36. //{
  37. //DT: {
  38. //0: 1654646400000
  39. //},
  40. //CLOSE: {
  41. //0: "CEDBService:: quota exceeded."
  42. //},
  43. //ErrorCode: {
  44. //0: -40522017
  45. //}
  46. //}
  47. // 设置服务器已超限
  48. SetIsLimitEdbCodeInWindUrl(windUrl)
  49. err = DeleteEdbCodeInWindUrl(edbCode)
  50. if err != nil {
  51. return
  52. }
  53. return GetEdbDataFromWind(edbCode, startDate, endDate)
  54. } else if item.ErrorCode["0"] == -40520005 {
  55. //.ErrorCode=-40520005
  56. //.Data=[No Python API Authority
  57. SetIsLimitEdbCodeInWindUrl(windUrl)
  58. err = DeleteEdbCodeInWindUrl(edbCode)
  59. if err != nil {
  60. return
  61. }
  62. go alarm_msg.SendAlarmMsg(fmt.Sprintf("wind数据服务异常,edbCode:%s,ErrorCode:%d,result:%s", edbCode, item.ErrorCode["0"], string(body)), 3)
  63. return GetEdbDataFromWind(edbCode, startDate, endDate)
  64. } else {
  65. go alarm_msg.SendAlarmMsg(fmt.Sprintf("wind数据服务异常,edbCode:%s,ErrorCode:%d,result:%s", edbCode, item.ErrorCode["0"], string(body)), 3)
  66. }
  67. }
  68. }
  69. return
  70. }
  71. // GetEdbDataFromWindUrl 通过url获取wind数据
  72. func GetEdbDataFromWindUrl(windUrl, edbCode, startDate, endDate string) (item *models.EdbDataFromWind, errorCode int, err error) {
  73. if windUrl == `` {
  74. return GetEdbDataFromWind(edbCode, startDate, endDate)
  75. }
  76. requestWindUrl := windUrl + `edbInfo/wind?EdbCode=%s&StartDate=%s&EndDate=%s`
  77. requestWindUrl = fmt.Sprintf(requestWindUrl, edbCode, startDate, endDate)
  78. utils.FileLog.Info(fmt.Sprintf("windUrl:%s", requestWindUrl))
  79. body, err := http.Get(requestWindUrl)
  80. fmt.Println("GetEdbDataByWind body:")
  81. fmt.Println(string(body))
  82. utils.FileLog.Info(fmt.Sprint("wind result:", string(body)))
  83. if err != nil {
  84. return
  85. }
  86. item = new(models.EdbDataFromWind)
  87. err = json.Unmarshal(body, &item)
  88. //异常的话,需要邮件通知
  89. if len(item.ErrorCode) > 0 {
  90. if item.ErrorCode["0"] != 0 {
  91. if item.ErrorCode["0"] == -40522017 {
  92. //{
  93. //DT: {
  94. //0: 1654646400000
  95. //},
  96. //CLOSE: {
  97. //0: "CEDBService:: quota exceeded."
  98. //},
  99. //ErrorCode: {
  100. //0: -40522017
  101. //}
  102. //}
  103. // 设置服务器已超限
  104. errorCode = 421
  105. go alarm_msg.SendAlarmMsg(fmt.Sprintf("wind数据接口超限,地址:%s", requestWindUrl), 2)
  106. //go utils.SendEmail("wind数据接口超限", "地址:"+requestWindUrl, utils.EmailSendToUsers)
  107. return
  108. } else if item.ErrorCode["0"] == -40520005 {
  109. //.ErrorCode=-40520005
  110. //.Data=[No Python API Authority
  111. err = errors.New("No Python API Authority")
  112. go alarm_msg.SendAlarmMsg(fmt.Sprintf("wind数据服务异常,edbCode:%s,ErrorCode:%d,result:%s", edbCode, item.ErrorCode["0"], string(body)), 3)
  113. return
  114. } else {
  115. go alarm_msg.SendAlarmMsg(fmt.Sprintf("wind数据服务异常,edbCode:%s,ErrorCode:%d,result:%s", edbCode, item.ErrorCode["0"], string(body)), 3)
  116. }
  117. }
  118. }
  119. return
  120. }
  121. // GetWindUrl 获取wind的url
  122. func GetWindUrl(edbCode string) (windUrl string, err error) {
  123. defer func() {
  124. if err == nil && windUrl == "" {
  125. err = errors.New("获取wind服务器地址失败,指标超限了")
  126. }
  127. }()
  128. //从缓存中获取
  129. cacheKey := utils.CACHE_WIND_URL + ":" + edbCode
  130. windUrl, _ = utils.Rc.RedisString(cacheKey)
  131. if windUrl != "" {
  132. return
  133. }
  134. //如果缓存中没有的话,那么从配置中获取
  135. for _, windUrlMap := range utils.Hz_Wind_Data_Url_LIST {
  136. //判断该url是否被占满了
  137. //count, tmpErr := GetCountEdbCodeInWindUrl(windUrlMap.Url)
  138. //if tmpErr != nil && tmpErr.Error() != "nil returned" {
  139. // err = tmpErr
  140. // return
  141. //}
  142. //if count < windUrlMap.Num {
  143. // windUrl = windUrlMap.Url
  144. // AddEdbCodeInWindUrl(windUrlMap.Url, edbCode)
  145. // return
  146. //}
  147. //如果超限了,那么进入下一循环
  148. isLimit, tmpErr := GetIsLimitEdbCodeInWindUrl(windUrlMap.Url)
  149. if isLimit {
  150. err = tmpErr
  151. continue
  152. }
  153. windUrl = windUrlMap.Url
  154. AddEdbCodeInWindUrl(windUrlMap.Url, edbCode)
  155. return
  156. }
  157. return
  158. }
  159. // GetCountEdbCodeInWindUrl 从缓存key中获取已经插入入的指标数
  160. func GetCountEdbCodeInWindUrl(windUrl string) (num int, err error) {
  161. cacheKey := utils.CACHE_WIND_URL + time.Now().Format(utils.FormatDateUnSpace) + ":" + utils.MD5(windUrl)
  162. num, err = utils.Rc.RedisInt(cacheKey)
  163. if err != nil && err.Error() == "redigo: nil returned" {
  164. err = nil
  165. }
  166. return
  167. }
  168. // GetIsLimitEdbCodeInWindUrl 从缓存key中获取是否超限
  169. func GetIsLimitEdbCodeInWindUrl(windUrl string) (isLimit bool, err error) {
  170. cacheKey := utils.CACHE_WIND_URL + ":limit:" + time.Now().Format(utils.FormatDateUnSpace) + ":" + utils.MD5(windUrl)
  171. num, err := utils.Rc.RedisInt(cacheKey)
  172. if err != nil && err.Error() == "redigo: nil returned" {
  173. err = nil
  174. }
  175. if num > 0 {
  176. isLimit = true
  177. }
  178. return
  179. }
  180. // SetIsLimitEdbCodeInWindUrl 设置服务器已超限
  181. func SetIsLimitEdbCodeInWindUrl(windUrl string) {
  182. cacheKey := utils.CACHE_WIND_URL + ":limit:" + time.Now().Format(utils.FormatDateUnSpace) + ":" + utils.MD5(windUrl)
  183. _ = utils.Rc.SetNX(cacheKey, 1, utils.GetTodayLastSecond())
  184. return
  185. }
  186. // AddEdbCodeInWindUrl 将指标插入到缓存key中
  187. // @return isInsert bool 是否插入数据,true时为插入数据,false表示数据已存在
  188. func AddEdbCodeInWindUrl(windUrl, edbCode string) (isInsert bool) {
  189. cacheKey := utils.CACHE_WIND_URL + ":" + edbCode
  190. isInsert = utils.Rc.SetNX(cacheKey, windUrl, utils.GetTodayLastSecond())
  191. cacheKey2 := utils.CACHE_WIND_URL + time.Now().Format(utils.FormatDateUnSpace) + ":" + utils.MD5(windUrl)
  192. utils.Rc.Incrby(cacheKey2, 1)
  193. return
  194. }
  195. // DeleteEdbCodeInWindUrl 删除指标编码 服务器归属 缓存
  196. func DeleteEdbCodeInWindUrl(edbCode string) (err error) {
  197. cacheKey := utils.CACHE_WIND_URL + ":" + edbCode
  198. err = utils.Rc.Delete(cacheKey)
  199. return
  200. }