Skip to content

Commit b87dcb8

Browse files
author
yang.zhao
committed
fix
1 parent 8d706dc commit b87dcb8

File tree

1 file changed

+161
-37
lines changed

1 file changed

+161
-37
lines changed

‎backend/baidu/baidu.go‎

Lines changed: 161 additions & 37 deletions
Original file line numberDiff line numberDiff line change
@@ -44,8 +44,8 @@ const (
4444
uriQuota = "/api/quota"
4545

4646
//
47-
chunkSize = 4 * 1024 * 1024 //分片大小4M
48-
rapidUploadThreshold = 256 * 1024 //秒传阈值256KB
47+
chunkSize = 4 * 1024 * 1024 // 分片大小锁定为 4M (官方黄金标准，普通用户与会员通用)
48+
rapidUploadThreshold = 256 * 1024 // 秒传阈值 256KB
4949
)
5050

5151
// Options defines the configuration for this backend
@@ -390,6 +390,7 @@ func (f *Fs) listDirFile(ctx context.Context, dir string, start, limit int) ([]F
390390
"web": {"1"},
391391
"folder": {"0"},
392392
"showempty": {"1"},
393+
"openapi": {"xpansdk"},
393394
},
394395
}
395396
resp := &FileListOut{}
@@ -615,17 +616,20 @@ func (f *Fs) Purge(ctx context.Context, dir string) error {
615616
// About gets quota information
616617
func (f *Fs) About(ctx context.Context) (usage *fs.Usage, err error) {
617618
opts := &rest.Opts{
618-
Method: "GET",
619-
RootURL: rootURL,
620-
Path: uriQuota,
621-
Parameters: map[string][]string{},
619+
Method: "GET",
620+
RootURL: rootURL,
621+
Path: uriQuota,
622+
Parameters: map[string][]string{
623+
"openapi": {"xpansdk"},
624+
},
622625
}
623626
resp := QuotaOut{}
624-
if err = f.call(ctx, opts, resp); err != nil {
627+
if err = f.call(ctx, opts, &resp); err != nil {
625628
return nil, err
626629
}
630+
free := resp.Total - resp.Used
627631
usage = &fs.Usage{
628-
Free: &resp.Free,
632+
Free: &free, // 百度接口返回 Free 常为 0，在此手动计算以满足 rclone 展示
629633
Total: &resp.Total,
630634
Used: &resp.Used,
631635
}
@@ -761,52 +765,167 @@ func (o *Object) upload(ctx context.Context, in io.Reader, size int64) error {
761765
}
762766
}
763767

764-
// 2. 对于 2GB 以下文件，使用单次上传 (Simple Upload) 绕过分片上传 Bug (31064)
765-
if size <= 2*1024*1024*1024 {
768+
// 2. 对于 4MB 以下文件，使用单次上传 (Simple Upload)
769+
// 根据官方文档，4MB 以上必须分片，PCS 的简单上传通道在超限时会不稳定 (易中断)
770+
if size <= 4*1024*1024 && size >= 0 {
766771
return o.simpleUpload(ctx, in, size)
767772
}
768773

769-
// 3. 超过 2GB 或流式大数据量时，使用 XPAN 三阶段分片上传
770-
// 注意：分片上传目前在部分账户/应用下可能报 31064 错误
774+
// 3. 超过 4MB 使用 XPAN 分片上传 (superfile2)
775+
// 3. 超过 4MB 使用 XPAN 分片上传 (superfile2)
776+
// 注意：由于 rclone 内部可能使用 AsyncReader 包装，无法可靠检测 Seek 支持
777+
// 因此统一使用 Disk Spooling 模式（写盘缓存）处理大文件，避免内存 OOM
778+
771779
var md5s []string
772-
var chunks [][]byte
773-
buf := make([]byte, chunkSize)
780+
var uploadSource io.ReaderAt
781+
// 如果是 Seeker（本地文件），可以直接使用
782+
if seeker, ok := in.(io.ReadSeeker); ok {
783+
// 再次尝试 Seek Detect，排除 AsyncReader 的假实现
784+
if _, err := seeker.Seek(0, io.SeekCurrent); err == nil {
785+
fs.Debugf(o, "检测到本地文件流，开启双读流式上传模式")
786+
uploadSource = seeker.(io.ReaderAt) // os.File implements ReaderAt
787+
788+
// 第一遍：计算所有分片的 MD5
789+
buf := make([]byte, chunkSize)
790+
for {
791+
n, err := io.ReadFull(seeker, buf)
792+
if n > 0 {
793+
h := md5.New()
794+
h.Write(buf[:n])
795+
md5s = append(md5s, hex.EncodeToString(h.Sum(nil)))
796+
}
797+
if err == io.EOF || err == io.ErrUnexpectedEOF {
798+
break
799+
}
800+
if err != nil {
801+
return fmt.Errorf("扫描文件 MD5 失败: %w", err)
802+
}
803+
}
774804

775-
for {
776-
n, err := io.ReadFull(in, buf)
777-
if n > 0 {
778-
chunkCopy := make([]byte, n)
779-
copy(chunkCopy, buf[:n])
780-
chunks = append(chunks, chunkCopy)
781-
782-
h := md5.New()
783-
h.Write(chunkCopy)
784-
md5s = append(md5s, hex.EncodeToString(h.Sum(nil)))
805+
// 复位偏移量
806+
if _, err := seeker.Seek(0, io.SeekStart); err != nil {
807+
return fmt.Errorf("复位文件指针失败: %w", err)
808+
}
809+
810+
goto DoUpload
785811
}
786-
if err == io.EOF || err == io.ErrUnexpectedEOF {
787-
break
812+
}
813+
814+
// 如果非 Seeker 且文件较大 (>128MB)，使用临时文件缓存，避免内存溢出
815+
if size > 128*1024*1024 {
816+
fs.Debugf(o, "大文件上传 (>128MB) 且无法 Seek，启用磁盘缓存模式...")
817+
tempFile, err := os.CreateTemp("", "rclone-baidu-upload-*")
818+
if err != nil {
819+
return fmt.Errorf("创建临时缓存文件失败: %w", err)
788820
}
821+
defer func() {
822+
tempFile.Close()
823+
os.Remove(tempFile.Name())
824+
}()
825+
826+
uploadSource = tempFile // os.File implements ReaderAt
827+
828+
buf := make([]byte, chunkSize)
829+
for {
830+
n, err := io.ReadFull(in, buf)
831+
if n > 0 {
832+
// 写入临时文件
833+
if _, err := tempFile.Write(buf[:n]); err != nil {
834+
return fmt.Errorf("写入临时缓存文件失败: %w", err)
835+
}
836+
837+
// 计算 MD5
838+
h := md5.New()
839+
h.Write(buf[:n])
840+
md5s = append(md5s, hex.EncodeToString(h.Sum(nil)))
841+
}
842+
if err == io.EOF || err == io.ErrUnexpectedEOF {
843+
break
844+
}
845+
if err != nil {
846+
return err
847+
}
848+
}
849+
850+
// 确保数据落盘
851+
if err := tempFile.Sync(); err != nil {
852+
return fmt.Errorf("同步临时文件失败: %w", err)
853+
}
854+
855+
goto DoUpload
856+
}
857+
858+
// 4. 对于较小的流式输入 (<=128MB)，缓冲至内存 (速度快)
859+
{
860+
fs.Debugf(o, "小文件流式上传 (<=128MB)，全量缓存至内存处理...")
861+
var chunks [][]byte
862+
buf := make([]byte, chunkSize)
863+
864+
for {
865+
n, err := io.ReadFull(in, buf)
866+
if n > 0 {
867+
chunkCopy := make([]byte, n)
868+
copy(chunkCopy, buf[:n])
869+
chunks = append(chunks, chunkCopy)
870+
871+
h := md5.New()
872+
h.Write(chunkCopy)
873+
md5s = append(md5s, hex.EncodeToString(h.Sum(nil)))
874+
}
875+
if err == io.EOF || err == io.ErrUnexpectedEOF {
876+
break
877+
}
878+
if err != nil {
879+
return err
880+
}
881+
}
882+
883+
preResp, err := o.precreate(ctx, remote, size, md5s)
789884
if err != nil {
790-
return err
885+
return fmt.Errorf("precreate 失败: %w", err)
791886
}
887+
uploadID := preResp.UploadID
888+
889+
for i, chunkData := range chunks {
890+
_, err := o.sliceUpload(ctx, remote, uploadID, i, bytes.NewReader(chunkData), int64(len(chunkData)))
891+
if err != nil {
892+
return fmt.Errorf("分片 %d 上传失败: %w", i, err)
893+
}
894+
}
895+
896+
md5ListBytes, _ := json.Marshal(md5s)
897+
file, err := o.complete(ctx, remote, uploadID, size, string(md5ListBytes))
898+
if err != nil {
899+
return fmt.Errorf("合并文件失败: %w", err)
900+
}
901+
o.id = strconv.FormatUint(file.FsID, 10)
902+
o.path = file.Path
903+
return nil
792904
}
793905

794-
// 预上传 (Precreate)
906+
DoUpload:
795907
preResp, err := o.precreate(ctx, remote, size, md5s)
796908
if err != nil {
797909
return fmt.Errorf("precreate 失败: %w", err)
798910
}
799911
uploadID := preResp.UploadID
800912

801-
// 分片上传 (SliceUpload)
802-
for i, chunkData := range chunks {
803-
_, err := o.sliceUpload(ctx, remote, uploadID, i, bytes.NewReader(chunkData), int64(len(chunkData)))
804-
if err != nil {
913+
for i := 0; i < len(md5s); i++ {
914+
var currentChunkSize int64
915+
if i == len(md5s)-1 {
916+
currentChunkSize = size - int64(i)*int64(chunkSize)
917+
} else {
918+
currentChunkSize = int64(chunkSize)
919+
}
920+
921+
// 使用 SectionReader 读取指定分片
922+
// uploadSource 必须是 io.ReaderAt (os.File 满足)
923+
sectionReader := io.NewSectionReader(uploadSource, int64(i)*int64(chunkSize), currentChunkSize)
924+
if _, err := o.sliceUpload(ctx, remote, uploadID, i, sectionReader, currentChunkSize); err != nil {
805925
return fmt.Errorf("分片 %d 上传失败: %w", i, err)
806926
}
807927
}
808928

809-
// 调用 Create 提交
810929
md5ListBytes, _ := json.Marshal(md5s)
811930
file, err := o.complete(ctx, remote, uploadID, size, string(md5ListBytes))
812931
if err != nil {
@@ -826,6 +945,7 @@ func (o *Object) precreate(ctx context.Context, remote string, size int64, md5s
826945
v.Set("autoinit", "1")
827946
v.Set("rtype", "3")
828947
v.Set("block_list", string(md5ListJSON))
948+
v.Set("openapi", "xpansdk")
829949

830950
opts := &rest.Opts{
831951
Method: "POST",
@@ -892,6 +1012,7 @@ func (o *Object) rapidUpload(ctx context.Context, remote, contentMD5, sliceMD5 s
8921012
"slice-md5": {sliceMD5},
8931013
"content-crc32": {fmt.Sprintf("%d", crc32Val)},
8941014
"ondup": {"overwrite"},
1015+
"openapi": {"xpansdk"},
8951016
},
8961017
}
8971018
resp := RapidUploadOut{}
@@ -917,9 +1038,10 @@ func (o *Object) simpleUpload(ctx context.Context, in io.Reader, size int64) err
9171038
"User-Agent": "netdisk;P2SP;8.3.1.2;PC;PC-Windows;10.0.19042;WindowsBaiduYunGuanJia",
9181039
},
9191040
Parameters: map[string][]string{
920-
"method": {"upload"},
921-
"path": {remote},
922-
"ondup": {"overwrite"},
1041+
"method": {"upload"},
1042+
"path": {remote},
1043+
"ondup": {"overwrite"},
1044+
"openapi": {"xpansdk"},
9231045
},
9241046
Body: formReader,
9251047
}
@@ -943,7 +1065,7 @@ func (o *Object) sliceUpload(ctx context.Context, remote, uploadID string, partS
9431065
opts := &rest.Opts{
9441066
Method: "POST",
9451067
RootURL: uploadURL,
946-
Path: uriPCSFile,
1068+
Path: uriSuperFile,
9471069
ContentType: contentType,
9481070
ContentLength: &contentLength,
9491071
ExtraHeaders: map[string]string{
@@ -955,6 +1077,7 @@ func (o *Object) sliceUpload(ctx context.Context, remote, uploadID string, partS
9551077
"path": {remote},
9561078
"uploadid": {uploadID},
9571079
"partseq": {strconv.Itoa(partSeq)},
1080+
"openapi": {"xpansdk"},
9581081
},
9591082
Body: formReader,
9601083
}
@@ -974,6 +1097,7 @@ func (o *Object) complete(ctx context.Context, remote, uploadID string, size int
9741097
v.Set("uploadid", uploadID)
9751098
// block_list 需要是 ["...", "..."] 的 JSON 数组格式
9761099
v.Set("block_list", md5ListJSON)
1100+
v.Set("openapi", "xpansdk")
9771101

9781102
opts := &rest.Opts{
9791103
Method: "POST",

0 commit comments

Comments
 (0)