eta_index_lib/models/predict_edb_data_base.go

package models

import (
	"encoding/json"
	"errors"
	"eta_gn/eta_index_lib/global"
	"eta_gn/eta_index_lib/utils"
	"fmt"

	"github.com/shopspring/decimal"
	"gorm.io/gorm"
	"reflect"
	"strconv"
	"strings"
	"time"
)

type PredictStandardBase struct {
}

func AddPredictEdb(item *EdbInfo, calculateMappingList []*EdbInfoCalculateMapping, predictEdbConfList []*PredictEdbConf, calculateRuleMap map[int]CalculateRule) (err error, errMsg string) {
	var latestDateStr string
	var latestValue float64
	to := global.DEFAULT_DmSQL.Begin()
	defer func() {
		if err != nil {
			to.Rollback()
		} else {
			to.Commit()
			go UnifiedModifyPredictEdbInfoMaxAndMinInfo(item, latestDateStr, latestValue)
		}
	}()
	err = to.Create(item).Error
	if err != nil {
		return
	}

	lenCalculateMapping := len(calculateMappingList)
	if lenCalculateMapping > 0 {
		for _, calculateMappingItem := range calculateMappingList {
			calculateMappingItem.EdbInfoId = item.EdbInfoId
			calculateMappingItem.EdbCode = item.EdbCode
		}
		err = to.CreateInBatches(calculateMappingList, 500).Error
		if err != nil {
			return
		}
	}

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

	for k, v := range predictEdbConfList {
		v.PredictEdbInfoId = item.EdbInfoId
		tmpErr := to.Create(v).Error
		if tmpErr != nil {
			err = tmpErr
			return
		}

		resultDataList := make([]*EdbInfoSearchData, 0)
		switch v.RuleType {
		case 9: //动态环差规则
			calculateRule := calculateRuleMap[k]
			calculateRule.ConfigId = v.ConfigId
			calculateRule.EdbInfoId = v.PredictEdbInfoId

			resultDataList, err = CalculateByRuleBy9(to, 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 = to.CreateInBatches(calculateRule.TrendsCalculateMappingList, 500).Error
				if err != nil {
					return
				}
			}
		case 14: //14：根据 一元线性拟合 规则获取预测数据
			calculateRule := calculateRuleMap[k]
			calculateRule.ConfigId = v.ConfigId
			calculateRule.EdbInfoId = v.PredictEdbInfoId

			err, errMsg = CalculateByRuleByRuleLineNh(to, *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 = to.CreateInBatches(calculateRule.TrendsCalculateMappingList, 500).Error
				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(to, item, predictEdbConfAndDataList, item.EdbCode, "")

	return
}

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

	to := global.DEFAULT_DmSQL.Begin()
	defer func() {
		if err != nil {
			to.Rollback()
		} else {
			to.Commit()
			go UnifiedModifyPredictEdbInfoMaxAndMinInfo(edbInfo, latestDateStr, latestValue)
		}
	}()
	err = to.Model(edbInfo).Select(updateEdbInfoCol).Updates(edbInfo).Error
	if err != nil {
		return
	}

	sql := ` DELETE FROM edb_info_calculate_mapping WHERE edb_info_id = ?`
	err = to.Exec(sql, edbInfo.EdbInfoId).Error
	if err != nil {
		return
	}

	sql = ` DELETE FROM predict_edb_conf WHERE predict_edb_info_id = ?`
	err = to.Exec(sql, edbInfo.EdbInfoId).Error
	if err != nil {
		return
	}

	sql = ` DELETE FROM predict_edb_conf_calculate_mapping WHERE edb_info_id = ?`
	err = to.Exec(sql, edbInfo.EdbInfoId).Error
	if err != nil {
		return
	}

	sql = ` DELETE FROM predict_edb_rule_data WHERE edb_info_id = ?`
	err = to.Exec(sql, edbInfo.EdbInfoId).Error
	if err != nil {
		return
	}

	lenCalculateMapping := len(calculateMappingList)
	if lenCalculateMapping > 0 {
		for _, calculateMappingItem := range calculateMappingList {
			calculateMappingItem.EdbInfoId = edbInfo.EdbInfoId
			calculateMappingItem.EdbCode = edbInfo.EdbCode
		}
		err = to.CreateInBatches(calculateMappingList, 500).Error
		if err != nil {
			return
		}
	}

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

	for confIndex, v := range predictEdbConfList {
		tmpErr := to.Create(v).Error
		if tmpErr != nil {
			err = tmpErr
			return
		}
		resultDataList := make([]*EdbInfoSearchData, 0)

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

			resultDataList, err = CalculateByRuleBy9(to, 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 = to.CreateInBatches(calculateRule.TrendsCalculateMappingList, 500).Error
				if err != nil {
					return
				}
			}
		case 14: //14：根据 一元线性拟合 规则获取预测数据
			calculateRule := calculateRuleMap[confIndex]
			calculateRule.ConfigId = v.ConfigId
			calculateRule.EdbInfoId = v.PredictEdbInfoId

			err, errMsg = CalculateByRuleByRuleLineNh(to, *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 = to.CreateInBatches(calculateRule.TrendsCalculateMappingList, 500).Error
				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(to, edbInfo, predictEdbConfAndDataList, edbInfo.EdbCode, "")

	return
}

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

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

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
	}

	to := global.DEFAULT_DmSQL.Begin()
	defer func() {
		if err != nil {
			to.Rollback()
		} else {
			to.Commit()
		}
	}()

	for k, v := range predictEdbConfList {
		tmpDataList, tmpErr := GetPredictEdbRuleDataByTo(to, v.ConfigId)
		if tmpErr != nil {
			err = tmpErr
			return
		}

		edbInfoSearchDataList := make([]*EdbInfoSearchData, 0)
		for _, tmpData := range tmpDataList {
			tmpValueDec, tmpErr := decimal.NewFromString(tmpData.Value)
			if tmpErr != nil {
				err = tmpErr
				return
			}
			tmpVal, _ := tmpValueDec.Float64()
			edbInfoSearchDataList = append(edbInfoSearchDataList, &EdbInfoSearchData{
				DataTime: tmpData.DataTime,
				Value:    tmpVal,
			})
		}
		predictEdbConfList[k].DataList = edbInfoSearchDataList
	}

	latestDateStr, latestValue, err, errMsg = obj.refresh(to, edbInfo, predictEdbConfList, edbInfo.EdbCode, "")

	return
}

func (obj PredictStandardBase) GetSource() int {
	return utils.DATA_SOURCE_PREDICT
}

func (obj PredictStandardBase) GetSourceName() string {
	return utils.DATA_SOURCE_NAME_PREDICT
}

func (obj PredictStandardBase) refresh(to *gorm.DB, edbInfo *EdbInfo, predictEdbConfAndDataList []*PredictEdbConfAndData, edbCode, startDate string) (latestDateStr string, latestValue float64, err error, errMsg string) {
	fmt.Println(predictEdbConfAndDataList)
	by, _ := json.Marshal(predictEdbConfAndDataList)
	utils.FileLog.Info(string(by))
	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, err := GetAllEdbDataListByTo(to, edbInfoId, edbInfo.Source, edbInfo.SubSource)
	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)

			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.Exec(sql, saveValue, edbInfoId, currDateStr).Error
				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.Exec(sql, edbInfo.EdbInfoId, removeDateList).Error
			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.Exec(addSql).Error
		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
}

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
}

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"))
		}
	}()

	var list []*PredictEdbConf
	sql := ` SELECT * FROM predict_edb_conf WHERE source_edb_info_id=? `
	err := global.DEFAULT_DmSQL.Raw(sql, sourceEdbInfoId).Scan(&list).Error
	if err != nil {
		errList = append(errList, fmt.Sprintf("根据来源换指标id获取配置项失败,来源指标ID:%d；err：%s", sourceEdbInfoId, err.Error()))
		return
	}
	total := len(list)
	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
}