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 }