Merge pull request #15 from MitchellThompkins/update-python-rclone-dep

Remove python-rclone dependency

Author: MitchellThompkins

poetry.lock

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "rclone_decrypt"
-version = "0.1.6"
+version = "0.1.7"
 description = "Wrapper around rclone to decrypt files encrypted with rclone"
 authors = ["Mitchell Thompkins <mitchell.thompkins@gmail.com>"]
 license = "MIT"
@@ -13,7 +13,6 @@ readme = "README.md"
 [tool.poetry.dependencies]
 python = ">=3.10"
 click = "^8.1.3"
-python-rclone = "^0.0.2"
 python-statemachine = "^2.0.0"
 tkinterdnd2 = "^0.3.0"
 

pyproject.toml

@@ -1 +1 @@
-__version__ = "0.1.6"
+__version__ = "0.1.7"

src/rclone_decrypt/__init__.py

@@ -1,7 +1,6 @@
 import click
 
 import rclone_decrypt.decrypt as decrypt
-import rclone_decrypt.gui as GUI
 
 
 help_str_config = f"""config file. default config file is:
@@ -23,22 +22,15 @@
     help=help_str_output,
     default=decrypt.default_output_dir,
 )
-@click.option("--gui", help="start the GUI", is_flag=True, default=False)
-@click.option(
-    "--gui_debug", help="print debug messages", is_flag=True, default=False
-)
-def cli(config, files, output_dir, gui, gui_debug):
-    if gui is True:
-        GUI.start_gui(gui_debug)
-    else:
-        try:
-            if files is None:
-                raise ValueError("files cannot be None")
-            else:
-                decrypt.decrypt(files, config, output_dir)
-
-        except (ValueError, decrypt.RCloneExecutableError) as err:
-            decrypt.print_error(err)
+def cli(config, files, output_dir):
+    try:
+        if files is None:
+            raise ValueError("files cannot be None")
+        else:
+            decrypt.decrypt(files, config, output_dir)
+
+    except (ValueError, decrypt.RCloneExecutableError) as err:
+        decrypt.print_error(err)
 
 
 if __name__ == "__main__":

src/rclone_decrypt/cli.py

@@ -2,17 +2,28 @@
 import re
 import shutil
 import tempfile
+import subprocess
 
-import rclone
 from statemachine import State, StateMachine
 
 default_output_dir = "out"
 
-# TODO(mitchellthompkins): This won't work on windows, check the rclone
-# documentation for the windows default location
-default_rclone_conf_dir = os.path.join(
-    os.environ["HOME"], ".config", "rclone", "rclone.conf"
-)
+try:
+    if shutil.which("rclone"):
+        # Get the rclone config file path dynamically
+        cmd = ["rclone", "config", "file"]
+        out = subprocess.check_output(cmd).decode().strip()
+        # The output format is usually:
+        # "Configuration file is stored at:\n/path/to/rclone.conf"
+        # We need to parse the last line
+        default_rclone_conf_dir = out.splitlines()[-1]
+    else:
+        raise FileNotFoundError("rclone executable not found")
+except (subprocess.CalledProcessError, FileNotFoundError):
+    # Fallback to the previous default if rclone command fails or is not found
+    default_rclone_conf_dir = os.path.join(
+        os.environ["HOME"], ".config", "rclone", "rclone.conf"
+    )
 
 
 class ConfigFileError(Exception):
@@ -77,99 +88,106 @@ def before_is_valid(self, line: str) -> None:
         self.cfg_file.write(line)
 
 
-def get_rclone_instance(
+def get_rclone_config_path(
     config: str, files: str, remote_folder_name: str
-) -> rclone.RClone:
+) -> str:
     """
     Opens a config file and strips out all of the non-crypt type entries,
     modifies the remote to be local directory.
 
-    Returns an rclone instance.
+    Returns the path to the temporary rclone config file.
     """
-    rclone_instance = None
+    config_path = None
 
     try:
         with open(config, "r") as f:
             config_file = f.readlines()
 
-            with tempfile.NamedTemporaryFile(
-                mode="wt", delete=True
-            ) as tmp_config_file:
-                with open(tmp_config_file.name, "w") as config:
-                    config_state = ConfigWriterControl(config)
-
-                    for line in config_file:
-                        state_id = config_state.current_state.id
-
-                        if state_id == "searching_for_start":
-                            start_of_entry = re.search("\\[.*?\\]", line)
-
-                            if start_of_entry is not None:
-                                config_state.validate(line)
+            # Create a temporary file that persists until manually deleted or
+            # cleaned up by caller.
+            # We use delete=False so we can return the path and use it later.
+            # It will be created in the system temp dir or temp_dir_name if
+            # passed? Actually, let's create it inside remote_folder_name
+            # (which is a temp dir).
+
+            config_path = os.path.join(remote_folder_name, "rclone.conf")
+
+            with open(config_path, "w") as config_out:
+                config_state = ConfigWriterControl(config_out)
+
+                for line in config_file:
+                    state_id = config_state.current_state.id
+
+                    if state_id == "searching_for_start":
+                        start_of_entry = re.search("\\[.*?\\]", line)
+
+                        if start_of_entry is not None:
+                            config_state.validate(line)
+                        else:
+                            config_state.search()
+
+                    elif state_id == "type_check":
+                        regex_str = "type\\s*=\\s*([\\S\\s]+)"
+                        entry_type = re.search(regex_str, line)
+                        if entry_type is not None:
+                            entry_type = entry_type.group(1).strip()
+                            if entry_type == "crypt":
+                                valid_str = f"type = {entry_type}\n"
+                                config_state.is_valid(valid_str)
                             else:
-                                config_state.search()
-
-                        elif state_id == "type_check":
-                            regex_str = "type\\s*=\\s*([\\S\\s]+)"
-                            entry_type = re.search(regex_str, line)
-                            if entry_type is not None:
-                                entry_type = entry_type.group(1).strip()
-                                if entry_type == "crypt":
-                                    valid_str = f"type = {entry_type}\n"
-                                    config_state.is_valid(valid_str)
-                                else:
-                                    config_state.is_invalid()
-
-                        elif state_id == "writing":
-                            regex_str = "remote\\s*=\\s*([\\S\\s]+)"
-                            remote = re.search(regex_str, line)
-                            if remote is not None:
-                                config_state.write(
-                                    f"remote =\
-                                        {remote_folder_name}/\n"
-                                )
-
-                            elif line == "\n":
-                                config_state.write(line)
-                                config_state.write_complete()
+                                config_state.is_invalid()
 
-                            else:
-                                config_state.write(line)
+                    elif state_id == "writing":
+                        regex_str = "remote\\s*=\\s*([\\S\\s]+)"
+                        remote = re.search(regex_str, line)
+                        if remote is not None:
+                            config_state.write(
+                                f"remote =\
+                                    {remote_folder_name}/\n"
+                            )
 
-                    config_state.complete()
+                        elif line == "\n":
+                            config_state.write(line)
+                            config_state.write_complete()
 
-                # Open the modified temporary file and create our instance
-                with open(tmp_config_file.name, "r") as t:
-                    o = t.read()
-                    rclone_instance = rclone.with_config(o)
+                        else:
+                            config_state.write(line)
 
-        # I think that given a file, any file, rclone.with_config() will always
-        # return _something_ as it doesn't validate the config file
-        if rclone_instance is None:
-            raise ConfigFileError("The rclone instance was not created.")
+                config_state.complete()
 
     except FileNotFoundError as err:
         print_error(err)
+        return None
 
-    return rclone_instance
+    return config_path
 
 
-def rclone_copy(rclone_instance: rclone.RClone, output_dir: str) -> None:
+def rclone_copy(config_path: str, output_dir: str) -> None:
     """
     Calls the rclone copy function via a shell instance and places the
     decrypted files into the output_dir
     """
     # convert list of remotes in str format into a list
-    remotes = rclone_instance.listremotes()["out"].decode().splitlines()
+    list_cmd = ["rclone", "--config", config_path, "listremotes"]
+    try:
+        out = subprocess.check_output(list_cmd).decode()
+        remotes = out.splitlines()
+    except subprocess.CalledProcessError as e:
+        print_error(f"Failed to list remotes: {e}")
+        return
 
     for r in remotes:
         print(f"Copying and decrypting: {r}")
-        rclone_instance.copy(f"{r}", f"{output_dir}")
-        # TODO(@mitchellthompkins): rclone.copy still returns 0 for an
-        # unsuccessful decryption. As long as the call itself doesn't fail, it
-        # will return 0.  Need to come up with someway to detect success
-        # if success['code'] == 0:
-        #    break
+        copy_cmd = [
+            "rclone",
+            "--config",
+            config_path,
+            "copy",
+            f"{r}",
+            f"{output_dir}",
+        ]
+        # TODO(@mitchellthompkins): check return code for success
+        subprocess.run(copy_cmd, check=True)
 
 
 def decrypt(
@@ -195,10 +213,10 @@ def decrypt(
 
     try:
         with tempfile.TemporaryDirectory(dir=os.getcwd()) as temp_dir_name:
-            rclone_instance = get_rclone_instance(config, files, temp_dir_name)
+            config_path = get_rclone_config_path(config, files, temp_dir_name)
 
-            if rclone_instance is None:
-                raise ConfigFileError("rclone_instance cannot be None")
+            if config_path is None:
+                raise ConfigFileError("config_path cannot be None")
 
             if output_dir is default_output_dir:
                 # If no output_dir is provided, put the de-crypted file into a
@@ -231,7 +249,7 @@ def decrypt(
                 # Do the copy, we wrap this in a try in case the user
                 # interrupts the process, otherwise the file won't be
                 # moved back
-                rclone_copy(rclone_instance, output_dir)
+                rclone_copy(config_path, output_dir)
                 print(f"Decryption complete. Files saved to: {output_dir}")
             except KeyboardInterrupt:
                 print("\n\tterminated rclone copy!")

src/rclone_decrypt/decrypt.py

@@ -183,7 +183,7 @@ def test_no_config_file():
     Test behavior when provided no config file
     """
     files = os.path.join("tests", "something_fake")
-    instance = decrypt.get_rclone_instance("", files, "a_dir_name")
+    instance = decrypt.get_rclone_config_path("", files, "a_dir_name")
 
     assert instance is None
 
@@ -193,8 +193,10 @@ def test_config_file():
     Test behavior when provided valid config file
     """
     files = os.path.join("tests", "something_fake")
-    instance = decrypt.get_rclone_instance(
-        decrypt_rclone_config_file, files, "a_dir_name"
-    )
+    with tempfile.TemporaryDirectory() as temp_dir:
+        instance = decrypt.get_rclone_config_path(
+            decrypt_rclone_config_file, files, temp_dir
+        )
 
-    assert instance is not None
+        assert instance is not None
+        assert os.path.exists(instance)