Coverage for src/drive_file_download.py: 100%

1"""File download utilities. 

2 

3This module provides file downloading functionality for iCloud Drive sync, 

4separating download logic from sync operations per SRP. 

5""" 

6 

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

8 

9import os 

10from datetime import timezone 

11from typing import Any 

12 

13from src import DEFAULT_REQUEST_TIMEOUT_SEC, configure_icloudpy_logging, get_logger 

14from src.drive_package_processing import process_package 

15 

16# Configure icloudpy logging immediately after import 

17configure_icloudpy_logging() 

18 

19LOGGER = get_logger() 

20 

21 

22def download_file( 

23 item: Any, 

24 local_file: str, 

25 timeout: int = DEFAULT_REQUEST_TIMEOUT_SEC, 

26 flatten_packages: bool = False, 

27) -> str | None: 

28 """Download a file from iCloud to local filesystem. 

29 

30 This function handles the actual download of files from iCloud, including 

31 package detection and processing, and sets the correct modification time. 

32 

33 Args: 

34 item: iCloud file item to download 

35 local_file: Local path to save the file 

36 timeout: HTTP read timeout in seconds. Without one a stalled 

37 connection blocks its worker thread for the life of the 

38 process -- nothing raises, so nothing retries and nothing 

39 restarts, and the sync simply stops. 

40 flatten_packages: When True, package downloads (``/packageDownload?`` 

41 URLs) skip the unpack step entirely and stay on disk as a 

42 single binary file. Useful for backup-style deployments where 

43 bundle-directory semantics aren't needed and single-file 

44 storage simplifies dedup / restoration. Opt-in via 

45 ``drive.flatten_packages: true`` in config.yaml. 

46 

47 Returns: 

48 Path to the downloaded/processed file, or None if download failed 

49 """ 

50 if not (item and local_file): 

51 return None 

52 

53 LOGGER.info(f"Downloading {local_file} ...") 

54 try: 

55 with item.open(stream=True, timeout=timeout) as response: 

56 with open(local_file, "wb") as file_out: 

57 for chunk in response.iter_content(4 * 1024 * 1024): 

58 file_out.write(chunk) 

59 

60 # Check if this is a package that needs processing. 

61 # 

62 # ``process_package`` now returns the local_file path for both 

63 # successful unpacks AND unrecognised mime types (the bytes 

64 # are still on disk as a flat bundle — see the docstring on 

65 # ``process_package`` for the rationale). It only returns 

66 # ``None`` on hard processing failure that leaves the file in 

67 # an unusable state. So we surface that as a download failure 

68 # but no longer treat "couldn't unpack" as failure when the 

69 # downloaded bytes are intact. 

70 if response.url and "/packageDownload?" in response.url: 

71 processed_file = process_package( 

72 local_file=local_file, flatten=flatten_packages, 

73 ) 

74 if processed_file is None: 

75 return None 

76 local_file = processed_file 

77 

78 # Set the file modification time to match the remote file. 

79 # iCloudPy produces date_modified via strptime(..., "%Y-%m-%dT%H:%M:%SZ") — always 

80 # naive UTC with no tzinfo. replace(tzinfo=UTC) is the correct conversion. 

81 item_modified_time = item.date_modified.replace(tzinfo=timezone.utc).timestamp() 

82 os.utime(local_file, (item_modified_time, item_modified_time)) 

83 

84 except Exception as e: 

85 # Enhanced error logging with file path context 

86 # This catches all exceptions including iCloudPy errors like ObjectNotFoundException 

87 error_msg = str(e) 

88 if "ObjectNotFoundException" in error_msg or "NOT_FOUND" in error_msg: 

89 LOGGER.error(f"File not found in iCloud Drive - {local_file}: {error_msg}") 

90 else: 

91 LOGGER.error(f"Failed to download {local_file}: {error_msg}") 

92 return None 

93 

94 return local_file