|
@@ -0,0 +1,308 @@
|
|
|
+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"})
|
|
|
+ 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.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.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, "unit")
|
|
|
+ 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
|
|
|
+}
|