package services import ( ccfService "eta/eta_data_analysis/services/base_from_ccf" oilchemService "eta/eta_data_analysis/services/base_from_oilchem" "eta/eta_data_analysis/services/fenwei" "eta/eta_data_analysis/services/liangyou" "eta/eta_data_analysis/utils" "fmt" "io/fs" "os" "path/filepath" "strings" "syscall" "time" "github.com/beego/beego/v2/task" "github.com/patrickmn/go-cache" "github.com/robfig/cron/v3" ) 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) } // 汾渭网络数据 if utils.FenweiNetOpen == "1" { // 汾渭网络数据处理 每天16点和19点 执行爬取 fenWeiNetDataDeal := task.NewTask("fenWeiNetDataDeal", "0 0 16,19 * * *", fenwei.FenWeiNetDataDeal) task.AddTask("汾渭网络数据处理", fenWeiNetDataDeal) } if utils.LY_OPEN == "1" { lyData := task.NewTask("refreshData", "0 0 12 * * *", liangyou.LyDataDeal) // 粮油商务网 task.AddTask("粮油商务网", lyData) } 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) } // 隆众资讯 if utils.OilchemDataInit == "1" && utils.OilchemOpen == "1" { err := oilchemService.OilchemList(nil) if err != nil { utils.FileLog.Info("InitOilchemList Err:" + err.Error()) } } else if utils.OilchemOpen == "1" { oilchemData := task.NewTask("oilchemData", "0 0 16 * * *", oilchemService.OilchemList) task.AddTask("卓创资讯", oilchemData) } 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 }) }