Coverage for src/sync_drive.py: 100%

1"""Sync drive module. 

2 

3This module provides the main entry point for iCloud Drive synchronization, 

4orchestrating the sync process using specialized utility modules per SRP. 

5""" 

6 

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

8 

9import os 

10import unicodedata 

11from pathlib import Path 

12from typing import Any 

13from urllib.parse import unquote 

14 

15from src import config_parser, configure_icloudpy_logging, get_logger 

16from src.drive_cleanup import remove_obsolete # noqa: F401 

17from src.drive_file_download import download_file # noqa: F401 

18from src.drive_file_existence import file_exists, is_package, package_exists # noqa: F401 

19from src.drive_filtering import ignored_path, wanted_file, wanted_folder, wanted_parent_folder # noqa: F401 

20from src.drive_folder_processing import process_folder # noqa: F401 

21from src.drive_package_processing import process_package # noqa: F401 

22from src.drive_parallel_download import collect_file_for_download, download_file_task, files_lock # noqa: F401 

23from src.drive_sync_directory import sync_directory 

24from src.drive_thread_config import get_max_threads # noqa: F401 

25 

26# Configure icloudpy logging immediately after import 

27configure_icloudpy_logging() 

28 

29LOGGER = get_logger() 

30 

31 

32def sync_drive(config: Any, drive: Any) -> set[str]: 

33 """Synchronize iCloud Drive to local filesystem. 

34 

35 This function serves as the main entry point for drive synchronization, 

36 preparing the destination and delegating to the sync_directory orchestrator. 

37 

38 Args: 

39 config: Configuration dictionary containing drive settings 

40 drive: iCloud drive service instance 

41 

42 Returns: 

43 Set of all synchronized file paths 

44 """ 

45 destination_path = config_parser.prepare_drive_destination(config=config) 

46 return sync_directory( 

47 drive=drive, 

48 destination_path=destination_path, 

49 root=destination_path, 

50 items=drive.dir(), 

51 top=True, 

52 filters=config["drive"]["filters"] if "drive" in config and "filters" in config["drive"] else None, 

53 ignore=config["drive"]["ignore"] if "drive" in config and "ignore" in config["drive"] else None, 

54 remove=config_parser.get_drive_remove_obsolete(config=config), 

55 config=config, 

56 ) 

57 

58 

59def process_file( 

60 item: Any, 

61 destination_path: str, 

62 filters: list[str], 

63 ignore: list[str], 

64 files: set[str], 

65 config: dict | None = None, 

66) -> bool: 

67 """Process given item as file (legacy compatibility function). 

68 

69 This function maintains backward compatibility with existing tests. 

70 New code should use the specialized modules directly. 

71 

72 Args: 

73 item: iCloud file item to process 

74 destination_path: Local destination directory 

75 filters: File extension filters 

76 ignore: Ignore patterns 

77 files: Set to track processed files 

78 config: Configuration dictionary (used to resolve request timeout) 

79 

80 Returns: 

81 True if file was processed successfully, False otherwise 

82 """ 

83 if not (item and destination_path and files is not None): 

84 return False 

85 # Decode URL-encoded filename from iCloud API 

86 # This handles special characters like %CC%88 (combining diacritical marks) 

87 decoded_name = unquote(item.name) 

88 local_file = os.path.join(destination_path, decoded_name) 

89 local_file = unicodedata.normalize("NFC", local_file) 

90 if not wanted_file(filters=filters, ignore=ignore, file_path=local_file): 

91 return False 

92 files.add(local_file) 

93 

94 # Check local existence FIRST to avoid unnecessary network requests. 

95 # is_package() makes an HTTP call for every file, which is very slow 

96 # when syncing thousands of already-up-to-date files. 

97 if os.path.isfile(local_file): 

98 if file_exists(item=item, local_file=local_file): 

99 return False 

100 # File exists locally but is outdated; need to determine type for re-download 

101 timeout = config_parser.get_drive_request_timeout(config) 

102 item_is_package = is_package(item=item, timeout=timeout) 

103 elif os.path.isdir(local_file): 

104 # A directory at this path means the item was previously downloaded as a 

105 # package. iCloud Drive items do not change type between file and package, 

106 # so package_exists() is the correct check here (no is_package() needed). 

107 # Note: package_exists() deletes the directory if it is outdated. 

108 if package_exists(item=item, local_package_path=local_file): 

109 for f in Path(local_file).glob("**/*"): 

110 files.add(str(f)) 

111 return False 

112 # Directory was deleted by package_exists(); re-download it as a package 

113 item_is_package = True 

114 else: 

115 # Item doesn't exist locally — call is_package() to determine the type 

116 timeout = config_parser.get_drive_request_timeout(config) 

117 item_is_package = is_package(item=item, timeout=timeout) 

118 

119 local_file = download_file(item=item, local_file=local_file) 

120 if local_file and item_is_package: 

121 for f in Path(local_file).glob("**/*"): 

122 f = str(f) 

123 f_normalized = unicodedata.normalize("NFD", f) 

124 if os.path.exists(f): 

125 os.rename(f, f_normalized) 

126 files.add(f_normalized) 

127 return bool(local_file)