edb_refresh.go 31 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945946947948
  1. package services
  2. import (
  3. "context"
  4. "encoding/json"
  5. "eta/eta_task/models"
  6. "eta/eta_task/models/data_manage"
  7. "eta/eta_task/models/data_manage/edb_refresh"
  8. "eta/eta_task/services/alarm_msg"
  9. "eta/eta_task/services/data"
  10. "eta/eta_task/utils"
  11. "fmt"
  12. "strings"
  13. "sync"
  14. "time"
  15. )
  16. // ConfigRefreshData
  17. // @Description: 配置刷新数据
  18. // @author: Roc
  19. // @datetime 2024-01-10 13:55:05
  20. // @param cont context.Context
  21. // @return err error
  22. func ConfigRefreshData(cont context.Context) (err error) {
  23. errMsgList := make([]string, 0)
  24. defer func() {
  25. if err != nil {
  26. fmt.Println(err)
  27. }
  28. }()
  29. // 一期是只做wind、同花顺、钢联、有色
  30. now := time.Now()
  31. //now = time.Date(2023, 12, 31, 19, 10, 59, 0, time.Local)
  32. //now = time.Date(2023, 12, 31, 16, 50, 59, 0, time.Local)
  33. defaultSourceEdbInfoListMap, err := getDefaultRefreshData(now)
  34. if err != nil {
  35. errMsgList = append(errMsgList, "获取默认刷新数据失败,Err:"+err.Error())
  36. }
  37. sourceEdbInfoListMap, err := getConfigRefreshData(now)
  38. if err != nil {
  39. errMsgList = append(errMsgList, "获取指标配置刷新数据失败,Err:"+err.Error())
  40. }
  41. // 将两个合并
  42. allSourceEdbInfoListMap := mergeMaps(defaultSourceEdbInfoListMap, sourceEdbInfoListMap)
  43. wgNum := len(allSourceEdbInfoListMap)
  44. if wgNum <= 0 {
  45. return
  46. }
  47. wg := sync.WaitGroup{}
  48. wg.Add(wgNum)
  49. for _, edbList := range allSourceEdbInfoListMap {
  50. if edbList == nil {
  51. wg.Done()
  52. continue
  53. }
  54. //if len(edbList) != 0 {
  55. // // debug环境仅测试刷新钢联
  56. // if edbList[0].Source == utils.DATA_SOURCE_MYSTEEL_CHEMICAL && utils.RunMode == "debug" {
  57. //
  58. // }
  59. //}
  60. if len(edbList) > 0 {
  61. go BaseRefreshData(&wg, edbList[0].Source, edbList[0].SubSource, edbList)
  62. } else {
  63. wg.Done()
  64. }
  65. }
  66. wg.Wait()
  67. fmt.Println("Refresh End")
  68. return
  69. }
  70. // Function to merge two maps
  71. func mergeMaps(dst map[string][]*edb_refresh.EdbInfoListAndRefreshConfig, src map[string][]*edb_refresh.EdbInfoListAndRefreshConfig) (newMap map[string][]*edb_refresh.EdbInfoListAndRefreshConfig) {
  72. if dst == nil {
  73. return src
  74. }
  75. if src == nil {
  76. return dst
  77. }
  78. newMap = dst
  79. for k, v := range src {
  80. if newK, ok := newMap[k]; ok {
  81. newK = append(newK, v...)
  82. newMap[k] = newK
  83. } else {
  84. newMap[k] = v
  85. }
  86. }
  87. return newMap
  88. }
  89. // getDefaultRefreshData
  90. // @Description: 根据默认配置获取需要刷新的指标列表
  91. // @author: Roc
  92. // @datetime 2024-01-10 13:55:38
  93. // @param now time.Time
  94. // @return sourceEdbInfoListMap map[string][]*edb_refresh.EdbInfoListAndRefreshConfig
  95. // @return err error
  96. func getDefaultRefreshData(now time.Time) (sourceEdbInfoListMap map[string][]*edb_refresh.EdbInfoListAndRefreshConfig, err error) {
  97. errMsgList := make([]string, 0)
  98. defer func() {
  99. if err != nil {
  100. fmt.Println(err)
  101. }
  102. }()
  103. // 一期是只做wind、同花顺、钢联、有色
  104. sourceEdbInfoListMap = make(map[string][]*edb_refresh.EdbInfoListAndRefreshConfig)
  105. currTimeStr := getPreviousHalfHour(now)
  106. fmt.Println(currTimeStr)
  107. // 所有默认配置刷新项
  108. list := make([]*edb_refresh.EdbRefreshDefaultConfig, 0)
  109. //刷新频率,枚举值:每自然日、每交易日、每周、每旬、每月、每季、每半年、每年
  110. refreshFrequencyList := []string{"每自然日", "每交易日", "每周", "每旬", "每月", "每季", "每半年", "每年"}
  111. conf, err := models.GetBusinessConf()
  112. if err != nil {
  113. fmt.Println(err)
  114. utils.FileLog.Info("获取业务配置失败,Err:" + err.Error())
  115. return
  116. }
  117. // 获取钢联化工的数据获取方式
  118. mySteelChemicalDataMethod := "excel"
  119. if v, ok := conf["MySteelDataMethod"]; ok {
  120. if v == "api" {
  121. mySteelChemicalDataMethod = v
  122. }
  123. }
  124. utils.FileLog.Info("获取业务配置,MySteelDataMethod:" + mySteelChemicalDataMethod)
  125. // 获取各个刷新频率的配置
  126. for _, refreshFrequency := range refreshFrequencyList {
  127. // 获取刷新频率条件
  128. condition, pars, isHandler := getRefreshFrequencyCondition(now, refreshFrequency)
  129. if !isHandler {
  130. // 可能是非交易日,所以过滤不处理
  131. continue
  132. }
  133. condition += ` AND refresh_frequency = ? AND refresh_time = ?`
  134. pars = append(pars, refreshFrequency, currTimeStr)
  135. // 这两个是excel的数据源,他是从公共机更新的,需要过滤掉
  136. if mySteelChemicalDataMethod == "api" {
  137. // 钢联化工使用api的方式获取数据的,不需要过滤
  138. condition += ` AND source not in (?)`
  139. pars = append(pars, utils.DATA_SOURCE_YS)
  140. } else {
  141. condition += ` AND source not in (?,?)`
  142. pars = append(pars, utils.DATA_SOURCE_MYSTEEL_CHEMICAL, utils.DATA_SOURCE_YS)
  143. }
  144. tmpList, tmpErr := edb_refresh.GetListByCondition(condition, pars)
  145. if tmpErr != nil {
  146. err = tmpErr
  147. return
  148. }
  149. list = append(list, tmpList...)
  150. }
  151. // 更新的单元格数
  152. refreshDataNumMap := make(map[string]*edb_refresh.EdbRefreshDefaultConfig)
  153. // 数据源刷新频度的列表数组
  154. refreshDataFrequencyListMap := make(map[int]map[int][]string)
  155. wgNum := 0
  156. // 处理待刷新的数据源,整理成数组,方便获取对应的指标
  157. for _, item := range list {
  158. // 更新的单元格数
  159. key := fmt.Sprintf("%d_%d_%s", item.Source, item.SubSource, item.Frequency)
  160. refreshDataNumMap[key] = item
  161. // 数据源刷新频度的列表数组
  162. subSourceFrequencyList, ok := refreshDataFrequencyListMap[item.Source]
  163. if !ok {
  164. subSourceFrequencyList = make(map[int][]string)
  165. }
  166. frequencyList, ok := subSourceFrequencyList[item.SubSource]
  167. if !ok {
  168. wgNum++
  169. frequencyList = make([]string, 0)
  170. }
  171. subSourceFrequencyList[item.SubSource] = append(frequencyList, item.Frequency)
  172. refreshDataFrequencyListMap[item.Source] = subSourceFrequencyList
  173. }
  174. for source, subSourceFrequencyListMap := range refreshDataFrequencyListMap {
  175. switch source {
  176. case utils.DATA_SOURCE_MYSTEEL_CHEMICAL, utils.DATA_SOURCE_YS:
  177. // 只处理钢联化工使用api方式获取数据的情况
  178. if mySteelChemicalDataMethod == "api" {
  179. for subSource, frequencyList := range subSourceFrequencyListMap {
  180. items, tmpErr := data_manage.GetBaseFromMysteelChemicalIndexItems(frequencyList)
  181. if tmpErr != nil {
  182. errMsgList = append(errMsgList, fmt.Sprint("source:", source, "subSource:", subSource, "frequencyList:", strings.Join(frequencyList, ","), "err:", tmpErr.Error()))
  183. }
  184. indexList := make([]*edb_refresh.EdbInfoListAndRefreshConfig, 0)
  185. for _, v := range items {
  186. tmpConf := new(edb_refresh.EdbInfoListAndRefreshConfig)
  187. // 数据刷新的期数
  188. dataRefreshNum := utils.DATA_REFRESH
  189. key := fmt.Sprintf("%d_%d_%s", utils.DATA_SOURCE_MYSTEEL_CHEMICAL, 0, v.Frequency)
  190. if edbRefreshDefaultConfig, ok := refreshDataNumMap[key]; ok {
  191. if edbRefreshDefaultConfig.RefreshAllData == 1 { // 刷新所有数据期数
  192. dataRefreshNum = 0
  193. } else if edbRefreshDefaultConfig.RefreshDataNum > 0 { //
  194. dataRefreshNum = edbRefreshDefaultConfig.RefreshDataNum
  195. }
  196. }
  197. tmpConf.EdbCode = v.IndexCode
  198. tmpConf.EdbName = v.IndexName
  199. tmpConf.Source = utils.DATA_SOURCE_MYSTEEL_CHEMICAL
  200. tmpConf.Frequency = v.Frequency
  201. tmpConf.Unit = v.Unit
  202. tmpConf.StartDate, err = time.Parse(utils.FormatDate, v.StartDate)
  203. tmpConf.EndDate, err = time.Parse(utils.FormatDate, v.EndDate)
  204. tmpConf.ClassifyId = int(v.BaseFromMysteelChemicalClassifyId)
  205. tmpConf.DataRefreshNum = dataRefreshNum
  206. tmpConf.EdbInfoId = v.EdbInfoId
  207. indexList = append(indexList, tmpConf)
  208. }
  209. key := fmt.Sprint(source, "_", subSource)
  210. sourceEdbInfoListMap[key] = indexList
  211. }
  212. }
  213. // 其他情况不处理
  214. default:
  215. for subSource, frequencyList := range subSourceFrequencyListMap {
  216. edbList, tmpErr := edb_refresh.GetDefaultRefreshEdbInfoListBySourceAndSubSource(source, subSource, frequencyList)
  217. if tmpErr != nil {
  218. errMsgList = append(errMsgList, fmt.Sprint("source:", source, "subSource:", subSource, "frequencyList:", strings.Join(frequencyList, ","), "err:", tmpErr.Error()))
  219. }
  220. for _, v := range edbList {
  221. // 数据刷新的期数
  222. dataRefreshNum := utils.DATA_REFRESH
  223. key := fmt.Sprintf("%d_%d_%s", v.Source, v.SubSource, v.Frequency)
  224. if edbRefreshDefaultConfig, ok := refreshDataNumMap[key]; ok {
  225. if edbRefreshDefaultConfig.RefreshAllData == 1 { // 刷新所有数据期数
  226. dataRefreshNum = 0
  227. } else if edbRefreshDefaultConfig.RefreshDataNum > 0 { //
  228. dataRefreshNum = edbRefreshDefaultConfig.RefreshDataNum
  229. }
  230. }
  231. v.DataRefreshNum = dataRefreshNum
  232. }
  233. key := fmt.Sprint(source, "_", subSource)
  234. sourceEdbInfoListMap[key] = edbList
  235. }
  236. }
  237. }
  238. fmt.Println("Get Refresh End")
  239. return
  240. }
  241. // getConfigRefreshData
  242. // @Description: 根据指标配置获取需要刷新的指标列表
  243. // @author: Roc
  244. // @datetime 2024-01-10 13:55:59
  245. // @param now time.Time
  246. // @return sourceEdbInfoListMap map[string][]*edb_refresh.EdbInfoListAndRefreshConfig
  247. // @return err error
  248. func getConfigRefreshData(now time.Time) (sourceEdbInfoListMap map[string][]*edb_refresh.EdbInfoListAndRefreshConfig, err error) {
  249. defer func() {
  250. if err != nil {
  251. fmt.Println(err)
  252. }
  253. }()
  254. // 一期是只做wind、同花顺、钢联、有色
  255. sourceEdbInfoListMap = make(map[string][]*edb_refresh.EdbInfoListAndRefreshConfig)
  256. currTimeStr := getPreviousHalfHour(now)
  257. // 所有默认配置刷新项
  258. list := make([]*edb_refresh.EdbRefreshConfig, 0)
  259. //刷新频率,枚举值:每自然日、每交易日、每周、每旬、每月、每季、每半年、每年
  260. refreshFrequencyList := []string{"每自然日", "每交易日", "每周", "每旬", "每月", "每季", "每半年", "每年"}
  261. // 获取各个刷新频率的配置
  262. for _, refreshFrequency := range refreshFrequencyList {
  263. // 获取刷新频率条件
  264. condition, pars, isHandler := getRefreshFrequencyCondition(now, refreshFrequency)
  265. if !isHandler {
  266. // 可能是非交易日,所以过滤不处理
  267. continue
  268. }
  269. condition += ` AND refresh_frequency = ? AND refresh_time = ?`
  270. pars = append(pars, refreshFrequency, currTimeStr)
  271. tmpList, tmpErr := edb_refresh.GetEdbRefreshConfigListByCondition(condition, pars)
  272. if tmpErr != nil {
  273. err = tmpErr
  274. return
  275. }
  276. list = append(list, tmpList...)
  277. }
  278. // 配置列表
  279. configIdEdbRefreshConfigMap := make(map[int]*edb_refresh.EdbRefreshConfig)
  280. configIdList := make([]int, 0)
  281. for _, v := range list {
  282. configIdList = append(configIdList, v.EdbRefreshConfigId)
  283. configIdEdbRefreshConfigMap[v.EdbRefreshConfigId] = v
  284. }
  285. conf, err := models.GetBusinessConf()
  286. if err != nil {
  287. fmt.Println(err)
  288. return
  289. }
  290. // 获取钢联化工的数据获取方式
  291. mySteelChemicalDataMethod := "excel"
  292. if v, ok := conf["MySteelDataMethod"]; ok {
  293. if v == "api" {
  294. mySteelChemicalDataMethod = v
  295. }
  296. }
  297. // 当钢联的数据获取方式是api时,不用过滤
  298. var sourceList []int
  299. if mySteelChemicalDataMethod == "api" {
  300. sourceList = []int{utils.DATA_SOURCE_YS}
  301. } else {
  302. sourceList = []int{utils.DATA_SOURCE_MYSTEEL_CHEMICAL, utils.DATA_SOURCE_YS}
  303. }
  304. edbInfoList, err := edb_refresh.GetConfigRefreshEdbInfoListBySourceAndSubSource(sourceList, configIdList)
  305. if err != nil {
  306. return
  307. }
  308. for _, v := range edbInfoList {
  309. key := fmt.Sprint(v.Source, "_", v.SubSource)
  310. tmpList, ok := sourceEdbInfoListMap[key]
  311. if !ok {
  312. tmpList = make([]*edb_refresh.EdbInfoListAndRefreshConfig, 0)
  313. }
  314. // 数据刷新的期数
  315. dataRefreshNum := utils.DATA_REFRESH
  316. if edbRefreshConfig, ok2 := configIdEdbRefreshConfigMap[v.EdbRefreshConfigId]; ok2 {
  317. if edbRefreshConfig.RefreshAllData == 1 { // 刷新所有数据期数
  318. dataRefreshNum = 0
  319. } else if edbRefreshConfig.RefreshDataNum > 0 { //
  320. dataRefreshNum = edbRefreshConfig.RefreshDataNum
  321. }
  322. }
  323. v.DataRefreshNum = dataRefreshNum
  324. sourceEdbInfoListMap[key] = append(tmpList, v)
  325. }
  326. fmt.Println("Get ConfigRefreshData End")
  327. return
  328. }
  329. func needForUpdate(date time.Time, frequency string) bool {
  330. //如果当前已经更新最新的数据则停止更新
  331. today := time.Now()
  332. switch frequency {
  333. case "日度":
  334. return !date.Equal(today)
  335. case "周度":
  336. // 获取本周的开始日期(周一)
  337. startOfWeek := today.AddDate(0, 0, int(time.Monday-today.Weekday()))
  338. return date.Before(startOfWeek)
  339. case "旬度":
  340. day := today.Day()
  341. var beginOfTenDays time.Time
  342. if day <= 10 {
  343. beginOfTenDays = time.Date(today.Year(), today.Month(), 1, 0, 0, 0, 0, time.Local)
  344. return date.Before(beginOfTenDays)
  345. } else if day <= 20 {
  346. beginOfTenDays = time.Date(today.Year(), today.Month(), 11, 0, 0, 0, 0, time.Local)
  347. return date.Before(beginOfTenDays)
  348. } else {
  349. beginOfTenDays = time.Date(today.Year(), today.Month(), 21, 0, 0, 0, 0, time.Local)
  350. return date.Before(beginOfTenDays)
  351. }
  352. case "月度":
  353. beginOfMonth := time.Date(today.Year(), today.Month(), 1, 0, 0, 0, 0, time.Local)
  354. return date.Before(beginOfMonth)
  355. case "季度":
  356. month := today.Month()
  357. var beginOfQuarter time.Time
  358. if month <= 3 {
  359. beginOfQuarter = time.Date(today.Year(), 1, 1, 0, 0, 0, 0, time.Local)
  360. } else if month <= 6 {
  361. beginOfQuarter = time.Date(today.Year(), 4, 1, 0, 0, 0, 0, time.Local)
  362. } else if month <= 9 {
  363. beginOfQuarter = time.Date(today.Year(), 7, 1, 0, 0, 0, 0, time.Local)
  364. } else {
  365. beginOfQuarter = time.Date(today.Year(), 10, 1, 0, 0, 0, 0, time.Local)
  366. }
  367. return date.Before(beginOfQuarter)
  368. case "半年度":
  369. month := today.Month()
  370. var beginOfHalfYear time.Time
  371. if month <= 6 {
  372. beginOfHalfYear = time.Date(today.Year(), 1, 1, 0, 0, 0, 0, time.Local)
  373. } else {
  374. beginOfHalfYear = time.Date(today.Year(), 7, 1, 0, 0, 0, 0, time.Local)
  375. }
  376. return date.Before(beginOfHalfYear)
  377. case "年度":
  378. startOfYear := time.Date(today.Year(), 1, 1, 0, 0, 0, 0, time.Local)
  379. return date.Before(startOfYear)
  380. default:
  381. return true
  382. }
  383. }
  384. // BaseRefreshData
  385. // @Description: 基础数据刷新
  386. // @author: Roc
  387. // @datetime 2024-01-09 16:27:45
  388. // @param wg *sync.WaitGroup
  389. // @return err error
  390. func BaseRefreshData(wg *sync.WaitGroup, source, subSource int, items []*edb_refresh.EdbInfoListAndRefreshConfig) (err error) {
  391. errMsgList := make([]string, 0)
  392. defer func() {
  393. if err != nil {
  394. fmt.Println("来源:", source, ";子来源:", subSource, ";BaseRefreshData Err:"+err.Error())
  395. go alarm_msg.SendAlarmMsg(fmt.Sprint("来源:", source, ";子来源:", subSource, ";BaseRefreshData ErrMsg:"+err.Error()), 3)
  396. }
  397. if len(errMsgList) > 0 {
  398. errMsg := fmt.Sprint("来源:", source, ";子来源:", subSource, ";BaseRefreshData Err:"+strings.Join(errMsgList, "\n"))
  399. fmt.Println(errMsg)
  400. go alarm_msg.SendAlarmMsg(errMsg, 3)
  401. }
  402. wg.Done()
  403. }()
  404. // 数据刷新的期数
  405. dataRefreshNum := utils.DATA_REFRESH
  406. // 是否从最开始的日期更新
  407. var isRefreshByStartDate bool
  408. if source != utils.DATA_SOURCE_THS {
  409. for _, v := range items {
  410. // 如果暂停更新,那就过滤
  411. if v.NoUpdate == 1 {
  412. continue
  413. }
  414. if !needForUpdate(v.EndDate, v.Frequency) {
  415. continue
  416. }
  417. if v.DataRefreshNum > 0 {
  418. dataRefreshNum = v.DataRefreshNum
  419. }
  420. startDate := ""
  421. if isRefreshByStartDate {
  422. startDate = v.StartDate.Format(utils.FormatDate)
  423. } else {
  424. if v.Frequency == "日度" {
  425. startDate = v.EndDate.AddDate(0, 0, -dataRefreshNum).Format(utils.FormatDate)
  426. } else if v.Frequency == "周度" {
  427. startDate = v.EndDate.AddDate(0, 0, -(dataRefreshNum * 7)).Format(utils.FormatDate)
  428. } else if v.Frequency == "旬度" {
  429. startDate = v.EndDate.AddDate(0, 0, -(dataRefreshNum * 10)).Format(utils.FormatDate)
  430. } else if v.Frequency == "月度" {
  431. startDate = v.EndDate.AddDate(0, -dataRefreshNum, 0).Format(utils.FormatDate)
  432. } else if v.Frequency == "季度" {
  433. startDate = v.EndDate.AddDate(0, -dataRefreshNum*3, 0).Format(utils.FormatDate)
  434. } else if v.Frequency == "半年度" {
  435. startDate = v.EndDate.AddDate(0, -dataRefreshNum*6, 0).Format(utils.FormatDate)
  436. } else if v.Frequency == "年度" {
  437. startDate = v.EndDate.AddDate(-dataRefreshNum, 0, 0).Format(utils.FormatDate)
  438. } else {
  439. startDate = v.EndDate.AddDate(0, 0, -utils.DATA_REFRESH).Format(utils.FormatDate)
  440. }
  441. }
  442. fmt.Println(startDate)
  443. // 数据更新
  444. resp, tmpErr := data.RefreshEdbData(v.EdbInfoId, v.Source, v.SubSource, v.EdbCode, startDate)
  445. if tmpErr != nil {
  446. errMsgList = append(errMsgList, v.EdbCode+"RefreshEdbData Err:"+tmpErr.Error())
  447. continue
  448. }
  449. if resp.Ret != 200 {
  450. errMsgList = append(errMsgList, v.EdbCode+";RefreshEdbData Err:"+resp.Msg+";ErrMsg:"+resp.ErrMsg)
  451. continue
  452. }
  453. }
  454. }
  455. // 同花顺控制速率, 每秒最多4次请求(同花顺EDB函数限制为5, 考虑到可能存在用户同时在使用, 这里批量刷新设置为4, 话说5确实是有点少了吧=_=!)
  456. if source == utils.DATA_SOURCE_THS {
  457. ticker := time.NewTicker(250 * time.Millisecond)
  458. defer ticker.Stop()
  459. for _, v := range items {
  460. <-ticker.C
  461. // 如果暂停更新,那就过滤
  462. if v.NoUpdate == 1 {
  463. continue
  464. }
  465. if v.DataRefreshNum > 0 {
  466. dataRefreshNum = v.DataRefreshNum
  467. }
  468. startDate := ""
  469. if isRefreshByStartDate {
  470. startDate = v.StartDate.Format(utils.FormatDate)
  471. } else {
  472. if v.Frequency == "日度" {
  473. startDate = v.EndDate.AddDate(0, 0, -dataRefreshNum).Format(utils.FormatDate)
  474. } else if v.Frequency == "周度" {
  475. startDate = v.EndDate.AddDate(0, 0, -(dataRefreshNum * 7)).Format(utils.FormatDate)
  476. } else if v.Frequency == "旬度" {
  477. startDate = v.EndDate.AddDate(0, 0, -(dataRefreshNum * 10)).Format(utils.FormatDate)
  478. } else if v.Frequency == "月度" {
  479. startDate = v.EndDate.AddDate(0, -dataRefreshNum, 0).Format(utils.FormatDate)
  480. } else if v.Frequency == "季度" {
  481. startDate = v.EndDate.AddDate(0, -dataRefreshNum*3, 0).Format(utils.FormatDate)
  482. } else if v.Frequency == "半年度" {
  483. startDate = v.EndDate.AddDate(0, -dataRefreshNum*6, 0).Format(utils.FormatDate)
  484. } else if v.Frequency == "年度" {
  485. startDate = v.EndDate.AddDate(-dataRefreshNum, 0, 0).Format(utils.FormatDate)
  486. } else {
  487. startDate = v.EndDate.AddDate(0, 0, -utils.DATA_REFRESH).Format(utils.FormatDate)
  488. }
  489. }
  490. fmt.Println(startDate)
  491. // 数据更新
  492. resp, tmpErr := data.RefreshEdbData(v.EdbInfoId, v.Source, v.SubSource, v.EdbCode, startDate)
  493. if tmpErr != nil {
  494. errMsgList = append(errMsgList, v.EdbCode+"RefreshEdbData Err:"+tmpErr.Error())
  495. continue
  496. }
  497. if resp.Ret != 200 {
  498. errMsgList = append(errMsgList, v.EdbCode+";RefreshEdbData Err:"+resp.Msg+";ErrMsg:"+resp.ErrMsg)
  499. continue
  500. }
  501. }
  502. }
  503. fmt.Println("来源:", source, ";子来源:", subSource, "刷新结束")
  504. return err
  505. }
  506. // getRefreshFrequencyCondition
  507. // @Description: 根据时间和刷新频率获取条件
  508. // @author: Roc
  509. // @datetime 2024-01-09 16:27:11
  510. // @param now time.Time
  511. // @param refreshFrequency string
  512. // @return condition string
  513. // @return pars []interface{}
  514. // @return isHandler bool
  515. func getRefreshFrequencyCondition(now time.Time, refreshFrequency string) (condition string, pars []interface{}, isHandler bool) {
  516. isHandler = true
  517. var dayNum int
  518. var isLastDay bool
  519. //刷新频率,枚举值:每自然日、每交易日、每周、每旬、每月、每季、每半年、每年
  520. switch refreshFrequency {
  521. case "每自然日":
  522. // 自然日不需要额外条件
  523. return
  524. case "每交易日":
  525. // 周六日不处理
  526. if now.Weekday() == time.Saturday || now.Weekday() == time.Sunday {
  527. isHandler = false
  528. }
  529. return
  530. case "每周":
  531. currWeekDay := now.Weekday()
  532. if currWeekDay == time.Sunday {
  533. currWeekDay = 7
  534. isLastDay = true
  535. }
  536. dayNum = int(currWeekDay)
  537. case "每旬":
  538. currDay := now.Day()
  539. if currDay <= 10 {
  540. dayNum = currDay
  541. // 如果是这旬的最后一天
  542. if currDay == 10 {
  543. isLastDay = true
  544. }
  545. } else if currDay <= 20 {
  546. dayNum = currDay - 10
  547. // 如果是这旬的最后一天
  548. if currDay == 20 {
  549. isLastDay = true
  550. }
  551. } else {
  552. dayNum = currDay - 20
  553. // 当月的最后一天
  554. monthLastDay := time.Date(now.Year(), now.Month()+1, 1, 0, 0, 0, 0, time.Local).AddDate(0, 0, -1)
  555. // 如果是这旬的最后一天
  556. if currDay == monthLastDay.Day() {
  557. isLastDay = true
  558. }
  559. }
  560. case "每月":
  561. // 当前日期
  562. currDay := now.Day()
  563. dayNum = currDay
  564. // 当期的最后一天
  565. monthLastDay := time.Date(now.Year(), now.Month()+1, 1, 0, 0, 0, 0, time.Local).AddDate(0, 0, -1)
  566. // 如果是这期的最后一天
  567. if currDay == monthLastDay.Day() {
  568. isLastDay = true
  569. }
  570. case "每季":
  571. // 当期的第一天 ; 当期的最后一天
  572. var startDay, endDay time.Time
  573. currMonth := now.Month()
  574. currDay := now.Day()
  575. if currMonth <= 3 {
  576. // 当季的第一天
  577. startDay = time.Date(now.Year(), 1, 1, 0, 0, 0, 0, time.Local)
  578. // 当季的最后一天
  579. endDay = time.Date(now.Year(), 4, 1, 0, 0, 0, 0, time.Local).AddDate(0, 0, -1)
  580. } else if currMonth <= 6 {
  581. // 当期的第一天
  582. startDay = time.Date(now.Year(), 4, 1, 0, 0, 0, 0, time.Local)
  583. // 当期的最后一天
  584. endDay = time.Date(now.Year(), 7, 1, 0, 0, 0, 0, time.Local).AddDate(0, 0, -1)
  585. } else if currMonth <= 9 {
  586. // 当期的第一天
  587. startDay = time.Date(now.Year(), 7, 1, 0, 0, 0, 0, time.Local)
  588. // 当期的最后一天
  589. endDay = time.Date(now.Year(), 10, 1, 0, 0, 0, 0, time.Local).AddDate(0, 0, -1)
  590. } else {
  591. // 当期的第一天
  592. startDay = time.Date(now.Year(), 10, 1, 0, 0, 0, 0, time.Local)
  593. // 当期的最后一天
  594. endDay = time.Date(now.Year()+1, 1, 1, 0, 0, 0, 0, time.Local).AddDate(0, 0, -1)
  595. }
  596. // 计算这期的第一天和当日的天数
  597. dayNum = utils.GetTimeSubDay(startDay, now) + 1
  598. // 如果是这期的最后一天
  599. if currMonth == endDay.Month() && currDay == endDay.Day() {
  600. isLastDay = true
  601. }
  602. case "每半年":
  603. // 当期的第一天 ; 当期的最后一天
  604. var startDay, endDay time.Time
  605. currMonth := now.Month()
  606. currDay := now.Day()
  607. if currMonth <= 6 {
  608. // 当期的第一天
  609. startDay = time.Date(now.Year(), 1, 1, 0, 0, 0, 0, time.Local)
  610. // 当期的最后一天
  611. endDay = time.Date(now.Year(), 7, 1, 0, 0, 0, 0, time.Local).AddDate(0, 0, -1)
  612. } else {
  613. // 当期的第一天
  614. startDay = time.Date(now.Year(), 7, 1, 0, 0, 0, 0, time.Local)
  615. // 当期的最后一天
  616. endDay = time.Date(now.Year()+1, 1, 1, 0, 0, 0, 0, time.Local).AddDate(0, 0, -1)
  617. }
  618. // 计算这期的第一天和当日的天数
  619. dayNum = utils.GetTimeSubDay(startDay, now) + 1
  620. // 如果是这期的最后一天
  621. if currMonth == endDay.Month() && currDay == endDay.Day() {
  622. isLastDay = true
  623. }
  624. case "每年":
  625. currMonth := now.Month()
  626. currDay := now.Day()
  627. // 当期的第一天
  628. startDay := time.Date(now.Year(), 1, 1, 0, 0, 0, 0, time.Local)
  629. // 当期的最后一天
  630. endDay := time.Date(now.Year()+1, 1, 1, 0, 0, 0, 0, time.Local).AddDate(0, 0, -1)
  631. // 计算这期的第一天和当日的天数
  632. dayNum = utils.GetTimeSubDay(startDay, now) + 1
  633. // 如果是这期的最后一天
  634. if currMonth == endDay.Month() && currDay == endDay.Day() {
  635. isLastDay = true
  636. }
  637. }
  638. // 如果是这期的最后一天,那么就是判断refresh_frequency_day是否配置为0,或者配置的天数大于这期的最大天数
  639. if isLastDay {
  640. condition += ` AND ( refresh_frequency_day = ? OR refresh_frequency_day >= ? )`
  641. pars = append(pars, 0, dayNum)
  642. } else {
  643. // 如果不是这期的最后一天,那么就是判断refresh_frequency_day是否等于配置的天数
  644. condition += ` AND refresh_frequency_day = ? `
  645. pars = append(pars, dayNum)
  646. }
  647. return
  648. }
  649. // getPreviousHalfHour
  650. // @Description: 根据传入的时间获取该时间的前整半小时的时间字符串
  651. // @author: Roc
  652. // @datetime 2024-01-09 14:27:34
  653. // @param now time.Time
  654. // @return string
  655. func getPreviousHalfHour(now time.Time) string {
  656. minute := now.Minute()
  657. if minute >= 30 {
  658. return fmt.Sprintf("%02d:%02d", now.Hour(), 30)
  659. }
  660. return fmt.Sprintf("%02d:%02d", now.Hour(), 0)
  661. }
  662. // 根据配置把钢联化工和wind指标设置成禁止刷新
  663. func DisableEdbRefresh(cont context.Context) (err error) {
  664. //设置刷新key,如果没有执行完 报错提示
  665. cacheKey := "eta_task:DisableEdbRefresh"
  666. deleteCache := true
  667. defer func() {
  668. if deleteCache {
  669. utils.Rc.Delete(cacheKey)
  670. }
  671. if err != nil {
  672. tips := "DisableEdbRefresh-钢联化工、wind、彭博、彭博财务、同花顺指标设置成禁止刷新失败, ErrMsg:\n" + err.Error()
  673. utils.FileLog.Info(tips)
  674. go alarm_msg.SendAlarmMsg(tips, 3)
  675. }
  676. }()
  677. if !utils.Rc.SetNX(cacheKey, 1, 2*time.Minute) {
  678. deleteCache = false
  679. err = fmt.Errorf("系统处理中,请稍后重试!")
  680. return
  681. }
  682. //查询配置,如果未开启自动设置禁止刷新,则无需处理
  683. obj := new(models.BusinessConf)
  684. conf, err := obj.GetItemByConfKey("EdbStopRefreshRule")
  685. if err != nil {
  686. if utils.IsErrNoRow(err) {
  687. utils.FileLog.Info("DisableEdbRefresh-未找到配置项,无需处理")
  688. return
  689. }
  690. return
  691. }
  692. if conf != nil && conf.Id <= 0 {
  693. utils.FileLog.Info("DisableEdbRefresh-未找到配置项,无需处理")
  694. return
  695. }
  696. //将json转为结构体
  697. rule := new(models.EdbStopRefreshRule)
  698. err = json.Unmarshal([]byte(conf.ConfVal), rule)
  699. if err != nil {
  700. return
  701. }
  702. //判断是否开启自动设置禁止刷新
  703. if rule.IsOpen == 0 {
  704. return
  705. }
  706. //获取当前时间
  707. now := time.Now()
  708. if rule.BaseIndexStopDays > 0 { //设置数据源钢联化工指标禁止更新
  709. baseIndexEndDate := now.AddDate(0, 0, -rule.BaseIndexStopDays+1).Format(utils.FormatDate)
  710. // 查询钢联化工指标,查询创建时间在baseIndexStartDate前,的所有钢联化工指标,分批查询,先查总数,再查列表
  711. totalCount, e := data_manage.GetCountRefreshBaseFromMysteelChemicalIndexItemByCreateTime(baseIndexEndDate)
  712. if e != nil {
  713. err = fmt.Errorf("查询钢联化工指标总数失败:%v", e)
  714. return
  715. }
  716. //分页查询
  717. pageSize := 100
  718. pageNum := (int(totalCount) + 99) / pageSize // 使用整数除法,并添加一页以防有余数
  719. stopRefreshIds := make([]int32, 0)
  720. for i := 0; i < pageNum; i++ {
  721. start := i * pageSize
  722. indexItems, e := data_manage.GetRefreshBaseFromMysteelChemicalIndexItemByCreateTime(baseIndexEndDate, start, pageSize)
  723. if e != nil {
  724. err = fmt.Errorf("分页查询钢联化工指标失败:%v", e)
  725. return
  726. }
  727. if len(indexItems) == 0 {
  728. continue
  729. }
  730. indexCodeList := make([]string, 0)
  731. for _, indexItem := range indexItems {
  732. indexCodeList = append(indexCodeList, indexItem.IndexCode)
  733. }
  734. condition := ` AND source=? AND edb_code in (` + utils.GetOrmInReplace(len(indexCodeList)) + `)`
  735. var pars []interface{}
  736. pars = append(pars, utils.DATA_SOURCE_MYSTEEL_CHEMICAL, indexCodeList)
  737. // 查询指标库里这些指标是否已创建
  738. edbList, e := data_manage.GetEdbInfoByCondition(condition, pars, 0)
  739. if e != nil {
  740. err = fmt.Errorf("查询指标库里这些指标是否被创建失败:%v", e)
  741. return
  742. }
  743. edbMap := make(map[string]bool)
  744. for _, edb := range edbList {
  745. edbMap[edb.EdbCode] = true
  746. }
  747. for _, indexItem := range indexItems {
  748. // 判断指标是否被创建
  749. if _, ok := edbMap[indexItem.IndexCode]; !ok {
  750. // 判断大写的指标编码是否存在
  751. if _, ok1 := edbMap[strings.ToLower(indexItem.IndexCode)]; !ok1 {
  752. stopRefreshIds = append(stopRefreshIds, indexItem.BaseFromMysteelChemicalIndexId)
  753. if len(stopRefreshIds) > 100 {
  754. err = data_manage.SetStopRefreshMysteelChemicalIndex(stopRefreshIds)
  755. if err != nil {
  756. err = fmt.Errorf("设置禁止刷新失败:%v", err)
  757. return
  758. }
  759. stopRefreshIds = make([]int32, 0)
  760. }
  761. }
  762. }
  763. }
  764. }
  765. // 未被创建,则设置禁止刷新
  766. if len(stopRefreshIds) > 0 {
  767. err = data_manage.SetStopRefreshMysteelChemicalIndex(stopRefreshIds)
  768. if err != nil {
  769. err = fmt.Errorf("设置禁止刷新失败:%v", err)
  770. return
  771. }
  772. }
  773. }
  774. if rule.EdbStopDays > 0 {
  775. // 查询钢联和wind、彭博、彭博财务、同花顺来源的指标
  776. edbEndDate := now.AddDate(0, 0, -rule.EdbStopDays+1).Format(utils.FormatDate)
  777. condition := ` AND no_update=0 AND source in (?,?,?,?,?) AND ((create_time < ? and set_update_time is null) or set_update_time < ? )`
  778. var pars []interface{}
  779. pars = append(pars, utils.DATA_SOURCE_MYSTEEL_CHEMICAL, utils.DATA_SOURCE_WIND, utils.DATA_SOURCE_PB_FINANCE, utils.DATA_SOURCE_PB, utils.DATA_SOURCE_THS, edbEndDate, edbEndDate)
  780. // 查询钢联化工指标和wind指标 分批查询,先查总数,再查列表
  781. totalCount, e := data_manage.GetEdbInfoCountByCondition(condition, pars)
  782. if e != nil {
  783. err = fmt.Errorf("查询钢联化工指标总数失败:%v", e)
  784. return
  785. }
  786. //分页查询
  787. pageSize := 100
  788. pageNum := (int(totalCount) + 99) / pageSize // 使用整数除法,并添加一页以防有余数
  789. stopRefreshIds := make([]int, 0)
  790. stopRefreshMysteelCode := make([]string, 0)
  791. fromEdbIdList := make([]int, 0)
  792. for i := 0; i < pageNum; i++ {
  793. start := i * pageSize
  794. edbItems, e := data_manage.GetEdbInfoPageByCondition(condition, pars, start, pageSize)
  795. if e != nil {
  796. err = fmt.Errorf("分页查询钢联化工指标失败:%v", e)
  797. return
  798. }
  799. if len(edbItems) == 0 {
  800. continue
  801. }
  802. edbInfoIds := make([]int, 0)
  803. fromEdbIdList = make([]int, 0)
  804. for _, item := range edbItems {
  805. edbInfoIds = append(edbInfoIds, item.EdbInfoId)
  806. }
  807. // 查询指标库里这些指标 引用情况
  808. relationList, e := data_manage.GetEdbInfoRelationByEdbInfoIds(edbInfoIds)
  809. if e != nil {
  810. err = fmt.Errorf("查询指标库里这些指标是否被创建失败:%v", e)
  811. return
  812. }
  813. edbMap := make(map[int]struct{})
  814. for _, item := range relationList {
  815. edbMap[item] = struct{}{}
  816. }
  817. for _, item := range edbItems {
  818. if _, ok := edbMap[item.EdbInfoId]; !ok {
  819. stopRefreshIds = append(stopRefreshIds, item.EdbInfoId)
  820. if item.Source == utils.DATA_SOURCE_MYSTEEL_CHEMICAL {
  821. stopRefreshMysteelCode = append(stopRefreshMysteelCode, item.EdbCode)
  822. }
  823. if item.EdbInfoType == 0 && item.EdbType == 1 {
  824. fromEdbIdList = append(fromEdbIdList, item.EdbInfoId)
  825. }
  826. // 更新指标禁止刷新状态
  827. if len(stopRefreshIds) > 100 {
  828. // 查询相关的计算指标
  829. calculateEdbIdList := make([]int, 0)
  830. if len(fromEdbIdList) > 0 {
  831. hasFind := make(map[int]struct{})
  832. calculateEdbIdList, err = GetCalculateEdbByFromEdbInfo(fromEdbIdList, calculateEdbIdList, hasFind)
  833. if err != nil {
  834. err = fmt.Errorf("查询计算指标信息失败:%v", err)
  835. return
  836. }
  837. }
  838. err = data_manage.ModifyEdbUpdateStatus(stopRefreshIds, stopRefreshMysteelCode, calculateEdbIdList)
  839. if err != nil {
  840. err = fmt.Errorf("更新指标禁止刷新状态失败:%v", err)
  841. return
  842. }
  843. stopRefreshIds = []int{}
  844. stopRefreshMysteelCode = []string{}
  845. }
  846. }
  847. }
  848. }
  849. // 更新指标禁止刷新状态
  850. if len(stopRefreshIds) > 0 {
  851. // 查询相关的计算指标
  852. calculateEdbIdList := make([]int, 0)
  853. if len(fromEdbIdList) > 0 {
  854. hasFind := make(map[int]struct{})
  855. calculateEdbIdList, err = GetCalculateEdbByFromEdbInfo(fromEdbIdList, calculateEdbIdList, hasFind)
  856. if err != nil {
  857. err = fmt.Errorf("查询计算指标信息失败:%v", err)
  858. return
  859. }
  860. }
  861. err = data_manage.ModifyEdbUpdateStatus(stopRefreshIds, stopRefreshMysteelCode, calculateEdbIdList)
  862. if err != nil {
  863. err = fmt.Errorf("更新指标禁止刷新状态失败:%v", err)
  864. return
  865. }
  866. }
  867. }
  868. return
  869. }