Coverage for src/photo_file_utils.py: 100%
61 statements
« prev ^ index » next coverage.py v7.14.1, created at 2026-06-06 02:49 +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 threading
12from datetime import timezone
14from src import get_logger
16LOGGER = get_logger()
18# Module-level lock to protect thread-safe mutation of photo._versions during retries
19_versions_refresh_lock = threading.Lock()
22def check_photo_exists(photo, file_size: str, local_path: str) -> bool:
23 """Check if photo exists locally with correct size.
25 Args:
26 photo: Photo object from iCloudPy
27 file_size: File size variant (original, medium, thumb, etc.)
28 local_path: Local file path to check
30 Returns:
31 True if photo exists locally with correct size, False otherwise
32 """
33 if not (photo and local_path and os.path.isfile(local_path)):
34 return False
36 local_size = os.path.getsize(local_path)
37 remote_size = int(photo.versions[file_size]["size"])
39 if local_size == remote_size:
40 LOGGER.debug(f"No changes detected. Skipping the file {local_path} ...")
41 return True
42 else:
43 LOGGER.debug(f"Change detected: local_file_size is {local_size} and remote_file_size is {remote_size}.")
44 return False
47def create_hardlink(source_path: str, destination_path: str) -> bool:
48 """Create a hard link from source to destination.
50 Args:
51 source_path: Path to existing file to link from
52 destination_path: Path where hardlink should be created
54 Returns:
55 True if hardlink was created successfully, False otherwise
56 """
57 try:
58 # Ensure destination directory exists
59 os.makedirs(os.path.dirname(destination_path), exist_ok=True)
60 # Create hard link
61 os.link(source_path, destination_path)
62 LOGGER.info(f"Created hard link: {destination_path} (linked to existing file: {source_path})")
63 return True
64 except (OSError, FileNotFoundError) as e:
65 LOGGER.warning(f"Failed to create hard link {destination_path}: {e!s}")
66 return False
69def download_photo_from_server(photo, file_size: str, destination_path: str, max_retries: int = 1) -> bool:
70 """Download photo from iCloud server to local path.
72 This function implements automatic retry logic for HTTP 410 (Gone) errors,
73 which occur when iCloud download URLs expire. When a 410 error is detected,
74 the function clears the cached URLs and retries the download.
76 Args:
77 photo: Photo object from iCloudPy
78 file_size: File size variant (original, medium, thumb, etc.)
79 destination_path: Local path where photo should be saved
80 max_retries: Maximum number of retries on 410 errors (default: 1)
82 Returns:
83 True if download was successful, False otherwise
84 """
85 if not (photo and file_size and destination_path):
86 return False
88 LOGGER.info(f"Downloading {destination_path} ...")
90 max_retries = max(0, max_retries) # Clamp to minimum 0 for predictable behavior
91 attempt = 0
92 max_attempts = max_retries + 1 # Initial attempt + retries
94 while attempt < max_attempts: # noqa: PERF203
95 try:
96 download = photo.download(file_size)
97 with open(destination_path, "wb") as file_out:
98 shutil.copyfileobj(download.raw, file_out)
100 # Set file modification time to photo's added date.
101 # iCloudPy returns added_date as an aware UTC datetime; replace() is a
102 # safe no-op here because tzinfo is already UTC. If it ever returns a
103 # naive datetime, replace(tzinfo=utc) correctly treats it as UTC.
104 local_modified_time = photo.added_date.replace(tzinfo=timezone.utc).timestamp()
105 os.utime(destination_path, (local_modified_time, local_modified_time))
107 return True
109 except Exception as e: # noqa: PERF203
110 # Enhanced error logging with file path context
111 # This catches all exceptions including iCloudPy errors like ObjectNotFoundException
112 error_msg = str(e)
114 # Check for HTTP 410 Gone error - download URL has expired
115 # The iCloudPy library raises exceptions with "Gone (410)" in the message
116 # when the download URL has expired (typically after 30-40 minutes)
117 if "Gone (410)" in error_msg:
118 attempt += 1
119 if attempt < max_attempts:
120 LOGGER.warning(
121 f"Download URL expired (410) for {destination_path}. "
122 f"Refreshing URL and retrying (attempt {attempt}/{max_attempts})...",
123 )
124 # Clear cached versions to force URL refresh on next download attempt
125 # This is necessary because iCloudPy caches the download URLs in _versions
126 # Lock is used to prevent concurrent _versions mutation from parallel threads
127 with _versions_refresh_lock:
128 if hasattr(photo, "_versions"):
129 photo._versions = None # noqa: SLF001
130 continue
131 else:
132 LOGGER.error(
133 f"Failed to download {destination_path} after {max_retries} retries: {error_msg}",
134 )
135 return False
137 # Handle other errors
138 if "ObjectNotFoundException" in error_msg or "NOT_FOUND" in error_msg:
139 LOGGER.error(f"Photo not found in iCloud Photos - {destination_path}: {error_msg}")
140 else:
141 LOGGER.error(f"Failed to download {destination_path}: {error_msg}")
142 return False
144 # This line should never be reached due to the logic above, but is kept as defensive programming
145 return False # pragma: no cover
148def rename_legacy_file_if_exists(old_path: str, new_path: str) -> None:
149 """Rename legacy file format to new format if it exists.
151 Args:
152 old_path: Path to legacy file format
153 new_path: Path to new file format
154 """
155 if os.path.isfile(old_path):
156 os.rename(old_path, new_path)