package models import ( "errors" "eta/eta_index_lib/utils" "fmt" "github.com/beego/beego/v2/client/orm" "github.com/shopspring/decimal" "strconv" "strings" "time" ) // AddCalculateBp 升频 func AddCalculateBp(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("AddCalculateBp,Err:" + err.Error()) _ = to.Rollback() } else { _ = to.Commit() } }() if req.EdbInfoId <= 0 { edbInfo = new(EdbInfo) edbInfo.Source = utils.DATA_SOURCE_CALCULATE_BP edbInfo.SourceName = utils.DATA_SOURCE_NAME_CALCULATE_BP 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 = edbInfo.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_BP, 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 = refreshAllCalculateBp(to, edbInfo.EdbInfoId, edbInfo.Source, edbInfo.SubSource, fromEdbInfo, edbInfo.EdbCode, "", "", 0) return } // EditCalculateBp 修改升频数据 func EditCalculateBp(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("EditCalculateBp,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.EdbNameEn = req.EdbNameEn edbInfo.UnitEn = req.UnitEn edbInfo.ModifyTime = time.Now() _, err = to.Update(edbInfo, "EdbName", "EdbNameSource", "Frequency", "Unit", "ClassifyId", "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 } //清空原有数据 sql = ` DELETE FROM edb_data_calculate_bp 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_BP, SourceName: utils.DATA_SOURCE_NAME_CALCULATE_BP, 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 = refreshAllCalculateBp(to, edbInfo.EdbInfoId, edbInfo.Source, edbInfo.SubSource, fromEdbInfo, edbInfo.EdbCode, "", "", 0) return } // RefreshAllCalculateBpBak 刷新所有升频数据 func RefreshAllCalculateBpBak(edbInfoId, source, subSource int, fromEdbInfo *EdbInfo, edbCode, startDate, endDate string) (err error) { o := orm.NewOrm() to, err := o.Begin() if err != nil { return } defer func() { if err != nil { fmt.Println("RefreshAllCalculateBp,Err:" + err.Error()) _ = to.Rollback() } else { _ = to.Commit() } }() if err != nil { return } edbInfoIdStr := strconv.Itoa(edbInfoId) //计算数据 dataList, err := GetEdbDataListAllByTo(to, fromEdbInfo.Source, fromEdbInfo.SubSource, FindEdbDataListAllCond{ EdbInfoId: fromEdbInfo.EdbInfoId, StartDataTime: startDate, StartDataTimeCond: ">=", EndDataTime: endDate, EndDataTimeCond: "<=", }, 0) if err != nil { return err } var dateArr []string dataMap := make(map[string]*EdbInfoSearchData) for _, v := range dataList { dateArr = append(dateArr, v.DataTime) dataMap[v.DataTime] = v } fmt.Println("source:", source) //获取指标所有数据 existDataList := make([]*EdbData, 0) dataTableName := GetEdbDataTableName(source, subSource) fmt.Println("dataTableName:", dataTableName) 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) for _, v := range existDataList { existDataMap[v.DataTime] = v.Value } fmt.Println("existDataMap:", existDataMap) addSql := ` INSERT INTO edb_data_calculate_bp(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) for i := 0; i < dataLen; i++ { //当期 currentItem := dataList[i] currentDate, _ := time.ParseInLocation(utils.FormatDate, currentItem.DataTime, time.Local) var day int var preItem *EdbInfoSearchData var preDate time.Time if i == 0 { day = int(time.Now().Sub(currentDate).Hours() / float64(24)) preDate = time.Now() } else { j := i + 1 if j < dataLen { preItem = dataList[j] preDate, _ = time.ParseInLocation(utils.FormatDate, preItem.DataTime, time.Local) day = int(currentDate.Sub(preDate).Hours() / float64(24)) } } for k := 0; k <= day; k++ { needDay := preDate.AddDate(0, 0, -k) needDayStr := needDay.Format(utils.FormatDate) existKey := edbCode + needDayStr if _, ok := existMap[existKey]; !ok { timestamp := needDay.UnixNano() / 1e6 timestampStr := fmt.Sprintf("%d", timestamp) valStr := decimal.NewFromFloat(currentItem.Value).String() if existVal, ok := existDataMap[needDayStr]; !ok { addSql += GetAddSql(edbInfoIdStr, edbCode, needDayStr, timestampStr, valStr) } else { if existVal != valStr { sql := ` UPDATE %s SET value=?,modify_time=NOW() WHERE edb_info_id=? AND data_time=? ` sql = fmt.Sprintf(sql, dataTableName) _, err = to.Raw(sql, valStr, edbInfoId, needDay).Exec() if err != nil { return err } } } } existMap[existKey] = needDayStr } existKey := edbCode + currentItem.DataTime if _, ok := existMap[existKey]; !ok { currentDate, _ := time.ParseInLocation(utils.FormatDate, currentItem.DataTime, time.Local) timestamp := currentDate.UnixNano() / 1e6 timestampStr := fmt.Sprintf("%d", timestamp) valStr := decimal.NewFromFloat(currentItem.Value).String() if existVal, ok := existDataMap[currentItem.DataTime]; !ok { addSql += GetAddSql(edbInfoIdStr, edbCode, currentItem.DataTime, timestampStr, valStr) } else { if existVal != valStr { sql := ` UPDATE %s SET value=?,modify_time=NOW() WHERE edb_info_id=? AND data_time=? ` sql = fmt.Sprintf(sql, dataTableName) _, err = to.Raw(sql, valStr, edbInfoId, currentItem.DataTime).Exec() if err != nil { return err } } } } existMap[existKey] = currentItem.DataTime } if isAdd { addSql = strings.TrimRight(addSql, ",") _, err = to.Raw(addSql).Exec() if err != nil { return err } } return } func RefreshAllCalculateBp(edbInfoId, source, subSource int, fromEdbInfo *EdbInfo, edbCode, startDate, endDate string) (err error) { o := orm.NewOrm() to, err := o.Begin() if err != nil { return } defer func() { if err != nil { fmt.Println("RefreshAllCalculateBp,Err:" + err.Error()) _ = to.Rollback() } else { _ = to.Commit() } }() // 计算数据 err = refreshAllCalculateBp(to, edbInfoId, source, subSource, fromEdbInfo, edbCode, startDate, endDate, 1) return } // refreshAllCalculateBp 刷新升频数据 func refreshAllCalculateBp(to orm.TxOrmer, edbInfoId, source, subSource int, fromEdbInfo *EdbInfo, edbCode, startDate, endDate string, order int) (err error) { edbInfoIdStr := strconv.Itoa(edbInfoId) //计算数据 //获取来源指标的数据 dataList, err := GetEdbDataListAllByTo(to, fromEdbInfo.Source, fromEdbInfo.SubSource, FindEdbDataListAllCond{ EdbInfoId: fromEdbInfo.EdbInfoId, }, order) if err != nil { return err } // 来源指标没有数据,那么需要删除所有的计算指标数据 if len(dataList) <= 0 { // todo 删除所有的计算指标数据 return } // 来源指标的第一个日期 fromFirstDate, err := time.ParseInLocation(utils.FormatDate, dataList[0].DataTime, time.Local) if err != nil { return } fromFirstDate = time.Date(fromFirstDate.Year(), fromFirstDate.Month(), fromFirstDate.Day(), 0, 0, 0, 0, time.Local) // 变频计算 newDataList, err := EdbInfoSearchDataToData(dataList) if err != nil { return } baseCalculate := BaseCalculate{ DataList: newDataList, Frequency: "", Formula: nil, Calendar: "", MoveType: 0, MoveFrequency: "", FromFrequency: "", Source: source, } dateDataMap, err, _ := baseCalculate.UpFrequency() if err != nil { return } // 获取升频指所有已经存在的计算指标数据 existDataList, err := GetAllEdbDataListByTo(to, edbInfoId, source, subSource) if err != nil { return } //计算指标的map existDataMap := make(map[string]*EdbData, 0) for _, v := range existDataList { existDataMap[v.DataTime] = v } addSql := ` INSERT INTO edb_data_calculate_bp(edb_info_id,edb_code,data_time,value,create_time,modify_time,data_timestamp) values ` var isAdd bool now := time.Date(time.Now().Year(), time.Now().Month(), time.Now().Day(), 0, 0, 0, 0, time.Local) for currDate := fromFirstDate; !currDate.After(now); currDate = currDate.AddDate(0, 0, 1) { currDateStr := currDate.Format(utils.FormatDate) timestamp := currDate.UnixNano() / 1e6 timestampStr := fmt.Sprintf("%d", timestamp) // 当前计算的值 currValue, ok := dateDataMap[currDate] if !ok { // 没有计算成功就过滤 continue } lastValueStr := decimal.NewFromFloat(currValue).Round(4).String() // 已经入库的值 existData, ok := existDataMap[currDateStr] if !ok { // 没有入库那么就插入添加 isAdd = true addSql += GetAddSql(edbInfoIdStr, edbCode, currDateStr, timestampStr, lastValueStr) continue } // 将已经入库的值转换为decimal类型,然后再保留4位小数,目的是为了做匹配,要不然取出来的数据与计算的数据不一致 existDataValueDec, tmpErr := decimal.NewFromString(existData.Value) if tmpErr != nil { err = tmpErr return } existDataValueStr := existDataValueDec.Round(4).String() // 如果该日期已经入库了,且两个值不匹配,那么就更新 if lastValueStr != existDataValueStr { err = ModifyEdbDataById(source, subSource, existData.EdbDataId, lastValueStr) if err != nil { return err } } // 该日期已经处理过了,所以需要移除,如果后面该map还有数据,那么需要删除该map里面的日期数据 delete(existDataMap, currDateStr) } // 删除不需要的指标数据 if len(existDataMap) > 0 { //待删除的日期 removeDateList := make([]string, 0) for date := range existDataMap { removeDateList = append(removeDateList, date) } removeDateStr := strings.Join(removeDateList, `","`) removeDateStr = `"` + removeDateStr + `"` //如果拼接指标变更了,那么需要删除所有的指标数据 tableName := GetEdbDataTableName(source, subSource) 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 } } // 新增的数据值 if isAdd { addSql = strings.TrimRight(addSql, ",") _, err = to.Raw(addSql).Exec() } return }