Fix issue when file names contain invalid characters

src/BetterADBSync/FileSystems/Android.py

@@ -1,4 +1,5 @@
 from typing import Iterable, Iterator, List, NoReturn, Tuple
+import posixpath
 import logging
 import os
 import re
@@ -94,7 +95,7 @@ def adb_shell(self, commands: List[str]) -> Iterator[str]:
 
     def line_not_captured(self, line: str) -> NoReturn:
         logging.critical("ADB line not captured")
-        logging_fatal(line)
+        logging_fatal(line, force_exit = True)
 
     def test_connection(self):
         for line in self.adb_shell([":"]):
@@ -186,14 +187,14 @@ def utime(self, path: str, times: Tuple[int, int]) -> None:
             self.line_not_captured(line)
 
     def join(self, base: str, leaf: str) -> str:
-        return os.path.join(base, leaf).replace("\\", "/") # for Windows
+        return posixpath.join(base, leaf)
 
     def split(self, path: str) -> Tuple[str, str]:
-        head, tail = os.path.split(path)
-        return head.replace("\\", "/"), tail # for Windows
+        head, tail = posixpath.split(path)
+        return head, tail
 
     def normpath(self, path: str) -> str:
-        return os.path.normpath(path).replace("\\", "/")
+        return posixpath.normpath(path)
 
     def push_file_here(self, source: str, destination: str, show_progress: bool = False) -> None:
         if show_progress:
@@ -205,3 +206,9 @@ def push_file_here(self, source: str, destination: str, show_progress: bool = Fa
             }
         if subprocess.call(self.adb_arguments + ["push", source, destination], **kwargs_call):
             logging_fatal("Non-zero exit code from adb push")
+
+    def convert_invalid_file_name(self, path_destination: str) -> str:
+        return path_destination  # no implement on other system
+
+    def validate_args_path(self, path: str) -> str:
+        return path              # no implement on other system

src/BetterADBSync/FileSystems/Base.py

@@ -9,6 +9,7 @@
 class FileSystem():
     def __init__(self, adb_arguments: List[str]) -> None:
         self.adb_arguments = adb_arguments
+        self.has_invalid_name_potential = False
 
     def _get_files_tree(self, tree_path: str, tree_path_stat: os.stat_result, follow_links: bool = False):
         # the reason to have two functions instead of one purely recursive one is to use self.lstat_in_dir ie ls
@@ -67,6 +68,7 @@ def push_tree_here(self,
         tree: Union[Tuple[int, int], dict],
         destination_root: str,
         fs_source: FileSystem,
+        fs_destination: FileSystem,
         dry_run: bool = True,
         show_progress: bool = False
         ) -> None:
@@ -88,12 +90,17 @@ def push_tree_here(self,
             except KeyError:
                 pass
             for key, value in tree.items():
+                # FIXME sync will work improperly if file with name converted already exist
+                destination_key: str = fs_destination.convert_invalid_file_name(key)    # might not need when copy a file, or folder with 1-layer deep
+                if key != destination_key:
+                    logging.info(f"convert {key} -> {destination_key}")
                 self.push_tree_here(
                     fs_source.normpath(fs_source.join(tree_path, key)),
                     fs_source.join(relative_tree_path, key),
                     value,
-                    self.normpath(self.join(destination_root, key)),
+                    self.normpath(self.join(destination_root, destination_key)),
                     fs_source,
+                    fs_destination,
                     dry_run = dry_run,
                     show_progress = show_progress
                 )
@@ -138,3 +145,9 @@ def normpath(self, path: str) -> str:
 
     def push_file_here(self, source: str, destination: str, show_progress: bool = False) -> None:
         raise NotImplementedError
+
+    def convert_invalid_file_name(self, path_destination: str) -> str:
+        raise NotImplementedError  # Problem only persist on Windows. No implement for other system
+
+    def validate_args_path(self, path: str) -> str:
+        raise NotImplementedError

src/BetterADBSync/FileSystems/Local.py

@@ -1,6 +1,7 @@
 from typing import Iterable, Tuple
 import os
 import subprocess
+import logging
 
 from ..SAOLogging import logging_fatal
 
@@ -50,5 +51,37 @@ def push_file_here(self, source: str, destination: str, show_progress: bool = Fa
                 "stdout": subprocess.DEVNULL,
                 "stderr": subprocess.DEVNULL
             }
+
+        # TODO add retries limit flag
         if subprocess.call(self.adb_arguments + ["pull", source, destination], **kwargs_call):
             logging_fatal("Non-zero exit code from adb pull")
+
+    def setup_invalid_name_check(self) -> None:
+        self.set_invalid_name_potential()
+        self.convert_table = str.maketrans('\/*:?"<>|', '_________')    # slash and backslash still needs to be converted
+
+    def set_invalid_name_potential(self) -> None:
+        self.has_invalid_name_potential = os.name == 'nt'
+        logging.debug("has_invalid_name_potential is {}".format(self.has_invalid_name_potential))
+
+    def convert_invalid_file_name(self, path_destination: str) -> str: # usually has this problem on Windows
+        # TODO implement flag for accepting dictionary of invalid-replacement pairs
+        # (or single character replacement if provide a character instead)
+        # TODO implement flag for accepting a list of invalid characters
+        # TODO make character map customizable
+        # TODO implement different list of invalid character for each file system
+        if self.has_invalid_name_potential:
+            return path_destination.translate(self.convert_table)
+        else:
+            return path_destination
+
+    def validate_args_path(self, path: str) -> str:
+        if os.name == 'nt':
+            invalid_chars = '*?"<>|'              # assume that user input won't contain slash or backslash as in file name
+            for char in invalid_chars:
+                if char in path:
+                    logging_fatal(f"{path} contains invalid string", force_exit = True)
+            for idx, path_char in enumerate(path):    # seperate ':' case out from above, in case user provide X:\folder\file format
+                if path_char == ':' and idx != 1:
+                    logging_fatal(f"{path} contains invalid string", force_exit = True)
+        return path

src/BetterADBSync/SAOLogging.py

@@ -59,11 +59,13 @@ def setup_root_logger(
     console_handler.setFormatter(formatter_class(fmt = messagefmt_to_use, datefmt = datefmt))
     root_logger.addHandler(console_handler)
 
-def logging_fatal(message, log_stack_info: bool = True, exit_code: int = 1):
+def logging_fatal(message, log_stack_info: bool = True, exit_code: int = 1, force_exit: bool = False):
+    # TODO collect all fatal errors, and show all of it when program terminated. Useful for tracking files that has problems.
     logging.critical(message)
     logging.debug("Stack Trace", stack_info = log_stack_info)
-    logging.critical("Exiting")
-    raise SystemExit(exit_code)
+    if force_exit:
+        logging.critical("Exiting")
+        raise SystemExit(exit_code)
 
 def log_tree(title, tree, finals = None, log_leaves_types = True, logging_level = logging.INFO):
     """Log tree nicely if it is a dictionary.
@@ -94,6 +96,6 @@ def perror(s: Union[str, Any], e: Exception, logging_level: int = logging.ERROR)
     strerror = e.strerror if (isinstance(e, OSError) and e.strerror is not None) else e.__class__.__name__
     msg = f"{s}{': ' if s else ''}{strerror}"
     if logging_level == FATAL:
-        logging_fatal(msg)
+        logging_fatal(msg, force_exit = True)
     else:
         logging.log(logging_level, msg)

src/BetterADBSync/__init__.py

@@ -2,7 +2,7 @@
 
 """Sync files between a computer and an Android device"""
 
-__version__ = "1.4.0"
+__version__ = "1.4.1"
 
 from typing import List, Tuple, Union
 import logging
@@ -28,6 +28,7 @@ def diff_trees(cls,
         path_join_function_source,
         path_join_function_destination,
         folder_file_overwrite_error: bool = True,
+        delete_args: bool = True,
         ) -> Tuple[
             Union[dict, Tuple[int, int], None], # delete
             Union[dict, Tuple[int, int], None], # copy
@@ -85,7 +86,8 @@ def diff_trees(cls,
                             destination_exclude_patterns,
                             path_join_function_source,
                             path_join_function_destination,
-                            folder_file_overwrite_error = folder_file_overwrite_error
+                            folder_file_overwrite_error = folder_file_overwrite_error,
+                            delete_args = delete_args
                         )
             else:
                 raise NotImplementedError
@@ -112,13 +114,15 @@ def diff_trees(cls,
                     unaccounted_destination = None
                     excluded_destination = destination
                 else:
-                    if source[1] > destination[1]:
+                    if source[1] > destination[1] and delete_args:
                         delete = destination
                         copy = source
                         excluded_source = None
                         unaccounted_destination = None
                         excluded_destination = None
                     else:
+                        if not delete_args:
+                            logging.info(f"--no-del: Newer file of {path_source} won't be updated")
                         delete = None
                         copy = None
                         excluded_source = None
@@ -139,7 +143,7 @@ def diff_trees(cls,
                     excluded_destination = {".": None}
                     if folder_file_overwrite_error:
                         logging.critical(f"Refusing to overwrite directory {path_destination} with file {path_source}")
-                        logging_fatal("Use --force if you are sure!")
+                        logging_fatal("Use --force if you are sure!", force_exit = True)
                     else:
                         logging.warning(f"Overwriting directory {path_destination} with file {path_source}")
             else:
@@ -169,7 +173,8 @@ def diff_trees(cls,
                             destination_exclude_patterns,
                             path_join_function_source,
                             path_join_function_destination,
-                            folder_file_overwrite_error = folder_file_overwrite_error
+                            folder_file_overwrite_error = folder_file_overwrite_error,
+                            delete_args = delete_args
                         )
             elif isinstance(destination, tuple):
                 if exclude:
@@ -194,11 +199,12 @@ def diff_trees(cls,
                             destination_exclude_patterns,
                             path_join_function_source,
                             path_join_function_destination,
-                            folder_file_overwrite_error = folder_file_overwrite_error
+                            folder_file_overwrite_error = folder_file_overwrite_error,
+                            delete_args = delete_args
                         )
                     if folder_file_overwrite_error:
                         logging.critical(f"Refusing to overwrite file {path_destination} with directory {path_source}")
-                        logging_fatal("Use --force if you are sure!")
+                        logging_fatal("Use --force if you are sure!", force_exit = True)
                     else:
                         logging.warning(f"Overwriting file {path_destination} with directory {path_source}")
                 excluded_destination = None
@@ -225,7 +231,8 @@ def diff_trees(cls,
                             destination_exclude_patterns,
                             path_join_function_source,
                             path_join_function_destination,
-                            folder_file_overwrite_error = folder_file_overwrite_error
+                            folder_file_overwrite_error = folder_file_overwrite_error,
+                            delete_args = delete_args
                         )
                     destination.pop(".")
                     for key, value in destination.items():
@@ -237,7 +244,8 @@ def diff_trees(cls,
                             destination_exclude_patterns,
                             path_join_function_source,
                             path_join_function_destination,
-                            folder_file_overwrite_error = folder_file_overwrite_error
+                            folder_file_overwrite_error = folder_file_overwrite_error,
+                            delete_args = delete_args
                         )
             else:
                 raise NotImplementedError
@@ -307,7 +315,7 @@ def paths_to_fixed_destination_paths(cls,
             perror(path_source, e, FATAL)
 
         if stat.S_ISLNK(lstat_destination.st_mode):
-            logging_fatal("Destination is a symlink. Not sure what to do. See GitHub issue #8")
+            logging_fatal("Destination is a symlink. Not sure what to do. See GitHub issue #8", force_exit = True)
 
         if not stat.S_ISDIR(lstat_destination.st_mode):
             return path_source, path_destination
@@ -327,7 +335,9 @@ def paths_to_fixed_destination_paths(cls,
             )
         return path_source, path_destination
 
+
 def main():
+    # TODO implement allow file override flag (default is True, except when use --no-del value will be False.)
     args = get_cli_args(__doc__, __version__)
 
     setup_root_logger(
@@ -352,18 +362,19 @@ def main():
     try:
         fs_android.test_connection()
     except BrokenPipeError:
-        logging_fatal("Connection test failed")
+        logging_fatal("Connection test failed", force_exit = True)
 
     if args.direction == "push":
         path_source = args.direction_push_local
         fs_source = fs_local
         path_destination = args.direction_push_android
         fs_destination = fs_android
     else:
-        path_source = args.direction_pull_android
+        fs_local.setup_invalid_name_check()
         fs_source = fs_android
-        path_destination = args.direction_pull_local
+        path_source = fs_source.validate_args_path(args.direction_pull_android)
         fs_destination = fs_local
+        path_destination = fs_destination.validate_args_path(args.direction_pull_local)
 
     path_source, path_destination = FileSyncer.paths_to_fixed_destination_paths(path_source, fs_source, path_destination, fs_destination)
 
@@ -412,7 +423,8 @@ def main():
         excludePatterns,
         fs_source.join,
         fs_destination.join,
-        folder_file_overwrite_error = not args.dry_run and not args.force
+        folder_file_overwrite_error = not args.dry_run and not args.force,
+        delete_args = args.delete
     )
 
     tree_delete                  = FileSyncer.prune_tree(tree_delete)
@@ -470,12 +482,13 @@ def main():
     logging.info("SYNCING")
     logging.info("")
 
-    if tree_delete is not None:
-        logging.info("Deleting delete tree")
-        fs_destination.remove_tree(path_destination, tree_delete, dry_run = args.dry_run)
-    else:
-        logging.info("Empty delete tree")
-    logging.info("")
+    if args.delete:
+        if tree_delete is not None:
+            logging.info("Deleting delete tree")
+            fs_destination.remove_tree(path_destination, tree_delete, dry_run = args.dry_run)
+        else:
+            logging.info("Empty delete tree")
+        logging.info("")
 
     if args.delete_excluded and args.delete:
         if tree_excluded_destination is not None:
@@ -513,6 +526,7 @@ def main():
             tree_copy,
             path_destination,
             fs_source,
+            fs_destination,
             dry_run = args.dry_run,
             show_progress = args.show_progress
         )

src/BetterADBSync/argparsing.py

@@ -84,7 +84,8 @@ def get_cli_args(docstring: str, version: str) -> Args:
     )
     parser.add_argument("--del",
         help = "Delete files at the destination that are not in the source",
-        action = "store_true",
+        action = argparse.BooleanOptionalAction,
+        default = True,
         dest = "delete"
     )
     parser.add_argument("--delete-excluded",