eta_index_lib/services/base_from_business.go

package services

import (
	"errors"
	"eta/eta_index_lib/logic"
	"eta/eta_index_lib/models"
	"eta/eta_index_lib/models/mgo"
	"eta/eta_index_lib/services/alarm_msg"
	"eta/eta_index_lib/utils"
	"fmt"
	"github.com/qiniu/qmgo"
	"go.mongodb.org/mongo-driver/bson"
	"strings"
	"time"
)

// HandleBusinessIndex
// @Description:  处理外部指标
// @author: Roc
// @datetime 2024-04-26 14:23:42
// @param indexReq *models.AddBusinessIndexReq
// @return err error
func HandleBusinessIndex(indexReq *models.AddBusinessIndexReq) (resp models.BaseFromBusinessIndexResp, err error) {
	defer func() {
		if err != nil {
			// 添加刷新失败日志
			dataUpdateResult := 2
			dataUpdateFailedReason := "服务异常"
			edbInfo, e := models.GetEdbInfoByEdbCode(utils.DATA_SOURCE_MYSTEEL_CHEMICAL, indexReq.IndexCode)
			if e == nil {
				//查询指标存在，才添加刷新日志
				_ = AddEdbInfoUpdateLog(edbInfo.EdbInfoId, 2, err.Error(), dataUpdateResult, dataUpdateFailedReason, 1, 0)
			}
		}
	}()

	// 没有数据就返回
	if indexReq.DataList == nil || len(indexReq.DataList) <= 0 {
		return
	}
	// 兼容频度缺少度的字段
	if !strings.Contains(indexReq.Frequency, "度") {
		indexReq.Frequency = indexReq.Frequency + "度"
	}
	if !utils.VerifyFrequency(indexReq.Frequency) {
		err = errors.New("指标频度不合法:" + indexReq.Frequency)
		return
	}

	// 判断来源，如果来源不存在的话，则创建
	sourceObj := new(models.EdbBusinessSource)
	sourceItem, err := sourceObj.GetEdbBusinessSourceItem(indexReq.SourceName)
	if err != nil {
		if err.Error() == utils.ErrNoRow() {
			sourceItem = &models.EdbBusinessSource{
				EdbBusinessSourceId: 0,
				SourceName:          indexReq.SourceName,
				CreateTime:          time.Now(),
			}
			err = sourceItem.Add()
		} else {
			return
		}
	}

	// 指标
	indexObj := new(models.BaseFromBusinessIndex)
	if indexReq.IndexCode == `` {
		// 如果指标编码为空，那么自动生成
		currId, tmpErr := indexObj.GetMaxId()
		if tmpErr != nil {
			err = tmpErr
			return
		}
		indexReq.IndexCode = fmt.Sprintf("SELF%07d", currId+1)
	}
	//判断指标是否存在
	item, err := indexObj.GetIndexItem(indexReq.IndexCode)
	if err != nil {
		if err.Error() != utils.ErrNoRow() {
			return
		}

		// 添加指标
		item = &models.BaseFromBusinessIndex{
			BaseFromBusinessIndexId: 0,
			IndexCode:               indexReq.IndexCode,
			IndexName:               indexReq.IndexName,
			Unit:                    indexReq.Unit,
			Frequency:               indexReq.Frequency,
			Source:                  int(sourceItem.EdbBusinessSourceId),
			SourceName:              sourceItem.SourceName,
			//StartDate:               time.Time{},
			//EndDate:                 time.Time{},
			Remark:         indexReq.Remark,
			BaseModifyTime: time.Now(),
			DataUpdateTime: time.Now(),
			CreateTime:     time.Now(),
			ModifyTime:     time.Now(),
		}
		err = item.Add()
		if err != nil {
			fmt.Println("add err:" + err.Error())
			return
		}
	} else {
		updateCols := make([]string, 0)
		if item.IndexName != indexReq.IndexName {
			item.IndexName = indexReq.IndexName
			updateCols = append(updateCols, "IndexName")
		}
		if item.Unit != indexReq.Unit {
			item.Unit = indexReq.Unit
			updateCols = append(updateCols, "Unit")
		}
		if item.Frequency != indexReq.Frequency {
			item.Frequency = indexReq.Frequency
			updateCols = append(updateCols, "Frequency")
		}
		if item.Source != int(sourceItem.EdbBusinessSourceId) {
			item.Source = int(sourceItem.EdbBusinessSourceId)
			item.SourceName = sourceItem.SourceName
			updateCols = append(updateCols, "Source", "SourceName")
		}

		if len(updateCols) > 0 {
			item.BaseModifyTime = time.Now()
			item.ModifyTime = time.Now()
			updateCols = append(updateCols, "BaseModifyTime", "ModifyTime")
			err = item.Update(updateCols)
			if err != nil {
				fmt.Println("update index err:" + err.Error())
				return
			}
		}
	}

	// 数据处理
	// 当前传入的最小日期
	var reqMinDate time.Time
	if utils.UseMongo {
		reqMinDate, err = handleBusinessDataByMongo(item, indexReq.DataList)
	} else {
		reqMinDate, err = handleBusinessDataByMysql(item, indexReq.DataList)
	}
	if err != nil {
		return
	}

	// 同步刷新指标库的指标
	go refreshEdbBusiness(item.IndexCode, reqMinDate)

	resp = models.BaseFromBusinessIndexResp{
		IndexCode:  item.IndexCode,
		IndexName:  item.IndexName,
		Unit:       item.Unit,
		Frequency:  item.Frequency,
		SourceName: item.SourceName,
	}

	return
}

// handleBusinessDataByMongo
// @Description: 处理外部指标数据（mongo）
// @author: Roc
// @datetime 2024-07-01 15:30:41
// @param item *models.BaseFromBusinessIndex
// @param reqDataList []*models.AddBusinessDataReq
// @return reqMinDate time.Time 当前传入的最小日期
// @return err error
func handleBusinessDataByMongo(item *models.BaseFromBusinessIndex, reqDataList []models.AddBusinessDataReq) (reqMinDate time.Time, err error) {
	mogDataObj := new(mgo.BaseFromBusinessData)

	//获取已存在的所有数据
	exitDataList, err := mogDataObj.GetAllDataList(bson.M{"index_code": item.IndexCode}, []string{"data_time"})
	if err != nil {
		fmt.Println("GetIndexDataList Err:" + err.Error())
		return
	}

	// 已经存在的数据集
	exitDataMap := make(map[string]*mgo.BaseFromBusinessData)
	for _, v := range exitDataList {
		exitDataMap[v.DataTime.Format(utils.FormatDate)] = v
	}

	// 待添加的数据集
	addDataList := make([]interface{}, 0)
	updateDataList := make([]mgo.BaseFromBusinessData, 0)
	//var hasUpdate bool
	// 遍历excel数据，然后跟现有的数据做校验，不存在则入库
	for _, data := range reqDataList {
		dateTime, tmpErr := utils.DealExcelDate(data.Date)
		if tmpErr != nil {
			fmt.Println("time.ParseInLocation Err:" + tmpErr.Error())
			err = tmpErr
			return
		}

		// 调整最小日期
		if reqMinDate.IsZero() || reqMinDate.After(dateTime) {
			reqMinDate = dateTime
		}

		date := dateTime.Format(utils.FormatDate)

		findData, ok := exitDataMap[date]
		if !ok {
			addDataList = append(addDataList, mgo.BaseFromBusinessData{
				BaseFromBusinessIndexId: item.BaseFromBusinessIndexId,
				IndexCode:               item.IndexCode,
				DataTime:                dateTime,
				Value:                   data.Value,
				CreateTime:              time.Now(),
				ModifyTime:              time.Now(),
				//DataTimestamp:           0,
			})
			continue
		}

		// 值不匹配，修改数据
		if findData.Value != data.Value {
			findData.Value = data.Value
			updateDataList = append(updateDataList, *findData)
		}
	}

	// 指标数据是否新增或修改
	var isIndexUpdateOrAdd bool

	// 入库
	{
		coll := mogDataObj.GetCollection()
		if len(addDataList) > 0 {
			isIndexUpdateOrAdd = true
			err = mogDataObj.BatchInsertDataByColl(coll, 500, addDataList)
			if err != nil {
				fmt.Println("mogDataObj.HandleData() Err:" + err.Error())
				return
			}
		}

		if len(updateDataList) > 0 {
			isIndexUpdateOrAdd = true
			for _, v := range updateDataList {
				err = mogDataObj.UpdateDataByColl(coll, bson.M{"_id": v.ID}, bson.M{"$set": bson.M{"value": v.Value, "modify_time": v.ModifyTime}})
				if err != nil {
					fmt.Println("UpdateDataByColl:Err:" + err.Error())
					return
				}
			}
		}
	}

	// 支持事务的话，下面操作
	//result, err := mogDataObj.HandleData(addDataList, updateDataList)
	//if err != nil {
	//	fmt.Println("mogDataObj.HandleData() Err:" + err.Error())
	//	return
	//}
	//fmt.Println("result", result)

	//修改最大最小日期
	indexMaxAndMinInfo, err := item.GetEdbInfoMaxAndMinInfo(item.IndexCode)
	if err != nil {
		return
	}
	if err == nil && indexMaxAndMinInfo != nil {
		e := item.ModifyIndexMaxAndMinInfo(item.IndexCode, indexMaxAndMinInfo, isIndexUpdateOrAdd)
		if e != nil {
			fmt.Println("ModifyIndexMaxAndMinInfo Err:" + e.Error())
		}
	}

	return
}

// handleBusinessDataByMysql
// @Description: 处理外部指标数据（mysql）
// @author: Roc
// @datetime 2024-07-01 15:59:43
// @param item *models.BaseFromBusinessIndex
// @param reqDataList []models.AddBusinessDataReq
// @return reqMinDate time.Time
// @return err error
func handleBusinessDataByMysql(item *models.BaseFromBusinessIndex, reqDataList []models.AddBusinessDataReq) (reqMinDate time.Time, err error) {
	businessDataObj := new(models.BaseFromBusinessData)

	var condition []string
	var pars []interface{}
	condition = append(condition, "index_code = ?")
	pars = append(pars, item.IndexCode)
	//获取已存在的所有数据
	exitDataList, err := businessDataObj.GetAllDataList(condition, pars, "data_time ASC")
	if err != nil {
		fmt.Println("GetIndexDataList Err:" + err.Error())
		return
	}

	// 已经存在的数据集
	exitDataMap := make(map[string]*models.BaseFromBusinessData)
	for _, v := range exitDataList {
		exitDataMap[v.DataTime.Format(utils.FormatDate)] = v
	}

	// 待添加的数据集
	addDataList := make([]*models.BaseFromBusinessData, 0)
	updateDataList := make([]*models.BaseFromBusinessData, 0)
	//var hasUpdate bool
	// 遍历excel数据，然后跟现有的数据做校验，不存在则入库
	for _, data := range reqDataList {
		dateTime, tmpErr := utils.DealExcelDate(data.Date)
		if tmpErr != nil {
			fmt.Println("time.ParseInLocation Err:" + tmpErr.Error())
			err = tmpErr
			return
		}

		// 调整最小日期
		if reqMinDate.IsZero() || reqMinDate.After(dateTime) {
			reqMinDate = dateTime
		}

		date := dateTime.Format(utils.FormatDate)

		findData, ok := exitDataMap[date]
		if !ok {
			addDataList = append(addDataList, &models.BaseFromBusinessData{
				BaseFromBusinessIndexId: int(item.BaseFromBusinessIndexId),
				IndexCode:               item.IndexCode,
				DataTime:                dateTime,
				Value:                   data.Value,
				CreateTime:              time.Now(),
				ModifyTime:              time.Now(),
				//DataTimestamp:           0,
			})
			continue
		}

		// 值不匹配，修改数据
		if findData.Value != data.Value {
			findData.Value = data.Value
			findData.ModifyTime = time.Now()
			updateDataList = append(updateDataList, findData)
		}
	}

	// 指标数据是否新增或修改
	var isIndexUpdateOrAdd bool

	// 入库
	{
		if len(addDataList) > 0 {
			isIndexUpdateOrAdd = true
		}

		if len(updateDataList) > 0 {
			isIndexUpdateOrAdd = true
		}
		err = businessDataObj.HandleData(addDataList, updateDataList)
		if err != nil {
			fmt.Println("UpdateDataByColl:Err:" + err.Error())
			return
		}
	}

	//修改最大最小日期
	indexMaxAndMinInfo, err := item.GetEdbInfoMaxAndMinInfo(item.IndexCode)
	if err != nil {
		return
	}
	if err == nil && indexMaxAndMinInfo != nil {
		e := item.ModifyIndexMaxAndMinInfo(item.IndexCode, indexMaxAndMinInfo, isIndexUpdateOrAdd)
		if e != nil {
			fmt.Println("ModifyIndexMaxAndMinInfo Err:" + e.Error())
		}
	}

	return
}

// DelBusinessIndexResp
// @Description: 删除外部指标的返回
type DelBusinessIndexResp struct {
	IsDeleteEdbCodeList []string `description:"已经删除了的指标编码"`
	NoDeleteEdbCodeList []string `description:"未删除的指标编码"`
}

// DelBusinessIndex
// @Description: 删除外部指标
// @author: Roc
// @datetime 2024-05-31 16:27:37
// @param indexCodeList []string
// @return err error
// @return errMsg string
func DelBusinessIndex(indexCodeList []string) (joinEdbCodeList, needDelEdbCodeList []string, err error, errMsg string) {
	defer func() {
		if err != nil {
			fmt.Println("DelBusinessIndex Err:" + err.Error())
		}
	}()
	errMsg = "删除失败"

	if len(indexCodeList) < 0 {
		errMsg = "指标编码不允许为空"
		err = errors.New(errMsg)
		return
	}

	// 指标
	indexObj := new(models.BaseFromBusinessIndex)

	//判断指标是否存在
	baseIndexList, err := indexObj.GetIndexItemList(indexCodeList)
	if err != nil {
		return
	}

	// 已加入到eta指标库的编码map
	joinEdbCodeList = make([]string, 0)
	joinEdbMap := make(map[string]string)

	// 未加入到eta指标库的编码
	needDelEdbCodeList = make([]string, 0)

	// 判断指标是否加入到指标库
	tmpEdbInfoList, err := models.GetEdbInfoByEdbCodeList(utils.DATA_SOURCE_BUSINESS, indexCodeList)
	if err != nil {
		return
	}
	for _, v := range tmpEdbInfoList {
		joinEdbCodeList = append(joinEdbCodeList, v.EdbCode)
		joinEdbMap[v.EdbCode] = v.EdbCode
	}

	for _, v := range baseIndexList {
		_, ok := joinEdbMap[v.IndexCode]
		if !ok {
			needDelEdbCodeList = append(needDelEdbCodeList, v.IndexCode)
		}
	}

	// 如果需要删除的指标，则直接返回
	if len(needDelEdbCodeList) <= 0 {
		return
	}

	// 删除指标
	err = indexObj.DelIndexItemList(needDelEdbCodeList)
	if err != nil {
		fmt.Println("删除自有指标失败， Err:" + err.Error())
		return
	}

	if utils.UseMongo {
		// 删除指标明细数据
		mogDataObj := new(mgo.BaseFromBusinessData)
		err = mogDataObj.RemoveMany(bson.M{"index_code": bson.M{"$in": needDelEdbCodeList}})
	} else {
		var condition []string
		var pars []interface{}
		delNum := len(needDelEdbCodeList)
		if delNum > 0 {
			condition = append(condition, "index_code in ("+utils.GetOrmInReplace(delNum)+")")
			pars = append(pars, needDelEdbCodeList)

			businessDataObj := models.BaseFromBusinessData{}
			err = businessDataObj.DelDataByCond(condition, pars)
		}
	}
	if err != nil {
		fmt.Println("删除自有指标明细数据 Err:" + err.Error())
		return
	}

	return
}

// DelBusinessIndexData
// @Description: 删除指标数据
// @author: Roc
// @datetime 2024-05-31 15:43:32
// @param indexCode string
// @param dateList []string
// @return err error
// @return errMsg string
func DelBusinessIndexData(indexCode string, startDate, endDate string) (err error, errMsg string) {
	defer func() {
		if err != nil {
			fmt.Println("DelBusinessIndexData Err:" + err.Error())
		}
	}()
	errMsg = "删除失败"

	if indexCode == `` {
		errMsg = "指标编码不允许为空"
		err = errors.New(errMsg)
		return
	}
	if startDate == `` && endDate == `` {
		errMsg = "开始日期和结束日期不允许同时为空"
		err = errors.New(errMsg)
		return
	}

	// 指标
	indexObj := new(models.BaseFromBusinessIndex)

	//判断指标是否存在
	item, err := indexObj.GetIndexItem(indexCode)
	if err != nil {
		if err.Error() != utils.ErrNoRow() {
			return
		}
		err = nil
		return
	}

	// 当前传入的最小日期
	var reqMinDate time.Time

	var startDateTime, endDateTime time.Time
	if startDate != `` {
		//获取已存在的所有数据
		startDateTime, err = time.ParseInLocation(utils.FormatDate, startDate, time.Local)
		if err != nil {
			return
		}
		// 调整最小日期
		if reqMinDate.IsZero() || reqMinDate.After(startDateTime) {
			reqMinDate = startDateTime
		}
	}

	if endDate != `` {
		//获取已存在的所有数据
		endDateTime, err = time.ParseInLocation(utils.FormatDate, endDate, time.Local)
		if err != nil {
			return
		}
		// 调整最小日期
		if reqMinDate.IsZero() || reqMinDate.After(endDateTime) {
			reqMinDate = endDateTime
		}
	}

	// 删除具体的数据
	var indexMaxAndMinInfo *models.EdbInfoMaxAndMinInfo
	if utils.UseMongo {
		indexMaxAndMinInfo, err = delBusinessIndexDataByMongo(item, startDateTime, endDateTime)
	} else {
		indexMaxAndMinInfo, err = delBusinessIndexDataByMysql(item, startDateTime, endDateTime)
	}
	if err != nil {
		return
	}

	// 修改指标的最早最晚日期
	if indexMaxAndMinInfo != nil {
		e := item.ModifyIndexMaxAndMinInfo(item.IndexCode, indexMaxAndMinInfo, true)
		if e != nil {
			fmt.Println("ModifyIndexMaxAndMinInfo Err:" + e.Error())
		}
	}

	// 同步刷新指标库的指标
	go refreshEdbBusiness(item.IndexCode, reqMinDate)

	return
}

// delBusinessIndexDataByMongo
// @Description: 删除指标数据（从mongo删除）
// @author: Roc
// @datetime 2024-07-01 18:00:07
// @param item *models.BaseFromBusinessIndex
// @param startDateTime time.Time
// @param endDateTime time.Time
// @return err error
func delBusinessIndexDataByMongo(item *models.BaseFromBusinessIndex, startDateTime, endDateTime time.Time) (indexMaxAndMinInfo *models.EdbInfoMaxAndMinInfo, err error) {
	defer func() {
		if err != nil {
			utils.FileLog.Error("delBusinessIndexDataByMongo 删除自有指标明细数据 Err:" + err.Error())
		}
	}()

	// 构建查询条件
	queryConditions := bson.M{
		"index_code": item.IndexCode,
	}

	dateCondition, err := mgo.BuildDateTimeCondition(startDateTime, endDateTime)
	if err != nil {
		return
	}
	if len(dateCondition) > 0 {
		queryConditions["data_time"] = dateCondition
	}

	// 删除数据源中的指标明细数据
	mogDataObj := new(mgo.BaseFromBusinessData)
	err = mogDataObj.RemoveMany(queryConditions)
	if err != nil {
		return
	}

	//修改最大最小日期
	indexMaxAndMinInfo, err = item.GetEdbInfoMaxAndMinInfo(item.IndexCode)
	// 如果有错误，且错误信息是取不到文档，那么就不修改了
	if err != nil && !errors.Is(err, qmgo.ErrNoSuchDocuments) {
		return
	}
	// 清空的目的是为了避免异常返回
	err = nil

	return
}

// delBusinessIndexDataByMysql
// @Description: 删除指标数据（从mysql删除）
// @author: Roc
// @datetime 2024-07-02 09:53:13
// @param item *models.BaseFromBusinessIndex
// @param startDateTime time.Time
// @param endDateTime time.Time
// @return err error
func delBusinessIndexDataByMysql(item *models.BaseFromBusinessIndex, startDateTime, endDateTime time.Time) (indexMaxAndMinInfo *models.EdbInfoMaxAndMinInfo, err error) {
	defer func() {
		if err != nil {
			utils.FileLog.Error("delBusinessIndexDataByMysql 删除自有指标明细数据 Err:" + err.Error())
		}
	}()

	// 构建查询条件
	var condition []string
	var pars []interface{}
	condition = append(condition, "index_code = ? ")
	pars = append(pars, item.IndexCode)

	if !startDateTime.IsZero() {
		condition = append(condition, " data_time >= ? ")
		pars = append(pars, startDateTime.Format(utils.FormatDate))
	}
	if !endDateTime.IsZero() {
		condition = append(condition, " data_time <= ? ")
		pars = append(pars, endDateTime.Format(utils.FormatDate))
	}

	// 删除数据源中的指标明细数据
	businessDataObj := new(models.BaseFromBusinessData)
	err = businessDataObj.DelDataByCond(condition, pars)
	if err != nil {
		return
	}

	//修改最大最小日期
	indexMaxAndMinInfo, err = item.GetEdbInfoMaxAndMinInfo(item.IndexCode)
	// 如果有错误，且错误信息是取不到文档，那么就不修改了
	if err != nil && err.Error() != utils.ErrNoRow() {
		return
	}
	// 清空的目的是为了避免异常返回
	err = nil

	return
}

func refreshEdbBusiness(indexCode string, reqMinDate time.Time) {
	var indexErr error
	var errMsg string
	defer func() {
		if indexErr != nil {
			tips := fmt.Sprintf("自有数据刷新-ETA指标刷新异常, 指标编码: %s, err: %s, 错误信息：%s", indexCode, indexErr.Error(), errMsg)
			alarm_msg.SendAlarmMsg(tips, 3)
		}
	}()

	edbInfo, e := models.GetEdbInfoByEdbCode(utils.DATA_SOURCE_BUSINESS, indexCode)
	if e != nil && e.Error() != utils.ErrNoRow() {
		indexErr = e
		return
	}

	if edbInfo != nil {
		startDate := ``
		if reqMinDate.IsZero() {
			startDate = edbInfo.EndDate
		} else {
			startDate = reqMinDate.Format(utils.FormatDate)
		}
		params := models.RefreshBaseParams{
			EdbInfo:   edbInfo,
			StartDate: startDate,
		}
		obj := models.Business{}

		indexErr, errMsg = obj.Refresh(params)
		if indexErr != nil {
			return
		}

		// 更新指标最大最小值
		indexErr = obj.UnifiedModifyEdbInfoMaxAndMinInfo(edbInfo)
		if indexErr != nil {
			return
		}

		// 更新ES
		go logic.UpdateEs(edbInfo.EdbInfoId)
	}
}