FranklinWH-Toolbox/service/repositories/api/fwh_pro/fwh_mqttLog.go

package fwh_pro

import (
	"encoding/json"
	"errors"
	"fmt"
	"fwh-toolbox-server/domain"
	"net/url"
	"strconv"
	"strings"
	"time"
)

type MqttLogListResponse struct {
	Code    int                  `json:"code"`
	Message string               `json:"message"`
	Result  []domain.MqttLogItem `json:"result"`
	Success bool                 `json:"success"`
	Total   uint                 `json:"total"`
}

type QueryMqttLogParam struct {
	SN        *string
	CmdTypes  []domain.MqttCmdType
	StartTime *time.Time
	EndTime   *time.Time
}

// 查询站点信息
func queryMqttLog(param QueryMqttLogParam) ([]byte, error) {
	// 构造 query 参数
	params := url.Values{}
	if param.SN != nil {
		params.Add("searchName", *param.SN)
	}
	if len(param.CmdTypes) > 0 {
		cmdTypesStr := make([]string, len(param.CmdTypes))
		for i, v := range param.CmdTypes {
			cmdTypesStr[i] = string(v) // 转成 string
		}
		params.Add("cmdTypes", strings.Join(cmdTypesStr, ","))
	}
	if param.StartTime != nil {
		params.Add("startTime", param.StartTime.Format(time.DateTime))
	}
	if param.EndTime != nil {
		params.Add("endTime", param.EndTime.Format(time.DateTime))
	}
	if param.StartTime != nil || param.EndTime != nil {
		params.Add("timeType", "1") // 0-通讯时间 1-设备时间时
	}
	params.Add("pageSize", strconv.FormatUint(REQ_PAGE_SIZE, 10))
	params.Add("pageNum", strconv.Itoa(1))
	params.Add("lang", "ZH_CN")

	return FwhHttpGet("https://energy.franklinwh.com/hes-gateway/manage/iotMqttCommlogs", params)
}

func queryMqttLogListOnePage(sn string, cmdType domain.MqttCmdType, startTime, endTime time.Time) ([]domain.MqttLogItem, error) {
	param := QueryMqttLogParam{
		SN:        &sn,
		CmdTypes:  []domain.MqttCmdType{cmdType},
		StartTime: &startTime,
		EndTime:   &endTime,
	}
	resByte, err := queryMqttLog(param)
	if err != nil {
		return nil, err
	}
	var mqttLogRes MqttLogListResponse
	// 反序列化（unmarshal）
	jsonErr := json.Unmarshal(resByte, &mqttLogRes)
	if jsonErr != nil {
		return nil, jsonErr
	}

	if mqttLogRes.Code == 401 {
		// TODO 尝试重新登录
		return nil, errors.New("certification expired")
	}

	// 判断是否超出可处理数据上限 10万条数据
	if mqttLogRes.Total > 10*10000 {
		return nil, fmt.Errorf("超出数据请求上限 - Total: %d (>10万)", mqttLogRes.Total)
	}

	return mqttLogRes.Result, nil
}

// 从FWH获取MQTT log数据
func QueryMqttLogListOnTimeRange(sn string, cmdType domain.MqttCmdType, startTime, endTime time.Time) ([]domain.MqttLogItem, error) {
	// 保证 startTime <= endTime
	if startTime.After(endTime) {
		return nil, errors.New("startTime > endTime")
	}

	// 按天拆分时间段
	var periods []struct{ Start, End time.Time }
	currTime := startTime
	for currTime.Before(endTime) {
		// 当天结束时间
		dayEnd := time.Date(currTime.Year(), currTime.Month(), currTime.Day(), 23, 59, 59, 0, currTime.Location())

		// 如果当天结束大于 endTime，使用 endTime
		if dayEnd.After(endTime) {
			dayEnd = endTime
		}

		periods = append(periods, struct{ Start, End time.Time }{Start: currTime, End: dayEnd})

		// 下一天开始
		currTime = dayEnd.Add(time.Second).Truncate(24 * time.Hour)
		currTime = time.Date(currTime.Year(), currTime.Month(), currTime.Day(), 0, 0, 0, 0, currTime.Location())
	}

	// 按天查询
	var mqttLogList []domain.MqttLogItem
	for _, period := range periods {
		var mqttLogRes []domain.MqttLogItem = nil
		periodEndTime := period.End
		// 检查数据是否超过1000(单次查询上限)
		for {
			var err error
			mqttLogRes, err = queryMqttLogListOnePage(sn, cmdType, period.Start, periodEndTime)
			if err != nil {
				return nil, err
			}

			time.Sleep(time.Second) // 延时1s

			mqttLogList = append(mqttLogList, mqttLogRes...)

			// 判断是否未查完
			if len(mqttLogRes) < int(REQ_PAGE_SIZE) {
				// 请求完成，退出
				break
			}

			// 跳转时间段，再次请求
			// 提取最后一条数据的设备时间
			periodEndTime, err = time.Parse(time.DateTime, mqttLogRes[len(mqttLogRes)-1].LogTime)
			if err != nil {
				return nil, err
			}
		}
	}

	return mqttLogList, nil
}