FranklinWH-Toolbox/service/services/mqttDataManageServices.go

/*
 * 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
}