yunzerwebsiteallinone/go/services/storage_migration.go

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
}