package syncdrive import ( "encoding/json" "fmt" "github.com/tickstep/aliyunpan/internal/config" "github.com/tickstep/aliyunpan/internal/log" "github.com/tickstep/aliyunpan/internal/utils" "github.com/tickstep/library-go/logger" "io/ioutil" "path" "strings" "time" ) type ( // SyncOption 同步选项 SyncOption struct { FileDownloadParallel int // 文件下载并发数 FileUploadParallel int // 文件上传并发数 FileDownloadBlockSize int64 // 文件下载分片大小 FileUploadBlockSize int64 // 文件上传分片大小 MaxDownloadRate int64 // 限制最大下载速度 MaxUploadRate int64 // 限制最大上传速度 // 优先级选项 SyncPriority SyncPriorityOption // 本地文件修改检测间隔 LocalFileModifiedCheckIntervalSec int // 文件记录器 FileRecorder *log.FileRecorder } // SyncTaskManager 同步任务管理器 SyncTaskManager struct { syncDriveConfig *SyncDriveConfig syncOption SyncOption PanUser *config.PanUser PanClient *config.PanClient SyncConfigFolderPath string // useConfigFile 是否使用配置文件启动 useConfigFile bool } // SyncDriveConfig 同步盘配置文件 SyncDriveConfig struct { ConfigVer string `json:"configVer"` SyncTaskList []*SyncTask `json:"syncTaskList"` } ) var ( ErrSyncTaskListEmpty error = fmt.Errorf("no sync task") ) func NewSyncTaskManager(user *config.PanUser, panClient *config.PanClient, syncConfigFolderPath string, option SyncOption) *SyncTaskManager { return &SyncTaskManager{ PanUser: user, PanClient: panClient, SyncConfigFolderPath: syncConfigFolderPath, syncOption: option, } } func (m *SyncTaskManager) parseConfigFile() error { /** 样例 { "configVer": "1.0", "syncTaskList": [ { "name": "NS游戏备份", "id": "5b2d7c10-e927-4e72-8f9d-5abb3bb04814", "driveId": "19519111", "localFolderPath": "D:\\smb\\datadisk\\game", "panFolderPath": "/sync_drive/game", "mode": "sync", "driveName": "backup", "lastSyncTime": "" } ] } */ configFilePath := m.ConfigFilePath() r := &SyncDriveConfig{ ConfigVer: "1.0", SyncTaskList: []*SyncTask{}, } m.syncDriveConfig = r if b, _ := utils.PathExists(configFilePath); b != true { //text := utils.ObjectToJsonStr(r, true) //ioutil.WriteFile(ConfigFilePath, []byte(text), 0755) return fmt.Errorf("备份配置文件不存在:" + m.ConfigFilePath()) } data, e := ioutil.ReadFile(configFilePath) if e != nil { return e } if len(data) > 0 { if err2 := json.Unmarshal(data, m.syncDriveConfig); err2 != nil { logger.Verboseln("parse sync drive config json error ", err2) return err2 } } return nil } func (m *SyncTaskManager) ConfigFilePath() string { return path.Join(m.SyncConfigFolderPath, "sync_drive_config.json") } // Start 启动同步进程 func (m *SyncTaskManager) Start(tasks []*SyncTask, cycleMode CycleMode, scanTimeInterval int64) (bool, error) { if tasks != nil && len(tasks) > 0 { m.syncDriveConfig = &SyncDriveConfig{ ConfigVer: "1.0", SyncTaskList: []*SyncTask{}, } m.syncDriveConfig.SyncTaskList = tasks m.useConfigFile = false } else { if er := m.parseConfigFile(); er != nil { return false, er } m.useConfigFile = true } if m.syncDriveConfig.SyncTaskList == nil || len(m.syncDriveConfig.SyncTaskList) == 0 { return false, ErrSyncTaskListEmpty } // start the sync task one by one for _, task := range m.syncDriveConfig.SyncTaskList { if len(task.Id) == 0 { task.Id = utils.UuidStr() } // cycle mode task.CycleModeType = cycleMode // scan time task.ScanTimeInterval = scanTimeInterval // check driveId if strings.ToLower(task.DriveName) == "backup" { task.DriveId = m.PanUser.DriveList.GetFileDriveId() } else if strings.ToLower(task.DriveName) == "resource" { task.DriveId = m.PanUser.DriveList.GetResourceDriveId() } if len(task.DriveId) == 0 { task.DriveId = m.PanUser.DriveList.GetFileDriveId() } // check userId if len(task.UserId) > 0 { if task.UserId != m.PanUser.UserId { // not this user task, skip logger.Verboseln("skip sync task: " + task.NameLabel()) continue } } else { task.UserId = m.PanUser.UserId } // check pan path if !utils.IsPanAbsPath(task.PanFolderPath) { task.PanFolderPath = "/" + task.PanFolderPath } // check local path if !utils.IsLocalAbsPath(task.LocalFolderPath) { fmt.Println("任务启动失败,本地路径不是绝对路径: ", task.LocalFolderPath) continue } task.panUser = m.PanUser task.syncDbFolderPath = m.SyncConfigFolderPath task.panClient = m.PanClient task.syncOption = m.syncOption if task.Priority != "" { task.syncOption.SyncPriority = task.Priority } else { task.Priority = SyncPriorityTimestampFirst task.syncOption.SyncPriority = SyncPriorityTimestampFirst } if task.Policy == "" { task.Policy = SyncPolicyIncrement } task.LocalFolderPath = path.Clean(task.LocalFolderPath) task.PanFolderPath = path.Clean(task.PanFolderPath) if e := task.Start(); e != nil { logger.Verboseln(e) fmt.Printf("启动同步任务[%s]出错: %s\n", task.Id, e.Error()) continue } fmt.Println("\n启动同步任务") fmt.Println(task) time.Sleep(200 * time.Millisecond) } // save config file if m.useConfigFile { ioutil.WriteFile(m.ConfigFilePath(), []byte(utils.ObjectToJsonStr(m.syncDriveConfig, true)), 0755) } return true, nil } // Stop 停止同步进程 func (m *SyncTaskManager) Stop() (bool, error) { // stop task one by one for _, task := range m.syncDriveConfig.SyncTaskList { var e error e = task.Stop() if e != nil { logger.Verboseln(e) fmt.Println("stop sync task error: ", task.NameLabel()) continue } fmt.Println("正在停止同步任务: ", task.NameLabel()) } // save config file if m.useConfigFile { ioutil.WriteFile(m.ConfigFilePath(), []byte(utils.ObjectToJsonStr(m.syncDriveConfig, true)), 0755) } return true, nil } func (m *SyncTaskManager) IsAllTaskCompletely() bool { for _, task := range m.syncDriveConfig.SyncTaskList { if !task.IsTaskCompletely() { return false } } return true }