eta_index_lib/services/base_from_radish_research.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/services/elastic"
	"eta/eta_index_lib/utils"
	"fmt"
	"strings"
	"time"
)

// HandleRadishResearchIndex 处理萝卜投研的excel数据
func HandleRadishResearchIndex(req *models.HandleRadishResearchEdbDataReq) (err error) {
	errMsgList := make([]string, 0)
	defer func() {
		if len(errMsgList) > 0 {
			msg := fmt.Sprint("HandleRadishResearchIndex, err:", strings.Join(errMsgList, "\n"))
			utils.FileLog.Info(msg)
			go alarm_msg.SendAlarmMsg(msg, 3)
		}
	}()
	for _, v := range req.List {
		if v.IndexName == "" || v.IndexCode == "" {
			errMsgList = append(errMsgList, fmt.Sprintf("新增指标异常，指标编码%s或者指标ID%s为空：", v.IndexCode, v.IndexName))
			continue
		}
		err = handleRadishResearchIndex(v, req.TerminalCode, req.FilePath)
		if err != nil {
			errMsgList = append(errMsgList, fmt.Sprintf("新增指标异常，指标编码：%s, Err: %s", v.IndexCode, err))
			return
		}
	}
	return
}

func handleRadishResearchIndex(req *models.HandleRadishResearchEdbData, terminalCode, filePath string) (err error) {
	indexName := req.IndexName
	indexCode := req.IndexCode
	excelDataMap := req.ExcelDataMap
	errMsgList := make([]string, 0)
	defer func() {
		if len(errMsgList) > 0 {
			msg := fmt.Sprint("handleRadishResearchIndex, err:", strings.Join(errMsgList, "\n"))
			utils.FileLog.Info(msg)
			go alarm_msg.SendAlarmMsg(msg, 3)
		}
	}()
	indexObj := new(models.BaseFromRadishResearchIndex)
	dataObj := new(models.BaseFromRadishResearchData)
	//classifyObj := new(models.BaseFromCCFClassify)

	var indexId int64

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

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

	// 修改指标信息
	if indexName == "" {
		utils.FileLog.Info("未刷新到指标数据:indexName:" + indexName)
		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
		}
	}
	if item != nil && item.BaseFromRadishResearchIndexId > 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 = req.ClassifyId
		if req.Unit == "" {
			req.Unit = "无"
		}
		indexObj.Unit = req.Unit
		indexObj.Source = req.Source
		indexObj.Sort = req.Sort
		indexObj.ModifyTime = time.Now()
		indexObj.CreateTime = time.Now()
		indexObj.TerminalCode = terminalCode
		indexObj.FilePath = filePath
		indexId, err = indexObj.Add()
		if err != nil {
			err = fmt.Errorf("数据源新增萝卜投研指标失败 Err：%s", err)
			return
		}
		indexObj.BaseFromRadishResearchIndexId = int(indexId)
	} else if isAdd == 2 {
		item.TerminalCode = terminalCode
		item.FilePath = filePath
		item.Source = req.Source
		item.ModifyTime = time.Now().Local()
		updateCols := []string{"TerminalCode", "FilePath", "Source", "ModifyTime"}
		err = item.Update(updateCols)
		if err != nil {
			err = fmt.Errorf("数据源更新萝卜投研指标失败 Err：%s", err)
			return
		}

		//获取已存在的所有数据
		var exitDataList []*models.BaseFromRadishResearchData
		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
		}
		indexId = int64(item.BaseFromRadishResearchIndexId)
	}

	// 遍历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.BaseFromRadishResearchData)
				dataItem.BaseFromRadishResearchIndexId = 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)
			}
		} else {
			if findData != nil && findData.Value != value && !strings.Contains(value, "#N/A") { //修改数据
				findData.Value = value
				findData.ModifyTime = time.Now().Local()
				updateDataList = append(updateDataList, findData)
			}
		}
	}

	var esUpdated bool
	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() {
			indexObj.ModifyIndexMaxAndMinDate(indexCode, dateItem)
			_ = UpdateRadishResearchIndexEs(indexCode)
		}()
		esUpdated = true
	}

	// 批量更新数据
	if len(updateDataList) > 0 {
		e := models.MultiUpdateBaseFromRadishResearchDataValue(updateDataList)
		if e != nil {
			err = fmt.Errorf("MultiUpdateBaseFromRadishResearchDataValue err: %s", e.Error())
			return
		}
	}

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

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

	// 写入es
	if !esUpdated {
		_ = UpdateRadishResearchIndexEs(indexCode)
	}
	return
}

// UpdateRadishResearchIndexEs 更新萝卜投研指标ES
func UpdateRadishResearchIndexEs(indexCode string) (err error) {
	defer func() {
		if err != nil {
			tips := fmt.Sprintf("UpdateRadishResearchIndexEs, IndexCode: %s, errMsg: %v", indexCode, err)
			fmt.Println(tips)
			utils.FileLog.Info(tips)
		}
	}()
	indexObj := new(models.BaseFromRadishResearchIndex)
	item, e := indexObj.GetByIndexCode(indexCode)
	if e != nil {
		if utils.IsErrNoRow(e) {
			err = fmt.Errorf("指标不存在, IndexCode: %s", indexCode)
			return
		}
	}
	if item.BaseFromRadishResearchIndexId <= 0 {
		err = fmt.Errorf("指标不存在, IndexCode: %s", indexCode)
		return
	}
	indexItem := new(models.SearchDataSource)
	indexItem.PrimaryId = item.BaseFromRadishResearchIndexId
	indexItem.IndexCode = item.IndexCode
	indexItem.IndexName = item.IndexName
	indexItem.ClassifyId = item.ClassifyId
	indexItem.Unit = item.Unit
	indexItem.Frequency = item.Frequency
	indexItem.StartDate = item.StartDate
	indexItem.EndDate = item.EndDate
	indexItem.LatestValue = fmt.Sprint(item.LatestValue)
	indexItem.Source = utils.DATA_SOURCE_RADISH_RESEARCH
	indexItem.SourceName = utils.DATA_SOURCE_NAME_RADISH_RESEARCH
	indexItem.IsDeleted = 0
	indexItem.CreateTime = item.CreateTime.Format(utils.FormatDateTime)
	indexItem.ModifyTime = item.ModifyTime.Format(utils.FormatDateTime)

	docId := fmt.Sprintf("%d-%d", utils.DATA_SOURCE_RADISH_RESEARCH, item.BaseFromRadishResearchIndexId)
	if e := elastic.EsAddOrEditDataSourceIndex(utils.EsDataSourceIndexName, docId, indexItem); e != nil {
		err = fmt.Errorf("指标写入ES失败, %v", e)
		return
	}
	return
}