613 lines
20 KiB
Python
613 lines
20 KiB
Python
import json
|
|
import re
|
|
import logging
|
|
from functools import wraps
|
|
from shutil import which
|
|
from typing import Optional, Union, List, Dict, Callable
|
|
|
|
from rclone_python import utils
|
|
from rclone_python.hash_types import HashTypes
|
|
from rclone_python.remote_types import RemoteTypes
|
|
|
|
# debug flag enables/disables raw output of rclone progresses in the terminal
|
|
DEBUG = False
|
|
|
|
|
|
def __check_installed(func):
|
|
@wraps(func)
|
|
def wrapper(*args, **kwargs):
|
|
if not is_installed():
|
|
raise Exception(
|
|
"rclone is not installed on this system. Please install it here: https://rclone.org/"
|
|
)
|
|
|
|
return func(*args, **kwargs)
|
|
|
|
return wrapper
|
|
|
|
|
|
def is_installed() -> bool:
|
|
"""
|
|
:return: True if rclone is correctly installed on the system.
|
|
"""
|
|
return which("rclone") is not None
|
|
|
|
|
|
@__check_installed
|
|
def about(remote_name: str):
|
|
"""
|
|
Executes the rclone about command and returns the retrieved json as a dictionary.
|
|
:param remote_name: The name of the remote to examine.
|
|
:return: Dictionary with remote properties.
|
|
"""
|
|
if not remote_name.endswith(":"):
|
|
# if the remote name missed the colon manually add it.
|
|
remote_name += ":"
|
|
|
|
process = utils.run_cmd(f"rclone about {remote_name} --json")
|
|
|
|
if process.returncode == 0:
|
|
return json.loads(process.stdout)
|
|
else:
|
|
raise Exception(
|
|
f"An error occurred while executing the about command: {process.stderr}"
|
|
)
|
|
|
|
|
|
@__check_installed
|
|
def check_remote_existing(remote_name: str) -> bool:
|
|
"""
|
|
Returns True, if the specified rclone remote is already configured.
|
|
:param remote_name: The name of the remote to check.
|
|
:return: True if the remote exists, False otherwise.
|
|
"""
|
|
# get the available remotes
|
|
remotes = get_remotes()
|
|
|
|
# add the trailing ':' if it is missing
|
|
if not remote_name.endswith(":"):
|
|
remote_name = f"{remote_name}:"
|
|
|
|
return remote_name in remotes
|
|
|
|
|
|
@__check_installed
|
|
def create_remote(
|
|
remote_name: str,
|
|
remote_type: Union[str, RemoteTypes],
|
|
client_id: Union[str, None] = None,
|
|
client_secret: Union[str, None] = None,
|
|
**kwargs,
|
|
):
|
|
"""Creates a new remote with name, type and options.
|
|
|
|
Args:
|
|
remote_name (str): Name of the new remote.
|
|
remote_type (Union[str, RemoteTypes]): The type of the remote (e.g. "onedrive", RemoteTypes.dropbox, ...)
|
|
client_id (str, optional): OAuth Client Id.
|
|
client_secret (str, optional): OAuth Client Secret.
|
|
**kwargs: Additional key value pairs that can be used with the "rclone config create" command.
|
|
"""
|
|
if isinstance(remote_type, RemoteTypes):
|
|
remote_type = remote_type.value
|
|
|
|
if not check_remote_existing(remote_name):
|
|
# set up the selected cloud
|
|
command = f'rclone config create "{remote_name}" "{remote_type}"'
|
|
|
|
if client_id and client_secret:
|
|
logging.info("Using the provided client id and client secret.")
|
|
|
|
kwargs["client_id"] = client_id
|
|
kwargs["client_secret"] = client_secret
|
|
else:
|
|
logging.warning(
|
|
"The drive client id and the client secret have not been set. Using defaults."
|
|
)
|
|
|
|
# add the options as key-value pairs
|
|
for key, value in kwargs.items():
|
|
command += f' {key}="{value}"'
|
|
|
|
# run the setup command
|
|
process = utils.run_cmd(command)
|
|
|
|
if process.returncode != 0:
|
|
raise Exception(process.stderr)
|
|
else:
|
|
raise Exception(
|
|
f"A rclone remote with the name '{remote_name}' already exists!"
|
|
)
|
|
|
|
|
|
def copy(
|
|
in_path: str,
|
|
out_path: str,
|
|
ignore_existing=False,
|
|
show_progress=True,
|
|
listener: Callable[[Dict], None] = None,
|
|
args=None,
|
|
):
|
|
"""
|
|
Copies a file or a directory from a src path to a destination path.
|
|
:param in_path: The source path to use. Specify the remote with 'remote_name:path_on_remote'
|
|
:param out_path: The destination path to use. Specify the remote with 'remote_name:path_on_remote'
|
|
:param ignore_existing: If True, all existing files are ignored and not overwritten.
|
|
:param show_progress: If true, show a progressbar.
|
|
:param listener: An event-listener that is called with every update of rclone.
|
|
:param args: List of additional arguments/ flags.
|
|
"""
|
|
if args is None:
|
|
args = []
|
|
|
|
_rclone_transfer_operation(
|
|
in_path,
|
|
out_path,
|
|
ignore_existing=ignore_existing,
|
|
command="rclone copy",
|
|
command_descr="Copying",
|
|
show_progress=show_progress,
|
|
listener=listener,
|
|
args=args,
|
|
)
|
|
|
|
|
|
def copyto(
|
|
in_path: str,
|
|
out_path: str,
|
|
ignore_existing=False,
|
|
show_progress=True,
|
|
listener: Callable[[Dict], None] = None,
|
|
args=None,
|
|
):
|
|
"""
|
|
Copies a file or a directory from a src path to a destination path and is typically used when renaming a file is necessary.
|
|
:param in_path: The source path to use. Specify the remote with 'remote_name:path_on_remote'
|
|
:param out_path: The destination path to use. Specify the remote with 'remote_name:path_on_remote'
|
|
:param ignore_existing: If True, all existing files are ignored and not overwritten.
|
|
:param show_progress: If true, show a progressbar.
|
|
:param listener: An event-listener that is called with every update of rclone.
|
|
:param args: List of additional arguments/ flags.
|
|
"""
|
|
if args is None:
|
|
args = []
|
|
|
|
_rclone_transfer_operation(
|
|
in_path,
|
|
out_path,
|
|
ignore_existing=ignore_existing,
|
|
command="rclone copyto",
|
|
command_descr="Copying",
|
|
show_progress=show_progress,
|
|
listener=listener,
|
|
args=args,
|
|
)
|
|
|
|
|
|
def move(
|
|
in_path: str,
|
|
out_path: str,
|
|
ignore_existing=False,
|
|
show_progress=True,
|
|
listener: Callable[[Dict], None] = None,
|
|
args=None,
|
|
):
|
|
"""
|
|
Moves a file or a directory from a src path to a destination path.
|
|
:param in_path: The source path to use. Specify the remote with 'remote_name:path_on_remote'
|
|
:param out_path: The destination path to use. Specify the remote with 'remote_name:path_on_remote'
|
|
:param ignore_existing: If True, all existing files are ignored and not overwritten.
|
|
:param show_progress: If true, show a progressbar.
|
|
:param listener: An event-listener that is called with every update of rclone.
|
|
:param args: List of additional arguments/ flags.
|
|
"""
|
|
if args is None:
|
|
args = []
|
|
|
|
_rclone_transfer_operation(
|
|
in_path,
|
|
out_path,
|
|
ignore_existing=ignore_existing,
|
|
command="rclone move",
|
|
command_descr="Moving",
|
|
show_progress=show_progress,
|
|
listener=listener,
|
|
args=args,
|
|
)
|
|
|
|
|
|
def moveto(
|
|
in_path: str,
|
|
out_path: str,
|
|
ignore_existing=False,
|
|
show_progress=True,
|
|
listener: Callable[[Dict], None] = None,
|
|
args=None,
|
|
):
|
|
"""
|
|
Moves a file or a directory from a src path to a destination path and is typically used when renaming is necessary.
|
|
:param in_path: The source path to use. Specify the remote with 'remote_name:path_on_remote'
|
|
:param out_path: The destination path to use. Specify the remote with 'remote_name:path_on_remote'
|
|
:param ignore_existing: If True, all existing files are ignored and not overwritten.
|
|
:param show_progress: If true, show a progressbar.
|
|
:param listener: An event-listener that is called with every update of rclone.
|
|
:param args: List of additional arguments/ flags.
|
|
"""
|
|
if args is None:
|
|
args = []
|
|
|
|
_rclone_transfer_operation(
|
|
in_path,
|
|
out_path,
|
|
ignore_existing=ignore_existing,
|
|
command="rclone moveto",
|
|
command_descr="Moving",
|
|
show_progress=show_progress,
|
|
listener=listener,
|
|
args=args,
|
|
)
|
|
|
|
|
|
def sync(
|
|
src_path: str,
|
|
dest_path: str,
|
|
show_progress=True,
|
|
listener: Callable[[Dict], None] = None,
|
|
args=None,
|
|
):
|
|
"""
|
|
Sync the source to the destination, changing the destination only. Doesn't transfer files that are identical on source and destination, testing by size and modification time or MD5SUM.
|
|
:param in_path: The source path to use. Specify the remote with 'remote_name:path_on_remote'
|
|
:param out_path: The destination path to use. Specify the remote with 'remote_name:path_on_remote'
|
|
:param show_progress: If true, show a progressbar.
|
|
:param listener: An event-listener that is called with every update of rclone.
|
|
:param args: List of additional arguments/ flags.
|
|
"""
|
|
if args is None:
|
|
args = []
|
|
|
|
_rclone_transfer_operation(
|
|
src_path,
|
|
dest_path,
|
|
command="rclone sync",
|
|
command_descr="Syncing",
|
|
show_progress=show_progress,
|
|
listener=listener,
|
|
args=args,
|
|
)
|
|
|
|
|
|
@__check_installed
|
|
def get_remotes() -> List[str]:
|
|
"""
|
|
:return: A list of all available remotes.
|
|
"""
|
|
command = "rclone listremotes"
|
|
remotes = utils.run_cmd(command).stdout.split()
|
|
if remotes is None:
|
|
remotes = []
|
|
|
|
return remotes
|
|
|
|
|
|
@__check_installed
|
|
def purge(path: str, args=None):
|
|
"""
|
|
Purges the specified folder. This means that unlike with delete, also all the folders are removed.
|
|
:param args: List of additional arguments/ flags.
|
|
:param path: The path of the folder that should be purged.
|
|
"""
|
|
if args is None:
|
|
args = []
|
|
|
|
command = f'rclone purge "{path}"'
|
|
process = utils.run_cmd(command, args)
|
|
if process.returncode == 0:
|
|
logging.info(f"Successfully purged {path}")
|
|
else:
|
|
raise Exception(
|
|
f'Purging path "{path}" failed with error message:\n{process.stderr}'
|
|
)
|
|
|
|
|
|
@__check_installed
|
|
def delete(path: str, args=None):
|
|
"""
|
|
Deletes a file or a folder. When deleting a folder, all the files in it and it's subdirectories are removed,
|
|
but not the folder structure itself.
|
|
:param args: List of additional arguments/ flags.
|
|
:param path: The path of the folder that should be deleted.
|
|
"""
|
|
if args is None:
|
|
args = []
|
|
|
|
command = f'rclone delete "{path}"'
|
|
process = utils.run_cmd(command, args)
|
|
|
|
if process.returncode == 0:
|
|
logging.info(f"Successfully deleted {path}")
|
|
else:
|
|
raise Exception(
|
|
f'Deleting path "{path}" failed with error message:\n{process.stderr}'
|
|
)
|
|
|
|
|
|
@__check_installed
|
|
def link(
|
|
path: str,
|
|
expire: Union[str, None] = None,
|
|
unlink=False,
|
|
args=None,
|
|
) -> str:
|
|
"""
|
|
Generates a public link to a file/directory.
|
|
:param path: The path to the file/directory that a public link should be created, retrieved or removed for.
|
|
:param expire: The amount of time that the link will be valid (not supported by all backends).
|
|
:param unlink: If true, remove existing public links to the file or directory (not supported by all backends).
|
|
:param args: List of additional arguments/ flags.
|
|
:return: The link to the given file or directory.
|
|
"""
|
|
if args is None:
|
|
args = []
|
|
|
|
command = f'rclone link "{path}"'
|
|
|
|
# add optional parameters
|
|
if expire is not None:
|
|
args.append(f"--expire {expire}")
|
|
if unlink:
|
|
args.append(f"--unlink")
|
|
|
|
process = utils.run_cmd(command, args)
|
|
|
|
if process.returncode != 0:
|
|
raise Exception(process.stderr)
|
|
else:
|
|
return process.stdout
|
|
|
|
|
|
@__check_installed
|
|
def ls(
|
|
path: str,
|
|
max_depth: Union[int, None] = None,
|
|
dirs_only=False,
|
|
files_only=False,
|
|
args=None,
|
|
) -> List[Dict[str, Union[int, str]]]:
|
|
"""
|
|
Lists the files in a directory.
|
|
:param path: The path to the folder that should be examined.
|
|
:param max_depth: The maximum depth for file search. If max_depth=1 only the files in the selected folder but not
|
|
in its subdirectories will be included.
|
|
:param files_only: If true only files will be returned.
|
|
:param dirs_only: If true, only dirs will be returned.
|
|
:param args: List of additional arguments/ flags.
|
|
:return: List of dicts containing file properties.
|
|
"""
|
|
if args is None:
|
|
args = []
|
|
|
|
command = f'rclone lsjson "{path}"'
|
|
|
|
# add optional parameters
|
|
if max_depth is not None:
|
|
args.append(f"--max-depth {max_depth}")
|
|
if dirs_only:
|
|
args.append(f"--dirs-only")
|
|
if files_only:
|
|
args.append("--files-only")
|
|
|
|
process = utils.run_cmd(command, args)
|
|
|
|
if process.returncode == 0:
|
|
return json.loads(process.stdout)
|
|
else:
|
|
raise Exception(f"ls operation on {path} failed with:\n{process.stderr}")
|
|
|
|
|
|
def tree(
|
|
path: str,
|
|
args: List[str] = None,
|
|
) -> str:
|
|
"""Returns the contents of the remote path in a tree like fashion.
|
|
|
|
Args:
|
|
path (str): The path from which the tree should be generated
|
|
args (List[str], optional): Optional additional list of flags.
|
|
|
|
Returns:
|
|
str: String containing the file tree.
|
|
"""
|
|
if args is None:
|
|
args = []
|
|
|
|
process = utils.run_cmd(f'rclone tree "{path}"', args)
|
|
|
|
if process.returncode != 0:
|
|
raise Exception(process.stderr)
|
|
else:
|
|
return process.stdout
|
|
|
|
|
|
@__check_installed
|
|
def hash(
|
|
hash: Union[str, HashTypes],
|
|
path: str,
|
|
download=False,
|
|
checkfile: Optional[str] = None,
|
|
output_file: Optional[str] = None,
|
|
args: List[str] = None,
|
|
) -> Union[None, str, bool, Dict[str, str], Dict[str, bool]]:
|
|
"""Produces a hashsum file for all the objects in the path.
|
|
|
|
Args:
|
|
hash (Union[str, HashTypes]): The hash algorithm to use, e.g. sha1. Depends on the backend used.
|
|
path (str): The path to the file/ folder to generate hashes for.
|
|
download (bool, optional): Download the file and hash it locally. Useful when the backend does not support the selected hash algorithm.
|
|
checkfile (Optional[str], optional): Validate hashes against a given SUM file instead of printing them.
|
|
output_file (Optional[str], optional): Output hashsums to a file rather than the terminal (same format as the checkfile).
|
|
args (List[str], optional): Optional additional list of flags.
|
|
|
|
Raises:
|
|
Exception: Raised when the rclone command does not succeed.
|
|
|
|
Returns:
|
|
Union[None, str, bool, Dict[str, str], Dict[str, bool]]: 3 different modes apply based on the inputs:
|
|
1) Nothing is returned when output file is set.
|
|
2) When checkfile is set, a dictionary is returned with file names as keys.
|
|
The values are either True or False, depending on wether the file is valid ot not.
|
|
In the special case of only a single file, True or False is directly returned.
|
|
3) If neither checkfile nor output_file is set, a dictionary is returned with file names as keys.
|
|
The values are the individual hash sums.
|
|
In the special case of only a single file, the hashsum is directly returned.
|
|
"""
|
|
|
|
if isinstance(hash, HashTypes):
|
|
hash = hash.value
|
|
|
|
if args is None:
|
|
args = []
|
|
|
|
if download:
|
|
args.append("--download")
|
|
|
|
if checkfile is not None:
|
|
args.append(f'--checkfile "{checkfile}"')
|
|
|
|
if output_file is not None:
|
|
args.append(f'--output-file "{output_file}"')
|
|
|
|
process: str = utils.run_cmd(f'rclone hashsum "{hash}" "{path}"', args)
|
|
|
|
lines = process.stdout.splitlines()
|
|
|
|
exception = False
|
|
|
|
if process.returncode != 0:
|
|
if checkfile is None:
|
|
exception = True
|
|
else:
|
|
# validate that the checkfile command succeeded, by checking if the output has the expected form
|
|
for l in lines:
|
|
if not (l.startswith("= ") or l.startswith("* ")):
|
|
exception = True
|
|
break
|
|
|
|
if exception:
|
|
raise Exception(
|
|
f"hashsum operation on {path} with hash='{hash}' failed with:\n{process.stderr}"
|
|
)
|
|
|
|
if output_file is None:
|
|
# each line contains the hashsum first, followed by the name of the file
|
|
hashsums = {}
|
|
|
|
for l in lines:
|
|
if len(l) > 0:
|
|
value, key = l.split()
|
|
|
|
if checkfile is None:
|
|
hashsums[key] = value
|
|
else:
|
|
# in checkfile mode, value is '=' for valid and '*' for invalid files
|
|
hashsums[key] = value == "="
|
|
|
|
# for only a single file return the value instead of the dict
|
|
if len(hashsums) == 1:
|
|
return next(iter(hashsums.values()))
|
|
|
|
return hashsums
|
|
|
|
|
|
@__check_installed
|
|
def version(
|
|
check=False,
|
|
args: List[str] = None,
|
|
) -> Union[str, List[str]]:
|
|
"""Get the rclone version number.
|
|
|
|
Args:
|
|
check (bool, optional): Whether to do an online check to compare your version with the latest release and the latest beta. Defaults to False.
|
|
args (List[str], optional): Optional additional list of flags. Defaults to None.
|
|
|
|
Returns:
|
|
Union[str, Set[str, str, str]]: When check is False, returns string of current version. When check is True returns installed version, lastest version and latest beta version.
|
|
"""
|
|
if args is None:
|
|
args = []
|
|
|
|
if check:
|
|
args.append("--check")
|
|
|
|
process = utils.run_cmd("rclone version", args)
|
|
|
|
if process.returncode != 0:
|
|
raise Exception(process.stderr)
|
|
|
|
stdout = process.stdout
|
|
|
|
if not check:
|
|
return stdout.split("\n")[0].replace("rclone ", "")
|
|
else:
|
|
yours = re.findall(r"yours:\s+([\d.]+)", stdout)[0]
|
|
latest = re.findall(r"latest:\s+([\d.]+)", stdout)[0]
|
|
# beta version might include dashes and word characters e.g. '1.64.0-beta.7161.9169b2b5a'
|
|
beta = re.findall(r"beta:\s+([.\w-]+)", stdout)[0]
|
|
return yours, latest, beta
|
|
|
|
|
|
@__check_installed
|
|
def _rclone_transfer_operation(
|
|
in_path: str,
|
|
out_path: str,
|
|
command: str,
|
|
command_descr: str,
|
|
ignore_existing=False,
|
|
show_progress=True,
|
|
listener: Callable[[Dict], None] = None,
|
|
args=None,
|
|
):
|
|
"""Executes the rclone transfer operation (e.g. copyto, move, ...) and displays the progress of every individual file.
|
|
|
|
Args:
|
|
in_path (str): The source path to use. Specify the remote with 'remote_name:path_on_remote'
|
|
out_path (str): The destination path to use. Specify the remote with 'remote_name:path_on_remote'
|
|
command (str): The rclone command to execute (e.g. rclone copyto)
|
|
command_descr (str): The description to this command that should be displayed.
|
|
ignore_existing (bool, optional): If True, all existing files are ignored and not overwritten.
|
|
show_progress (bool, optional): If true, show a progressbar.
|
|
listener (Callable[[Dict], None], optional): An event-listener that is called with every update of rclone.
|
|
args: List of additional arguments/ flags.
|
|
"""
|
|
if args is None:
|
|
args = []
|
|
|
|
prog_title = f"{command_descr} [bold magenta]{utils.shorten_filepath(in_path, 20)}[/bold magenta] to [bold magenta]{utils.shorten_filepath(out_path, 20)}"
|
|
|
|
# add global rclone flags
|
|
if ignore_existing:
|
|
command += " --ignore-existing"
|
|
command += " --progress"
|
|
|
|
# in path
|
|
command += f' "{in_path}"'
|
|
# out path
|
|
command += f' "{out_path}"'
|
|
|
|
# optional named arguments/flags
|
|
command += utils.args2string(args)
|
|
|
|
# execute the upload command
|
|
process = utils.rclone_progress(
|
|
command, prog_title, listener=listener, show_progress=show_progress, debug=DEBUG
|
|
)
|
|
|
|
if process.wait() == 0:
|
|
logging.info("Cloud upload completed.")
|
|
else:
|
|
_, err = process.communicate()
|
|
raise Exception(
|
|
f"Copy/Move operation from {in_path} to {out_path}"
|
|
f' failed with error message:\n{err.decode("utf-8")}'
|
|
)
|