FranklinWH-Toolbox/service/taskQueue/taskBatchDownload.go

package taskQueue

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.String()))
		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.String()))
			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.String()), 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.String()), 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)
				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.String())
			if err != nil {
				zap.L().Error("BatchDownload - 读取CallFileTaskStandbyList失败", zap.String("id", payload.Id.String()), zap.String("err", err.Error()))
				return nil, err
			}

			if len(standbyList) == 0 {
				break
			}

			// 等待1分钟后再判断
			time.Sleep(time.Minute)
		}

		// 处理失败列表
		failedList, err := repositories.GetBatchDownloadCallFileTaskFailedList(payload.Id.String())
		if err != nil {
			zap.L().Error("BatchDownload - 读取CallFileTaskFailedList失败", zap.String("id", payload.Id.String()), zap.String("err", err.Error()))
			return nil, err
		}

		// 删除task list
		repositories.DeleteBatchDownloadCallFileTaskList(payload.Id.String())

		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
	}

	// 提取符合日期范围的选项
	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
}