package fwhQueueOld import ( "context" "errors" "fmt" "fwh-toolbox-server/domain" "fwh-toolbox-server/dto" "fwh-toolbox-server/jobs" "fwh-toolbox-server/logger" "fwh-toolbox-server/repositories" "strconv" "strings" "time" "go.uber.org/zap" ) var ( TaskTypeBatchDownload = "task_BatchDownload" ) // type TaskBatchDownload = types.TaskBatchDownload func handleTaskBatchDownload(ctx context.Context, payload domain.TaskBatchDownload) error { // 更新运行状态 fmt.Println("更新任务运行状态") if repoErr := repositories.UpdateBatchDownloadResult(payload.Id, domain.ProgressRunning, []string{}); repoErr != nil { zap.L().Error("更新批量数据下载结果状态失败(Running)", zap.String("id", payload.Id.Hex())) return repoErr } fmt.Printf("开始执行任务 %v\n", payload) reasons, err := executeTaskBatchDownload(ctx, payload) // 记录任务执行结果 if err != nil { fmt.Printf("任务执行失败 %s\n", err.Error()) if repoErr := repositories.UpdateBatchDownloadResult(payload.Id, domain.ProgressError, []string{err.Error()}); repoErr != nil { zap.L().Error("更新批量数据下载结果状态失败(Error)", zap.String("id", payload.Id.Hex())) return repoErr } } else { if len(reasons) == 0 { fmt.Println("任务执行成功") if repoErr := repositories.UpdateBatchDownloadResult(payload.Id, domain.ProgressComplete, []string{}); repoErr != nil { zap.L().Error("更新批量数据下载结果状态失败(Complete)", zap.String("id", payload.Id.Hex()), zap.String("err", repoErr.Error())) return repoErr } } else { fmt.Println("任务执行失败") if repoErr := repositories.UpdateBatchDownloadResult(payload.Id, domain.ProgressFailed, reasons); repoErr != nil { zap.L().Error("更新批量数据下载结果状态失败(Complete)", zap.String("id", payload.Id.Hex()), zap.String("err", repoErr.Error())) return repoErr } } } return err } var executeTaskBatchDownload = func(ctx context.Context, payload domain.TaskBatchDownload) ([]string, error) { // 转换时间范围,只取用日期部分 startTime, err := time.Parse(time.DateOnly, payload.TimePeriod[0]) if err != nil { zap.L().Error("StartTime 格式错误", zap.String("time", payload.TimePeriod[0]), zap.String("err", err.Error())) return nil, err } endTime, err := time.Parse(time.DateOnly, payload.TimePeriod[1]) if err != nil { zap.L().Error("EndTime 格式错误", zap.String("time", payload.TimePeriod[1]), zap.String("err", err.Error())) return nil, err } // 将 endTime 调整到当天的 23:59:59 endTime = endTime.Add(23*time.Hour + 59*time.Minute + 59*time.Second) // 保证 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()) } fmt.Println(payload.Options) // 优先处理文件下载任务(并行处理缩短时间) downloadList := []domain.BatchDownloadFileTaskInfo{} for _, sn := range payload.Sn { for _, period := range periods { dateStr := period.Start.Format(time.DateOnly) // 捞取EMS日志 for _, logType := range payload.Options.EmsLogs { downloadList = append(downloadList, domain.BatchDownloadFileTaskInfo{ DevType: -1, Sn: sn, Date: dateStr, FileType: jobs.FileTypeEmsLog, FileSupType: logType, Result: 0, }) } } } for _, sn := range payload.Sn { // 捞取故障录波数据 if payload.Options.FaultRecord { // 召唤aGate故障录波 err := requestCallFaultRecord(&downloadList, 1, sn, startTime, endTime) if err != nil { return nil, err } // 提取设备的子设备SN - 暂时不支持MPPT和aHub siteBaseInfoList, err := repositories.QuerySiteBaseInfoList(dto.SiteBaseInfoPara{ SiteList: []string{sn}, PageSize: 10, PageNum: 1, }) if err != nil { zap.L().Error("查询站点基本信息失败", zap.String("sn", sn), zap.String("err", err.Error())) return nil, err } if len(*siteBaseInfoList.Result) > 0 { aPowerSnList := strings.Split((*siteBaseInfoList.Result)[0].FhpSn, ",") for i := range aPowerSnList { aPowerSn := strings.TrimSpace(aPowerSnList[i]) // 召唤aGate故障录波 err := requestCallFaultRecord(&downloadList, 0, aPowerSn, startTime, endTime) if err != nil { return nil, err } } } } } // 插入文件下载任务 var isDownloadFile bool = false if len(downloadList) > 0 { isDownloadFile = true err := jobs.CreateCallFwhFileTask(payload.Id, downloadList) if err != nil { return nil, err } } // 再处理其他数据下载任务 for _, sn := range payload.Sn { // 生成功率流图文件 if payload.Options.PowerFlow { for _, period := range periods { // 查询功率流数据 fmt.Printf("查询功率流数据 - %s\n", period.Start.Format(time.DateOnly)) res, err := repositories.QuerySitePowerFlow(sn, period.Start) if err != nil { zap.L().Error("查询功率流数据失败", zap.String("sn", sn), zap.String("date", period.Start.Format(time.DateOnly)), zap.String("err", err.Error())) return nil, err } dateStr := period.Start.Format(time.DateOnly) objName, err := jobs.SaveBatchDownloadPowerFlow(sn, dateStr, *res.Result) if err != nil { zap.L().Error("站点运行时数据保存失败", zap.String("sn", sn), zap.String("err", err.Error())) return nil, err } if err := repositories.InsertBatchDownloadPathPowerFlow(payload.Id, sn, dateStr, objName); err != nil { zap.L().Error("BatchDownload - PowerFlow文件地址写入失败", zap.String("sn", sn), zap.String("date", dateStr), zap.String("objName", objName), zap.String("err", err.Error())) return nil, err } } } // 生成Mqtt日志数据文件 for _, logType := range payload.Options.MqttLogs { // 查询MQTT日志 fmt.Printf("查询MQTT日志%s\n", logType) for _, period := range periods { logList, err := repositories.QueryMqttLogListOnTimeRange(sn, domain.MqttCmdType(logType), period.Start, period.End) if err != nil { zap.L().Error("查询MQTT日志失败", zap.String("type", logType), zap.String("err", err.Error())) return nil, err } // 将结果保存到文件服务器 dateStr := period.Start.Format(time.DateOnly) objName, err := jobs.SaveBatchDownloadMqttLog(sn, logType, dateStr, logList) if err != nil { zap.L().Error("站点运行时数据保存失败", zap.String("sn", sn), zap.String("err", err.Error())) return nil, err } if err := repositories.InsertBatchDownloadPathMqttLogs(payload.Id, sn, logType, dateStr, objName); err != nil { zap.L().Error("BatchDownload - MqttLog文件地址写入失败", zap.String("sn", sn), zap.String("date", dateStr), zap.String("objName", objName), zap.String("err", err.Error())) return nil, err } } } // 生成事件&告警日志文件 if payload.Options.EventAlarm { // 查询事件&告警日志 fmt.Printf("查询事件&告警日志\n") for _, period := range periods { eventAlarmList, err := repositories.QueryEventOrAlarmListOnTimeRange(sn, period.Start, period.End, repositories.EventAndAlarm) if err != nil { zap.L().Error("查询事件&告警日志失败", zap.String("sn", sn), zap.String("startTime", period.Start.Format(time.DateTime)), zap.String("endTime", period.End.Format(time.DateTime)), zap.String("err", err.Error())) return nil, err } // 将结果保存到文件服务器 dateStr := period.Start.Format(time.DateOnly) objName, err := jobs.SaveBatchDownloadEventAlarm(sn, dateStr, eventAlarmList) if err != nil { zap.L().Error("站点事件&告警数据保存失败", zap.String("sn", sn), zap.String("err", err.Error())) return nil, err } if err := repositories.InsertBatchDownloadPathEventAlarm(payload.Id, sn, dateStr, objName); err != nil { zap.L().Error("BatchDownload - EventAlarm文件地址写入失败", zap.String("sn", sn), zap.String("date", dateStr), zap.String("objName", objName), zap.String("err", err.Error())) return nil, err } } } // 生成告警快照文件 if payload.Options.AlarmSnapshot { // 查询告警快照日志 fmt.Printf("查询告警快照日志\n") for _, period := range periods { dateStr := period.Start.Format(time.DateOnly) alarmSnapshotList, err := jobs.GetAlarmSnapshotData(ctx, sn, dateStr) if err != nil { zap.L().Error("查询告警快照日志失败", zap.String("sn", sn), zap.String("date", dateStr), zap.String("err", err.Error())) return nil, err } // 将结果保存到文件服务器 objName, err := jobs.SaveBatchDownloadAlarmSnapshot(sn, dateStr, alarmSnapshotList) if err != nil { zap.L().Error("站点告警快照数据保存失败", zap.String("sn", sn), zap.String("err", err.Error())) return nil, err } if err := repositories.InsertBatchDownloadPathAlarmSnapshot(payload.Id, sn, dateStr, objName); err != nil { zap.L().Error("BatchDownload - AlarmSnapshot文件地址写入失败", zap.String("sn", sn), zap.String("date", dateStr), zap.String("objName", objName), zap.String("err", err.Error())) return nil, err } } } // 生成站点升级日志数据文件 if payload.Options.UpdateLog { fmt.Printf("查询站点升级日志\n") updateLogList := []domain.BatchDownloadSystemUpdateLogInfo{} // 查询系统升级日志 systemUpdateLogList, err := repositories.QuerySystemUpdateLog(sn) if err != nil { zap.L().Error("查询站点系统升级日志失败", zap.String("sn", sn), zap.String("err", err.Error())) return nil, err } // 查询系统升级日志详细信息 for _, logList := range systemUpdateLogList { var updateLogInfo domain.BatchDownloadSystemUpdateLogInfo updateLogInfo.OrderNo = logList.OrderNo updateLogInfo.GateSn = sn updateLogInfo.FileName = logList.Name // 解析 UTC 时间 upgradeTime, err := time.ParseInLocation(time.DateTime, logList.UpgradeTime, time.UTC) if err != nil { panic(err) } // 转换成北京时间 beijing, _ := time.LoadLocation("Asia/Shanghai") bt := upgradeTime.In(beijing) updateLogInfo.UpdateTime = bt.Format(time.DateTime) // 判断升级触发类型 if strings.HasPrefix(logList.OrderNo, "CH") { // 自动升级 updateLogInfo.TriggerType = "Auto" } else if strings.HasPrefix(logList.OrderNo, "DH") { // 手动升级 updateLogInfo.TriggerType = "Manual" } else if strings.HasPrefix(logList.OrderNo, "SH") { // SPAN自动升级 updateLogInfo.TriggerType = "Span Auto" } else { // 其他未知 updateLogInfo.TriggerType = "unknown" } // 判断升级结果 switch logList.Result { case 1: // 升级中 updateLogInfo.Result = "Updating" case 2: // 成功 updateLogInfo.Result = "Success" case 3: // 失败 updateLogInfo.Result = "Fault" default: updateLogInfo.Result = fmt.Sprintf("unknown(%d)", logList.Result) } systemUpdateLogListDetail, err := repositories.QuerySystemUpdateLogDetail(logList.Id, sn) if err != nil { zap.L().Error("查询站点系统升级详细日志失败", zap.String("sn", sn), zap.String("err", err.Error())) return nil, err } // 提取对应SN数据,并提取版本数据 for _, logDetail := range systemUpdateLogListDetail { if logDetail.GatewayId == sn { // 提取Gate版本数据 for _, ibgVer := range logDetail.IbgVersionNew { if ibgVer.Children != nil { for _, childVerInfo := range ibgVer.Children { var verInfo domain.BatchDownloadUpdateLogVersionInfo verInfo.Name = childVerInfo.Type verInfo.OldVer = childVerInfo.OldVersion verInfo.NewVer = childVerInfo.Version updateLogInfo.GateVerList = append(updateLogInfo.GateVerList, verInfo) } } else { var verInfo domain.BatchDownloadUpdateLogVersionInfo verInfo.Name = ibgVer.Type verInfo.OldVer = ibgVer.OldVersion verInfo.NewVer = ibgVer.Version updateLogInfo.GateVerList = append(updateLogInfo.GateVerList, verInfo) } } // 提取子设备版本数据 for _, subInfo := range logDetail.SonNewDevices { var subDevInfo domain.BatchDownloadUpdateLogSubDevInfo subDevInfo.DevSn = subInfo.ApowerSn for _, firstItem := range subInfo.FirstList { for _, secondItem := range firstItem.Secondlist { for _, thirdItem := range secondItem.Thirdlist { var verInfo domain.BatchDownloadUpdateLogVersionInfo verInfo.Name = thirdItem.SubType verInfo.OldVer = thirdItem.OldVersion verInfo.NewVer = thirdItem.Version subDevInfo.DevVerList = append(subDevInfo.DevVerList, verInfo) } } } updateLogInfo.SubDevList = append(updateLogInfo.SubDevList, subDevInfo) } break } } updateLogList = append(updateLogList, updateLogInfo) } // 查询aHub升级日志 aHubUpdateLogList, err := repositories.QueryAHubUpdateLog(sn) if err != nil { zap.L().Error("查询站点aHub升级日志失败", zap.String("sn", sn), zap.String("err", err.Error())) return nil, err } var aHubUpdateLogData []domain.BatchDownloadAHubUpdateLogInfo for _, aHubVerInfo := range aHubUpdateLogList { var updateLogInfo domain.BatchDownloadAHubUpdateLogInfo updateLogInfo.OrderNo = aHubVerInfo.OrderNo updateLogInfo.GateSn = sn updateLogInfo.AHubSn = aHubVerInfo.AHubSn // 判断升级触发类型 if strings.HasPrefix(aHubVerInfo.OrderNo, "CH") { // 自动升级 updateLogInfo.TriggerType = "Auto" } else if strings.HasPrefix(aHubVerInfo.OrderNo, "DH") { // 手动升级 updateLogInfo.TriggerType = "Manual" } else if strings.HasPrefix(aHubVerInfo.OrderNo, "SH") { // SPAN自动升级 updateLogInfo.TriggerType = "Span Auto" } else { // 其他未知 updateLogInfo.TriggerType = "unknown" } // 判断升级结果 switch aHubVerInfo.UpgradeResult { case 0: // 成功 updateLogInfo.Result = "Success" case -1: // 失败 updateLogInfo.Result = "Fault" default: updateLogInfo.Result = fmt.Sprintf("unknown(%d)", aHubVerInfo.UpgradeResult) } for _, detailInfo := range aHubVerInfo.DetailList { var aHubDetailInfo domain.BatchDownloadUpdateLogAHubVersionInfo aHubDetailInfo.FileName = detailInfo.UpgradeFileName aHubDetailInfo.FileOrder = detailInfo.UpgradeFileOrder aHubDetailInfo.Result = detailInfo.UpgradeResult aHubDetailInfo.ResultDesc = detailInfo.UpgradeResultDesc aHubDetailInfo.OldVer = detailInfo.OldVersion aHubDetailInfo.NewVer = detailInfo.NewVersion updateLogInfo.DetailList = append(updateLogInfo.DetailList, aHubDetailInfo) } aHubUpdateLogData = append(aHubUpdateLogData, updateLogInfo) } // 将结果保存到文件服务器 objName, err := jobs.SaveBatchDownloadUpdateLog(sn, updateLogList, aHubUpdateLogData) if err != nil { zap.L().Error("站点升级记录数据保存失败", zap.String("sn", sn), zap.String("err", err.Error())) return nil, err } if err := repositories.InsertBatchDownloadPathUpdateLog(payload.Id, sn, objName); err != nil { zap.L().Error("BatchDownload - UpdateLog文件地址写入失败", zap.String("sn", sn), zap.String("objName", objName), zap.String("err", err.Error())) return nil, err } } // TODO 生成站点设置数据文件(仅当前配置,和选择时间范围无关) // if payload.Options.Settings { // } } if isDownloadFile { for { // 等待文件下载完成,持续判断文件下载进度 standbyList, err := repositories.GetBatchDownloadCallFileTaskStandbyList(payload.Id.Hex()) if err != nil { zap.L().Error("BatchDownload - 读取CallFileTaskStandbyList失败", zap.String("id", payload.Id.Hex()), zap.String("err", err.Error())) return nil, err } if len(standbyList) == 0 { break } // 等待1分钟后再判断 time.Sleep(time.Minute) } // 处理失败列表 failedList, err := repositories.GetBatchDownloadCallFileTaskFailedList(payload.Id.Hex()) if err != nil { zap.L().Error("BatchDownload - 读取CallFileTaskFailedList失败", zap.String("id", payload.Id.Hex()), zap.String("err", err.Error())) return nil, err } // 删除task list repositories.DeleteBatchDownloadCallFileTaskList(payload.Id.Hex()) var failReasons []string = []string{} for _, f := range failedList { failReasons = append(failReasons, f.Reason) } return failReasons, nil } return nil, nil } // 插入批量数据下载队列 // payload - 任务参数 func EnqueueBatchDownloadTask(payload domain.TaskBatchDownload) error { p := TaskPayload[domain.TaskBatchDownload]{ TaskPayloadBase: TaskPayloadBase{ Type: TaskTypeBatchDownload, }, Payload: payload, } if err := EnqueueTask[domain.TaskBatchDownload](p); err != nil { logger.Std.Printf("EnqueueBatchDownloadTask - %s : %s\n", payload.Id, err.Error()) return err } return nil } func requestCallFaultRecord(downloadList *[]domain.BatchDownloadFileTaskInfo, devType int, sn string, startTime time.Time, endTime time.Time) error { // 召唤aGate、aPower故障录波 // 查询要召唤的选项 callInfoList, err := repositories.QueryFaultRecordCallableList(devType, sn) if err != nil { zap.L().Error("查询站点故障录波召唤选项失败", zap.String("sn", sn), zap.String("err", err.Error())) return err } // 未召唤到记录,不添加任务 if len(callInfoList) == 0 { return nil } // 提取符合日期范围的选项 var indexList string = "" for _, info := range callInfoList { // 转换成标准时间格式 parts := strings.Split(info.FailureTime, " ") date := parts[0] timeStr := parts[1] d := strings.Split(date, "-") t2 := strings.Split(timeStr, ":") failureStandardTime := fmt.Sprintf("%04s-%02s-%02s %02s:%02s:%02s", d[0], d[1], d[2], t2[0], t2[1], t2[2]) failureTime, err := time.Parse(time.DateTime, failureStandardTime) if err != nil { zap.L().Error("故障录波召唤选项格式错误", zap.String("sn", sn), zap.String("FailureTime", info.FailureTime), zap.String("err", err.Error())) return err } // 判断是否在范围内(包含边界) inRange := (failureTime.Equal(startTime) || failureTime.After(startTime)) && (failureTime.Equal(endTime) || failureTime.Before(endTime)) if inRange { if indexList == "" { indexList = indexList + strconv.Itoa(info.Idx) } else { indexList = indexList + "," + strconv.Itoa(info.Idx) } } } *downloadList = append(*downloadList, domain.BatchDownloadFileTaskInfo{ DevType: devType, Sn: sn, Date: startTime.Format(time.DateTime) + "," + endTime.Format(time.DateTime), // 故障录波的日期是一个范围,用","隔开 FileType: jobs.FileTypeFaultRecord, FileSupType: indexList, Result: 0, }) return nil }