fix:添加降频指标

models/edb_data_calculate_jp.go

@@ -0,0 +1,464 @@
+package models
+
+import (
+	"errors"
+	"fmt"
+	"github.com/beego/beego/v2/client/orm"
+	"github.com/shopspring/decimal"
+	"hongze/hongze_edb_lib/utils"
+	"strconv"
+	"strings"
+	"time"
+)
+
+// CheckFrequency 判断是否可以降频
+func CheckFrequency(fromEdbFrequency, edbFrequency string) (ok bool) {
+	frequencyList := make([]string, 0)
+	switch fromEdbFrequency {
+	case "日度":
+		frequencyList = []string{"日度"}
+	case "周度":
+		frequencyList = []string{"日度", "周度"}
+	case "旬度":
+		frequencyList = []string{"日度", "周度", "旬度"}
+	case "月度":
+		frequencyList = []string{"日度", "周度", "旬度", "月度"}
+	case "季度":
+		frequencyList = []string{"日度", "周度", "旬度", "月度", "季度"}
+	default:
+		return
+	}
+
+	// 如果存在,那么就标识不允许降频
+	if utils.InArrayByStr(frequencyList, edbFrequency) {
+		return
+	}
+	ok = true
+
+	return
+}
+
+// AddCalculateJp 降频
+func AddCalculateJp(req *EdbInfoCalculateBatchSaveReq, fromEdbInfo *EdbInfo, edbCode, uniqueCode string, sysUserId int, sysUserRealName string) (edbInfo *EdbInfo, err error) {
+	o := orm.NewOrm()
+	to, err := o.Begin()
+	if err != nil {
+		return
+	}
+	defer func() {
+		if err != nil {
+			fmt.Println("AddCalculateJp,Err:" + err.Error())
+			_ = to.Rollback()
+		} else {
+			_ = to.Commit()
+		}
+	}()
+	if req.EdbInfoId <= 0 {
+		edbInfo = new(EdbInfo)
+		edbInfo.Source = utils.DATA_SOURCE_CALCULATE_JP
+		edbInfo.SourceName = "降频"
+		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 = sysUserId
+		edbInfo.SysUserRealName = sysUserRealName
+		edbInfo.CreateTime = time.Now()
+		edbInfo.ModifyTime = time.Now()
+		edbInfo.UniqueCode = uniqueCode
+		edbInfo.CalculateFormula = req.Formula
+		edbInfo.EdbType = 2
+		newEdbInfoId, tmpErr := to.Insert(edbInfo)
+		if tmpErr != nil {
+			err = tmpErr
+			return
+		}
+		edbInfo.EdbInfoId = int(newEdbInfoId)
+		//关联关系
+		{
+			calculateMappingItem := new(EdbInfoCalculateMapping)
+			calculateMappingItem.CreateTime = time.Now()
+			calculateMappingItem.ModifyTime = time.Now()
+			calculateMappingItem.Sort = 1
+			calculateMappingItem.EdbCode = edbCode
+			calculateMappingItem.EdbInfoId = edbInfo.EdbInfoId
+			calculateMappingItem.FromEdbInfoId = fromEdbInfo.EdbInfoId
+			calculateMappingItem.FromEdbCode = fromEdbInfo.EdbCode
+			calculateMappingItem.FromEdbName = fromEdbInfo.EdbName
+			calculateMappingItem.FromSource = fromEdbInfo.Source
+			calculateMappingItem.FromSourceName = fromEdbInfo.SourceName
+			calculateMappingItem.FromTag = ""
+			calculateMappingItem.Source = edbInfo.Source
+			calculateMappingItem.SourceName = edbInfo.SourceName
+			_, err = to.Insert(calculateMappingItem)
+			if err != nil {
+				return
+			}
+		}
+	} else {
+		edbInfo, err = GetEdbInfoById(req.EdbInfoId)
+		if err != nil {
+			return
+		}
+		dataTableName := GetEdbDataTableName(utils.DATA_SOURCE_CALCULATE_JP)
+		fmt.Println("dataTableName:" + dataTableName)
+		deleteSql := ` DELETE FROM %s WHERE edb_info_id=? `
+		deleteSql = fmt.Sprintf(deleteSql, dataTableName)
+		_, err = to.Raw(deleteSql, req.EdbInfoId).Exec()
+		if err != nil {
+			return
+		}
+	}
+
+	//计算数据
+	err = refreshAllCalculateJp(to, edbInfo.EdbInfoId, edbInfo.Source, fromEdbInfo, edbInfo.EdbCode, edbInfo.Frequency, fromEdbInfo.Frequency, edbInfo.CalculateFormula)
+
+	return
+}
+
+// EditCalculateJp 修改降频数据
+func EditCalculateJp(edbInfo *EdbInfo, req *EdbInfoCalculateBatchEditReq, fromEdbInfo *EdbInfo) (err error) {
+	o := orm.NewOrm()
+	to, err := o.Begin()
+	if err != nil {
+		return
+	}
+	defer func() {
+		if err != nil {
+			fmt.Println("EditCalculateJp,Err:" + err.Error())
+			_ = to.Rollback()
+		} else {
+			_ = to.Commit()
+		}
+	}()
+
+	//修改指标信息
+	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.ModifyTime = time.Now()
+	_, err = to.Update(edbInfo, "EdbName", "EdbNameSource", "Frequency", "Unit", "ClassifyId", "CalculateFormula", "ModifyTime")
+	if err != nil {
+		return
+	}
+
+	//判断计算指标是否被更换
+	var existCondition string
+	var existPars []interface{}
+	existCondition += " AND edb_info_id=? AND from_edb_info_id=? "
+	existPars = append(existPars, edbInfo.EdbInfoId, req.FromEdbInfoId)
+
+	count, err := GetEdbInfoCalculateCountByCondition(existCondition, existPars)
+	if err != nil {
+		err = errors.New("判断指标是否改变失败,Err:" + err.Error())
+		return
+	}
+	if count > 0 { // 指标未被替换，无需重新计算
+		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)
+	sql = ` DELETE FROM ` + tableName + ` WHERE edb_info_id = ? `
+	_, err = to.Raw(sql, edbInfo.EdbInfoId).Exec()
+	if err != nil {
+		return
+	}
+	//关联关系
+	{
+		calculateMappingItem := &EdbInfoCalculateMapping{
+			EdbInfoCalculateMappingId: 0,
+			EdbInfoId:                 edbInfo.EdbInfoId,
+			Source:                    utils.DATA_SOURCE_CALCULATE_JP,
+			SourceName:                "降频",
+			EdbCode:                   edbInfo.EdbCode,
+			FromEdbInfoId:             fromEdbInfo.EdbInfoId,
+			FromEdbCode:               fromEdbInfo.EdbCode,
+			FromEdbName:               fromEdbInfo.EdbName,
+			FromSource:                fromEdbInfo.Source,
+			FromSourceName:            fromEdbInfo.SourceName,
+			FromTag:                   "",
+			Sort:                      1,
+			CreateTime:                time.Now(),
+			ModifyTime:                time.Now(),
+		}
+		_, err = to.Insert(calculateMappingItem)
+		if err != nil {
+			return
+		}
+	}
+
+	//计算数据
+	err = refreshAllCalculateJp(to, edbInfo.EdbInfoId, edbInfo.Source, fromEdbInfo, edbInfo.EdbCode, edbInfo.Frequency, fromEdbInfo.Frequency, edbInfo.CalculateFormula)
+
+	return
+}
+
+func RefreshAllCalculateJp(edbInfoId, source int, fromEdbInfo *EdbInfo, edbCode, edbFrequency, formula string) (err error) {
+	o := orm.NewOrm()
+	to, err := o.Begin()
+	if err != nil {
+		return
+	}
+	defer func() {
+		if err != nil {
+			fmt.Println("RefreshAllCalculateJp,Err:" + err.Error())
+			_ = to.Rollback()
+		} else {
+			_ = to.Commit()
+		}
+	}()
+
+	// 计算数据
+	err = refreshAllCalculateJp(to, edbInfoId, source, fromEdbInfo, edbCode, edbFrequency, fromEdbInfo.Frequency, formula)
+
+	return
+}
+
+// refreshAllCalculateJp 刷新降频数据
+func refreshAllCalculateJp(to orm.TxOrmer, edbInfoId, source int, fromEdbInfo *EdbInfo, edbCode, edbFrequency, fromEdbFrequency, formula string) (err error) {
+	edbInfoIdStr := strconv.Itoa(edbInfoId)
+
+	//计算数据
+	var condition string
+	var pars []interface{}
+	condition += " AND edb_info_id=? "
+	pars = append(pars, fromEdbInfo.EdbInfoId)
+
+	//if startDate != "" {
+	//	condition += " AND data_time>=? "
+	//	pars = append(pars, startDate)
+	//}
+	//if endDate != "" {
+	//	condition += " AND data_time<=? "
+	//	pars = append(pars, endDate)
+	//}
+	//获取来源指标的数据
+	dataList, err := GetEdbDataListAllByTo(to, condition, pars, fromEdbInfo.Source, 1)
+	if err != nil {
+		return err
+	}
+	var dateArr []string
+	dataMap := make(map[string]*EdbInfoSearchData)
+	fromDataMap := make(map[string]float64)
+	//来源指指标数据
+	for _, v := range dataList {
+		dateArr = append(dateArr, v.DataTime)
+		dataMap[v.DataTime] = v
+		fromDataMap[v.DataTime] = v.Value
+	}
+	fmt.Println("source:", source)
+
+	//获取降频指标所有数据
+	existDataList, err := GetAllEdbDataListByTo(to, edbInfoId, source)
+	if err != nil {
+		return
+	}
+	//计算指标的map
+	existDataMap := make(map[string]*EdbData, 0)
+	//existDataMap := make(map[string]string, 0)
+	for _, v := range existDataList {
+		existDataMap[v.DataTime] = v
+	}
+
+	tableName := GetEdbDataTableName(utils.DATA_SOURCE_CALCULATE_JP)
+	addSql := ` INSERT INTO ` + tableName + ` (edb_info_id,edb_code,data_time,value,create_time,modify_time,data_timestamp) values `
+	var isAdd bool
+
+	//existMap := make(map[string]string)
+	dataLen := len(dataList)
+	if dataLen <= 0 {
+		return
+	}
+	startDataTime, _ := time.ParseInLocation(utils.FormatDate, dataList[0].DataTime, time.Local)
+	endDataTime, _ := time.ParseInLocation(utils.FormatDate, dataList[dataLen-1].DataTime, time.Local)
+
+	var lastValue float64     // 最近的值
+	var nextEndDate time.Time // 下一个节点的日期
+	weekDayDataList := make([]float64, 0)
+	for tmpStartDataTime := startDataTime; !tmpStartDataTime.After(endDataTime); tmpStartDataTime = tmpStartDataTime.AddDate(0, 0, 1) {
+		// 将当前数据加入到 weekDayDataList
+		if tmpData, ok := dataMap[tmpStartDataTime.Format(utils.FormatDate)]; ok {
+			tmpValue := decimal.NewFromFloat(tmpData.Value)
+			tmpValueFloat, _ := tmpValue.Truncate(4).Float64()
+			weekDayDataList = append(weekDayDataList, tmpValueFloat)
+		}
+		// 如果下个节点的日期不存在,那么就先给赋值(兼容时间区间内只有一组数据的情况)
+		if nextEndDate.IsZero() {
+			nextEndDate = utils.GetFrequencyEndDay(tmpStartDataTime, edbFrequency)
+		}
+
+		// 日期处理过滤
+		switch edbFrequency {
+		case "周度":
+			if tmpStartDataTime.Weekday() != 0 {
+				//不是周日,代表需要进入下一个循环获取数据并计算
+				continue
+			} else {
+				//记录下一个结束节点的日期
+				nextEndDate = tmpStartDataTime.AddDate(0, 0, 7)
+			}
+		case "旬度":
+			nextDay := tmpStartDataTime.AddDate(0, 0, 1)
+			if nextDay.Day() != 1 && nextDay.Day() != 11 && nextDay.Day() != 21 {
+				//不是每月10、20、最后一天,代表需要进入下一个循环获取数据并计算
+				continue
+			} else {
+				//记录下一个结束节点的日期
+				if nextDay.Day() == 1 || nextDay.Day() == 11 {
+					//月初或者月末的时候,加10天就好了
+					nextEndDate = nextDay.AddDate(0, 0, 9)
+				} else {
+					tmpNextMonth := nextDay.AddDate(0, 1, 0)
+					nextEndDate = time.Date(tmpNextMonth.Year(), tmpNextMonth.Month(), 1, 0, 0, 0, 0, time.Local).AddDate(0, 0, -1)
+				}
+			}
+		case "月度":
+			nextDay := tmpStartDataTime.AddDate(0, 0, 1)
+			if nextDay.Day() != 1 {
+				//不是每月最后一天,代表需要进入下一个循环获取数据并计算
+				continue
+			} else {
+				//记录下一个结束节点的日期
+				nextEndDate = nextDay.AddDate(0, 1, -1)
+			}
+		case "季度":
+			nextDay := tmpStartDataTime.AddDate(0, 0, 1)
+			if (nextDay.Month() == 1 || nextDay.Month() == 4 || nextDay.Month() == 7 || nextDay.Month() == 10) && nextDay.Day() == 1 {
+				//记录下一个结束节点的日期
+				nextEndDate = nextDay.AddDate(0, 3, -1)
+			} else {
+				//不是3,6,9,12 月份的最后一天,代表需要进入下一个循环获取数据并计算
+				continue
+			}
+		case "年度":
+			if tmpStartDataTime.Month() == 12 && tmpStartDataTime.Day() == 31 {
+				//记录下一个结束节点的日期
+				nextEndDate = tmpStartDataTime.AddDate(1, 0, 0)
+			} else {
+				//不是每年的12-31日,代表需要进入下一个循环获取数据并计算
+				continue
+			}
+		default:
+			err = errors.New("错误的频度:" + edbFrequency)
+			return
+		}
+
+		// 当前时间段内的数据计算,得出实际值
+		var currVal float64
+		lenWeekDayDataList := len(weekDayDataList)
+		// 如果这个时间区间内没有数据,那么就采用上一个时间区间的值
+		if len(weekDayDataList) <= 0 {
+			currVal = lastValue
+		} else {
+			if formula == "期末值" {
+				currVal = weekDayDataList[lenWeekDayDataList-1]
+			} else {
+				// 平均值
+				sumValDeci := decimal.NewFromFloat(0)
+				for _, v := range weekDayDataList {
+					tmpValDeci := decimal.NewFromFloat(v)
+					sumValDeci = sumValDeci.Add(tmpValDeci)
+				}
+				lenDeci := decimal.NewFromInt(int64(lenWeekDayDataList))
+				currVal, _ = sumValDeci.Div(lenDeci).Truncate(4).Float64()
+			}
+		}
+
+		// 判断降频指标是否存在数据
+		if existData, ok := existDataMap[tmpStartDataTime.Format(utils.FormatDate)]; ok {
+			// 处理降频数据的值
+			existValStr := existData.Value
+			existValDeci, tmpErr := decimal.NewFromString(existValStr)
+			if tmpErr != nil {
+				err = tmpErr
+				return
+			}
+			existVal, _ := existValDeci.Truncate(4).Float64()
+			// 判断降频数据的值 与 当前计算出来的结果, 如果两个数据结果不相等的话,那么就修改咯
+			if existVal != currVal {
+				err = ModifyEdbDataById(source, existData.EdbDataId, fmt.Sprint(currVal))
+				if err != nil {
+					return err
+				}
+			}
+		} else {
+			// 直接入库
+			timestamp := tmpStartDataTime.UnixNano() / 1e6
+			timestampStr := fmt.Sprintf("%d", timestamp)
+			addSql += GetAddSql(edbInfoIdStr, edbCode, tmpStartDataTime.Format(utils.FormatDate), timestampStr, fmt.Sprint(currVal))
+			isAdd = true
+		}
+
+		// 一轮结束后,数据清空
+		weekDayDataList = make([]float64, 0)
+	}
+
+	// 最后已有的日期处理完成后,需要对剩余不在时间段内的数据做处理
+	if len(weekDayDataList) > 0 {
+		// 当前时间段内的数据计算,得出实际值
+		var currVal float64
+		lenWeekDayDataList := len(weekDayDataList)
+		// 如果这个时间区间内没有数据,那么就采用上一个时间区间的值
+		if len(weekDayDataList) < 0 {
+			currVal = lastValue
+		} else {
+			if formula == "期末值" {
+				currVal = weekDayDataList[lenWeekDayDataList-1]
+			} else {
+				// 平均值
+				sumValDeci := decimal.NewFromFloat(0)
+				for _, v := range weekDayDataList {
+					tmpValDeci := decimal.NewFromFloat(v)
+					sumValDeci = sumValDeci.Add(tmpValDeci)
+				}
+				lenDeci := decimal.NewFromInt(int64(lenWeekDayDataList))
+				currVal, _ = sumValDeci.Div(lenDeci).Truncate(4).Float64()
+			}
+		}
+
+		// 判断降频指标是否存在数据
+		if existData, ok := existDataMap[nextEndDate.Format(utils.FormatDate)]; ok {
+			// 处理降频数据的值
+			existValStr := existData.Value
+			existValDeci, tmpErr := decimal.NewFromString(existValStr)
+			if tmpErr != nil {
+				err = tmpErr
+				return
+			}
+			existVal, _ := existValDeci.Truncate(4).Float64()
+			// 判断降频数据的值 与 当前计算出来的结果, 如果两个数据结果不相等的话,那么就修改咯
+			if existVal != currVal {
+				err = ModifyEdbDataById(source, existData.EdbDataId, fmt.Sprint(currVal))
+				if err != nil {
+					return err
+				}
+			}
+		} else {
+			// 直接入库
+			timestamp := nextEndDate.UnixNano() / 1e6
+			timestampStr := fmt.Sprintf("%d", timestamp)
+			addSql += GetAddSql(edbInfoIdStr, edbCode, nextEndDate.Format(utils.FormatDate), timestampStr, fmt.Sprint(currVal))
+			isAdd = true
+		}
+	}
+
+	if isAdd {
+		addSql = strings.TrimRight(addSql, ",")
+		_, err = to.Raw(addSql).Exec()
+	}
+
+	return
+}