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" ) // Ljz 累计值 type Ljz struct { } // Add 添加 func (obj Ljz) Add(params AddCalculateBatchParams) (edbInfo *EdbInfo, err error, errMsg string) { req := params.Req fromEdbInfo := params.FromEdbInfo edbCode := params.EdbCode frequencyInt := utils.CheckFrequency(fromEdbInfo.Frequency, req.Frequency) if frequencyInt < 0 { errMsg = "频度异常,不允许低频转为高频" err = errors.New(errMsg) return } if frequencyInt == 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(), ";Add,Err:"+err.Error()) } else { _ = to.Commit() } }() edbInfo = &EdbInfo{ //EdbInfoId: 0, SourceName: obj.GetSourceName(), Source: obj.GetSource(), EdbCode: edbCode, EdbName: req.EdbName, EdbNameSource: req.EdbName, Frequency: req.Frequency, Unit: req.Unit, StartDate: "", EndDate: "", ClassifyId: req.ClassifyId, SysUserId: params.SysUserId, SysUserRealName: params.SysUserRealName, UniqueCode: params.UniqueCode, CreateTime: time.Now(), ModifyTime: time.Now(), MinValue: 0, MaxValue: 0, CalculateFormula: req.Formula, EdbType: obj.GetEdbType(), Sort: 0, MoveType: 0, MoveFrequency: "", NoUpdate: 0, ServerUrl: "", EdbInfoType: 0, EdbNameEn: "", UnitEn: "", LatestDate: "", LatestValue: 0, ChartImage: "", Calendar: "", Extra: req.Extra, } 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 calculateMappingItem.FromSubSource = fromEdbInfo.SubSource _, err = to.Insert(calculateMappingItem) if err != nil { return } } //计算数据 err = obj.refresh(to, edbInfo.EdbInfoId, edbInfo.Source, edbInfo.SubSource, edbInfo, fromEdbInfo, edbInfo.EdbCode, "", edbInfo.Extra) return } // Edit 编辑 func (obj Ljz) Edit(params EditCalculateBatchParams) (err error, errMsg string) { req := params.Req edbInfo := params.EdbInfo fromEdbInfo := params.FromEdbInfo frequencyInt := utils.CheckFrequency(fromEdbInfo.Frequency, req.Frequency) if frequencyInt < 0 { errMsg = "频度异常,不允许低频转为高频" err = errors.New(errMsg) return } if frequencyInt == 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(), ";Edit,Err:"+err.Error()) } else { _ = to.Commit() } }() tableName := GetEdbDataTableName(edbInfo.Source, edbInfo.SubSource) var isRecalculate bool if edbInfo.Frequency != req.Frequency || edbInfo.Extra != req.Extra { isRecalculate = true } //修改指标信息 edbInfo.EdbName = req.EdbName edbInfo.EdbNameSource = req.EdbName edbInfo.Frequency = req.Frequency edbInfo.Unit = req.Unit edbInfo.ClassifyId = req.ClassifyId edbInfo.Extra = req.Extra edbInfo.ModifyTime = time.Now() _, err = to.Update(edbInfo, "EdbName", "EdbNameSource", "Frequency", "Unit", "ClassifyId", "ModifyTime", "Extra") 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 { // 指标未被替换,无需删除关联数据 // todo 频度被换了,需要重新计算 if isRecalculate { err = obj.refresh(to, edbInfo.EdbInfoId, edbInfo.Source, edbInfo.SubSource, edbInfo, fromEdbInfo, edbInfo.EdbCode, "", req.Extra) } return } //删除,计算指标关联的,基础指标的关联关系 sql := ` DELETE FROM edb_info_calculate_mapping WHERE edb_info_id = ? ` _, err = to.Raw(sql, edbInfo.EdbInfoId).Exec() if err != nil { return } //清空原有数据 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: obj.GetSource(), SourceName: obj.GetSourceName(), 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(), FromSubSource: fromEdbInfo.SubSource, } _, err = to.Insert(calculateMappingItem) if err != nil { return } } //计算数据 err = obj.refresh(to, edbInfo.EdbInfoId, edbInfo.Source, edbInfo.SubSource, edbInfo, fromEdbInfo, edbInfo.EdbCode, "", edbInfo.Extra) return } // Refresh 刷新 func (obj Ljz) Refresh(params RefreshParams) (err error, errMsg string) { calculateMapping, err := GetEdbInfoCalculateMappingDetail(params.EdbInfo.EdbInfoId) if err != nil { errMsg = "GetEdbInfoCalculateLjzDetail Err:" + err.Error() return } fromEdbInfo, err := GetEdbInfoById(calculateMapping.FromEdbInfoId) if err != nil { errMsg = "GetEdbInfoById Err:" + err.Error() 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() } }() // 计算数据 err = obj.refresh(to, params.EdbInfo.EdbInfoId, params.EdbInfo.Source, params.EdbInfo.SubSource, params.EdbInfo, fromEdbInfo, params.EdbInfo.EdbCode, params.StartDate, params.EdbInfo.Extra) return } // GetSource 获取来源编码id func (obj Ljz) GetSource() int { return utils.DATA_SOURCE_CALCULATE_LJZ } // GetSourceName 获取来源名称 func (obj Ljz) GetSourceName() string { return utils.DATA_SOURCE_NAME_CALCULATE_LJZ } // GetEdbType 获取指标类型 func (obj Ljz) GetEdbType() int { return utils.CALCULATE_EDB_TYPE } func (obj Ljz) refresh(to orm.TxOrmer, edbInfoId, source, subSource int, edbInfo, fromEdbInfo *EdbInfo, edbCode, startDate string, extra string) (err error) { dataTableName := GetEdbDataTableName(source, subSource) 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) //} var isWeekData bool // 是否周度数据,如果是周度数据的话,是需要变频的,最后结果还需要除以7 // 周度数据需要先变成日度的 if fromEdbInfo.Frequency == `周度` { isWeekData = true } dataList, err := GetEdbDataListAllByTo(to, condition, pars, fromEdbInfo.Source, fromEdbInfo.SubSource, 1) if err != nil { return err } fromEdbDataMap := make(map[string]float64) if isWeekData { dataList, err = HandleDataByLinearRegression(dataList, fromEdbDataMap) if err != nil { return } } var lastValType int if extra != "" { var lastValConfig CalculateLjzEdbExtra err = json.Unmarshal([]byte(extra), &lastValConfig) if err != nil { err = fmt.Errorf("refreshAllCalculate,extra解析失败,Err:%s", err.Error()) return } lastValType = lastValConfig.LastValType } //日度转周度:日期选周五,计算上周六到本周五的日度值的加总,最新日期为最新值对应的周五。 //日度转旬度:日期选每个旬的最后一天,计算当旬所有日度值的加总,最新日期为最新值对应当旬的最后一天。 //日度转月度:日期选每个月最后一天,计算当月所有日度值的加总,最新日期为最新值对应当月最后一天。 //日度转季度、年度:方法类似转月度。 //周度转月度/季度/年度:将周度值转成日度,空值用插值法插值,计算当月/当季/当年所有值的加总,然后除以7。 //月度转季度/年度: 当季/当年月度值相加。 dateList := make([]time.Time, 0) valueMap := make(map[time.Time]float64) switch edbInfo.Frequency { case "年度": yearMap := make(map[int]float64) yearList := make([]int, 0) var lastNewDate time.Time for _, item := range dataList { itemDate, tmpErr := time.ParseInLocation(utils.FormatDate, item.DataTime, time.Local) if tmpErr != nil { err = tmpErr return } year := itemDate.Year() yearVal, ok := yearMap[year] if ok { yearMap[year] = item.Value + yearVal } else { yearList = append(yearList, year) yearMap[year] = item.Value } } for _, v := range yearList { currTime := time.Date(v, 12, 31, 0, 0, 0, 0, time.Local) dateList = append(dateList, currTime) valueMap[currTime] = yearMap[v] lastNewDate = currTime } // 根据配置处理最新值, 1 表示均值填充 if lastValType == 1 { lastItem := dataList[len(dataList)-1] // 最后一天的累计值 initVal := valueMap[lastNewDate] itemDate, tmpErr := time.ParseInLocation(utils.FormatDate, lastItem.DataTime, time.Local) if tmpErr != nil { err = tmpErr return } //获取当季的总天数, 当季最后一天减去当季第一天 // 获取当年的总天数 startDateT := time.Date(itemDate.Year(), 1, 1, 0, 0, 0, 0, time.Local) allDays := int(lastNewDate.Sub(startDateT).Hours() / 24) //获取距离当年第一天的天数 days := int(itemDate.Sub(startDateT).Hours() / 24) daysT := decimal.NewFromInt(int64(days)) initValAndMonthT := decimal.NewFromFloat(initVal * float64(allDays)) val, _ := initValAndMonthT.Div(daysT).RoundCeil(4).Float64() fmt.Printf("最新值 计算公式:%d*%f/(%d) = %f\n", allDays, initVal, days, val) valueMap[lastNewDate] = val } case "半年度": yearMonthMap := make(map[string]float64) yearMonthList := make([]string, 0) var lastNewDate time.Time for _, item := range dataList { itemDate, tmpErr := time.ParseInLocation(utils.FormatDate, item.DataTime, time.Local) if tmpErr != nil { err = tmpErr return } year := itemDate.Year() var tmpK string if itemDate.Month() <= 6 { tmpK = fmt.Sprint(year, "06") } else { tmpK = fmt.Sprint(year, "12") } yearVal, ok := yearMonthMap[tmpK] if ok { yearMonthMap[tmpK] = item.Value + yearVal } else { yearMonthList = append(yearMonthList, tmpK) yearMonthMap[tmpK] = item.Value } } for _, v := range yearMonthList { currTime, tmpErr := time.ParseInLocation(utils.FormatYearMonthUnSpace, v, time.Local) if tmpErr != nil { err = tmpErr return } currTime = currTime.AddDate(0, 1, -1) dateList = append(dateList, currTime) valueMap[currTime] = yearMonthMap[v] lastNewDate = currTime } // 根据配置处理最新值, 1 表示均值填充 if lastValType == 1 { lastItem := dataList[len(dataList)-1] // 最后一天的累计值 initVal := valueMap[lastNewDate] itemDate, tmpErr := time.ParseInLocation(utils.FormatDate, lastItem.DataTime, time.Local) if tmpErr != nil { err = tmpErr return } //获取上半年或者下半年的总天数, 半年度最后一天减去半年度第一天 startDateT := lastNewDate.AddDate(0, -6, 0) allDays := int(lastNewDate.Sub(startDateT).Hours() / 24) //获取距离半年度第一天的天数 days := int(itemDate.Sub(startDateT).Hours() / 24) daysT := decimal.NewFromInt(int64(days)) initValAndMonthT := decimal.NewFromFloat(initVal * float64(allDays)) val, _ := initValAndMonthT.Div(daysT).RoundCeil(4).Float64() fmt.Printf("最新值 计算公式:%d*%f/(%d) = %f\n", allDays, initVal, days, val) valueMap[lastNewDate] = val } case "季度": yearMonthMap := make(map[string]float64) yearMonthList := make([]string, 0) var lastNewDate time.Time for _, item := range dataList { itemDate, tmpErr := time.ParseInLocation(utils.FormatDate, item.DataTime, time.Local) if tmpErr != nil { err = tmpErr return } year := itemDate.Year() var tmpK string if itemDate.Month() <= 3 { tmpK = fmt.Sprint(year, "03") } else if itemDate.Month() <= 6 { tmpK = fmt.Sprint(year, "06") } else if itemDate.Month() <= 9 { tmpK = fmt.Sprint(year, "09") } else { tmpK = fmt.Sprint(year, "12") } yearVal, ok := yearMonthMap[tmpK] if ok { yearMonthMap[tmpK] = item.Value + yearVal } else { yearMonthList = append(yearMonthList, tmpK) yearMonthMap[tmpK] = item.Value } } for _, v := range yearMonthList { currTime, tmpErr := time.ParseInLocation(utils.FormatYearMonthUnSpace, v, time.Local) if tmpErr != nil { err = tmpErr return } currTime = currTime.AddDate(0, 1, -1) dateList = append(dateList, currTime) valueMap[currTime] = yearMonthMap[v] lastNewDate = currTime } // 根据配置处理最新值, 1 表示均值填充 if lastValType == 1 { lastItem := dataList[len(dataList)-1] // 最后一天的累计值 initVal := valueMap[lastNewDate] itemDate, tmpErr := time.ParseInLocation(utils.FormatDate, lastItem.DataTime, time.Local) if tmpErr != nil { err = tmpErr return } //获取当季的总天数, 当季最后一天减去当季第一天 startDateT := lastNewDate.AddDate(0, -3, 0) allDays := int(lastNewDate.Sub(startDateT).Hours() / 24) //获取距离当季第一天的天数 days := int(itemDate.Sub(startDateT).Hours() / 24) daysT := decimal.NewFromInt(int64(days)) initValAndMonthT := decimal.NewFromFloat(initVal * float64(allDays)) val, _ := initValAndMonthT.Div(daysT).RoundCeil(4).Float64() fmt.Printf("最新值 计算公式:%d*%f/(%d) = %f\n", allDays, initVal, days, val) valueMap[lastNewDate] = val } case "月度": yearMonthMap := make(map[string]float64) yearMonthList := make([]string, 0) var lastNewDate time.Time for _, item := range dataList { itemDate, tmpErr := time.ParseInLocation(utils.FormatDate, item.DataTime, time.Local) if tmpErr != nil { err = tmpErr return } year := itemDate.Year() var tmpK string tmpK = fmt.Sprint(year*100 + int(itemDate.Month())) yearVal, ok := yearMonthMap[tmpK] if ok { yearMonthMap[tmpK] = item.Value + yearVal fmt.Printf("%s 累加值:%f = %.6f + %.6f\n", tmpK, yearMonthMap[tmpK], item.Value, yearVal) } else { yearMonthList = append(yearMonthList, tmpK) yearMonthMap[tmpK] = item.Value } } for _, v := range yearMonthList { currTime, tmpErr := time.ParseInLocation(utils.FormatYearMonthUnSpace, v, time.Local) if tmpErr != nil { err = tmpErr return } currTime = currTime.AddDate(0, 1, -1) dateList = append(dateList, currTime) valueMap[currTime] = yearMonthMap[v] lastNewDate = currTime } // 根据配置处理最新值, 1 表示均值填充 if lastValType == 1 { lastItem := dataList[len(dataList)-1] // 最后一天的累计值 initVal := valueMap[lastNewDate] itemDate, tmpErr := time.ParseInLocation(utils.FormatDate, lastItem.DataTime, time.Local) if tmpErr != nil { err = tmpErr return } //获取当月的天数 monthDays := lastNewDate.Day() //获取自然日的天数 days := itemDate.Day() daysT := decimal.NewFromInt(int64(days)) initValAndMonthT := decimal.NewFromFloat(initVal * float64(monthDays)) val, _ := initValAndMonthT.Div(daysT).RoundCeil(4).Float64() fmt.Printf("最新值 计算公式:%d*%f/(%d) = %f\n", monthDays, initVal, days, val) valueMap[lastNewDate] = val } case "旬度": tmpDateDataMap := make(map[time.Time]float64) tmpDateList := make([]time.Time, 0) for _, item := range dataList { itemDate, tmpErr := time.ParseInLocation(utils.FormatDate, item.DataTime, time.Local) if tmpErr != nil { err = tmpErr return } dayInt := itemDate.Year()*100 + int(itemDate.Month()) var currTime time.Time if itemDate.Day() <= 10 { //本月上旬 tmpK := fmt.Sprint(dayInt, "10") currTime, err = time.ParseInLocation(utils.FormatDateUnSpace, tmpK, time.Local) if err != nil { return } } else if itemDate.Day() <= 20 { // 本月中旬 tmpK := fmt.Sprint(dayInt, "20") currTime, err = time.ParseInLocation(utils.FormatDateUnSpace, tmpK, time.Local) if err != nil { return } } else { // 本月下旬 currTime, err = time.ParseInLocation(utils.FormatYearMonthUnSpace, fmt.Sprint(dayInt), time.Local) if err != nil { return } currTime = currTime.AddDate(0, 1, -1) } yearVal, ok := tmpDateDataMap[currTime] if ok { tmpDateDataMap[currTime] = item.Value + yearVal } else { tmpDateList = append(tmpDateList, currTime) tmpDateDataMap[currTime] = item.Value } } for _, currTime := range tmpDateList { dateList = append(dateList, currTime) valueMap[currTime] = tmpDateDataMap[currTime] } case "周度": tmpDateDataMap := make(map[time.Time]float64) tmpDateList := make([]time.Time, 0) for _, item := range dataList { itemDate, tmpErr := time.ParseInLocation(utils.FormatDate, item.DataTime, time.Local) if tmpErr != nil { err = tmpErr return } var currTime time.Time // 周六周日,这是下一个周五的数据 if itemDate.Weekday() == 0 { currTime = itemDate.AddDate(0, 0, 5) } else if itemDate.Weekday() == 6 { currTime = itemDate.AddDate(0, 0, 6) } else { currTime = itemDate.AddDate(0, 0, 5-int(itemDate.Weekday())) } yearVal, ok := tmpDateDataMap[currTime] if ok { tmpDateDataMap[currTime] = item.Value + yearVal } else { tmpDateList = append(tmpDateList, currTime) tmpDateDataMap[currTime] = item.Value } } for _, currTime := range tmpDateList { dateList = append(dateList, currTime) valueMap[currTime] = tmpDateDataMap[currTime] } } //获取指标所有数据 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 err } 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 _, currTime := range dateList { currDateStr := currTime.Format(utils.FormatDate) existVal, ok := existDataMap[currDateStr] tmpVal, ok2 := valueMap[currTime] if !ok2 { err = errors.New("数据异常,date:" + currDateStr) return } var saveValue string if isWeekData { //周度指标转的话,最后结果要除以7 saveValue = decimal.NewFromFloat(tmpVal).Div(decimal.NewFromInt(7)).Round(4).String() } else { saveValue = decimal.NewFromFloat(tmpVal).Round(4).String() } // 如果库中已经存在该数据的话,那么就进行值的变更操作 if ok { //校验待删除日期数据里面是否存在该元素,如果存在的话,那么移除该日期 delete(removeDataTimeMap, currDateStr) 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 } } return }