123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314 |
- 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
- }
|