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 

13 

14from src import config_parser, configure_icloudpy_logging, get_logger 

15from src.drive_cleanup import remove_obsolete # noqa: F401 

16from src.drive_file_download import download_file # noqa: F401 

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

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

19from src.drive_folder_processing import process_folder # noqa: F401 

20from src.drive_package_processing import process_package # noqa: F401 

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

22from src.drive_sync_directory import sync_directory 

23from src.drive_thread_config import get_max_threads # noqa: F401 

24 

25# Configure icloudpy logging immediately after import 

26configure_icloudpy_logging() 

27 

28LOGGER = get_logger() 

29 

30 

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

32 """Synchronize iCloud Drive to local filesystem. 

33 

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

35 preparing the destination and delegating to the sync_directory orchestrator. 

36 

37 Args: 

38 config: Configuration dictionary containing drive settings 

39 drive: iCloud drive service instance 

40 

41 Returns: 

42 Set of all synchronized file paths 

43 """ 

44 destination_path = config_parser.prepare_drive_destination(config=config) 

45 return sync_directory( 

46 drive=drive, 

47 destination_path=destination_path, 

48 root=destination_path, 

49 items=drive.dir(), 

50 top=True, 

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

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

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

54 config=config, 

55 ) 

56 

57 

58def process_file(item: Any, destination_path: str, filters: list[str], ignore: list[str], files: set[str]) -> bool: 

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

60 

61 This function maintains backward compatibility with existing tests. 

62 New code should use the specialized modules directly. 

63 

64 Args: 

65 item: iCloud file item to process 

66 destination_path: Local destination directory 

67 filters: File extension filters 

68 ignore: Ignore patterns 

69 files: Set to track processed files 

70 

71 Returns: 

72 True if file was processed successfully, False otherwise 

73 """ 

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

75 return False 

76 local_file = os.path.join(destination_path, item.name) 

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

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

79 return False 

80 files.add(local_file) 

81 item_is_package = is_package(item=item) 

82 if item_is_package: 

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

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

85 files.add(str(f)) 

86 return False 

87 elif file_exists(item=item, local_file=local_file): 

88 return False 

89 local_file = download_file(item=item, local_file=local_file) 

90 if local_file and item_is_package: 

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

92 f = str(f) 

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

94 if os.path.exists(f): 

95 os.rename(f, f_normalized) 

96 files.add(f_normalized) 

97 return bool(local_file)