eta_index_lib/services/base_from_usda_fas.go

package services

import (
	"eta/eta_index_lib/logic"
	"eta/eta_index_lib/models"
	"eta/eta_index_lib/services/alarm_msg"
	"eta/eta_index_lib/utils"
	"fmt"
	"strconv"
	"strings"
	"time"
)

// HandleUsdaFasIndex 处理美国农业部的excel数据
func HandleUsdaFasIndex(req *models.HandleUsdaFasExcelDataReq) (err error) {
	errMsgList := make([]string, 0)
	defer func() {
		if len(errMsgList) > 0 {
			msg := fmt.Sprint("数据源-美国农业部数据处理失败,err:", strings.Join(errMsgList, "\n"))
			utils.FileLog.Info(msg)
			go alarm_msg.SendAlarmMsg(msg, 3)
		}
	}()
	// 查询所有的一级分类
	classifyObj := new(models.BaseFromUsdaFasClassify)
	classifyList, err := classifyObj.GetParentClassify()
	if err != nil {
		err = fmt.Errorf("查询一级目录信息失败 Err：%s", err)
		return
	}
	classifyMap := make(map[string]int, 0)
	for _, v := range classifyList {
		classifyMap[v.ClassifyName] = int(v.ClassifyId)
	}

	for _, v := range req.List {
		if v.IndexName == "" || v.IndexCode == "" {
			errMsgList = append(errMsgList, fmt.Sprintf("新增指标异常，指标编码%s或者指标ID%s为空：", v.IndexCode, v.IndexName))
			continue
		}
		err = handleUsdaFasIndex(v, req.TerminalCode, classifyMap)
		if err != nil {
			errMsgList = append(errMsgList, fmt.Sprintf("新增指标异常，指标编码：%s, Err: %s", v.IndexCode, err))
			return
		}
	}
	return
}

func handleUsdaFasIndex(req *models.HandleUsdaFasExcelData, terminalCode string, classifyMap map[string]int) (err error) {
	indexName := req.IndexName
	indexCode := req.IndexCode
	excelDataMap := req.ExcelDataMap
	errMsgList := make([]string, 0)
	defer func() {
		if len(errMsgList) > 0 {
			msg := fmt.Sprint("数据源-美国农业部数据处理失败,err:", strings.Join(errMsgList, "\n"))
			utils.FileLog.Info(msg)
			go alarm_msg.SendAlarmMsg(msg, 3)
		}
	}()
	indexObj := new(models.BaseFromUsdaFasIndex)
	dataObj := new(models.BaseFromUsdaFasData)
	classifyObj := new(models.BaseFromUsdaFasClassify)

	var indexId int64

	addDataList := make([]*models.BaseFromUsdaFasData, 0)

	exitDataMap := make(map[string]*models.BaseFromUsdaFasData)

	// 修改指标信息
	if indexName == "" {
		utils.FileLog.Info("未刷新到指标数据:indexName:" + indexName)
		return
	}
	// 判断目录是否存在
	var classifyId int64
	now := time.Now()
	if req.ClassifyName != "" {
		classifyParentId := 0
		level := 1
		classifyParentId, _ = classifyMap[req.ParentClassifyName]
		if classifyParentId > 0 {
			level = 2
		}
		classifyObj, err = classifyObj.GetByClassifyName(req.ClassifyName)
		if err != nil {
			if err.Error() == utils.ErrNoRow() {
				//新增分类
				classifyObj = &models.BaseFromUsdaFasClassify{
					ClassifyName:    req.ClassifyName,
					ClassifyNameEn:  req.ClassifyName,
					ParentId:        classifyParentId,
					SysUserId:       0,
					SysUserRealName: "",
					Level:           level,
					Sort:            req.ClassifySort,
					ModifyTime:      now,
					CreateTime:      now,
				}

				classifyId, err = classifyObj.Add()
				if err != nil {
					err = fmt.Errorf("新增分类失败 Err：%s", err)
					return
				}
				classifyObj.ClassifyId = classifyId
			} else {
				return
			}
		} else {
			classifyId = classifyObj.ClassifyId
			classifyObj.ModifyTime = now
			//classifyObj.Sort = req.ClassifySort
			classifyObj.ParentId = classifyParentId
			//e := classifyObj.Update([]string{"ParentId", "Sort", "ModifyTime"})
			e := classifyObj.Update([]string{"ParentId", "ModifyTime"})
			if e != nil {
				fmt.Println("classifyObj Update Err:" + e.Error())
				return
			}
		}
	}

	//判断指标是否存在
	var isAdd int
	item, err := indexObj.GetByIndexCode(indexCode)
	if err != nil {
		if err.Error() == utils.ErrNoRow() {
			isAdd = 1
			err = nil
		} else {
			isAdd = -1
			err = fmt.Errorf("查询数据源指标库失败 GetByIndexCode Err：%s", err)
			return
		}
	}
	oldIndexObj := item
	if item != nil && item.BaseFromUsdaFasIndexId > 0 {
		fmt.Println("item:", item)
		isAdd = 2
	} else {
		isAdd = 1
	}

	if isAdd == 1 {
		indexObj.IndexCode = indexCode
		indexObj.IndexName = indexName
		indexObj.Frequency = req.Frequency
		indexObj.ClassifyId = classifyId
		indexObj.Country = req.Country
		indexObj.Commodity = req.Commodity
		indexObj.Unit = req.Unit
		indexObj.Sort = req.Sort
		indexObj.ModifyTime = time.Now()
		indexObj.CreateTime = time.Now()
		indexObj.TerminalCode = terminalCode
		indexId, err = indexObj.Add()
		if err != nil {
			err = fmt.Errorf("数据源新增美国农业部指标失败 Err：%s", err)
			return
		}
		indexObj.BaseFromUsdaFasIndexId = indexId
	} else if isAdd == 2 {
		indexId = item.BaseFromUsdaFasIndexId
		if item.TerminalCode == `` && terminalCode != `` {
			item.TerminalCode = terminalCode
			err = item.Update([]string{"TerminalCode"})
			if err != nil {
				err = fmt.Errorf("数据源更新美国农业部指标失败 Err：%s", err)
				return
			}
		}

		indexObj.BaseFromUsdaFasIndexId = item.BaseFromUsdaFasIndexId
		indexObj.IndexName = indexName
		indexObj.Frequency = req.Frequency
		indexObj.ClassifyId = classifyId
		indexObj.Country = req.Country
		indexObj.Commodity = req.Commodity
		indexObj.Unit = req.Unit
		indexObj.Sort = req.Sort
		indexObj.ModifyTime = time.Now()

		//修改数据
		updateColsArr := make([]string, 0)
		updateColsArr = append(updateColsArr, "index_name")
		updateColsArr = append(updateColsArr, "classify_id")
		updateColsArr = append(updateColsArr, "country")
		updateColsArr = append(updateColsArr, "commodity")
		updateColsArr = append(updateColsArr, "frequency")
		updateColsArr = append(updateColsArr, "sort")
		updateColsArr = append(updateColsArr, "modify_time")
		// 指标数据有变化才更新
		if oldIndexObj.IndexName != indexName || oldIndexObj.ClassifyId != classifyId || oldIndexObj.Country != req.Country || oldIndexObj.Commodity != req.Commodity || oldIndexObj.Frequency != req.Frequency || oldIndexObj.Sort != req.Sort {
		    e := indexObj.Update(updateColsArr)
		    if e != nil {
			   fmt.Println("Index Update Err:" + e.Error())
			   return
		    }
		}
	}

	//获取已存在的所有数据
	var exitDataList []*models.BaseFromUsdaFasData
	exitDataList, err = dataObj.GetByIndexCode(indexCode)
	if err != nil {
		err = fmt.Errorf("数据源查询美国农业部指标数据失败 Err：%s", err)
		return
	}
	fmt.Println("exitDataListLen:", len(exitDataList))
	for _, v := range exitDataList {
		dateStr := v.DataTime
		exitDataMap[dateStr] = v
	}

	// 遍历excel数据，然后跟现有的数据做校验，不存在则入库
	for date, value := range excelDataMap {
		if findData, ok := exitDataMap[date]; !ok {
			_, err = time.ParseInLocation(utils.FormatDate, date, time.Local)
			if err != nil {
				err = fmt.Errorf("%s 转换日期格式失败 Err：%s", date, err)
				return
			}
			//if !strings.Contains(value, "#N/A") {
			var saveDataTime time.Time
			if strings.Contains(date, "00:00:00") {
				saveDataTime, err = time.Parse(utils.FormatDateTime, date)
			} else {
				saveDataTime, err = time.Parse(utils.FormatDate, date)
			}
			if err != nil {
				err = fmt.Errorf("%s 转换日期格式失败 Err：%s", date, err)
				continue
			}
			timestamp := saveDataTime.UnixNano() / 1e6

			dataItem := new(models.BaseFromUsdaFasData)
			dataItem.BaseFromUsdaFasIndexId = int(indexId)
			dataItem.IndexCode = indexCode
			dataItem.DataTime = date
			dataItem.Value = value
			dataItem.CreateTime = time.Now()
			dataItem.ModifyTime = time.Now()
			dataItem.DataTimestamp = timestamp
			addDataList = append(addDataList, dataItem)
			if len(addDataList) > 500 {
				err = dataObj.AddMulti(addDataList)
				if err != nil {
					err = fmt.Errorf("批量新增指标失败 Err：%s", err)
					return
				}
				addDataList = make([]*models.BaseFromUsdaFasData, 0)
			}
			//}
		} else {
			if findData != nil && findData.Value != value && !strings.Contains(value, "#N/A") { //修改数据
				// 过滤0.50和0.5的比较
				oldV, _ := strconv.ParseFloat(findData.Value, 64)
				newV, _ := strconv.ParseFloat(value, 64)
				if oldV == newV {
					continue
				}
				dataObj.BaseFromUsdaFasIndexId = findData.BaseFromUsdaFasIndexId
				dataObj.Value = value
				dataObj.ModifyTime = time.Now()

				updateDataColsArr := make([]string, 0)
				updateDataColsArr = append(updateDataColsArr, "value")
				updateDataColsArr = append(updateDataColsArr, "modify_time")
				dataObj.Update(updateDataColsArr)
			}
		}
	}

	if len(addDataList) > 0 {
		err = dataObj.AddMulti(addDataList)
		if err != nil {
			err = fmt.Errorf("批量新增指标失败 Err：%s", err)
			return
		}
	}

	var dateItem *models.EdbInfoMaxAndMinInfo
	dateItem, err = dataObj.GetMaxAndMinDateByIndexCode(indexCode)
	if err != nil {
		err = fmt.Errorf("查询指标最新日期失败 Err：%s", err)
		return
	}

	go func() {
		if isAdd == 2 && oldIndexObj != nil && oldIndexObj.BaseFromUsdaFasIndexId > 0 {
			// 判断指标数据是否有变更
			if oldIndexObj.StartDate != dateItem.MinDate || oldIndexObj.EndDate != dateItem.MaxDate || oldIndexObj.EndValue != dateItem.LatestValue {
				indexObj.ModifyIndexMaxAndMinDate(indexCode, dateItem)
			}
		} else {
			indexObj.ModifyIndexMaxAndMinDate(indexCode, dateItem)
		}
	}()

	// 同步刷新ETA指标库的指标
	{
		// 获取指标详情
		baseObj := new(models.BaseFromUsdaFas)
		var edbInfo *models.EdbInfo
		edbInfo, err = models.GetEdbInfoByEdbCode(baseObj.GetSource(), indexCode)
		if err != nil {
			if !utils.IsErrNoRow(err) {
				errMsgList = append(errMsgList, fmt.Sprint("刷新ETA指标异常，指标编码：", indexCode, err.Error()))
				return
			} else {
				err = nil
			}
		}

		// 已经加入到指标库的话，那么就去更新ETA指标库吧
		if edbInfo != nil {
			go logic.RefreshBaseEdbInfo(edbInfo, ``)
		}
	}
	return
}