package models import ( "errors" "eta/eta_index_lib/utils" "fmt" "github.com/beego/beego/v2/client/orm" "github.com/shopspring/decimal" "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 = utils.DATA_SOURCE_NAME_CALCULATE_JP 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.EdbNameEn = req.EdbName edbInfo.UnitEn = req.Unit 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 calculateMappingItem.FromSubSource = fromEdbInfo.SubSource _, 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, utils.DATA_SUB_SOURCE_EDB) 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, edbInfo.SubSource, fromEdbInfo, edbInfo.EdbCode, edbInfo.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.EdbNameEn = req.EdbNameEn edbInfo.UnitEn = req.UnitEn edbInfo.ModifyTime = time.Now() _, err = to.Update(edbInfo, "EdbName", "EdbNameSource", "Frequency", "Unit", "ClassifyId", "CalculateFormula", "ModifyTime", "EdbNameEn", "UnitEn") 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, edbInfo.SubSource) 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: edbInfo.Source, SourceName: edbInfo.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(), FromSubSource: fromEdbInfo.SubSource, } _, err = to.Insert(calculateMappingItem) if err != nil { return } } //计算数据 err = refreshAllCalculateJp(to, edbInfo.EdbInfoId, edbInfo.Source, edbInfo.SubSource, fromEdbInfo, edbInfo.EdbCode, edbInfo.Frequency, edbInfo.CalculateFormula) return } func RefreshAllCalculateJp(edbInfoId, source, subSource 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, subSource, fromEdbInfo, edbCode, edbFrequency, formula) return } // refreshAllCalculateJp 刷新降频数据 func refreshAllCalculateJp(to orm.TxOrmer, edbInfoId, source, subSource int, fromEdbInfo *EdbInfo, edbCode, edbFrequency, formula string) (err error) { edbInfoIdStr := strconv.Itoa(edbInfoId) //计算数据 //获取来源指标的数据 dataList, err := GetEdbDataListAllByTo(to, fromEdbInfo.Source, fromEdbInfo.SubSource, FindEdbDataListAllCond{ EdbInfoId: fromEdbInfo.EdbInfoId, }, 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, subSource) if err != nil { return } //计算指标的map existDataMap := make(map[string]*EdbData) // 已经入库的日期map existDelDateMap := make(map[string]string) //existDataMap := make(map[string]string, 0) for _, v := range existDataList { existDataMap[v.DataTime] = v existDelDateMap[v.DataTime] = v.DataTime } tableName := GetEdbDataTableName(utils.DATA_SOURCE_CALCULATE_JP, subSource) 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) nextEndDate := utils.GetFrequencyEndDay(startDataTime, edbFrequency) // 下一个节点的日期 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.Round(4).Float64() weekDayDataList = append(weekDayDataList, tmpValueFloat) } // 日期处理过滤 switch edbFrequency { case "周度": if tmpStartDataTime.Weekday() != 5 { //不是周五,代表需要进入下一个循环获取数据并计算 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 } // 本期的数据值 lenWeekDayDataList := len(weekDayDataList) if lenWeekDayDataList <= 0 { continue } // 当前时间段内的数据计算,得出实际值 var currVal float64 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).Round(4).Float64() } tmpStartDataTimeStr := tmpStartDataTime.Format(utils.FormatDate) // 判断降频指标是否存在数据 if existData, ok := existDataMap[tmpStartDataTimeStr]; ok { // 处理降频数据的值 existValStr := existData.Value existValDeci, tmpErr := decimal.NewFromString(existValStr) if tmpErr != nil { err = tmpErr return } existVal, _ := existValDeci.Round(4).Float64() // 判断降频数据的值 与 当前计算出来的结果, 如果两个数据结果不相等的话,那么就修改咯 if existVal != currVal { err = ModifyEdbDataById(source, subSource, existData.EdbDataId, fmt.Sprint(currVal)) if err != nil { return err } } // 移除待删除的日期 delete(existDelDateMap, tmpStartDataTimeStr) } else { // 直接入库 timestamp := tmpStartDataTime.UnixNano() / 1e6 timestampStr := fmt.Sprintf("%d", timestamp) addSql += GetAddSql(edbInfoIdStr, edbCode, tmpStartDataTimeStr, timestampStr, fmt.Sprint(currVal)) isAdd = true // 移除待删除的日期 delete(existDelDateMap, tmpStartDataTimeStr) } // 一轮结束后,数据清空 weekDayDataList = make([]float64, 0) } // 最后已有的日期处理完成后,需要对剩余不在时间段内的数据做处理 lenWeekDayDataList := len(weekDayDataList) if lenWeekDayDataList > 0 { // 当前时间段内的数据计算,得出实际值 var currVal float64 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).Round(4).Float64() } nextEndDateStr := nextEndDate.Format(utils.FormatDate) // 判断降频指标是否存在数据 if existData, ok := existDataMap[nextEndDateStr]; ok { // 处理降频数据的值 existValStr := existData.Value existValDeci, tmpErr := decimal.NewFromString(existValStr) if tmpErr != nil { err = tmpErr return } existVal, _ := existValDeci.Round(4).Float64() // 判断降频数据的值 与 当前计算出来的结果, 如果两个数据结果不相等的话,那么就修改咯 if existVal != currVal { err = ModifyEdbDataById(source, subSource, existData.EdbDataId, fmt.Sprint(currVal)) if err != nil { return err } } // 移除待删除的日期 delete(existDelDateMap, nextEndDateStr) } else { // 直接入库 timestamp := nextEndDate.UnixNano() / 1e6 timestampStr := fmt.Sprintf("%d", timestamp) addSql += GetAddSql(edbInfoIdStr, edbCode, nextEndDate.Format(utils.FormatDate), timestampStr, fmt.Sprint(currVal)) isAdd = true // 移除待删除的日期 delete(existDelDateMap, nextEndDate.Format(utils.FormatDate)) } } if isAdd { addSql = strings.TrimRight(addSql, ",") _, err = to.Raw(addSql).Exec() } // 移除不存在的日期数据 if len(existDelDateMap) > 0 { removeDateList := make([]string, 0) //需要移除的日期 for k := range existDelDateMap { 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 }