Skip to content

Commit 08dae4f

Browse files
xrgzsxrgzs
authored
feat(123_open): update upload api v2 (#976)
1 parent 9ac0484 commit 08dae4f

File tree

4 files changed

+133
-106
lines changed

4 files changed

+133
-106
lines changed

‎drivers/123_open/driver.go‎

Lines changed: 43 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -2,7 +2,9 @@ package _123_open
22

33
import (
44
"context"
5+
"fmt"
56
"strconv"
7+
"time"
68

79
"github.com/OpenListTeam/OpenList/v4/internal/driver"
810
"github.com/OpenListTeam/OpenList/v4/internal/errs"
@@ -95,6 +97,22 @@ func (d *Open123) Rename(ctx context.Context, srcObj model.Obj, newName string)
9597
}
9698

9799
func (d *Open123) Copy(ctx context.Context, srcObj, dstDir model.Obj) error {
100+
// 尝试使用上传+MD5秒传功能实现复制
101+
// 1. 创建文件
102+
// parentFileID 父目录id，上传到根目录时填写 0
103+
parentFileId, err := strconv.ParseInt(dstDir.GetID(), 10, 64)
104+
if err != nil {
105+
return fmt.Errorf("parse parentFileID error: %v", err)
106+
}
107+
etag := srcObj.(File).Etag
108+
createResp, err := d.create(parentFileId, srcObj.GetName(), etag, srcObj.GetSize(), 2, false)
109+
if err != nil {
110+
return err
111+
}
112+
// 是否秒传
113+
if createResp.Data.Reuse {
114+
return nil
115+
}
98116
return errs.NotSupport
99117
}
100118

@@ -105,9 +123,14 @@ func (d *Open123) Remove(ctx context.Context, obj model.Obj) error {
105123
}
106124

107125
func (d *Open123) Put(ctx context.Context, dstDir model.Obj, file model.FileStreamer, up driver.UpdateProgress) error {
126+
// 1. 创建文件
127+
// parentFileID 父目录id，上传到根目录时填写 0
108128
parentFileId, err := strconv.ParseInt(dstDir.GetID(), 10, 64)
129+
if err != nil {
130+
return fmt.Errorf("parse parentFileID error: %v", err)
131+
}
132+
// etag 文件md5
109133
etag := file.GetHash().GetHash(utils.MD5)
110-
111134
if len(etag) < utils.MD5.Width {
112135
cacheFileProgress := model.UpdateProgressWithRange(up, 0, 50)
113136
up = model.UpdateProgressWithRange(up, 50, 100)
@@ -120,11 +143,29 @@ func (d *Open123) Put(ctx context.Context, dstDir model.Obj, file model.FileStre
120143
if err != nil {
121144
return err
122145
}
146+
// 是否秒传
123147
if createResp.Data.Reuse {
124148
return nil
125149
}
126150

127-
return d.Upload(ctx, file, createResp, up)
151+
// 2. 上传分片
152+
err = d.Upload(ctx, file, createResp, up)
153+
if err != nil {
154+
return err
155+
}
156+
157+
// 3. 上传完毕
158+
for range 60 {
159+
uploadCompleteResp, err := d.complete(createResp.Data.PreuploadID)
160+
// 返回错误代码未知，如：20103，文档也没有具体说
161+
if err == nil && uploadCompleteResp.Data.Completed && uploadCompleteResp.Data.FileID != 0 {
162+
break
163+
}
164+
// 若接口返回的completed为 false 时，则需间隔1秒继续轮询此接口，获取上传最终结果。
165+
time.Sleep(time.Second)
166+
}
167+
up(100)
168+
return nil
128169
}
129170

130171
var _ driver.Driver = (*Open123)(nil)

‎drivers/123_open/types.go‎

Lines changed: 7 additions & 36 deletions
Original file line numberDiff line numberDiff line change
@@ -154,52 +154,23 @@ type DownloadInfoResp struct {
154154
} `json:"data"`
155155
}
156156

157+
// 创建文件V2返回
157158
type UploadCreateResp struct {
158159
BaseResp
159160
Data struct {
160-
FileID int64 `json:"fileID"`
161-
PreuploadID string `json:"preuploadID"`
162-
Reuse bool `json:"reuse"`
163-
SliceSize int64 `json:"sliceSize"`
161+
FileID int64 `json:"fileID"`
162+
PreuploadID string `json:"preuploadID"`
163+
Reuse bool `json:"reuse"`
164+
SliceSize int64 `json:"sliceSize"`
165+
Servers []string `json:"servers"`
164166
} `json:"data"`
165167
}
166168

167-
type UploadUrlResp struct {
168-
BaseResp
169-
Data struct {
170-
PresignedURL string `json:"presignedURL"`
171-
}
172-
}
173-
169+
// 上传完毕V2返回
174170
type UploadCompleteResp struct {
175171
BaseResp
176172
Data struct {
177-
Async bool `json:"async"`
178173
Completed bool `json:"completed"`
179174
FileID int64 `json:"fileID"`
180175
} `json:"data"`
181176
}
182-
183-
type UploadAsyncResp struct {
184-
BaseResp
185-
Data struct {
186-
Completed bool `json:"completed"`
187-
FileID int64 `json:"fileID"`
188-
} `json:"data"`
189-
}
190-
191-
type UploadResp struct {
192-
BaseResp
193-
Data struct {
194-
AccessKeyId string `json:"AccessKeyId"`
195-
Bucket string `json:"Bucket"`
196-
Key string `json:"Key"`
197-
SecretAccessKey string `json:"SecretAccessKey"`
198-
SessionToken string `json:"SessionToken"`
199-
FileId int64 `json:"FileId"`
200-
Reuse bool `json:"Reuse"`
201-
EndPoint string `json:"EndPoint"`
202-
StorageNode string `json:"StorageNode"`
203-
UploadId string `json:"UploadId"`
204-
} `json:"data"`
205-
}

‎drivers/123_open/upload.go‎

Lines changed: 80 additions & 63 deletions
Original file line numberDiff line numberDiff line change
@@ -1,9 +1,14 @@
11
package _123_open
22

33
import (
4+
"bytes"
45
"context"
6+
"encoding/json"
7+
"fmt"
58
"io"
9+
"mime/multipart"
610
"net/http"
11+
"strconv"
712
"strings"
813
"time"
914

@@ -17,6 +22,7 @@ import (
1722
"github.com/go-resty/resty/v2"
1823
)
1924

25+
// 创建文件 V2
2026
func (d *Open123) create(parentFileID int64, filename string, etag string, size int64, duplicate int, containDir bool) (*UploadCreateResp, error) {
2127
var resp UploadCreateResp
2228
_, err := d.Request(UploadCreate, http.MethodPost, func(req *resty.Request) {
@@ -35,48 +41,9 @@ func (d *Open123) create(parentFileID int64, filename string, etag string, size
3541
return &resp, nil
3642
}
3743

38-
func (d *Open123) url(preuploadID string, sliceNo int64) (string, error) {
39-
// get upload url
40-
var resp UploadUrlResp
41-
_, err := d.Request(UploadUrl, http.MethodPost, func(req *resty.Request) {
42-
req.SetBody(base.Json{
43-
"preuploadId": preuploadID,
44-
"sliceNo": sliceNo,
45-
})
46-
}, &resp)
47-
if err != nil {
48-
return "", err
49-
}
50-
return resp.Data.PresignedURL, nil
51-
}
52-
53-
func (d *Open123) complete(preuploadID string) (*UploadCompleteResp, error) {
54-
var resp UploadCompleteResp
55-
_, err := d.Request(UploadComplete, http.MethodPost, func(req *resty.Request) {
56-
req.SetBody(base.Json{
57-
"preuploadID": preuploadID,
58-
})
59-
}, &resp)
60-
if err != nil {
61-
return nil, err
62-
}
63-
return &resp, nil
64-
}
65-
66-
func (d *Open123) async(preuploadID string) (*UploadAsyncResp, error) {
67-
var resp UploadAsyncResp
68-
_, err := d.Request(UploadAsync, http.MethodPost, func(req *resty.Request) {
69-
req.SetBody(base.Json{
70-
"preuploadID": preuploadID,
71-
})
72-
}, &resp)
73-
if err != nil {
74-
return nil, err
75-
}
76-
return &resp, nil
77-
}
78-
44+
// 上传分片 V2
7945
func (d *Open123) Upload(ctx context.Context, file model.FileStreamer, createResp *UploadCreateResp, up driver.UpdateProgress) error {
46+
uploadDomain := createResp.Data.Servers[0]
8047
size := file.GetSize()
8148
chunkSize := createResp.Data.SliceSize
8249
uploadNums := (size + chunkSize - 1) / chunkSize
@@ -90,7 +57,7 @@ func (d *Open123) Upload(ctx context.Context, file model.FileStreamer, createRes
9057
if err != nil {
9158
return err
9259
}
93-
for partIndex := int64(0); partIndex < uploadNums; partIndex++ {
60+
for partIndex := range uploadNums {
9461
if utils.IsCanceled(uploadCtx) {
9562
break
9663
}
@@ -100,36 +67,90 @@ func (d *Open123) Upload(ctx context.Context, file model.FileStreamer, createRes
10067
size := min(chunkSize, size-offset)
10168
var reader *stream.SectionReader
10269
var rateLimitedRd io.Reader
70+
sliceMD5 := ""
10371
threadG.GoWithLifecycle(errgroup.Lifecycle{
10472
Before: func(ctx context.Context) error {
10573
if reader == nil {
10674
var err error
75+
// 每个分片一个reader
10776
reader, err = ss.GetSectionReader(offset, size)
10877
if err != nil {
10978
return err
11079
}
80+
// 计算当前分片的MD5
81+
sliceMD5, err = utils.HashReader(utils.MD5, reader)
82+
if err != nil {
83+
return err
84+
}
11185
rateLimitedRd = driver.NewLimitedUploadStream(ctx, reader)
11286
}
11387
return nil
11488
},
11589
Do: func(ctx context.Context) error {
90+
// 重置分片reader位置，因为HashReader、上一次失败已经读取到分片EOF
11691
reader.Seek(0, io.SeekStart)
117-
uploadPartUrl, err := d.url(createResp.Data.PreuploadID, partNumber)
92+
93+
// 创建表单数据
94+
var b bytes.Buffer
95+
w := multipart.NewWriter(&b)
96+
// 添加表单字段
97+
err = w.WriteField("preuploadID", createResp.Data.PreuploadID)
98+
if err != nil {
99+
return err
100+
}
101+
err = w.WriteField("sliceNo", strconv.FormatInt(partNumber, 10))
102+
if err != nil {
103+
return err
104+
}
105+
err = w.WriteField("sliceMD5", sliceMD5)
106+
if err != nil {
107+
return err
108+
}
109+
// 写入文件内容
110+
fw, err := w.CreateFormFile("slice", fmt.Sprintf("%s.part%d", file.GetName(), partNumber))
111+
if err != nil {
112+
return err
113+
}
114+
_, err = utils.CopyWithBuffer(fw, rateLimitedRd)
115+
if err != nil {
116+
return err
117+
}
118+
err = w.Close()
118119
if err != nil {
119120
return err
120121
}
121122

122-
req, err := http.NewRequestWithContext(ctx, http.MethodPut, uploadPartUrl, rateLimitedRd)
123+
// 创建请求并设置header
124+
req, err := http.NewRequestWithContext(ctx, http.MethodPost, uploadDomain+"/upload/v2/file/slice", &b)
123125
if err != nil {
124126
return err
125127
}
126-
req.ContentLength = size
128+
129+
// 设置请求头
130+
req.Header.Add("Authorization", "Bearer "+d.AccessToken)
131+
req.Header.Add("Content-Type", w.FormDataContentType())
132+
req.Header.Add("Platform", "open_platform")
127133

128134
res, err := base.HttpClient.Do(req)
129135
if err != nil {
130136
return err
131137
}
132-
_ = res.Body.Close()
138+
defer res.Body.Close()
139+
if res.StatusCode != 200 {
140+
return fmt.Errorf("slice %d upload failed, status code: %d", partNumber, res.StatusCode)
141+
}
142+
var resp BaseResp
143+
respBody, err := io.ReadAll(res.Body)
144+
if err != nil {
145+
return err
146+
}
147+
err = json.Unmarshal(respBody, &resp)
148+
if err != nil {
149+
return err
150+
}
151+
if resp.Code != 0 {
152+
return fmt.Errorf("slice %d upload failed: %s", partNumber, resp.Message)
153+
}
133154

134155
progress := 10.0 + 85.0*float64(threadG.Success())/float64(uploadNums)
135156
up(progress)
@@ -145,23 +166,19 @@ func (d *Open123) Upload(ctx context.Context, file model.FileStreamer, createRes
145166
return err
146167
}
147168

148-
uploadCompleteResp, err := d.complete(createResp.Data.PreuploadID)
149-
if err != nil {
150-
return err
151-
}
152-
if uploadCompleteResp.Data.Async == false || uploadCompleteResp.Data.Completed {
153-
return nil
154-
}
169+
return nil
170+
}
155171

156-
for {
157-
uploadAsyncResp, err := d.async(createResp.Data.PreuploadID)
158-
if err != nil {
159-
return err
160-
}
161-
if uploadAsyncResp.Data.Completed {
162-
break
163-
}
172+
// 上传完毕
173+
func (d *Open123) complete(preuploadID string) (*UploadCompleteResp, error) {
174+
var resp UploadCompleteResp
175+
_, err := d.Request(UploadComplete, http.MethodPost, func(req *resty.Request) {
176+
req.SetBody(base.Json{
177+
"preuploadID": preuploadID,
178+
})
179+
}, &resp)
180+
if err != nil {
181+
return nil, err
164182
}
165-
up(100)
166-
return nil
183+
return &resp, nil
167184
}

‎drivers/123_open/util.go‎

Lines changed: 3 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -19,16 +19,14 @@ var ( //不同情况下获取的AccessTokenQPS限制不同 如下模块化易于
1919
AccessToken = InitApiInfo(Api+"/api/v1/access_token", 1)
2020
RefreshToken = InitApiInfo(Api+"/api/v1/oauth2/access_token", 1)
2121
UserInfo = InitApiInfo(Api+"/api/v1/user/info", 1)
22-
FileList = InitApiInfo(Api+"/api/v2/file/list", 4)
22+
FileList = InitApiInfo(Api+"/api/v2/file/list", 3)
2323
DownloadInfo = InitApiInfo(Api+"/api/v1/file/download_info", 0)
2424
Mkdir = InitApiInfo(Api+"/upload/v1/file/mkdir", 2)
2525
Move = InitApiInfo(Api+"/api/v1/file/move", 1)
2626
Rename = InitApiInfo(Api+"/api/v1/file/name", 1)
2727
Trash = InitApiInfo(Api+"/api/v1/file/trash", 2)
28-
UploadCreate = InitApiInfo(Api+"/upload/v1/file/create", 2)
29-
UploadUrl = InitApiInfo(Api+"/upload/v1/file/get_upload_url", 0)
30-
UploadComplete = InitApiInfo(Api+"/upload/v1/file/upload_complete", 0)
31-
UploadAsync = InitApiInfo(Api+"/upload/v1/file/upload_async_result", 1)
28+
UploadCreate = InitApiInfo(Api+"/upload/v2/file/create", 2)
29+
UploadComplete = InitApiInfo(Api+"/upload/v2/file/upload_complete", 0)
3230
)
3331

3432
func (d *Open123) Request(apiInfo *ApiInfo, method string, callback base.ReqCallback, resp interface{}) ([]byte, error) {

0 commit comments

Comments
 (0)