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" ) type CalculateAvg struct { } // Add 新增 func (obj CalculateAvg) Add(params AddCalculateBatchParams) (edbInfo *EdbInfo, err error, errMsg string) { req := params.Req edbCode := params.EdbCode o := orm.NewOrm() to, err := o.Begin() if err != nil { return } defer func() { if err != nil { fmt.Println("CalculateAvg.Add,Err:" + err.Error()) _ = to.Rollback() } else { _ = to.Commit() } }() if req.EdbInfoId > 0 { err = errors.New("无法新增") return } edbInfo = new(EdbInfo) edbInfo.Source = obj.GetSource() edbInfo.SourceName = obj.GetSourceName() 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 = params.SysUserId edbInfo.SysUserRealName = params.SysUserRealName edbInfo.CreateTime = time.Now() edbInfo.ModifyTime = time.Now() edbInfo.UniqueCode = params.UniqueCode edbInfo.EdbType = obj.GetEdbType() edbInfo.EmptyType = req.EmptyType edbInfo.MaxEmptyType = req.MaxEmptyType edbInfo.Extra = req.Extra edbInfo.EdbNameEn = req.EdbName edbInfo.UnitEn = req.Unit newEdbInfoId, tmpErr := to.Insert(edbInfo) if tmpErr != nil { err = tmpErr return } edbInfo.EdbInfoId = int(newEdbInfoId) //关联关系 tagMap := make(map[string]int) relationEdbInfoList := make([]*EdbInfo, 0) calculateMappingItemList := make([]*EdbInfoCalculateMapping, 0) for k, v := range req.EdbInfoIdArr { tmpEdbInfo, e := GetEdbInfoById(v.EdbInfoId) if e != nil { err = e return } relationEdbInfoList = append(relationEdbInfoList, tmpEdbInfo) calculateMappingItem := new(EdbInfoCalculateMapping) calculateMappingItem.CreateTime = time.Now() calculateMappingItem.ModifyTime = time.Now() calculateMappingItem.Sort = k calculateMappingItem.EdbCode = edbCode calculateMappingItem.EdbInfoId = edbInfo.EdbInfoId calculateMappingItem.FromEdbInfoId = tmpEdbInfo.EdbInfoId calculateMappingItem.FromEdbCode = tmpEdbInfo.EdbCode calculateMappingItem.FromEdbName = tmpEdbInfo.EdbName calculateMappingItem.FromSource = tmpEdbInfo.Source calculateMappingItem.FromSourceName = tmpEdbInfo.SourceName calculateMappingItem.FromTag = v.FromTag calculateMappingItem.Source = edbInfo.Source calculateMappingItem.SourceName = edbInfo.SourceName calculateMappingItem.FromSubSource = edbInfo.SubSource calculateMappingItemList = append(calculateMappingItemList, calculateMappingItem) tagMap[v.FromTag] = v.EdbInfoId } _, err = to.InsertMulti(len(calculateMappingItemList), calculateMappingItemList) if err != nil { return } //计算数据 err = obj.refresh(to, edbInfo.EdbInfoId, edbInfo.Source, edbInfo.SubSource, relationEdbInfoList, edbInfo.EdbCode, edbInfo.Extra, tagMap, edbInfo.EmptyType) return } func (obj CalculateAvg) Edit(params EditCalculateBatchParams) (err error, errMsg string) { req := params.Req edbInfo := params.EdbInfo o := orm.NewOrm() to, err := o.Begin() if err != nil { return } defer func() { if err != nil { fmt.Println("CalculateAvg.Edit,Err:" + err.Error()) _ = to.Rollback() } else { _ = to.Commit() } }() if len(req.EdbInfoIdArr) <= 0 { err = fmt.Errorf("请选择要关联的指标") return } needCalculate := false if edbInfo.EmptyType != req.EmptyType || edbInfo.Extra != req.Extra { needCalculate = true } //修改指标信息 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.Extra = req.Extra edbInfo.EmptyType = req.EmptyType edbInfo.MaxEmptyType = req.MaxEmptyType edbInfo.EdbNameEn = req.EdbNameEn edbInfo.UnitEn = req.UnitEn edbInfo.ModifyTime = time.Now() _, err = to.Update(edbInfo, "EdbName", "EdbNameSource", "Frequency", "Unit", "ClassifyId", "CalculateFormula", "Extra", "EmptyType", "MaxEmptyType", "ModifyTime", "EdbNameEn", "UnitEn") if err != nil { return } //判断是否重复指标 //edbInfoMap := make(map[int]string) if !needCalculate { //用到的指标 newEdbInfoIdArr := make([]string, 0) for _, tmpEdbInfo := range req.EdbInfoIdArr { /*_, ok := edbInfoMap[tmpEdbInfo.EdbInfoId] if ok { br.Msg = "选择指标失败,请勿选择重复指标!" return } edbInfoMap[tmpEdbInfo.EdbInfoId] = tmpEdbInfo.FromTag*/ newEdbInfoIdArr = append(newEdbInfoIdArr, strconv.Itoa(tmpEdbInfo.EdbInfoId)) } edbInfoIdStr := strings.Join(newEdbInfoIdArr, ",") oldEdbInfoIdStr, e := GetCalculateEdbInfo(req.EdbInfoId) if e != nil { errMsg = "获取数据失败,GetCalculateEdbInfo:" + e.Error() err = fmt.Errorf("查询关联指标失败") return } if edbInfoIdStr != oldEdbInfoIdStr { // 不需要重复计算 needCalculate = true } } fmt.Printf("needCalculate:%v\n", needCalculate) if !needCalculate { 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 } //关联关系 tagMap := make(map[string]int) relationEdbInfoList := make([]*EdbInfo, 0) calculateMappingItemList := make([]*EdbInfoCalculateMapping, 0) for k, v := range req.EdbInfoIdArr { tmpEdbInfo, tmpErr := GetEdbInfoById(v.EdbInfoId) if tmpErr != nil { err = tmpErr return } relationEdbInfoList = append(relationEdbInfoList, tmpEdbInfo) calculateMappingItem := new(EdbInfoCalculateMapping) calculateMappingItem.CreateTime = time.Now() calculateMappingItem.ModifyTime = time.Now() calculateMappingItem.Sort = k calculateMappingItem.EdbCode = edbInfo.EdbCode calculateMappingItem.EdbInfoId = edbInfo.EdbInfoId calculateMappingItem.FromEdbInfoId = tmpEdbInfo.EdbInfoId calculateMappingItem.FromEdbCode = tmpEdbInfo.EdbCode calculateMappingItem.FromEdbName = tmpEdbInfo.EdbName calculateMappingItem.FromSource = tmpEdbInfo.Source calculateMappingItem.FromSourceName = tmpEdbInfo.SourceName calculateMappingItem.FromTag = v.FromTag calculateMappingItem.Source = edbInfo.Source calculateMappingItem.SourceName = edbInfo.SourceName calculateMappingItem.FromSubSource = edbInfo.SubSource calculateMappingItemList = append(calculateMappingItemList, calculateMappingItem) tagMap[v.FromTag] = v.EdbInfoId } _, err = to.InsertMulti(len(calculateMappingItemList), calculateMappingItemList) if err != nil { return } //计算数据 err = obj.refresh(to, edbInfo.EdbInfoId, edbInfo.Source, edbInfo.SubSource, relationEdbInfoList, edbInfo.EdbCode, edbInfo.Extra, tagMap, edbInfo.EmptyType) return } func (obj CalculateAvg) Refresh(params RefreshParams) (err error, errMsg string) { edbInfo := params.EdbInfo edbInfoCalculateDetailList, err := GetEdbInfoCalculateDetailList(edbInfo.EdbInfoId) if err != nil { return } tagMap := make(map[string]int) relationEdbInfoList := make([]*EdbInfo, 0) for _, v := range edbInfoCalculateDetailList { tagMap[v.FromTag] = v.FromEdbInfoId fromEdbInfo, _ := GetEdbInfoById(v.FromEdbInfoId) relationEdbInfoList = append(relationEdbInfoList, fromEdbInfo) } o := orm.NewOrm() to, err := o.Begin() if err != nil { return } defer func() { if err != nil { fmt.Println(reflect.TypeOf(obj).Name() + ",Err:" + err.Error()) _ = to.Rollback() } else { _ = to.Commit() } }() // 计算数据 err = obj.refresh(to, edbInfo.EdbInfoId, edbInfo.Source, edbInfo.SubSource, relationEdbInfoList, edbInfo.EdbCode, edbInfo.Extra, tagMap, edbInfo.EmptyType) return } func (obj CalculateAvg) refresh(to orm.TxOrmer, edbInfoId, source, subSource int, edbInfoIdArr []*EdbInfo, edbCode, extra string, edbInfoTag map[string]int, emptyType int) (err error) { //to orm.TxOrmer, edbInfoId, source, subSource int, edbInfoIdArr []*EdbInfo, edbCode, extra string, edbInfoTag map[string]int, emptyType int realSaveDataMap := make(map[string]map[int]float64) saveDataMap := make(map[string]map[int]float64) // 最小的结束日期 , 最晚的数据开始日期 var minLatestDate, maxStartDate time.Time dateList := make([]string, 0) // 最终的日期数据 dateMap := make(map[string]struct{}) // 最终的日期数据 edbInfoIdStr := strconv.Itoa(edbInfoId) tableName := GetEdbDataTableName(source, subSource) // 获取多指标求和关联的指标id dateTagConfig := "" if extra != "" { var dateConfig CalculateEdbExtra err = json.Unmarshal([]byte(extra), &dateConfig) if err != nil { err = fmt.Errorf("refreshAllCalculate,extra解析失败,Err:%s", err.Error()) return } dateTagConfig = dateConfig.DateTag } for edbInfoIndex, v := range edbInfoIdArr { dataList, err := GetEdbDataListAllByTo(to, v.Source, v.SubSource, FindEdbDataListAllCond{ EdbInfoId: v.EdbInfoId, }, 1) if err != nil { return err } for _, dv := range dataList { if val, ok := realSaveDataMap[dv.DataTime]; ok { if _, ok := val[v.EdbInfoId]; !ok { val[v.EdbInfoId] = dv.Value } } else { temp := make(map[int]float64) temp[v.EdbInfoId] = dv.Value realSaveDataMap[dv.DataTime] = temp } // saveDataMap if val, ok := saveDataMap[dv.DataTime]; ok { if _, ok := val[v.EdbInfoId]; !ok { val[v.EdbInfoId] = dv.Value } } else { temp2 := make(map[int]float64) temp2[v.EdbInfoId] = dv.Value saveDataMap[dv.DataTime] = temp2 } if dateTagConfig == "all" { if _, ok := dateMap[dv.DataTime]; !ok { dateList = append(dateList, dv.DataTime) dateMap[dv.DataTime] = struct{}{} } } else if dateTagConfig == "" { // 默认取第一个指标的时间序列 if edbInfoIndex == 0 { if _, ok := dateMap[dv.DataTime]; !ok { dateList = append(dateList, dv.DataTime) dateMap[dv.DataTime] = struct{}{} } } } else { if eId, ok := edbInfoTag[dateTagConfig]; ok { if v.EdbInfoId == eId { if _, ok1 := dateMap[dv.DataTime]; !ok1 { dateList = append(dateList, dv.DataTime) dateMap[dv.DataTime] = struct{}{} } } } } } } // 处理最大日期和最小日期 for _, v := range dateList { tmpDate, _ := time.ParseInLocation(utils.FormatDate, v, time.Local) if minLatestDate.IsZero() || tmpDate.After(minLatestDate) { minLatestDate = tmpDate } if maxStartDate.IsZero() || tmpDate.Before(maxStartDate) { maxStartDate = tmpDate } } //数据处理,将日期内不全的数据做补全 HandleDateSaveDataMap(dateList, maxStartDate, minLatestDate, realSaveDataMap, saveDataMap, edbInfoIdArr, emptyType) addSql := ` INSERT INTO ` + tableName + ` (edb_info_id,edb_code,data_time,value,create_time,modify_time,data_timestamp) values ` var isAdd bool //获取指标所有数据 dataList := make([]*EdbData, 0) dataTableName := GetEdbDataTableName(source, subSource) sql := `SELECT * FROM %s WHERE edb_info_id=? ` sql = fmt.Sprintf(sql, dataTableName) _, err = to.Raw(sql, edbInfoId).QueryRows(&dataList) if err != nil { return err } dataMap := make(map[string]string) removeDateMap := make(map[string]int) //需要移除的日期 for _, v := range dataList { dataMap[v.DataTime] = v.Value removeDateMap[v.DataTime] = 1 } existDataMap := make(map[string]string) for sk, sv := range saveDataMap { if _, ok := dateMap[sk]; !ok { continue } // 当空值处理类型选择了不计算时,只要有一个指标在某个日期没有值(即空值),则计算指标在该日期没有值 if len(sv) != len(edbInfoIdArr) { continue } var calVal float64 for _, value := range sv { calVal += value } calVal = calVal / float64(len(sv)) // 有计算出来值,那么就从待删除指标中移除 delete(removeDateMap, sk) saveValue := decimal.NewFromFloat(calVal).RoundCeil(4).String() if existVal, ok := dataMap[sk]; !ok { dataTime, _ := time.ParseInLocation(utils.FormatDate, sk, time.Local) timestamp := dataTime.UnixNano() / 1e6 timeStr := fmt.Sprintf("%d", timestamp) if _, existOk := existDataMap[sk]; !existOk { addSql += GetAddSql(edbInfoIdStr, edbCode, sk, timeStr, saveValue) isAdd = true } existDataMap[sk] = sk } else { existValDecimal, err := decimal.NewFromString(existVal) existStr := existValDecimal.String() if existStr != 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, sk).Exec() if err != nil { return err } } } } if isAdd { addSql = strings.TrimRight(addSql, ",") _, err = to.Raw(addSql).Exec() if err != nil { fmt.Println("RefreshAllCalculate add Err", err.Error()) return } } if len(removeDateMap) > 0 { removeDateList := make([]string, 0) //需要移除的日期 for k := range removeDateMap { 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 } // GetSource 获取来源编码id func (obj CalculateAvg) GetSource() int { return utils.DATA_SOURCE_CALCULATE_AVG } // GetSourceName 获取来源名称 func (obj CalculateAvg) GetSourceName() string { return utils.DATA_SOURCE_NAME_CALCULATE_AVG } // GetEdbType 获取指标类型 func (obj CalculateAvg) GetEdbType() int { return utils.CALCULATE_EDB_TYPE }