Coverage for src/drive_file_download.py: 100%

31 statements  

« prev     ^ index     » next       coverage.py v7.10.7, created at 2025-10-16 04:41 +0000

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 

10import time 

11from typing import Any 

12 

13from src import 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(item: Any, local_file: str) -> str | None: 

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

24 

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

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

27 

28 Args: 

29 item: iCloud file item to download 

30 local_file: Local path to save the file 

31 

32 Returns: 

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

34 """ 

35 if not (item and local_file): 

36 return None 

37 

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

39 try: 

40 with item.open(stream=True) as response: 

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

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

43 file_out.write(chunk) 

44 

45 # Check if this is a package that needs processing 

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

47 processed_file = process_package(local_file=local_file) 

48 if processed_file: 

49 local_file = processed_file 

50 else: 

51 return None 

52 

53 # Set the file modification time to match the remote file 

54 item_modified_time = time.mktime(item.date_modified.timetuple()) 

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

56 

57 except Exception as e: 

58 # Enhanced error logging with file path context 

59 # This catches all exceptions including iCloudPy errors like ObjectNotFoundException 

60 error_msg = str(e) 

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

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

63 else: 

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

65 return None 

66 

67 return local_file