fixes error with unpacking packages

src/sync_drive.py

@@ -6,8 +6,9 @@
 import re
 import time
 import unicodedata
+import zipfile
 from pathlib import Path
-from shutil import copyfileobj, rmtree, unpack_archive
+from shutil import copyfileobj, rmtree
 
 import magic
 from icloudpy import exceptions
@@ -153,11 +154,11 @@ def process_package(local_file):
         archive_file += ".zip"
         os.rename(local_file, archive_file)
         LOGGER.info(f"Unpacking {archive_file} to {os.path.dirname(archive_file)}")
-        unpack_archive(filename=archive_file, extract_dir=os.path.dirname(archive_file))
-        try:
-            os.rename(unicodedata.normalize("NFD", local_file), local_file)
-        except Exception as e:
-            LOGGER.warning("Normalizing failed - " + str(e))
+        zipfile.ZipFile(archive_file).extractall(path=os.path.dirname(archive_file))
+        normalized_path = unicodedata.normalize("NFD", local_file)
+        if normalized_path is not local_file:
+            os.rename(local_file, normalized_path)
+            local_file = normalized_path
         os.remove(archive_file)
     elif "application/gzip" == magic_object.from_file(filename=local_file):
         archive_file += ".gz"
@@ -174,7 +175,7 @@ def process_package(local_file):
         )
         return False
     LOGGER.info(f"Successfully unpacked the package {archive_file}.")
-    return True
+    return local_file
 
 
 def is_package(item):
@@ -195,13 +196,13 @@ def download_file(item, local_file):
             with open(local_file, "wb") as file_out:
                 copyfileobj(response.raw, file_out)
             if response.url and "/packageDownload?" in response.url:
-                process_package(local_file=local_file)
+                local_file = process_package(local_file=local_file)
         item_modified_time = time.mktime(item.date_modified.timetuple())
         os.utime(local_file, (item_modified_time, item_modified_time))
     except (exceptions.ICloudPyAPIResponseException, FileNotFoundError, Exception) as e:
         LOGGER.error(f"Failed to download {local_file}: {str(e)}")
         return False
-    return True
+    return local_file
 
 
 def process_file(item, destination_path, filters, ignore, files):
@@ -221,10 +222,7 @@ def process_file(item, destination_path, filters, ignore, files):
             return False
     elif file_exists(item=item, local_file=local_file):
         return False
-    download_file(item=item, local_file=local_file)
-    if item_is_package:
-        for f in Path(local_file).glob("**/*"):
-            files.add(str(f))
+    local_file = download_file(item=item, local_file=local_file)
     return True
 
 

tests/data/__init__.py

@@ -2675,6 +2675,20 @@
                                 "etag": "4ioq::4eu4",
                                 "type": "FILE",
                             },
+                            {
+                                "dateCreated": "2021-08-02T19:21:54Z",
+                                "drivewsid": "FILE::com.apple.CloudDocs::111F1760-D940-480F-8C4F-005824A4E05G",
+                                "docwsid": "111F1760-D940-480F-8C4F-005824A4E05G",
+                                "zone": "com.apple.CloudDocs",
+                                "name": "Fotoksiążka-Wzór",
+                                "extension": "xmcf",
+                                "parentId": "FOLDER::com.apple.CloudDocs::117F1760-D940-480F-8C4F-005824A4E05D",
+                                "dateModified": "2021-08-01T19:21:54Z",
+                                "dateChanged": "2021-12-02T23:44:55Z",
+                                "size": 31514,
+                                "etag": "4ioq::4eu4",
+                                "type": "FILE",
+                            },
                         ],
                     }
                 ],
@@ -3316,6 +3330,28 @@
     }
 ]
 
+DRIVE_PACKAGE_SPECIAL_CHARS_WORKING = {
+    "document_id": "111F1760-D940-480F-8C4F-005824A4E05G",
+    "package_token": {
+        "url": "https://cvws.icloud-content.com/B/signature1ref_signature1/Fotoksiążka-Wzór.xmcf?o=object1&v=1&x=3&a"
+        "=token1&e=1588472097&k=wrapping_key1&fl=&r=request&ckc=com.apple.clouddocs&ckz=com.apple.CloudDocs&p"
+        "=31&s=s1",
+        "token": "token1",
+        "signature": "signature1",
+        "wrapping_key": "wrapping_key1==",
+        "reference_signature": "ref_signature1",
+    },
+    "thumbnail_token": {
+        "url": "https://cvws.icloud-content.com/B/signature2ref_signature2/Scanned+document+1.jpg?o=object2&v=1&x=3&a"
+        "=token2&e=1588472097&k=wrapping_key2&fl=&r=request&ckc=com.apple.clouddocs&ckz=com.apple.CloudDocs&p"
+        "=31&s=s2",
+        "token": "token2",
+        "signature": "signature2",
+        "wrapping_key": "wrapping_key2==",
+        "reference_signature": "ref_signature2",
+    },
+    "double_etag": "32::2x",
+}
 DRIVE_PACKAGE_DOWNLOAD_WORKING = {
     "document_id": "111F1760-D940-480F-8C4F-005824A4E05E",
     "package_token": {
@@ -3749,11 +3785,24 @@ def request(self, method, url, **kwargs):
                 "111F1760-D940-480F-8C4F-005824A4E05F",
             ]:
                 return ResponseMock(DRIVE_PACKAGE_DOWNLOAD_WORKING_EXTRACTION_ERROR)
+            if params.get("document_id") in ["111F1760-D940-480F-8C4F-005824A4E05G"]:
+                return ResponseMock(DRIVE_PACKAGE_SPECIAL_CHARS_WORKING)
         if "icloud-content.com" in url and method == "GET":
             if "Scanned+document+1.pdf" in url or "Document scanne 2.pdf" in url:
                 return ResponseMock({}, raw=open(__file__, "rb"))
             if "This is a title.md" in url:
                 return ResponseMock({}, raw=open("This is a title.md", "rb"))
+            if "Fotoksiążka-Wzór.xmcf" in url:
+                return ResponseMock(
+                    {},
+                    url="/packageDownload?",
+                    raw=open(
+                        os.path.join(
+                            os.path.dirname(__file__), "Fotoksiążka-Wzór.xmcf.zip"
+                        ),
+                        "rb",
+                    ),
+                )
             if "Project.band" in url:
                 return ResponseMock(
                     {},

tests/data/test_config.yaml

@@ -48,6 +48,7 @@ drive:
       - jpeg
       - md
       - band
+      - xmcf
 photos:
   destination: photos
   remove_obsolete: false

tests/test_sync_drive.py

@@ -32,6 +32,9 @@ def setUp(self) -> None:
         self.folder_item = self.drive[self.items[5]]
         self.file_item = self.drive[self.items[4]]["Test"]["Scanned document 1.pdf"]
         self.package_item = self.drive[self.items[6]]["Sample"]["Project.band"]
+        self.special_chars_package_item = self.drive[self.items[6]]["Sample"][
+            "Fotoksiążka-Wzór.xmcf"
+        ]
         self.package_item_nested = self.drive[self.items[6]]["Sample"]["ms.band"]
         self.file_name = "Scanned document 1.pdf"
         self.package_name = "Project.band"
@@ -437,7 +440,6 @@ def test_wanted_folder_ignore_takes_precedence_to_filters(self):
             )
         )
 
-
     def test_wanted_folder_empty(self):
         """Test for empty wanted folder."""
         original_filters = dict(self.filters)
@@ -456,21 +458,29 @@ def test_wanted_folder_none_folder_path(self):
         """Test for wanted folder path as None."""
         self.assertTrue(
             sync_drive.wanted_folder(
-                filters=self.filters["folders"], ignore=None, root=self.root, folder_path=None
+                filters=self.filters["folders"],
+                ignore=None,
+                root=self.root,
+                folder_path=None,
             )
         )
 
     def test_wanted_folder_none_filters(self):
         """Test for wanted folder filters as None."""
         self.assertTrue(
-            sync_drive.wanted_folder(filters=None, ignore=None, root=self.root, folder_path="dir1")
+            sync_drive.wanted_folder(
+                filters=None, ignore=None, root=self.root, folder_path="dir1"
+            )
         )
 
     def test_wanted_folder_none_root(self):
         """Test for wanted folder root as None."""
         self.assertTrue(
             sync_drive.wanted_folder(
-                filters=self.filters["folders"], ignore=None, root=None, folder_path="dir1"
+                filters=self.filters["folders"],
+                ignore=None,
+                root=None,
+                folder_path="dir1",
             )
         )
 
@@ -487,15 +497,19 @@ def test_wanted_file_missing(self):
         """Test for a missing wanted file."""
         self.assertFalse(
             sync_drive.wanted_file(
-                filters=self.filters["file_extensions"], ignore=None, file_path=tests.CONFIG_PATH
+                filters=self.filters["file_extensions"],
+                ignore=None,
+                file_path=tests.CONFIG_PATH,
             )
         )
 
     def test_wanted_file_check_log(self):
         """Test for valid unwanted file."""
         with self.assertLogs(logger=LOGGER, level="DEBUG") as captured:
             sync_drive.wanted_file(
-                filters=self.filters["file_extensions"], ignore=None, file_path=tests.CONFIG_PATH
+                filters=self.filters["file_extensions"],
+                ignore=None,
+                file_path=tests.CONFIG_PATH,
             )
             self.assertTrue(len(captured.records) > 0)
             self.assertIn(
@@ -504,7 +518,9 @@ def test_wanted_file_check_log(self):
 
     def test_wanted_file_none_file_path(self):
         """Test for unexpected wanted file path."""
-        self.assertTrue(sync_drive.wanted_file(filters=None, ignore=None, file_path=__file__))
+        self.assertTrue(
+            sync_drive.wanted_file(filters=None, ignore=None, file_path=__file__)
+        )
         self.assertFalse(
             sync_drive.wanted_file(
                 filters=self.filters["file_extensions"], ignore=None, file_path=None
@@ -925,7 +941,7 @@ def test_sync_directory_without_remove(self):
             filters=self.filters,
             remove=False,
         )
-        self.assertTrue(len(actual) == 22)
+        self.assertTrue(len(actual) == 11)
         self.assertTrue(os.path.isdir(os.path.join(self.destination_path, "icloudpy")))
         self.assertTrue(
             os.path.isdir(os.path.join(self.destination_path, "icloudpy", "Test"))
@@ -961,7 +977,7 @@ def test_sync_directory_with_remove(self):
             ignore=self.ignore,
             remove=True,
         )
-        self.assertTrue(len(actual) == 22)
+        self.assertTrue(len(actual) == 11)
         self.assertTrue(os.path.isdir(os.path.join(self.destination_path, "icloudpy")))
         self.assertTrue(
             os.path.isdir(os.path.join(self.destination_path, "icloudpy", "Test"))
@@ -995,7 +1011,7 @@ def test_sync_directory_without_folder_filter(self):
             ignore=self.ignore,
             remove=False,
         )
-        self.assertTrue(len(actual) == 26)
+        self.assertTrue(len(actual) == 15)
         self.assertTrue(os.path.isdir(os.path.join(self.destination_path, "icloudpy")))
         self.assertTrue(
             os.path.isdir(os.path.join(self.destination_path, "icloudpy", "Test"))
@@ -1155,6 +1171,20 @@ def test_sync_drive_valids(
             ),
         )
 
+    def test_process_file_special_chars_package(self):
+        """Test for special characters package."""
+        files = set()
+        # Download the package
+        self.assertTrue(
+            sync_drive.process_file(
+                item=self.special_chars_package_item,
+                destination_path=self.destination_path,
+                filters=self.filters["file_extensions"],
+                ignore=None,
+                files=files,
+            )
+        )
+
     def test_process_file_existing_package(self):
         """Test for existing package."""
         files = set()
@@ -1200,18 +1230,16 @@ def test_process_file_nested_package_extraction(self):
                 files=files,
             )
         )
-        self.assertTrue(
-            os.path.exists(os.path.join(self.destination_path, "My Song.band"))
-        )
+        self.assertTrue(os.path.exists(os.path.join(self.destination_path, "ms.band")))
         self.assertEqual(
             sum(
                 os.path.getsize(f)
-                for f in os.listdir(os.path.join(self.destination_path, "My Song.band"))
+                for f in os.listdir(os.path.join(self.destination_path, "ms.band"))
                 if os.path.isfile(f)
             ),
             sum(
                 os.path.getsize(f)
-                for f in os.listdir(os.path.join(tests.DATA_DIR, "My Song.band"))
+                for f in os.listdir(os.path.join(tests.DATA_DIR, "ms.band"))
                 if os.path.isfile(f)
             ),
         )
@@ -1242,7 +1270,7 @@ def test_execution_continuation_on_icloudpy_exception(self):
                 ignore=self.ignore,
                 remove=False,
             )
-            self.assertTrue(len(actual) == 23)
+            self.assertTrue(len(actual) == 12)
             self.assertTrue(
                 os.path.isdir(os.path.join(self.destination_path, "icloudpy"))
             )