fix: UPF流量总计7,30天累计数据查询

features/pm/kpi_c_report/controller.go

@@ -36,7 +36,7 @@ func (k *KpiCReport) Get(c *gin.Context) {
 	dbg := db.DB("").Model(&KpiCReport{}).Table(tableName)
 
 	if querys.NeID != "" {
-		conditions = append(conditions, "rm_uid = (select n.rm_uid from ne_info n where n.ne_type=? and n.ne_id=? and n.status=1)")
+		conditions = append(conditions, "rm_uid = (select n.rm_uid from ne_info n where n.ne_type=? and n.ne_id=?)")
 		params = append(params, strings.ToUpper(querys.NeType), querys.NeID)
 	} else {
 		c.JSON(http.StatusBadRequest, services.ErrResp("Not found required parameter NE ID"))

features/pm/performance.go

@@ -202,7 +202,7 @@ func saveKPIData(kpiReport KpiReport, index int64) int64 {
 				// 更新UPF总流量
 				upValue := parse.Number(kpiEvent["UPF.03"])
 				downValue := parse.Number(kpiEvent["UPF.06"])
-				neDataService.NewKpiReport.UPFTodayFlowUpdate(neInfo.RmUID, upValue, downValue, false)
+				neDataService.NewKpiReport.UPFTodayFlowUpdate(neInfo.RmUID, upValue, downValue)
 			}
 		}
 	}

src/modules/network_data/network_data.go

@@ -357,5 +357,5 @@ func Setup(router *gin.Engine) {
 // InitLoad 初始参数
 func InitLoad() {
 	// 启动时，加载UPF上下行流量
-	go service.NewKpiReport.UPFTodayFlowLoad()
+	go service.NewKpiReport.UPFTodayFlowLoad(30)
 }

src/modules/network_data/service/kpi_report.go

@@ -142,13 +142,14 @@ func (r KpiReport) FindTitle(neType string) []model.KpiTitle {
 }
 
 // UPFTodayFlowFind 查询UPF总流量 N3上行 N6下行
+// day 统计天数
 // down * 8 / 1000 / 1000 单位M
 func (r KpiReport) UPFTodayFlowFind(rmUID string, day int) (int64, int64) {
 	// 获取当前日期
 	now := time.Now()
 	var upTotal, downTotal int64
 
-	// 查询最近7天的数据
+	// 查询最近day天的数据
 	for i := 0; i <= day; i++ {
 		dateKey := now.AddDate(0, 0, -i).Format("2006-01-02")
 		key := fmt.Sprintf("%s:UPF_FLOW:%s:%s", constants.CACHE_NE_DATA, rmUID, dateKey)
@@ -171,28 +172,11 @@ func (r KpiReport) UPFTodayFlowFind(rmUID string, day int) (int64, int64) {
 }
 
 // UPFTodayFlow UPF流量今日统计
-func (r KpiReport) UPFTodayFlowUpdate(rmUID string, upValue, downValue int64, rest bool) error {
+func (r KpiReport) UPFTodayFlowUpdate(rmUID string, upValue, downValue int64) error {
 	// 按日期存储统计数据
 	dateKey := time.Now().Format("2006-01-02")
 	key := fmt.Sprintf("%s:UPF_FLOW:%s:%s", constants.CACHE_NE_DATA, rmUID, dateKey)
 
-	// 重置数据
-	if rest {
-		err := redis.SetHash("", key, map[string]any{
-			"up":   upValue,
-			"down": downValue,
-		})
-		if err != nil {
-			return err
-		}
-		// 设置key的过期时间为30天，自动清理旧数据
-		err = redis.Expire("", key, 30*24*time.Hour)
-		if err != nil {
-			return err
-		}
-		return nil
-	}
-
 	// 使用HIncrBy实时累加统计值
 	if err := redis.IncrBy("", key, "up", upValue); err != nil {
 		return err
@@ -204,17 +188,14 @@ func (r KpiReport) UPFTodayFlowUpdate(rmUID string, upValue, downValue int64, re
 }
 
 // UPFTodayFlowLoad UPF上下行数据到redis
-func (r KpiReport) UPFTodayFlowLoad() {
+// day 统计天数
+func (r KpiReport) UPFTodayFlowLoad(day int) {
 	cacheKeys, _ := redis.GetKeys("", constants.CACHE_NE_INFO+":UPF:*")
 	if len(cacheKeys) == 0 {
 		return
 	}
 
-	// 今日流量
 	now := time.Now()
-	beginTime := now.Truncate(24 * time.Hour).UnixMilli()
-	endTime := beginTime + 24*60*60*1000 - 1
-
 	for _, key := range cacheKeys {
 		var v neModel.NeInfo
 		jsonStr, _ := redis.Get("", key)
@@ -222,31 +203,50 @@ func (r KpiReport) UPFTodayFlowLoad() {
 			json.Unmarshal([]byte(jsonStr), &v)
 		}
 		if v.NeType == "UPF" && v.RmUID != "" {
-			// 查询历史数据
-			rows := r.kpiReportRepository.SelectUPF(v.RmUID, beginTime, endTime)
-			var upTotal, downTotal int64
+			// 查询最近day天的数据
+			for i := 0; i <= day; i++ {
+				dateKey := now.AddDate(0, 0, -i).Format("2006-01-02")
+				key := fmt.Sprintf("%s:UPF_FLOW:%s:%s", constants.CACHE_NE_INFO, v.RmUID, dateKey)
+				// 根据传入天数计算时间范围
+				beginTime := now.AddDate(0, 0, -i).Truncate(24 * time.Hour).UnixMilli()
+				endTime := beginTime + 24*60*60*1000 - 1
+				// 查询历史数据
+				rows := r.kpiReportRepository.SelectUPF(v.RmUID, beginTime, endTime)
+				var upTotal, downTotal int64
 
-			// 处理历史数据
-			for _, row := range rows {
-				var kpiValues []map[string]any
-				if err := json.Unmarshal([]byte(row.KpiValues), &kpiValues); err != nil {
-					continue
-				}
+				// 处理历史数据
+				for _, row := range rows {
+					var kpiValues []map[string]any
+					if err := json.Unmarshal([]byte(row.KpiValues), &kpiValues); err != nil {
+						continue
+					}
 
-				for _, v := range kpiValues {
-					if k, ok := v["kpiId"]; ok {
-						if k == "UPF.03" {
-							upTotal += parse.Number(v["value"])
-						}
-						if k == "UPF.06" {
-							downTotal += parse.Number(v["value"])
+					for _, v := range kpiValues {
+						if k, ok := v["kpiId"]; ok {
+							if k == "UPF.03" {
+								upTotal += parse.Number(v["value"])
+							}
+							if k == "UPF.06" {
+								downTotal += parse.Number(v["value"])
+							}
 						}
 					}
 				}
-			}
 
-			// 将历史数据添加到Redis
-			r.UPFTodayFlowUpdate(v.RmUID, upTotal, downTotal, true)
+				err := redis.SetHash("", key, map[string]any{
+					"up":   upTotal,
+					"down": downTotal,
+				})
+				if err != nil {
+					continue
+				}
+				// 设置key的过期时间为30天，自动清理旧数据
+				daySub := (30 - i) * 24
+				err = redis.Expire("", key, time.Duration(daySub)*time.Hour)
+				if err != nil {
+					continue
+				}
+			}
 		}
 	}
 }