Add include-hidden option to sync

Author: const-cloudinary

cloudinary_cli/modules/sync.py

@@ -23,17 +23,20 @@
 @argument("cloudinary_folder")
 @option("--push", help="Push changes from your local folder to your Cloudinary folder.", is_flag=True)
 @option("--pull", help="Pull changes from your Cloudinary folder to your local folder.", is_flag=True)
+@option("-H", "--include-hidden", is_flag=True, help="Include hidden files in sync.")
 @option("-w", "--concurrent_workers", type=int, default=_DEFAULT_CONCURRENT_WORKERS,
         help="Specify the number of concurrent network threads.")
 @option("-F", "--force", is_flag=True, help="Skip confirmation when deleting files.")
 @option("-K", "--keep-unique", is_flag=True, help="Keep unique files in the destination folder.")
 @option("-D", "--deletion-batch-size", type=int, default=_DEFAULT_DELETION_BATCH_SIZE,
         help="Specify the batch size for deleting remote assets.")
-def sync(local_folder, cloudinary_folder, push, pull, concurrent_workers, force, keep_unique, deletion_batch_size):
+def sync(local_folder, cloudinary_folder, push, pull, include_hidden, concurrent_workers, force, keep_unique,
+         deletion_batch_size):
     if push == pull:
         raise Exception("Please use either the '--push' OR '--pull' options")
 
-    sync_dir = SyncDir(local_folder, cloudinary_folder, concurrent_workers, force, keep_unique, deletion_batch_size)
+    sync_dir = SyncDir(local_folder, cloudinary_folder, include_hidden, concurrent_workers, force, keep_unique,
+                       deletion_batch_size)
 
     if push:
         sync_dir.push()
@@ -44,17 +47,19 @@ def sync(local_folder, cloudinary_folder, push, pull, concurrent_workers, force,
 
 
 class SyncDir:
-    def __init__(self, local_dir, remote_dir, concurrent_workers, force, keep_deleted, deletion_batch_size):
+    def __init__(self, local_dir, remote_dir, include_hidden, concurrent_workers, force, keep_deleted,
+                 deletion_batch_size):
         self.local_dir = local_dir
         self.remote_dir = remote_dir
+        self.include_hidden = include_hidden
         self.concurrent_workers = concurrent_workers
         self.force = force
         self.keep_unique = keep_deleted
         self.deletion_batch_size = deletion_batch_size
 
         self.verbose = logger.getEffectiveLevel() < logging.INFO
 
-        self.local_files = walk_dir(abspath(self.local_dir))
+        self.local_files = walk_dir(abspath(self.local_dir), include_hidden)
         logger.info(f"Found {len(self.local_files)} items in local folder '{local_dir}'")
 
         self.remote_files = query_cld_folder(self.remote_dir)

cloudinary_cli/utils/file_utils.py

@@ -1,13 +1,19 @@
+import os
+import stat
 from os import walk, path, listdir, rmdir, sep
 from os.path import split, relpath, abspath
 
 from cloudinary_cli.defaults import logger
 from cloudinary_cli.utils.utils import etag
 
 
-def walk_dir(root_dir):
+def walk_dir(root_dir, include_hidden=False):
     all_files = {}
-    for root, _, files in walk(root_dir):
+    for root, dirs, files in walk(root_dir):
+        if not include_hidden:
+            files = [f for f in files if not is_hidden(root, f)]
+            dirs[:] = [d for d in dirs if not is_hidden(root, d)]
+
         relative_path = relpath(root, root_dir) if root_dir != root else ""
         for file in files:
             full_path = path.join(root, file)
@@ -19,6 +25,25 @@ def walk_dir(root_dir):
     return all_files
 
 
+def is_hidden(root, relative_path):
+    return is_hidden_path(path.join(root, relative_path))
+
+
+def is_hidden_path(filepath):
+    name = os.path.basename(filepath)
+    return name.startswith('.') or has_hidden_attribute(filepath)
+
+
+def has_hidden_attribute(filepath):
+    st = os.stat(filepath)
+
+    if not hasattr(st, 'st_file_attributes'):  # not a pythonic way, but it's relevant only for windows, no need to try
+        return False
+
+    # noinspection PyUnresolvedReferences
+    return bool(st.st_file_attributes & stat.FILE_ATTRIBUTE_HIDDEN)
+
+
 def delete_empty_dirs(root, remove_root=False):
     if not path.isdir(root):
         return

test/test_file_utils.py

@@ -1,9 +1,9 @@
 import unittest
 
-from cloudinary_cli.utils.file_utils import get_destination_folder
+from cloudinary_cli.utils.file_utils import get_destination_folder, walk_dir
 
 
-class UtilsTest(unittest.TestCase):
+class FileUtilsTest(unittest.TestCase):
     def test_get_destination_folder(self):
         """ should parse option values correctly """
 
@@ -14,3 +14,10 @@ def test_get_destination_folder(self):
                                                 "/Users/user/myfolder/subfolder/file.jpg",
                                                 parent="/Users/user/"))
 
+    def test_walk_dir(self):
+        """ should skip hidden files in the directory """
+
+        test_dir = "test_resources/test_file_utils"
+
+        self.assertEqual(1, len(walk_dir(test_dir, include_hidden=False)))
+        self.assertEqual(4, len(walk_dir(test_dir, include_hidden=True)))