github-mirror/alist
github-mirror/alist
2
0
Fork 0
mirror of https://github.com/AlistGo/alist.git synced 2025-04-21 20:18:47 +08:00
Code Issues Actions 1 Packages Projects Releases Wiki Activity

chore: remove useless code

Browse Source
This commit is contained in:
Andy Hsu 2023-11-19 20:05:09 +08:00
parent 8d5283604c
commit de9647a5fa
12 changed files with 1 additions and 1242 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

61
internal/aria2/add.go
View File

@ -1,61 +0,0 @@
package aria2
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"
)
func AddURI(ctx context.Context, uri string, dstDirPath string) error {
// check storage
storage, dstDirActualPath, err := op.GetStorageAndActualPath(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)
}
}
// call aria2 rpc
tempDir := filepath.Join(conf.Conf.TempDir, "aria2", uuid.NewString())
options := map[string]interface{}{
"dir": tempDir,
}
gid, err := client.AddURI([]string{uri}, options)
if err != nil {
return errors.Wrapf(err, "failed to add uri %s", uri)
}
DownTaskManager.Submit(task.WithCancelCtx(&task.Task[string]{
ID: gid,
Name: fmt.Sprintf("download %s to [%s](%s)", uri, storage.GetStorage().MountPath, dstDirActualPath),
Func: func(tsk *task.Task[string]) error {
m := &Monitor{
tsk: tsk,
tempDir: tempDir,
retried: 0,
dstDirPath: dstDirPath,
}
return m.Loop()
},
}))
return nil
}

42
internal/aria2/aria2.go
View File

@ -1,42 +0,0 @@
package aria2
import (
"context"
"time"
"github.com/alist-org/alist/v3/internal/conf"
"github.com/alist-org/alist/v3/internal/setting"
"github.com/alist-org/alist/v3/pkg/aria2/rpc"
"github.com/alist-org/alist/v3/pkg/task"
"github.com/pkg/errors"
log "github.com/sirupsen/logrus"
)
var DownTaskManager = task.NewTaskManager[string](3)
var notify = NewNotify()
var client rpc.Client
func InitClient(timeout int) (string, error) {
client = nil
uri := setting.GetStr(conf.Aria2Uri)
secret := setting.GetStr(conf.Aria2Secret)
return InitAria2Client(uri, secret, timeout)
}
func InitAria2Client(uri string, secret string, timeout int) (string, error) {
c, err := rpc.New(context.Background(), uri, secret, time.Duration(timeout)*time.Second, notify)
if err != nil {
return "", errors.Wrap(err, "failed to init aria2 client")
}
version, err := c.GetVersion()
if err != nil {
return "", errors.Wrapf(err, "failed get aria2 version")
}
client = c
log.Infof("using aria2 version: %s", version.Version)
return version.Version, nil
}
func IsAria2Ready() bool {
return client != nil
}

87
internal/aria2/aria2_test.go
View File

@ -1,87 +0,0 @@
package aria2
import (
"context"
"path/filepath"
"testing"
"time"
_ "github.com/alist-org/alist/v3/drivers"
conf2 "github.com/alist-org/alist/v3/internal/conf"
"github.com/alist-org/alist/v3/internal/db"
"github.com/alist-org/alist/v3/internal/model"
"github.com/alist-org/alist/v3/internal/op"
"github.com/alist-org/alist/v3/pkg/task"
"gorm.io/driver/sqlite"
"gorm.io/gorm"
)
func init() {
conf2.Conf = conf2.DefaultConfig()
absPath, err := filepath.Abs("../../data/temp")
if err != nil {
panic(err)
}
conf2.Conf.TempDir = absPath
dB, err := gorm.Open(sqlite.Open("file::memory:?cache=shared"), &gorm.Config{})
if err != nil {
panic("failed to connect database")
}
db.Init(dB)
}
func TestConnect(t *testing.T) {
_, err := InitAria2Client("http://localhost:16800/jsonrpc", "secret", 3)
if err != nil {
t.Errorf("failed to init aria2: %+v", err)
}
}
func TestDown(t *testing.T) {
TestConnect(t)
_, err := op.CreateStorage(context.Background(), model.Storage{
ID: 0,
MountPath: "/",
Order: 0,
Driver: "Local",
Status: "",
Addition: `{"root_folder":"../../data"}`,
Remark: "",
})
if err != nil {
t.Fatalf("failed to create storage: %+v", err)
}
err = AddURI(context.Background(), "https://nodejs.org/dist/index.json", "/test")
if err != nil {
t.Errorf("failed to add uri: %+v", err)
}
tasks := DownTaskManager.GetAll()
if len(tasks) != 1 {
t.Errorf("failed to get tasks: %+v", tasks)
}
for {
tsk := tasks[0]
t.Logf("task: %+v", tsk)
if tsk.GetState() == task.SUCCEEDED {
break
}
if tsk.GetState() == task.ERRORED {
t.Fatalf("failed to download: %+v", tsk)
}
time.Sleep(time.Second)
}
for {
if len(TransferTaskManager.GetAll()) == 0 {
continue
}
tsk := TransferTaskManager.GetAll()[0]
t.Logf("task: %+v", tsk)
if tsk.GetState() == task.SUCCEEDED {
break
}
if tsk.GetState() == task.ERRORED {
t.Fatalf("failed to download: %+v", tsk)
}
time.Sleep(time.Second)
}
}

192
internal/aria2/monitor.go
View File

@ -1,192 +0,0 @@
package aria2
import (
"fmt"
"os"
"path"
"path/filepath"
"strconv"
"sync"
"sync/atomic"
"time"
"github.com/alist-org/alist/v3/internal/stream"
"github.com/alist-org/alist/v3/internal/model"
"github.com/alist-org/alist/v3/internal/op"
"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 {
tsk *task.Task[string]
tempDir string
retried int
c chan int
dstDirPath string
finish chan struct{}
}
func (m *Monitor) Loop() error {
defer func() {
notify.Signals.Delete(m.tsk.ID)
// clear temp dir, should do while complete
//_ = os.RemoveAll(m.tempDir)
}()
m.c = make(chan int)
m.finish = make(chan struct{})
notify.Signals.Store(m.tsk.ID, m.c)
var (
err error
ok bool
)
outer:
for {
select {
case <-m.tsk.Ctx.Done():
_, err := client.Remove(m.tsk.ID)
return err
case <-m.c:
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
}
func (m *Monitor) Update() (bool, error) {
info, err := client.TellStatus(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
if len(info.FollowedBy) != 0 {
log.Debugf("followen by: %+v", info.FollowedBy)
gid := info.FollowedBy[0]
notify.Signals.Delete(m.tsk.ID)
oldId := m.tsk.ID
m.tsk.ID = gid
DownTaskManager.RawTasks().Delete(oldId)
DownTaskManager.RawTasks().Store(m.tsk.ID, m.tsk)
notify.Signals.Store(gid, m.c)
return false, nil
}
// update download status
total, err := strconv.ParseUint(info.TotalLength, 10, 64)
if err != nil {
total = 0
}
downloaded, err := strconv.ParseUint(info.CompletedLength, 10, 64)
if err != nil {
downloaded = 0
}
progress := float64(downloaded) / float64(total) * 100
m.tsk.SetProgress(progress)
switch info.Status {
case "complete":
err := m.Complete()
return true, errors.WithMessage(err, "failed to transfer file")
case "error":
return true, errors.Errorf("failed to download %s, error: %s", m.tsk.ID, info.ErrorMessage)
case "active":
m.tsk.SetStatus("aria2: " + info.Status)
if info.Seeder == "true" {
err := m.Complete()
return true, errors.WithMessage(err, "failed to transfer file")
}
return false, nil
case "waiting", "paused":
m.tsk.SetStatus("aria2: " + info.Status)
return false, nil
case "removed":
return true, errors.Errorf("failed to download %s, removed", m.tsk.ID)
default:
return true, errors.Errorf("failed to download %s, unknown status %s", m.tsk.ID, info.Status)
}
}
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")
}
// get files
files, err := client.GetFiles(m.tsk.ID)
log.Debugf("files len: %d", len(files))
if err != nil {
return errors.Wrapf(err, "failed to get files of %s", m.tsk.ID)
}
// 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()
size, _ := strconv.ParseInt(file.Length, 10, 64)
mimetype := utils.GetMimeType(file.Path)
f, err := os.Open(file.Path)
if err != nil {
return errors.Wrapf(err, "failed to open file %s", file.Path)
}
s := stream.FileStream{
Obj: &model.Object{
Name: path.Base(file.Path),
Size: size,
Modified: time.Now(),
IsFolder: false,
},
Reader: f,
Closers: utils.NewClosers(f),
Mimetype: mimetype,
}
ss, err := stream.NewSeekableStream(s, nil)
if err != nil {
return err
}
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, ss, tsk.SetProgress)
},
}))
}
return nil
}

70
internal/aria2/notify.go
View File

@ -1,70 +0,0 @@
package aria2
import (
"github.com/alist-org/alist/v3/pkg/aria2/rpc"
"github.com/alist-org/alist/v3/pkg/generic_sync"
)
const (
Downloading = iota
Paused
Stopped
Completed
Errored
)
type Notify struct {
Signals generic_sync.MapOf[string, chan int]
}
func NewNotify() *Notify {
return &Notify{Signals: generic_sync.MapOf[string, chan int]{}}
}
func (n *Notify) OnDownloadStart(events []rpc.Event) {
for _, e := range events {
if signal, ok := n.Signals.Load(e.Gid); ok {
signal <- Downloading
}
}
}
func (n *Notify) OnDownloadPause(events []rpc.Event) {
for _, e := range events {
if signal, ok := n.Signals.Load(e.Gid); ok {
signal <- Paused
}
}
}
func (n *Notify) OnDownloadStop(events []rpc.Event) {
for _, e := range events {
if signal, ok := n.Signals.Load(e.Gid); ok {
signal <- Stopped
}
}
}
func (n *Notify) OnDownloadComplete(events []rpc.Event) {
for _, e := range events {
if signal, ok := n.Signals.Load(e.Gid); ok {
signal <- Completed
}
}
}
func (n *Notify) OnDownloadError(events []rpc.Event) {
for _, e := range events {
if signal, ok := n.Signals.Load(e.Gid); ok {
signal <- Errored
}
}
}
func (n *Notify) OnBtDownloadComplete(events []rpc.Event) {
for _, e := range events {
if signal, ok := n.Signals.Load(e.Gid); ok {
signal <- Completed
}
}
}

2
internal/offline_download/qbit/qbit.go
View File

@ -4,8 +4,8 @@ import (
"github.com/alist-org/alist/v3/internal/conf"
"github.com/alist-org/alist/v3/internal/model"
"github.com/alist-org/alist/v3/internal/offline_download/tool"
"github.com/alist-org/alist/v3/internal/qbittorrent"
"github.com/alist-org/alist/v3/internal/setting"
"github.com/alist-org/alist/v3/pkg/qbittorrent"
"github.com/pkg/errors"
)

60
internal/qbittorrent/add.go
View File

@ -1,60 +0,0 @@
package qbittorrent
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/internal/setting"
"github.com/alist-org/alist/v3/pkg/task"
"github.com/google/uuid"
"github.com/pkg/errors"
)
func AddURL(ctx context.Context, url string, dstDirPath string) error {
// check storage
storage, dstDirActualPath, err := op.GetStorageAndActualPath(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)
}
}
// call qbittorrent
id := uuid.NewString()
tempDir := filepath.Join(conf.Conf.TempDir, "qbittorrent", id)
err = qbclient.AddFromLink(url, tempDir, id)
if err != nil {
return errors.Wrapf(err, "failed to add url %s", url)
}
DownTaskManager.Submit(task.WithCancelCtx(&task.Task[string]{
ID: id,
Name: fmt.Sprintf("download %s to [%s](%s)", url, storage.GetStorage().MountPath, dstDirActualPath),
Func: func(tsk *task.Task[string]) error {
m := &Monitor{
tsk: tsk,
tempDir: tempDir,
dstDirPath: dstDirPath,
seedtime: setting.GetInt(conf.QbittorrentSeedtime, 0),
}
return m.Loop()
},
}))
return nil
}

366
internal/qbittorrent/client.go
View File

@ -1,366 +0,0 @@
package qbittorrent
import (
"bytes"
"errors"
"io"
"mime/multipart"
"net/http"
"net/http/cookiejar"
"net/url"
"github.com/alist-org/alist/v3/pkg/utils"
)
type Client interface {
AddFromLink(link string, savePath string, id string) error
GetInfo(id string) (TorrentInfo, error)
GetFiles(id string) ([]FileInfo, error)
Delete(id string, deleteFiles bool) error
}
type client struct {
url *url.URL
client http.Client
Client
}
func New(webuiUrl string) (Client, error) {
u, err := url.Parse(webuiUrl)
if err != nil {
return nil, err
}
jar, err := cookiejar.New(nil)
if err != nil {
return nil, err
}
var c = &client{
url: u,
client: http.Client{Jar: jar},
}
err = c.checkAuthorization()
if err != nil {
return nil, err
}
return c, nil
}
func (c *client) checkAuthorization() error {
// check authorization
if c.authorized() {
return nil
}
// check authorization after logging in
err := c.login()
if err != nil {
return err
}
if c.authorized() {
return nil
}
return errors.New("unauthorized qbittorrent url")
}
func (c *client) authorized() bool {
resp, err := c.post("/api/v2/app/version", nil)
if err != nil {
return false
}
return resp.StatusCode == 200 // the status code will be 403 if not authorized
}
func (c *client) login() error {
// prepare HTTP request
v := url.Values{}
v.Set("username", c.url.User.Username())
passwd, _ := c.url.User.Password()
v.Set("password", passwd)
resp, err := c.post("/api/v2/auth/login", v)
if err != nil {
return err
}
// check result
body := make([]byte, 2)
_, err = resp.Body.Read(body)
if err != nil {
return err
}
if string(body) != "Ok" {
return errors.New("failed to login into qBittorrent webui with url: " + c.url.String())
}
return nil
}
func (c *client) post(path string, data url.Values) (*http.Response, error) {
u := c.url.JoinPath(path)
u.User = nil // remove userinfo for requests
req, err := http.NewRequest("POST", u.String(), bytes.NewReader([]byte(data.Encode())))
if err != nil {
return nil, err
}
if data != nil {
req.Header.Add("Content-Type", "application/x-www-form-urlencoded")
}
resp, err := c.client.Do(req)
if err != nil {
return nil, err
}
if resp.Cookies() != nil {
c.client.Jar.SetCookies(u, resp.Cookies())
}
return resp, nil
}
func (c *client) AddFromLink(link string, savePath string, id string) error {
err := c.checkAuthorization()
if err != nil {
return err
}
buf := new(bytes.Buffer)
writer := multipart.NewWriter(buf)
addField := func(name string, value string) {
if err != nil {
return
}
err = writer.WriteField(name, value)
}
addField("urls", link)
addField("savepath", savePath)
addField("tags", "alist-"+id)
addField("autoTMM", "false")
if err != nil {
return err
}
err = writer.Close()
if err != nil {
return err
}
u := c.url.JoinPath("/api/v2/torrents/add")
u.User = nil // remove userinfo for requests
req, err := http.NewRequest("POST", u.String(), buf)
if err != nil {
return err
}
req.Header.Add("Content-Type", writer.FormDataContentType())
resp, err := c.client.Do(req)
if err != nil {
return err
}
// check result
body := make([]byte, 2)
_, err = resp.Body.Read(body)
if err != nil {
return err
}
if resp.StatusCode != 200 || string(body) != "Ok" {
return errors.New("failed to add qBittorrent task: " + link)
}
return nil
}
type TorrentStatus string
const (
ERROR TorrentStatus = "error"
MISSINGFILES TorrentStatus = "missingFiles"
UPLOADING TorrentStatus = "uploading"
PAUSEDUP TorrentStatus = "pausedUP"
QUEUEDUP TorrentStatus = "queuedUP"
STALLEDUP TorrentStatus = "stalledUP"
CHECKINGUP TorrentStatus = "checkingUP"
FORCEDUP TorrentStatus = "forcedUP"
ALLOCATING TorrentStatus = "allocating"
DOWNLOADING TorrentStatus = "downloading"
METADL TorrentStatus = "metaDL"
PAUSEDDL TorrentStatus = "pausedDL"
QUEUEDDL TorrentStatus = "queuedDL"
STALLEDDL TorrentStatus = "stalledDL"
CHECKINGDL TorrentStatus = "checkingDL"
FORCEDDL TorrentStatus = "forcedDL"
CHECKINGRESUMEDATA TorrentStatus = "checkingResumeData"
MOVING TorrentStatus = "moving"
UNKNOWN TorrentStatus = "unknown"
)
// https://github.com/DGuang21/PTGo/blob/main/app/client/client_distributer.go
type TorrentInfo struct {
AddedOn int `json:"added_on"` // 将 torrent 添加到客户端的时间（Unix Epoch）
AmountLeft int64 `json:"amount_left"` // 剩余大小（字节）
AutoTmm bool `json:"auto_tmm"` // 此 torrent 是否由 Automatic Torrent Management 管理
Availability float64 `json:"availability"` // 当前百分比
Category string `json:"category"` //
Completed int64 `json:"completed"` // 完成的传输数据量（字节）
CompletionOn int `json:"completion_on"` // Torrent 完成的时间（Unix Epoch）
ContentPath string `json:"content_path"` // torrent 内容的绝对路径（多文件 torrent 的根路径，单文件 torrent 的绝对文件路径）
DlLimit int `json:"dl_limit"` // Torrent 下载速度限制（字节/秒）
Dlspeed int `json:"dlspeed"` // Torrent 下载速度（字节/秒）
Downloaded int64 `json:"downloaded"` // 已经下载大小
DownloadedSession int64 `json:"downloaded_session"` // 此会话下载的数据量
Eta int `json:"eta"` //
FLPiecePrio bool `json:"f_l_piece_prio"` // 如果第一个最后一块被优先考虑，则为true
ForceStart bool `json:"force_start"` // 如果为此 torrent 启用了强制启动，则为true
Hash string `json:"hash"` //
LastActivity int `json:"last_activity"` // 上次活跃的时间（Unix Epoch）
MagnetURI string `json:"magnet_uri"` // 与此 torrent 对应的 Magnet URI
MaxRatio float64 `json:"max_ratio"` // 种子/上传停止种子前的最大共享比率
MaxSeedingTime int `json:"max_seeding_time"` // 停止种子种子前的最长种子时间（秒）
Name string `json:"name"` //
NumComplete int `json:"num_complete"` //
NumIncomplete int `json:"num_incomplete"` //
NumLeechs int `json:"num_leechs"` // 连接到的 leechers 的数量
NumSeeds int `json:"num_seeds"` // 连接到的种子数
Priority int `json:"priority"` // 速度优先。如果队列被禁用或 torrent 处于种子模式，则返回 -1
Progress float64 `json:"progress"` // 进度
Ratio float64 `json:"ratio"` // Torrent 共享比率
RatioLimit int `json:"ratio_limit"` //
SavePath string `json:"save_path"`
SeedingTime int `json:"seeding_time"` // Torrent 完成用时（秒）
SeedingTimeLimit int `json:"seeding_time_limit"` // max_seeding_time
SeenComplete int `json:"seen_complete"` // 上次 torrent 完成的时间
SeqDl bool `json:"seq_dl"` // 如果启用顺序下载，则为true
Size int64 `json:"size"` //
State TorrentStatus `json:"state"` // 参见https://github.com/qbittorrent/qBittorrent/wiki/WebUI-API-(qBittorrent-4.1)#get-torrent-list
SuperSeeding bool `json:"super_seeding"` // 如果启用超级播种，则为true
Tags string `json:"tags"` // Torrent 的逗号连接标签列表
TimeActive int `json:"time_active"` // 总活动时间（秒）
TotalSize int64 `json:"total_size"` // 此 torrent 中所有文件的总大小（字节）（包括未选择的文件）
Tracker string `json:"tracker"` // 第一个具有工作状态的tracker。如果没有tracker在工作，则返回空字符串。
TrackersCount int `json:"trackers_count"` //
UpLimit int `json:"up_limit"` // 上传限制
Uploaded int64 `json:"uploaded"` // 累计上传
UploadedSession int64 `json:"uploaded_session"` // 当前session累计上传
Upspeed int `json:"upspeed"` // 上传速度（字节/秒）
}
type InfoNotFoundError struct {
Id string
Err error
}
func (i InfoNotFoundError) Error() string {
return "there should be exactly one task with tag \"alist-" + i.Id + "\""
}
func NewInfoNotFoundError(id string) InfoNotFoundError {
return InfoNotFoundError{Id: id}
}
func (c *client) GetInfo(id string) (TorrentInfo, error) {
var infos []TorrentInfo
err := c.checkAuthorization()
if err != nil {
return TorrentInfo{}, err
}
v := url.Values{}
v.Set("tag", "alist-"+id)
response, err := c.post("/api/v2/torrents/info", v)
if err != nil {
return TorrentInfo{}, err
}
body, err := io.ReadAll(response.Body)
if err != nil {
return TorrentInfo{}, err
}
err = utils.Json.Unmarshal(body, &infos)
if err != nil {
return TorrentInfo{}, err
}
if len(infos) != 1 {
return TorrentInfo{}, NewInfoNotFoundError(id)
}
return infos[0], nil
}
type FileInfo struct {
Index int `json:"index"`
Name string `json:"name"`
Size int64 `json:"size"`
Progress float32 `json:"progress"`
Priority int `json:"priority"`
IsSeed bool `json:"is_seed"`
PieceRange []int `json:"piece_range"`
Availability float32 `json:"availability"`
}
func (c *client) GetFiles(id string) ([]FileInfo, error) {
var infos []FileInfo
err := c.checkAuthorization()
if err != nil {
return []FileInfo{}, err
}
tInfo, err := c.GetInfo(id)
if err != nil {
return []FileInfo{}, err
}
v := url.Values{}
v.Set("hash", tInfo.Hash)
response, err := c.post("/api/v2/torrents/files", v)
if err != nil {
return []FileInfo{}, err
}
body, err := io.ReadAll(response.Body)
if err != nil {
return []FileInfo{}, err
}
err = utils.Json.Unmarshal(body, &infos)
if err != nil {
return []FileInfo{}, err
}
return infos, nil
}
func (c *client) Delete(id string, deleteFiles bool) error {
err := c.checkAuthorization()
if err != nil {
return err
}
info, err := c.GetInfo(id)
if err != nil {
return err
}
v := url.Values{}
v.Set("hashes", info.Hash)
if deleteFiles {
v.Set("deleteFiles", "true")
} else {
v.Set("deleteFiles", "false")
}
response, err := c.post("/api/v2/torrents/delete", v)
if err != nil {
return err
}
if response.StatusCode != 200 {
return errors.New("failed to delete qbittorrent task")
}
v = url.Values{}
v.Set("tags", "alist-"+id)
response, err = c.post("/api/v2/torrents/deleteTags", v)
if err != nil {
return err
}
if response.StatusCode != 200 {
return errors.New("failed to delete qbittorrent tag")
}
return nil
}

154
internal/qbittorrent/client_test.go
View File

@ -1,154 +0,0 @@
package qbittorrent
import (
"net/http"
"net/http/cookiejar"
"net/url"
"testing"
)
func TestLogin(t *testing.T) {
// test logging in with wrong password
u, err := url.Parse("http://admin:admin@127.0.0.1:8080/")
if err != nil {
t.Error(err)
}
jar, err := cookiejar.New(nil)
if err != nil {
t.Error(err)
}
var c = &client{
url: u,
client: http.Client{Jar: jar},
}
err = c.login()
if err == nil {
t.Error(err)
}
// test logging in with correct password
u, err = url.Parse("http://admin:adminadmin@127.0.0.1:8080/")
if err != nil {
t.Error(err)
}
c.url = u
err = c.login()
if err != nil {
t.Error(err)
}
}
// in this test, the `Bypass authentication for clients on localhost` option in qBittorrent webui should be disabled
func TestAuthorized(t *testing.T) {
// init client
u, err := url.Parse("http://admin:adminadmin@127.0.0.1:8080/")
if err != nil {
t.Error(err)
}
jar, err := cookiejar.New(nil)
if err != nil {
t.Error(err)
}
var c = &client{
url: u,
client: http.Client{Jar: jar},
}
// test without logging in, which should be unauthorized
authorized := c.authorized()
if authorized {
t.Error("Should not be authorized")
}
// test after logging in
err = c.login()
if err != nil {
t.Error(err)
}
authorized = c.authorized()
if !authorized {
t.Error("Should be authorized")
}
}
func TestNew(t *testing.T) {
_, err := New("http://admin:adminadmin@127.0.0.1:8080/")
if err != nil {
t.Error(err)
}
_, err = New("http://admin:wrong_password@127.0.0.1:8080/")
if err == nil {
t.Error("Should get an error")
}
}
func TestAdd(t *testing.T) {
// init client
c, err := New("http://admin:adminadmin@127.0.0.1:8080/")
if err != nil {
t.Error(err)
}
err = c.AddFromLink(
"https://releases.ubuntu.com/22.04/ubuntu-22.04.1-desktop-amd64.iso.torrent",
"D:\\qBittorrentDownload\\alist",
"uuid-1",
)
if err != nil {
t.Error(err)
}
err = c.AddFromLink(
"magnet:?xt=urn:btih:375ae3280cd80a8e9d7212e11dfaf7c45069dd35&dn=archlinux-2023.02.01-x86_64.iso",
"D:\\qBittorrentDownload\\alist",
"uuid-2",
)
if err != nil {
t.Error(err)
}
}
func TestGetInfo(t *testing.T) {
// init client
c, err := New("http://admin:adminadmin@127.0.0.1:8080/")
if err != nil {
t.Error(err)
}
_, err = c.GetInfo("uuid-1")
if err != nil {
t.Error(err)
}
}
func TestGetFiles(t *testing.T) {
// init client
c, err := New("http://admin:adminadmin@127.0.0.1:8080/")
if err != nil {
t.Error(err)
}
files, err := c.GetFiles("uuid-1")
if err != nil {
t.Error(err)
}
if len(files) != 1 {
t.Error("should have exactly one file")
}
}
func TestDelete(t *testing.T) {
// init client
c, err := New("http://admin:adminadmin@127.0.0.1:8080/")
if err != nil {
t.Error(err)
}
err = c.AddFromLink(
"https://releases.ubuntu.com/22.04/ubuntu-22.04.1-desktop-amd64.iso.torrent",
"D:\\qBittorrentDownload\\alist",
"uuid-1",
)
if err != nil {
t.Error(err)
}
err = c.Delete("uuid-1", true)
if err != nil {
t.Error(err)
}
}

181
internal/qbittorrent/monitor.go
View File

@ -1,181 +0,0 @@
package qbittorrent
import (
"fmt"
"os"
"path/filepath"
"sync"
"sync/atomic"
"time"
"github.com/alist-org/alist/v3/internal/stream"
"github.com/alist-org/alist/v3/internal/model"
"github.com/alist-org/alist/v3/internal/op"
"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 {
tsk *task.Task[string]
tempDir string
dstDirPath string
seedtime int
finish chan struct{}
}
func (m *Monitor) Loop() error {
var (
err error
completed bool
)
m.finish = make(chan struct{})
// wait for qbittorrent to parse torrent and create task
m.tsk.SetStatus("waiting for qbittorrent to parse torrent and create task")
waitCount := 0
for {
_, err := qbclient.GetInfo(m.tsk.ID)
if err == nil {
break
}
switch err.(type) {
case InfoNotFoundError:
break
default:
return err
}
waitCount += 1
if waitCount >= 60 {
return errors.New("torrent parse timeout")
}
timer := time.NewTimer(time.Second)
<-timer.C
}
outer:
for {
select {
case <-m.tsk.Ctx.Done():
// delete qbittorrent task and downloaded files when the task exits with error
return qbclient.Delete(m.tsk.ID, true)
case <-time.After(time.Second * 2):
completed, err = m.update()
if completed {
break outer
}
}
}
if err != nil {
return err
}
m.tsk.SetStatus("qbittorrent download completed, transferring")
<-m.finish
m.tsk.SetStatus("completed")
return nil
}
func (m *Monitor) update() (bool, error) {
info, err := qbclient.GetInfo(m.tsk.ID)
if err != nil {
m.tsk.SetStatus("qbittorrent " + string(info.State))
return true, err
}
progress := float64(info.Completed) / float64(info.Size) * 100
m.tsk.SetProgress(progress)
switch info.State {
case UPLOADING, PAUSEDUP, QUEUEDUP, STALLEDUP, FORCEDUP, CHECKINGUP:
err = m.complete()
return true, errors.WithMessage(err, "failed to transfer file")
case ALLOCATING, DOWNLOADING, METADL, PAUSEDDL, QUEUEDDL, STALLEDDL, CHECKINGDL, FORCEDDL, CHECKINGRESUMEDATA, MOVING:
m.tsk.SetStatus("qbittorrent downloading")
return false, nil
case ERROR, MISSINGFILES, UNKNOWN:
return true, errors.Errorf("failed to download %s, error: %s", m.tsk.ID, info.State)
}
return true, errors.New("unknown error occurred downloading qbittorrent") // should never happen
}
var TransferTaskManager = task.NewTaskManager(3, func(k *uint64) {
atomic.AddUint64(k, 1)
})
func (m *Monitor) complete() error {
// check dstDir again
storage, dstBaseDir, err := op.GetStorageAndActualPath(m.dstDirPath)
if err != nil {
return errors.WithMessage(err, "failed get storage")
}
// get files
files, err := qbclient.GetFiles(m.tsk.ID)
if err != nil {
return errors.Wrapf(err, "failed to get files of %s", m.tsk.ID)
}
log.Debugf("files len: %d", len(files))
// delete qbittorrent task but do not delete the files before transferring to avoid qbittorrent
// accessing downloaded files and throw `cannot access the file because it is being used by another process` error
// err = qbclient.Delete(m.tsk.ID, false)
// if err != nil {
// return err
// }
// upload files
var wg sync.WaitGroup
wg.Add(len(files))
go func() {
wg.Wait()
m.finish <- struct{}{}
if m.seedtime < 0 {
log.Debugf("do not delete qb task %s", m.tsk.ID)
return
}
log.Debugf("delete qb task %s after %d minutes", m.tsk.ID, m.seedtime)
<-time.After(time.Duration(m.seedtime) * time.Minute)
err := qbclient.Delete(m.tsk.ID, true)
if err != nil {
log.Errorln(err.Error())
}
err = os.RemoveAll(m.tempDir)
if err != nil {
log.Errorf("failed to remove qbittorrent temp dir: %+v", err.Error())
}
}()
for _, file := range files {
tempPath := filepath.Join(m.tempDir, file.Name)
dstPath := filepath.Join(dstBaseDir, file.Name)
dstDir := filepath.Dir(dstPath)
fileName := filepath.Base(dstPath)
TransferTaskManager.Submit(task.WithCancelCtx(&task.Task[uint64]{
Name: fmt.Sprintf("transfer %s to [%s](%s)", tempPath, storage.GetStorage().MountPath, dstPath),
Func: func(tsk *task.Task[uint64]) error {
defer wg.Done()
size := file.Size
mimetype := utils.GetMimeType(tempPath)
f, err := os.Open(tempPath)
if err != nil {
return errors.Wrapf(err, "failed to open file %s", tempPath)
}
s := stream.FileStream{
Obj: &model.Object{
Name: fileName,
Size: size,
Modified: time.Now(),
IsFolder: false,
},
Reader: f,
Closers: utils.NewClosers(f),
Mimetype: mimetype,
}
ss, err := stream.NewSeekableStream(s, nil)
if err != nil {
return err
}
return op.Put(tsk.Ctx, storage, dstDir, ss, tsk.SetProgress)
},
}))
}
return nil
}

23
internal/qbittorrent/qbittorrent.go
View File

@ -1,23 +0,0 @@
package qbittorrent
import (
"github.com/alist-org/alist/v3/internal/conf"
"github.com/alist-org/alist/v3/internal/setting"
"github.com/alist-org/alist/v3/pkg/task"
)
var DownTaskManager = task.NewTaskManager[string](3)
var qbclient Client
func InitClient() error {
var err error
qbclient = nil
url := setting.GetStr(conf.QbittorrentUrl)
qbclient, err = New(url)
return err
}
func IsQbittorrentReady() bool {
return qbclient != nil
}

5
server/handles/task.go
View File

@ -5,7 +5,6 @@ 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/internal/qbittorrent"
"github.com/alist-org/alist/v3/pkg/task"
"github.com/alist-org/alist/v3/server/common"
"github.com/gin-gonic/gin"
@ -118,10 +117,6 @@ func taskRoute[K comparable](g *gin.RouterGroup, manager *task.Manager[K], k2Str
func SetupTaskRoute(g *gin.RouterGroup) {
taskRoute(g.Group("/upload"), fs.UploadTaskManager, uint64K2Str, str2Uint64K)
taskRoute(g.Group("/copy"), fs.CopyTaskManager, uint64K2Str, str2Uint64K)
taskRoute(g.Group("/qbit_down"), qbittorrent.DownTaskManager, strK2Str, str2StrK)
taskRoute(g.Group("/qbit_transfer"), qbittorrent.TransferTaskManager, uint64K2Str, str2Uint64K)
//taskRoute(g.Group("/aria2_down"), aria2.DownTaskManager, strK2Str, str2StrK)
//taskRoute(g.Group("/aria2_transfer"), aria2.TransferTaskManager, uint64K2Str, str2Uint64K)
taskRoute(g.Group("/offline_download"), tool.DownTaskManager, strK2Str, str2StrK)
taskRoute(g.Group("/offline_download_transfer"), tool.TransferTaskManager, uint64K2Str, str2Uint64K)
}