Studio/aliyunpan
5
0
Fork 0
mirror of https://gitee.com/zhangsken/aliyunpan.git synced 2026-10-16 02:23:49 +08:00
Code Issues Packages Projects Releases Wiki Activity

Merge branch 'main' of github.com:tickstep/aliyunpan

Browse Source
This commit is contained in:
tickstep
2026-02-11 19:33:53 +08:00
parent 6672aa4cd4 1916dda8f8
commit 1822d3c2b2
11 changed files with 217 additions and 48 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/command/rm.go
+18 -1
View File
@@ -74,6 +74,21 @@ func CmdRm() cli.Command {
}
}
func queryFileListByPath(activeUser *config.PanUser, driveId, filePathPattern string) (files []*aliyunpan.FileEntity, e error) {
if !strings.ContainsAny(filePathPattern, aliyunpan.ShellPatternCharacters) {
// 没有通配符，直接查询
fi, err := activeUser.PanClient().OpenapiPanClient().FileInfoByPath(driveId, filePathPattern)
if err != nil {
return nil, err
}
files = append(files, fi)
} else {
// 包含通配符，使用通配符查询
files, e = matchPathByShellPattern(driveId, filePathPattern)
}
return files, e
}
// RunRemove 执行 批量删除文件/目录
func RunRemove(driveId string, paths ...string) {
activeUser := GetActiveUser()
@@ -86,7 +101,9 @@ func RunRemove(driveId string, paths ...string) {
for _, p := range paths {
absolutePath := path.Clean(activeUser.PathJoin(driveId, p))
fileList, err1 := matchPathByShellPattern(driveId, absolutePath)
// 查询路径对应的文件信息
fileList, err1 := queryFileListByPath(activeUser, driveId, absolutePath)
if err1 != nil {
failedRmPaths = append(failedRmPaths, absolutePath)
continue
internal/file/downloader/monitor.go
+1
View File
@@ -346,6 +346,7 @@ func (mt *Monitor) ResetWorker(worker *Worker) {
// 忽略, 返回
return
case StatusCodeDownloadUrlExpired: // 下载链接已经过期
logger.Verbosef("download url expired, reset worker: %d\n", worker.ID())
worker.RefreshDownloadUrl()
break
case StatusCodeDownloadUrlExceedMaxConcurrency: // 下载遇到限流报错
internal/file/downloader/worker.go
+5 -1
View File
@@ -205,6 +205,7 @@ func (wer *Worker) Cancel() error {
// Reset 重设连接
func (wer *Worker) Reset() {
logger.Verbosef("Reset worker: %d\n", wer.ID())
if wer.resetFunc == nil {
logger.Verbosef("DEBUG: worker: resetFunc not set")
return
@@ -220,13 +221,15 @@ func (wer *Worker) Reset() {
// RefreshDownloadUrl 重新刷新下载链接
func (wer *Worker) RefreshDownloadUrl() {
var apierr *apierror.ApiError
logger.Verbosef("get new download url for worker: %d\n", wer.ID())
durl, apierr := wer.panClient.OpenapiPanClient().GetFileDownloadUrl(&aliyunpan.GetFileDownloadUrlParam{DriveId: wer.driveId, FileId: wer.fileId})
if apierr != nil {
logger.Verbosef("get new download url for worker: %d, error: %+v\n", wer.ID(), apierr)
wer.status.statusCode = StatusCodeTooManyConnections
return
}
wer.url = durl.Url
logger.Verbosef("get new download url for worker: %d, new url: %s\n", wer.ID(), wer.url)
}
// Canceled 是否已经取消
@@ -266,6 +269,7 @@ func (wer *Worker) Err() error {
// Execute 执行任务
func (wer *Worker) Execute() {
logger.Verbosef("Execute worker: %d\n", wer.ID())
wer.lazyInit()
wer.execMu.Lock()
internal/file/uploader/error.go
+7 -6
View File
@@ -16,12 +16,13 @@ package uploader
import "fmt"
var (
UploadUrlExpired = fmt.Errorf("UrlExpired")
UploadPartNotSeq = fmt.Errorf("PartNotSequential")
UploadNoSuchUpload = fmt.Errorf("NoSuchUpload")
UploadTerminate = fmt.Errorf("UploadErrorTerminate")
UploadPartAlreadyExist = fmt.Errorf("PartAlreadyExist")
UploadHttpError = fmt.Errorf("HttpError")
UploadUrlExpired = fmt.Errorf("UrlExpired")
UploadPartNotSeq = fmt.Errorf("PartNotSequential")
UploadNoSuchUpload = fmt.Errorf("NoSuchUpload")
UploadTerminate = fmt.Errorf("UploadErrorTerminate")
UploadPartAlreadyExist = fmt.Errorf("PartAlreadyExist")
UploadHttpError = fmt.Errorf("HttpError")
UploadLocalFileAlreadyClosedError = fmt.Errorf("LocalFileAlreadyClosedError")
)
type (
internal/file/uploader/multiworker.go
+8 -5
View File
@@ -150,9 +150,10 @@ func (muer *MultiUploader) upload() (uperr error) {
}()
wg.Wait()
if uperr != nil {
if errors.Is(uperr, UploadPartNotSeq) || errors.Is(uperr, UploadNoSuchUpload) {
// 分片出现乱序，停止上传
// 清空数据，准备重新上传
if errors.Is(uperr, UploadPartNotSeq) ||
errors.Is(uperr, UploadNoSuchUpload) ||
errors.Is(uperr, UploadLocalFileAlreadyClosedError) {
// 清空数据，停止上传
uploadDeque = lane.NewDeque() // 清空待上传列表
}
}
@@ -165,8 +166,10 @@ func (muer *MultiUploader) upload() (uperr error) {
// 释放链路
uploadClient.CloseIdleConnections()
// 返回错误，通知上层客户端
if errors.Is(uperr, UploadPartNotSeq) || errors.Is(uperr, UploadNoSuchUpload) {
// 返回错误，通知上层客户端进行处理
if errors.Is(uperr, UploadPartNotSeq) ||
errors.Is(uperr, UploadNoSuchUpload) ||
errors.Is(uperr, UploadLocalFileAlreadyClosedError) {
return uperr
}
internal/functions/pandownload/download_task_unit.go
+2
View File
@@ -647,6 +647,8 @@ func (dtu *DownloadTaskUnit) Run() (result *taskframework.TaskUnitRunResult) {
}
dtu.logf("[%s] 将会下载到路径: %s\n", dtu.taskInfo.Id(), dtu.SavePath)
// 更新UI面板任务状态
dtu.updateUITaskState(ui.TaskRunning, "")
var ok bool
er := dtu.download()
internal/functions/panupload/upload.go
+35 -6
View File
@@ -16,12 +16,16 @@ package panupload
import (
"context"
"encoding/xml"
"errors"
"github.com/tickstep/aliyunpan/internal/config"
"github.com/tickstep/library-go/logger"
"github.com/tickstep/library-go/requester"
"io"
"net/http"
"net/url"
"os"
"strconv"
"strings"
"time"
"github.com/tickstep/aliyunpan-api/aliyunpan"
@@ -106,6 +110,7 @@ func (pu *PanUpload) UploadFile(ctx context.Context, partseq int, partOffset int
}
var respErr *uploader.MultiError
// 分片上传回调函数，供阿里云盘API接口使用
uploadFunc := func(httpMethod, fullUrl string, headers map[string]string) (*http.Response, error) {
var resp *http.Response
var respError error = nil
@@ -128,10 +133,10 @@ func (pu *PanUpload) UploadFile(ctx context.Context, partseq int, partOffset int
if blen > 0 {
buf := make([]byte, blen)
resp.Body.Read(buf)
logger.Verbosef("分片上传出错: 分片%d => %s\n", partseq+1, string(buf))
logger.Verbosef("分片上传出错: 分片%d, 错误信息: %s\n", partseq+1, string(buf))
errResp := &apierror.ErrorXmlResp{}
if err := xml.Unmarshal(buf, errResp); err == nil {
if errXml := xml.Unmarshal(buf, errResp); errXml == nil {
if errResp.Code != "" {
if "PartNotSequential" == errResp.Code {
respError = uploader.UploadPartNotSeq
@@ -188,10 +193,31 @@ func (pu *PanUpload) UploadFile(ctx context.Context, partseq int, partOffset int
}
}
} else {
// 默认错误
respError = uploader.UploadTerminate
respErr = &uploader.MultiError{
Terminated: true,
}
// 解析HTTP请求出现的错误信息
if err != nil {
var netErr *url.Error
if errors.As(err, &netErr) {
var pathErr *os.PathError
if errors.As(netErr.Err, &pathErr) { // 本地文件读取异常
if strings.Contains(pathErr.Err.Error(), "file already closed") {
// 本地文件流触发"file already closed"错误
// 一般是文件上传过程中，文件被删除了，或者文件所在的硬盘异常导致文件流被系统关闭
respError = uploader.UploadLocalFileAlreadyClosedError
respErr = &uploader.MultiError{
Terminated: true,
NeedStartOver: false,
Err: uploader.UploadLocalFileAlreadyClosedError,
}
}
}
}
}
}
return resp, respError
}
@@ -201,7 +227,7 @@ func (pu *PanUpload) UploadFile(ctx context.Context, partseq int, partOffset int
apiError := pu.panClient.OpenapiPanClient().UploadFileData(uploadUrl, uploadFunc)
if respErr != nil {
if respErr.Err == uploader.UploadUrlExpired {
if errors.Is(respErr.Err, uploader.UploadUrlExpired) {
// URL过期，获取新的URL
guur, er := pu.panClient.OpenapiPanClient().GetUploadUrl(&aliyunpan.GetUploadUrlParam{
DriveId: pu.driveId,
@@ -217,16 +243,19 @@ func (pu *PanUpload) UploadFile(ctx context.Context, partseq int, partOffset int
// 获取新的上传URL重试一次
pu.uploadOpEntity.PartInfoList[partseq] = guur.PartInfoList[0]
uploadUrl := pu.uploadOpEntity.PartInfoList[partseq].UploadURL
uploadUrl = pu.uploadOpEntity.PartInfoList[partseq].UploadURL
apiError = pu.panClient.OpenapiPanClient().UploadFileData(uploadUrl, uploadFunc)
} else if respErr.Err == uploader.UploadPartAlreadyExist {
} else if errors.Is(respErr.Err, uploader.UploadPartAlreadyExist) {
// already upload
// success
return true, nil
} else if respErr.Err == uploader.UploadPartNotSeq {
} else if errors.Is(respErr.Err, uploader.UploadPartNotSeq) {
// 上传分片乱序了
// 返回错误，然后重传
return false, respErr
} else if errors.Is(respErr.Err, uploader.UploadLocalFileAlreadyClosedError) {
// 本地文件读取错误
return false, respErr
} else {
return false, respErr
}
internal/functions/panupload/upload_task_unit.go
+27 -1
View File
@@ -170,6 +170,13 @@ func (utu *UploadTaskUnit) upload() (result *taskframework.TaskUnitRunResult) {
MaxRate: config.Config.MaxUploadRate,
}, utu.LocalFileChecksum.UploadOpEntity, utu.PanClient, utu.GlobalSpeedsStat)
// 模拟 file already closed 错误
//go func(f *os.File) {
// time.Sleep(10 * time.Second)
// f.Close()
// fmt.Printf("关闭文件流: %s\n", f.Name())
//}(utu.LocalFileChecksum.GetFile())
// 设置断点续传
if utu.state != nil {
muer.SetInstanceState(utu.state)
@@ -192,7 +199,10 @@ func (utu *UploadTaskUnit) upload() (result *taskframework.TaskUnitRunResult) {
} else {
leftStr = left.String()
}
uploadedPercentage := fmt.Sprintf("%.2f%%", float64(status.Uploaded())/float64(status.TotalSize())*100)
uploadedPercentage := "0.00%"
if status.TotalSize() > 0 {
uploadedPercentage = fmt.Sprintf("%.2f%%", float64(status.Uploaded())/float64(status.TotalSize())*100)
}
fmt.Printf("\r[%s] ↑ %s/%s(%s) %s/s(%s/s) in %s, left %s ............", utu.taskInfo.Id(),
converter.ConvertFileSize(status.Uploaded(), 2),
converter.ConvertFileSize(status.TotalSize(), 2),
@@ -595,6 +605,22 @@ stepUploadUpload:
utu.state = nil
goto StepUploadPrepareUpload
}
if errors.Is(uploadResult.Err, uploader.UploadLocalFileAlreadyClosedError) {
// 本地文件读取错误
fmt.Printf("[%s] %s 本地文件读取错误，建议检查文件是否存在，是否有文件读取权限，或者硬盘是否有异常\n", utu.taskInfo.Id(), time.Now().Format("2006-01-02 15:04:06"))
uploadResult = &taskframework.TaskUnitRunResult{
Succeed: false,
NeedRetry: false,
Cancel: false,
Err: uploadResult.Err,
ResultCode: 0,
ResultMessage: "",
Extra: nil,
}
return uploadResult
}
// 处理阿里云盘API接口错误
var apier *apierror.ApiError
if errors.As(uploadResult.Err, &apier) {
// 上传任务过期
internal/panupdate/panupdate.go
+19 -18
View File
@@ -17,6 +17,16 @@ import (
"archive/zip"
"bytes"
"fmt"
"io/ioutil"
"net/http"
"path/filepath"
"regexp"
"runtime"
"strconv"
"strings"
"sync"
"time"
jsoniter "github.com/json-iterator/go"
"github.com/tickstep/aliyunpan/cmder/cmdliner"
"github.com/tickstep/aliyunpan/cmder/cmdutil"
@@ -27,19 +37,9 @@ import (
"github.com/tickstep/library-go/cachepool"
"github.com/tickstep/library-go/checkaccess"
"github.com/tickstep/library-go/converter"
"github.com/tickstep/library-go/getip"
"github.com/tickstep/library-go/jsonhelper"
"github.com/tickstep/library-go/logger"
"github.com/tickstep/library-go/requester"
"io/ioutil"
"net/http"
"path/filepath"
"regexp"
"runtime"
"strconv"
"strings"
"sync"
"time"
)
const (
@@ -58,16 +58,16 @@ type tsResp struct {
Msg string `json:"msg"`
}
func getReleaseFromTicstep(client *requester.HTTPClient, showPrompt bool) *ReleaseInfo {
func getReleaseFromTickstep(client *requester.HTTPClient, showPrompt bool) *ReleaseInfo {
tsReleaseInfo := &ReleaseInfo{}
tsRespObj := &tsResp{Data: tsReleaseInfo}
fullUrl := strings.Builder{}
ipAddr, err := getip.IPInfoFromTechainBaidu()
ipAddr, err := utils.GetPublicIP()
if err != nil {
ipAddr = "127.0.0.1"
}
fmt.Fprintf(&fullUrl, "http://api.tickstep.com/update/tickstep/aliyunpan/releases/latest?ip=%s&os=%s&arch=%s&version=%s",
ipAddr, runtime.GOOS, runtime.GOARCH, global.AppVersion)
fmt.Fprintf(&fullUrl, "http://api.tickstep.com/update/tickstep/aliyunpan/releases/latest?ip=%s&os=%s&arch=%s&version=%s", ipAddr, runtime.GOOS, runtime.GOARCH, global.AppVersion)
//fmt.Fprintf(&fullUrl, "http://localhost:8997/update/tickstep/aliyunpan/releases/latest?ip=%s&os=%s&arch=%s&version=%s", ipAddr, runtime.GOOS, runtime.GOARCH, global.AppVersion)
resp, err := client.Req(http.MethodGet, fullUrl.String(), nil, nil)
if resp != nil {
defer resp.Body.Close()
@@ -136,10 +136,10 @@ func GetLatestReleaseInfo(showPrompt bool) *ReleaseInfo {
client.SetTimeout(time.Duration(0) * time.Second)
client.SetKeepAlive(true)
// check tickstep srv
var tsReleaseInfo *ReleaseInfo = getReleaseFromTicstep(client, showPrompt)
// tickstep server
var tsReleaseInfo = getReleaseFromTickstep(client, showPrompt)
// github
// github server
var ghReleaseInfo *ReleaseInfo = nil
for idx := 0; idx < 2; idx++ {
ghReleaseInfo = getReleaseFromGithub(client, showPrompt)
@@ -151,10 +151,11 @@ func GetLatestReleaseInfo(showPrompt bool) *ReleaseInfo {
var releaseInfo *ReleaseInfo = nil
if config.Config.UpdateCheckInfo.PreferUpdateSrv == "tickstep" {
// theoretically, tickstep server will be faster at mainland
// theoretically, tickstep server will be faster in China mainland
releaseInfo = tsReleaseInfo
} else {
releaseInfo = ghReleaseInfo
// 如果github的版本信息获取不到（例如网络不稳定），可以尝试使用tickstep服务进行升级
if ghReleaseInfo == nil || ghReleaseInfo.TagName == "" {
releaseInfo = tsReleaseInfo
}
internal/ui/download_dashboard.go
+9 -6
View File
@@ -15,6 +15,7 @@ package ui
import (
"fmt"
"github.com/tickstep/aliyunpan/internal/utils"
"io"
"math"
"os"
@@ -430,9 +431,8 @@ func (db *DownloadDashboard) headerLines(snapshot dashboardSnapshot, innerWidth
}
}
speedStr := converter.ConvertFileSize(speed, 2) + "/s"
line1 := fmt.Sprintf("总速度: %s | 文件: %d/%d | 失败: %d | 已用时间: %s | 剩余时间: %s | 状态: %s",
speedStr, doneFiles, totalFiles, failedFiles, formatDurationShort(elapsed), left, status)
line1 := fmt.Sprintf("总速度: %s | 文件: %d/%d | 失败: %d | 状态: %s | 已用时间: %s | 剩余时间: %s",
utils.FormatSpeedFixedWidth(speed, 11), doneFiles, totalFiles, failedFiles, status, formatDurationShort(elapsed), left)
line1 = fitLine(line1, innerWidth)
progressPct := 0.0
@@ -611,7 +611,10 @@ func renderTaskLine(task *downloadTask, index int, width int) string {
percent := "--.-%"
if task.total > 0 {
progressPct = float64(task.downloaded) / float64(task.total)
percent = fmt.Sprintf("%5.1f%%", math.Min(100, progressPct*100))
percent = fmt.Sprintf("%5.1f%%", math.Min(100, progressPct*100)) // 将浮点数格式化为：总宽度 5 个字符，包含 1 位小数，后面加百分号
} else {
// 空闲队列左边多一个空格，对齐活跃队列
percent = " --.-%"
}
// 速度
@@ -644,7 +647,7 @@ func renderTaskLine(task *downloadTask, index int, width int) string {
rateWidth := 12
etaWidth := 8
percentWidth := displayWidth(percent)
fixed := displayWidth(prefix) + 1 + 2 + barWidth + 1 + percentWidth + 1 + rateWidth + 1 + etaWidth
fixed := displayWidth(prefix) + 1 + 2 /*进度条左右[]括号*/ + barWidth + 1 + percentWidth + 1 + rateWidth + 1 + etaWidth
nameWidth := width - fixed
if nameWidth < 8 {
reduce := 8 - nameWidth
@@ -663,7 +666,7 @@ func renderTaskLine(task *downloadTask, index int, width int) string {
etaWidth -= cut
reduce -= cut
}
fixed = displayWidth(prefix) + 1 + 2 + barWidth + 1 + percentWidth + 1 + rateWidth + 1 + etaWidth
fixed = displayWidth(prefix) + 1 /*1个空格分隔*/ + 2 /*进度条左右[]括号*/ + barWidth /*进度条长度*/ + 1 /*1个空格分隔*/ + percentWidth + 1 /*1个空格分隔*/ + rateWidth + 1 /*1个空格分隔*/ + etaWidth
nameWidth = width - fixed
if nameWidth < 4 {
nameWidth = 4
internal/utils/utils.go
+86 -4
View File
@@ -16,15 +16,13 @@ package utils
import (
"compress/gzip"
"crypto/md5"
"encoding/json"
"flag"
"fmt"
jsoniter "github.com/json-iterator/go"
uuid "github.com/satori/go.uuid"
"github.com/tickstep/aliyunpan-api/aliyunpan"
"github.com/tickstep/library-go/ids"
"io"
"io/ioutil"
"math"
"net/http"
"net/http/cookiejar"
"net/url"
"os"
@@ -34,6 +32,11 @@ import (
"strconv"
"strings"
"time"
jsoniter "github.com/json-iterator/go"
uuid "github.com/satori/go.uuid"
"github.com/tickstep/aliyunpan-api/aliyunpan"
"github.com/tickstep/library-go/ids"
)
// TrimPathPrefix 去除目录的前缀
@@ -301,3 +304,82 @@ func UnixTime2LocalFormatStr(unixTime int64) string {
cz := time.FixedZone("CST", 8*3600) // 东8区
return t.In(cz).Format("2006-01-02T15:04:05.000Z")
}
// FormatSpeedFixedWidth 格式化速度显示
// speed: 速度值，单位 Byte/s
// 返回格式: " xxx.xx MB/s" (固定长度，左侧用空格补齐)
func FormatSpeedFixedWidth(speed int64, totalWidth int) string {
const (
kb = 1024.0
mb = kb * 1024
gb = mb * 1024
tb = gb * 1024
)
var value float64
var unit string
speedFloat := float64(speed)
switch {
case speedFloat >= tb:
value = speedFloat / tb
unit = "TB/s"
case speedFloat >= gb:
value = speedFloat / gb
unit = "GB/s"
case speedFloat >= mb:
value = speedFloat / mb
unit = "MB/s"
case speedFloat >= kb:
value = speedFloat / kb
unit = "KB/s"
default:
value = speedFloat
unit = "B/s"
}
// 格式化为两位小数
formatted := fmt.Sprintf("%.2f %s", value, unit)
// 如果不足指定宽度，左侧用空格补齐
if len(formatted) < totalWidth {
spaces := strings.Repeat(" ", totalWidth-len(formatted))
return spaces + formatted
}
return formatted
}
func GetPublicIP() (string, error) {
type IPResponse struct {
Origin string `json:"origin"`
}
resp, err := http.Get("https://httpbin.org/ip")
if err != nil {
return "", fmt.Errorf("请求失败: %w", err)
}
defer resp.Body.Close()
if resp.StatusCode != http.StatusOK {
return "", fmt.Errorf("HTTP 状态码异常: %d", resp.StatusCode)
}
body, err := io.ReadAll(resp.Body)
if err != nil {
return "", fmt.Errorf("读取响应体失败: %w", err)
}
var ipResp IPResponse
if err := json.Unmarshal(body, &ipResp); err != nil {
return "", fmt.Errorf("解析 JSON 失败: %w", err)
}
if ipResp.Origin == "" {
return "", fmt.Errorf("响应中无 IP 信息")
}
// 取第一个 IP（处理代理/X-Forwarded-For 情况）
ip := strings.TrimSpace(strings.Split(ipResp.Origin, ",")[0])
return ip, nil
}