FranklinWH-Toolbox/service/services/batchDownloadServices.go

package services

import (
	"archive/zip"
	"fwh-toolbox-server/domain"
	"fwh-toolbox-server/dto"
	"fwh-toolbox-server/repositories"
	"fwh-toolbox-server/taskQueue"
	"fwh-toolbox-server/utils/minio"
	"io"

	"go.mongodb.org/mongo-driver/bson/primitive"
	"go.uber.org/zap"
)

// 从数据库获取批量数据下载列表
func GetBatchDownloadManageList(page, size uint, sn, progress *string) (*dto.ResPage[domain.BatchDownloadInfo], error) {
	return repositories.GetBatchDownloadManageList(page, size, sn, progress)
}

// 插入一条批量数据下载配置到数据库
func InsertBatchDownloadConfig(writeConfig domain.BatchDownloadConfigWrite) error {
	id, err := repositories.InsertBatchDownloadConfig(writeConfig)
	if err != nil {
		return err
	}

	taskQueue.EnqueueBatchDownloadTask(domain.TaskBatchDownload{
		Id:         id,
		Sn:         writeConfig.Sn,
		TimePeriod: writeConfig.TimePeriod,
		Options:    writeConfig.Options,
	})

	return nil
}

// 更新一条批量数据下载配置到数据库
func UpdateBatchDownloadDescription(id primitive.ObjectID, desc string) error {
	err := repositories.UpdateBatchDownloadDescription(id, desc)
	if err != nil {
		return err
	}

	return nil
}

// 删除一条批量数据下载配置
func DeleteBatchDownloadConfig(id primitive.ObjectID) error {
	err := repositories.DeleteBatchDownloadConfig(id)
	if err != nil {
		return err
	}

	// TODO 需删除相关数据

	return nil
}

// 从数据库获取批量数据下载文件地址并打包成zip文件
func GetBatchDownloadData(id primitive.ObjectID) (*io.PipeReader, error) {
	dataPathMap, err := repositories.GetBatchDownloadData(id)
	if err != nil {
		return nil, err
	}

	pr, pw := io.Pipe()
	zw := zip.NewWriter(pw)

	go func() {
		defer zw.Close()
		defer pw.Close()
		var InsertZipFile = func(path string) error {
			obj, err := minio.GetObject(path)
			if err != nil {
				zap.L().Error("获取对象失败", zap.String("objName", path), zap.String("err", err.Error()))
				return err
			}

			zipFileWriter, err := zw.Create(path)
			if err != nil {
				zap.L().Error("创建zip条目失败", zap.String("path", path), zap.String("err", err.Error()))
				return err
			}

			if _, err := io.Copy(zipFileWriter, obj); err != nil {
				zap.L().Error("写入zip失败", zap.String("path", path), zap.String("err", err.Error()))
				return err
			}
			return nil
		}

		// map[sn]batchData
		for _, batchData := range dataPathMap {
			// PowerFlow
			// map[date]path
			for _, path := range batchData.PowerFlow {
				if err := InsertZipFile(path); err != nil {
					continue
				}
			}

			// MqttLogs
			// map[date]logs
			for _, logs := range batchData.MqttLogs {
				// map[logType]path
				for _, path := range logs {
					if err := InsertZipFile(path); err != nil {
						continue
					}
				}
			}

			// EventAlarm
			// map[date]path
			for _, path := range batchData.EventAlarm {
				if err := InsertZipFile(path); err != nil {
					continue
				}
			}

			// EmsLogs
			// map[date]logs
			for _, logs := range batchData.EmsLogs {
				// map[logType]path
				for _, path := range logs {
					if err := InsertZipFile(path); err != nil {
						continue
					}
				}
			}

			// FaultRecord
			// map[date]path
			for _, path := range batchData.FaultRecord {
				if err := InsertZipFile(path); err != nil {
					continue
				}
			}

			// UpdateLog
			if batchData.UpdateLog != "" {
				if err := InsertZipFile(batchData.UpdateLog); err != nil {
					continue
				}
			}

			// Settings
			if batchData.Settings != "" {
				if err := InsertZipFile(batchData.Settings); err != nil {
					continue
				}
			}
		}
	}()

	// 返回流给客户端
	return pr, nil
}