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
« prev ^ index » next coverage.py v7.10.7, created at 2025-10-16 04:41 +0000
1"""File download utilities.
3This module provides file downloading functionality for iCloud Drive sync,
4separating download logic from sync operations per SRP.
5"""
7__author__ = "Mandar Patil (mandarons@pm.me)"
9import os
10import time
11from typing import Any
13from src import configure_icloudpy_logging, get_logger
14from src.drive_package_processing import process_package
16# Configure icloudpy logging immediately after import
17configure_icloudpy_logging()
19LOGGER = get_logger()
22def download_file(item: Any, local_file: str) -> str | None:
23 """Download a file from iCloud to local filesystem.
25 This function handles the actual download of files from iCloud, including
26 package detection and processing, and sets the correct modification time.
28 Args:
29 item: iCloud file item to download
30 local_file: Local path to save the file
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
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)
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
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))
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
67 return local_file