eta_index_lib/models/edb_data_calculate_avg.go

package models

import (
	"encoding/json"
	"errors"
	"eta/eta_index_lib/utils"
	"fmt"
	"github.com/beego/beego/v2/client/orm"
	"github.com/shopspring/decimal"
	"reflect"
	"strconv"
	"strings"
	"time"
)

type CalculateAvg struct {
}

// Add 新增
func (obj CalculateAvg) Add(params AddCalculateBatchParams) (edbInfo *EdbInfo, err error, errMsg string) {
	req := params.Req
	edbCode := params.EdbCode
	o := orm.NewOrm()
	to, err := o.Begin()
	if err != nil {
		return
	}
	defer func() {
		if err != nil {
			fmt.Println("CalculateAvg.Add,Err:" + err.Error())
			_ = to.Rollback()
		} else {
			_ = to.Commit()
		}
	}()
	if req.EdbInfoId > 0 {
		err = errors.New("无法新增")
		return
	}

	edbInfo = new(EdbInfo)
	edbInfo.Source = obj.GetSource()
	edbInfo.SourceName = obj.GetSourceName()
	edbInfo.EdbCode = edbCode
	edbInfo.EdbName = req.EdbName
	edbInfo.EdbNameSource = req.EdbName
	edbInfo.Frequency = req.Frequency
	edbInfo.Unit = req.Unit
	edbInfo.ClassifyId = req.ClassifyId
	edbInfo.SysUserId = params.SysUserId
	edbInfo.SysUserRealName = params.SysUserRealName
	edbInfo.CreateTime = time.Now()
	edbInfo.ModifyTime = time.Now()
	edbInfo.UniqueCode = params.UniqueCode
	edbInfo.EdbType = obj.GetEdbType()
	edbInfo.Sort = GetAddEdbMaxSortByClassifyId(req.ClassifyId, utils.EDB_INFO_TYPE)
	edbInfo.EmptyType = req.EmptyType
	edbInfo.MaxEmptyType = req.MaxEmptyType
	edbInfo.Extra = req.Extra
	edbInfo.EdbNameEn = req.EdbName
	edbInfo.UnitEn = req.Unit
	newEdbInfoId, tmpErr := to.Insert(edbInfo)
	if tmpErr != nil {
		err = tmpErr
		return
	}
	edbInfo.EdbInfoId = int(newEdbInfoId)

	//关联关系
	tagMap := make(map[string]int)
	relationEdbInfoList := make([]*EdbInfo, 0)
	calculateMappingItemList := make([]*EdbInfoCalculateMapping, 0)
	for k, v := range req.EdbInfoIdArr {
		tmpEdbInfo, e := GetEdbInfoById(v.EdbInfoId)
		if e != nil {
			err = e
			return
		}
		relationEdbInfoList = append(relationEdbInfoList, tmpEdbInfo)

		calculateMappingItem := new(EdbInfoCalculateMapping)
		calculateMappingItem.CreateTime = time.Now()
		calculateMappingItem.ModifyTime = time.Now()
		calculateMappingItem.Sort = k
		calculateMappingItem.EdbCode = edbCode
		calculateMappingItem.EdbInfoId = edbInfo.EdbInfoId
		calculateMappingItem.FromEdbInfoId = tmpEdbInfo.EdbInfoId
		calculateMappingItem.FromEdbCode = tmpEdbInfo.EdbCode
		calculateMappingItem.FromEdbName = tmpEdbInfo.EdbName
		calculateMappingItem.FromSource = tmpEdbInfo.Source
		calculateMappingItem.FromSourceName = tmpEdbInfo.SourceName
		calculateMappingItem.FromTag = v.FromTag
		calculateMappingItem.Source = edbInfo.Source
		calculateMappingItem.SourceName = edbInfo.SourceName
		calculateMappingItem.FromSubSource = edbInfo.SubSource
		calculateMappingItemList = append(calculateMappingItemList, calculateMappingItem)

		tagMap[v.FromTag] = v.EdbInfoId
	}
	_, err = to.InsertMulti(len(calculateMappingItemList), calculateMappingItemList)
	if err != nil {
		return
	}

	//计算数据
	err = obj.refresh(to, edbInfo.EdbInfoId, edbInfo.Source, edbInfo.SubSource, relationEdbInfoList, edbInfo.EdbCode, edbInfo.Extra, tagMap, edbInfo.EmptyType)

	return
}

func (obj CalculateAvg) Edit(params EditCalculateBatchParams) (err error, errMsg string) {
	req := params.Req
	edbInfo := params.EdbInfo
	o := orm.NewOrm()
	to, err := o.Begin()
	if err != nil {
		return
	}
	defer func() {
		if err != nil {
			fmt.Println("CalculateAvg.Edit,Err:" + err.Error())
			_ = to.Rollback()
		} else {
			_ = to.Commit()
		}
	}()

	if len(req.EdbInfoIdArr) <= 0 {
		err = fmt.Errorf("请选择要关联的指标")
		return
	}
	needCalculate := false
	if edbInfo.EmptyType != req.EmptyType || edbInfo.Extra != req.Extra {
		needCalculate = true
	}

	//修改指标信息
	edbInfo.EdbName = req.EdbName
	edbInfo.EdbNameSource = req.EdbName
	edbInfo.Frequency = req.Frequency
	edbInfo.Unit = req.Unit
	edbInfo.ClassifyId = req.ClassifyId
	edbInfo.CalculateFormula = req.Formula
	edbInfo.Extra = req.Extra
	edbInfo.EmptyType = req.EmptyType
	edbInfo.MaxEmptyType = req.MaxEmptyType
	edbInfo.EdbNameEn = req.EdbNameEn
	edbInfo.UnitEn = req.UnitEn
	edbInfo.ModifyTime = time.Now()
	_, err = to.Update(edbInfo, "EdbName", "EdbNameSource", "Frequency", "Unit", "ClassifyId", "CalculateFormula", "Extra", "EmptyType", "MaxEmptyType", "ModifyTime", "EdbNameEn", "UnitEn")
	if err != nil {
		return
	}
	//判断是否重复指标
	//edbInfoMap := make(map[int]string)
	if !needCalculate {
		//用到的指标
		newEdbInfoIdArr := make([]string, 0)
		for _, tmpEdbInfo := range req.EdbInfoIdArr {
			/*_, ok := edbInfoMap[tmpEdbInfo.EdbInfoId]
			if ok {
				br.Msg = "选择指标失败，请勿选择重复指标！"
				return
			}
			edbInfoMap[tmpEdbInfo.EdbInfoId] = tmpEdbInfo.FromTag*/
			newEdbInfoIdArr = append(newEdbInfoIdArr, strconv.Itoa(tmpEdbInfo.EdbInfoId))
		}

		edbInfoIdStr := strings.Join(newEdbInfoIdArr, ",")
		oldEdbInfoIdStr, e := GetCalculateEdbInfo(req.EdbInfoId)
		if e != nil {
			errMsg = "获取数据失败，GetCalculateEdbInfo:" + e.Error()
			err = fmt.Errorf("查询关联指标失败")
			return
		}

		if edbInfoIdStr != oldEdbInfoIdStr { // 不需要重复计算
			needCalculate = true
		}
	}
	fmt.Printf("needCalculate:%v\n", needCalculate)
	if !needCalculate {
		return
	}
	//删除，计算指标关联的，基础指标的关联关系
	sql := ` DELETE FROM edb_info_calculate_mapping WHERE edb_info_id = ? `
	_, err = to.Raw(sql, edbInfo.EdbInfoId).Exec()
	if err != nil {
		err = errors.New("删除计算指标关联关系失败,Err:" + err.Error())
		return
	}
	//清空原有数据
	tableName := GetEdbDataTableName(edbInfo.Source, edbInfo.SubSource)
	sql = ` DELETE FROM ` + tableName + ` WHERE edb_info_id = ? `
	_, err = to.Raw(sql, edbInfo.EdbInfoId).Exec()
	if err != nil {
		return
	}

	//关联关系
	tagMap := make(map[string]int)
	relationEdbInfoList := make([]*EdbInfo, 0)
	calculateMappingItemList := make([]*EdbInfoCalculateMapping, 0)
	for k, v := range req.EdbInfoIdArr {
		tmpEdbInfo, tmpErr := GetEdbInfoById(v.EdbInfoId)
		if tmpErr != nil {
			err = tmpErr
			return
		}
		relationEdbInfoList = append(relationEdbInfoList, tmpEdbInfo)

		calculateMappingItem := new(EdbInfoCalculateMapping)
		calculateMappingItem.CreateTime = time.Now()
		calculateMappingItem.ModifyTime = time.Now()
		calculateMappingItem.Sort = k
		calculateMappingItem.EdbCode = edbInfo.EdbCode
		calculateMappingItem.EdbInfoId = edbInfo.EdbInfoId
		calculateMappingItem.FromEdbInfoId = tmpEdbInfo.EdbInfoId
		calculateMappingItem.FromEdbCode = tmpEdbInfo.EdbCode
		calculateMappingItem.FromEdbName = tmpEdbInfo.EdbName
		calculateMappingItem.FromSource = tmpEdbInfo.Source
		calculateMappingItem.FromSourceName = tmpEdbInfo.SourceName
		calculateMappingItem.FromTag = v.FromTag
		calculateMappingItem.Source = edbInfo.Source
		calculateMappingItem.SourceName = edbInfo.SourceName
		calculateMappingItem.FromSubSource = edbInfo.SubSource
		calculateMappingItemList = append(calculateMappingItemList, calculateMappingItem)

		tagMap[v.FromTag] = v.EdbInfoId
	}
	_, err = to.InsertMulti(len(calculateMappingItemList), calculateMappingItemList)
	if err != nil {
		return
	}

	//计算数据
	err = obj.refresh(to, edbInfo.EdbInfoId, edbInfo.Source, edbInfo.SubSource, relationEdbInfoList, edbInfo.EdbCode, edbInfo.Extra, tagMap, edbInfo.EmptyType)

	return
}

func (obj CalculateAvg) Refresh(params RefreshParams) (err error, errMsg string) {
	edbInfo := params.EdbInfo
	edbInfoCalculateDetailList, err := GetEdbInfoCalculateDetailList(edbInfo.EdbInfoId)
	if err != nil {
		return
	}
	tagMap := make(map[string]int)
	relationEdbInfoList := make([]*EdbInfo, 0)
	for _, v := range edbInfoCalculateDetailList {
		tagMap[v.FromTag] = v.FromEdbInfoId
		fromEdbInfo, _ := GetEdbInfoById(v.FromEdbInfoId)
		relationEdbInfoList = append(relationEdbInfoList, fromEdbInfo)
	}
	o := orm.NewOrm()
	to, err := o.Begin()
	if err != nil {
		return
	}
	defer func() {
		if err != nil {
			fmt.Println(reflect.TypeOf(obj).Name() + ",Err:" + err.Error())
			_ = to.Rollback()
		} else {
			_ = to.Commit()
		}
	}()

	// 计算数据
	err = obj.refresh(to, edbInfo.EdbInfoId, edbInfo.Source, edbInfo.SubSource, relationEdbInfoList, edbInfo.EdbCode, edbInfo.Extra, tagMap, edbInfo.EmptyType)

	return
}

func (obj CalculateAvg) refresh(to orm.TxOrmer, edbInfoId, source, subSource int, edbInfoIdArr []*EdbInfo, edbCode, extra string, edbInfoTag map[string]int, emptyType int) (err error) {

	//to orm.TxOrmer, edbInfoId, source, subSource int, edbInfoIdArr []*EdbInfo, edbCode, extra string, edbInfoTag map[string]int, emptyType int
	realSaveDataMap := make(map[string]map[int]float64)
	saveDataMap := make(map[string]map[int]float64)

	// 最小的结束日期 , 最晚的数据开始日期
	var minLatestDate, maxStartDate time.Time
	dateList := make([]string, 0)        // 最终的日期数据
	dateMap := make(map[string]struct{}) // 最终的日期数据

	edbInfoIdStr := strconv.Itoa(edbInfoId)
	tableName := GetEdbDataTableName(source, subSource)

	// 获取多指标求和关联的指标id
	dateTagConfig := ""
	if extra != "" {
		var dateConfig CalculateEdbExtra
		err = json.Unmarshal([]byte(extra), &dateConfig)
		if err != nil {
			err = fmt.Errorf("refreshAllCalculate,extra解析失败,Err:%s", err.Error())
			return
		}
		dateTagConfig = dateConfig.DateTag
	}

	for edbInfoIndex, v := range edbInfoIdArr {
		dataList, err := GetEdbDataListAllByTo(to, v.Source, v.SubSource, FindEdbDataListAllCond{
			EdbInfoId: v.EdbInfoId,
		}, 1)
		if err != nil {
			return err
		}

		for _, dv := range dataList {
			if val, ok := realSaveDataMap[dv.DataTime]; ok {
				if _, ok := val[v.EdbInfoId]; !ok {
					val[v.EdbInfoId] = dv.Value
				}
			} else {
				temp := make(map[int]float64)
				temp[v.EdbInfoId] = dv.Value
				realSaveDataMap[dv.DataTime] = temp
			}

			// saveDataMap
			if val, ok := saveDataMap[dv.DataTime]; ok {
				if _, ok := val[v.EdbInfoId]; !ok {
					val[v.EdbInfoId] = dv.Value
				}
			} else {
				temp2 := make(map[int]float64)
				temp2[v.EdbInfoId] = dv.Value
				saveDataMap[dv.DataTime] = temp2
			}

			if dateTagConfig == "all" {
				if _, ok := dateMap[dv.DataTime]; !ok {
					dateList = append(dateList, dv.DataTime)
					dateMap[dv.DataTime] = struct{}{}
				}
			} else if dateTagConfig == "" { // 默认取第一个指标的时间序列
				if edbInfoIndex == 0 {
					if _, ok := dateMap[dv.DataTime]; !ok {
						dateList = append(dateList, dv.DataTime)
						dateMap[dv.DataTime] = struct{}{}
					}
				}
			} else {
				if eId, ok := edbInfoTag[dateTagConfig]; ok {
					if v.EdbInfoId == eId {
						if _, ok1 := dateMap[dv.DataTime]; !ok1 {
							dateList = append(dateList, dv.DataTime)
							dateMap[dv.DataTime] = struct{}{}
						}
					}
				}
			}
		}
	}
	// 处理最大日期和最小日期
	for _, v := range dateList {
		tmpDate, _ := time.ParseInLocation(utils.FormatDate, v, time.Local)
		if minLatestDate.IsZero() || tmpDate.After(minLatestDate) {
			minLatestDate = tmpDate
		}
		if maxStartDate.IsZero() || tmpDate.Before(maxStartDate) {
			maxStartDate = tmpDate
		}
	}

	//数据处理，将日期内不全的数据做补全
	HandleDateSaveDataMap(dateList, maxStartDate, minLatestDate, realSaveDataMap, saveDataMap, edbInfoIdArr, emptyType)

	addSql := ` INSERT INTO ` + tableName + ` (edb_info_id,edb_code,data_time,value,create_time,modify_time,data_timestamp) values `
	var isAdd bool

	//获取指标所有数据
	dataList := make([]*EdbData, 0)
	dataTableName := GetEdbDataTableName(source, subSource)
	sql := `SELECT * FROM %s WHERE edb_info_id=? `
	sql = fmt.Sprintf(sql, dataTableName)
	_, err = to.Raw(sql, edbInfoId).QueryRows(&dataList)
	if err != nil {
		return err
	}
	dataMap := make(map[string]string)

	removeDateMap := make(map[string]int) //需要移除的日期
	for _, v := range dataList {
		dataMap[v.DataTime] = v.Value
		removeDateMap[v.DataTime] = 1
	}
	existDataMap := make(map[string]string)

	for sk, sv := range saveDataMap {
		if _, ok := dateMap[sk]; !ok {
			continue
		}
		// 当空值处理类型选择了不计算时，只要有一个指标在某个日期没有值(即空值)，则计算指标在该日期没有值
		if len(sv) != len(edbInfoIdArr) {
			continue
		}
		var calVal float64
		for _, value := range sv {
			calVal += value
		}
		calVal = calVal / float64(len(sv))
		// 有计算出来值，那么就从待删除指标中移除
		delete(removeDateMap, sk)
		saveValue := decimal.NewFromFloat(calVal).Round(4).String()
		if existVal, ok := dataMap[sk]; !ok {
			dataTime, _ := time.ParseInLocation(utils.FormatDate, sk, time.Local)
			timestamp := dataTime.UnixNano() / 1e6
			timeStr := fmt.Sprintf("%d", timestamp)

			if _, existOk := existDataMap[sk]; !existOk {
				addSql += GetAddSql(edbInfoIdStr, edbCode, sk, timeStr, saveValue)
				isAdd = true
			}
			existDataMap[sk] = sk
		} else {
			existValDecimal, err := decimal.NewFromString(existVal)
			existStr := existValDecimal.String()
			if existStr != saveValue {
				sql := ` UPDATE %s SET value=?,modify_time=NOW() WHERE edb_info_id=? AND data_time=? `
				sql = fmt.Sprintf(sql, dataTableName)
				_, err = to.Raw(sql, saveValue, edbInfoId, sk).Exec()
				if err != nil {
					return err
				}
			}
		}
	}

	if isAdd {
		addSql = strings.TrimRight(addSql, ",")
		_, err = to.Raw(addSql).Exec()
		if err != nil {
			fmt.Println("RefreshAllCalculate add Err", err.Error())
			return
		}
	}

	if len(removeDateMap) > 0 {
		removeDateList := make([]string, 0) //需要移除的日期
		for k := range removeDateMap {
			removeDateList = append(removeDateList, k)
		}
		removeDateStr := strings.Join(removeDateList, `","`)
		removeDateStr = `"` + removeDateStr + `"`
		//如果拼接指标变更了，那么需要删除所有的指标数据
		sql := fmt.Sprintf(` DELETE FROM %s WHERE edb_info_id = ? and data_time in (%s) `, tableName, removeDateStr)

		_, err = to.Raw(sql, edbInfoId).Exec()
		if err != nil {
			err = fmt.Errorf("删除计算失败的计算指标数据失败,Err:" + err.Error())
			return
		}
	}
	return
}

// GetSource 获取来源编码id
func (obj CalculateAvg) GetSource() int {
	return utils.DATA_SOURCE_CALCULATE_AVG
}

// GetSourceName 获取来源名称
func (obj CalculateAvg) GetSourceName() string {
	return utils.DATA_SOURCE_NAME_CALCULATE_AVG
}

// GetEdbType 获取指标类型
func (obj CalculateAvg) GetEdbType() int {
	return utils.CALCULATE_EDB_TYPE
}