/* * MQTT 报文管理 Service(数据来自 iotcore-stat Elasticsearch) * 站点即 equipNo。看板统计 + 逐条查询。 */ package services import ( "fmt" "fwh-toolbox-server/domain" "fwh-toolbox-server/repositories" "time" "go.uber.org/zap" ) /*************** 看板:报文总量 ***************/ // GetMqttDailyTotal 日报文总量 func GetMqttDailyTotal(date string) (*domain.MessageTotal, error) { res, err := repositories.EsGetDailyMessageTotal(date) if err != nil { zap.L().Error("EsGetDailyMessageTotal failed.", zap.String("err", err.Error()), zap.String("date", date)) return nil, err } return res, nil } // GetMqttMonthlyTotal 月报文总量(仅统计总数,代价可控;不涉及慢的站点月聚合) func GetMqttMonthlyTotal(year, month int) (*domain.MessageTotal, error) { res, err := repositories.EsGetMonthlyMessageTotal(year, month) if err != nil { zap.L().Error("EsGetMonthlyMessageTotal failed.", zap.String("err", err.Error()), zap.Int("year", year), zap.Int("month", month)) return nil, err } return res, nil } // GetMqttMonthlyDailyTrend 全站本月每日报文总量(看板柱状图)。 // 结果按整月 1~月末补齐:缺失的天补 0,保证横轴完整。 func GetMqttMonthlyDailyTrend(year, month int) ([]domain.DailyTotal, error) { list, err := repositories.EsGetMonthlyDailyTrend(year, month) if err != nil { zap.L().Error("EsGetMonthlyDailyTrend failed.", zap.String("err", err.Error()), zap.Int("year", year), zap.Int("month", month)) return nil, err } // 已有数据按日期建索引 totalByDate := make(map[string]int64, len(list)) for _, d := range list { totalByDate[d.Date] = d.Total } // 生成整月 1~月末的完整日期序列(缺失补 0) // 用下月第 0 天得到当月天数 daysInMonth := time.Date(year, time.Month(month)+1, 0, 0, 0, 0, 0, time.UTC).Day() out := make([]domain.DailyTotal, 0, daysInMonth) for day := 1; day <= daysInMonth; day++ { date := fmt.Sprintf("%04d-%02d-%02d", year, month, day) out = append(out, domain.DailyTotal{Date: date, Total: totalByDate[date]}) } return out, nil } /*************** 看板:站点日总量(分页降序) ***************/ // GetMqttSiteDailyTotals 站点日总量(分页) func GetMqttSiteDailyTotals(date string, page, pageSize int) (*domain.SiteTotalPage, error) { res, err := repositories.EsGetSiteDailyTotals(date, page, pageSize) if err != nil { zap.L().Error("EsGetSiteDailyTotals failed.", zap.String("err", err.Error()), zap.String("date", date)) return nil, err } return res, nil } /*************** 下钻:站点当日命令类型总量 ***************/ // GetMqttSiteCmdTypeTotals 某站点某天各命令类型总量(分页) func GetMqttSiteCmdTypeTotals(equipNo, date string, page, pageSize int) (*domain.CmdTypeTotalPage, error) { res, err := repositories.EsGetSiteCmdTypeTotals(equipNo, date, page, pageSize) if err != nil { zap.L().Error("EsGetSiteCmdTypeTotals failed.", zap.String("err", err.Error()), zap.String("equipNo", equipNo), zap.String("date", date)) return nil, err } return res, nil } /*************** 查询:逐条原始报文 ***************/ // QueryMqttMessages 按过滤条件查询逐条报文 func QueryMqttMessages(equipNo *string, cmdType *int64, startTime, endTime *time.Time, page, pageSize int) (*domain.MessagePage, error) { param := repositories.EsQueryMessageParam{ EquipNo: equipNo, CmdType: cmdType, StartTime: startTime, EndTime: endTime, Page: page, PageSize: pageSize, } res, err := repositories.EsQueryMessages(param) if err != nil { zap.L().Error("EsQueryMessages failed.", zap.String("err", err.Error())) return nil, err } return res, nil }