eta_index_lib/services/base_from_fenwei.go

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 {
		// 汾渭不存在数据更新和预测值情况，所以此处未做更新逻辑
		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()
}