feat(drivers): add teldrive

drivers/all.go

@@ -61,6 +61,7 @@ import (
 	_ "github.com/OpenListTeam/OpenList/v4/drivers/smb"
 	_ "github.com/OpenListTeam/OpenList/v4/drivers/strm"
 	_ "github.com/OpenListTeam/OpenList/v4/drivers/teambition"
+	_ "github.com/OpenListTeam/OpenList/v4/drivers/teldrive"
 	_ "github.com/OpenListTeam/OpenList/v4/drivers/terabox"
 	_ "github.com/OpenListTeam/OpenList/v4/drivers/thunder"
 	_ "github.com/OpenListTeam/OpenList/v4/drivers/thunder_browser"

drivers/teldrive/copy.go

@@ -0,0 +1,137 @@
+package teldrive
+
+import (
+	"fmt"
+	"net/http"
+
+	"github.com/OpenListTeam/OpenList/v4/drivers/base"
+	"github.com/OpenListTeam/OpenList/v4/internal/model"
+	"github.com/OpenListTeam/OpenList/v4/pkg/utils"
+	"github.com/go-resty/resty/v2"
+	"golang.org/x/net/context"
+	"golang.org/x/sync/errgroup"
+	"golang.org/x/sync/semaphore"
+)
+
+func NewCopyManager(ctx context.Context, concurrent int, d *Teldrive) *CopyManager {
+	g, ctx := errgroup.WithContext(ctx)
+
+	return &CopyManager{
+		TaskChan: make(chan CopyTask, concurrent*2),
+		Sem:      semaphore.NewWeighted(int64(concurrent)),
+		G:        g,
+		Ctx:      ctx,
+		d:        d,
+	}
+}
+
+func (cm *CopyManager) startWorkers() {
+	workerCount := cap(cm.TaskChan) / 2
+	for i := 0; i < workerCount; i++ {
+		cm.G.Go(func() error {
+			return cm.worker()
+		})
+	}
+}
+
+func (cm *CopyManager) worker() error {
+	for {
+		select {
+		case task, ok := <-cm.TaskChan:
+			if !ok {
+				return nil
+			}
+
+			if err := cm.Sem.Acquire(cm.Ctx, 1); err != nil {
+				return err
+			}
+
+			var err error
+
+			err = cm.processFile(task)
+
+			cm.Sem.Release(1)
+
+			if err != nil {
+				return fmt.Errorf("task processing failed: %w", err)
+			}
+
+		case <-cm.Ctx.Done():
+			return cm.Ctx.Err()
+		}
+	}
+}
+
+func (cm *CopyManager) generateTasks(ctx context.Context, srcObj, dstDir model.Obj) error {
+	if srcObj.IsDir() {
+		return cm.generateFolderTasks(ctx, srcObj, dstDir)
+	} else {
+		// add single file task directly
+		select {
+		case cm.TaskChan <- CopyTask{SrcObj: srcObj, DstDir: dstDir}:
+			return nil
+		case <-ctx.Done():
+			return ctx.Err()
+		}
+	}
+}
+
+func (cm *CopyManager) generateFolderTasks(ctx context.Context, srcDir, dstDir model.Obj) error {
+	objs, err := cm.d.List(ctx, srcDir, model.ListArgs{})
+	if err != nil {
+		return fmt.Errorf("failed to list directory %s: %w", srcDir.GetPath(), err)
+	}
+
+	err = cm.d.MakeDir(cm.Ctx, dstDir, srcDir.GetName())
+	if err != nil || len(objs) == 0 {
+		return err
+	}
+	newDstDir := &model.Object{
+		ID:       dstDir.GetID(),
+		Path:     dstDir.GetPath() + "/" + srcDir.GetName(),
+		Name:     srcDir.GetName(),
+		IsFolder: true,
+	}
+
+	for _, file := range objs {
+		if utils.IsCanceled(ctx) {
+			return ctx.Err()
+		}
+
+		srcFile := &model.Object{
+			ID:       file.GetID(),
+			Path:     srcDir.GetPath() + "/" + file.GetName(),
+			Name:     file.GetName(),
+			IsFolder: file.IsDir(),
+		}
+
+		// 递归生成任务
+		if err := cm.generateTasks(ctx, srcFile, newDstDir); err != nil {
+			return err
+		}
+	}
+
+	return nil
+}
+
+func (cm *CopyManager) processFile(task CopyTask) error {
+	return cm.copySingleFile(cm.Ctx, task.SrcObj, task.DstDir)
+}
+
+func (cm *CopyManager) copySingleFile(ctx context.Context, srcObj, dstDir model.Obj) error {
+	// `override copy mode` should delete the existing file
+	if obj, err := cm.d.getFile(dstDir.GetPath(), srcObj.GetName(), srcObj.IsDir()); err == nil {
+		if err := cm.d.Remove(ctx, obj); err != nil {
+			return fmt.Errorf("failed to remove existing file: %w", err)
+		}
+	}
+
+	// Do copy
+	return cm.d.request(http.MethodPost, "/api/files/{id}/copy", func(req *resty.Request) {
+		req.SetPathParam("id", srcObj.GetID())
+		req.SetBody(base.Json{
+			"newName":     srcObj.GetName(),
+			"destination": dstDir.GetPath(),
+		})
+	}, nil)
+}

drivers/teldrive/driver.go

@@ -0,0 +1,217 @@
+package teldrive
+
+import (
+	"context"
+	"fmt"
+	"math"
+	"net/http"
+	"net/url"
+	"strings"
+
+	"github.com/OpenListTeam/OpenList/v4/drivers/base"
+	"github.com/OpenListTeam/OpenList/v4/internal/driver"
+	"github.com/OpenListTeam/OpenList/v4/internal/errs"
+	"github.com/OpenListTeam/OpenList/v4/internal/model"
+	"github.com/OpenListTeam/OpenList/v4/internal/op"
+	"github.com/OpenListTeam/OpenList/v4/pkg/utils"
+	"github.com/go-resty/resty/v2"
+	"github.com/google/uuid"
+)
+
+type Teldrive struct {
+	model.Storage
+	Addition
+}
+
+func (d *Teldrive) Config() driver.Config {
+	return config
+}
+
+func (d *Teldrive) GetAddition() driver.Additional {
+	return &d.Addition
+}
+
+func (d *Teldrive) Init(ctx context.Context) error {
+	d.Address = strings.TrimSuffix(d.Address, "/")
+	if d.Cookie == "" || !strings.HasPrefix(d.Cookie, "access_token=") {
+		return fmt.Errorf("cookie must start with 'access_token='")
+	}
+	if d.UploadConcurrency == 0 {
+		d.UploadConcurrency = 4
+	}
+	if d.ChunkSize == 0 {
+		d.ChunkSize = 10
+	}
+
+	op.MustSaveDriverStorage(d)
+	return nil
+}
+
+func (d *Teldrive) Drop(ctx context.Context) error {
+	return nil
+}
+
+func (d *Teldrive) List(ctx context.Context, dir model.Obj, args model.ListArgs) ([]model.Obj, error) {
+	var listResp ListResp
+	err := d.request(http.MethodGet, "/api/files", func(req *resty.Request) {
+		req.SetQueryParams(map[string]string{
+			"path":  dir.GetPath(),
+			"limit": "1000", // overide default 500, TODO pagination
+		})
+	}, &listResp)
+	if err != nil {
+		return nil, err
+	}
+
+	return utils.SliceConvert(listResp.Items, func(src Object) (model.Obj, error) {
+		return &model.Object{
+			ID:   src.ID,
+			Name: src.Name,
+			Size: func() int64 {
+				if src.Type == "folder" {
+					return 0
+				}
+				return src.Size
+			}(),
+			IsFolder: src.Type == "folder",
+			Modified: src.UpdatedAt,
+		}, nil
+	})
+}
+
+func (d *Teldrive) Link(ctx context.Context, file model.Obj, args model.LinkArgs) (*model.Link, error) {
+	if d.UseShareLink {
+		shareObj, err := d.getShareFileById(file.GetID())
+		if err != nil || shareObj == nil {
+			if err := d.createShareFile(file.GetID()); err != nil {
+				return nil, err
+			}
+			shareObj, err = d.getShareFileById(file.GetID())
+			if err != nil {
+				return nil, err
+			}
+		}
+		return &model.Link{
+			URL: d.Address + "/api/shares/" + url.PathEscape(shareObj.Id) + "/files/" + url.PathEscape(file.GetID()) + "/" + url.PathEscape(file.GetName()),
+		}, nil
+	}
+	return &model.Link{
+		URL: d.Address + "/api/files/" + url.PathEscape(file.GetID()) + "/" + url.PathEscape(file.GetName()),
+		Header: http.Header{
+			"Cookie": {d.Cookie},
+		},
+	}, nil
+}
+
+func (d *Teldrive) MakeDir(ctx context.Context, parentDir model.Obj, dirName string) error {
+	return d.request(http.MethodPost, "/api/files/mkdir", func(req *resty.Request) {
+		req.SetBody(map[string]interface{}{
+			"path": parentDir.GetPath() + "/" + dirName,
+		})
+	}, nil)
+}
+
+func (d *Teldrive) Move(ctx context.Context, srcObj, dstDir model.Obj) error {
+	body := base.Json{
+		"ids":               []string{srcObj.GetID()},
+		"destinationParent": dstDir.GetID(),
+	}
+	return d.request(http.MethodPost, "/api/files/move", func(req *resty.Request) {
+		req.SetBody(body)
+	}, nil)
+}
+
+func (d *Teldrive) Rename(ctx context.Context, srcObj model.Obj, newName string) error {
+	body := base.Json{
+		"name": newName,
+	}
+	return d.request(http.MethodPatch, "/api/files/{id}", func(req *resty.Request) {
+		req.SetPathParam("id", srcObj.GetID())
+		req.SetBody(body)
+	}, nil)
+}
+
+func (d *Teldrive) Copy(ctx context.Context, srcObj, dstDir model.Obj) error {
+	copyConcurrentLimit := 4
+	copyManager := NewCopyManager(ctx, copyConcurrentLimit, d)
+	copyManager.startWorkers()
+	copyManager.G.Go(func() error {
+		defer close(copyManager.TaskChan)
+		return copyManager.generateTasks(ctx, srcObj, dstDir)
+	})
+	return copyManager.G.Wait()
+}
+
+func (d *Teldrive) Remove(ctx context.Context, obj model.Obj) error {
+	body := base.Json{
+		"ids": []string{obj.GetID()},
+	}
+	return d.request(http.MethodPost, "/api/files/delete", func(req *resty.Request) {
+		req.SetBody(body)
+	}, nil)
+}
+
+func (d *Teldrive) Put(ctx context.Context, dstDir model.Obj, file model.FileStreamer, up driver.UpdateProgress) error {
+	fileId := uuid.New().String()
+	chunkSizeInMB := d.ChunkSize
+	chunkSize := chunkSizeInMB * 1024 * 1024 // Convert MB to bytes
+	totalSize := file.GetSize()
+	totalParts := int(math.Ceil(float64(totalSize) / float64(chunkSize)))
+	maxRetried := 3
+
+	// delete the upload task when finished or failed
+	defer func() {
+		_ = d.request(http.MethodDelete, "/api/uploads/{id}", func(req *resty.Request) {
+			req.SetPathParam("id", fileId)
+		}, nil)
+	}()
+
+	if obj, err := d.getFile(dstDir.GetPath(), file.GetName(), file.IsDir()); err == nil {
+		if err = d.Remove(ctx, obj); err != nil {
+			return err
+		}
+	}
+	// start the upload process
+	if err := d.request(http.MethodGet, "/api/uploads/fileId", func(req *resty.Request) {
+		req.SetPathParam("id", fileId)
+	}, nil); err != nil {
+		return err
+	}
+	if totalSize == 0 {
+		return d.touch(file.GetName(), dstDir.GetPath())
+	}
+
+	if totalParts <= 1 {
+		return d.doSingleUpload(ctx, dstDir, file, up, totalParts, chunkSize, fileId)
+	}
+
+	return d.doMultiUpload(ctx, dstDir, file, up, maxRetried, totalParts, chunkSize, fileId)
+}
+
+func (d *Teldrive) GetArchiveMeta(ctx context.Context, obj model.Obj, args model.ArchiveArgs) (model.ArchiveMeta, error) {
+	// TODO get archive file meta-info, return errs.NotImplement to use an internal archive tool, optional
+	return nil, errs.NotImplement
+}
+
+func (d *Teldrive) ListArchive(ctx context.Context, obj model.Obj, args model.ArchiveInnerArgs) ([]model.Obj, error) {
+	// TODO list args.InnerPath in the archive obj, return errs.NotImplement to use an internal archive tool, optional
+	return nil, errs.NotImplement
+}
+
+func (d *Teldrive) Extract(ctx context.Context, obj model.Obj, args model.ArchiveInnerArgs) (*model.Link, error) {
+	// TODO return link of file args.InnerPath in the archive obj, return errs.NotImplement to use an internal archive tool, optional
+	return nil, errs.NotImplement
+}
+
+func (d *Teldrive) ArchiveDecompress(ctx context.Context, srcObj, dstDir model.Obj, args model.ArchiveDecompressArgs) ([]model.Obj, error) {
+	// TODO extract args.InnerPath path in the archive srcObj to the dstDir location, optional
+	// a folder with the same name as the archive file needs to be created to store the extracted results if args.PutIntoNewDir
+	// return errs.NotImplement to use an internal archive tool
+	return nil, errs.NotImplement
+}
+
+//func (d *Teldrive) Other(ctx context.Context, args model.OtherArgs) (interface{}, error) {
+//	return nil, errs.NotSupport
+//}
+
+var _ driver.Driver = (*Teldrive)(nil)

drivers/teldrive/meta.go

@@ -0,0 +1,26 @@
+package teldrive
+
+import (
+	"github.com/OpenListTeam/OpenList/v4/internal/driver"
+	"github.com/OpenListTeam/OpenList/v4/internal/op"
+)
+
+type Addition struct {
+	driver.RootPath
+	Address           string `json:"url" required:"true"`
+	Cookie            string `json:"cookie" type:"string" required:"true" help:"access_token=xxx"`
+	UseShareLink      bool   `json:"use_share_link" type:"bool" default:"false" help:"Create share link when getting link to support 302. If disabled, you need to enable web proxy."`
+	ChunkSize         int64  `json:"chunk_size" type:"number" default:"10" help:"Chunk size in MiB"`
+	UploadConcurrency int64  `json:"upload_concurrency" type:"number" default:"4" help:"Concurrency upload requests"`
+}
+
+var config = driver.Config{
+	Name:        "Teldrive",
+	DefaultRoot: "/",
+}
+
+func init() {
+	op.RegisterDriver(func() driver.Driver {
+		return &Teldrive{}
+	})
+}