mirror of
https://github.com/rclone/rclone.git
synced 2024-11-25 09:41:44 +08:00
1780 lines
52 KiB
Go
1780 lines
52 KiB
Go
// Package pikpak provides an interface to the PikPak
|
|
package pikpak
|
|
|
|
// ------------------------------------------------------------
|
|
// NOTE
|
|
// ------------------------------------------------------------
|
|
|
|
// md5sum is not always available, sometimes given empty.
|
|
|
|
// Trashed files are not restored to the original location when using `batchUntrash`
|
|
|
|
// Can't stream without `--vfs-cache-mode=full`
|
|
|
|
// ------------------------------------------------------------
|
|
// TODO
|
|
// ------------------------------------------------------------
|
|
|
|
// * List() with options starred-only
|
|
// * user-configurable list chunk
|
|
// * backend command: untrash, iscached
|
|
// * api(event,task)
|
|
|
|
import (
|
|
"bytes"
|
|
"context"
|
|
"encoding/json"
|
|
"errors"
|
|
"fmt"
|
|
"io"
|
|
"math"
|
|
"net/http"
|
|
"net/url"
|
|
"path"
|
|
"reflect"
|
|
"strconv"
|
|
"strings"
|
|
"sync"
|
|
"time"
|
|
|
|
"github.com/aws/aws-sdk-go-v2/aws"
|
|
awsconfig "github.com/aws/aws-sdk-go-v2/config"
|
|
"github.com/aws/aws-sdk-go-v2/credentials"
|
|
"github.com/aws/aws-sdk-go-v2/feature/s3/manager"
|
|
"github.com/aws/aws-sdk-go-v2/service/s3"
|
|
"github.com/rclone/rclone/backend/pikpak/api"
|
|
"github.com/rclone/rclone/fs"
|
|
"github.com/rclone/rclone/fs/accounting"
|
|
"github.com/rclone/rclone/fs/chunksize"
|
|
"github.com/rclone/rclone/fs/config"
|
|
"github.com/rclone/rclone/fs/config/configmap"
|
|
"github.com/rclone/rclone/fs/config/configstruct"
|
|
"github.com/rclone/rclone/fs/config/obscure"
|
|
"github.com/rclone/rclone/fs/fserrors"
|
|
"github.com/rclone/rclone/fs/hash"
|
|
"github.com/rclone/rclone/lib/atexit"
|
|
"github.com/rclone/rclone/lib/dircache"
|
|
"github.com/rclone/rclone/lib/encoder"
|
|
"github.com/rclone/rclone/lib/oauthutil"
|
|
"github.com/rclone/rclone/lib/pacer"
|
|
"github.com/rclone/rclone/lib/random"
|
|
"github.com/rclone/rclone/lib/rest"
|
|
"golang.org/x/oauth2"
|
|
)
|
|
|
|
// Constants
|
|
const (
|
|
rcloneClientID = "YNxT9w7GMdWvEOKa"
|
|
rcloneEncryptedClientSecret = "aqrmB6M1YJ1DWCBxVxFSjFo7wzWEky494YMmkqgAl1do1WKOe2E"
|
|
minSleep = 100 * time.Millisecond
|
|
maxSleep = 2 * time.Second
|
|
taskWaitTime = 500 * time.Millisecond
|
|
decayConstant = 2 // bigger for slower decay, exponential
|
|
rootURL = "https://api-drive.mypikpak.com"
|
|
minChunkSize = fs.SizeSuffix(manager.MinUploadPartSize)
|
|
defaultUploadConcurrency = manager.DefaultUploadConcurrency
|
|
)
|
|
|
|
// Globals
|
|
var (
|
|
// Description of how to auth for this app
|
|
oauthConfig = &oauth2.Config{
|
|
Scopes: nil,
|
|
Endpoint: oauth2.Endpoint{
|
|
AuthURL: "https://user.mypikpak.com/v1/auth/signin",
|
|
TokenURL: "https://user.mypikpak.com/v1/auth/token",
|
|
AuthStyle: oauth2.AuthStyleInParams,
|
|
},
|
|
ClientID: rcloneClientID,
|
|
ClientSecret: obscure.MustReveal(rcloneEncryptedClientSecret),
|
|
RedirectURL: oauthutil.RedirectURL,
|
|
}
|
|
)
|
|
|
|
// Returns OAuthOptions modified for pikpak
|
|
func pikpakOAuthOptions() []fs.Option {
|
|
opts := []fs.Option{}
|
|
for _, opt := range oauthutil.SharedOptions {
|
|
if opt.Name == config.ConfigClientID {
|
|
opt.Advanced = true
|
|
} else if opt.Name == config.ConfigClientSecret {
|
|
opt.Advanced = true
|
|
}
|
|
opts = append(opts, opt)
|
|
}
|
|
return opts
|
|
}
|
|
|
|
// pikpakAutorize retrieves OAuth token using user/pass and save it to rclone.conf
|
|
func pikpakAuthorize(ctx context.Context, opt *Options, name string, m configmap.Mapper) error {
|
|
// override default client id/secret
|
|
if id, ok := m.Get("client_id"); ok && id != "" {
|
|
oauthConfig.ClientID = id
|
|
}
|
|
if secret, ok := m.Get("client_secret"); ok && secret != "" {
|
|
oauthConfig.ClientSecret = secret
|
|
}
|
|
pass, err := obscure.Reveal(opt.Password)
|
|
if err != nil {
|
|
return fmt.Errorf("failed to decode password - did you obscure it?: %w", err)
|
|
}
|
|
t, err := oauthConfig.PasswordCredentialsToken(ctx, opt.Username, pass)
|
|
if err != nil {
|
|
return fmt.Errorf("failed to retrieve token using username/password: %w", err)
|
|
}
|
|
return oauthutil.PutToken(name, m, t, false)
|
|
}
|
|
|
|
// Register with Fs
|
|
func init() {
|
|
fs.Register(&fs.RegInfo{
|
|
Name: "pikpak",
|
|
Description: "PikPak",
|
|
NewFs: NewFs,
|
|
CommandHelp: commandHelp,
|
|
Config: func(ctx context.Context, name string, m configmap.Mapper, config fs.ConfigIn) (*fs.ConfigOut, error) {
|
|
// Parse config into Options struct
|
|
opt := new(Options)
|
|
err := configstruct.Set(m, opt)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("couldn't parse config into struct: %w", err)
|
|
}
|
|
|
|
switch config.State {
|
|
case "":
|
|
// Check token exists
|
|
if _, err := oauthutil.GetToken(name, m); err != nil {
|
|
return fs.ConfigGoto("authorize")
|
|
}
|
|
return fs.ConfigConfirm("authorize_ok", false, "consent_to_authorize", "Re-authorize for new token?")
|
|
case "authorize_ok":
|
|
if config.Result == "false" {
|
|
return nil, nil
|
|
}
|
|
return fs.ConfigGoto("authorize")
|
|
case "authorize":
|
|
if err := pikpakAuthorize(ctx, opt, name, m); err != nil {
|
|
return nil, err
|
|
}
|
|
return nil, nil
|
|
}
|
|
return nil, fmt.Errorf("unknown state %q", config.State)
|
|
},
|
|
Options: append(pikpakOAuthOptions(), []fs.Option{{
|
|
Name: "user",
|
|
Help: "Pikpak username.",
|
|
Required: true,
|
|
Sensitive: true,
|
|
}, {
|
|
Name: "pass",
|
|
Help: "Pikpak password.",
|
|
Required: true,
|
|
IsPassword: true,
|
|
}, {
|
|
Name: "root_folder_id",
|
|
Help: `ID of the root folder.
|
|
Leave blank normally.
|
|
|
|
Fill in for rclone to use a non root folder as its starting point.
|
|
`,
|
|
Advanced: true,
|
|
Sensitive: true,
|
|
}, {
|
|
Name: "use_trash",
|
|
Default: true,
|
|
Help: "Send files to the trash instead of deleting permanently.\n\nDefaults to true, namely sending files to the trash.\nUse `--pikpak-use-trash=false` to delete files permanently instead.",
|
|
Advanced: true,
|
|
}, {
|
|
Name: "trashed_only",
|
|
Default: false,
|
|
Help: "Only show files that are in the trash.\n\nThis will show trashed files in their original directory structure.",
|
|
Advanced: true,
|
|
}, {
|
|
Name: "hash_memory_limit",
|
|
Help: "Files bigger than this will be cached on disk to calculate hash if required.",
|
|
Default: fs.SizeSuffix(10 * 1024 * 1024),
|
|
Advanced: true,
|
|
}, {
|
|
Name: "chunk_size",
|
|
Help: `Chunk size for multipart uploads.
|
|
|
|
Large files will be uploaded in chunks of this size.
|
|
|
|
Note that this is stored in memory and there may be up to
|
|
"--transfers" * "--pikpak-upload-concurrency" chunks stored at once
|
|
in memory.
|
|
|
|
If you are transferring large files over high-speed links and you have
|
|
enough memory, then increasing this will speed up the transfers.
|
|
|
|
Rclone will automatically increase the chunk size when uploading a
|
|
large file of known size to stay below the 10,000 chunks limit.
|
|
|
|
Increasing the chunk size decreases the accuracy of the progress
|
|
statistics displayed with "-P" flag.`,
|
|
Default: minChunkSize,
|
|
Advanced: true,
|
|
}, {
|
|
Name: "upload_concurrency",
|
|
Help: `Concurrency for multipart uploads.
|
|
|
|
This is the number of chunks of the same file that are uploaded
|
|
concurrently for multipart uploads.
|
|
|
|
Note that chunks are stored in memory and there may be up to
|
|
"--transfers" * "--pikpak-upload-concurrency" chunks stored at once
|
|
in memory.
|
|
|
|
If you are uploading small numbers of large files over high-speed links
|
|
and these uploads do not fully utilize your bandwidth, then increasing
|
|
this may help to speed up the transfers.`,
|
|
Default: defaultUploadConcurrency,
|
|
Advanced: true,
|
|
}, {
|
|
Name: config.ConfigEncoding,
|
|
Help: config.ConfigEncodingHelp,
|
|
Advanced: true,
|
|
Default: (encoder.EncodeCtl |
|
|
encoder.EncodeDot |
|
|
encoder.EncodeBackSlash |
|
|
encoder.EncodeSlash |
|
|
encoder.EncodeDoubleQuote |
|
|
encoder.EncodeAsterisk |
|
|
encoder.EncodeColon |
|
|
encoder.EncodeLtGt |
|
|
encoder.EncodeQuestion |
|
|
encoder.EncodePipe |
|
|
encoder.EncodeLeftSpace |
|
|
encoder.EncodeRightSpace |
|
|
encoder.EncodeRightPeriod |
|
|
encoder.EncodeInvalidUtf8),
|
|
}}...),
|
|
})
|
|
}
|
|
|
|
// Options defines the configuration for this backend
|
|
type Options struct {
|
|
Username string `config:"user"`
|
|
Password string `config:"pass"`
|
|
RootFolderID string `config:"root_folder_id"`
|
|
UseTrash bool `config:"use_trash"`
|
|
TrashedOnly bool `config:"trashed_only"`
|
|
HashMemoryThreshold fs.SizeSuffix `config:"hash_memory_limit"`
|
|
ChunkSize fs.SizeSuffix `config:"chunk_size"`
|
|
UploadConcurrency int `config:"upload_concurrency"`
|
|
Enc encoder.MultiEncoder `config:"encoding"`
|
|
}
|
|
|
|
// Fs represents a remote pikpak
|
|
type Fs struct {
|
|
name string // name of this remote
|
|
root string // the path we are working on
|
|
opt Options // parsed options
|
|
features *fs.Features // optional features
|
|
rst *rest.Client // the connection to the server
|
|
dirCache *dircache.DirCache // Map of directory path to directory id
|
|
pacer *fs.Pacer // pacer for API calls
|
|
rootFolderID string // the id of the root folder
|
|
deviceID string // device id used for api requests
|
|
client *http.Client // authorized client
|
|
m configmap.Mapper
|
|
tokenMu *sync.Mutex // when renewing tokens
|
|
}
|
|
|
|
// Object describes a pikpak object
|
|
type Object struct {
|
|
fs *Fs // what this object is part of
|
|
remote string // The remote path
|
|
hasMetaData bool // whether info below has been set
|
|
id string // ID of the object
|
|
size int64 // size of the object
|
|
modTime time.Time // modification time of the object
|
|
mimeType string // The object MIME type
|
|
parent string // ID of the parent directories
|
|
gcid string // custom hash of the object
|
|
md5sum string // md5sum of the object
|
|
link *api.Link // link to download the object
|
|
linkMu *sync.Mutex
|
|
}
|
|
|
|
// ------------------------------------------------------------
|
|
|
|
// Name of the remote (as passed into NewFs)
|
|
func (f *Fs) Name() string {
|
|
return f.name
|
|
}
|
|
|
|
// Root of the remote (as passed into NewFs)
|
|
func (f *Fs) Root() string {
|
|
return f.root
|
|
}
|
|
|
|
// String converts this Fs to a string
|
|
func (f *Fs) String() string {
|
|
return fmt.Sprintf("PikPak root '%s'", f.root)
|
|
}
|
|
|
|
// Features returns the optional features of this Fs
|
|
func (f *Fs) Features() *fs.Features {
|
|
return f.features
|
|
}
|
|
|
|
// Precision return the precision of this Fs
|
|
func (f *Fs) Precision() time.Duration {
|
|
return fs.ModTimeNotSupported
|
|
// meaning that the modification times from the backend shouldn't be used for syncing
|
|
// as they can't be set.
|
|
}
|
|
|
|
// DirCacheFlush resets the directory cache - used in testing as an
|
|
// optional interface
|
|
func (f *Fs) DirCacheFlush() {
|
|
f.dirCache.ResetRoot()
|
|
}
|
|
|
|
// Hashes returns the supported hash sets.
|
|
func (f *Fs) Hashes() hash.Set {
|
|
return hash.Set(hash.MD5)
|
|
}
|
|
|
|
// parsePath parses a remote path
|
|
func parsePath(path string) (root string) {
|
|
root = strings.Trim(path, "/")
|
|
return
|
|
}
|
|
|
|
// parentIDForRequest returns ParentId for api requests
|
|
func parentIDForRequest(dirID string) string {
|
|
if dirID == "root" {
|
|
return ""
|
|
}
|
|
return dirID
|
|
}
|
|
|
|
// retryErrorCodes is a slice of error codes that we will retry
|
|
var retryErrorCodes = []int{
|
|
429, // Too Many Requests.
|
|
500, // Internal Server Error
|
|
502, // Bad Gateway
|
|
503, // Service Unavailable
|
|
504, // Gateway Timeout
|
|
509, // Bandwidth Limit Exceeded
|
|
}
|
|
|
|
// reAuthorize re-authorize oAuth token during runtime
|
|
func (f *Fs) reAuthorize(ctx context.Context) (err error) {
|
|
f.tokenMu.Lock()
|
|
defer f.tokenMu.Unlock()
|
|
|
|
if err := pikpakAuthorize(ctx, &f.opt, f.name, f.m); err != nil {
|
|
return err
|
|
}
|
|
return f.newClientWithPacer(ctx)
|
|
}
|
|
|
|
// shouldRetry returns a boolean as to whether this resp and err
|
|
// deserve to be retried. It returns the err as a convenience
|
|
func (f *Fs) shouldRetry(ctx context.Context, resp *http.Response, err error) (bool, error) {
|
|
if fserrors.ContextError(ctx, &err) {
|
|
return false, err
|
|
}
|
|
if err == nil {
|
|
return false, nil
|
|
}
|
|
if fserrors.ShouldRetry(err) {
|
|
return true, err
|
|
}
|
|
authRetry := false
|
|
|
|
// traceback to possible api.Error wrapped in err, and re-authorize if necessary
|
|
// "unauthenticated" (16): when access_token is invalid, but should be handled by oauthutil
|
|
var terr *oauth2.RetrieveError
|
|
if errors.As(err, &terr) {
|
|
apiErr := new(api.Error)
|
|
if err := json.Unmarshal(terr.Body, apiErr); err == nil {
|
|
if apiErr.Reason == "invalid_grant" {
|
|
// "invalid_grant" (4126): The refresh token is incorrect or expired //
|
|
// Invalid refresh token. It may have been refreshed by another process.
|
|
authRetry = true
|
|
}
|
|
}
|
|
}
|
|
// Once err was processed by maybeWrapOAuthError() in lib/oauthutil,
|
|
// the above code is no longer sufficient to handle the 'invalid_grant' error.
|
|
if strings.Contains(err.Error(), "invalid_grant") {
|
|
authRetry = true
|
|
}
|
|
|
|
if authRetry {
|
|
if authErr := f.reAuthorize(ctx); authErr != nil {
|
|
return false, fserrors.FatalError(authErr)
|
|
}
|
|
}
|
|
|
|
switch apiErr := err.(type) {
|
|
case *api.Error:
|
|
if apiErr.Reason == "file_rename_uncompleted" {
|
|
// "file_rename_uncompleted" (9): Renaming uncompleted file or folder is not supported
|
|
// This error occurs when you attempt to rename objects
|
|
// right after some server-side changes, e.g. DirMove, Move, Copy
|
|
return true, err
|
|
} else if apiErr.Reason == "file_duplicated_name" {
|
|
// "file_duplicated_name" (3): File name cannot be repeated
|
|
// This error may occur when attempting to rename temp object (newly uploaded)
|
|
// right after the old one is removed.
|
|
return true, err
|
|
} else if apiErr.Reason == "task_daily_create_limit_vip" {
|
|
// "task_daily_create_limit_vip" (11): Sorry, you have submitted too many tasks and have exceeded the current processing capacity, please try again tomorrow
|
|
return false, fserrors.FatalError(err)
|
|
} else if apiErr.Reason == "file_space_not_enough" {
|
|
// "file_space_not_enough" (8): Storage space is not enough
|
|
return false, fserrors.FatalError(err)
|
|
}
|
|
}
|
|
|
|
return authRetry || fserrors.ShouldRetryHTTP(resp, retryErrorCodes), err
|
|
}
|
|
|
|
// errorHandler parses a non 2xx error response into an error
|
|
func errorHandler(resp *http.Response) error {
|
|
// Decode error response
|
|
errResponse := new(api.Error)
|
|
err := rest.DecodeJSON(resp, &errResponse)
|
|
if err != nil {
|
|
fs.Debugf(nil, "Couldn't decode error response: %v", err)
|
|
}
|
|
if errResponse.Reason == "" {
|
|
errResponse.Reason = resp.Status
|
|
}
|
|
if errResponse.Code == 0 {
|
|
errResponse.Code = resp.StatusCode
|
|
}
|
|
return errResponse
|
|
}
|
|
|
|
// newClientWithPacer sets a new http/rest client with a pacer to Fs
|
|
func (f *Fs) newClientWithPacer(ctx context.Context) (err error) {
|
|
f.client, _, err = oauthutil.NewClient(ctx, f.name, f.m, oauthConfig)
|
|
if err != nil {
|
|
return fmt.Errorf("failed to create oauth client: %w", err)
|
|
}
|
|
f.rst = rest.NewClient(f.client).SetRoot(rootURL).SetErrorHandler(errorHandler)
|
|
f.pacer = fs.NewPacer(ctx, pacer.NewDefault(pacer.MinSleep(minSleep), pacer.MaxSleep(maxSleep), pacer.DecayConstant(decayConstant)))
|
|
return nil
|
|
}
|
|
|
|
// newFs partially constructs Fs from the path
|
|
//
|
|
// It constructs a valid Fs but doesn't attempt to figure out whether
|
|
// it is a file or a directory.
|
|
func newFs(ctx context.Context, name, path string, m configmap.Mapper) (*Fs, error) {
|
|
// Parse config into Options struct
|
|
opt := new(Options)
|
|
if err := configstruct.Set(m, opt); err != nil {
|
|
return nil, err
|
|
}
|
|
if opt.ChunkSize < minChunkSize {
|
|
return nil, fmt.Errorf("chunk size must be at least %s", minChunkSize)
|
|
}
|
|
|
|
root := parsePath(path)
|
|
|
|
f := &Fs{
|
|
name: name,
|
|
root: root,
|
|
opt: *opt,
|
|
m: m,
|
|
tokenMu: new(sync.Mutex),
|
|
}
|
|
f.features = (&fs.Features{
|
|
ReadMimeType: true, // can read the mime type of objects
|
|
CanHaveEmptyDirectories: true, // can have empty directories
|
|
NoMultiThreading: true, // can't have multiple threads downloading
|
|
}).Fill(ctx, f)
|
|
f.deviceID = genDeviceID()
|
|
|
|
if err := f.newClientWithPacer(ctx); err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
return f, nil
|
|
}
|
|
|
|
// NewFs constructs an Fs from the path, container:path
|
|
func NewFs(ctx context.Context, name, path string, m configmap.Mapper) (fs.Fs, error) {
|
|
f, err := newFs(ctx, name, path, m)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
// Set the root folder ID
|
|
if f.opt.RootFolderID != "" {
|
|
// use root_folder ID if set
|
|
f.rootFolderID = f.opt.RootFolderID
|
|
} else {
|
|
// pseudo-root
|
|
f.rootFolderID = "root"
|
|
}
|
|
|
|
f.dirCache = dircache.New(f.root, f.rootFolderID, f)
|
|
|
|
// Find the current root
|
|
err = f.dirCache.FindRoot(ctx, false)
|
|
if err != nil {
|
|
// Assume it is a file
|
|
newRoot, remote := dircache.SplitPath(f.root)
|
|
tempF := *f
|
|
tempF.dirCache = dircache.New(newRoot, f.rootFolderID, &tempF)
|
|
tempF.root = newRoot
|
|
// Make new Fs which is the parent
|
|
err = tempF.dirCache.FindRoot(ctx, false)
|
|
if err != nil {
|
|
// No root so return old f
|
|
return f, nil
|
|
}
|
|
_, err := tempF.NewObject(ctx, remote)
|
|
if err != nil {
|
|
if err == fs.ErrorObjectNotFound {
|
|
// File doesn't exist so return old f
|
|
return f, nil
|
|
}
|
|
return nil, err
|
|
}
|
|
f.features.Fill(ctx, &tempF)
|
|
// XXX: update the old f here instead of returning tempF, since
|
|
// `features` were already filled with functions having *f as a receiver.
|
|
// See https://github.com/rclone/rclone/issues/2182
|
|
f.dirCache = tempF.dirCache
|
|
f.root = tempF.root
|
|
// return an error with an fs which points to the parent
|
|
return f, fs.ErrorIsFile
|
|
}
|
|
return f, nil
|
|
}
|
|
|
|
// readMetaDataForPath reads the metadata from the path
|
|
func (f *Fs) readMetaDataForPath(ctx context.Context, path string) (info *api.File, err error) {
|
|
leaf, dirID, err := f.dirCache.FindPath(ctx, path, false)
|
|
if err != nil {
|
|
if err == fs.ErrorDirNotFound {
|
|
return nil, fs.ErrorObjectNotFound
|
|
}
|
|
return nil, err
|
|
}
|
|
|
|
// checking whether fileObj with name of leaf exists in dirID
|
|
trashed := "false"
|
|
if f.opt.TrashedOnly {
|
|
trashed = "true"
|
|
}
|
|
found, err := f.listAll(ctx, dirID, api.KindOfFile, trashed, func(item *api.File) bool {
|
|
if item.Name == leaf {
|
|
info = item
|
|
return true
|
|
}
|
|
return false
|
|
})
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
if !found {
|
|
return nil, fs.ErrorObjectNotFound
|
|
}
|
|
return info, nil
|
|
}
|
|
|
|
// Return an Object from a path
|
|
//
|
|
// If it can't be found it returns the error fs.ErrorObjectNotFound.
|
|
func (f *Fs) newObjectWithInfo(ctx context.Context, remote string, info *api.File) (fs.Object, error) {
|
|
o := &Object{
|
|
fs: f,
|
|
remote: remote,
|
|
linkMu: new(sync.Mutex),
|
|
}
|
|
var err error
|
|
if info != nil {
|
|
err = o.setMetaData(info)
|
|
} else {
|
|
err = o.readMetaData(ctx) // reads info and meta, returning an error
|
|
}
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
return o, nil
|
|
}
|
|
|
|
// NewObject finds the Object at remote. If it can't be found
|
|
// it returns the error fs.ErrorObjectNotFound.
|
|
func (f *Fs) NewObject(ctx context.Context, remote string) (fs.Object, error) {
|
|
return f.newObjectWithInfo(ctx, remote, nil)
|
|
}
|
|
|
|
// FindLeaf finds a directory of name leaf in the folder with ID pathID
|
|
func (f *Fs) FindLeaf(ctx context.Context, pathID, leaf string) (pathIDOut string, found bool, err error) {
|
|
// Find the leaf in pathID
|
|
trashed := "false"
|
|
if f.opt.TrashedOnly {
|
|
// still need to list folders
|
|
trashed = ""
|
|
}
|
|
found, err = f.listAll(ctx, pathID, api.KindOfFolder, trashed, func(item *api.File) bool {
|
|
if item.Name == leaf {
|
|
pathIDOut = item.ID
|
|
return true
|
|
}
|
|
return false
|
|
})
|
|
return pathIDOut, found, err
|
|
}
|
|
|
|
// list the objects into the function supplied
|
|
//
|
|
// If directories is set it only sends directories
|
|
// User function to process a File item from listAll
|
|
//
|
|
// Should return true to finish processing
|
|
type listAllFn func(*api.File) bool
|
|
|
|
// Lists the directory required calling the user function on each item found
|
|
//
|
|
// If the user fn ever returns true then it early exits with found = true
|
|
func (f *Fs) listAll(ctx context.Context, dirID, kind, trashed string, fn listAllFn) (found bool, err error) {
|
|
// Url Parameters
|
|
params := url.Values{}
|
|
params.Set("thumbnail_size", api.ThumbnailSizeM)
|
|
params.Set("limit", strconv.Itoa(api.ListLimit))
|
|
params.Set("with_audit", strconv.FormatBool(true))
|
|
if parentID := parentIDForRequest(dirID); parentID != "" {
|
|
params.Set("parent_id", parentID)
|
|
}
|
|
|
|
// Construct filter string
|
|
filters := &api.Filters{}
|
|
filters.Set("Phase", "eq", api.PhaseTypeComplete)
|
|
filters.Set("Trashed", "eq", trashed)
|
|
filters.Set("Kind", "eq", kind)
|
|
if filterStr, err := json.Marshal(filters); err == nil {
|
|
params.Set("filters", string(filterStr))
|
|
}
|
|
// fs.Debugf(f, "list params: %v", params)
|
|
|
|
opts := rest.Opts{
|
|
Method: "GET",
|
|
Path: "/drive/v1/files",
|
|
Parameters: params,
|
|
}
|
|
|
|
pageToken := ""
|
|
OUTER:
|
|
for {
|
|
opts.Parameters.Set("page_token", pageToken)
|
|
|
|
var info api.FileList
|
|
var resp *http.Response
|
|
err = f.pacer.Call(func() (bool, error) {
|
|
resp, err = f.rst.CallJSON(ctx, &opts, nil, &info)
|
|
return f.shouldRetry(ctx, resp, err)
|
|
})
|
|
if err != nil {
|
|
return found, fmt.Errorf("couldn't list files: %w", err)
|
|
}
|
|
if len(info.Files) == 0 {
|
|
break
|
|
}
|
|
for _, item := range info.Files {
|
|
item.Name = f.opt.Enc.ToStandardName(item.Name)
|
|
if fn(item) {
|
|
found = true
|
|
break OUTER
|
|
}
|
|
}
|
|
if info.NextPageToken == "" {
|
|
break
|
|
}
|
|
pageToken = info.NextPageToken
|
|
}
|
|
return
|
|
}
|
|
|
|
// itemToDirEntry converts a api.File to an fs.DirEntry.
|
|
// When the api.File cannot be represented as an fs.DirEntry
|
|
// (nil, nil) is returned.
|
|
func (f *Fs) itemToDirEntry(ctx context.Context, remote string, item *api.File) (entry fs.DirEntry, err error) {
|
|
switch {
|
|
case item.Kind == api.KindOfFolder:
|
|
// cache the directory ID for later lookups
|
|
f.dirCache.Put(remote, item.ID)
|
|
d := fs.NewDir(remote, time.Time(item.ModifiedTime)).SetID(item.ID)
|
|
if item.ParentID == "" {
|
|
d.SetParentID("root")
|
|
} else {
|
|
d.SetParentID(item.ParentID)
|
|
}
|
|
return d, nil
|
|
case f.opt.TrashedOnly && !item.Trashed:
|
|
// ignore object
|
|
default:
|
|
entry, err = f.newObjectWithInfo(ctx, remote, item)
|
|
if err == fs.ErrorObjectNotFound {
|
|
return nil, nil
|
|
}
|
|
return entry, err
|
|
}
|
|
return nil, nil
|
|
}
|
|
|
|
// List the objects and directories in dir into entries. The
|
|
// entries can be returned in any order but should be for a
|
|
// complete directory.
|
|
//
|
|
// dir should be "" to list the root, and should not have
|
|
// trailing slashes.
|
|
//
|
|
// This should return ErrDirNotFound if the directory isn't
|
|
// found.
|
|
func (f *Fs) List(ctx context.Context, dir string) (entries fs.DirEntries, err error) {
|
|
// fs.Debugf(f, "List(%q)\n", dir)
|
|
dirID, err := f.dirCache.FindDir(ctx, dir, false)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
var iErr error
|
|
trashed := "false"
|
|
if f.opt.TrashedOnly {
|
|
// still need to list folders
|
|
trashed = ""
|
|
}
|
|
_, err = f.listAll(ctx, dirID, "", trashed, func(item *api.File) bool {
|
|
entry, err := f.itemToDirEntry(ctx, path.Join(dir, item.Name), item)
|
|
if err != nil {
|
|
iErr = err
|
|
return true
|
|
}
|
|
if entry != nil {
|
|
entries = append(entries, entry)
|
|
}
|
|
return false
|
|
})
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
if iErr != nil {
|
|
return nil, iErr
|
|
}
|
|
return entries, nil
|
|
}
|
|
|
|
// CreateDir makes a directory with pathID as parent and name leaf
|
|
func (f *Fs) CreateDir(ctx context.Context, pathID, leaf string) (newID string, err error) {
|
|
// fs.Debugf(f, "CreateDir(%q, %q)\n", pathID, leaf)
|
|
req := api.RequestNewFile{
|
|
Name: f.opt.Enc.FromStandardName(leaf),
|
|
Kind: api.KindOfFolder,
|
|
ParentID: parentIDForRequest(pathID),
|
|
}
|
|
info, err := f.requestNewFile(ctx, &req)
|
|
if err != nil {
|
|
return "", err
|
|
}
|
|
return info.File.ID, nil
|
|
}
|
|
|
|
// Mkdir creates the container if it doesn't exist
|
|
func (f *Fs) Mkdir(ctx context.Context, dir string) error {
|
|
_, err := f.dirCache.FindDir(ctx, dir, true)
|
|
return err
|
|
}
|
|
|
|
// About gets quota information
|
|
func (f *Fs) About(ctx context.Context) (usage *fs.Usage, err error) {
|
|
info, err := f.getAbout(ctx)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("failed to get drive quota: %w", err)
|
|
}
|
|
q := info.Quota
|
|
usage = &fs.Usage{
|
|
Used: fs.NewUsageValue(q.Usage), // bytes in use
|
|
// Trashed: fs.NewUsageValue(q.UsageInTrash), // bytes in trash but this seems not working
|
|
}
|
|
if q.Limit > 0 {
|
|
usage.Total = fs.NewUsageValue(q.Limit) // quota of bytes that can be used
|
|
usage.Free = fs.NewUsageValue(q.Limit - q.Usage) // bytes which can be uploaded before reaching the quota
|
|
}
|
|
return usage, nil
|
|
}
|
|
|
|
// PublicLink adds a "readable by anyone with link" permission on the given file or folder.
|
|
func (f *Fs) PublicLink(ctx context.Context, remote string, expire fs.Duration, unlink bool) (string, error) {
|
|
id, err := f.dirCache.FindDir(ctx, remote, false)
|
|
if err == nil {
|
|
fs.Debugf(f, "attempting to share directory '%s'", remote)
|
|
} else {
|
|
fs.Debugf(f, "attempting to share single file '%s'", remote)
|
|
o, err := f.NewObject(ctx, remote)
|
|
if err != nil {
|
|
return "", err
|
|
}
|
|
id = o.(*Object).id
|
|
}
|
|
expiry := -1
|
|
if expire < fs.DurationOff {
|
|
expiry = int(math.Ceil(time.Duration(expire).Hours() / 24))
|
|
}
|
|
req := api.RequestShare{
|
|
FileIDs: []string{id},
|
|
ShareTo: "publiclink",
|
|
ExpirationDays: expiry,
|
|
PassCodeOption: "NOT_REQUIRED",
|
|
}
|
|
info, err := f.requestShare(ctx, &req)
|
|
if err != nil {
|
|
return "", err
|
|
}
|
|
return info.ShareURL, err
|
|
}
|
|
|
|
// delete a file or directory by ID w/o using trash
|
|
func (f *Fs) deleteObjects(ctx context.Context, IDs []string, useTrash bool) (err error) {
|
|
if len(IDs) == 0 {
|
|
return nil
|
|
}
|
|
action := "batchDelete"
|
|
if useTrash {
|
|
action = "batchTrash"
|
|
}
|
|
req := api.RequestBatch{
|
|
IDs: IDs,
|
|
}
|
|
if err := f.requestBatchAction(ctx, action, &req); err != nil {
|
|
return fmt.Errorf("delete object failed: %w", err)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// purgeCheck removes the root directory, if check is set then it
|
|
// refuses to do so if it has anything in
|
|
func (f *Fs) purgeCheck(ctx context.Context, dir string, check bool) error {
|
|
root := path.Join(f.root, dir)
|
|
if root == "" {
|
|
return errors.New("can't purge root directory")
|
|
}
|
|
rootID, err := f.dirCache.FindDir(ctx, dir, false)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
trashedFiles := false
|
|
if check {
|
|
found, err := f.listAll(ctx, rootID, "", "", func(item *api.File) bool {
|
|
if !item.Trashed {
|
|
fs.Debugf(dir, "Rmdir: contains file: %q", item.Name)
|
|
return true
|
|
}
|
|
fs.Debugf(dir, "Rmdir: contains trashed file: %q", item.Name)
|
|
trashedFiles = true
|
|
return false
|
|
})
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if found {
|
|
return fs.ErrorDirectoryNotEmpty
|
|
}
|
|
}
|
|
if root != "" {
|
|
// trash the directory if it had trashed files
|
|
// in or the user wants to trash, otherwise
|
|
// delete it.
|
|
err = f.deleteObjects(ctx, []string{rootID}, trashedFiles || f.opt.UseTrash)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
} else if check {
|
|
return errors.New("can't purge root directory")
|
|
}
|
|
f.dirCache.FlushDir(dir)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// Rmdir deletes the root folder
|
|
//
|
|
// Returns an error if it isn't empty
|
|
func (f *Fs) Rmdir(ctx context.Context, dir string) error {
|
|
return f.purgeCheck(ctx, dir, true)
|
|
}
|
|
|
|
// Purge deletes all the files and the container
|
|
//
|
|
// Optional interface: Only implement this if you have a way of
|
|
// deleting all the files quicker than just running Remove() on the
|
|
// result of List()
|
|
func (f *Fs) Purge(ctx context.Context, dir string) error {
|
|
return f.purgeCheck(ctx, dir, false)
|
|
}
|
|
|
|
// CleanUp empties the trash
|
|
func (f *Fs) CleanUp(ctx context.Context) (err error) {
|
|
opts := rest.Opts{
|
|
Method: "PATCH",
|
|
Path: "/drive/v1/files/trash:empty",
|
|
}
|
|
info := struct {
|
|
TaskID string `json:"task_id"`
|
|
}{}
|
|
var resp *http.Response
|
|
err = f.pacer.Call(func() (bool, error) {
|
|
resp, err = f.rst.CallJSON(ctx, &opts, nil, &info)
|
|
return f.shouldRetry(ctx, resp, err)
|
|
})
|
|
if err != nil {
|
|
return fmt.Errorf("couldn't empty trash: %w", err)
|
|
}
|
|
return f.waitTask(ctx, info.TaskID)
|
|
}
|
|
|
|
// Move the object
|
|
func (f *Fs) moveObjects(ctx context.Context, IDs []string, dirID string) (err error) {
|
|
if len(IDs) == 0 {
|
|
return nil
|
|
}
|
|
req := api.RequestBatch{
|
|
IDs: IDs,
|
|
To: map[string]string{"parent_id": parentIDForRequest(dirID)},
|
|
}
|
|
if err := f.requestBatchAction(ctx, "batchMove", &req); err != nil {
|
|
return fmt.Errorf("move object failed: %w", err)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// renames the object
|
|
func (f *Fs) renameObject(ctx context.Context, ID, newName string) (info *api.File, err error) {
|
|
req := api.File{
|
|
Name: f.opt.Enc.FromStandardName(newName),
|
|
}
|
|
info, err = f.patchFile(ctx, ID, &req)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("rename object failed: %w", err)
|
|
}
|
|
return
|
|
}
|
|
|
|
// DirMove moves src, srcRemote to this remote at dstRemote
|
|
// using server-side move operations.
|
|
//
|
|
// Will only be called if src.Fs().Name() == f.Name()
|
|
//
|
|
// If it isn't possible then return fs.ErrorCantDirMove
|
|
//
|
|
// If destination exists then return fs.ErrorDirExists
|
|
func (f *Fs) DirMove(ctx context.Context, src fs.Fs, srcRemote, dstRemote string) error {
|
|
srcFs, ok := src.(*Fs)
|
|
if !ok {
|
|
fs.Debugf(srcFs, "Can't move directory - not same remote type")
|
|
return fs.ErrorCantDirMove
|
|
}
|
|
|
|
srcID, srcParentID, srcLeaf, dstParentID, dstLeaf, err := f.dirCache.DirMove(ctx, srcFs.dirCache, srcFs.root, srcRemote, f.root, dstRemote)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
if srcParentID != dstParentID {
|
|
// Do the move
|
|
err = f.moveObjects(ctx, []string{srcID}, dstParentID)
|
|
if err != nil {
|
|
return fmt.Errorf("couldn't dir move: %w", err)
|
|
}
|
|
}
|
|
|
|
// Can't copy and change name in one step so we have to check if we have
|
|
// the correct name after copy
|
|
if srcLeaf != dstLeaf {
|
|
_, err = f.renameObject(ctx, srcID, dstLeaf)
|
|
if err != nil {
|
|
return fmt.Errorf("dirmove: couldn't rename moved dir: %w", err)
|
|
}
|
|
}
|
|
srcFs.dirCache.FlushDir(srcRemote)
|
|
return nil
|
|
}
|
|
|
|
// Creates from the parameters passed in a half finished Object which
|
|
// must have setMetaData called on it
|
|
//
|
|
// Returns the object, leaf, dirID and error.
|
|
//
|
|
// Used to create new objects
|
|
func (f *Fs) createObject(ctx context.Context, remote string, modTime time.Time, size int64) (o *Object, leaf string, dirID string, err error) {
|
|
// Create the directory for the object if it doesn't exist
|
|
leaf, dirID, err = f.dirCache.FindPath(ctx, remote, true)
|
|
if err != nil {
|
|
return
|
|
}
|
|
// Temporary Object under construction
|
|
o = &Object{
|
|
fs: f,
|
|
remote: remote,
|
|
parent: dirID,
|
|
size: size,
|
|
modTime: modTime,
|
|
linkMu: new(sync.Mutex),
|
|
}
|
|
return o, leaf, dirID, nil
|
|
}
|
|
|
|
// Move src to this remote using server-side move operations.
|
|
//
|
|
// This is stored with the remote path given.
|
|
//
|
|
// It returns the destination Object and a possible error.
|
|
//
|
|
// Will only be called if src.Fs().Name() == f.Name()
|
|
//
|
|
// If it isn't possible then return fs.ErrorCantMove
|
|
func (f *Fs) Move(ctx context.Context, src fs.Object, remote string) (fs.Object, error) {
|
|
srcObj, ok := src.(*Object)
|
|
if !ok {
|
|
fs.Debugf(src, "Can't move - not same remote type")
|
|
return nil, fs.ErrorCantMove
|
|
}
|
|
err := srcObj.readMetaData(ctx)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
srcLeaf, srcParentID, err := srcObj.fs.dirCache.FindPath(ctx, srcObj.remote, false)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
// Create temporary object - still missing id, mimeType, gcid, md5sum
|
|
dstObj, dstLeaf, dstParentID, err := f.createObject(ctx, remote, srcObj.modTime, srcObj.size)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
if srcParentID != dstParentID {
|
|
// Do the move
|
|
if err = f.moveObjects(ctx, []string{srcObj.id}, dstParentID); err != nil {
|
|
return nil, err
|
|
}
|
|
}
|
|
// Manually update info of moved object to save API calls
|
|
dstObj.id = srcObj.id
|
|
dstObj.mimeType = srcObj.mimeType
|
|
dstObj.gcid = srcObj.gcid
|
|
dstObj.md5sum = srcObj.md5sum
|
|
dstObj.hasMetaData = true
|
|
|
|
if srcLeaf != dstLeaf {
|
|
// Rename
|
|
info, err := f.renameObject(ctx, srcObj.id, dstLeaf)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("move: couldn't rename moved file: %w", err)
|
|
}
|
|
return dstObj, dstObj.setMetaData(info)
|
|
}
|
|
return dstObj, nil
|
|
}
|
|
|
|
// copy objects
|
|
func (f *Fs) copyObjects(ctx context.Context, IDs []string, dirID string) (err error) {
|
|
if len(IDs) == 0 {
|
|
return nil
|
|
}
|
|
req := api.RequestBatch{
|
|
IDs: IDs,
|
|
To: map[string]string{"parent_id": parentIDForRequest(dirID)},
|
|
}
|
|
if err := f.requestBatchAction(ctx, "batchCopy", &req); err != nil {
|
|
return fmt.Errorf("copy object failed: %w", err)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// Copy src to this remote using server side copy operations.
|
|
//
|
|
// This is stored with the remote path given.
|
|
//
|
|
// It returns the destination Object and a possible error.
|
|
//
|
|
// Will only be called if src.Fs().Name() == f.Name()
|
|
//
|
|
// If it isn't possible then return fs.ErrorCantCopy
|
|
func (f *Fs) Copy(ctx context.Context, src fs.Object, remote string) (fs.Object, error) {
|
|
srcObj, ok := src.(*Object)
|
|
if !ok {
|
|
fs.Debugf(src, "Can't copy - not same remote type")
|
|
return nil, fs.ErrorCantCopy
|
|
}
|
|
err := srcObj.readMetaData(ctx)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
// Create temporary object - still missing id, mimeType, gcid, md5sum
|
|
dstObj, dstLeaf, dstParentID, err := f.createObject(ctx, remote, srcObj.modTime, srcObj.size)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
if srcObj.parent == dstParentID {
|
|
// api restriction
|
|
fs.Debugf(src, "Can't copy - same parent")
|
|
return nil, fs.ErrorCantCopy
|
|
}
|
|
// Copy the object
|
|
if err := f.copyObjects(ctx, []string{srcObj.id}, dstParentID); err != nil {
|
|
return nil, fmt.Errorf("couldn't copy file: %w", err)
|
|
}
|
|
// Update info of the copied object with new parent but source name
|
|
if info, err := dstObj.fs.readMetaDataForPath(ctx, srcObj.remote); err != nil {
|
|
return nil, fmt.Errorf("copy: couldn't locate copied file: %w", err)
|
|
} else if err = dstObj.setMetaData(info); err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
// Can't copy and change name in one step so we have to check if we have
|
|
// the correct name after copy
|
|
srcLeaf, _, err := srcObj.fs.dirCache.FindPath(ctx, srcObj.remote, false)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
if srcLeaf != dstLeaf {
|
|
// Rename
|
|
info, err := f.renameObject(ctx, dstObj.id, dstLeaf)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("copy: couldn't rename copied file: %w", err)
|
|
}
|
|
return dstObj, dstObj.setMetaData(info)
|
|
}
|
|
return dstObj, nil
|
|
}
|
|
|
|
func (f *Fs) uploadByForm(ctx context.Context, in io.Reader, name string, size int64, form *api.Form, options ...fs.OpenOption) (err error) {
|
|
// struct to map. transferring values from MultParts to url parameter
|
|
params := url.Values{}
|
|
iVal := reflect.ValueOf(&form.MultiParts).Elem()
|
|
iTyp := iVal.Type()
|
|
for i := 0; i < iVal.NumField(); i++ {
|
|
params.Set(iTyp.Field(i).Tag.Get("json"), iVal.Field(i).String())
|
|
}
|
|
formReader, contentType, overhead, err := rest.MultipartUpload(ctx, in, params, "file", name)
|
|
if err != nil {
|
|
return fmt.Errorf("failed to make multipart upload: %w", err)
|
|
}
|
|
|
|
contentLength := overhead + size
|
|
opts := rest.Opts{
|
|
Method: form.Method,
|
|
RootURL: form.URL,
|
|
Body: formReader,
|
|
ContentType: contentType,
|
|
ContentLength: &contentLength,
|
|
Options: options,
|
|
TransferEncoding: []string{"identity"},
|
|
NoResponse: true,
|
|
}
|
|
|
|
var resp *http.Response
|
|
err = f.pacer.CallNoRetry(func() (bool, error) {
|
|
resp, err = f.rst.Call(ctx, &opts)
|
|
return f.shouldRetry(ctx, resp, err)
|
|
})
|
|
return
|
|
}
|
|
|
|
func (f *Fs) uploadByResumable(ctx context.Context, in io.Reader, name string, size int64, resumable *api.Resumable) (err error) {
|
|
p := resumable.Params
|
|
|
|
// Create a credentials provider
|
|
creds := credentials.NewStaticCredentialsProvider(p.AccessKeyID, p.AccessKeySecret, p.SecurityToken)
|
|
|
|
cfg, err := awsconfig.LoadDefaultConfig(ctx,
|
|
awsconfig.WithCredentialsProvider(creds),
|
|
awsconfig.WithRegion("pikpak"))
|
|
if err != nil {
|
|
return
|
|
}
|
|
|
|
client := s3.NewFromConfig(cfg, func(o *s3.Options) {
|
|
o.BaseEndpoint = aws.String("https://mypikpak.com/")
|
|
})
|
|
partSize := chunksize.Calculator(name, size, int(manager.MaxUploadParts), f.opt.ChunkSize)
|
|
|
|
// Create an uploader with custom options
|
|
uploader := manager.NewUploader(client, func(u *manager.Uploader) {
|
|
u.PartSize = int64(partSize)
|
|
u.Concurrency = f.opt.UploadConcurrency
|
|
})
|
|
// Perform an upload
|
|
_, err = uploader.Upload(ctx, &s3.PutObjectInput{
|
|
Bucket: &p.Bucket,
|
|
Key: &p.Key,
|
|
Body: in,
|
|
})
|
|
return
|
|
}
|
|
|
|
func (f *Fs) upload(ctx context.Context, in io.Reader, leaf, dirID, gcid string, size int64, options ...fs.OpenOption) (info *api.File, err error) {
|
|
// determine upload type
|
|
uploadType := api.UploadTypeResumable
|
|
// if size >= 0 && size < int64(5*fs.Mebi) {
|
|
// uploadType = api.UploadTypeForm
|
|
// }
|
|
// stop using uploadByForm() cause it is not as reliable as uploadByResumable() for a large number of small files
|
|
|
|
// request upload ticket to API
|
|
req := api.RequestNewFile{
|
|
Kind: api.KindOfFile,
|
|
Name: f.opt.Enc.FromStandardName(leaf),
|
|
ParentID: parentIDForRequest(dirID),
|
|
FolderType: "NORMAL",
|
|
Size: size,
|
|
Hash: strings.ToUpper(gcid),
|
|
UploadType: uploadType,
|
|
}
|
|
if uploadType == api.UploadTypeResumable {
|
|
req.Resumable = map[string]string{"provider": "PROVIDER_ALIYUN"}
|
|
}
|
|
new, err := f.requestNewFile(ctx, &req)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("failed to create a new file: %w", err)
|
|
}
|
|
if new.File == nil {
|
|
return nil, fmt.Errorf("invalid response: %+v", new)
|
|
} else if new.File.Phase == api.PhaseTypeComplete {
|
|
// early return; in case of zero-byte objects
|
|
if acc, ok := in.(*accounting.Account); ok && acc != nil {
|
|
// if `in io.Reader` is still in type of `*accounting.Account` (meaning that it is unused)
|
|
// it is considered as a server side copy as no incoming/outgoing traffic occur at all
|
|
acc.ServerSideTransferStart()
|
|
acc.ServerSideCopyEnd(size)
|
|
}
|
|
return new.File, nil
|
|
}
|
|
|
|
defer atexit.OnError(&err, func() {
|
|
fs.Debugf(leaf, "canceling upload: %v", err)
|
|
if cancelErr := f.deleteObjects(ctx, []string{new.File.ID}, false); cancelErr != nil {
|
|
fs.Logf(leaf, "failed to cancel upload: %v", cancelErr)
|
|
}
|
|
if cancelErr := f.deleteTask(ctx, new.Task.ID, false); cancelErr != nil {
|
|
fs.Logf(leaf, "failed to cancel upload: %v", cancelErr)
|
|
}
|
|
fs.Debugf(leaf, "waiting %v for the cancellation to be effective", taskWaitTime)
|
|
time.Sleep(taskWaitTime)
|
|
})()
|
|
|
|
if uploadType == api.UploadTypeForm && new.Form != nil {
|
|
err = f.uploadByForm(ctx, in, req.Name, size, new.Form, options...)
|
|
} else if uploadType == api.UploadTypeResumable && new.Resumable != nil {
|
|
err = f.uploadByResumable(ctx, in, leaf, size, new.Resumable)
|
|
} else {
|
|
err = fmt.Errorf("no method available for uploading: %+v", new)
|
|
}
|
|
|
|
if err != nil {
|
|
return nil, fmt.Errorf("failed to upload: %w", err)
|
|
}
|
|
return new.File, f.waitTask(ctx, new.Task.ID)
|
|
}
|
|
|
|
// Put the object
|
|
//
|
|
// Copy the reader in to the new object which is returned.
|
|
//
|
|
// The new object may have been created if an error is returned
|
|
func (f *Fs) Put(ctx context.Context, in io.Reader, src fs.ObjectInfo, options ...fs.OpenOption) (fs.Object, error) {
|
|
existingObj, err := f.NewObject(ctx, src.Remote())
|
|
switch err {
|
|
case nil:
|
|
return existingObj, existingObj.Update(ctx, in, src, options...)
|
|
case fs.ErrorObjectNotFound:
|
|
// Not found so create it
|
|
newObj := &Object{
|
|
fs: f,
|
|
remote: src.Remote(),
|
|
linkMu: new(sync.Mutex),
|
|
}
|
|
return newObj, newObj.upload(ctx, in, src, false, options...)
|
|
default:
|
|
return nil, err
|
|
}
|
|
}
|
|
|
|
// UserInfo fetches info about the current user
|
|
func (f *Fs) UserInfo(ctx context.Context) (userInfo map[string]string, err error) {
|
|
user, err := f.getUserInfo(ctx)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
userInfo = map[string]string{
|
|
"Id": user.Sub,
|
|
"Username": user.Name,
|
|
"Email": user.Email,
|
|
"PhoneNumber": user.PhoneNumber,
|
|
"Password": user.Password,
|
|
"Status": user.Status,
|
|
"CreatedAt": time.Time(user.CreatedAt).String(),
|
|
"PasswordUpdatedAt": time.Time(user.PasswordUpdatedAt).String(),
|
|
}
|
|
if vip, err := f.getVIPInfo(ctx); err == nil && vip.Result == "ACCEPTED" {
|
|
userInfo["VIPExpiresAt"] = time.Time(vip.Data.Expire).String()
|
|
userInfo["VIPStatus"] = vip.Data.Status
|
|
userInfo["VIPType"] = vip.Data.Type
|
|
}
|
|
return userInfo, nil
|
|
}
|
|
|
|
// ------------------------------------------------------------
|
|
|
|
// add offline download task for url
|
|
func (f *Fs) addURL(ctx context.Context, url, path string) (*api.Task, error) {
|
|
req := api.RequestNewTask{
|
|
Kind: api.KindOfFile,
|
|
UploadType: "UPLOAD_TYPE_URL",
|
|
URL: &api.URL{
|
|
URL: url,
|
|
},
|
|
FolderType: "DOWNLOAD",
|
|
}
|
|
if parentID, err := f.dirCache.FindDir(ctx, path, false); err == nil {
|
|
req.ParentID = parentIDForRequest(parentID)
|
|
req.FolderType = ""
|
|
}
|
|
return f.requestNewTask(ctx, &req)
|
|
}
|
|
|
|
type decompressDirResult struct {
|
|
Decompressed int
|
|
SourceDeleted int
|
|
Errors int
|
|
}
|
|
|
|
func (r decompressDirResult) Error() string {
|
|
return fmt.Sprintf("%d error(s) while decompressing - see log", r.Errors)
|
|
}
|
|
|
|
// decompress file/files in a directory of an ID
|
|
func (f *Fs) decompressDir(ctx context.Context, filename, id, password string, srcDelete bool) (r decompressDirResult, err error) {
|
|
_, err = f.listAll(ctx, id, api.KindOfFile, "false", func(item *api.File) bool {
|
|
if item.MimeType == "application/zip" || item.MimeType == "application/x-7z-compressed" || item.MimeType == "application/x-rar-compressed" {
|
|
if filename == "" || filename == item.Name {
|
|
res, err := f.requestDecompress(ctx, item, password)
|
|
if err != nil {
|
|
err = fmt.Errorf("unexpected error while requesting decompress of %q: %w", item.Name, err)
|
|
r.Errors++
|
|
fs.Errorf(f, "%v", err)
|
|
} else if res.Status != "OK" {
|
|
r.Errors++
|
|
fs.Errorf(f, "%q: %d files: %s", item.Name, res.FilesNum, res.Status)
|
|
} else {
|
|
r.Decompressed++
|
|
fs.Infof(f, "%q: %d files: %s", item.Name, res.FilesNum, res.Status)
|
|
if srcDelete {
|
|
derr := f.deleteObjects(ctx, []string{item.ID}, f.opt.UseTrash)
|
|
if derr != nil {
|
|
derr = fmt.Errorf("failed to delete %q: %w", item.Name, derr)
|
|
r.Errors++
|
|
fs.Errorf(f, "%v", derr)
|
|
} else {
|
|
r.SourceDeleted++
|
|
}
|
|
}
|
|
}
|
|
}
|
|
}
|
|
return false
|
|
})
|
|
if err != nil {
|
|
err = fmt.Errorf("couldn't list files to decompress: %w", err)
|
|
r.Errors++
|
|
fs.Errorf(f, "%v", err)
|
|
}
|
|
if r.Errors != 0 {
|
|
return r, r
|
|
}
|
|
return r, nil
|
|
}
|
|
|
|
var commandHelp = []fs.CommandHelp{{
|
|
Name: "addurl",
|
|
Short: "Add offline download task for url",
|
|
Long: `This command adds offline download task for url.
|
|
|
|
Usage:
|
|
|
|
rclone backend addurl pikpak:dirpath url
|
|
|
|
Downloads will be stored in 'dirpath'. If 'dirpath' is invalid,
|
|
download will fallback to default 'My Pack' folder.
|
|
`,
|
|
}, {
|
|
Name: "decompress",
|
|
Short: "Request decompress of a file/files in a folder",
|
|
Long: `This command requests decompress of file/files in a folder.
|
|
|
|
Usage:
|
|
|
|
rclone backend decompress pikpak:dirpath {filename} -o password=password
|
|
rclone backend decompress pikpak:dirpath {filename} -o delete-src-file
|
|
|
|
An optional argument 'filename' can be specified for a file located in
|
|
'pikpak:dirpath'. You may want to pass '-o password=password' for a
|
|
password-protected files. Also, pass '-o delete-src-file' to delete
|
|
source files after decompression finished.
|
|
|
|
Result:
|
|
|
|
{
|
|
"Decompressed": 17,
|
|
"SourceDeleted": 0,
|
|
"Errors": 0
|
|
}
|
|
`,
|
|
}}
|
|
|
|
// Command the backend to run a named command
|
|
//
|
|
// The command run is name
|
|
// args may be used to read arguments from
|
|
// opts may be used to read optional arguments from
|
|
//
|
|
// The result should be capable of being JSON encoded
|
|
// If it is a string or a []string it will be shown to the user
|
|
// otherwise it will be JSON encoded and shown to the user like that
|
|
func (f *Fs) Command(ctx context.Context, name string, arg []string, opt map[string]string) (out interface{}, err error) {
|
|
switch name {
|
|
case "addurl":
|
|
if len(arg) != 1 {
|
|
return nil, errors.New("need exactly 1 argument")
|
|
}
|
|
return f.addURL(ctx, arg[0], "")
|
|
case "decompress":
|
|
filename := ""
|
|
if len(arg) > 0 {
|
|
filename = arg[0]
|
|
}
|
|
id, err := f.dirCache.FindDir(ctx, "", false)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("failed to get an ID of dirpath: %w", err)
|
|
}
|
|
password := ""
|
|
if pass, ok := opt["password"]; ok {
|
|
password = pass
|
|
}
|
|
_, srcDelete := opt["delete-src-file"]
|
|
return f.decompressDir(ctx, filename, id, password, srcDelete)
|
|
default:
|
|
return nil, fs.ErrorCommandNotFound
|
|
}
|
|
}
|
|
|
|
// ------------------------------------------------------------
|
|
|
|
// parseFileID gets fid parameter from url query
|
|
func parseFileID(s string) string {
|
|
if u, err := url.Parse(s); err == nil {
|
|
if q, err := url.ParseQuery(u.RawQuery); err == nil {
|
|
return q.Get("fid")
|
|
}
|
|
}
|
|
return ""
|
|
}
|
|
|
|
// setMetaData sets the metadata from info
|
|
func (o *Object) setMetaData(info *api.File) (err error) {
|
|
if info.Kind == api.KindOfFolder {
|
|
return fs.ErrorIsDir
|
|
}
|
|
if info.Kind != api.KindOfFile {
|
|
return fmt.Errorf("%q is %q: %w", o.remote, info.Kind, fs.ErrorNotAFile)
|
|
}
|
|
o.hasMetaData = true
|
|
o.id = info.ID
|
|
o.size = info.Size
|
|
o.modTime = time.Time(info.ModifiedTime)
|
|
o.mimeType = info.MimeType
|
|
if info.ParentID == "" {
|
|
o.parent = "root"
|
|
} else {
|
|
o.parent = info.ParentID
|
|
}
|
|
o.gcid = info.Hash
|
|
o.md5sum = info.Md5Checksum
|
|
if info.Links.ApplicationOctetStream != nil {
|
|
o.link = info.Links.ApplicationOctetStream
|
|
if fid := parseFileID(o.link.URL); fid != "" {
|
|
for mid, media := range info.Medias {
|
|
if media.Link == nil {
|
|
continue
|
|
}
|
|
if mfid := parseFileID(media.Link.URL); fid == mfid {
|
|
fs.Debugf(o, "Using a media link from Medias[%d]", mid)
|
|
o.link = media.Link
|
|
break
|
|
}
|
|
}
|
|
}
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// setMetaDataWithLink ensures a link for opening an object
|
|
func (o *Object) setMetaDataWithLink(ctx context.Context) error {
|
|
o.linkMu.Lock()
|
|
defer o.linkMu.Unlock()
|
|
|
|
// check if the current link is valid
|
|
if o.link.Valid() {
|
|
return nil
|
|
}
|
|
|
|
info, err := o.fs.getFile(ctx, o.id)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
return o.setMetaData(info)
|
|
}
|
|
|
|
// readMetaData gets the metadata if it hasn't already been fetched
|
|
//
|
|
// it also sets the info
|
|
func (o *Object) readMetaData(ctx context.Context) (err error) {
|
|
if o.hasMetaData {
|
|
return nil
|
|
}
|
|
info, err := o.fs.readMetaDataForPath(ctx, o.remote)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
return o.setMetaData(info)
|
|
}
|
|
|
|
// Fs returns the parent Fs
|
|
func (o *Object) Fs() fs.Info {
|
|
return o.fs
|
|
}
|
|
|
|
// Return a string version
|
|
func (o *Object) String() string {
|
|
if o == nil {
|
|
return "<nil>"
|
|
}
|
|
return o.remote
|
|
}
|
|
|
|
// Remote returns the remote path
|
|
func (o *Object) Remote() string {
|
|
return o.remote
|
|
}
|
|
|
|
// Hash returns the Md5sum of an object returning a lowercase hex string
|
|
func (o *Object) Hash(ctx context.Context, t hash.Type) (string, error) {
|
|
if t != hash.MD5 {
|
|
return "", hash.ErrUnsupported
|
|
}
|
|
return strings.ToLower(o.md5sum), nil
|
|
}
|
|
|
|
// Size returns the size of an object in bytes
|
|
func (o *Object) Size() int64 {
|
|
err := o.readMetaData(context.TODO())
|
|
if err != nil {
|
|
fs.Logf(o, "Failed to read metadata: %v", err)
|
|
return 0
|
|
}
|
|
return o.size
|
|
}
|
|
|
|
// MimeType of an Object if known, "" otherwise
|
|
func (o *Object) MimeType(ctx context.Context) string {
|
|
return o.mimeType
|
|
}
|
|
|
|
// ID returns the ID of the Object if known, or "" if not
|
|
func (o *Object) ID() string {
|
|
return o.id
|
|
}
|
|
|
|
// ParentID returns the ID of the Object parent if known, or "" if not
|
|
func (o *Object) ParentID() string {
|
|
return o.parent
|
|
}
|
|
|
|
// ModTime returns the modification time of the object
|
|
func (o *Object) ModTime(ctx context.Context) time.Time {
|
|
err := o.readMetaData(ctx)
|
|
if err != nil {
|
|
fs.Logf(o, "Failed to read metadata: %v", err)
|
|
return time.Now()
|
|
}
|
|
return o.modTime
|
|
}
|
|
|
|
// SetModTime sets the modification time of the local fs object
|
|
func (o *Object) SetModTime(ctx context.Context, modTime time.Time) error {
|
|
return fs.ErrorCantSetModTime
|
|
}
|
|
|
|
// Storable returns a boolean showing whether this object storable
|
|
func (o *Object) Storable() bool {
|
|
return true
|
|
}
|
|
|
|
// Remove an object
|
|
func (o *Object) Remove(ctx context.Context) error {
|
|
return o.fs.deleteObjects(ctx, []string{o.id}, o.fs.opt.UseTrash)
|
|
}
|
|
|
|
// httpResponse gets an http.Response object for the object
|
|
// using the url and method passed in
|
|
func (o *Object) httpResponse(ctx context.Context, url, method string, options []fs.OpenOption) (res *http.Response, err error) {
|
|
if url == "" {
|
|
return nil, errors.New("forbidden to download - check sharing permission")
|
|
}
|
|
req, err := http.NewRequestWithContext(ctx, method, url, nil)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
fs.FixRangeOption(options, o.size)
|
|
fs.OpenOptionAddHTTPHeaders(req.Header, options)
|
|
if o.size == 0 {
|
|
// Don't supply range requests for 0 length objects as they always fail
|
|
delete(req.Header, "Range")
|
|
}
|
|
err = o.fs.pacer.Call(func() (bool, error) {
|
|
res, err = o.fs.client.Do(req)
|
|
return o.fs.shouldRetry(ctx, res, err)
|
|
})
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
return res, nil
|
|
}
|
|
|
|
// open a url for reading
|
|
func (o *Object) open(ctx context.Context, url string, options ...fs.OpenOption) (in io.ReadCloser, err error) {
|
|
res, err := o.httpResponse(ctx, url, "GET", options)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("open file failed: %w", err)
|
|
}
|
|
return res.Body, nil
|
|
}
|
|
|
|
// Open an object for read
|
|
func (o *Object) Open(ctx context.Context, options ...fs.OpenOption) (in io.ReadCloser, err error) {
|
|
if o.id == "" {
|
|
return nil, errors.New("can't download: no id")
|
|
}
|
|
if o.size == 0 {
|
|
// zero-byte objects may have no download link
|
|
return io.NopCloser(bytes.NewBuffer([]byte(nil))), nil
|
|
}
|
|
if err = o.setMetaDataWithLink(ctx); err != nil {
|
|
return nil, fmt.Errorf("can't download: %w", err)
|
|
}
|
|
return o.open(ctx, o.link.URL, options...)
|
|
}
|
|
|
|
// Update the object with the contents of the io.Reader, modTime and size
|
|
//
|
|
// If existing is set then it updates the object rather than creating a new one.
|
|
//
|
|
// The new object may have been created if an error is returned
|
|
func (o *Object) Update(ctx context.Context, in io.Reader, src fs.ObjectInfo, options ...fs.OpenOption) (err error) {
|
|
return o.upload(ctx, in, src, true, options...)
|
|
}
|
|
|
|
// upload uploads the object with or without using a temporary file name
|
|
func (o *Object) upload(ctx context.Context, in io.Reader, src fs.ObjectInfo, withTemp bool, options ...fs.OpenOption) (err error) {
|
|
size := src.Size()
|
|
remote := o.Remote()
|
|
|
|
// Create the directory for the object if it doesn't exist
|
|
leaf, dirID, err := o.fs.dirCache.FindPath(ctx, remote, true)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
// Calculate gcid; grabbed from package jottacloud
|
|
gcid, err := o.fs.getGcid(ctx, src)
|
|
if err != nil || gcid == "" {
|
|
fs.Debugf(o, "calculating gcid: %v", err)
|
|
if srcObj := fs.UnWrapObjectInfo(src); srcObj != nil && srcObj.Fs().Features().IsLocal {
|
|
// No buffering; directly calculate gcid from source
|
|
rc, err := srcObj.Open(ctx)
|
|
if err != nil {
|
|
return fmt.Errorf("failed to open src: %w", err)
|
|
}
|
|
defer fs.CheckClose(rc, &err)
|
|
|
|
if gcid, err = calcGcid(rc, srcObj.Size()); err != nil {
|
|
return fmt.Errorf("failed to calculate gcid: %w", err)
|
|
}
|
|
} else {
|
|
var cleanup func()
|
|
gcid, in, cleanup, err = readGcid(in, size, int64(o.fs.opt.HashMemoryThreshold))
|
|
defer cleanup()
|
|
if err != nil {
|
|
return fmt.Errorf("failed to calculate gcid: %w", err)
|
|
}
|
|
}
|
|
}
|
|
fs.Debugf(o, "gcid = %s", gcid)
|
|
|
|
if !withTemp {
|
|
info, err := o.fs.upload(ctx, in, leaf, dirID, gcid, size, options...)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
return o.setMetaData(info)
|
|
}
|
|
|
|
// We have to fall back to upload + rename
|
|
tempName := "rcloneTemp" + random.String(8)
|
|
info, err := o.fs.upload(ctx, in, tempName, dirID, gcid, size, options...)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
// upload was successful, need to delete old object before rename
|
|
if err = o.Remove(ctx); err != nil {
|
|
return fmt.Errorf("failed to remove old object: %w", err)
|
|
}
|
|
|
|
// rename also updates metadata
|
|
if info, err = o.fs.renameObject(ctx, info.ID, leaf); err != nil {
|
|
return fmt.Errorf("failed to rename temp object: %w", err)
|
|
}
|
|
return o.setMetaData(info)
|
|
}
|
|
|
|
// Check the interfaces are satisfied
|
|
var (
|
|
// _ fs.ListRer = (*Fs)(nil)
|
|
// _ fs.ChangeNotifier = (*Fs)(nil)
|
|
// _ fs.PutStreamer = (*Fs)(nil)
|
|
_ fs.Fs = (*Fs)(nil)
|
|
_ fs.Purger = (*Fs)(nil)
|
|
_ fs.CleanUpper = (*Fs)(nil)
|
|
_ fs.Copier = (*Fs)(nil)
|
|
_ fs.Mover = (*Fs)(nil)
|
|
_ fs.DirMover = (*Fs)(nil)
|
|
_ fs.Commander = (*Fs)(nil)
|
|
_ fs.DirCacheFlusher = (*Fs)(nil)
|
|
_ fs.PublicLinker = (*Fs)(nil)
|
|
_ fs.Abouter = (*Fs)(nil)
|
|
_ fs.UserInfoer = (*Fs)(nil)
|
|
_ fs.Object = (*Object)(nil)
|
|
_ fs.MimeTyper = (*Object)(nil)
|
|
_ fs.IDer = (*Object)(nil)
|
|
_ fs.ParentIDer = (*Object)(nil)
|
|
)
|