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 } } 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") 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() { indexObj.ModifyIndexMaxAndMinDate(indexCode, dateItem) }() // 同步刷新ETA指标库的指标 { // 获取指标详情 baseObj := new(models.BaseFromUsdaFas) 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 }