github
/
alist
mirror of https://github.com/Xhofe/alist
1
0
Fork 0
Code Issues Releases Wiki Activity

feat: add task info to resp of add task api (close #5579)

pull/5643/head
Andy Hsu 2023-12-03 14:44:20 +08:00
parent 8bdfc7ac8e
commit 026e944cbb
8 changed files with 79 additions and 50 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

23
internal/fs/copy.go
View File

@ -39,23 +39,23 @@ var CopyTaskManager *tache.Manager[*CopyTask]
// Copy if in the same storage, call move method
// if not, add copy task
func _copy(ctx context.Context, srcObjPath, dstDirPath string, lazyCache ...bool) (bool, error) {
func _copy(ctx context.Context, srcObjPath, dstDirPath string, lazyCache ...bool) (tache.TaskWithInfo, error) {
srcStorage, srcObjActualPath, err := op.GetStorageAndActualPath(srcObjPath)
if err != nil {
return false, errors.WithMessage(err, "failed get src storage")
return nil, errors.WithMessage(err, "failed get src storage")
}
dstStorage, dstDirActualPath, err := op.GetStorageAndActualPath(dstDirPath)
if err != nil {
return false, errors.WithMessage(err, "failed get dst storage")
return nil, errors.WithMessage(err, "failed get dst storage")
}
// copy if in the same storage, just call driver.Copy
if srcStorage.GetStorage() == dstStorage.GetStorage() {
return false, op.Copy(ctx, srcStorage, srcObjActualPath, dstDirActualPath, lazyCache...)
return nil, op.Copy(ctx, srcStorage, srcObjActualPath, dstDirActualPath, lazyCache...)
}
if ctx.Value(conf.NoTaskKey) != nil {
srcObj, err := op.Get(ctx, srcStorage, srcObjActualPath)
if err != nil {
return false, errors.WithMessagef(err, "failed get src [%s] file", srcObjPath)
return nil, errors.WithMessagef(err, "failed get src [%s] file", srcObjPath)
}
if !srcObj.IsDir() {
// copy file directly
@ -63,7 +63,7 @@ func _copy(ctx context.Context, srcObjPath, dstDirPath string, lazyCache ...bool
Header: http.Header{},
})
if err != nil {
return false, errors.WithMessagef(err, "failed get [%s] link", srcObjPath)
return nil, errors.WithMessagef(err, "failed get [%s] link", srcObjPath)
}
fs := stream.FileStream{
Obj: srcObj,
@ -72,19 +72,20 @@ func _copy(ctx context.Context, srcObjPath, dstDirPath string, lazyCache ...bool
// any link provided is seekable
ss, err := stream.NewSeekableStream(fs, link)
if err != nil {
return false, errors.WithMessagef(err, "failed get [%s] stream", srcObjPath)
return nil, errors.WithMessagef(err, "failed get [%s] stream", srcObjPath)
}
return false, op.Put(ctx, dstStorage, dstDirActualPath, ss, nil, false)
return nil, op.Put(ctx, dstStorage, dstDirActualPath, ss, nil, false)
}
}
// not in the same storage
CopyTaskManager.Add(&CopyTask{
t := &CopyTask{
srcStorage: srcStorage,
dstStorage: dstStorage,
srcObjPath: srcObjActualPath,
dstDirPath: dstDirActualPath,
})
return true, nil
}
CopyTaskManager.Add(t)
return t, nil
}
func copyBetween2Storages(t *CopyTask, srcStorage, dstStorage driver.Driver, srcObjPath, dstDirPath string) error {

9
internal/fs/fs.go
View File

@ -6,6 +6,7 @@ import (
"github.com/alist-org/alist/v3/internal/model"
"github.com/alist-org/alist/v3/internal/op"
log "github.com/sirupsen/logrus"
"github.com/xhofe/tache"
)
// the param named path of functions in this package is a mount path
@ -68,7 +69,7 @@ func Move(ctx context.Context, srcPath, dstDirPath string, lazyCache ...bool) er
return err
}
func Copy(ctx context.Context, srcObjPath, dstDirPath string, lazyCache ...bool) (bool, error) {
func Copy(ctx context.Context, srcObjPath, dstDirPath string, lazyCache ...bool) (tache.TaskWithInfo, error) {
res, err := _copy(ctx, srcObjPath, dstDirPath, lazyCache...)
if err != nil {
log.Errorf("failed copy %s to %s: %+v", srcObjPath, dstDirPath, err)
@ -100,12 +101,12 @@ func PutDirectly(ctx context.Context, dstDirPath string, file model.FileStreamer
return err
}
func PutAsTask(dstDirPath string, file model.FileStreamer) error {
err := putAsTask(dstDirPath, file)
func PutAsTask(dstDirPath string, file model.FileStreamer) (tache.TaskWithInfo, error) {
t, err := putAsTask(dstDirPath, file)
if err != nil {
log.Errorf("failed put %s: %+v", dstDirPath, err)
}
return err
return t, err
}
type GetStoragesArgs struct {

15
internal/fs/put.go
View File

@ -33,28 +33,29 @@ func (t *UploadTask) Run() error {
var UploadTaskManager *tache.Manager[*UploadTask]
// putAsTask add as a put task and return immediately
func putAsTask(dstDirPath string, file model.FileStreamer) error {
func putAsTask(dstDirPath string, file model.FileStreamer) (tache.TaskWithInfo, error) {
storage, dstDirActualPath, err := op.GetStorageAndActualPath(dstDirPath)
if err != nil {
return errors.WithMessage(err, "failed get storage")
return nil, errors.WithMessage(err, "failed get storage")
}
if storage.Config().NoUpload {
return errors.WithStack(errs.UploadNotSupported)
return nil, errors.WithStack(errs.UploadNotSupported)
}
if file.NeedStore() {
_, err := file.CacheFullInTempFile()
if err != nil {
return errors.Wrapf(err, "failed to create temp file")
return nil, errors.Wrapf(err, "failed to create temp file")
}
//file.SetReader(tempFile)
//file.SetTmpFile(tempFile)
}
UploadTaskManager.Add(&UploadTask{
t := &UploadTask{
storage: storage,
dstDirActualPath: dstDirActualPath,
file: file,
})
return nil
}
UploadTaskManager.Add(t)
return t, nil
}
// putDirect put the file and return after finish

17
internal/offline_download/tool/add.go
View File

@ -7,6 +7,7 @@ import (
"github.com/alist-org/alist/v3/internal/op"
"github.com/google/uuid"
"github.com/pkg/errors"
"github.com/xhofe/tache"
"path/filepath"
)
@ -26,38 +27,38 @@ type AddURLArgs struct {
DeletePolicy DeletePolicy
}
func AddURL(ctx context.Context, args *AddURLArgs) error {
func AddURL(ctx context.Context, args *AddURLArgs) (tache.TaskWithInfo, error) {
// get tool
tool, err := Tools.Get(args.Tool)
if err != nil {
return errors.Wrapf(err, "failed get tool")
return nil, errors.Wrapf(err, "failed get tool")
}
// check tool is ready
if !tool.IsReady() {
// try to init tool
if _, err := tool.Init(); err != nil {
return errors.Wrapf(err, "failed init tool %s", args.Tool)
return nil, errors.Wrapf(err, "failed init tool %s", args.Tool)
}
}
// check storage
storage, dstDirActualPath, err := op.GetStorageAndActualPath(args.DstDirPath)
if err != nil {
return errors.WithMessage(err, "failed get storage")
return nil, errors.WithMessage(err, "failed get storage")
}
// check is it could upload
if storage.Config().NoUpload {
return errors.WithStack(errs.UploadNotSupported)
return nil, 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")
return nil, errors.WithMessage(err, "failed get object")
}
} else {
if !obj.IsDir() {
// can't add to a file
return errors.WithStack(errs.NotFolder)
return nil, errors.WithStack(errs.NotFolder)
}
}
@ -71,5 +72,5 @@ func AddURL(ctx context.Context, args *AddURLArgs) error {
tool: tool,
}
DownloadTaskManager.Add(t)
return nil
return t, nil
}

17
server/handles/fsmanage.go
View File

@ -2,6 +2,7 @@ package handles
import (
"fmt"
"github.com/xhofe/tache"
"io"
stdpath "path"
@ -120,22 +121,20 @@ func FsCopy(c *gin.Context) {
common.ErrorResp(c, err, 403)
return
}
var addedTask []string
var addedTasks []tache.TaskWithInfo
for i, name := range req.Names {
ok, err := fs.Copy(c, stdpath.Join(srcDir, name), dstDir, len(req.Names) > i+1)
if ok {
addedTask = append(addedTask, name)
t, err := fs.Copy(c, stdpath.Join(srcDir, name), dstDir, len(req.Names) > i+1)
if t != nil {
addedTasks = append(addedTasks, t)
}
if err != nil {
common.ErrorResp(c, err, 500)
return
}
}
if len(addedTask) > 0 {
common.SuccessResp(c, fmt.Sprintf("Added %d tasks", len(addedTask)))
} else {
common.SuccessResp(c)
}
common.SuccessResp(c, gin.H{
"tasks": getTaskInfos(addedTasks),
})
}
type RenameReq struct {

23
server/handles/fsup.go
View File

@ -1,6 +1,7 @@
package handles
import (
"github.com/xhofe/tache"
"io"
"net/url"
stdpath "path"
@ -57,8 +58,9 @@ func FsStream(c *gin.Context) {
Mimetype: c.GetHeader("Content-Type"),
WebPutAsTask: asTask,
}
var t tache.TaskWithInfo
if asTask {
err = fs.PutAsTask(dir, s)
t, err = fs.PutAsTask(dir, s)
} else {
err = fs.PutDirectly(c, dir, s, true)
}
@ -67,7 +69,13 @@ func FsStream(c *gin.Context) {
common.ErrorResp(c, err, 500)
return
}
common.SuccessResp(c)
if t == nil {
common.SuccessResp(c)
return
}
common.SuccessResp(c, gin.H{
"task": getTaskInfo(t),
})
}
func FsForm(c *gin.Context) {
@ -115,11 +123,12 @@ func FsForm(c *gin.Context) {
Mimetype: file.Header.Get("Content-Type"),
WebPutAsTask: asTask,
}
var t tache.TaskWithInfo
if asTask {
s.Reader = struct {
io.Reader
}{f}
err = fs.PutAsTask(dir, &s)
t, err = fs.PutAsTask(dir, &s)
} else {
ss, err := stream.NewSeekableStream(s, nil)
if err != nil {
@ -132,5 +141,11 @@ func FsForm(c *gin.Context) {
common.ErrorResp(c, err, 500)
return
}
common.SuccessResp(c)
if t == nil {
common.SuccessResp(c)
return
}
common.SuccessResp(c, gin.H{
"task": getTaskInfo(t),
})
}

9
server/handles/offline_download.go
View File

@ -7,6 +7,7 @@ import (
"github.com/alist-org/alist/v3/internal/op"
"github.com/alist-org/alist/v3/server/common"
"github.com/gin-gonic/gin"
"github.com/xhofe/tache"
)
type SetAria2Req struct {
@ -97,8 +98,9 @@ func AddOfflineDownload(c *gin.Context) {
common.ErrorResp(c, err, 403)
return
}
var tasks []tache.TaskWithInfo
for _, url := range req.Urls {
err := tool.AddURL(c, &tool.AddURLArgs{
t, err := tool.AddURL(c, &tool.AddURLArgs{
URL: url,
DstDirPath: reqPath,
Tool: req.Tool,
@ -108,6 +110,9 @@ func AddOfflineDownload(c *gin.Context) {
common.ErrorResp(c, err, 500)
return
}
tasks = append(tasks, t)
}
common.SuccessResp(c)
common.SuccessResp(c, gin.H{
"tasks": getTaskInfos(tasks),
})
}

16
server/handles/task.go
View File

@ -3,6 +3,7 @@ package handles
import (
"github.com/alist-org/alist/v3/internal/fs"
"github.com/alist-org/alist/v3/internal/offline_download/tool"
"github.com/alist-org/alist/v3/pkg/utils"
"github.com/alist-org/alist/v3/server/common"
"github.com/gin-gonic/gin"
"github.com/xhofe/tache"
@ -33,11 +34,7 @@ func getTaskInfo[T tache.TaskWithInfo](task T) TaskInfo {
}
func getTaskInfos[T tache.TaskWithInfo](tasks []T) []TaskInfo {
var infos []TaskInfo
for _, t := range tasks {
infos = append(infos, getTaskInfo(t))
}
return infos
return utils.MustSliceConvert(tasks, getTaskInfo[T])
}
func taskRoute[T tache.TaskWithInfo](g *gin.RouterGroup, manager *tache.Manager[T]) {
@ -48,6 +45,15 @@ func taskRoute[T tache.TaskWithInfo](g *gin.RouterGroup, manager *tache.Manager[
g.GET("/done", func(c *gin.Context) {
common.SuccessResp(c, getTaskInfos(manager.GetByState(tache.StateCanceled, tache.StateFailed, tache.StateSucceeded)))
})
g.POST("/info", func(c *gin.Context) {
tid := c.Query("tid")
task, ok := manager.GetByID(tid)
if !ok {
common.ErrorStrResp(c, "task not found", 404)
return
}
common.SuccessResp(c, getTaskInfo(task))
})
g.POST("/cancel", func(c *gin.Context) {
tid := c.Query("tid")
manager.Cancel(tid)