wip: use items in offline_download

internal/offline_download/tool/add.go

@@ -0,0 +1,80 @@
+package tool
+
+import (
+	"context"
+	"fmt"
+	"path/filepath"
+
+	"github.com/alist-org/alist/v3/internal/conf"
+	"github.com/alist-org/alist/v3/internal/errs"
+	"github.com/alist-org/alist/v3/internal/op"
+	"github.com/alist-org/alist/v3/pkg/task"
+	"github.com/google/uuid"
+	"github.com/pkg/errors"
+)
+
+type AddURIArgs struct {
+	URI        string
+	DstDirPath string
+	Tool       string
+}
+
+func AddURI(ctx context.Context, args *AddURIArgs) error {
+	// get tool
+	tool, err := Tools.Get(args.Tool)
+	if err != nil {
+		return errors.Wrapf(err, "failed get tool")
+	}
+	// check tool is ready
+	if !tool.IsReady() {
+		return errors.Wrapf(err, "tool %s is not ready", args.Tool)
+	}
+	// check storage
+	storage, dstDirActualPath, err := op.GetStorageAndActualPath(args.DstDirPath)
+	if err != nil {
+		return errors.WithMessage(err, "failed get storage")
+	}
+	// check is it could upload
+	if storage.Config().NoUpload {
+		return errors.WithStack(errs.UploadNotSupported)
+	}
+	// check path is valid
+	obj, err := op.Get(ctx, storage, dstDirActualPath)
+	if err != nil {
+		if !errs.IsObjectNotFound(err) {
+			return errors.WithMessage(err, "failed get object")
+		}
+	} else {
+		if !obj.IsDir() {
+			// can't add to a file
+			return errors.WithStack(errs.NotFolder)
+		}
+	}
+
+	uid := uuid.NewString()
+	tempDir := filepath.Join(conf.Conf.TempDir, args.Tool, uid)
+	signal := make(chan int)
+	gid, err := tool.AddURI(&AddUriArgs{
+		Uri:     args.URI,
+		UID:     uid,
+		TempDir: tempDir,
+		Signal:  signal,
+	})
+	if err != nil {
+		return errors.Wrapf(err, "[%s] failed to add uri %s", args.Tool, args.URI)
+	}
+	DownTaskManager.Submit(task.WithCancelCtx(&task.Task[string]{
+		ID:   gid,
+		Name: fmt.Sprintf("download %s to [%s](%s)", args.URI, storage.GetStorage().MountPath, dstDirActualPath),
+		Func: func(tsk *task.Task[string]) error {
+			m := &Monitor{
+				tsk:        tsk,
+				tempDir:    tempDir,
+				dstDirPath: args.DstDirPath,
+				signal:     signal,
+			}
+			return m.Loop()
+		},
+	}))
+	return nil
+}

internal/offline_download/tool/all_test.go

@@ -0,0 +1,17 @@
+package tool_test
+
+import (
+	"testing"
+
+	"github.com/alist-org/alist/v3/internal/offline_download/tool"
+)
+
+func TestGetFiles(t *testing.T) {
+	files, err := tool.GetFiles("..")
+	if err != nil {
+		t.Fatal(err)
+	}
+	for _, file := range files {
+		t.Log(file.Name, file.Size, file.Path, file.Modified)
+	}
+}

internal/offline_download/tool/base.go

@@ -0,0 +1,58 @@
+package tool
+
+import (
+	"io"
+	"os"
+	"time"
+
+	"github.com/alist-org/alist/v3/internal/model"
+)
+
+type AddUriArgs struct {
+	Uri     string
+	UID     string
+	TempDir string
+	Signal  chan int
+}
+
+type Status struct {
+	Progress  float64
+	NewTID    string
+	Completed bool
+	Status    string
+	Err       error
+}
+
+type Tool interface {
+	// Items return the setting items the tool need
+	Items() []model.SettingItem
+	Init() (string, error)
+	IsReady() bool
+	// AddURI add an uri to download, return the task id
+	AddURI(args *AddUriArgs) (string, error)
+	// Remove the download if task been canceled
+	Remove(tid string) error
+	// Status return the status of the download task, if an error occurred, return the error in Status.Err
+	Status(tid string) (*Status, error)
+	// GetFile return an io.ReadCloser as the download file, if nil, means walk the temp dir to get the files
+	GetFile(tid string) *File
+}
+
+type File struct {
+	io.ReadCloser
+	Name     string
+	Size     int64
+	Path     string
+	Modified time.Time
+}
+
+func (f *File) GetReadCloser() (io.ReadCloser, error) {
+	if f.ReadCloser != nil {
+		return f.ReadCloser, nil
+	}
+	file, err := os.Open(f.Path)
+	if err != nil {
+		return nil, err
+	}
+	return file, nil
+}

internal/offline_download/tool/monitor.go

@@ -0,0 +1,160 @@
+package tool
+
+import (
+	"fmt"
+	"os"
+	"path"
+	"path/filepath"
+	"sync"
+	"sync/atomic"
+	"time"
+
+	"github.com/alist-org/alist/v3/internal/model"
+	"github.com/alist-org/alist/v3/internal/op"
+	"github.com/alist-org/alist/v3/internal/stream"
+	"github.com/alist-org/alist/v3/pkg/task"
+	"github.com/alist-org/alist/v3/pkg/utils"
+	"github.com/pkg/errors"
+	log "github.com/sirupsen/logrus"
+)
+
+type Monitor struct {
+	tool       Tool
+	tsk        *task.Task[string]
+	tempDir    string
+	retried    int
+	dstDirPath string
+	finish     chan struct{}
+	signal     chan int
+}
+
+func (m *Monitor) Loop() error {
+	m.finish = make(chan struct{})
+	var (
+		err error
+		ok  bool
+	)
+outer:
+	for {
+		select {
+		case <-m.tsk.Ctx.Done():
+			err := m.tool.Remove(m.tsk.ID)
+			return err
+		case <-m.signal:
+			ok, err = m.Update()
+			if ok {
+				break outer
+			}
+		case <-time.After(time.Second * 2):
+			ok, err = m.Update()
+			if ok {
+				break outer
+			}
+		}
+	}
+	if err != nil {
+		return err
+	}
+	m.tsk.SetStatus("aria2 download completed, transferring")
+	<-m.finish
+	m.tsk.SetStatus("completed")
+	return nil
+}
+
+// Update download status, return true if download completed
+func (m *Monitor) Update() (bool, error) {
+	info, err := m.tool.Status(m.tsk.ID)
+	if err != nil {
+		m.retried++
+		log.Errorf("failed to get status of %s, retried %d times", m.tsk.ID, m.retried)
+		return false, nil
+	}
+	if m.retried > 5 {
+		return true, errors.Errorf("failed to get status of %s, retried %d times", m.tsk.ID, m.retried)
+	}
+	m.retried = 0
+	m.tsk.SetProgress(info.Progress)
+	m.tsk.SetStatus("tool: " + info.Status)
+	if info.NewTID != "" {
+		log.Debugf("followen by: %+v", info.NewTID)
+		DownTaskManager.RawTasks().Delete(m.tsk.ID)
+		m.tsk.ID = info.NewTID
+		DownTaskManager.RawTasks().Store(m.tsk.ID, m.tsk)
+		return false, nil
+	}
+	// if download completed
+	if info.Completed {
+		err := m.Complete()
+		return true, errors.WithMessage(err, "failed to transfer file")
+	}
+	// if download failed
+	if info.Err != nil {
+		return true, errors.Errorf("failed to download %s, error: %s", m.tsk.ID, info.Err.Error())
+	}
+	return false, nil
+}
+
+var TransferTaskManager = task.NewTaskManager(3, func(k *uint64) {
+	atomic.AddUint64(k, 1)
+})
+
+func (m *Monitor) Complete() error {
+	// check dstDir again
+	storage, dstDirActualPath, err := op.GetStorageAndActualPath(m.dstDirPath)
+	if err != nil {
+		return errors.WithMessage(err, "failed get storage")
+	}
+	var files []*File
+	if f := m.tool.GetFile(m.tsk.ID); f != nil {
+		files = append(files, f)
+	} else {
+		files, err = GetFiles(m.tempDir)
+		if err != nil {
+			return errors.Wrapf(err, "failed to get files")
+		}
+	}
+	// upload files
+	var wg sync.WaitGroup
+	wg.Add(len(files))
+	go func() {
+		wg.Wait()
+		err := os.RemoveAll(m.tempDir)
+		m.finish <- struct{}{}
+		if err != nil {
+			log.Errorf("failed to remove aria2 temp dir: %+v", err.Error())
+		}
+	}()
+	for i, _ := range files {
+		file := files[i]
+		TransferTaskManager.Submit(task.WithCancelCtx(&task.Task[uint64]{
+			Name: fmt.Sprintf("transfer %s to [%s](%s)", file.Path, storage.GetStorage().MountPath, dstDirActualPath),
+			Func: func(tsk *task.Task[uint64]) error {
+				defer wg.Done()
+				mimetype := utils.GetMimeType(file.Path)
+				rc, err := file.GetReadCloser()
+				if err != nil {
+					return errors.Wrapf(err, "failed to open file %s", file.Path)
+				}
+				s := &stream.FileStream{
+					Ctx: nil,
+					Obj: &model.Object{
+						Name:     path.Base(file.Path),
+						Size:     file.Size,
+						Modified: file.Modified,
+						IsFolder: false,
+					},
+					Reader:   rc,
+					Mimetype: mimetype,
+					Closers:  utils.NewClosers(rc),
+				}
+				relDir, err := filepath.Rel(m.tempDir, filepath.Dir(file.Path))
+				if err != nil {
+					log.Errorf("find relation directory error: %v", err)
+				}
+				newDistDir := filepath.Join(dstDirActualPath, relDir)
+				return op.Put(tsk.Ctx, storage, newDistDir, s, tsk.SetProgress)
+			},
+		}))
+	}
+	return nil
+}

internal/offline_download/tool/tools.go

@@ -0,0 +1,42 @@
+package tool
+
+import (
+	"fmt"
+
+	"github.com/alist-org/alist/v3/internal/model"
+	"github.com/alist-org/alist/v3/pkg/task"
+)
+
+var (
+	Tools           = make(ToolsManager)
+	DownTaskManager = task.NewTaskManager[string](3)
+)
+
+type ToolsManager map[string]Tool
+
+func (t ToolsManager) Get(name string) (Tool, error) {
+	if tool, ok := t[name]; ok {
+		return tool, nil
+	}
+	return nil, fmt.Errorf("tool %s not found", name)
+}
+
+func (t ToolsManager) Add(name string, tool Tool) {
+	t[name] = tool
+}
+
+func (t ToolsManager) Names() []string {
+	names := make([]string, 0, len(t))
+	for name := range t {
+		names = append(names, name)
+	}
+	return names
+}
+
+func (t ToolsManager) Items() []model.SettingItem {
+	var items []model.SettingItem
+	for _, tool := range t {
+		items = append(items, tool.Items()...)
+	}
+	return items
+}

internal/offline_download/tool/util.go

@@ -0,0 +1,28 @@
+package tool
+
+import (
+	"os"
+	"path/filepath"
+)
+
+func GetFiles(dir string) ([]*File, error) {
+	var files []*File
+	err := filepath.Walk(dir, func(path string, info os.FileInfo, err error) error {
+		if err != nil {
+			return err
+		}
+		if !info.IsDir() {
+			files = append(files, &File{
+				Name:     info.Name(),
+				Size:     info.Size(),
+				Path:     path,
+				Modified: info.ModTime(),
+			})
+		}
+		return nil
+	})
+	if err != nil {
+		return nil, err
+	}
+	return files, nil
+}