Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

feat(115): enhance cache #7479

Merged
merged 1 commit into from
Nov 8, 2024
+146 −40
Merged

feat(115): enhance cache #7479

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
100 changes: 74 additions & 26 deletions drivers/115/driver.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -79,28 +79,60 @@ func (d *Pan115) Link(ctx context.Context, file model.Obj, args model.LinkArgs)
return link, nil
}

func (d *Pan115) MakeDir(ctx context.Context, parentDir model.Obj, dirName string) error {
func (d *Pan115) MakeDir(ctx context.Context, parentDir model.Obj, dirName string) (model.Obj, error) {
if err := d.WaitLimit(ctx); err != nil {
return err
return nil, err
}
if _, err := d.client.Mkdir(parentDir.GetID(), dirName); err != nil {
return err

result := driver115.MkdirResp{}
form := map[string]string{
"pid": parentDir.GetID(),
"cname": dirName,
}
return nil
req := d.client.NewRequest().
SetFormData(form).
SetResult(&result).
ForceContentType("application/json;charset=UTF-8")

resp, err := req.Post(driver115.ApiDirAdd)

err = driver115.CheckErr(err, &result, resp)
if err != nil {
return nil, err
}
f, err := d.getNewFile(result.FileID)
if err != nil {
return nil, nil
}
return f, nil
}

func (d *Pan115) Move(ctx context.Context, srcObj, dstDir model.Obj) error {
func (d *Pan115) Move(ctx context.Context, srcObj, dstDir model.Obj) (model.Obj, error) {
if err := d.WaitLimit(ctx); err != nil {
return err
return nil, err
}
return d.client.Move(dstDir.GetID(), srcObj.GetID())
if err := d.client.Move(dstDir.GetID(), srcObj.GetID()); err != nil {
return nil, err
}
f, err := d.getNewFile(srcObj.GetID())
if err != nil {
return nil, nil
}
return f, nil
}

func (d *Pan115) Rename(ctx context.Context, srcObj model.Obj, newName string) error {
func (d *Pan115) Rename(ctx context.Context, srcObj model.Obj, newName string) (model.Obj, error) {
if err := d.WaitLimit(ctx); err != nil {
return err
return nil, err
}
if err := d.client.Rename(srcObj.GetID(), newName); err != nil {
return nil, err
}
return d.client.Rename(srcObj.GetID(), newName)
f, err := d.getNewFile((srcObj.GetID()))
if err != nil {
return nil, nil
}
return f, nil
}

func (d *Pan115) Copy(ctx context.Context, srcObj, dstDir model.Obj) error {
Expand All @@ -117,9 +149,9 @@ func (d *Pan115) Remove(ctx context.Context, obj model.Obj) error {
return d.client.Delete(obj.GetID())
}

func (d *Pan115) Put(ctx context.Context, dstDir model.Obj, stream model.FileStreamer, up driver.UpdateProgress) error {
func (d *Pan115) Put(ctx context.Context, dstDir model.Obj, stream model.FileStreamer, up driver.UpdateProgress) (model.Obj, error) {
if err := d.WaitLimit(ctx); err != nil {
return err
return nil, err
}

var (
Expand All @@ -128,10 +160,10 @@ func (d *Pan115) Put(ctx context.Context, dstDir model.Obj, stream model.FileStr
)

if ok, err := d.client.UploadAvailable(); err != nil || !ok {
return err
return nil, err
}
if stream.GetSize() > d.client.UploadMetaInfo.SizeLimit {
return driver115.ErrUploadTooLarge
return nil, driver115.ErrUploadTooLarge
}
//if digest, err = d.client.GetDigestResult(stream); err != nil {
// return err
Expand All @@ -144,22 +176,22 @@ func (d *Pan115) Put(ctx context.Context, dstDir model.Obj, stream model.FileStr
}
reader, err := stream.RangeRead(http_range.Range{Start: 0, Length: hashSize})
if err != nil {
return err
return nil, err
}
preHash, err := utils.HashReader(utils.SHA1, reader)
if err != nil {
return err
return nil, err
}
preHash = strings.ToUpper(preHash)
fullHash := stream.GetHash().GetHash(utils.SHA1)
if len(fullHash) <= 0 {
tmpF, err := stream.CacheFullInTempFile()
if err != nil {
return err
return nil, err
}
fullHash, err = utils.HashFile(utils.SHA1, tmpF)
if err != nil {
return err
return nil, err
}
}
fullHash = strings.ToUpper(fullHash)
Expand All @@ -168,20 +200,36 @@ func (d *Pan115) Put(ctx context.Context, dstDir model.Obj, stream model.FileStr
// note that 115 add timeout for rapid-upload,
// and "sig invalid" err is thrown even when the hash is correct after timeout.
if fastInfo, err = d.rapidUpload(stream.GetSize(), stream.GetName(), dirID, preHash, fullHash, stream); err != nil {
return err
return nil, err
}
if matched, err := fastInfo.Ok(); err != nil {
return err
return nil, err
} else if matched {
return nil
f, err := d.getNewFileByPickCode(fastInfo.PickCode)
if err != nil {
return nil, nil
}
return f, nil
}

var uploadResult *UploadResult
// 闪传失败，上传
if stream.GetSize() <= utils.KB { // 文件大小小于1KB，改用普通模式上传
return d.client.UploadByOSS(&fastInfo.UploadOSSParams, stream, dirID)
if stream.GetSize() <= 10*utils.MB { // 文件大小小于10MB，改用普通模式上传
if uploadResult, err = d.UploadByOSS(&fastInfo.UploadOSSParams, stream, dirID); err != nil {
return nil, err
}
} else {
// 分片上传
if uploadResult, err = d.UploadByMultipart(&fastInfo.UploadOSSParams, stream.GetSize(), stream, dirID); err != nil {
return nil, err
}
}

file, err := d.getNewFile(uploadResult.Data.FileID)
if err != nil {
return nil, nil
}
// 分片上传
return d.UploadByMultipart(&fastInfo.UploadOSSParams, stream.GetSize(), stream, dirID)
return file, nil
}

func (d *Pan115) OfflineList(ctx context.Context) ([]*driver115.OfflineTask, error) {
Expand Down
2 changes: 1 addition & 1 deletion drivers/115/meta.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -10,7 +10,7 @@ type Addition struct {
QRCodeToken string `json:"qrcode_token" type:"text" help:"one of QR code token and cookie required"`
QRCodeSource string `json:"qrcode_source" type:"select" options:"web,android,ios,tv,alipaymini,wechatmini,qandroid" default:"linux" help:"select the QR code device, default linux"`
PageSize int64 `json:"page_size" type:"number" default:"1000" help:"list api per page size of 115 driver"`
LimitRate float64 `json:"limit_rate" type:"number" default:"2" help:"limit all api request rate (1r/[limit_rate]s)"`
LimitRate float64 `json:"limit_rate" type:"number" default:"2" help:"limit all api request rate ([limit]r/1s)"`
driver.RootID
}

Expand Down
84 changes: 71 additions & 13 deletions drivers/115/util.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -74,6 +74,34 @@ func (d *Pan115) getFiles(fileId string) ([]FileObj, error) {
return res, nil
}

func (d *Pan115) getNewFile(fileId string) (*FileObj, error) {
file, err := d.client.GetFile(fileId)
if err != nil {
return nil, err
}
return &FileObj{*file}, nil
}

func (d *Pan115) getNewFileByPickCode(pickCode string) (*FileObj, error) {
result := driver115.GetFileInfoResponse{}
req := d.client.NewRequest().
SetQueryParam("pick_code", pickCode).
ForceContentType("application/json;charset=UTF-8").
SetResult(&result)
resp, err := req.Get(driver115.ApiFileInfo)
if err := driver115.CheckErr(err, &result, resp); err != nil {
return nil, err
}
if len(result.Files) == 0 {
return nil, errors.New("not get file info")
}
fileInfo := result.Files[0]

f := &FileObj{}
f.From(fileInfo)
return f, nil
}

func (d *Pan115) getUA() string {
return fmt.Sprintf("Mozilla/5.0 115Browser/%s", appVer)
}
Expand Down Expand Up @@ -244,8 +272,38 @@ func UploadDigestRange(stream model.FileStreamer, rangeSpec string) (result stri
return
}

// UploadByOSS use aliyun sdk to upload
func (c *Pan115) UploadByOSS(params *driver115.UploadOSSParams, r io.Reader, dirID string) (*UploadResult, error) {
ossToken, err := c.client.GetOSSToken()
if err != nil {
return nil, err
}
ossClient, err := oss.New(driver115.OSSEndpoint, ossToken.AccessKeyID, ossToken.AccessKeySecret)
if err != nil {
return nil, err
}
bucket, err := ossClient.Bucket(params.Bucket)
if err != nil {
return nil, err
}

var bodyBytes []byte
if err = bucket.PutObject(params.Object, r, append(
driver115.OssOption(params, ossToken),
oss.CallbackResult(&bodyBytes),
)...); err != nil {
return nil, err
}

var uploadResult UploadResult
if err = json.Unmarshal(bodyBytes, &uploadResult); err != nil {
return nil, err
}
return &uploadResult, uploadResult.Err(string(bodyBytes))
}

// UploadByMultipart upload by mutipart blocks
func (d *Pan115) UploadByMultipart(params *driver115.UploadOSSParams, fileSize int64, stream model.FileStreamer, dirID string, opts ...driver115.UploadMultipartOption) error {
func (d *Pan115) UploadByMultipart(params *driver115.UploadOSSParams, fileSize int64, stream model.FileStreamer, dirID string, opts ...driver115.UploadMultipartOption) (*UploadResult, error) {
var (
chunks []oss.FileChunk
parts []oss.UploadPart
Expand All @@ -259,7 +317,7 @@ func (d *Pan115) UploadByMultipart(params *driver115.UploadOSSParams, fileSize i

tmpF, err := stream.CacheFullInTempFile()
if err != nil {
return err
return nil, err
}

options := driver115.DefalutUploadMultipartOptions()
Expand All @@ -272,15 +330,15 @@ func (d *Pan115) UploadByMultipart(params *driver115.UploadOSSParams, fileSize i
options.ThreadsNum = 1

if ossToken, err = d.client.GetOSSToken(); err != nil {
return err
return nil, err
}

if ossClient, err = oss.New(driver115.OSSEndpoint, ossToken.AccessKeyID, ossToken.AccessKeySecret, oss.EnableMD5(true), oss.EnableCRC(true)); err != nil {
return err
return nil, err
}

if bucket, err = ossClient.Bucket(params.Bucket); err != nil {
return err
return nil, err
}

// ossToken一小时后就会失效，所以每50分钟重新获取一次
Expand All @@ -290,15 +348,15 @@ func (d *Pan115) UploadByMultipart(params *driver115.UploadOSSParams, fileSize i
timeout := time.NewTimer(options.Timeout)

if chunks, err = SplitFile(fileSize); err != nil {
return err
return nil, err
}

if imur, err = bucket.InitiateMultipartUpload(params.Object,
oss.SetHeader(driver115.OssSecurityTokenHeaderName, ossToken.SecurityToken),
oss.UserAgentHeader(driver115.OSSUserAgent),
oss.EnableSha1(), oss.Sequential(),
); err != nil {
return err
return nil, err
}

wg := sync.WaitGroup{}
Expand Down Expand Up @@ -364,14 +422,14 @@ LOOP:
case <-ticker.C:
// 到时重新获取ossToken
if ossToken, err = d.client.GetOSSToken(); err != nil {
return err
return nil, err
}
case <-quit:
break LOOP
case <-errCh:
return err
return nil, err
case <-timeout.C:
return fmt.Errorf("time out")
return nil, fmt.Errorf("time out")
}
}

Expand All @@ -381,14 +439,14 @@ LOOP:
driver115.OssOption(params, ossToken),
oss.CallbackResult(&bodyBytes),
)...); err != nil {
return err
return nil, err
}

var uploadResult UploadResult
if err = json.Unmarshal(bodyBytes, &uploadResult); err != nil {
return err
return nil, err
}
return uploadResult.Err(string(bodyBytes))
return &uploadResult, uploadResult.Err(string(bodyBytes))
}

func chunksProducer(ch chan oss.FileChunk, chunks []oss.FileChunk) {
Expand Down
Loading