package services import ( "fmt" "mime/multipart" "os" "path/filepath" "sync" "server/models" ) // MigrationProgress 迁移进度 type MigrationProgress struct { Total int Success int Failed int Skipped int Current string Errors []string mu sync.Mutex } // AddSuccess 增加成功计数 func (p *MigrationProgress) AddSuccess() { p.mu.Lock() defer p.mu.Unlock() p.Success++ } // AddSkipped 增加跳过计数 func (p *MigrationProgress) AddSkipped() { p.mu.Lock() defer p.mu.Unlock() p.Skipped++ } // AddFailed 增加失败计数 func (p *MigrationProgress) AddFailed(err string) { p.mu.Lock() defer p.mu.Unlock() p.Failed++ p.Errors = append(p.Errors, err) } // SetCurrent 设置当前处理的文件 func (p *MigrationProgress) SetCurrent(filename string) { p.mu.Lock() defer p.mu.Unlock() p.Current = filename } // GetProgress 获取进度信息 func (p *MigrationProgress) GetProgress() (int, int, int, string) { p.mu.Lock() defer p.mu.Unlock() return p.Total, p.Success, p.Failed, p.Current } // StorageMigration 存储迁移服务 type StorageMigration struct { fromService StorageService toService StorageService progress *MigrationProgress } // NewStorageMigration 创建存储迁移服务 func NewStorageMigration(from, to StorageService) *StorageMigration { return &StorageMigration{ fromService: from, toService: to, progress: &MigrationProgress{ Errors: make([]string, 0), }, } } // TargetKey 计算某条文件记录改造后的新 key func TargetKey(f *models.SystemFile) string { source := f.Source if source == "" { source = SourceBackend } var tuid uint64 if f.Tuid != nil { tuid = *f.Tuid } return BuildObjectKey(UploadContext{Source: source, Tid: f.Tid, Tuid: tuid}, FileExt(f.Name)) } // MigrateFile 迁移单个文件到新的分层目录 // // 同类型存储(七牛→七牛 / 本地→本地)走服务端改名,不重新上传、不消耗流量; // 本地→七牛 走"读取 + 上传",其余跨存储方向暂不支持。 func (m *StorageMigration) MigrateFile(file *models.SystemFile) error { m.progress.SetCurrent(file.Name) oldKey := file.ObjectKey if oldKey == "" { oldKey = KeyFromSrc(file.Src, m.fromService) } if oldKey == "" { return fmt.Errorf("无法解析原存储路径: %s", file.Src) } newKey := TargetKey(file) if oldKey == newKey { m.progress.AddSkipped() return nil } if m.fromService.Type() == m.toService.Type() { if err := m.toService.Move(oldKey, newKey); err != nil { return fmt.Errorf("移动文件失败: %w", err) } } else { // 跨存储:仅支持 本地 → 七牛 localFrom, ok := m.fromService.(*LocalStorage) if !ok { return fmt.Errorf("暂不支持从 %s 迁出到 %s", m.fromService.Type(), m.toService.Type()) } localPath := filepath.Join(localFrom.BaseDir, filepath.FromSlash(oldKey)) f, err := os.Open(localPath) if err != nil { return fmt.Errorf("打开本地文件失败: %w", err) } defer f.Close() stat, err := f.Stat() if err != nil { return fmt.Errorf("获取文件信息失败: %w", err) } header := &multipart.FileHeader{Filename: file.Name, Size: stat.Size()} staged, err := m.toService.Stage(f, header) if err != nil { return err } source := file.Source if source == "" { source = SourceBackend } var tuid uint64 if file.Tuid != nil { tuid = *file.Tuid } if _, err := m.toService.Commit(staged, UploadContext{Source: source, Tid: file.Tid, Tuid: tuid}); err != nil { _ = m.toService.Discard(staged) return err } } newSrc := m.toService.GetPublicURL(newKey) if _, err := models.Orm.QueryTable(new(models.SystemFile)). Filter("id", file.ID). Update(map[string]interface{}{ "src": newSrc, "object_key": newKey, "storage": m.toService.Type(), }); err != nil { return fmt.Errorf("更新数据库失败: %w", err) } m.progress.AddSuccess() return nil } // MigrateAll 迁移所有文件 func (m *StorageMigration) MigrateAll(tid uint64) error { var files []models.SystemFile qs := models.Orm.QueryTable(new(models.SystemFile)).Filter("delete_time__isnull", true) if tid > 0 { qs = qs.Filter("tid", tid) } if _, err := qs.All(&files); err != nil { return fmt.Errorf("获取文件列表失败: %w", err) } m.progress.Total = len(files) concurrency := 5 sem := make(chan struct{}, concurrency) var wg sync.WaitGroup for i := range files { wg.Add(1) go func(file *models.SystemFile) { defer wg.Done() sem <- struct{}{} defer func() { <-sem }() if err := m.MigrateFile(file); err != nil { m.progress.AddFailed(fmt.Sprintf("%s: %v", file.Name, err)) } }(&files[i]) } wg.Wait() return nil } // GetProgress 获取迁移进度 func (m *StorageMigration) GetProgress() *MigrationProgress { return m.progress } // MigrateLocalToQiniu 从本地存储迁移到七牛云 func MigrateLocalToQiniu(tid uint64) (*MigrationProgress, error) { cfg, err := models.GetStorageConfig() if err != nil { return nil, fmt.Errorf("获取存储配置失败: %w", err) } if cfg.StorageType != StorageTypeQiniu { return nil, fmt.Errorf("当前存储类型不是七牛云") } migration := NewStorageMigration(NewLocalStorage(), NewQiniuStorage(cfg)) if err := migration.MigrateAll(tid); err != nil { return migration.GetProgress(), err } return migration.GetProgress(), nil }