eta_index_lib/models/predict_edb_data_base.go

package models

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

// PredictStandardBase 基础预测指标
type PredictStandardBase struct {
}

// AddPredictEdb 添加预测指标
// edbInfo, calculateMappingList, predictEdbConfList,calculateRule9List,trendsMappingList
func AddPredictEdb(item *EdbInfo, calculateMappingList []*EdbInfoCalculateMapping, predictEdbConfList []*PredictEdbConf, calculateRuleMap map[int]CalculateRule) (err error, errMsg string) {
	var latestDateStr string
	var latestValue float64
	o := orm.NewOrm()
	tx, err := o.Begin()
	if err != nil {
		return
	}
	defer func() {
		if err != nil {
			tx.Rollback()
		} else {
			err = tx.Commit()

			// 更新指标最大最小值
			go UnifiedModifyPredictEdbInfoMaxAndMinInfo(item, latestDateStr, latestValue)
		}
	}()
	// 新增预测指标
	edbInfoId, err := tx.Insert(item)
	if err != nil {
		return
	}
	item.EdbInfoId = int(edbInfoId)

	// 新增预测指标的关联关系
	lenCalculateMapping := len(calculateMappingList)
	if lenCalculateMapping > 0 {
		for _, calculateMappingItem := range calculateMappingList {
			calculateMappingItem.EdbInfoId = item.EdbInfoId
			calculateMappingItem.EdbCode = item.EdbCode
		}
		_, err = tx.InsertMulti(lenCalculateMapping, calculateMappingList)
		if err != nil {
			return
		}
	}

	predictEdbConfAndDataList := make([]*PredictEdbConfAndData, 0)

	// 新增预测指标配置
	for k, v := range predictEdbConfList {
		v.PredictEdbInfoId = item.EdbInfoId
		configId, tmpErr := tx.Insert(v)
		if tmpErr != nil {
			err = tmpErr
			return
		}
		v.ConfigId = int(configId)

		// 每次规则计算的时候，产生的临时数据
		resultDataList := make([]*EdbInfoSearchData, 0)
		switch v.RuleType {
		case 9: //动态环差规则
			calculateRule := calculateRuleMap[k]
			calculateRule.ConfigId = v.ConfigId
			calculateRule.EdbInfoId = v.PredictEdbInfoId

			// 指标与规则的动态数据生成入库
			resultDataList, err = CalculateByRuleBy9(tx, calculateRule)
			if err != nil {
				return
			}

			// 规则与指标的关系入库
			lenTrendsCalculateMapping := len(calculateRule.TrendsCalculateMappingList)
			if lenTrendsCalculateMapping > 0 {
				for _, vv := range calculateRule.TrendsCalculateMappingList {
					vv.EdbInfoId = item.EdbInfoId
					vv.ConfigId = v.ConfigId
				}
				_, err = tx.InsertMulti(lenTrendsCalculateMapping, calculateRule.TrendsCalculateMappingList)
				if err != nil {
					return
				}
			}
		case 14: //14：根据 一元线性拟合 规则获取预测数据
			calculateRule := calculateRuleMap[k]
			calculateRule.ConfigId = v.ConfigId
			calculateRule.EdbInfoId = v.PredictEdbInfoId

			// 指标与规则的动态数据（拟合数据）生成入库
			err, errMsg = CalculateByRuleByRuleLineNh(tx, *item, predictEdbConfAndDataList, *v)
			if err != nil {
				return
			}

			// 规则与指标的关系入库
			lenTrendsCalculateMapping := len(calculateRule.TrendsCalculateMappingList)
			if lenTrendsCalculateMapping > 0 {
				for _, vv := range calculateRule.TrendsCalculateMappingList {
					vv.EdbInfoId = item.EdbInfoId
					vv.ConfigId = v.ConfigId
				}
				_, err = tx.InsertMulti(lenTrendsCalculateMapping, calculateRule.TrendsCalculateMappingList)
				if err != nil {
					return
				}
			}

		}

		// 规则配置（含数据）
		tmpPredictEdbConfAndData := &PredictEdbConfAndData{
			ConfigId:         0,
			PredictEdbInfoId: 0,
			SourceEdbInfoId:  v.SourceEdbInfoId,
			RuleType:         v.RuleType,
			FixedValue:       v.FixedValue,
			Value:            v.Value,
			EndDate:          v.EndDate,
			ModifyTime:       v.ModifyTime,
			CreateTime:       v.CreateTime,
			DataList:         resultDataList,
		}
		predictEdbConfAndDataList = append(predictEdbConfAndDataList, tmpPredictEdbConfAndData)
	}

	// 重新预测数据
	obj := new(PredictStandardBase)
	latestDateStr, latestValue, err, errMsg = obj.refresh(tx, item, predictEdbConfAndDataList, item.EdbCode, "")

	return
}

// EditPredictEdb 修改预测指标
func EditPredictEdb(edbInfo *EdbInfo, updateEdbInfoCol []string, calculateMappingList []*EdbInfoCalculateMapping, predictEdbConfList []*PredictEdbConf, calculateRuleMap map[int]CalculateRule) (err error, errMsg string) {
	var latestDateStr string
	var latestValue float64

	o := orm.NewOrm()
	tx, err := o.Begin()
	if err != nil {
		return
	}
	defer func() {
		if err != nil {
			tx.Rollback()
		} else {
			err = tx.Commit()

			// 更新指标最大最小值
			go UnifiedModifyPredictEdbInfoMaxAndMinInfo(edbInfo, latestDateStr, latestValue)
		}
	}()
	// 修改预测指标
	_, err = tx.Update(edbInfo, updateEdbInfoCol...)
	if err != nil {
		return
	}

	// 先删除原有的预测指标 与 其他指标的 关联关系
	sql := ` DELETE FROM edb_info_calculate_mapping WHERE edb_info_id = ?`
	_, err = tx.Raw(sql, edbInfo.EdbInfoId).Exec()
	if err != nil {
		return
	}

	// 先删除原有的配置
	sql = ` DELETE FROM predict_edb_conf WHERE predict_edb_info_id = ?`
	_, err = tx.Raw(sql, edbInfo.EdbInfoId).Exec()
	if err != nil {
		return
	}

	// 删除基础预测指标 规则配置 与 其他指标的 关联关系
	sql = ` DELETE FROM predict_edb_conf_calculate_mapping WHERE edb_info_id = ?`
	_, err = tx.Raw(sql, edbInfo.EdbInfoId).Exec()
	if err != nil {
		return
	}

	// 删除基础预测指标 规则配置 生成的动态数据值
	sql = ` DELETE FROM predict_edb_rule_data WHERE edb_info_id = ?`
	_, err = tx.Raw(sql, edbInfo.EdbInfoId).Exec()
	if err != nil {
		return
	}

	// 新增预测指标的关联关系
	lenCalculateMapping := len(calculateMappingList)
	if lenCalculateMapping > 0 {
		for _, calculateMappingItem := range calculateMappingList {
			calculateMappingItem.EdbInfoId = edbInfo.EdbInfoId
			calculateMappingItem.EdbCode = edbInfo.EdbCode
		}
		_, err = tx.InsertMulti(lenCalculateMapping, calculateMappingList)
		if err != nil {
			return
		}
	}

	calculateRuleIndex := 0 // 预测计算规则下标
	predictEdbConfAndDataList := make([]*PredictEdbConfAndData, 0)

	// 新增预测指标配置
	for confIndex, v := range predictEdbConfList {
		configId, tmpErr := tx.Insert(v)
		if tmpErr != nil {
			err = tmpErr
			return
		}
		v.ConfigId = int(configId)
		// 每次规则计算的时候，产生的临时数据
		resultDataList := make([]*EdbInfoSearchData, 0)

		switch v.RuleType {
		case 9: //动态环差规则
			calculateRule := calculateRuleMap[confIndex]
			calculateRule.ConfigId = v.ConfigId
			calculateRule.EdbInfoId = v.PredictEdbInfoId

			// 指标与规则的动态数据生成入库
			resultDataList, err = CalculateByRuleBy9(tx, calculateRule)
			if err != nil {
				return
			}

			// 规则与指标的关系入库
			lenTrendsCalculateMapping := len(calculateRule.TrendsCalculateMappingList)
			if lenTrendsCalculateMapping > 0 {
				for _, vv := range calculateRule.TrendsCalculateMappingList {
					vv.EdbInfoId = edbInfo.EdbInfoId
					vv.ConfigId = v.ConfigId
				}
				_, err = tx.InsertMulti(lenTrendsCalculateMapping, calculateRule.TrendsCalculateMappingList)
				if err != nil {
					return
				}
			}
		case 14: //14：根据 一元线性拟合 规则获取预测数据
			calculateRule := calculateRuleMap[confIndex]
			calculateRule.ConfigId = v.ConfigId
			calculateRule.EdbInfoId = v.PredictEdbInfoId

			// 指标与规则的动态数据（拟合数据）生成入库
			err, errMsg = CalculateByRuleByRuleLineNh(tx, *edbInfo, predictEdbConfAndDataList, *v)
			if err != nil {
				return
			}

			// 规则与指标的关系入库
			lenTrendsCalculateMapping := len(calculateRule.TrendsCalculateMappingList)
			if lenTrendsCalculateMapping > 0 {
				for _, vv := range calculateRule.TrendsCalculateMappingList {
					vv.EdbInfoId = edbInfo.EdbInfoId
					vv.ConfigId = v.ConfigId
				}
				_, err = tx.InsertMulti(lenTrendsCalculateMapping, calculateRule.TrendsCalculateMappingList)
				if err != nil {
					return
				}
			}

		}

		calculateRuleIndex++

		// 规则配置（含数据）
		tmpPredictEdbConfAndData := &PredictEdbConfAndData{
			ConfigId:         0,
			PredictEdbInfoId: 0,
			SourceEdbInfoId:  v.SourceEdbInfoId,
			RuleType:         v.RuleType,
			FixedValue:       v.FixedValue,
			Value:            v.Value,
			EndDate:          v.EndDate,
			ModifyTime:       v.ModifyTime,
			CreateTime:       v.CreateTime,
			DataList:         resultDataList,
		}
		predictEdbConfAndDataList = append(predictEdbConfAndDataList, tmpPredictEdbConfAndData)
	}

	// 重新预测数据
	obj := new(PredictStandardBase)
	latestDateStr, latestValue, err, errMsg = obj.refresh(tx, edbInfo, predictEdbConfAndDataList, edbInfo.EdbCode, "")

	return
}

// Add 添加
func (obj PredictStandardBase) Add(params BatchSaveCalculateBatchParams) (edbInfo *EdbInfo, latestDateStr string, latestValue float64, err error, errMsg string) {
	err = errors.New("暂不支持该方法")
	return
}

// Edit 编辑
func (obj PredictStandardBase) Edit(params BatchSaveCalculateBatchParams) (latestDateStr string, latestValue float64, err error, errMsg string) {
	err = errors.New("暂不支持该方法")
	return
}

// Refresh 刷新
func (obj PredictStandardBase) Refresh(params RefreshParams) (latestDateStr string, latestValue float64, err error, errMsg string) {
	edbInfo := params.EdbInfo
	if edbInfo == nil {
		errMsg = `错误的指标`
		err = errors.New(errMsg)
		return
	}

	// 查找该预测指标配置
	predictEdbConfList, err := GetPredictEdbConfAndDataListById(edbInfo.EdbInfoId)
	if err != nil && err.Error() != utils.ErrNoRow() {
		errMsg = "获取预测指标配置信息失败"
		return
	}
	if len(predictEdbConfList) == 0 {
		errMsg = "获取预测指标配置信息失败"
		err = errors.New(errMsg)
		return
	}

	o := orm.NewOrm()
	to, err := o.Begin()
	if err != nil {
		return
	}
	defer func() {
		if err != nil {
			_ = to.Rollback()
			fmt.Println(reflect.TypeOf(obj).Name(), "；Refresh,Err:"+err.Error())
		} else {
			_ = to.Commit()
		}
	}()

	// 重新预测数据
	latestDateStr, latestValue, err, errMsg = obj.refresh(to, edbInfo, predictEdbConfList, edbInfo.EdbCode, "")

	return
}

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

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

func (obj PredictStandardBase) refresh(to orm.TxOrmer, edbInfo *EdbInfo, predictEdbConfAndDataList []*PredictEdbConfAndData, edbCode, startDate string) (latestDateStr string, latestValue float64, err error, errMsg string) {
	edbInfoId := edbInfo.EdbInfoId
	dataTableName := GetEdbDataTableName(edbInfo.Source, edbInfo.SubSource)
	edbInfoIdStr := strconv.Itoa(edbInfoId)

	// 获取标准差图表的指标数据
	fromDataList, fromEdbInfo, err, errMsg := obj.GetCalculateDataList(edbInfo, predictEdbConfAndDataList, 1, startDate)
	if err != nil {
		return
	}
	latestDateStr = fromEdbInfo.LatestDate

	//获取指标所有数据
	existDataList := make([]*EdbData, 0)
	sql := `SELECT * FROM %s WHERE edb_info_id=? `
	sql = fmt.Sprintf(sql, dataTableName)
	_, err = to.Raw(sql, edbInfoId).QueryRows(&existDataList)
	if err != nil {
		return
	}
	existDataMap := make(map[string]string)
	removeDataTimeMap := make(map[string]int) //需要移除的日期数据
	for _, v := range existDataList {
		existDataMap[v.DataTime] = v.Value
		removeDataTimeMap[v.DataTime] = 1
	}
	needAddDateMap := make(map[time.Time]int)

	addSql := ` INSERT INTO ` + dataTableName + `(edb_info_id,edb_code,data_time,value,create_time,modify_time,data_timestamp) values `
	var isAdd bool
	for _, tmpData := range fromDataList {
		currDateStr := tmpData.DataTime
		currTime, tmpErr := time.ParseInLocation(utils.FormatDate, currDateStr, time.Local)
		if tmpErr != nil {
			err = tmpErr
			return
		}
		// 当前的实际值
		saveValue := decimal.NewFromFloat(tmpData.Value).Round(4).String()

		existVal, ok := existDataMap[currDateStr]
		// 如果库中已经存在该数据的话，那么就进行值的变更操作
		if ok {
			//校验待删除日期数据里面是否存在该元素，如果存在的话，那么移除该日期
			delete(removeDataTimeMap, currDateStr)

			// 格式化库中的指标数据，并保留4位小数
			existValDec, tmpErr := decimal.NewFromString(existVal)
			if tmpErr != nil {
				err = tmpErr
				return
			}
			existVal = existValDec.Round(4).String()

			if existVal != 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, currDateStr).Exec()
				if err != nil {
					return
				}
			}

			continue
		}

		// 库中不存在该日期的数据
		timestamp := currTime.UnixNano() / 1e6
		timeStr := fmt.Sprintf("%d", timestamp)
		if _, existOk := needAddDateMap[currTime]; !existOk {
			addSql += GetAddSql(edbInfoIdStr, edbCode, currDateStr, timeStr, saveValue)
			isAdd = true
		}
		needAddDateMap[currTime] = 1
	}

	//删除已经不存在的指标数据（由于该指标当日的数据删除了）
	{
		removeDateList := make([]string, 0)
		for dateTime := range removeDataTimeMap {
			removeDateList = append(removeDateList, dateTime)
		}
		removeNum := len(removeDateList)
		if removeNum > 0 {
			sql := fmt.Sprintf(` DELETE FROM %s WHERE edb_info_id = ? and data_time in (`+utils.GetOrmInReplace(removeNum)+`) `, dataTableName)
			_, err = to.Raw(sql, edbInfo.EdbInfoId, removeDateList).Exec()
			if err != nil {
				fmt.Println(reflect.TypeOf(obj).Name(), " add data ;delete Err", err.Error())
				err = fmt.Errorf("删除不存在的指标数据失败,Err:" + err.Error())
				return
			}
		}
	}

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

	//确定实际数据的最终值
	{
		finalLast, tmpErr := GetFinalLastByTo(to, edbInfoId, edbInfo.Source, edbInfo.SubSource, fromEdbInfo.LatestDate)
		if tmpErr != nil && tmpErr.Error() != utils.ErrNoRow() {
			return
		}
		if tmpErr == nil {
			latestDateStr = finalLast.DataTime
			latestValue = finalLast.Value
		}
	}

	return
}

// GetCalculateDataList
// @Description: 获取计算后的数据
// @author: Roc
// @receiver obj
// @datetime 2024-04-24 10:42:51
// @param edbInfo *EdbInfo
// @param predictEdbConfList []*PredictEdbConfAndData
// @param order int order:1升序，其余值为降序
// @param startDate string
// @return dataList []*EdbInfoSearchData
// @return sourceEdbInfoItem *EdbInfo
// @return err error
// @return errMsg string
func (obj PredictStandardBase) GetCalculateDataList(edbInfo *EdbInfo, predictEdbConfList []*PredictEdbConfAndData, order int, startDate string) (dataList []*EdbInfoSearchData, sourceEdbInfoItem *EdbInfo, err error, errMsg string) {
	if len(predictEdbConfList) == 0 {
		errMsg = "获取预测指标配置信息失败"
		err = errors.New(errMsg)
		return
	}
	predictEdbConf := predictEdbConfList[0]

	// 来源指标
	sourceEdbInfoItem, err = GetEdbInfoById(predictEdbConf.SourceEdbInfoId)
	if err != nil {
		if err.Error() == utils.ErrNoRow() {
			errMsg = "找不到来源指标信息"
			err = errors.New(errMsg)
		}
		return
	}

	dataList, err, errMsg = GetPredictDataListByPredictEdbConfList(edbInfo, sourceEdbInfoItem, predictEdbConfList, order, startDate)

	return
}

// RefreshPredictStandardBaseByGeneralEdbInfoId
// @Description:  根据普通指标id刷新预测标准基础数据
// @author: Roc
// @datetime 2024-04-24 13:40:59
// @param sourceEdbInfoId int
// @return err error
func RefreshPredictStandardBaseByGeneralEdbInfoId(sourceEdbInfoId int) {
	errList := make([]string, 0)
	defer func() {
		if len(errList) > 0 {
			fmt.Println(errList)
			utils.FileLog.Info("RefreshPredictStandardBaseByGeneralEdbInfoId更新失败，来源ID： ErrMsg:" + strings.Join(errList, "\n"))
		}
	}()
	o := orm.NewOrm()
	var list []*PredictEdbConf
	sql := ` SELECT * FROM predict_edb_conf WHERE source_edb_info_id=? `
	total, err := o.Raw(sql, sourceEdbInfoId).QueryRows(&list)
	if err != nil {
		errList = append(errList, fmt.Sprintf("根据来源换指标id获取配置项失败,来源指标ID:%d；err：%s", sourceEdbInfoId, err.Error()))
		return
	}

	if total > 0 {
		idList := make([]int, 0)
		for _, v := range list {
			idList = append(idList, v.PredictEdbInfoId)
			edbInfo, tmpErr := GetEdbInfoById(v.PredictEdbInfoId)
			if tmpErr != nil {
				errList = append(errList, fmt.Sprintf("获取预测指标配置信息失败,预测指标配置ID:%d；err：%s", v.PredictEdbInfoId, tmpErr.Error()))
				continue
			}
			refreshParams := RefreshParams{
				EdbInfo: edbInfo,
			}
			obj := new(PredictStandardBase)
			latestDateStr, latestValue, tmpErr, errMsg := obj.Refresh(refreshParams)
			if tmpErr != nil {
				errList = append(errList, fmt.Sprintf("更新失败,预测指标配置ID:%d；err：%s；errMsg:%s", v.PredictEdbInfoId, tmpErr.Error(), errMsg))
				continue
			}

			// 更新指标最大最小值
			tmpErr, errMsg = UnifiedModifyPredictEdbInfoMaxAndMinInfo(edbInfo, latestDateStr, latestValue)
			if tmpErr != nil {
				errList = append(errList, fmt.Sprintf("更新指标最大最小值失败,预测指标配置ID:%d；err：%s；errMsg:%s", v.PredictEdbInfoId, tmpErr.Error(), errMsg))
				continue
			}
		}
	}

	return
}