package fwh_pro import ( "encoding/json" "errors" "fmt" "fwh-toolbox-server/config" "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") reqUrl := fmt.Sprintf("https://%s/hes-gateway/manage/iotMqttCommlogs", config.FWH_ENERGY) response, err := FwhHttpGet(reqUrl, params) if err != nil { return nil, err } return response.Body, nil } 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 }