diff --git a/go.mod b/go.mod index fd609b9..43917a0 100644 --- a/go.mod +++ b/go.mod @@ -3,6 +3,7 @@ module github.com/ShukeBta/MMTL go 1.25.0 require ( + github.com/aliyun/alibabacloud-oss-go-sdk-v2 v1.5.2 github.com/fsnotify/fsnotify v1.7.0 github.com/gin-gonic/gin v1.9.1 github.com/glebarez/sqlite v1.11.0 @@ -16,6 +17,7 @@ require ( go.uber.org/zap v1.27.0 golang.org/x/crypto v0.21.0 golang.org/x/sys v0.20.0 + golang.org/x/time v0.15.0 gorm.io/driver/postgres v1.5.7 gorm.io/gorm v1.30.0 ) @@ -72,7 +74,6 @@ require ( golang.org/x/exp v0.0.0-20230905200255-921286631fa9 // indirect golang.org/x/net v0.21.0 // indirect golang.org/x/text v0.20.0 // indirect - golang.org/x/time v0.15.0 // indirect google.golang.org/protobuf v1.31.0 // indirect gopkg.in/ini.v1 v1.67.0 // indirect gopkg.in/yaml.v3 v3.0.1 // indirect diff --git a/go.sum b/go.sum index 3445a7a..00ae618 100644 --- a/go.sum +++ b/go.sum @@ -1,3 +1,5 @@ +github.com/aliyun/alibabacloud-oss-go-sdk-v2 v1.5.2 h1:40yUSXwdkWN851BHCq6uiDhleh7A4+0yIBS+IUAqZVY= +github.com/aliyun/alibabacloud-oss-go-sdk-v2 v1.5.2/go.mod h1:FTzydeQVmR24FI0D6XWUOMKckjXehM/jgMn1xC+DA9M= github.com/bsm/ginkgo/v2 v2.12.0 h1:Ny8MWAHyOepLGlLKYmXG4IEkioBysk6GpaRTLC8zwWs= github.com/bsm/ginkgo/v2 v2.12.0/go.mod h1:SwYbGRRDovPVboqFv0tPTcG1sN61LM1Z4ARdbAV9g4c= github.com/bsm/gomega v1.27.10 h1:yeMWxP2pV2fG3FgAODIY8EiRE3dy0aeFYt4l7wh6yKA= diff --git a/internal/service/cloud/pan115_openapi.go b/internal/service/cloud/pan115_openapi.go index 603f0fd..6731550 100644 --- a/internal/service/cloud/pan115_openapi.go +++ b/internal/service/cloud/pan115_openapi.go @@ -17,6 +17,9 @@ package cloud import ( "context" "fmt" + "io" + "os" + "path/filepath" "strings" "github.com/ShukeBta/MMTL/internal/service/cloud115" @@ -61,16 +64,16 @@ func (p *openAPI115Provider) List(ctx context.Context, dirID string) ([]FileEntr if err != nil { return nil, err } - for _, f := range files { - out = append(out, FileEntry{ - ID: f.FileId, - Name: f.FileName, - IsDir: f.Category == cloud115.TypeDir, - Size: f.FileSize, - MTime: f.Utime, - PickCode: f.PickCode, - }) - } + for _, f := range files { + out = append(out, FileEntry{ + ID: f.FileId, + Name: f.FileName, + IsDir: f.Category == cloud115.TypeDir, + Size: f.FileSize, + MTime: f.Utime, + PickCode: f.PickCode, + }) + } if len(files) < pageSize { break } @@ -105,6 +108,39 @@ func (p *openAPI115Provider) ResolveWithUA(ctx context.Context, fileRef, ua stri // OpenClient 暴露底层客户端（token 刷新用）。 func (p *openAPI115Provider) OpenClient() *cloud115.OpenClient { return p.c } +// PutFileNamed 把本地元数据上传到 115 指定父目录（parentCID 为父目录 cid）。 +// io.Reader 无法携带文件名，因此走独立的 named 上传接口。将内容落为临时文件后 +// 重命名为目标文件名，再交给 115 上传（/open/upload/init 的 file_name 取真实文件名）。 +func (p *openAPI115Provider) PutFileNamed(ctx context.Context, parentCID, fileName string, r io.Reader) error { + tmp, err := os.CreateTemp("", "mmtl-upload-*") + if err != nil { + return fmt.Errorf("115: 创建临时文件失败：%w", err) + } + tmpPath := tmp.Name() + defer func() { + _ = tmp.Close() + _ = os.Remove(tmpPath) + }() + if _, err := io.Copy(tmp, r); err != nil { + return fmt.Errorf("115: 写入临时文件失败：%w", err) + } + if err := tmp.Close(); err != nil { + return fmt.Errorf("115: 关闭临时文件失败：%w", err) + } + // 重命名为目标文件名，保证上传到 115 后保留原始文件名 + if fileName != "" && fileName != filepath.Base(tmpPath) { + namedPath := filepath.Join(filepath.Dir(tmpPath), fileName) + if err := os.Rename(tmpPath, namedPath); err == nil { + tmpPath = namedPath + } + } + _, err = p.c.Upload(ctx, tmpPath, parentCID, "", "") + if err != nil { + return err + } + return nil +} + // RefreshToken 刷新访问令牌并返回新令牌；refresh_token 失效时返回 // cloud115.IsRefreshTokenDead(err) 为 true 的错误。 func (p *openAPI115Provider) RefreshToken(refreshToken string) (*cloud115.TokenData, error) { diff --git a/internal/service/cloud115/cloud115_test.go b/internal/service/cloud115/cloud115_test.go index aaf5dfd..bf2e09c 100644 --- a/internal/service/cloud115/cloud115_test.go +++ b/internal/service/cloud115/cloud115_test.go @@ -371,62 +371,61 @@ func TestThrottleCodeHandling(t *testing.T) { if err == nil { t.Fatal("expected throttle error") } - if !tm.IsThrottled() { - t.Fatal("code 770004 should trigger throttle status") - } + if !tm.IsThrottled() { + t.Fatal("code 770004 should trigger throttle status") + } +} + +func TestRemoteFileDetailRelativePath(t *testing.T) { + rootCID := "3238787832374488117" // 影视库 + + // 场景 1：115 目录 paths 中只有祖先目录链，不包含自身 + d1 := &RemoteFileDetail{ + FileId: "3251154147730910635", + FileName: "出包王女", + Paths: []struct { + FileId string `json:"file_id"` + Name string `json:"file_name"` + }{ + {FileId: "0", Name: "根目录"}, + {FileId: "3238787832374488117", Name: "影视库"}, + {FileId: "3238787913223892116", Name: "动漫"}, + }, + } + if got := d1.RelativePath(rootCID); got != "动漫/出包王女" { + t.Errorf("d1.RelativePath = %q, want %q", got, "动漫/出包王女") } - func TestRemoteFileDetailRelativePath(t *testing.T) { - rootCID := "3238787832374488117" // 影视库 - - // 场景 1：115 目录 paths 中只有祖先目录链，不包含自身 - d1 := &RemoteFileDetail{ - FileId: "3251154147730910635", - FileName: "出包王女", - Paths: []struct { - FileId string `json:"file_id"` - Name string `json:"file_name"` - }{ - {FileId: "0", Name: "根目录"}, - {FileId: "3238787832374488117", Name: "影视库"}, - {FileId: "3238787913223892116", Name: "动漫"}, - }, - } - if got := d1.RelativePath(rootCID); got != "动漫/出包王女" { - t.Errorf("d1.RelativePath = %q, want %q", got, "动漫/出包王女") - } - - // 场景 2：祖先中间目录，自身在 paths 末尾 - d2 := &RemoteFileDetail{ - FileId: "3238787913223892116", - FileName: "动漫", - Paths: []struct { - FileId string `json:"file_id"` - Name string `json:"file_name"` - }{ - {FileId: "0", Name: "根目录"}, - {FileId: "3238787832374488117", Name: "影视库"}, - {FileId: "3238787913223892116", Name: "动漫"}, - }, - } - if got := d2.RelativePath(rootCID); got != "动漫" { - t.Errorf("d2.RelativePath = %q, want %q", got, "动漫") - } - - // 场景 3：根同步目录自身 - d3 := &RemoteFileDetail{ - FileId: rootCID, - FileName: "影视库", - Paths: []struct { - FileId string `json:"file_id"` - Name string `json:"file_name"` - }{ - {FileId: "0", Name: "根目录"}, - {FileId: rootCID, Name: "影视库"}, - }, - } - if got := d3.RelativePath(rootCID); got != "" { - t.Errorf("d3.RelativePath = %q, want %q", got, "") - } + // 场景 2：祖先中间目录，自身在 paths 末尾 + d2 := &RemoteFileDetail{ + FileId: "3238787913223892116", + FileName: "动漫", + Paths: []struct { + FileId string `json:"file_id"` + Name string `json:"file_name"` + }{ + {FileId: "0", Name: "根目录"}, + {FileId: "3238787832374488117", Name: "影视库"}, + {FileId: "3238787913223892116", Name: "动漫"}, + }, + } + if got := d2.RelativePath(rootCID); got != "动漫" { + t.Errorf("d2.RelativePath = %q, want %q", got, "动漫") } + // 场景 3：根同步目录自身 + d3 := &RemoteFileDetail{ + FileId: rootCID, + FileName: "影视库", + Paths: []struct { + FileId string `json:"file_id"` + Name string `json:"file_name"` + }{ + {FileId: "0", Name: "根目录"}, + {FileId: rootCID, Name: "影视库"}, + }, + } + if got := d3.RelativePath(rootCID); got != "" { + t.Errorf("d3.RelativePath = %q, want %q", got, "") + } +} diff --git a/internal/service/cloud115/open.go b/internal/service/cloud115/open.go index 44c64c6..9a51c1d 100644 --- a/internal/service/cloud115/open.go +++ b/internal/service/cloud115/open.go @@ -185,7 +185,6 @@ func (d *RemoteFileDetail) RelativePath(rootCID string) string { return strings.Join(segments, "/") } - // ─── 下载直链 ────────────────────────────────────────────────────────────────── type downloadURLData struct { diff --git a/internal/service/cloud115/oss_multipart.go b/internal/service/cloud115/oss_multipart.go new file mode 100644 index 0000000..00d4a8d --- /dev/null +++ b/internal/service/cloud115/oss_multipart.go @@ -0,0 +1,347 @@ +// 阿里云 OSS multipart 分片上传（用于 115 元数据上传直传）。 +// 使用 115 下发的临时 STS 凭证，将本地文件分片上传到 OSS，并经 complete 回调 +// 通知 115 完成落盘。参考 QMediaSync 的 OSSMultipartUploader 实现。 +package cloud115 + +import ( + "context" + "errors" + "fmt" + "io" + "os" + "sort" + + "github.com/aliyun/alibabacloud-oss-go-sdk-v2/oss" + "github.com/aliyun/alibabacloud-oss-go-sdk-v2/oss/credentials" +) + +const ( + defaultMultipartPartSize int64 = 32 * 1024 * 1024 + multipartPartAlign int64 = 1024 * 1024 + maxMultipartParts int64 = 9999 + maxMultipartPartSize int64 = 5 * 1024 * 1024 * 1024 +) + +type ossMultipartClient interface { + InitiateMultipartUpload(context.Context, *oss.InitiateMultipartUploadRequest, ...func(*oss.Options)) (*oss.InitiateMultipartUploadResult, error) + UploadPart(context.Context, *oss.UploadPartRequest, ...func(*oss.Options)) (*oss.UploadPartResult, error) + ListParts(context.Context, *oss.ListPartsRequest, ...func(*oss.Options)) (*oss.ListPartsResult, error) + CompleteMultipartUpload(context.Context, *oss.CompleteMultipartUploadRequest, ...func(*oss.Options)) (*oss.CompleteMultipartUploadResult, error) + AbortMultipartUpload(context.Context, *oss.AbortMultipartUploadRequest, ...func(*oss.Options)) (*oss.AbortMultipartUploadResult, error) +} + +// OSSMultipartUploader 封装 OSS multipart 上传。 +type OSSMultipartUploader struct { + client ossMultipartClient +} + +// OSSMultipartUploadInput 是 multipart 上传输入。 +type OSSMultipartUploadInput struct { + Bucket string + Object string + Callback string + CallbackVar string + FilePath string + FileSize int64 + UploadId string + PartSize int64 + PartRetryMax int + refreshClient func(context.Context) (ossMultipartClient, error) +} + +// OSSMultipartUploadResult 是 multipart 上传后的结果。 +type OSSMultipartUploadResult struct { + CallbackResult map[string]any + UploadId string + PartSize int64 + TotalParts int + UploadedBytes int64 + UploadedParts int +} + +// CalculateMultipartPartSize 计算 OSS multipart 分片大小与分片数量。 +func CalculateMultipartPartSize(fileSize int64) (int64, int, error) { + if fileSize < 0 { + return 0, 0, fmt.Errorf("文件大小不能为负数：%d", fileSize) + } + partSize := defaultMultipartPartSize + minPartSize := ceilDiv(fileSize, maxMultipartParts) + if minPartSize > partSize { + partSize = roundUp(minPartSize, multipartPartAlign) + } + if partSize > maxMultipartPartSize { + return 0, 0, fmt.Errorf("文件过大，所需分片大小 %d 超过 OSS 上限 %d", partSize, maxMultipartPartSize) + } + totalParts := int(ceilDiv(fileSize, partSize)) + if totalParts == 0 { + totalParts = 1 + } + if int64(totalParts) > maxMultipartParts { + return 0, 0, fmt.Errorf("分片数量 %d 超过上限 %d", totalParts, maxMultipartParts) + } + return partSize, totalParts, nil +} + +// NewOSSMultipartUploader 创建 OSS multipart 上传器。 +func NewOSSMultipartUploader(endpoint, accessKeyId, accessKeySecret, securityToken string) *OSSMultipartUploader { + return &OSSMultipartUploader{client: newOSSMultipartClient(endpoint, accessKeyId, accessKeySecret, securityToken)} +} + +func newOSSMultipartClient(endpoint, accessKeyId, accessKeySecret, securityToken string) ossMultipartClient { + cfg := oss.LoadDefaultConfig(). + WithCredentialsProvider(credentials.NewStaticCredentialsProvider(accessKeyId, accessKeySecret, securityToken)). + WithRegion("cn-shenzhen"). + WithEndpoint(endpoint) + return oss.NewClient(cfg) +} + +// UploadFile 上传文件并完成 OSS multipart，返回 complete callback 结果。 +func (u *OSSMultipartUploader) UploadFile(ctx context.Context, input OSSMultipartUploadInput) (map[string]any, error) { + result, err := u.UploadFileWithResult(ctx, input) + if err != nil { + return nil, err + } + return result.CallbackResult, nil +} + +// UploadFileWithResult 上传文件并返回 multipart 结果。 +func (u *OSSMultipartUploader) UploadFileWithResult(ctx context.Context, input OSSMultipartUploadInput) (OSSMultipartUploadResult, error) { + if input.PartRetryMax <= 0 { + input.PartRetryMax = 3 + } + partSize := input.PartSize + totalParts := 0 + var err error + if partSize <= 0 { + partSize, totalParts, err = CalculateMultipartPartSize(input.FileSize) + if err != nil { + return OSSMultipartUploadResult{}, err + } + } else { + totalParts = int(ceilDiv(input.FileSize, partSize)) + } + + uploadId := input.UploadId + if uploadId == "" { + initResult, err := u.client.InitiateMultipartUpload(ctx, &oss.InitiateMultipartUploadRequest{ + Bucket: oss.Ptr(input.Bucket), + Key: oss.Ptr(input.Object), + RequestCommon: oss.RequestCommon{ + Parameters: map[string]string{"sequential": "1"}, + }, + }) + if err != nil { + return OSSMultipartUploadResult{}, fmt.Errorf("初始化 OSS multipart 失败：%w", err) + } + if initResult.UploadId == nil || *initResult.UploadId == "" { + return OSSMultipartUploadResult{}, fmt.Errorf("初始化 OSS multipart 返回空 upload_id") + } + uploadId = *initResult.UploadId + } + + existingPartMap := make(map[int32]int64) + existingParts, err := u.ListUploadedParts(ctx, input.Bucket, input.Object, uploadId) + if err == nil { + for _, part := range existingParts { + existingPartMap[part.PartNumber] = part.Size + } + } + + file, err := os.Open(input.FilePath) + if err != nil { + return OSSMultipartUploadResult{}, fmt.Errorf("打开待上传文件失败：%w", err) + } + defer file.Close() + + var uploadedBytes int64 + uploadedParts := 0 + completeParts := make([]oss.UploadPart, 0, totalParts) + for partNumber := 1; partNumber <= totalParts; partNumber++ { + offset := int64(partNumber-1) * partSize + length := minInt64(partSize, input.FileSize-offset) + if length < 0 { + length = 0 + } + if existingSize, ok := existingPartMap[int32(partNumber)]; ok && existingSize == length { + uploadedBytes += length + uploadedParts++ + } + etag, err := u.uploadPartWithRetry(ctx, input, uploadId, int32(partNumber), file, offset, length) + if err != nil { + return OSSMultipartUploadResult{}, err + } + uploadedBytes += length + uploadedParts++ + completeParts = append(completeParts, oss.UploadPart{ + PartNumber: int32(partNumber), + ETag: oss.Ptr(etag), + }) + } + sort.Slice(completeParts, func(i, j int) bool { + return completeParts[i].PartNumber < completeParts[j].PartNumber + }) + + completeResult, err := u.client.CompleteMultipartUpload(ctx, &oss.CompleteMultipartUploadRequest{ + Bucket: oss.Ptr(input.Bucket), + Key: oss.Ptr(input.Object), + UploadId: oss.Ptr(uploadId), + CompleteMultipartUpload: &oss.CompleteMultipartUpload{ + Parts: completeParts, + }, + Callback: oss.Ptr(input.Callback), + CallbackVar: oss.Ptr(input.CallbackVar), + }) + if err != nil { + return OSSMultipartUploadResult{}, fmt.Errorf("完成 OSS multipart 失败：%w", err) + } + return OSSMultipartUploadResult{ + CallbackResult: completeResult.CallbackResult, + UploadId: uploadId, + PartSize: partSize, + TotalParts: totalParts, + UploadedBytes: uploadedBytes, + UploadedParts: uploadedParts, + }, nil +} + +// ListUploadedParts 查询 OSS 已上传分片。 +func (u *OSSMultipartUploader) ListUploadedParts(ctx context.Context, bucket, object, uploadId string) ([]struct { + PartNumber int32 + Size int64 +}, error) { + parts := []struct { + PartNumber int32 + Size int64 + }{} + result, err := u.client.ListParts(ctx, &oss.ListPartsRequest{ + Bucket: oss.Ptr(bucket), + Key: oss.Ptr(object), + UploadId: oss.Ptr(uploadId), + MaxParts: 1000, + }) + if err != nil { + return nil, fmt.Errorf("查询 OSS 已上传分片失败：%w", err) + } + for _, part := range result.Parts { + parts = append(parts, struct { + PartNumber int32 + Size int64 + }{PartNumber: part.PartNumber, Size: part.Size}) + } + return parts, nil +} + +func (u *OSSMultipartUploader) uploadPartWithRetry( + ctx context.Context, + input OSSMultipartUploadInput, + uploadId string, + partNumber int32, + file *os.File, + offset, length int64, +) (string, error) { + var lastErr error + for attempt := 0; attempt < input.PartRetryMax; attempt++ { + reader := io.NewSectionReader(file, offset, length) + result, err := u.client.UploadPart(ctx, &oss.UploadPartRequest{ + Bucket: oss.Ptr(input.Bucket), + Key: oss.Ptr(input.Object), + PartNumber: partNumber, + UploadId: oss.Ptr(uploadId), + Body: reader, + ContentLength: oss.Ptr(length), + }) + if err == nil { + if result.ETag == nil || *result.ETag == "" { + return "", fmt.Errorf("OSS part %d 返回空 ETag", partNumber) + } + return *result.ETag, nil + } + lastErr = err + if attempt < input.PartRetryMax-1 && input.refreshClient != nil { + refreshed, refreshErr := input.refreshClient(ctx) + if refreshErr != nil { + lastErr = refreshErr + continue + } + u.client = refreshed + } + } + return "", fmt.Errorf("上传 OSS part %d 失败：%w", partNumber, lastErr) +} + +// ParseCompleteCallbackResult 校验并解析 OSS complete 后的 115 callback 结果。 +func ParseCompleteCallbackResult(result map[string]any) (UploadCompleteResult, error) { + if result == nil { + return UploadCompleteResult{}, errors.New("OSS complete callback 结果为空") + } + if state, ok := result["state"].(bool); ok && !state { + return UploadCompleteResult{}, fmt.Errorf("115 callback 返回失败：%s", anyToString(result["message"])) + } + if message := anyToString(result["message"]); message != "" { + return UploadCompleteResult{}, fmt.Errorf("115 callback 返回错误：%s", message) + } + data, ok := result["data"].(map[string]any) + if !ok { + return UploadCompleteResult{}, errors.New("115 callback 缺少 data") + } + complete := UploadCompleteResult{ + FileId: anyToString(data["file_id"]), + PickCode: anyToString(data["pick_code"]), + ParentId: anyToString(data["parent_id"]), + Sha1: anyToString(data["sha1"]), + Size: anyToInt64(data["size"]), + Mtime: anyToInt64(data["mtime"]), + } + if complete.FileId == "" || complete.PickCode == "" { + return UploadCompleteResult{}, errors.New("115 callback 缺少 file_id/pick_code") + } + return complete, nil +} + +func ceilDiv(n, d int64) int64 { + if d <= 0 { + return 0 + } + if n <= 0 { + return 0 + } + return (n + d - 1) / d +} + +func roundUp(n, align int64) int64 { + if align <= 0 { + return n + } + return ceilDiv(n, align) * align +} + +func minInt64(a, b int64) int64 { + if a < b { + return a + } + return b +} + +func anyToInt64(v any) int64 { + switch t := v.(type) { + case string: + var n int64 + fmt.Sscanf(t, "%d", &n) + return n + case float64: + return int64(t) + case int64: + return t + case int: + return int64(t) + default: + return 0 + } +} + +func anyToString(v any) string { + if s, ok := v.(string); ok { + return s + } + return fmt.Sprintf("%v", v) +} diff --git a/internal/service/cloud115/sha1.go b/internal/service/cloud115/sha1.go new file mode 100644 index 0000000..f8a2cb4 --- /dev/null +++ b/internal/service/cloud115/sha1.go @@ -0,0 +1,47 @@ +package cloud115 + +import ( + "crypto/sha1" + "encoding/hex" + "io" + "os" +) + +// FileSHA1 计算文件完整 SHA1（小写 hex）。 +func FileSHA1(path string) (string, error) { + f, err := os.Open(path) + if err != nil { + return "", err + } + defer f.Close() + h := sha1.New() + if _, err := io.Copy(h, f); err != nil { + return "", err + } + return hex.EncodeToString(h.Sum(nil)), nil +} + +// FileSHA1Partial 计算文件 [start,end]（含）字节区间的 SHA1（小写 hex）。 +// 用于 115 上传二次签名按 sign_check 指定的区间重算哈希。 +func FileSHA1Partial(path string, start, end int64) (string, error) { + if start < 0 { + start = 0 + } + if end < start { + end = start + } + f, err := os.Open(path) + if err != nil { + return "", err + } + defer f.Close() + if _, err := f.Seek(start, io.SeekStart); err != nil { + return "", err + } + length := end - start + 1 + h := sha1.New() + if _, err := io.CopyN(h, f, length); err != nil { + return "", err + } + return hex.EncodeToString(h.Sum(nil)), nil +} diff --git a/internal/service/cloud115/upload.go b/internal/service/cloud115/upload.go new file mode 100644 index 0000000..eae8e9d --- /dev/null +++ b/internal/service/cloud115/upload.go @@ -0,0 +1,355 @@ +// 115 网盘元数据上传能力：115 开放平台调度 + 阿里云 OSS 直传。 +// 参考 QMediaSync 的上传流程实现： +// +// POST /open/upload/init 上传初始化/秒传调度（含二次签名） +// GET /open/upload/get_token 获取 OSS 临时上传凭证（STS） +// OSS multipart 分片直传 + callback 完成 +// +// 上传目标父目录为 115 目录 ID（cid），而非路径字符串。 +package cloud115 + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "os" + "strconv" + "strings" +) + +// 115 上传状态码。 +const ( + UploadInitStatusNeedUpload = 1 // 需要真实上传 + UploadInitStatusRapidUploaded = 2 // 秒传成功 + UploadInitStatusSignFailed = 6 // 签名验证失败 + UploadInitStatusNeedSign = 7 // 需要二次签名 + UploadInitStatusSignRejected = 8 // 签名认证失败 +) + +// UploadInitRequest 是 /open/upload/init 的结构化请求。 +type UploadInitRequest struct { + FileName string + FileSize int64 + ParentFileId string + FileSha1 string + Preid string + PickCode string + TopUpload string + SignKey string + SignVal string +} + +// UploadInitResult 是 /open/upload/init 的调度结果。 +type UploadInitResult struct { + PickCode string + Status int + FileId string + Target string + Bucket string + Object string + SignKey string + SignCheck string + Callback UploadResultCallBack +} + +type uploadScheduleAPIResult struct { + PickCode string `json:"pick_code"` + Status int `json:"status"` + FileId string `json:"file_id"` + Target string `json:"target"` + Version string `json:"version"` + Bucket string `json:"bucket"` + Object string `json:"object"` + SignKey string `json:"sign_key"` + SignCheck string `json:"sign_check"` + Callback json.RawMessage `json:"callback"` +} + +// UploadResultCallBack 是 init 返回给 OSS complete 使用的 callback 内容。 +type UploadResultCallBack struct { + Callback string `json:"callback"` + CallbackVar string `json:"callback_var"` +} + +// UploadToken 是 /open/upload/get_token 返回的 OSS STS 临时凭证。 +type UploadToken struct { + Endpoint string `json:"endpoint"` + AccessKeySecret string `json:"AccessKeySecret"` + AccessKeySecrett string `json:"AccessKeySecrett"` + SecurityToken string `json:"SecurityToken"` + Expiration string `json:"Expiration"` + AccessKeyId string `json:"AccessKeyId"` +} + +func (token *UploadToken) normalize() { + if token == nil { + return + } + if token.AccessKeySecret == "" { + token.AccessKeySecret = token.AccessKeySecrett + } +} + +// UploadCompleteResult 是 OSS complete callback 成功后的远端文件定位结果。 +type UploadCompleteResult struct { + FileId string + PickCode string + ParentId string + Sha1 string + Size int64 + Mtime int64 +} + +// SignCheckRange 是 115 二次认证要求的闭区间 [start,end]。 +type SignCheckRange struct { + Start int64 + End int64 +} + +// UploadInit 调用 115 上传初始化/秒传调度接口。 +func (c *OpenClient) UploadInit(ctx context.Context, input UploadInitRequest) (*UploadInitResult, error) { + params := buildUploadInitForm(input) + resp, err := c.doAuthJSON(ctx, "POST", ProAPIBase+"/open/upload/init", params, 2) + if err != nil { + return nil, err + } + var raw uploadScheduleAPIResult + if err := json.Unmarshal(resp.Data, &raw); err != nil { + return nil, fmt.Errorf("115: 解析 upload/init 结果失败：%w", err) + } + callback, err := decodeUploadCallback(raw.Callback) + if err != nil { + return nil, err + } + return &UploadInitResult{ + PickCode: raw.PickCode, + Status: raw.Status, + FileId: raw.FileId, + Target: raw.Target, + Bucket: raw.Bucket, + Object: raw.Object, + SignKey: raw.SignKey, + SignCheck: raw.SignCheck, + Callback: callback, + }, nil +} + +func buildUploadInitForm(input UploadInitRequest) map[string]string { + topUpload := input.TopUpload + if topUpload == "" { + topUpload = "0" + } + params := map[string]string{ + "file_name": input.FileName, + "file_size": strconv.FormatInt(input.FileSize, 10), + "target": fmt.Sprintf("U_1_%s", input.ParentFileId), + "fileid": input.FileSha1, + "preid": input.Preid, + "topupload": topUpload, + } + if input.PickCode != "" { + params["pick_code"] = input.PickCode + } + if input.SignKey != "" && input.SignVal != "" { + params["sign_key"] = input.SignKey + params["sign_val"] = input.SignVal + } + return params +} + +func decodeUploadCallback(raw json.RawMessage) (UploadResultCallBack, error) { + if len(raw) == 0 || string(raw) == "null" { + return UploadResultCallBack{}, nil + } + if raw[0] == '[' { + var callbacks []UploadResultCallBack + if err := json.Unmarshal(raw, &callbacks); err != nil { + return UploadResultCallBack{}, err + } + if len(callbacks) == 0 { + return UploadResultCallBack{}, nil + } + return callbacks[0], nil + } + var callback UploadResultCallBack + if err := json.Unmarshal(raw, &callback); err != nil { + return UploadResultCallBack{}, err + } + return callback, nil +} + +func parseSignCheckRange(value string) (SignCheckRange, error) { + parts := strings.Split(value, "-") + if len(parts) != 2 { + return SignCheckRange{}, fmt.Errorf("sign_check 格式错误：%s", value) + } + start, err := strconv.ParseInt(strings.TrimSpace(parts[0]), 10, 64) + if err != nil { + return SignCheckRange{}, err + } + end, err := strconv.ParseInt(strings.TrimSpace(parts[1]), 10, 64) + if err != nil { + return SignCheckRange{}, err + } + if start < 0 || end < start { + return SignCheckRange{}, fmt.Errorf("sign_check 范围非法：%s", value) + } + return SignCheckRange{Start: start, End: end}, nil +} + +// GetUploadToken 获取 115 下发的 OSS 临时上传凭证。 +func (c *OpenClient) GetUploadToken(ctx context.Context) (*UploadToken, error) { + resp, err := c.doAuthJSON(ctx, "GET", ProAPIBase+"/open/upload/get_token", nil, 2) + if err != nil { + return nil, err + } + var token UploadToken + if err := json.Unmarshal(resp.Data, &token); err != nil { + return nil, fmt.Errorf("115: 解析 get_token 结果失败：%w", err) + } + token.normalize() + return &token, nil +} + +// Upload 上传单个本地文件到 115 指定父目录（cid），返回成功后的远端文件信息。 +// filePath 必须是落到磁盘的真实文件路径（调用方负责把 io.Reader 落盘为临时文件）。 +func (c *OpenClient) Upload(ctx context.Context, filePath, parentCID, signKey, signVal string) (*UploadCompleteResult, error) { + fileSize := fileSizeOf(filePath) + if fileSize < 0 { + return nil, fmt.Errorf("115: 无法获取文件大小：%s", filePath) + } + fileSha1, err := FileSHA1(filePath) + if err != nil { + return nil, fmt.Errorf("115: 计算文件 SHA1 失败：%w", err) + } + preSha1, err := FileSHA1Partial(filePath, 0, 128*1024-1) + if err != nil { + return nil, fmt.Errorf("115: 计算文件前 128 KiB SHA1 失败：%w", err) + } + request := UploadInitRequest{ + FileName: baseNameOf(filePath), + FileSize: fileSize, + ParentFileId: parentCID, + FileSha1: fileSha1, + Preid: preSha1, + TopUpload: "0", + SignKey: signKey, + SignVal: signVal, + } + initResult, err := c.UploadInit(ctx, request) + if err != nil { + return nil, fmt.Errorf("115: 上传初始化失败：%w", err) + } + status := initResult.Status + if status == UploadInitStatusNeedSign { + // 二次签名：按 sign_check 指定区间重算 sha1 + rng, err := parseSignCheckRange(initResult.SignCheck) + if err != nil { + return nil, err + } + signValue, err := FileSHA1Partial(filePath, rng.Start, rng.End) + if err != nil { + return nil, err + } + request.SignKey = initResult.SignKey + request.SignVal = signValue + initResult, err = c.UploadInit(ctx, request) + if err != nil { + return nil, fmt.Errorf("115: 上传二次签名失败：%w", err) + } + status = initResult.Status + } + switch status { + case UploadInitStatusRapidUploaded: + // 秒传成功 + return &UploadCompleteResult{FileId: initResult.FileId, PickCode: initResult.PickCode}, nil + case UploadInitStatusSignFailed: + return nil, fmt.Errorf("115: 签名验证后失败") + case UploadInitStatusSignRejected: + return nil, fmt.Errorf("115: 签名认证失败") + case UploadInitStatusNeedUpload: + // 真实上传：OSS multipart + default: + return &UploadCompleteResult{FileId: initResult.FileId, PickCode: initResult.PickCode}, nil + } + + if initResult.Bucket == "" || initResult.Object == "" { + return nil, fmt.Errorf("115: upload/init 缺少 bucket/object 信息") + } + token, err := c.GetUploadToken(ctx) + if err != nil { + return nil, fmt.Errorf("115: 获取上传凭证失败：%w", err) + } + if token == nil || token.Endpoint == "" || token.AccessKeyId == "" || token.AccessKeySecret == "" { + return nil, fmt.Errorf("115: 上传凭证不完整") + } + uploader := NewOSSMultipartUploader(token.Endpoint, token.AccessKeyId, token.AccessKeySecret, token.SecurityToken) + result, err := uploader.UploadFile(ctx, OSSMultipartUploadInput{ + Bucket: initResult.Bucket, + Object: initResult.Object, + Callback: initResult.Callback.Callback, + CallbackVar: initResult.Callback.CallbackVar, + FilePath: filePath, + FileSize: fileSize, + refreshClient: func(ctx context.Context) (ossMultipartClient, error) { + refreshed, rerr := c.GetUploadToken(ctx) + if rerr != nil || refreshed == nil { + return nil, rerr + } + return newOSSMultipartClient(refreshed.Endpoint, refreshed.AccessKeyId, refreshed.AccessKeySecret, refreshed.SecurityToken), nil + }, + }) + if err != nil { + return nil, fmt.Errorf("115: OSS 上传失败：%w", err) + } + complete, err := ParseCompleteCallbackResult(result) + if err != nil { + return nil, err + } + return &complete, nil +} + +// MkDir 在 115 的 parentCid 下创建目录，返回新目录 cid。 +func (c *OpenClient) MkDir(ctx context.Context, parentCID, name string) (string, error) { + params := map[string]string{ + "cname": name, + "pid": parentCID, + } + resp, err := c.doAuthJSON(ctx, "POST", ProAPIBase+"/open/folder/add", params, 2) + if err != nil { + return "", err + } + // /open/folder/add 结构：{ aid, cid, fid, name, pid, ... }，单一对象 + var r struct { + Cid string `json:"cid"` + } + if err := json.Unmarshal(resp.Data, &r); err != nil { + return "", fmt.Errorf("115: 解析 folder/add 结果失败：%w", err) + } + if r.Cid == "" { + return "", errors.New("115: folder/add 未返回 cid") + } + return r.Cid, nil +} + +func fileSizeOf(path string) int64 { + info, err := os.Stat(path) + if err != nil { + return -1 + } + if info.IsDir() { + return -1 + } + return info.Size() +} + +func baseNameOf(path string) string { + s := path + for i := len(s) - 1; i >= 0; i-- { + if s[i] == '/' || s[i] == '\\' { + return s[i+1:] + } + } + return s +} diff --git a/internal/service/cloud115/upload_test.go b/internal/service/cloud115/upload_test.go new file mode 100644 index 0000000..5664e1f --- /dev/null +++ b/internal/service/cloud115/upload_test.go @@ -0,0 +1,94 @@ +package cloud115 + +import ( + "os" + "path/filepath" + "testing" +) + +func TestFileSHA1(t *testing.T) { + dir := t.TempDir() + path := filepath.Join(dir, "a.txt") + if err := os.WriteFile(path, []byte("hello"), 0o644); err != nil { + t.Fatal(err) + } + sum, err := FileSHA1(path) + if err != nil { + t.Fatal(err) + } + // sha1("hello") = aaf4c61ddcc5e8a2dabede0f3b482cd9aea9434d + if sum != "aaf4c61ddcc5e8a2dabede0f3b482cd9aea9434d" { + t.Errorf("unexpected sha1: %s", sum) + } +} + +func TestFileSHA1Partial(t *testing.T) { + dir := t.TempDir() + path := filepath.Join(dir, "b.txt") + // 10 bytes: "0123456789" + if err := os.WriteFile(path, []byte("0123456789"), 0o644); err != nil { + t.Fatal(err) + } + // bytes [2,4] = "234" + sum, err := FileSHA1Partial(path, 2, 4) + if err != nil { + t.Fatal(err) + } + if sum != "0ec09ef9836da03f1add21e3ef607627e687e790" { + t.Errorf("unexpected partial sha1: %s", sum) + } +} + +func TestParseSignCheckRange(t *testing.T) { + rng, err := parseSignCheckRange("0-131071") + if err != nil { + t.Fatal(err) + } + if rng.Start != 0 || rng.End != 131071 { + t.Errorf("unexpected range: %+v", rng) + } + if _, err := parseSignCheckRange("bad"); err == nil { + t.Error("expected error for bad range") + } + if _, err := parseSignCheckRange("100-50"); err == nil { + t.Error("expected error for end partSize 32MiB, 1 part + ps, parts, err := CalculateMultipartPartSize(1 << 20) + if err != nil { + t.Fatal(err) + } + if ps != defaultMultipartPartSize { + t.Errorf("partSize=%d, want %d", ps, defaultMultipartPartSize) + } + if parts != 1 { + t.Errorf("parts=%d, want 1", parts) + } + // zero-size -> 1 part + _, parts, err = CalculateMultipartPartSize(0) + if err != nil { + t.Fatal(err) + } + if parts != 1 { + t.Errorf("zero-size parts=%d, want 1", parts) + } + // negative -> error + if _, _, err := CalculateMultipartPartSize(-1); err == nil { + t.Error("expected error for negative size") + } +} + +func TestBaseNameOf(t *testing.T) { + if got := baseNameOf("/a/b/file.nfo"); got != "file.nfo" { + t.Errorf("got %s", got) + } + if got := baseNameOf("a\\b\\c.jpg"); got != "c.jpg" { + t.Errorf("got %s", got) + } + if got := baseNameOf("top.txt"); got != "top.txt" { + t.Errorf("got %s", got) + } +} diff --git a/internal/service/strm_queue.go b/internal/service/strm_queue.go index f3ff3f3..4e48de0 100644 --- a/internal/service/strm_queue.go +++ b/internal/service/strm_queue.go @@ -157,7 +157,7 @@ func (s *StrmService) processUploadTask(ctx context.Context, task *model.StrmUpl } } if task.Provider == model.StrmProvider115 { - finish(model.StrmTaskFailed, "115 网盘暂不支持元数据上传") + s.processUpload115(ctx, task) return } acct, err := s.repo.StrmAccount.FindByID(ctx, task.AccountID) @@ -196,6 +196,48 @@ func (s *StrmService) processUploadTask(ctx context.Context, task *model.StrmUpl finish(model.StrmTaskDone, "") } +// processUpload115 115 元数据上传：task.RemotePath 存的是父目录 cid，FileName 为远端文件名。 +func (s *StrmService) processUpload115(ctx context.Context, task *model.StrmUploadTask) { + finish := func(status, message string) { + now := time.Now() + task.Status = status + task.Error = message + task.FinishedAt = &now + if err := s.repo.StrmUpload.Update(context.Background(), task); err != nil { + s.log.Warn("update strm upload task failed", zap.Error(err)) + } + } + acct, err := s.repo.StrmAccount.FindByID(ctx, task.AccountID) + if err != nil || acct == nil { + finish(model.StrmTaskFailed, "网盘账号不存在") + return + } + provider, err := s.providerFor(ctx, acct) + if err != nil { + s.uploadTaskFailWithRetry(task, err.Error()) + return + } + named, ok := provider.(interface { + PutFileNamed(ctx context.Context, parentCID, fileName string, r io.Reader) error + }) + if !ok { + finish(model.StrmTaskFailed, "该网盘不支持元数据上传") + return + } + f, err := os.Open(task.LocalPath) + if err != nil { + s.uploadTaskFailWithRetry(task, "打开本地文件失败："+err.Error()) + return + } + if err := named.PutFileNamed(ctx, task.RemotePath, task.FileName, f); err != nil { + _ = f.Close() + s.uploadTaskFailWithRetry(task, "上传失败："+err.Error()) + return + } + _ = f.Close() + finish(model.StrmTaskDone, "") +} + // downloadTaskFailWithRetry 下载失败任务按退避重试，超过上限标记 failed。 func (s *StrmService) downloadTaskFailWithRetry(task *model.StrmDownloadTask, message string) { if !retryTask(&task.RetryCount, &task.Status, &task.Error, &task.NextTryAt, &task.FinishedAt, message) { diff --git a/internal/service/strm_sync.go b/internal/service/strm_sync.go index f16a1b6..8038fd3 100644 --- a/internal/service/strm_sync.go +++ b/internal/service/strm_sync.go @@ -46,7 +46,10 @@ type strmSyncState struct { activeUploadPaths map[string]bool // 本地已在排队/进行的上传任务路径（内存去重） pendingDownloads []*model.StrmDownloadTask pendingUploads []*model.StrmUploadTask - dirCache sync.Map // dirID (string) -> relativePath (string) + dirCache sync.Map // dirID (string) -> relativePath (string) + dirPathToID map[string]string // relativePath (string) -> dirID（115 上传父目录寻址用，walk 后构建） + + scanIncomplete atomic.Bool // 远端目录树/文件列表本次扫描不完整 → 禁止增量 prune 误删本地文件 } // StartSync 启动一次同步（异步执行，同一目录同时只允许一个任务）。 @@ -280,7 +283,21 @@ func (st *strmSyncState) run() error { } st.flushPendingDownloads() st.flushProgress() - if st.cfg.UploadMeta && st.provider != nil && st.p.Provider != model.StrmProvider115 { + if st.cfg.UploadMeta && st.provider != nil { + // 115 上传需要父目录 cid，先用 dirCache 构建「路径 → cid」反向索引 + if st.p.Provider == model.StrmProvider115 { + reversed := map[string]string{} + st.dirCache.Range(func(key, value any) bool { + path, ok := value.(string) + if ok && path != "" { + if id, ok2 := key.(string); ok2 { + reversed[path] = id + } + } + return true + }) + st.dirPathToID = reversed + } if err := st.scanLocalMetaForUpload(); err != nil { return err } @@ -439,6 +456,29 @@ func (st *strmSyncState) isMetaExt(ext string) bool { return false } +// cleanDirRel 对 115 扁平化拉取的目录相对路径逐段套用目录级文件名清洗， +// 确保与 walkRemote / joinLocalRel（sanitizeRelativePath）使用同一套清洗规则。 +// 若不清洗，目录名中的冒号等非法字符会直达 rel，而 seenVideo/seenMeta 的 key +// 与磁盘实际路径不一致，导致 pruneLocal 误删已下载的 strm / 元数据。 +// 空 rel（根目录）原样返回。 +func cleanDirRel(rel string) string { + if rel == "" { + return "" + } + parts := strings.Split(rel, "/") + out := make([]string, 0, len(parts)) + for _, part := range parts { + if part == "" { + continue + } + clean := cleanEntryName(part, true) + if clean != "" && clean != "." && clean != ".." { + out = append(out, clean) + } + } + return strings.Join(out, "/") +} + // walk115Flat 使用 115 开放平台扁平化分页批量拉取机制与目录拓扑缓存（参考 QMediaSync）。 // 极大地降低 API 请求次数并支持毫秒级/秒级增量同步。 func (st *strmSyncState) walk115Flat(open115 *cloud115.OpenClient) error { @@ -456,23 +496,23 @@ func (st *strmSyncState) walk115Flat(open115 *cloud115.OpenClient) error { if err := st.s.repo.StrmDirCache.DeleteBySyncPathID(ctx, st.p.ID); err != nil { st.s.log.Warn("delete strm dir cache failed", zap.Error(err)) } - } else { - // 增量同步：预加载历史目录缓存（过滤历史一对多塌陷冲突的脏数据以自愈刷新） - cached, err := st.s.repo.StrmDirCache.ListBySyncPathID(ctx, st.p.ID) - if err == nil { - pathCounts := make(map[string]int, len(cached)) - for _, item := range cached { - pathCounts[item.Path]++ - } - for _, item := range cached { - // 若同一个 path 对应了多个不同 dir_id，说明包含历史层级塌陷的脏数据，不预加载，让后续步骤重新向 115 获取精确路径 - if pathCounts[item.Path] > 1 { - continue - } - st.dirCache.Store(item.DirID, item.Path) + } else { + // 增量同步：预加载历史目录缓存（过滤历史一对多塌陷冲突的脏数据以自愈刷新） + cached, err := st.s.repo.StrmDirCache.ListBySyncPathID(ctx, st.p.ID) + if err == nil { + pathCounts := make(map[string]int, len(cached)) + for _, item := range cached { + pathCounts[item.Path]++ + } + for _, item := range cached { + // 若同一个 path 对应了多个不同 dir_id，说明包含历史层级塌陷的脏数据，不预加载，让后续步骤重新向 115 获取精确路径 + if pathCounts[item.Path] > 1 { + continue } + st.dirCache.Store(item.DirID, cleanDirRel(item.Path)) } } + } // 2. 探测文件总数 const pageSize = 1150 @@ -602,9 +642,12 @@ func (st *strmSyncState) walk115Flat(open115 *cloud115.OpenClient) error { detail, err := open115.GetFsDetailByCid(ctx, pid) if err != nil { st.s.log.Warn("115: 获取目录详情失败", zap.String("pid", pid), zap.Error(err)) + // 目录详情解析失败会导致下游文件 rel 无法还原真实父路径， + // seen key 与磁盘路径对不上，增量 prune 会误删本地文件，标记本次扫描不完整。 + st.scanIncomplete.Store(true) } else if detail != nil { // 解析相对路径 - relPath := detail.RelativePath(rootCID) + relPath := cleanDirRel(detail.RelativePath(rootCID)) st.dirCache.Store(pid, relPath) _ = st.s.repo.StrmDirCache.Set(ctx, st.p.ID, pid, relPath) @@ -625,7 +668,7 @@ func (st *strmSyncState) walk115Flat(open115 *cloud115.OpenClient) error { break } } - ancestorRel := subDetail.RelativePath(rootCID) + ancestorRel := cleanDirRel(subDetail.RelativePath(rootCID)) st.dirCache.Store(ancestor.FileId, ancestorRel) _ = st.s.repo.StrmDirCache.Set(ctx, st.p.ID, ancestor.FileId, ancestorRel) } @@ -653,8 +696,13 @@ func (st *strmSyncState) walk115Flat(open115 *cloud115.OpenClient) error { rel = cleanName } else { if parentVal, ok := st.dirCache.Load(f.Pid); ok && parentVal.(string) != "" { - rel = parentVal.(string) + "/" + cleanName + rel = cleanDirRel(parentVal.(string)) + "/" + cleanName } else { + // 父目录不在目录缓存（目录详情先前解析失败），无法还原真实相对路径。 + // 该文件会落到根/错误路径，seen key 与磁盘路径不符，增量 prune 会误删，标记扫描不完整。 + if st.syncType == model.StrmSyncTypeIncremental { + st.scanIncomplete.Store(true) + } rel = cleanName } } @@ -966,124 +1014,124 @@ func (st *strmSyncState) walkLocalSource() error { if err != nil { return nil } - target, err := joinLocalRel(st.p.LocalPath, relSansExt+".strm") - if err != nil { - return nil - } - mTime := info.ModTime() - if st.syncType == model.StrmSyncTypeIncremental { - if tInfo, err := os.Stat(target); err == nil && tInfo.Size() > 0 && tInfo.ModTime().Unix() == mTime.Unix() { - st.mu.Lock() - st.rec.Skipped++ - st.mu.Unlock() - st.touchProgress() - return nil - } - } - if data, err := os.ReadFile(target); err == nil && string(data) == content { - _ = os.Chtimes(target, mTime, mTime) + target, err := joinLocalRel(st.p.LocalPath, relSansExt+".strm") + if err != nil { + return nil + } + mTime := info.ModTime() + if st.syncType == model.StrmSyncTypeIncremental { + if tInfo, err := os.Stat(target); err == nil && tInfo.Size() > 0 && tInfo.ModTime().Unix() == mTime.Unix() { st.mu.Lock() st.rec.Skipped++ st.mu.Unlock() st.touchProgress() return nil } - if err := os.MkdirAll(filepath.Dir(target), 0o755); err != nil { - return nil - } - tmp := target + ".tmp" - if err := os.WriteFile(tmp, []byte(content), 0o644); err == nil { - _ = os.Rename(tmp, target) - _ = os.Chtimes(target, mTime, mTime) - } else { - _ = os.Remove(tmp) - } + } + if data, err := os.ReadFile(target); err == nil && string(data) == content { + _ = os.Chtimes(target, mTime, mTime) st.mu.Lock() - st.rec.NewStrm++ + st.rec.Skipped++ st.mu.Unlock() st.touchProgress() return nil + } + if err := os.MkdirAll(filepath.Dir(target), 0o755); err != nil { + return nil + } + tmp := target + ".tmp" + if err := os.WriteFile(tmp, []byte(content), 0o644); err == nil { + _ = os.Rename(tmp, target) + _ = os.Chtimes(target, mTime, mTime) + } else { + _ = os.Remove(tmp) + } + st.mu.Lock() + st.rec.NewStrm++ + st.mu.Unlock() + st.touchProgress() + return nil }) } - // scanLocalMetaForUpload 扫描本地元数据，与远端比对后入上传队列。 - func (st *strmSyncState) scanLocalMetaForUpload() error { - defer st.flushPendingUploads() - if st.activeUploadPaths == nil { - if active, err := st.s.repo.StrmUpload.GetActiveLocalPathMap(st.ctx, st.p.ID); err == nil { - st.activeUploadPaths = active - } else { - st.activeUploadPaths = map[string]bool{} - } +// scanLocalMetaForUpload 扫描本地元数据，与远端比对后入上传队列。 +func (st *strmSyncState) scanLocalMetaForUpload() error { + defer st.flushPendingUploads() + if st.activeUploadPaths == nil { + if active, err := st.s.repo.StrmUpload.GetActiveLocalPathMap(st.ctx, st.p.ID); err == nil { + st.activeUploadPaths = active + } else { + st.activeUploadPaths = map[string]bool{} } - localRoot := filepath.Clean(st.p.LocalPath) - return filepath.WalkDir(localRoot, func(path string, d os.DirEntry, err error) error { - if err != nil { - return nil - } - if path == localRoot { - return nil - } - select { - case <-st.ctx.Done(): - return st.ctx.Err() - default: - } - if d.IsDir() { - return nil - } - rel, err := filepath.Rel(localRoot, path) - if err != nil { - return nil - } - rel = filepath.ToSlash(rel) - ext := strings.ToLower(filepath.Ext(rel)) - if !st.isMetaExt(ext) { - return nil - } - info, err := d.Info() - if err != nil { - return nil - } - st.mu.Lock() - _, exists := st.remoteMeta["m:"+rel] - st.mu.Unlock() - if exists { - // 网盘端已存在该元数据文件，跳过上传 - return nil - } - st.mu.Lock() - if st.activeUploadPaths != nil && st.activeUploadPaths[path] { - st.mu.Unlock() - return nil - } - if st.activeUploadPaths != nil { - st.activeUploadPaths[path] = true - } - st.mu.Unlock() - - task := &model.StrmUploadTask{ - SyncPathID: st.p.ID, - AccountID: st.p.AccountID, - Provider: st.p.Provider, - FileName: filepath.Base(rel), - LocalPath: path, - RemotePath: st.remoteUploadPath(rel), - Size: info.Size(), - Status: model.StrmTaskPending, - } - st.mu.Lock() - st.pendingUploads = append(st.pendingUploads, task) - shouldFlush := len(st.pendingUploads) >= 100 - st.rec.Uploaded++ - st.mu.Unlock() - - if shouldFlush { - st.flushPendingUploads() - } - return nil - }) } + localRoot := filepath.Clean(st.p.LocalPath) + return filepath.WalkDir(localRoot, func(path string, d os.DirEntry, err error) error { + if err != nil { + return nil + } + if path == localRoot { + return nil + } + select { + case <-st.ctx.Done(): + return st.ctx.Err() + default: + } + if d.IsDir() { + return nil + } + rel, err := filepath.Rel(localRoot, path) + if err != nil { + return nil + } + rel = filepath.ToSlash(rel) + ext := strings.ToLower(filepath.Ext(rel)) + if !st.isMetaExt(ext) { + return nil + } + info, err := d.Info() + if err != nil { + return nil + } + st.mu.Lock() + _, exists := st.remoteMeta["m:"+rel] + st.mu.Unlock() + if exists { + // 网盘端已存在该元数据文件，跳过上传 + return nil + } + st.mu.Lock() + if st.activeUploadPaths != nil && st.activeUploadPaths[path] { + st.mu.Unlock() + return nil + } + if st.activeUploadPaths != nil { + st.activeUploadPaths[path] = true + } + st.mu.Unlock() + + task := &model.StrmUploadTask{ + SyncPathID: st.p.ID, + AccountID: st.p.AccountID, + Provider: st.p.Provider, + FileName: filepath.Base(rel), + LocalPath: path, + RemotePath: st.uploadRemoteTarget(rel), + Size: info.Size(), + Status: model.StrmTaskPending, + } + st.mu.Lock() + st.pendingUploads = append(st.pendingUploads, task) + shouldFlush := len(st.pendingUploads) >= 100 + st.rec.Uploaded++ + st.mu.Unlock() + + if shouldFlush { + st.flushPendingUploads() + } + return nil + }) +} // remoteUploadPath 远端元数据目标路径 = 同步目录远端根 + 相对路径。 func (st *strmSyncState) remoteUploadPath(rel string) string { @@ -1094,6 +1142,31 @@ func (st *strmSyncState) remoteUploadPath(rel string) string { return root + "/" + rel } +// uploadRemoteTarget 返回上传任务的目标远端描述。 +// - 115：返回父目录 cid（供 PutFileNamed 定位），基于 dirPathToID 把父目录相对路径映射到 cid。 +// - 网盘桥接（clouddrive2/openlist）：返回完整远端路径。 +func (st *strmSyncState) uploadRemoteTarget(rel string) string { + if st.p.Provider == model.StrmProvider115 { + dir := rel + if idx := strings.LastIndexByte(dir, '/'); idx >= 0 { + dir = dir[:idx] + } else { + dir = "" + } + if dir == "" { + // 文件在同步根目录下，父目录即 115 同步根目录 ID + return st.p.RemotePath + } + if cid, ok := st.dirPathToID[dir]; ok && cid != "" { + return cid + } + // 父目录未在缓存中（父目录可能本次未扫描到），降级为用户配置的同步根 cid， + // 由上传端尽力处理（可能失败记日志，不影响下载）。 + return st.p.RemotePath + } + return st.remoteUploadPath(rel) +} + // taskExists 检查是否已有同目录、同目标的进行中/已完成任务（避免重复入队）。 func (st *strmSyncState) taskExists(kind, syncPathID, localPath string) bool { ctx := st.ctx @@ -1109,6 +1182,14 @@ func (st *strmSyncState) taskExists(kind, syncPathID, localPath string) bool { // pruneLocal 清理本地多余 .strm 与元数据（远端已不存在），可选删除空目录。 func (st *strmSyncState) pruneLocal() error { + // 增量同步保护：本次远端扫描不完整（目录详情解析失败 / 文件父路径降级）时， + // seenVideo/seenMeta 覆盖不全，按"远端不存在"清理会误删刚下载或已存在的本地文件， + // 进而触发"下次增量重新下载"的循环。此时跳过清理，仅做进度落库。 + if st.syncType == model.StrmSyncTypeIncremental && st.scanIncomplete.Load() { + st.s.log.Warn("strm 增量同步跳过清理：本次远端扫描不完整，prune 已禁用", + zap.String("path_id", st.p.ID)) + return nil + } localRoot := filepath.Clean(st.p.LocalPath) var dirs []string err := filepath.WalkDir(localRoot, func(path string, d os.DirEntry, err error) error {