rclone/backend/pikpak/pikpak.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)
)