eta_data_analysis/services/task.go

package services

import (
	ccfService "eta/eta_data_analysis/services/base_from_ccf"
	"eta/eta_data_analysis/utils"
	"fmt"
	"github.com/beego/beego/v2/task"
	"github.com/patrickmn/go-cache"
	"github.com/robfig/cron/v3"
	"io/fs"
	"os"
	"path/filepath"
	"strings"
	"syscall"
	"time"
)

func Task() {
	fmt.Println("task start")
	if utils.YongyiOpen == "1" {
		// 每隔两分钟检测文件是否变动，如果发生变动，则自动解析数据到库里
		readWatchYongyiFile := task.NewTask("ReadWatchYongyiFile", "0 */2 * * * *", ReadWatchYongyiFile)
		task.AddTask("监听涌溢咨询文件夹并解析Excel", readWatchYongyiFile)

		yongyiDownloadDaily := task.NewTask("YongyiDownloadDaily", "0 */25 13-23 * * *", YongyiDownloadDaily)
		task.AddTask("涌益咨询日度指标下载", yongyiDownloadDaily)

		yongyiDownloadWeekly := task.NewTask("YongyiDownloadWeekly", "0 */30 16-23 * * 4,5,6,0", YongyiDownloadWeekyly)
		task.AddTask("涌益咨询周度指标下载", yongyiDownloadWeekly)
	}

	if utils.CoalMineOpen == "1" {
		c := cron.New(cron.WithSeconds())
		//每2分钟检测一次指标文件是否更新
		_, err := c.AddFunc("0 */2 * * * *", ReadWatchIndexFile)
		if err != nil {
			fmt.Println("watch.ReadWatchIndexFile err" + err.Error())
			utils.FileLog.Info("watch.ReadWatchIndexFile err" + err.Error())
		}
		c.Start()
	}

	// 汾渭数据
	if utils.FenweiOpen == "1" {
		// 每5分钟检测一次目录是否有新文件
		fenWeiReadWatchIndexFile := task.NewTask("fenWeiReadWatchIndexFile", "0 */5 * * * *", FenweiReadWatchIndexFile)
		task.AddTask("汾渭数据指标文件检测", fenWeiReadWatchIndexFile)

		// 汾渭网络数据处理 todo 上线放开
		/*fenWeiNetDataDeal := task.NewTask("fenWeiNetDataDeal", "0 0 1 * * *", fenwei.FenWeiNetDataDeal)
		task.AddTask("汾渭网络数据处理", fenWeiNetDataDeal)*/

	}

	if utils.MtjhOpen == "1" {
		c := cron.New(cron.WithSeconds())
		//每2分钟检测一次指标文件是否更新
		_, err := c.AddFunc("0 */2 * * * *", mtjhWatch)
		if err != nil {
			fmt.Println("watch.mtjhWatch err" + err.Error())
			utils.FileLog.Info("watch.mtjhWatch err" + err.Error())
		}
		c.Start()
	}

	if utils.CoalMailAttachmentOpen == "1" {
		mailAttachment := task.NewTask("MailAttachment", utils.CoalMailAttachmentTime, MailAttachment)
		task.AddTask("启动获取邮件附件脚本", mailAttachment)
	}

	// CCF化纤信息
	if utils.CCFOpen == "1" {
		// 原油石化早报-默认每日9:00
		ccfDailyTime := utils.CCFDailyTaskTime
		if ccfDailyTime == "" {
			ccfDailyTime = "0 0 9 * * *"
		}
		taskCCFOilDailyEdb := task.NewTask("taskCCFOilDailyEdb", ccfDailyTime, ccfService.TaskOilDailyEdb)
		task.AddTask("CCF原油石化早报", taskCCFOilDailyEdb)

		// 各品种周报-默认每周五17-19点每隔半小时
		ccfWeeklyTime := utils.CCFWeeklyTaskTime
		if ccfWeeklyTime == "" {
			ccfWeeklyTime = "0 */30 17-19 * * 5"
		}
		taskCCFWeeklyEdb := task.NewTask("taskCCFWeeklyEdb", ccfWeeklyTime, ccfService.TaskWeeklyEdb)
		task.AddTask("CCF周度指标", taskCCFWeeklyEdb)

		// 各品种装置-默认每周四15-18点每隔半小时
		ccfStockTime := utils.CCFStockTaskTime
		if ccfStockTime == "" {
			ccfStockTime = "0 */30 15-18 * * 4"
		}
		taskCCFStockTable := task.NewTask("taskCCFStockTable", ccfStockTime, ccfService.TaskStockTable)
		task.AddTask("CCF装置检修", taskCCFStockTable)
	}

	task.StartTask()

	fmt.Println("task end")
}

func ReadWatchIndexFile() {
	fmt.Println("ReadWatchIndexFile start")
	var err error
	defer func() {
		if err != nil {
			fmt.Println("ReadWatchIndexFile Err:" + err.Error())
		}
	}()
	var cacheClient *cache.Cache
	if cacheClient == nil {
		cacheClient = cache.New(365*24*time.Hour, 365*24*time.Hour)
	}
	err = filepath.Walk(utils.CoalMineFilePath, func(path string, info fs.FileInfo, err error) error {
		if err != nil {
			return err
		}
		if !info.IsDir() {
			fileInfo, err := os.Stat(path)
			if err != nil {
				fmt.Println("os.Stat:", err.Error())
			}
			winFileAttr := fileInfo.Sys().(*syscall.Win32FileAttributeData)
			modifyTimeStr := utils.SecondToTime(winFileAttr.LastWriteTime.Nanoseconds() / 1e9).Format(utils.FormatDateTime)

			existModifyTime, ok := cacheClient.Get(path)
			if ok {
				existModifyTimeStr := existModifyTime.(string)
				if existModifyTimeStr != modifyTimeStr {
					if strings.Contains(path, "442家晋陕蒙煤矿周度产量数据") {
						err = Jsm(path)
					} else if strings.Contains(path, "内陆17省动力煤终端用户供耗存") {
						err = Inland(path)
					} else if strings.Contains(path, "沿海八省动力煤终端用户供耗存数据更新") {
						err = Coastal(path)
					} else if strings.Contains(path, "442家晋陕蒙历史数据") {
						err = JsmHistory(path)
					} else if strings.Contains(path, "CⅢ-8-16 25省市库存和日耗情况") {
						err = CoastalHistory(path)
						time.Sleep(time.Second * 10)
						err = InlandHistory(path)
					} else if strings.Contains(path, "分企业煤炭产量旬度数据") {
						err = Firm(path)
					}
				}
			} else {
				if strings.Contains(path, "442家晋陕蒙煤矿周度产量数据") {
					err = Jsm(path)
				} else if strings.Contains(path, "内陆17省动力煤终端用户供耗存") {
					err = Inland(path)
				} else if strings.Contains(path, "沿海八省动力煤终端用户供耗存数据更新") {
					err = Coastal(path)
				} else if strings.Contains(path, "442家晋陕蒙历史数据") {
					err = JsmHistory(path)
				} else if strings.Contains(path, "CⅢ-8-16 25省市库存和日耗情况") {
					err = CoastalHistory(path)
					time.Sleep(time.Second * 10)
					err = InlandHistory(path)
				} else if strings.Contains(path, "分企业煤炭产量旬度数据") {
					err = Firm(path)
				}
			}
			cacheClient.Delete(path)
			cacheClient.Set(path, modifyTimeStr, 24*time.Hour)
		}
		return nil
	})
}