eta_task/services/edb_refresh.go

package services

import (
	"context"
	"eta_gn/eta_task/models/data_manage/edb_refresh"
	"eta_gn/eta_task/services/alarm_msg"
	"eta_gn/eta_task/services/data"
	"eta_gn/eta_task/utils"
	"fmt"
	"strings"
	"sync"
	"time"
)

func mergeMaps(dst map[string][]*edb_refresh.EdbInfoListAndRefreshConfig, src map[string][]*edb_refresh.EdbInfoListAndRefreshConfig) (newMap map[string][]*edb_refresh.EdbInfoListAndRefreshConfig) {
	if dst == nil {
		return src
	}
	if src == nil {
		return dst
	}
	newMap = dst
	for k, v := range src {
		if newK, ok := newMap[k]; ok {
			newK = append(newK, v...)
			newMap[k] = newK
		} else {
			newMap[k] = v
		}
	}

	return newMap
}

func getDefaultRefreshData(now time.Time) (sourceEdbInfoListMap map[string][]*edb_refresh.EdbInfoListAndRefreshConfig, err error) {
	errMsgList := make([]string, 0)
	defer func() {
		if err != nil {
			fmt.Println(err)
		}
	}()

	sourceEdbInfoListMap = make(map[string][]*edb_refresh.EdbInfoListAndRefreshConfig)

	currTimeStr := getPreviousHalfHour(now)
	fmt.Println(currTimeStr)

	list := make([]*edb_refresh.EdbRefreshDefaultConfig, 0)

	refreshFrequencyList := []string{"每自然日", "每交易日", "每周", "每旬", "每月", "每季", "每半年", "每年"}


	for _, refreshFrequency := range refreshFrequencyList {
		condition, pars, isHandler := getRefreshFrequencyCondition(now, refreshFrequency)
		if !isHandler {
			continue
		}

		condition += ` AND refresh_frequency = ? AND refresh_time = ?`
		pars = append(pars, refreshFrequency, currTimeStr)

		tmpList, tmpErr := edb_refresh.GetListByCondition(condition, pars)
		if tmpErr != nil {
			err = tmpErr
			return
		}
		list = append(list, tmpList...)
	}

	refreshDataNumMap := make(map[string]*edb_refresh.EdbRefreshDefaultConfig)
	refreshDataFrequencyListMap := make(map[int]map[int][]string)

	wgNum := 0
	for _, item := range list {
		key := fmt.Sprintf("%d_%d_%s", item.Source, item.SubSource, item.Frequency)
		refreshDataNumMap[key] = item

		subSourceFrequencyList, ok := refreshDataFrequencyListMap[item.Source]
		if !ok {
			subSourceFrequencyList = make(map[int][]string)
		}
		frequencyList, ok := subSourceFrequencyList[item.SubSource]
		if !ok {
			wgNum++
			frequencyList = make([]string, 0)
		}
		subSourceFrequencyList[item.SubSource] = append(frequencyList, item.Frequency)
		refreshDataFrequencyListMap[item.Source] = subSourceFrequencyList
	}

	for source, subSourceFrequencyListMap := range refreshDataFrequencyListMap {
		for subSource, frequencyList := range subSourceFrequencyListMap {
			edbList, tmpErr := edb_refresh.GetDefaultRefreshEdbInfoListBySourceAndSubSource(source, subSource, frequencyList)
			if tmpErr != nil {
				errMsgList = append(errMsgList, fmt.Sprint("source:", source, "subSource:", subSource, "frequencyList:", strings.Join(frequencyList, ","), "err:", tmpErr.Error()))
			}

			for _, v := range edbList {
				dataRefreshNum := utils.DATA_REFRESH
				key := fmt.Sprintf("%d_%d_%s", v.Source, v.SubSource, v.Frequency)
				if edbRefreshDefaultConfig, ok := refreshDataNumMap[key]; ok {
					if edbRefreshDefaultConfig.RefreshAllData == 1 { // 刷新所有数据期数
						dataRefreshNum = 0
					} else if edbRefreshDefaultConfig.RefreshDataNum > 0 { //
						dataRefreshNum = edbRefreshDefaultConfig.RefreshDataNum
					}
				}
				v.DataRefreshNum = dataRefreshNum
			}

			key := fmt.Sprint(source, "_", subSource)
			sourceEdbInfoListMap[key] = edbList
		}
	}

	fmt.Println("Get Refresh End")
	return
}

func getConfigRefreshData(now time.Time) (sourceEdbInfoListMap map[string][]*edb_refresh.EdbInfoListAndRefreshConfig, err error) {
	defer func() {
		if err != nil {
			fmt.Println(err)
		}
	}()

	sourceEdbInfoListMap = make(map[string][]*edb_refresh.EdbInfoListAndRefreshConfig)

	currTimeStr := getPreviousHalfHour(now)

	list := make([]*edb_refresh.EdbRefreshConfig, 0)

	refreshFrequencyList := []string{"每自然日", "每交易日", "每周", "每旬", "每月", "每季", "每半年", "每年"}

	for _, refreshFrequency := range refreshFrequencyList {
		condition, pars, isHandler := getRefreshFrequencyCondition(now, refreshFrequency)
		if !isHandler {
			continue
		}

		condition += ` AND refresh_frequency = ? AND refresh_time = ?`
		pars = append(pars, refreshFrequency, currTimeStr)

		tmpList, tmpErr := edb_refresh.GetEdbRefreshConfigListByCondition(condition, pars)
		if tmpErr != nil {
			err = tmpErr
			return
		}
		list = append(list, tmpList...)
	}

	configIdEdbRefreshConfigMap := make(map[int]*edb_refresh.EdbRefreshConfig)
	configIdList := make([]int, 0)
	for _, v := range list {
		configIdList = append(configIdList, v.EdbRefreshConfigId)
		configIdEdbRefreshConfigMap[v.EdbRefreshConfigId] = v
	}

	var sourceList []int
	edbInfoList, err := edb_refresh.GetConfigRefreshEdbInfoListBySourceAndSubSource(sourceList, configIdList)
	if err != nil {
		return
	}

	for _, v := range edbInfoList {
		key := fmt.Sprint(v.Source, "_", v.SubSource)
		tmpList, ok := sourceEdbInfoListMap[key]
		if !ok {
			tmpList = make([]*edb_refresh.EdbInfoListAndRefreshConfig, 0)
		}
		dataRefreshNum := utils.DATA_REFRESH
		if edbRefreshConfig, ok2 := configIdEdbRefreshConfigMap[v.EdbRefreshConfigId]; ok2 {
			if edbRefreshConfig.RefreshAllData == 1 { // 刷新所有数据期数
				dataRefreshNum = 0
			} else if edbRefreshConfig.RefreshDataNum > 0 { //
				dataRefreshNum = edbRefreshConfig.RefreshDataNum
			}
		}
		v.DataRefreshNum = dataRefreshNum
		sourceEdbInfoListMap[key] = append(tmpList, v)
	}

	fmt.Println("Get ConfigRefreshData End")
	return
}

func BaseRefreshData(wg *sync.WaitGroup, source, subSource int, items []*edb_refresh.EdbInfoListAndRefreshConfig) (err error) {
	errMsgList := make([]string, 0)
	defer func() {
		if err != nil {
			fmt.Println("来源：", source, "；子来源：", subSource, "；BaseRefreshData Err:"+err.Error())
			go alarm_msg.SendAlarmMsg(fmt.Sprint("来源：", source, "；子来源：", subSource, "；BaseRefreshData ErrMsg:"+err.Error()), 3)
		}
		if len(errMsgList) > 0 {
			errMsg := fmt.Sprint("来源：", source, "；子来源：", subSource, "；BaseRefreshData Err:"+strings.Join(errMsgList, "\n"))
			fmt.Println(errMsg)
			go alarm_msg.SendAlarmMsg(errMsg, 3)
		}
		wg.Done()
	}()

	dataRefreshNum := utils.DATA_REFRESH

	for _, v := range items {
		if v.NoUpdate == 1 {
			continue
		}
		if v.DataRefreshNum > 0 {
			dataRefreshNum = v.DataRefreshNum
		}

		startDate := ""
		if v.EndDate.IsZero() {
			startDate = utils.BaseEdbRefreshStartDate
		} else {
			if v.Frequency == "日度" {
				startDate = v.EndDate.AddDate(0, 0, -dataRefreshNum).Format(utils.FormatDate)
			} else if v.Frequency == "周度" {
				startDate = v.EndDate.AddDate(0, 0, -(dataRefreshNum * 7)).Format(utils.FormatDate)
			} else if v.Frequency == "旬度" {
				startDate = v.EndDate.AddDate(0, 0, -(dataRefreshNum * 10)).Format(utils.FormatDate)
			} else if v.Frequency == "月度" {
				startDate = v.EndDate.AddDate(0, -dataRefreshNum, 0).Format(utils.FormatDate)
			} else if v.Frequency == "季度" {
				startDate = v.EndDate.AddDate(0, -dataRefreshNum*3, 0).Format(utils.FormatDate)
			} else if v.Frequency == "半年度" {
				startDate = v.EndDate.AddDate(0, -dataRefreshNum*6, 0).Format(utils.FormatDate)
			} else if v.Frequency == "年度" {
				startDate = v.EndDate.AddDate(-dataRefreshNum, 0, 0).Format(utils.FormatDate)
			} else {
				startDate = v.EndDate.AddDate(0, 0, -utils.DATA_REFRESH).Format(utils.FormatDate)
			}
		}
		fmt.Println(startDate)

		resp, tmpErr := data.RefreshEdbData(v.EdbInfoId, v.Source, v.SubSource, v.EdbCode, startDate)
		if tmpErr != nil {
			errMsgList = append(errMsgList, v.EdbCode+"RefreshEdbData Err:"+tmpErr.Error())
			continue
		}
		if resp.Ret != 200 {
			errMsgList = append(errMsgList, v.EdbCode+";RefreshEdbData Err:"+resp.Msg+";ErrMsg:"+resp.ErrMsg)
			continue
		}
	}

	fmt.Println("来源：", source, ";子来源：", subSource, "刷新结束")
	return err
}

func getRefreshFrequencyCondition(now time.Time, refreshFrequency string) (condition string, pars []interface{}, isHandler bool) {
	isHandler = true

	var dayNum int
	var isLastDay bool

	switch refreshFrequency {
	case "每自然日":
		return
	case "每交易日":
		if now.Weekday() == time.Saturday || now.Weekday() == time.Sunday {
			isHandler = false
		}
		return
	case "每周":
		currWeekDay := now.Weekday()
		if currWeekDay == time.Sunday {
			currWeekDay = 7
			isLastDay = true
		}
		dayNum = int(currWeekDay)
	case "每旬":
		currDay := now.Day()
		if currDay <= 10 {
			dayNum = currDay
			if currDay == 10 {
				isLastDay = true
			}
		} else if currDay <= 20 {
			dayNum = currDay - 10
			if currDay == 20 {
				isLastDay = true
			}
		} else {
			dayNum = currDay - 20

			monthLastDay := time.Date(now.Year(), now.Month()+1, 1, 0, 0, 0, 0, time.Local).AddDate(0, 0, -1)
			if currDay == monthLastDay.Day() {
				isLastDay = true
			}
		}
	case "每月":
		currDay := now.Day()
		dayNum = currDay

		monthLastDay := time.Date(now.Year(), now.Month()+1, 1, 0, 0, 0, 0, time.Local).AddDate(0, 0, -1)

		if currDay == monthLastDay.Day() {
			isLastDay = true
		}
	case "每季":
		var startDay, endDay time.Time
		currMonth := now.Month()
		currDay := now.Day()
		if currMonth <= 3 {
			startDay = time.Date(now.Year(), 1, 1, 0, 0, 0, 0, time.Local)
			endDay = time.Date(now.Year(), 4, 1, 0, 0, 0, 0, time.Local).AddDate(0, 0, -1)
		} else if currMonth <= 6 {
			startDay = time.Date(now.Year(), 4, 1, 0, 0, 0, 0, time.Local)
			endDay = time.Date(now.Year(), 7, 1, 0, 0, 0, 0, time.Local).AddDate(0, 0, -1)
		} else if currMonth <= 9 {
			startDay = time.Date(now.Year(), 7, 1, 0, 0, 0, 0, time.Local)
			endDay = time.Date(now.Year(), 10, 1, 0, 0, 0, 0, time.Local).AddDate(0, 0, -1)
		} else {
			startDay = time.Date(now.Year(), 10, 1, 0, 0, 0, 0, time.Local)
			endDay = time.Date(now.Year()+1, 1, 1, 0, 0, 0, 0, time.Local).AddDate(0, 0, -1)
		}

		dayNum = utils.GetTimeSubDay(startDay, now) + 1

		if currMonth == endDay.Month() && currDay == endDay.Day() {
			isLastDay = true
		}
	case "每半年":
		var startDay, endDay time.Time
		currMonth := now.Month()
		currDay := now.Day()
		if currMonth <= 6 {
			startDay = time.Date(now.Year(), 1, 1, 0, 0, 0, 0, time.Local)
			endDay = time.Date(now.Year(), 7, 1, 0, 0, 0, 0, time.Local).AddDate(0, 0, -1)
		} else {
			startDay = time.Date(now.Year(), 7, 1, 0, 0, 0, 0, time.Local)
			endDay = time.Date(now.Year()+1, 1, 1, 0, 0, 0, 0, time.Local).AddDate(0, 0, -1)
		}

		dayNum = utils.GetTimeSubDay(startDay, now) + 1

		if currMonth == endDay.Month() && currDay == endDay.Day() {
			isLastDay = true
		}
	case "每年":
		currMonth := now.Month()
		currDay := now.Day()

		startDay := time.Date(now.Year(), 1, 1, 0, 0, 0, 0, time.Local)
		endDay := time.Date(now.Year()+1, 1, 1, 0, 0, 0, 0, time.Local).AddDate(0, 0, -1)

		dayNum = utils.GetTimeSubDay(startDay, now) + 1

		if currMonth == endDay.Month() && currDay == endDay.Day() {
			isLastDay = true
		}

	}

	if isLastDay {
		condition += ` AND ( refresh_frequency_day = ? OR refresh_frequency_day >= ? )`
		pars = append(pars, 0, dayNum)
	} else {
		condition += ` AND refresh_frequency_day = ?  `
		pars = append(pars, dayNum)
	}

	return
}

func getPreviousHalfHour(now time.Time) string {
	minute := now.Minute()

	if minute >= 30 {
		return fmt.Sprintf("%02d:%02d", now.Hour(), 30)
	}
	return fmt.Sprintf("%02d:%02d", now.Hour(), 0)
}

func ConfigRefreshDataGn(cont context.Context) (err error) {
	errMsgList := make([]string, 0)
	defer func() {
		if err != nil {
			fmt.Println(err)
		}
	}()

	now := time.Now()
	defaultSourceEdbInfoListMap, err := getDefaultRefreshData(now)
	if err != nil {
		errMsgList = append(errMsgList, "获取默认刷新数据失败，Err:"+err.Error())
	}
	sourceEdbInfoListMap, err := getConfigRefreshData(now)
	if err != nil {
		errMsgList = append(errMsgList, "获取指标配置刷新数据失败，Err:"+err.Error())
	}

	allSourceEdbInfoListMap := mergeMaps(defaultSourceEdbInfoListMap, sourceEdbInfoListMap)
	wgNum := len(allSourceEdbInfoListMap)
	if wgNum <= 0 {
		return
	}
	wg := sync.WaitGroup{}
	wg.Add(wgNum)

	for _, edbList := range allSourceEdbInfoListMap {
		if edbList == nil {
			wg.Done()
			continue
		}
		if len(edbList) != 0 {
			go BaseRefreshData(&wg, edbList[0].Source, edbList[0].SubSource, edbList)
		}
	}

	wg.Wait()

	fmt.Println("Refresh End")
	return
}