Clone/aliyunpan
1
0
Fork 0
mirror of https://github.com/tickstep/aliyunpan.git synced 2025-01-23 14:32:14 +08:00
Code Issues Actions Packages Projects Releases Wiki Activity

add multi user download option for download command

Browse Source
This commit is contained in:
tickstep 2024-08-09 21:20:40 +08:00
parent c05148df16
commit 97ee446b32
4 changed files with 192 additions and 28 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

53
internal/command/download.go
View File

@ -25,13 +25,13 @@ import (
"github.com/tickstep/aliyunpan/internal/utils"
"github.com/tickstep/aliyunpan/library/requester/transfer"
"github.com/tickstep/library-go/converter"
"github.com/tickstep/library-go/logger"
"github.com/tickstep/library-go/requester/rio/speeds"
"github.com/urfave/cli"
"os"
"path"
"path/filepath"
"runtime"
"time"
)
type (
@ -49,6 +49,7 @@ type (
ShowProgress bool
DriveId string
ExcludeNames []string // 排除的文件名，包括文件夹和文件。即这些文件/文件夹不进行下载，支持正则表达式
IsMultiUserDownload bool // 是否启用多用户混合下载
}
// LocateDownloadOption 获取下载链接可选参数
@ -138,6 +139,7 @@ func CmdDownload() cli.Command {
ShowProgress: !c.Bool("np"),
DriveId: parseDriveId(c),
ExcludeNames: c.StringSlice("exn"),
IsMultiUserDownload: c.Bool("md"),
}
// 获取下载文件锁，保证下载操作单实例
@ -210,6 +212,10 @@ func CmdDownload() cli.Command {
Usage: "exclude name，指定排除的文件夹或者文件的名称，被排除的文件不会进行下载，只支持正则表达式。支持同时排除多个名称，每一个名称就是一个exn参数",
Value: nil,
},
cli.BoolFlag{
Name: "md",
Usage: "(BETA) Multi-user Download，使用多用户混合下载，可以叠加所有用户的下载速度",
},
},
}
}
@ -309,14 +315,46 @@ func RunDownload(paths []string, options *DownloadOptions) {
fmt.Println(err)
return
}
fmt.Printf("\n[0] 当前文件下载最大并发量为: %d, 单文件下载分片线程数为: %d, 下载缓存为: %s\n", options.Parallel, options.SliceParallel, converter.ConvertFileSize(int64(cfg.CacheSize), 2))
// 阿里OpenAPI规定：文件分片下载的并发数为3，即某用户使用 App 时，可以同时下载 1 个文件的 3 个分片，或者同时下载 3 个文件的各 1 个分片。
// 超过并发，调用接口，报错 http status：403，并且下载速度为0
if options.Parallel*options.SliceParallel > 3 {
fmt.Println("\n####### 当前文件下载的并发数已经超过阿里云盘的限制，可能会导致下载速度为0！ #######\n")
time.Sleep(3 * time.Second)
// 多用户下载的辅助账号列表
var subPanClientList []*config.PanClient
if options.IsMultiUserDownload { // 多用户下载
c := config.Config
for _, u := range config.Config.UserList {
if u.UserId == activeUser.UserId {
// 当前登录用户，作为主用户，跳过
continue
}
// 初始化客户端
user, err := config.SetupUserByCookie(u.OpenapiToken, u.WebapiToken,
u.TicketId, u.UserId,
c.DeviceId, c.DeviceName,
c.ClientId, c.ClientSecret)
if err != nil {
logger.Verboseln("setup user error")
continue
}
if subPanClientList == nil {
subPanClientList = []*config.PanClient{}
}
subPanClientList = append(subPanClientList, user.PanClient())
}
if subPanClientList == nil || len(subPanClientList) == 0 {
fmt.Printf("\n当前登录用户只有一个，无法启用多用户混合下载\n")
subPanClientList = nil
}
}
if subPanClientList != nil || len(subPanClientList) > 0 {
// 已启用多用户下载
userCount := len(subPanClientList) + 1
fmt.Printf("\n*** 已启用多用户混合下载，用户数: %d ***\n", userCount)
// 多用户下载，并发数必须为1，以获得最大下载速度
options.Parallel = 1
// 阿里OpenAPI规定：文件分片下载的并发数为3，即某用户使用 App 时，可以同时下载 1 个文件的 3 个分片，或者同时下载 3 个文件的各 1 个分片。
options.SliceParallel = userCount * 3
}
fmt.Printf("\n[0] 当前文件下载最大并发量为: %d, 单文件下载分片线程数为: %d, 下载缓存为: %s\n", options.Parallel, options.SliceParallel, converter.ConvertFileSize(int64(cfg.CacheSize), 2))
var (
panClient = activeUser.PanClient()
@ -365,6 +403,7 @@ func RunDownload(paths []string, options *DownloadOptions) {
unit := pandownload.DownloadTaskUnit{
Cfg: &newCfg, // 复制一份新的cfg
PanClient: panClient,
SubPanClientList: subPanClientList,
VerbosePrinter: panCommandVerbose,
PrintFormat: downloadPrintFormat(options.Load),
ParentTaskExecutor: &executor,

9
internal/config/pan_user.go
View File

@ -62,6 +62,15 @@ func (d DriveInfoList) GetResourceDriveId() string {
return ""
}
func (d DriveInfoList) GetDriveIdByName(name string) string {
for _, drive := range d {
if drive.DriveTag == name {
return drive.DriveId
}
}
return ""
}
// PanClientToken 授权Token
type PanClientToken struct {
// AccessToken AccessToken

155
internal/file/downloader/downloader.go
View File

@ -64,6 +64,7 @@ type (
writer io.WriterAt
client *requester.HTTPClient
panClient *config.PanClient
subPanClientList []*config.PanClient // 辅助下载子账号列表
config *Config
monitor *Monitor
instanceState *InstanceState
@ -73,14 +74,24 @@ type (
DURLCheckFunc func(client *requester.HTTPClient, durl string) (contentLength int64, resp *http.Response, err error)
// StatusCodeBodyCheckFunc 响应状态码出错的检查函数
StatusCodeBodyCheckFunc func(respBody io.Reader) error
// panClientDownloadUrlEntity 下载url实体，和网盘(用户)客户端绑定
panClientDownloadUrlEntity struct {
PanClient *config.PanClient
FileInfo *aliyunpan.FileEntity
DriveId string
FileId string
FileUrl string
}
)
// NewDownloader 初始化Downloader
func NewDownloader(writer io.WriterAt, config *Config, p *config.PanClient, globalSpeedsStat *speeds.Speeds) (der *Downloader) {
func NewDownloader(writer io.WriterAt, config *Config, p *config.PanClient, sp []*config.PanClient, globalSpeedsStat *speeds.Speeds) (der *Downloader) {
der = &Downloader{
config: config,
writer: writer,
panClient: p,
subPanClientList: sp,
globalSpeedsStat: globalSpeedsStat,
}
return
@ -379,41 +390,51 @@ func (der *Downloader) Execute() error {
)
// 获取下载链接
var apierr *apierror.ApiError
durl, apierr := der.panClient.OpenapiPanClient().GetFileDownloadUrl(&aliyunpan.GetFileDownloadUrlParam{
DriveId: der.driveId,
FileId: der.fileInfo.FileId,
})
time.Sleep(time.Duration(200) * time.Millisecond)
if apierr != nil {
//var apierr *apierror.ApiError
//durl, apierr := der.panClient.OpenapiPanClient().GetFileDownloadUrl(&aliyunpan.GetFileDownloadUrlParam{
// DriveId: der.driveId,
// FileId: der.fileInfo.FileId,
//})
//time.Sleep(time.Duration(200) * time.Millisecond)
//if apierr != nil {
// logger.Verbosef("ERROR: get download url error: %s\n", der.fileInfo.FileId)
// cmdutil.Trigger(der.onCancelEvent)
// return apierr
//}
//if durl == nil || durl.Url == "" || strings.HasPrefix(durl.Url, aliyunpan.IllegalDownloadUrlPrefix) {
// logger.Verbosef("无法获取有效的下载链接: %+v\n", durl)
// cmdutil.Trigger(der.onCancelEvent)
// der.removeInstanceState() // 移除断点续传文件
// cmdutil.Trigger(der.onFailedEvent)
// return ErrFileDownloadForbidden
//}
// 获取各个网盘客户端的下载链接
panClientFileUrl, err := der.getFileAllClientDownloadUrl()
if err != nil || panClientFileUrl == nil {
logger.Verbosef("ERROR: get download url error: %s\n", der.fileInfo.FileId)
cmdutil.Trigger(der.onCancelEvent)
return apierr
}
if durl == nil || durl.Url == "" || strings.HasPrefix(durl.Url, aliyunpan.IllegalDownloadUrlPrefix) {
logger.Verbosef("无法获取有效的下载链接: %+v\n", durl)
cmdutil.Trigger(der.onCancelEvent)
der.removeInstanceState() // 移除断点续传文件
cmdutil.Trigger(der.onFailedEvent)
return ErrFileDownloadForbidden
return err
}
// 初始化下载worker
factorNum := len(bii.Ranges) / len(panClientFileUrl)
for k, r := range bii.Ranges {
panClientUrl := panClientFileUrl[int(k/factorNum)]
loadBalancer := loadBalancerResponseList.SequentialGet()
if loadBalancer == nil {
continue
}
logger.Verbosef("work id: %d, download url: %v\n", k, durl)
logger.Verbosef("work id: %d, download url: %v\n", k, panClientUrl.FileUrl)
client := requester.NewHTTPClient()
client.SetKeepAlive(true)
client.SetTimeout(10 * time.Minute)
realUrl := durl.Url
worker := NewWorker(k, der.driveId, der.fileInfo.FileId, realUrl, writer, der.globalSpeedsStat)
realUrl := panClientUrl.FileUrl
worker := NewWorker(k, panClientUrl.DriveId, panClientUrl.FileInfo.FileId, realUrl, writer, der.globalSpeedsStat)
worker.SetClient(client)
worker.SetPanClient(der.panClient)
worker.SetPanClient(panClientUrl.PanClient)
worker.SetWriteMutex(writeMu)
worker.SetTotalSize(der.fileInfo.FileSize)
@ -455,6 +476,100 @@ func (der *Downloader) Execute() error {
return err
}
// 获取对应网盘的下载链接
func (der *Downloader) getFileAllClientDownloadUrl() ([]*panClientDownloadUrlEntity, error) {
result := []*panClientDownloadUrlEntity{}
// 主账号（必须存在）
// 获取下载链接
var apierr *apierror.ApiError
durl, apierr := der.panClient.OpenapiPanClient().GetFileDownloadUrl(&aliyunpan.GetFileDownloadUrlParam{
DriveId: der.driveId,
FileId: der.fileInfo.FileId,
})
time.Sleep(time.Duration(200) * time.Millisecond)
if apierr != nil {
logger.Verbosef("ERROR: get download url error: %s\n", der.fileInfo.FileId)
cmdutil.Trigger(der.onCancelEvent)
return nil, apierr
}
if durl == nil || durl.Url == "" || strings.HasPrefix(durl.Url, aliyunpan.IllegalDownloadUrlPrefix) {
logger.Verbosef("无法获取有效的下载链接: %+v\n", durl)
cmdutil.Trigger(der.onCancelEvent)
der.removeInstanceState() // 移除断点续传文件
cmdutil.Trigger(der.onFailedEvent)
return nil, ErrFileDownloadForbidden
}
result = append(result, &panClientDownloadUrlEntity{
PanClient: der.panClient,
FileInfo: der.fileInfo,
DriveId: der.driveId,
FileId: der.fileInfo.FileId,
FileUrl: durl.Url,
})
// 网盘名称
mainDriveName := ""
openUserInfo, err := der.panClient.OpenapiPanClient().GetUserInfo()
if err != nil || openUserInfo == nil {
return nil, err
}
if openUserInfo.FileDriveId == der.driveId {
mainDriveName = "File"
} else if openUserInfo.ResourceDriveId == der.driveId {
mainDriveName = "Resource"
}
// 网盘文件路径
mainFileFullPath := der.fileInfo.Path
// 铺助账号的下载链接
if der.subPanClientList != nil {
for _, spc := range der.subPanClientList {
driveId := ""
userInfo, err1 := spc.OpenapiPanClient().GetUserInfo()
if err1 != nil {
continue
}
if mainDriveName == "File" {
driveId = userInfo.FileDriveId
} else if mainDriveName == "Resource" {
driveId = userInfo.ResourceDriveId
}
// 文件信息
panfileInfo, err2 := spc.OpenapiPanClient().FileInfoByPath(driveId, mainFileFullPath)
if err2 != nil || panfileInfo == nil {
continue
}
// 下载链接
durl2, apierr := spc.OpenapiPanClient().GetFileDownloadUrl(&aliyunpan.GetFileDownloadUrlParam{
DriveId: driveId,
FileId: panfileInfo.FileId,
})
time.Sleep(time.Duration(200) * time.Millisecond)
if apierr != nil {
logger.Verbosef("ERROR: get download url error: %s\n", der.fileInfo.FileId)
continue
}
if durl2 == nil || durl2.Url == "" || strings.HasPrefix(durl2.Url, aliyunpan.IllegalDownloadUrlPrefix) {
logger.Verbosef("无法获取有效的下载链接: %+v\n", durl2)
continue
}
result = append(result, &panClientDownloadUrlEntity{
PanClient: spc,
FileInfo: panfileInfo,
DriveId: driveId,
FileId: panfileInfo.FileId,
FileUrl: durl2.Url,
})
}
}
return result, nil
}
// downloadStatusEvent 执行状态处理事件
func (der *Downloader) downloadStatusEvent() {
if der.onDownloadStatusEvent == nil {

3
internal/functions/pandownload/download_task_unit.go
View File

@ -48,6 +48,7 @@ type (
Cfg *downloader.Config
PanClient *config.PanClient
SubPanClientList []*config.PanClient // 辅助下载子账号列表
ParentTaskExecutor *taskframework.TaskExecutor
DownloadStatistic *DownloadStatistic // 下载统计
@ -170,7 +171,7 @@ func (dtu *DownloadTaskUnit) download() (err error) {
}
defer file.Close()
der := downloader.NewDownloader(writer, dtu.Cfg, dtu.PanClient, dtu.GlobalSpeedsStat)
der := downloader.NewDownloader(writer, dtu.Cfg, dtu.PanClient, dtu.SubPanClientList, dtu.GlobalSpeedsStat)
der.SetFileInfo(dtu.fileInfo)
der.SetDriveId(dtu.DriveId)
der.SetStatusCodeBodyCheckFunc(func(respBody io.Reader) error {