Coverage for src/drive_package_processing.py: 100%

1"""Package processing utilities. 

2 

3This module provides package extraction and processing functionality, 

4separating archive handling logic from sync operations per SRP. 

5""" 

6 

7__author__ = "Mandar Patil (mandarons@pm.me)" 

8 

9import gzip 

10import os 

11import unicodedata 

12import zipfile 

13from shutil import copyfileobj 

14 

15import magic 

16 

17from src import configure_icloudpy_logging, get_logger 

18 

19# Configure icloudpy logging immediately after import 

20configure_icloudpy_logging() 

21 

22LOGGER = get_logger() 

23 

24 

25def process_package(local_file: str) -> str | None: 

26 """Process and extract a downloaded package file. 

27 

28 This function handles different archive types (ZIP, gzip) and extracts them 

29 to the appropriate location. It also handles Unicode normalization for 

30 cross-platform compatibility. 

31 

32 Args: 

33 local_file: Path to the downloaded package file 

34 

35 Returns: 

36 Path to the processed file/directory, or False if processing failed 

37 """ 

38 archive_file = local_file 

39 

40 # zipfile.is_zipfile() rather than libmagic's MIME string: libmagic reports 

41 # "application/octet-stream" for many of Apple's packageDownload zips even though 

42 # they begin with PK\x03\x04 and open fine with zipfile. is_zipfile() reads the End 

43 # of Central Directory record and is the authority on whether ZipFile() will 

44 # succeed, which is all _process_zip_package() needs to know. 

45 if zipfile.is_zipfile(local_file): 

46 return _process_zip_package(local_file, archive_file) 

47 

48 magic_object = magic.Magic(mime=True) 

49 file_mime_type = magic_object.from_file(filename=local_file) 

50 

51 if file_mime_type == "application/gzip": 

52 return _process_gzip_package(local_file, archive_file) 

53 else: 

54 LOGGER.error( 

55 f"Unhandled file type - cannot unpack the package {local_file} ({file_mime_type}). " 

56 "The downloaded archive is left in place under the package's name.", 

57 ) 

58 return None 

59 

60 

61def _process_zip_package(local_file: str, archive_file: str) -> str: 

62 """Process a ZIP package file. 

63 

64 Args: 

65 local_file: Original file path 

66 archive_file: Archive file path 

67 

68 Returns: 

69 Path to the processed file 

70 """ 

71 archive_file += ".zip" 

72 os.rename(local_file, archive_file) 

73 LOGGER.info(f"Unpacking {archive_file} to {os.path.dirname(archive_file)}") 

74 zipfile.ZipFile(archive_file).extractall(path=os.path.dirname(archive_file)) 

75 

76 # Handle Unicode normalization for cross-platform compatibility 

77 normalized_path = unicodedata.normalize("NFD", local_file) 

78 if normalized_path != local_file: 

79 os.rename(local_file, normalized_path) 

80 local_file = normalized_path 

81 

82 os.remove(archive_file) 

83 LOGGER.info(f"Successfully unpacked the package {archive_file}.") 

84 return local_file 

85 

86 

87def _process_gzip_package(local_file: str, archive_file: str) -> str | None: 

88 """Process a gzip package file. 

89 

90 Args: 

91 local_file: Original file path 

92 archive_file: Archive file path 

93 

94 Returns: 

95 Path to the processed file, or None if processing failed 

96 """ 

97 archive_file += ".gz" 

98 os.rename(local_file, archive_file) 

99 LOGGER.info(f"Unpacking {archive_file} to {os.path.dirname(local_file)}") 

100 

101 with gzip.GzipFile(filename=archive_file, mode="rb") as gz_file: 

102 with open(file=local_file, mode="wb") as package_file: 

103 copyfileobj(gz_file, package_file) 

104 

105 os.remove(archive_file) 

106 

107 # Recursively process the extracted file (might be another archive) 

108 return process_package(local_file=local_file)