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" "strconv" "strings" "time" ) // HandleYongyiIndex 处理涌益咨询的excel数据 func HandleYongyiIndex(req *models.HandleYongyiExcelDataReq) (err error) { for _, v := range req.List { if v.IndexName == "" || v.IndexCode == "" { continue } err = handleYongyiIndex(v) if err != nil { return } } return } func handleYongyiIndex(req *models.HandleYongyiExcelData) (err error) { terminalCode := req.TerminalCode indexName := req.IndexName indexCode := req.IndexCode excelDataMap := req.ExcelDataMap errMsgList := make([]string, 0) defer func() { if len(errMsgList) > 0 { go alarm_msg.SendAlarmMsg(fmt.Sprint("数据源-涌益咨询数据处理失败,err:", strings.Join(errMsgList, "\n")), 3) } }() indexObj := new(models.BaseFromYongyiIndex) dataObj := new(models.BaseFromYongyiData) classifyObj := new(models.BaseFromYongyiClassify) var indexId int64 addDataList := make([]*models.BaseFromYongyiData, 0) exitDataMap := make(map[string]*models.BaseFromYongyiData) // 修改指标信息 if indexName == "" { utils.FileLog.Info("未刷新到指标数据:indexName:" + indexName) return } // 判断目录是否存在 var classifyId int64 now := time.Now() if req.ClassifyName != "" { classifyObj, err = classifyObj.GetByClassifyName(req.ClassifyName) if err != nil { if err.Error() == utils.ErrNoRow() { err = nil //新增分类 classifyObj = &models.BaseFromYongyiClassify{ ClassifyName: req.ClassifyName, ParentId: 0, SysUserId: 0, SysUserRealName: "", Level: 1, Sort: 0, ModifyTime: now, CreateTime: now, } classifyId, err = classifyObj.Add() if err != nil { fmt.Println("add err:" + err.Error()) return } classifyObj.ClassifyId = classifyId //修改排序 classifyObj.Sort = int(classifyId) err = classifyObj.Update([]string{"Sort"}) if err != nil { return } } return } else { classifyId = classifyObj.ClassifyId } } //判断指标是否存在 var isAdd int item, err := indexObj.GetByIndexCode(indexCode) if err != nil { if err.Error() == utils.ErrNoRow() { isAdd = 1 err = nil } else { isAdd = -1 fmt.Println("GetByIndexCode Err:" + err.Error()) return } } if item != nil && item.YongyiIndexId > 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.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 { fmt.Println("add err:" + err.Error()) return } indexObj.YongyiIndexId = indexId } else if isAdd == 2 { if item.TerminalCode == `` && terminalCode != `` { item.TerminalCode = terminalCode err = item.Update([]string{"TerminalCode"}) if err != nil { return } } //获取已存在的所有数据 var exitDataList []*models.BaseFromYongyiData exitDataList, err = dataObj.GetByIndexCode(indexCode) if err != nil { fmt.Println("GetIndexDataList Err:" + err.Error()) return } fmt.Println("exitDataListLen:", len(exitDataList)) for _, v := range exitDataList { dateStr := v.DataTime exitDataMap[dateStr] = v } indexId = item.YongyiIndexId } // 遍历excel数据,然后跟现有的数据做校验,不存在则入库 for date, value := range excelDataMap { if findData, ok := exitDataMap[date]; !ok { _, err = time.ParseInLocation(utils.FormatDate, date, time.Local) if err != nil { fmt.Println("time.ParseInLocation Err:" + err.Error()) 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 { errMsg := "saveDataTime parse err:" + err.Error() + " dataTime:" + date fmt.Println(errMsg) continue } timestamp := saveDataTime.UnixNano() / 1e6 dataItem := new(models.BaseFromYongyiData) dataItem.YongyiIndexId = 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") { //修改数据 } } } if len(addDataList) > 0 { err = dataObj.AddMulti(addDataList) if err != nil { fmt.Println("dataObj.Add() Err:" + err.Error()) return } var dateItem *models.EdbInfoMaxAndMinInfo dateItem, err = dataObj.GetMaxAndMinDateByIndexCode(indexCode) if err != nil { return } go func() { indexObj.ModifyIndexMaxAndMinDate(indexCode, dateItem) UpdateEsBaseFromYongyi(int(indexObj.YongyiIndexId)) }() } // 同步刷新ETA指标库的指标 { // 获取指标详情 baseObj := new(models.BaseFromYongyi) 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, ``) } } return } // UpdateEsBaseFromYongyi 更新数据源的es func UpdateEsBaseFromYongyi(yongyiIndexId int) (err error) { indexObj := new(models.BaseFromYongyiIndex) //添加es itemInfo, err := indexObj.GetByIndexId(yongyiIndexId) if err == nil { err = elastic.EsAddOrEditBaseFromYongyiIndex(utils.ES_INDEX_BASE_FROM_YONGYI_INDEX, strconv.Itoa(yongyiIndexId), itemInfo) return } return }