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 }