package fwh_pro import ( "errors" "fmt" "fwh-toolbox-server/domain" "net/url" "strconv" "strings" "time" "go.uber.org/zap" ) type QueryMqttLogParam struct { SN *string CmdTypes []domain.MqttCmdType StartTime *time.Time EndTime *time.Time } // 查询站点信息 func queryMqttLog(param QueryMqttLogParam) (*domain.FwhPageResponse[[]domain.MqttLogItem], 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") response, err := FwhHttpGetUnmarshalPage[[]domain.MqttLogItem](getFullEnergyUrl("/hes-gateway/manage/iotMqttCommlogs"), params) if err != nil { return nil, err } return response, 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, } mqttLogRes, err := queryMqttLog(param) if err != nil { return nil, err } if mqttLogRes.Code != 200 { return nil, fmt.Errorf("[queryMqttLogListOnePage] 查询失败 code: %d, message: %s", mqttLogRes.Code, mqttLogRes.Message) } // 判断是否超出可处理数据上限 10万条数据 if mqttLogRes.Total > 10*10000 { return nil, fmt.Errorf("超出数据请求上限 - Total: %d (>10万)", mqttLogRes.Total) } return *mqttLogRes.Result, nil } // 从FWH获取MQTT log数据 // // 限制条件(重要):startTime/endTime 必须为「不含时区」的字面时间(用 UTC 表示)。 // 本函数按天拆分时间段时用 time.Truncate(24h) 推进游标,Truncate 按 UTC 零点对齐; // 若传入带非 UTC 时区(如 America/Chicago)的时间,游标会回退到前一天, // currTime.Before(endTime) 恒成立,periods 切片无限 append,导致内存 OOM。 // 调用方(DataDownload)需先用 stripTimeLocation 剥离时区再传入。 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()) } zap.L().Debug("[fwh_pro] MQTT periods 构建完成", zap.String("sn", sn), zap.Any("cmdType", cmdType), zap.Time("start", startTime), zap.Time("end", endTime), zap.Int("periods", len(periods)), ) // 按天查询 var mqttLogList []domain.MqttLogItem for _, period := range periods { var mqttLogRes []domain.MqttLogItem = nil periodEndTime := period.End pages := 0 // 检查数据是否超过1000(单次查询上限) for { var err error pages++ zap.L().Debug("[fwh_pro] MQTT 分页请求发起", zap.String("sn", sn), zap.Any("cmdType", cmdType), zap.Time("periodStart", period.Start), zap.Time("periodEnd", periodEndTime), zap.Int("page", pages), zap.Int("accumulated", len(mqttLogList)), ) mqttLogRes, err = queryMqttLogListOnePage(sn, cmdType, period.Start, periodEndTime) if err != nil { return nil, err } time.Sleep(time.Second) // 延时1s mqttLogList = append(mqttLogList, mqttLogRes...) zap.L().Debug("[fwh_pro] MQTT 分页请求返回", zap.String("sn", sn), zap.Any("cmdType", cmdType), zap.Int("page", pages), zap.Int("pageRows", len(mqttLogRes)), zap.Int("accumulated", len(mqttLogList)), ) // 判断是否未查完 if len(mqttLogRes) < int(REQ_PAGE_SIZE) { // 请求完成,退出 break } // 跳转时间段,再次请求 // 提取最后一条数据的设备时间 next, err := time.Parse(time.DateTime, mqttLogRes[len(mqttLogRes)-1].LogTime) if err != nil { return nil, err } // 游标未能前进:说明同一秒内 >= REQ_PAGE_SIZE(1000) 条数据, // 单页游标(按最后一条 LogTime)无法翻过这一秒,继续循环会死循环并 OOM。 // 现实中基本不会发生,此处仅作防护:记录错误并返回,不再无限拉取。 if !next.Before(periodEndTime) { zap.L().Error("[fwh_pro] MQTT 日志分页游标停滞,疑似同一秒内数据超过单页上限", zap.String("sn", sn), zap.Any("cmdType", cmdType), zap.Time("stuckAt", next), zap.Int("pageSize", int(REQ_PAGE_SIZE)), zap.Int("accumulated", len(mqttLogList)), ) return nil, fmt.Errorf("mqtt log pagination stalled at %s: more than %d rows within one second", next.Format(time.DateTime), REQ_PAGE_SIZE) } periodEndTime = next } } return mqttLogList, nil }