task.go 4.5 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135
  1. package services
  2. import (
  3. "eta/eta_data_analysis/utils"
  4. "fmt"
  5. "github.com/beego/beego/v2/task"
  6. "github.com/patrickmn/go-cache"
  7. "github.com/robfig/cron/v3"
  8. "io/fs"
  9. "os"
  10. "path/filepath"
  11. "strings"
  12. "syscall"
  13. "time"
  14. )
  15. func Task() {
  16. fmt.Println("task start")
  17. if utils.YongyiOpen == "1" {
  18. handleYongyiExcelDaily := task.NewTask("handleYongyiExcelDaily", "0 35 13,23 * * *", HandleYongyiExcelDaily)
  19. task.AddTask("涌益咨询日度指标处理", handleYongyiExcelDaily)
  20. handleYongyiExcelWeekly := task.NewTask("handleYongyiExcelWeekly", "0 5 17,23 * * *", HandleYongyiExcelWeekly)
  21. task.AddTask("涌益咨询周度指标处理", handleYongyiExcelWeekly)
  22. //HandleYongyiExcelDaily("/Users/xiexiaoyuan/Downloads/2023年11月21日涌益咨询日度数据 (1).xlsx")
  23. //HandleYongyiExcelWeekly()
  24. yongyiDownloadDaily := task.NewTask("YongyiDownloadDaily", "0 30 13,23 * * *", YongyiDownloadDaily)
  25. task.AddTask("涌益咨询日度指标下载", yongyiDownloadDaily)
  26. yongyiDownloadWeekly := task.NewTask("YongyiDownloadWeekly", "0 0 17,23 * * *", YongyiDownloadWeeyly)
  27. task.AddTask("涌益咨询周度指标下载", yongyiDownloadWeekly)
  28. }
  29. if utils.CoalMineOpen == "1" {
  30. c := cron.New(cron.WithSeconds())
  31. //每2分钟检测一次指标文件是否更新
  32. _, err := c.AddFunc("0 */2 * * * *", ReadWatchIndexFile)
  33. if err != nil {
  34. fmt.Println("watch.ReadWatchIndexFile err" + err.Error())
  35. utils.FileLog.Info("watch.ReadWatchIndexFile err" + err.Error())
  36. }
  37. c.Start()
  38. }
  39. // 汾渭数据
  40. if utils.FenweiOpen == "1" {
  41. // 每5分钟检测一次目录是否有新文件
  42. fenWeiReadWatchIndexFile := task.NewTask("fenWeiReadWatchIndexFile", "0 */5 * * * *", FenweiReadWatchIndexFile)
  43. task.AddTask("汾渭数据指标文件检测", fenWeiReadWatchIndexFile)
  44. }
  45. if utils.MtjhOpen == "1" {
  46. c := cron.New(cron.WithSeconds())
  47. //每2分钟检测一次指标文件是否更新
  48. _, err := c.AddFunc("0 */2 * * * *", mtjhWatch)
  49. if err != nil {
  50. fmt.Println("watch.mtjhWatch err" + err.Error())
  51. utils.FileLog.Info("watch.mtjhWatch err" + err.Error())
  52. }
  53. c.Start()
  54. }
  55. task.StartTask()
  56. fmt.Println("task end")
  57. }
  58. func ReadWatchIndexFile() {
  59. fmt.Println("ReadWatchIndexFile start")
  60. var err error
  61. defer func() {
  62. if err != nil {
  63. fmt.Println("ReadWatchIndexFile Err:" + err.Error())
  64. }
  65. }()
  66. var cacheClient *cache.Cache
  67. if cacheClient == nil {
  68. cacheClient = cache.New(365*24*time.Hour, 365*24*time.Hour)
  69. }
  70. err = filepath.Walk(utils.CoalMineFilePath, func(path string, info fs.FileInfo, err error) error {
  71. if err != nil {
  72. return err
  73. }
  74. if !info.IsDir() {
  75. fileInfo, err := os.Stat(path)
  76. if err != nil {
  77. fmt.Println("os.Stat:", err.Error())
  78. }
  79. winFileAttr := fileInfo.Sys().(*syscall.Win32FileAttributeData)
  80. modifyTimeStr := utils.SecondToTime(winFileAttr.LastWriteTime.Nanoseconds() / 1e9).Format(utils.FormatDateTime)
  81. existModifyTime, ok := cacheClient.Get(path)
  82. if ok {
  83. existModifyTimeStr := existModifyTime.(string)
  84. if existModifyTimeStr != modifyTimeStr {
  85. if strings.Contains(path, "442家晋陕蒙煤矿周度产量数据") {
  86. err = Jsm(path)
  87. } else if strings.Contains(path, "内陆17省动力煤终端用户供耗存") {
  88. err = Inland(path)
  89. } else if strings.Contains(path, "沿海八省动力煤终端用户供耗存数据更新") {
  90. err = Coastal(path)
  91. } else if strings.Contains(path, "442家晋陕蒙历史数据") {
  92. err = JsmHistory(path)
  93. } else if strings.Contains(path, "CⅢ-8-16 25省市库存和日耗情况") {
  94. err = CoastalHistory(path)
  95. time.Sleep(time.Second * 10)
  96. err = InlandHistory(path)
  97. } else if strings.Contains(path, "分企业煤炭产量旬度数据") {
  98. err = Firm(path)
  99. }
  100. }
  101. } else {
  102. if strings.Contains(path, "442家晋陕蒙煤矿周度产量数据") {
  103. err = Jsm(path)
  104. } else if strings.Contains(path, "内陆17省动力煤终端用户供耗存") {
  105. err = Inland(path)
  106. } else if strings.Contains(path, "沿海八省动力煤终端用户供耗存数据更新") {
  107. err = Coastal(path)
  108. } else if strings.Contains(path, "442家晋陕蒙历史数据") {
  109. err = JsmHistory(path)
  110. } else if strings.Contains(path, "CⅢ-8-16 25省市库存和日耗情况") {
  111. err = CoastalHistory(path)
  112. time.Sleep(time.Second * 10)
  113. err = InlandHistory(path)
  114. } else if strings.Contains(path, "分企业煤炭产量旬度数据") {
  115. err = Firm(path)
  116. }
  117. }
  118. cacheClient.Delete(path)
  119. cacheClient.Set(path, modifyTimeStr, 24*time.Hour)
  120. }
  121. return nil
  122. })
  123. }