task.go 6.8 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197
  1. package services
  2. import (
  3. ccfService "eta/eta_data_analysis/services/base_from_ccf"
  4. oilchemService "eta/eta_data_analysis/services/base_from_oilchem"
  5. "eta/eta_data_analysis/services/fenwei"
  6. "eta/eta_data_analysis/services/liangyou"
  7. "eta/eta_data_analysis/utils"
  8. "fmt"
  9. "io/fs"
  10. "path/filepath"
  11. "time"
  12. "github.com/beego/beego/v2/task"
  13. "github.com/patrickmn/go-cache"
  14. "github.com/robfig/cron/v3"
  15. )
  16. func Task() {
  17. fmt.Println("task start")
  18. if utils.YongyiOpen == "1" {
  19. // 每隔两分钟检测文件是否变动,如果发生变动,则自动解析数据到库里
  20. readWatchYongyiFile := task.NewTask("ReadWatchYongyiFile", "0 */2 * * * *", ReadWatchYongyiFile)
  21. task.AddTask("监听涌溢咨询文件夹并解析Excel", readWatchYongyiFile)
  22. yongyiDownloadDaily := task.NewTask("YongyiDownloadDaily", "0 */25 13-23 * * *", YongyiDownloadDaily)
  23. task.AddTask("涌益咨询日度指标下载", yongyiDownloadDaily)
  24. yongyiDownloadWeekly := task.NewTask("YongyiDownloadWeekly", "0 */30 16-23 * * 4,5,6,0", YongyiDownloadWeekyly)
  25. task.AddTask("涌益咨询周度指标下载", yongyiDownloadWeekly)
  26. }
  27. if utils.CoalMineOpen == "1" {
  28. c := cron.New(cron.WithSeconds())
  29. //每2分钟检测一次指标文件是否更新
  30. _, err := c.AddFunc("0 */2 * * * *", ReadWatchIndexFile)
  31. if err != nil {
  32. fmt.Println("watch.ReadWatchIndexFile err" + err.Error())
  33. utils.FileLog.Info("watch.ReadWatchIndexFile err" + err.Error())
  34. }
  35. c.Start()
  36. }
  37. // 汾渭数据
  38. if utils.FenweiOpen == "1" {
  39. // 每5分钟检测一次目录是否有新文件
  40. fenWeiReadWatchIndexFile := task.NewTask("fenWeiReadWatchIndexFile", "0 */5 * * * *", FenweiReadWatchIndexFile)
  41. task.AddTask("汾渭数据指标文件检测", fenWeiReadWatchIndexFile)
  42. }
  43. // 汾渭网络数据
  44. if utils.FenweiNetOpen == "1" {
  45. // 汾渭网络数据处理 每天16点和19点 执行爬取
  46. fenWeiNetDataDeal := task.NewTask("fenWeiNetDataDeal", "0 0 16,19 * * *", fenwei.FenWeiNetDataDeal)
  47. task.AddTask("汾渭网络数据处理", fenWeiNetDataDeal)
  48. }
  49. if utils.LY_OPEN == "1" {
  50. lyData := task.NewTask("refreshData", "0 0 12 * * *", liangyou.LyDataDeal) // 粮油商务网
  51. task.AddTask("粮油商务网", lyData)
  52. }
  53. if utils.MtjhOpen == "1" {
  54. c := cron.New(cron.WithSeconds())
  55. //每2分钟检测一次指标文件是否更新
  56. _, err := c.AddFunc("0 */2 * * * *", mtjhWatch)
  57. if err != nil {
  58. fmt.Println("watch.mtjhWatch err" + err.Error())
  59. utils.FileLog.Info("watch.mtjhWatch err" + err.Error())
  60. }
  61. c.Start()
  62. }
  63. if utils.CoalMailAttachmentOpen == "1" {
  64. mailAttachment := task.NewTask("MailAttachment", utils.CoalMailAttachmentTime, MailAttachment)
  65. task.AddTask("启动获取邮件附件脚本", mailAttachment)
  66. }
  67. // CCF化纤信息
  68. if utils.CCFOpen == "1" {
  69. // 原油石化早报-默认每日9:00
  70. ccfDailyTime := utils.CCFDailyTaskTime
  71. if ccfDailyTime == "" {
  72. ccfDailyTime = "0 0 9 * * *"
  73. }
  74. taskCCFOilDailyEdb := task.NewTask("taskCCFOilDailyEdb", ccfDailyTime, ccfService.TaskOilDailyEdb)
  75. task.AddTask("CCF原油石化早报", taskCCFOilDailyEdb)
  76. // 各品种周报-默认每周五17-19点每隔半小时
  77. ccfWeeklyTime := utils.CCFWeeklyTaskTime
  78. if ccfWeeklyTime == "" {
  79. ccfWeeklyTime = "0 */30 17-19 * * 5"
  80. }
  81. taskCCFWeeklyEdb := task.NewTask("taskCCFWeeklyEdb", ccfWeeklyTime, ccfService.TaskWeeklyEdb)
  82. task.AddTask("CCF周度指标", taskCCFWeeklyEdb)
  83. // 各品种装置-默认每周四15-18点每隔半小时
  84. ccfStockTime := utils.CCFStockTaskTime
  85. if ccfStockTime == "" {
  86. ccfStockTime = "0 */30 15-18 * * 4"
  87. }
  88. taskCCFStockTable := task.NewTask("taskCCFStockTable", ccfStockTime, ccfService.TaskStockTable)
  89. task.AddTask("CCF装置检修", taskCCFStockTable)
  90. // ccf图表数据-默认每天下午4点
  91. ccfChartTime := utils.CCFChartTaskTime
  92. if ccfChartTime == "" {
  93. ccfChartTime = "0 16 * * * *"
  94. }
  95. taskGetCCFChartEdb := task.NewTask("taskGetCCFChartEdb", ccfChartTime, ccfService.TaskGetCCFChartEdb)
  96. task.AddTask("CCF图表数据", taskGetCCFChartEdb)
  97. }
  98. // 隆众资讯
  99. if utils.OilchemDataInit == "1" && utils.OilchemOpen == "1" {
  100. err := oilchemService.OilchemList(nil)
  101. if err != nil {
  102. utils.FileLog.Info("InitOilchemList Err:" + err.Error())
  103. }
  104. } else if utils.OilchemOpen == "1" {
  105. oilchemData := task.NewTask("oilchemData", "0 0 16 * * *", oilchemService.OilchemList)
  106. task.AddTask("卓创资讯", oilchemData)
  107. }
  108. task.StartTask()
  109. fmt.Println("task end")
  110. }
  111. func ReadWatchIndexFile() {
  112. fmt.Println("ReadWatchIndexFile start")
  113. var err error
  114. defer func() {
  115. if err != nil {
  116. fmt.Println("ReadWatchIndexFile Err:" + err.Error())
  117. }
  118. }()
  119. var cacheClient *cache.Cache
  120. if cacheClient == nil {
  121. cacheClient = cache.New(365*24*time.Hour, 365*24*time.Hour)
  122. }
  123. err = filepath.Walk(utils.CoalMineFilePath, func(path string, info fs.FileInfo, err error) error {
  124. if err != nil {
  125. return err
  126. }
  127. if !info.IsDir() {
  128. /* fileInfo, err := os.Stat(path)
  129. if err != nil {
  130. fmt.Println("os.Stat:", err.Error())
  131. }
  132. winFileAttr := fileInfo.Sys().(*syscall.Win32FileAttributeData)
  133. modifyTimeStr := utils.SecondToTime(winFileAttr.LastWriteTime.Nanoseconds() / 1e9).Format(utils.FormatDateTime)
  134. existModifyTime, ok := cacheClient.Get(path)
  135. if ok {
  136. existModifyTimeStr := existModifyTime.(string)
  137. if existModifyTimeStr != modifyTimeStr {
  138. if strings.Contains(path, "442家晋陕蒙煤矿周度产量数据") {
  139. err = Jsm(path)
  140. } else if strings.Contains(path, "内陆17省动力煤终端用户供耗存") {
  141. err = Inland(path)
  142. } else if strings.Contains(path, "沿海八省动力煤终端用户供耗存数据更新") {
  143. err = Coastal(path)
  144. } else if strings.Contains(path, "442家晋陕蒙历史数据") {
  145. err = JsmHistory(path)
  146. } else if strings.Contains(path, "CⅢ-8-16 25省市库存和日耗情况") {
  147. err = CoastalHistory(path)
  148. time.Sleep(time.Second * 10)
  149. err = InlandHistory(path)
  150. } else if strings.Contains(path, "分企业煤炭产量旬度数据") {
  151. err = Firm(path)
  152. }
  153. }
  154. } else {
  155. if strings.Contains(path, "442家晋陕蒙煤矿周度产量数据") {
  156. err = Jsm(path)
  157. } else if strings.Contains(path, "内陆17省动力煤终端用户供耗存") {
  158. err = Inland(path)
  159. } else if strings.Contains(path, "沿海八省动力煤终端用户供耗存数据更新") {
  160. err = Coastal(path)
  161. } else if strings.Contains(path, "442家晋陕蒙历史数据") {
  162. err = JsmHistory(path)
  163. } else if strings.Contains(path, "CⅢ-8-16 25省市库存和日耗情况") {
  164. err = CoastalHistory(path)
  165. time.Sleep(time.Second * 10)
  166. err = InlandHistory(path)
  167. } else if strings.Contains(path, "分企业煤炭产量旬度数据") {
  168. err = Firm(path)
  169. }
  170. }
  171. cacheClient.Delete(path)
  172. cacheClient.Set(path, modifyTimeStr, 24*time.Hour)*/
  173. }
  174. return nil
  175. })
  176. }