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" "github.com/beego/beego/v2/core/logs" "github.com/mozillazg/go-pinyin" "strings" "time" "unicode" ) // HandleFenweiIndex 处理汾渭数据的excel数据 func HandleFenweiIndex(req *models.HandleFenweiExcelDataReq) (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) } }() for _, v := range req.List { if v.IndexName == "" || v.IndexCode == "" { errMsgList = append(errMsgList, fmt.Sprintf("新增指标异常,指标编码%s或者指标ID%s为空:", v.IndexCode, v.IndexName)) continue } err = handleFenweiIndex(v, req.TerminalCode) if err != nil { errMsgList = append(errMsgList, fmt.Sprintf("新增指标异常,指标编码:%s, Err: %s", v.IndexCode, err)) return } } return } func handleFenweiIndex(req *models.HandleFenweiExcelData, terminalCode 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("数据源-汾渭数据数据处理失败,err:", strings.Join(errMsgList, "\n")) utils.FileLog.Info(msg) go alarm_msg.SendAlarmMsg(msg, 3) } }() indexObj := new(models.BaseFromFenweiIndex) dataObj := new(models.BaseFromFenweiData) //classifyObj := new(models.BaseFromFenweiClassify) var indexId int64 addDataList := make([]*models.BaseFromFenweiData, 0) updateDataList := make([]*models.BaseFromFenweiData, 0) exitDataMap := make(map[string]*models.BaseFromFenweiData) // 修改指标信息 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() { // //新增分类 // classifyObj = &models.BaseFromFenweiClassify{ // ClassifyName: req.ClassifyName, // ParentId: 0, // SysUserId: 0, // SysUserRealName: "", // Level: 1, // 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 // } //} //判断指标是否存在 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.FenweiIndexId > 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 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.FenweiIndexId = indexId } else if isAdd == 2 { if item.TerminalCode == `` && terminalCode != `` { item.TerminalCode = terminalCode err = item.Update([]string{"TerminalCode"}) if err != nil { err = fmt.Errorf("数据源更新涌溢指标失败 Err:%s", err) return } } //获取已存在的所有数据 var exitDataList []*models.BaseFromFenweiData 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 = item.FenweiIndexId } // 遍历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.BaseFromFenweiData) dataItem.FenweiIndexId = 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) } } } 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) }() } // 批量更新数据 if len(updateDataList) > 0 { e := models.MultiUpdateBaseFromFenweiDataValue(updateDataList) if e != nil { err = fmt.Errorf("MultiUpdateBaseFromSmmDataValue err: %s", e.Error()) return } } // 同步刷新ETA指标库的指标 { // 获取指标详情 baseObj := new(models.BaseFromFenwei) 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 } func NetDataHandle(req models.HandleFenWeiNetDataReq) error { indexInfoList := req.List classifyObj := new(models.BaseFromFenweiClassify) // 因为传递过来得List是 指标数据 维度得,获取全量得指标数据比对,会导致分类和指标 频繁查询,会频繁访问数据库,做个简单优化吧 classifyMap := make(map[string]models.FenWeiNetIndexInfo) indexMap := make(map[string]models.FenWeiNetIndexInfo) for _, indexInfo := range indexInfoList { classifyMap[indexInfo.ClassifyName] = *indexInfo indexMap[indexInfo.IndexName] = *indexInfo } // 处理分类信息 classifyIdMap := make(map[string]int) for classifyName, reqData := range classifyMap { var classifyId int classify, err := classifyObj.GetByClassifyName(classifyName) if err != nil { return err } if classify == nil { logs.Info("NetDataHandle addClassify classifyName: ", classifyName) classifyId, err = addFenWeiClassify(&reqData) if err != nil { return err } } else { classifyId = classify[0].ClassifyId } // 保存分类ID classifyIdMap[classifyName] = classifyId } // 处理指标信息 indexIdMap := make(map[string]int64) indexCodeMap := make(map[string]string) for indexName, reqData := range indexMap { indexCode := GenerateIndexCode(indexName) indexCodeMap[indexName] = indexCode index, err := models.GetBaseFromFenWeiIndexByIndexName(indexName) if err != nil { return err } if index == nil { logs.Info("NetDataHandle indexName: ", indexName) classifyId := classifyIdMap[reqData.ClassifyName] logs.Info("NetDataHandle addIndex indexName: ", indexName) indexId, err := addIndex(&reqData, classifyId, indexCode) if err != nil { return err } indexIdMap[reqData.IndexName] = indexId } else { logs.Info("NetDataHandle exist indexName: ", indexName) indexIdMap[reqData.IndexName] = index.FenweiIndexId } } indexDataList := make([]*models.BaseFromFenweiData, 0) for _, indexInfo := range indexInfoList { // 处理指标数据 indexData, err := handleIndexData(indexInfo, indexIdMap, indexCodeMap) if err != nil { return err } if indexData != nil { indexDataList = append(indexDataList, indexData) } } // 批量插入指标数据 logs.Info("NetDataHandle addIndexData indexDataList.size: ", len(indexDataList)) if len(indexDataList) == 0 { return nil } err := models.BatchAddBaseFromFenWeiData(indexDataList) if err != nil { return err } logs.Info("NetDataHandle addIndexData success") return nil } func handleIndexData(indexInfo *models.FenWeiNetIndexInfo, indexIdMap map[string]int64, indexCodeMap map[string]string) (*models.BaseFromFenweiData, error) { var format string if isYearMonth(indexInfo.DataTime) { format = convertYearMonthToLastDay(indexInfo.DataTime) } else if indexInfo.Frequency == "月度" && isFirstDayOfMonth(indexInfo.DataTime) { format = convertYearMonthDayToLastDay(indexInfo.DataTime) } else { format = indexInfo.DataTime } // 获取指标数据 indexCode := indexCodeMap[indexInfo.IndexName] indexData, err := models.GetBaseFromFenweiDataByIndexCodeAndDataTime(indexCode, format) if err != nil { return nil, err } if indexData != nil { // 对数据做更新处理 if indexData.Value != fmt.Sprintf("%v", indexInfo.Value) { indexData.Value = fmt.Sprintf("%v", indexInfo.Value) indexData.ModifyTime = time.Now() logs.Info("handleIndexData update indexData : ", indexData) err = models.UpdateBaseFromFenweiData(indexData) if err != nil { return nil, err } } return nil, nil } timestamp := time.Now().UnixNano() / 1e6 indexId := indexIdMap[indexInfo.IndexName] data := new(models.BaseFromFenweiData) data.FenweiIndexId = int(indexId) data.IndexCode = indexCode data.DataTime = format data.Value = fmt.Sprintf("%v", indexInfo.Value) data.CreateTime = time.Now() data.ModifyTime = time.Now() data.DataTimestamp = timestamp // 打印data对象日志 logs.Info("handleIndexData data : ", data) return data, nil } func addIndex(info *models.FenWeiNetIndexInfo, classifyId int, indexCode string) (indexId int64, err error) { byIndexCode, err := models.GetBaseFromFenWeiIndexByIndexCode(indexCode) if err != nil { return 0, err } if byIndexCode != nil { return byIndexCode.FenweiIndexId, nil } index := new(models.BaseFromFenweiIndex) index.IndexName = info.IndexName index.IndexCode = indexCode index.IndexCode = indexCode index.Frequency = info.Frequency index.ClassifyId = classifyId index.Unit = info.Unit index.ModifyTime = time.Now() index.CreateTime = time.Now() index.TerminalCode = info.TerminalCode indexId, err = index.Add() if err != nil { return 0, err } return indexId, nil } // addFenWeiClassify 添加分类 func addFenWeiClassify(info *models.FenWeiNetIndexInfo) (int, error) { // 初始化分类结构 classify := new(models.BaseFromFenweiClassify) classify.ClassifyName = info.ClassifyName // 递归处理分类层级 return handleClassify(classify, 0, 1) } func handleClassify(classify *models.BaseFromFenweiClassify, parentId int, level int) (int, error) { // 将分类名按 "-" 分割 classifyArr := strings.Split(classify.ClassifyName, "-") // 当前分类名称 currentClassifyName := classifyArr[0] classify.ClassifyName = currentClassifyName classify.ParentId = parentId classify.Level = level // 检查当前分类是否已存在 existingClassifyList, err := classify.GetByClassifyName(currentClassifyName) if err != nil { return 0, err } var existingClassify *models.BaseFromFenweiClassify for _, item := range existingClassifyList { if item.ParentId == parentId { existingClassify = item } } // 如果分类存在,且父级ID相同,则递归处理下一级 if existingClassify != nil && existingClassify != nil { // 如果存在且有下一级,则递归处理下一级 if len(classifyArr) > 1 { nextClassifyName := strings.Join(classifyArr[1:], "-") nextClassify := new(models.BaseFromFenweiClassify) nextClassify.ClassifyName = nextClassifyName return handleClassify(nextClassify, existingClassify.ClassifyId, existingClassify.Level+1) } return existingClassify.ClassifyId, nil } // 如果不存在,新增当前分类 classify.SysUserId = 0 classify.SysUserRealName = "" classify.ModifyTime = time.Now() classify.CreateTime = time.Now() classifyId, err := models.AddBaseFromFenWeiClassify(classify) if err != nil { return 0, err } // 如果有下一级,递归处理下一级 if len(classifyArr) > 1 { nextClassifyName := strings.Join(classifyArr[1:], "-") nextClassify := new(models.BaseFromFenweiClassify) nextClassify.ClassifyName = nextClassifyName return handleClassify(nextClassify, int(classifyId), level+1) } return int(classifyId), nil } // 判断传入参数 dataText 是否是yyyy-MM格式,如果是则返回true,否则返回false func isYearMonth(dataText string) bool { _, err := time.Parse("2006-01", dataText) if err != nil { return false } return true } // 判断传入参数 dataText yyyy-MM-dd格式,是否是该月第一天,如果是则返回true,否则返回false func isFirstDayOfMonth(dataText string) bool { t, _ := time.Parse("2006-01-02", dataText) if t.Day() == 1 { return true } return false } // 转换时间 dataText yyyy-MM-dd格式,获取该月最后一天的日期 func convertYearMonthDayToLastDay(dataText string) string { t, _ := time.Parse("2006-01-02", dataText) lastDay := t.AddDate(0, 1, -1) return lastDay.Format("2006-01-02") } // 转换时间 dataText 2024-08 yyyy-MM 格式的时间字符串转换为该月最后一天的日期Date --> 2024-08-31 func convertYearMonthToLastDay(dataText string) string { t, _ := time.Parse("2006-01", dataText) lastDay := t.AddDate(0, 1, -1) return lastDay.Format("2006-01-02") } // GenerateIndexCode 指标编码规则:指标名称拼音首字母,数字、字母保留,特殊字符拿掉 // 例:美湾:9月U:国际大豆进口成本价:期货收盘:张家港 -----> lyswwmw9yUgjddjkcbjqhspzjg func GenerateIndexCode(indexName string) string { // 获取汉字的拼音首字母,保留数字和大写字母 indexCode := getFirstLetters(indexName) return indexCode } // getFirstLetters 获取汉字的拼音首字母,并保留数字和大写字母 func getFirstLetters(input string) string { // 设置拼音转换选项,只获取首字母 args := pinyin.NewArgs() args.Style = pinyin.FirstLetter // 定义用于存储结果的字符串 var result strings.Builder // 遍历输入字符串中的每个字符 for _, r := range input { if unicode.IsDigit(r) || unicode.IsUpper(r) { // 保留数字和大写字母 result.WriteRune(r) } else if unicode.Is(unicode.Han, r) { // 如果是汉字,则获取其拼音首字母 py := pinyin.Pinyin(string(r), args) if len(py) > 0 && len(py[0]) > 0 { result.WriteString(py[0][0]) } } // 对于其他字符,忽略处理 } return result.String() }