mirror of https://github.com/Xhofe/alist
fix(pikpak): webdav upload issue (#7050)
YangXu
GitHub
parent
489b28bdf7
commit
ef5e192c3b
|
|
@ -4,24 +4,18 @@ import (
|
|||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"github.com/alist-org/alist/v3/internal/op"
|
||||
"golang.org/x/oauth2"
|
||||
"io"
|
||||
"net/http"
|
||||
"strconv"
|
||||
"strings"
|
||||
|
||||
"github.com/alist-org/alist/v3/drivers/base"
|
||||
"github.com/alist-org/alist/v3/internal/driver"
|
||||
"github.com/alist-org/alist/v3/internal/model"
|
||||
"github.com/alist-org/alist/v3/internal/op"
|
||||
"github.com/alist-org/alist/v3/pkg/utils"
|
||||
hash_extend "github.com/alist-org/alist/v3/pkg/utils/hash"
|
||||
"github.com/aws/aws-sdk-go/aws"
|
||||
"github.com/aws/aws-sdk-go/aws/credentials"
|
||||
"github.com/aws/aws-sdk-go/aws/session"
|
||||
"github.com/aws/aws-sdk-go/service/s3/s3manager"
|
||||
"github.com/go-resty/resty/v2"
|
||||
log "github.com/sirupsen/logrus"
|
||||
"golang.org/x/oauth2"
|
||||
"net/http"
|
||||
"strconv"
|
||||
"strings"
|
||||
)
|
||||
|
||||
type PikPak struct {
|
||||
|
|
@ -123,10 +117,16 @@ func (d *PikPak) Init(ctx context.Context) (err error) {
|
|||
d.AccessToken = token.AccessToken
|
||||
|
||||
// 获取CaptchaToken
|
||||
err = d.RefreshCaptchaTokenAtLogin(GetAction(http.MethodGet, "https://api-drive.mypikpak.com/drive/v1/files"), d.Common.UserID)
|
||||
err = d.RefreshCaptchaTokenAtLogin(GetAction(http.MethodGet, "https://api-drive.mypikpak.com/drive/v1/files"), d.Username)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
// 获取用户ID
|
||||
userID := token.Extra("sub").(string)
|
||||
if userID != "" {
|
||||
d.Common.SetUserID(userID)
|
||||
}
|
||||
// 更新UserAgent
|
||||
if d.Platform == "android" {
|
||||
d.Common.UserAgent = BuildCustomUserAgent(utils.GetMD5EncodeStr(d.Username+d.Password), AndroidClientID, AndroidPackageName, AndroidSdkVersion, AndroidClientVersion, AndroidPackageName, d.Common.UserID)
|
||||
|
|
@ -271,27 +271,17 @@ func (d *PikPak) Put(ctx context.Context, dstDir model.Obj, stream model.FileStr
|
|||
}
|
||||
|
||||
params := resp.Resumable.Params
|
||||
endpoint := strings.Join(strings.Split(params.Endpoint, ".")[1:], ".")
|
||||
cfg := &aws.Config{
|
||||
Credentials: credentials.NewStaticCredentials(params.AccessKeyID, params.AccessKeySecret, params.SecurityToken),
|
||||
Region: aws.String("pikpak"),
|
||||
Endpoint: &endpoint,
|
||||
//endpoint := strings.Join(strings.Split(params.Endpoint, ".")[1:], ".")
|
||||
// web 端上传 返回的endpoint 为 `mypikpak.com` | android 端上传 返回的endpoint 为 `vip-lixian-07.mypikpak.com`·
|
||||
if d.Addition.Platform == "android" {
|
||||
params.Endpoint = "mypikpak.com"
|
||||
}
|
||||
ss, err := session.NewSession(cfg)
|
||||
if err != nil {
|
||||
return err
|
||||
|
||||
if stream.GetSize() <= 10*utils.MB { // 文件大小 小于10MB，改用普通模式上传
|
||||
return d.UploadByOSS(¶ms, stream, up)
|
||||
}
|
||||
uploader := s3manager.NewUploader(ss)
|
||||
if stream.GetSize() > s3manager.MaxUploadParts*s3manager.DefaultUploadPartSize {
|
||||
uploader.PartSize = stream.GetSize() / (s3manager.MaxUploadParts - 1)
|
||||
}
|
||||
input := &s3manager.UploadInput{
|
||||
Bucket: ¶ms.Bucket,
|
||||
Key: ¶ms.Key,
|
||||
Body: io.TeeReader(stream, driver.NewProgress(stream.GetSize(), up)),
|
||||
}
|
||||
_, err = uploader.UploadWithContext(ctx, input)
|
||||
return err
|
||||
// 分片上传
|
||||
return d.UploadByMultipart(¶ms, stream.GetSize(), stream, up)
|
||||
}
|
||||
|
||||
// 离线下载文件
|
||||
|
|
|
|||
|
|
@ -80,22 +80,24 @@ type UploadTaskData struct {
|
|||
UploadType string `json:"upload_type"`
|
||||
//UPLOAD_TYPE_RESUMABLE
|
||||
Resumable *struct {
|
||||
Kind string `json:"kind"`
|
||||
Params struct {
|
||||
AccessKeyID string `json:"access_key_id"`
|
||||
AccessKeySecret string `json:"access_key_secret"`
|
||||
Bucket string `json:"bucket"`
|
||||
Endpoint string `json:"endpoint"`
|
||||
Expiration time.Time `json:"expiration"`
|
||||
Key string `json:"key"`
|
||||
SecurityToken string `json:"security_token"`
|
||||
} `json:"params"`
|
||||
Provider string `json:"provider"`
|
||||
Kind string `json:"kind"`
|
||||
Params S3Params `json:"params"`
|
||||
Provider string `json:"provider"`
|
||||
} `json:"resumable"`
|
||||
|
||||
File File `json:"file"`
|
||||
}
|
||||
|
||||
type S3Params struct {
|
||||
AccessKeyID string `json:"access_key_id"`
|
||||
AccessKeySecret string `json:"access_key_secret"`
|
||||
Bucket string `json:"bucket"`
|
||||
Endpoint string `json:"endpoint"`
|
||||
Expiration time.Time `json:"expiration"`
|
||||
Key string `json:"key"`
|
||||
SecurityToken string `json:"security_token"`
|
||||
}
|
||||
|
||||
// 添加离线下载响应
|
||||
type OfflineDownloadResp struct {
|
||||
File *string `json:"file"`
|
||||
|
|
|
|||
|
|
@ -1,17 +1,24 @@
|
|||
package pikpak
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"crypto/md5"
|
||||
"crypto/sha1"
|
||||
"encoding/hex"
|
||||
"errors"
|
||||
"fmt"
|
||||
"github.com/alist-org/alist/v3/internal/driver"
|
||||
"github.com/alist-org/alist/v3/internal/model"
|
||||
"github.com/alist-org/alist/v3/internal/op"
|
||||
"github.com/alist-org/alist/v3/pkg/utils"
|
||||
"github.com/aliyun/aliyun-oss-go-sdk/oss"
|
||||
jsoniter "github.com/json-iterator/go"
|
||||
"github.com/pkg/errors"
|
||||
"io"
|
||||
"net/http"
|
||||
"path/filepath"
|
||||
"regexp"
|
||||
"strings"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"github.com/alist-org/alist/v3/drivers/base"
|
||||
|
|
@ -56,6 +63,12 @@ var WebAlgorithms = []string{
|
|||
"NhXXU9rg4XXdzo7u5o",
|
||||
}
|
||||
|
||||
const (
|
||||
OSSUserAgent = "aliyun-sdk-android/2.9.13(Linux/Android 14/M2004j7ac;UKQ1.231108.001)"
|
||||
OssSecurityTokenHeaderName = "X-OSS-Security-Token"
|
||||
ThreadsNum = 10
|
||||
)
|
||||
|
||||
const (
|
||||
AndroidClientID = "YNxT9w7GMdWvEOKa"
|
||||
AndroidClientSecret = "dbw2OtmVEeuUvIptb1Coyg"
|
||||
|
|
@ -393,3 +406,237 @@ func (d *PikPak) refreshCaptchaToken(action string, metas map[string]string) err
|
|||
d.Common.SetCaptchaToken(resp.CaptchaToken)
|
||||
return nil
|
||||
}
|
||||
|
||||
func (d *PikPak) UploadByOSS(params *S3Params, stream model.FileStreamer, up driver.UpdateProgress) error {
|
||||
ossClient, err := oss.New(params.Endpoint, params.AccessKeyID, params.AccessKeySecret)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
bucket, err := ossClient.Bucket(params.Bucket)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
err = bucket.PutObject(params.Key, stream, OssOption(params)...)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (d *PikPak) UploadByMultipart(params *S3Params, fileSize int64, stream model.FileStreamer, up driver.UpdateProgress) error {
|
||||
var (
|
||||
chunks []oss.FileChunk
|
||||
parts []oss.UploadPart
|
||||
imur oss.InitiateMultipartUploadResult
|
||||
ossClient *oss.Client
|
||||
bucket *oss.Bucket
|
||||
err error
|
||||
)
|
||||
|
||||
tmpF, err := stream.CacheFullInTempFile()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
if ossClient, err = oss.New(params.Endpoint, params.AccessKeyID, params.AccessKeySecret); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
if bucket, err = ossClient.Bucket(params.Bucket); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
ticker := time.NewTicker(time.Hour * 12)
|
||||
defer ticker.Stop()
|
||||
// 设置超时
|
||||
timeout := time.NewTimer(time.Hour * 24)
|
||||
|
||||
if chunks, err = SplitFile(fileSize); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
if imur, err = bucket.InitiateMultipartUpload(params.Key,
|
||||
oss.SetHeader(OssSecurityTokenHeaderName, params.SecurityToken),
|
||||
oss.UserAgentHeader(OSSUserAgent),
|
||||
); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
wg := sync.WaitGroup{}
|
||||
wg.Add(len(chunks))
|
||||
|
||||
chunksCh := make(chan oss.FileChunk)
|
||||
errCh := make(chan error)
|
||||
UploadedPartsCh := make(chan oss.UploadPart)
|
||||
quit := make(chan struct{})
|
||||
|
||||
// producer
|
||||
go chunksProducer(chunksCh, chunks)
|
||||
go func() {
|
||||
wg.Wait()
|
||||
quit <- struct{}{}
|
||||
}()
|
||||
|
||||
// consumers
|
||||
for i := 0; i < ThreadsNum; i++ {
|
||||
go func(threadId int) {
|
||||
defer func() {
|
||||
if r := recover(); r != nil {
|
||||
errCh <- fmt.Errorf("recovered in %v", r)
|
||||
}
|
||||
}()
|
||||
for chunk := range chunksCh {
|
||||
var part oss.UploadPart // 出现错误就继续尝试，共尝试3次
|
||||
for retry := 0; retry < 3; retry++ {
|
||||
select {
|
||||
case <-ticker.C:
|
||||
errCh <- errors.Wrap(err, "ossToken 过期")
|
||||
default:
|
||||
}
|
||||
|
||||
buf := make([]byte, chunk.Size)
|
||||
if _, err = tmpF.ReadAt(buf, chunk.Offset); err != nil && !errors.Is(err, io.EOF) {
|
||||
continue
|
||||
}
|
||||
|
||||
b := bytes.NewBuffer(buf)
|
||||
if part, err = bucket.UploadPart(imur, b, chunk.Size, chunk.Number, OssOption(params)...); err == nil {
|
||||
break
|
||||
}
|
||||
}
|
||||
if err != nil {
|
||||
errCh <- errors.Wrap(err, fmt.Sprintf("上传 %s 的第%d个分片时出现错误：%v", stream.GetName(), chunk.Number, err))
|
||||
}
|
||||
UploadedPartsCh <- part
|
||||
}
|
||||
}(i)
|
||||
}
|
||||
|
||||
go func() {
|
||||
for part := range UploadedPartsCh {
|
||||
parts = append(parts, part)
|
||||
wg.Done()
|
||||
}
|
||||
}()
|
||||
LOOP:
|
||||
for {
|
||||
select {
|
||||
case <-ticker.C:
|
||||
// ossToken 过期
|
||||
return err
|
||||
case <-quit:
|
||||
break LOOP
|
||||
case <-errCh:
|
||||
return err
|
||||
case <-timeout.C:
|
||||
return fmt.Errorf("time out")
|
||||
}
|
||||
}
|
||||
|
||||
// EOF错误是xml的Unmarshal导致的，响应其实是json格式，所以实际上上传是成功的
|
||||
if _, err = bucket.CompleteMultipartUpload(imur, parts, OssOption(params)...); err != nil && !errors.Is(err, io.EOF) {
|
||||
// 当文件名含有 &< 这两个字符之一时响应的xml解析会出现错误，实际上上传是成功的
|
||||
if filename := filepath.Base(stream.GetName()); !strings.ContainsAny(filename, "&<") {
|
||||
return err
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func chunksProducer(ch chan oss.FileChunk, chunks []oss.FileChunk) {
|
||||
for _, chunk := range chunks {
|
||||
ch <- chunk
|
||||
}
|
||||
}
|
||||
|
||||
func SplitFile(fileSize int64) (chunks []oss.FileChunk, err error) {
|
||||
for i := int64(1); i < 10; i++ {
|
||||
if fileSize < i*utils.GB { // 文件大小小于iGB时分为i*100片
|
||||
if chunks, err = SplitFileByPartNum(fileSize, int(i*100)); err != nil {
|
||||
return
|
||||
}
|
||||
break
|
||||
}
|
||||
}
|
||||
if fileSize > 9*utils.GB { // 文件大小大于9GB时分为1000片
|
||||
if chunks, err = SplitFileByPartNum(fileSize, 1000); err != nil {
|
||||
return
|
||||
}
|
||||
}
|
||||
// 单个分片大小不能小于1MB
|
||||
if chunks[0].Size < 1*utils.MB {
|
||||
if chunks, err = SplitFileByPartSize(fileSize, 1*utils.MB); err != nil {
|
||||
return
|
||||
}
|
||||
}
|
||||
return
|
||||
}
|
||||
|
||||
// SplitFileByPartNum splits big file into parts by the num of parts.
|
||||
// Split the file with specified parts count, returns the split result when error is nil.
|
||||
func SplitFileByPartNum(fileSize int64, chunkNum int) ([]oss.FileChunk, error) {
|
||||
if chunkNum <= 0 || chunkNum > 10000 {
|
||||
return nil, errors.New("chunkNum invalid")
|
||||
}
|
||||
|
||||
if int64(chunkNum) > fileSize {
|
||||
return nil, errors.New("oss: chunkNum invalid")
|
||||
}
|
||||
|
||||
var chunks []oss.FileChunk
|
||||
chunk := oss.FileChunk{}
|
||||
chunkN := (int64)(chunkNum)
|
||||
for i := int64(0); i < chunkN; i++ {
|
||||
chunk.Number = int(i + 1)
|
||||
chunk.Offset = i * (fileSize / chunkN)
|
||||
if i == chunkN-1 {
|
||||
chunk.Size = fileSize/chunkN + fileSize%chunkN
|
||||
} else {
|
||||
chunk.Size = fileSize / chunkN
|
||||
}
|
||||
chunks = append(chunks, chunk)
|
||||
}
|
||||
|
||||
return chunks, nil
|
||||
}
|
||||
|
||||
// SplitFileByPartSize splits big file into parts by the size of parts.
|
||||
// Splits the file by the part size. Returns the FileChunk when error is nil.
|
||||
func SplitFileByPartSize(fileSize int64, chunkSize int64) ([]oss.FileChunk, error) {
|
||||
if chunkSize <= 0 {
|
||||
return nil, errors.New("chunkSize invalid")
|
||||
}
|
||||
|
||||
chunkN := fileSize / chunkSize
|
||||
if chunkN >= 10000 {
|
||||
return nil, errors.New("Too many parts, please increase part size")
|
||||
}
|
||||
|
||||
var chunks []oss.FileChunk
|
||||
chunk := oss.FileChunk{}
|
||||
for i := int64(0); i < chunkN; i++ {
|
||||
chunk.Number = int(i + 1)
|
||||
chunk.Offset = i * chunkSize
|
||||
chunk.Size = chunkSize
|
||||
chunks = append(chunks, chunk)
|
||||
}
|
||||
|
||||
if fileSize%chunkSize > 0 {
|
||||
chunk.Number = len(chunks) + 1
|
||||
chunk.Offset = int64(len(chunks)) * chunkSize
|
||||
chunk.Size = fileSize % chunkSize
|
||||
chunks = append(chunks, chunk)
|
||||
}
|
||||
|
||||
return chunks, nil
|
||||
}
|
||||
|
||||
// OssOption get options
|
||||
func OssOption(params *S3Params) []oss.Option {
|
||||
options := []oss.Option{
|
||||
oss.SetHeader(OssSecurityTokenHeaderName, params.SecurityToken),
|
||||
oss.UserAgentHeader(OSSUserAgent),
|
||||
}
|
||||
return options
|
||||
}
|
||||
|
|
|
|||
Loading…
Reference in New Issue