/* * MQTT 报文管理 API List(数据来自 iotcore-stat Elasticsearch,站点即 equipNo) * * 看板: * /mqttDataManage/total/daily - GET 日总量 query: date=yyyy.MM.dd * /mqttDataManage/total/monthly - GET 月总量 query: year, month * /mqttDataManage/total/dailyTrend - GET 月度每日总量 query: year, month * /mqttDataManage/site/daily - GET 站点日总量 query: date, page, size * /mqttDataManage/site/cmdType - GET 站点当日命令类型总量 query: equipNo, date, page, size * 查询: * /mqttDataManage/messages - GET 逐条报文 query: equipNo, cmdType, startTime, endTime, page, size * * 注:站点月总量列表(site/monthly)因查询非常慢、非常耗资源已下线,见 fwh_es_mqtt.GetSiteMonthlyTotals 备注。 * * 时区:全部统一使用 UTC。date 参数(yyyy.MM.dd)对应 UTC 日索引; * startTime/endTime(yyyy-MM-dd HH:mm:ss)按 UTC 解析;聚合按 UTC 分桶。 */ package controllers import ( "errors" "fmt" "fwh-toolbox-server/repositories" "fwh-toolbox-server/services" "regexp" "strconv" "time" "github.com/gofiber/fiber/v2" "go.uber.org/zap" ) // 日期格式 yyyy.MM.dd(对应 ES 日索引) var esDateRegexp = regexp.MustCompile(`^\d{4}\.(0[1-9]|1[0-2])\.(0[1-9]|[12]\d|3[01])$`) // 时间格式 yyyy-MM-dd HH:mm:ss const mqttTimeLayout = "2006-01-02 15:04:05" // 解析分页参数(page 默认 1,size 默认 20) func parsePaging(c *fiber.Ctx) (int, int, error) { page, err := strconv.Atoi(c.Query("page", "1")) if err != nil || page <= 0 { return 0, 0, err } size, err := strconv.Atoi(c.Query("size", "20")) if err != nil || size <= 0 { return 0, 0, err } return page, size, nil } // 解析年月 func parseYearMonth(c *fiber.Ctx) (int, int, bool) { year, err := strconv.Atoi(c.Query("year", "")) if err != nil || year < 2000 || year > 9999 { return 0, 0, false } month, err := strconv.Atoi(c.Query("month", "")) if err != nil || month < 1 || month > 12 { return 0, 0, false } return year, month, true } // GetMqttDailyTotal 日总量 // /mqttDataManage/total/daily func GetMqttDailyTotal(c *fiber.Ctx) error { date := c.Query("date", "") if !esDateRegexp.MatchString(date) { zap.L().Warn("date parameter is in the wrong format", zap.String("date", date)) return Fail(c, "date parameter is in the wrong format (yyyy.MM.dd)") } res, err := services.GetMqttDailyTotal(date) if err != nil { return Error(c, "Get MQTT daily total failed.") } return Success(c, res) } // GetMqttMonthlyTotal 月总量 // /mqttDataManage/total/monthly func GetMqttMonthlyTotal(c *fiber.Ctx) error { year, month, ok := parseYearMonth(c) if !ok { return Fail(c, "year/month parameter is invalid") } res, err := services.GetMqttMonthlyTotal(year, month) if err != nil { return Error(c, "Get MQTT monthly total failed.") } return Success(c, res) } // GetMqttMonthlyDailyTrend 月度每日总量(看板柱状图,整月补零) // /mqttDataManage/total/dailyTrend func GetMqttMonthlyDailyTrend(c *fiber.Ctx) error { year, month, ok := parseYearMonth(c) if !ok { return Fail(c, "year/month parameter is invalid") } res, err := services.GetMqttMonthlyDailyTrend(year, month) if err != nil { return Error(c, "Get MQTT monthly daily trend failed.") } return Success(c, res) } // GetMqttSiteDailyTotals 站点日总量(分页) // /mqttDataManage/site/daily func GetMqttSiteDailyTotals(c *fiber.Ctx) error { date := c.Query("date", "") if !esDateRegexp.MatchString(date) { return Fail(c, "date parameter is in the wrong format (yyyy.MM.dd)") } page, size, err := parsePaging(c) if err != nil { return Fail(c, "page/size parameter is invalid") } res, err := services.GetMqttSiteDailyTotals(date, page, size) if err != nil { return Error(c, "Get MQTT site daily totals failed.") } return Success(c, res) } // GetMqttSiteCmdTypeTotals 站点当日命令类型总量(下钻,分页) // /mqttDataManage/site/cmdType func GetMqttSiteCmdTypeTotals(c *fiber.Ctx) error { equipNo := c.Query("equipNo", "") if equipNo == "" { return Fail(c, "equipNo parameter is empty") } date := c.Query("date", "") if !esDateRegexp.MatchString(date) { return Fail(c, "date parameter is in the wrong format (yyyy.MM.dd)") } page, size, err := parsePaging(c) if err != nil { return Fail(c, "page/size parameter is invalid") } res, err := services.GetMqttSiteCmdTypeTotals(equipNo, date, page, size) if err != nil { return Error(c, "Get MQTT site cmdType totals failed.") } return Success(c, res) } // QueryMqttMessages 逐条报文查询 // /mqttDataManage/messages func QueryMqttMessages(c *fiber.Ctx) error { // 可选:equipNo var equipNoPtr *string if v := c.Query("equipNo", ""); v != "" { equipNoPtr = &v } // 可选:cmdType var cmdTypePtr *int64 if v := c.Query("cmdType", ""); v != "" { cmdType, err := strconv.ParseInt(v, 10, 64) if err != nil { return Fail(c, "cmdType parameter is invalid") } cmdTypePtr = &cmdType } // 可选:startTime / endTime(yyyy-MM-dd HH:mm:ss,按 UTC 解析) var startPtr, endPtr *time.Time if v := c.Query("startTime", ""); v != "" { t, err := time.ParseInLocation(mqttTimeLayout, v, time.UTC) if err != nil { return Fail(c, "startTime parameter is in the wrong format (yyyy-MM-dd HH:mm:ss)") } startPtr = &t } if v := c.Query("endTime", ""); v != "" { t, err := time.ParseInLocation(mqttTimeLayout, v, time.UTC) if err != nil { return Fail(c, "endTime parameter is in the wrong format (yyyy-MM-dd HH:mm:ss)") } endPtr = &t } page, size, err := parsePaging(c) if err != nil { return Fail(c, "page/size parameter is invalid") } res, err := services.QueryMqttMessages(equipNoPtr, cmdTypePtr, startPtr, endPtr, page, size) if err != nil { // 深分页超出 ES 窗口上限:返回 400 友好提示,而非 500 if errors.Is(err, repositories.ErrEsDeepPaging) { return Fail(c, fmt.Sprintf("查询超出可翻页范围:最多可浏览前 %d 条,请缩小时间范围或增加设备号/命令类型等过滤条件。", repositories.EsMaxMessagePageWindow)) } return Error(c, "Query MQTT messages failed.") } return Success(c, res) }