eta_task/services/task.go

package services

import (
	"context"
	"eta_gn/eta_task/services/data"
	"eta_gn/eta_task/services/data/future_good"
	"eta_gn/eta_task/services/data_stat"
	"eta_gn/eta_task/utils"
	"fmt"
	"sync"
	"time"

	"github.com/beego/beego/v2/task"
)

func Task() {
	fmt.Println("task start")
	// 如果开启了需要走生产环境的配置，才需要走这些任务
	if utils.RunMode == "release" || utils.IsStartReleaseTask {
		releaseTask()
	}
	// 定时发布智能研报
	publishSmartReport := task.NewTask("publishSmartReport", "0 */1 * * * *", PublishSmartReport)
	task.AddTask("定时发布智能研报", publishSmartReport)

	// 定时发布研报
	publishReport := task.NewTask("publishReport", "0 */1 * * * *", PublishReport)
	task.AddTask("定时发布研报", publishReport)

	// 定时汇总数据源终端指标更新情况
	setEdbSourceStatTask := task.NewTask("setEdbSourceStatTask", "0 20 19,23 * * *", data_stat.SetEdbSourceStatTask)
	task.AddTask("数据源统计表", setEdbSourceStatTask)

	GnTask()

	// 测试-刷新同花顺高频, 上线前删除
	//if utils.RunMode == "debug" {
	//	refreshThsHfBase := task.NewTask("refreshThsHfBase", "0 30 12 * * *", data.RefreshBaseFromThsHfIndex)
	//	task.AddTask("refreshThsHfBase", refreshThsHfBase)
	//}

	task.StartTask()
	fmt.Println("task end")
}

// 生产环境需要走的任务
func releaseTask() {
	//同步指标
	if utils.BusinessCode != utils.BusinessCodeRelease {
		syncHzDataIndex := task.NewTask("syncHzDataIndex", "0 10,20,40,50 16,18 * * *", SyncHzDataIndex)
		task.AddTask("syncHzDataIndex", syncHzDataIndex)

		syncRankingFromDalian := task.NewTask("syncRankingFromDalian", "0 30,40,50 16,18 * * *", SyncRankingFromDalian)
		task.AddTask("syncRankingFromDalian", syncRankingFromDalian)
	}
	// 定时统计交易所的持仓分析数据
	initPositionTask := task.NewTask("initPositionTask", "0 20,40 16-19 * * *", data.InitPositionTask)
	task.AddTask("initPositionTask", initPositionTask)

	//刷新指标数据
	refreshData := task.NewTask("refreshData", "0 30 0,19 * * *", RefreshData)
	task.AddTask("refreshData", refreshData)

	// 根据配置刷新指标数据
	configRefreshData := task.NewTask("syncBaseDataExt", "0 */30 * * * * ", ConfigRefreshDataGn)
	task.AddTask("configRefreshData", configRefreshData)

	// 定时禁用钢联化工和wind指标的刷新状态
	//disableEdbRefresh := task.NewTask("disableEdbRefresh", "0 0 10 * * *", DisableEdbRefresh)
	//task.AddTask("disableEdbRefresh", disableEdbRefresh)

	//同步弘则数据库中来自，钢联，隆众，有色,人工等基础数据--每隔五分钟，同步一次最新数据
	//syncBaseData := task.NewTask("syncBaseData", "0 */5 * * * * ", SyncBaseData)
	//task.AddTask("syncBaseData", syncBaseData)

	syncBaseDataExt := task.NewTask("syncBaseDataExt", "0 */30 * * * * ", SyncBaseDataExt)
	task.AddTask("syncBaseDataExt", syncBaseDataExt)

	//初始化指标更新状态
	resetEdbInfoIsUpdate := task.NewTask("resetEdbInfoIsUpdate", "0 0 0 * * *", data.ResetEdbInfoIsUpdate)
	task.AddTask("resetEdbInfoIsUpdate", resetEdbInfoIsUpdate)

	// 刷新商品数据
	refreshFutureGoodData := task.NewTask("refreshFutureGoodData", "0 30 0,19 * * *", RefreshFutureGoodData)
	task.AddTask("refreshFutureGoodData", refreshFutureGoodData)

	//刷新交易所指标数据
	refreshTradeData := task.NewTask("refreshData", "0 1 4 * * *", RefreshTradeData)
	task.AddTask("refreshTradeData", refreshTradeData)

	//刷新欧洲天然气指标数据
	refreshEicData := task.NewTask("refreshData", "0 1 3,7 * * *", RefreshEicData)
	task.AddTask("refreshEicData", refreshEicData)

	//检测数据服务器
	checkDataServer := task.NewTask("checkDataServer", "0 */2 * * * * ", checkDataServer)
	task.AddTask("checkDataServer", checkDataServer)

	//删除日志 report_save_log，ppt_v2_save_log，保留一个月的
	deleteLog := task.NewTask("syncSubStatus", "0 0 2 2 * *", DeleteLog)
	task.AddTask("deleteLog", deleteLog)

	// 定时发布英文研报
	publishEnglishReport := task.NewTask("publishEnglishReport", "0 */1 * * * *", PublishEnglishReport)
	task.AddTask("定时发布英文研报", publishEnglishReport)

	// 每天清理两周前的报告保存日志
	clearReportSaveLog := task.NewTask("clearReportSaveLog", "0 15 23 * * *", ClearReportSaveLog)
	task.AddTask("定时清理报告保存日志", clearReportSaveLog)

	// 每天清理三个月前的用户操作日志
	clearAdminOperateLog := task.NewTask("clearAdminOperateLog", "0 20 23 * * *", ClearAdminOperateLog)
	task.AddTask("定时清理用户操作日志", clearAdminOperateLog)

	// 刷新同花顺高频
	refreshThsHfBase := task.NewTask("refreshThsHfBase", "0 0 0,6,9,12,15,18,21 * * *", data.RefreshBaseFromThsHfIndex)
	task.AddTask("refreshThsHfBase", refreshThsHfBase)
}

func RefreshData(cont context.Context) (err error) {
	wg := sync.WaitGroup{}

	//手工数据
	go func() {
		wg.Add(1)
		data.RefreshDataFromManual(&wg)
	}()

	// 持仓分析
	//go func() {
	//	wg.Add(1)
	//	_ = data.RefreshDataFromTradeAnalysis(&wg)
	//}()

	// 刷新未配置刷新时间点的来源(可能含未知的来源, 统一在这个时间刷新)
	go func() {
		wg.Add(1)
		data.NoneConfigRefreshDataGn(&wg)
	}()

	wg.Wait()
	////计算指标
	data.RefreshDataFromCalculateAll()

	// 刷新所有的基础预测指标
	data.RefreshBasePredictDataAll()

	// 预测计算指标
	data.RefreshPredictDataFromCalculateAll()

	// 指标系列计算数据
	_ = data.RefreshFactorEdbCalculateData()

	// 指标系列图表计算数据
	_ = data.RefreshFactorEdbChartCalculateData()

	time.Sleep(5 * time.Second)
	//data.RefreshNotice()

	fmt.Println("Refresh End")
	return
}

// SyncBaseData 刷新基础数据
func SyncBaseData(cont context.Context) (err error) {
	now := time.Now()
	if now.Hour() == 0 || now.Hour() == 19 {
		return nil
	}
	//同步钢联基础数据
	if utils.BusinessCode == utils.BusinessCodeRelease {
		go data.SyncGlDataBase()
	}
	return
}

// SyncBaseDataExt 刷新基础数据
func SyncBaseDataExt(cont context.Context) (err error) {
	now := time.Now()
	if now.Hour() == 0 || now.Hour() == 19 {
		return nil
	}
	//同步手工数据
	go data.SyncManualDataBase()
	//刷新图表中，指标的最新日期
	go data.SetChartEdbEndDate()
	return
}

// RefreshFutureGoodData 刷新商品数据
func RefreshFutureGoodData(cont context.Context) (err error) {
	future_good.RefreshFutureGoodDataFromThs()
	fmt.Println("Refresh End")
	return
}

// RefreshTradeData 刷新交易所数据
func RefreshTradeData(cont context.Context) (err error) {
	wg := sync.WaitGroup{}
	wg.Add(6)

	//郑商所
	go data.RefreshBaseDataFromSource(&wg, utils.DATA_SOURCE_ZZ)
	//上期所
	go data.RefreshBaseDataFromSource(&wg, utils.DATA_SOURCE_SH)
	//上期能源
	go data.RefreshBaseDataFromSource(&wg, utils.DATA_SOURCE_SHFE)
	//中金所
	go data.RefreshBaseDataFromSource(&wg, utils.DATA_SOURCE_CFFEX)
	//大商所
	go data.RefreshBaseDataFromSource(&wg, utils.DATA_SOURCE_DL)
	//广期所数据
	go data.RefreshBaseDataFromSource(&wg, utils.DATA_SOURCE_GFEX)

	wg.Wait()

	//计算指标
	data.RefreshDataFromCalculateAll()

	// 刷新所有的基础预测指标
	data.RefreshBasePredictDataAll()

	// 计算预测指标
	data.RefreshPredictDataFromCalculateAll()

	return
}

// RefreshEicData 刷新欧洲天然气数据
func RefreshEicData(cont context.Context) (err error) {
	wg := sync.WaitGroup{}
	wg.Add(1)

	//欧洲天然气
	go data.RefreshDataFromEic(&wg)
	wg.Wait()
	////计算指标
	//data.RefreshDataFromCalculateAll()
	//
	//// 刷新所有的基础预测指标
	//data.RefreshBasePredictDataAll()
	//
	//// 计算预测指标
	//data.RefreshPredictDataFromCalculateAll()

	return
}

// 检测数据服务
func checkDataServer(cont context.Context) (err error) {
	//检测wind新服务器
	go data.CheckWindDataInterface(cont)
	//检测同花顺数据服务器
	//go data.CheckThsDataInterface(cont)

	//检测路透数据服务器
	go data.CheckLtDataInterface(cont)
	//检测彭博
	go data.CheckPbDataInterface(cont)
	return
}

func TaskInit() {
	fmt.Println("TaskInit start")
	//data.InitGuangzhouPositionTask()
	fmt.Println("TaskInit end")
	return
}