Coverage for src/photo_file_utils.py: 100%
45 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"""Photo file operations module.
3This module contains utilities for photo file operations including
4downloading, hardlink creation, and file existence checking.
5"""
7___author___ = "Mandar Patil <mandarons@pm.me>"
9import os
10import shutil
11import time
13from src import get_logger
15LOGGER = get_logger()
18def check_photo_exists(photo, file_size: str, local_path: str) -> bool:
19 """Check if photo exists locally with correct size.
21 Args:
22 photo: Photo object from iCloudPy
23 file_size: File size variant (original, medium, thumb, etc.)
24 local_path: Local file path to check
26 Returns:
27 True if photo exists locally with correct size, False otherwise
28 """
29 if not (photo and local_path and os.path.isfile(local_path)):
30 return False
32 local_size = os.path.getsize(local_path)
33 remote_size = int(photo.versions[file_size]["size"])
35 if local_size == remote_size:
36 LOGGER.debug(f"No changes detected. Skipping the file {local_path} ...")
37 return True
38 else:
39 LOGGER.debug(f"Change detected: local_file_size is {local_size} and remote_file_size is {remote_size}.")
40 return False
43def create_hardlink(source_path: str, destination_path: str) -> bool:
44 """Create a hard link from source to destination.
46 Args:
47 source_path: Path to existing file to link from
48 destination_path: Path where hardlink should be created
50 Returns:
51 True if hardlink was created successfully, False otherwise
52 """
53 try:
54 # Ensure destination directory exists
55 os.makedirs(os.path.dirname(destination_path), exist_ok=True)
56 # Create hard link
57 os.link(source_path, destination_path)
58 LOGGER.info(f"Created hard link: {destination_path} (linked to existing file: {source_path})")
59 return True
60 except (OSError, FileNotFoundError) as e:
61 LOGGER.warning(f"Failed to create hard link {destination_path}: {e!s}")
62 return False
65def download_photo_from_server(photo, file_size: str, destination_path: str) -> bool:
66 """Download photo from iCloud server to local path.
68 Args:
69 photo: Photo object from iCloudPy
70 file_size: File size variant (original, medium, thumb, etc.)
71 destination_path: Local path where photo should be saved
73 Returns:
74 True if download was successful, False otherwise
75 """
76 if not (photo and file_size and destination_path):
77 return False
79 LOGGER.info(f"Downloading {destination_path} ...")
80 try:
81 download = photo.download(file_size)
82 with open(destination_path, "wb") as file_out:
83 shutil.copyfileobj(download.raw, file_out)
85 # Set file modification time to photo's added date
86 local_modified_time = time.mktime(photo.added_date.timetuple())
87 os.utime(destination_path, (local_modified_time, local_modified_time))
89 except Exception as e:
90 # Enhanced error logging with file path context
91 # This catches all exceptions including iCloudPy errors like ObjectNotFoundException
92 error_msg = str(e)
93 if "ObjectNotFoundException" in error_msg or "NOT_FOUND" in error_msg:
94 LOGGER.error(f"Photo not found in iCloud Photos - {destination_path}: {error_msg}")
95 else:
96 LOGGER.error(f"Failed to download {destination_path}: {error_msg}")
97 return False
99 return True
102def rename_legacy_file_if_exists(old_path: str, new_path: str) -> None:
103 """Rename legacy file format to new format if it exists.
105 Args:
106 old_path: Path to legacy file format
107 new_path: Path to new file format
108 """
109 if os.path.isfile(old_path):
110 os.rename(old_path, new_path)