diff --git a/README.md b/README.md index 0e13cc193..48e287e37 100644 --- a/README.md +++ b/README.md @@ -159,6 +159,10 @@ photos: - "original" # - "medium" # - "thumb" + # For Live Photos, add live_video_original to also download the paired .mov + # (live_video_medium / live_video_thumb for smaller variants). Non-Live + # Photos lack these versions and are skipped. + # - "live_video_original" extensions: # Optional, media extensions to be included in syncing iCloud Photos content # - jpg # - heic diff --git a/config.yaml b/config.yaml index 8a3c0e40d..05e2bdf8e 100644 --- a/config.yaml +++ b/config.yaml @@ -96,3 +96,7 @@ photos: - "original" # - "medium" # - "thumb" + # For Live Photos, add live_video_original to also download the paired .mov + # (live_video_medium / live_video_thumb for smaller variants). Non-Live + # Photos don't have these versions and are skipped. + # - "live_video_original" diff --git a/src/album_sync_orchestrator.py b/src/album_sync_orchestrator.py index e5da95810..5c4d7ec7f 100644 --- a/src/album_sync_orchestrator.py +++ b/src/album_sync_orchestrator.py @@ -137,6 +137,10 @@ def _collect_photo_download_tasks( ) if download_info: tasks.append(download_info) + # Live Photos: add "live_video_original" (or _medium/_thumb) to + # photos.filters.file_sizes to pull the paired .mov. It flows through + # the loop above like any other version; non-Live-Photos don't have + # those versions and are skipped (quietly -- see collect_download_task). return tasks except Exception as e: try: diff --git a/src/config_parser.py b/src/config_parser.py index 4099b0766..6eecebebf 100644 --- a/src/config_parser.py +++ b/src/config_parser.py @@ -210,7 +210,12 @@ def get_drive_sync_interval(config: dict, log_messages: bool = True) -> int: Drive sync interval in seconds """ config_path = ["drive", "sync_interval"] - return get_sync_interval(config=config, config_path=config_path, service_name="drive", log_messages=log_messages) + return get_sync_interval( + config=config, + config_path=config_path, + service_name="drive", + log_messages=log_messages, + ) def get_drive_request_timeout(config: dict) -> int: @@ -241,7 +246,12 @@ def get_photos_sync_interval(config: dict, log_messages: bool = True) -> int: Photos sync interval in seconds """ config_path = ["photos", "sync_interval"] - return get_sync_interval(config=config, config_path=config_path, service_name="photos", log_messages=log_messages) + return get_sync_interval( + config=config, + config_path=config_path, + service_name="photos", + log_messages=log_messages, + ) # ============================================================================= @@ -577,6 +587,12 @@ def get_photos_folder_format(config: dict) -> str | None: def validate_file_sizes(file_sizes: list[str]) -> list[str]: """Validate and filter file sizes against valid options. + Accepts any key in ``PhotoAsset.PHOTO_VERSION_LOOKUP``, including the + ``live_video_*`` keys: add ``live_video_original`` to ``file_sizes`` to + download the paired ``.mov`` of a Live Photo (and ``live_video_medium`` / + ``live_video_thumb`` for smaller variants). Photos that aren't Live Photos + simply don't have those versions and are skipped. + Args: file_sizes: List of file size strings to validate @@ -925,6 +941,7 @@ def get_pushover_api_token(config: dict) -> str | None: """ return get_notification_config_value(config, "pushover", "api_token") + def get_pushover_notification_priority(config: dict) -> int | None: """Return Pushover notification priority from config. diff --git a/src/photo_download_manager.py b/src/photo_download_manager.py index e497bc7d7..87d39c3b1 100644 --- a/src/photo_download_manager.py +++ b/src/photo_download_manager.py @@ -14,6 +14,7 @@ from src.hardlink_registry import HardlinkRegistry from src.photo_file_utils import create_hardlink, download_photo_from_server from src.photo_path_utils import ( + _LIVE_VIDEO_SIZES, create_folder_path_if_needed, generate_photo_filename_with_metadata, normalize_file_path, @@ -29,9 +30,14 @@ class DownloadTaskInfo: """Information about a photo download task.""" - def __init__(self, photo, file_size: str, photo_path: str, - hardlink_source: str | None = None, - hardlink_registry: HardlinkRegistry | None = None): + def __init__( + self, + photo, + file_size: str, + photo_path: str, + hardlink_source: str | None = None, + hardlink_registry: HardlinkRegistry | None = None, + ): """Initialize download task info. Args: @@ -60,8 +66,7 @@ def get_max_threads_for_download(config) -> int: return config_parser.get_app_max_threads(config) -def generate_photo_path(photo, file_size: str, destination_path: str, - folder_format: str | None) -> str: +def generate_photo_path(photo, file_size: str, destination_path: str, folder_format: str | None) -> str: """Generate full file path for photo with legacy file renaming. This function combines path generation, folder creation, and legacy @@ -90,7 +95,7 @@ def generate_photo_path(photo, file_size: str, destination_path: str, file_path = os.path.join(destination_path, filename) file_size_path = os.path.join( destination_path, - f"{'__'.join([name, file_size])}" if extension == "" else f"{'__'.join([name, file_size])}.{extension}", + (f"{'__'.join([name, file_size])}" if extension == "" else f"{'__'.join([name, file_size])}.{extension}"), ) # Final path with normalization @@ -101,6 +106,18 @@ def generate_photo_path(photo, file_size: str, destination_path: str, rename_legacy_file_if_exists(file_path, normalized_path) rename_legacy_file_if_exists(file_size_path, normalized_path) + # Self-heal the earlier .HEIC mislabeling of Live Photo videos: an older + # version wrote the paired video with the still's extension. Rename that + # file to the corrected path instead of re-downloading it (which would also + # leave the broken duplicate behind). + if file_size in _LIVE_VIDEO_SIZES and extension: + root, _ = os.path.splitext(filename_with_metadata) + legacy_mislabeled = normalize_file_path( + os.path.join(final_destination, f"{root}.{extension}"), + ) + if legacy_mislabeled != normalized_path: + rename_legacy_file_if_exists(legacy_mislabeled, normalized_path) + # Handle existing file with different normalization if os.path.isfile(final_file_path) and final_file_path != normalized_path: rename_legacy_file_if_exists(final_file_path, normalized_path) @@ -108,9 +125,14 @@ def generate_photo_path(photo, file_size: str, destination_path: str, return normalized_path -def collect_download_task(photo, file_size: str, destination_path: str, - files: set[str] | None, folder_format: str | None, - hardlink_registry: HardlinkRegistry | None) -> DownloadTaskInfo | None: +def collect_download_task( + photo, + file_size: str, + destination_path: str, + files: set[str] | None, + folder_format: str | None, + hardlink_registry: HardlinkRegistry | None, +) -> DownloadTaskInfo | None: """Collect photo info for parallel download without immediately downloading. Args: @@ -127,7 +149,14 @@ def collect_download_task(photo, file_size: str, destination_path: str, # Check if file size exists on server if file_size not in photo.versions: photo_path = generate_photo_path(photo, file_size, destination_path, folder_format) - LOGGER.warning(f"File size {file_size} not found on server. Skipping the photo {photo_path} ...") + # A missing live_video_* version just means this isn't a Live Photo -- + # expected for most assets, so log at DEBUG to avoid warning-spam when + # live_video_original is in file_sizes. Other sizes warn as before. + msg = f"File size {file_size} not found on server. Skipping the photo {photo_path} ..." + if file_size.startswith("live_video_"): + LOGGER.debug(msg) + else: + LOGGER.warning(msg) return None # Generate photo path @@ -140,6 +169,7 @@ def collect_download_task(photo, file_size: str, destination_path: str, # Check if photo already exists with correct size from src.photo_file_utils import check_photo_exists + if check_photo_exists(photo, file_size, photo_path): return None @@ -183,7 +213,9 @@ def execute_download_task(task_info: DownloadTaskInfo) -> bool: if result and task_info.hardlink_registry is not None: # Register for future hard links if enabled task_info.hardlink_registry.register_photo_path( - task_info.photo.id, task_info.file_size, task_info.photo_path, + task_info.photo.id, + task_info.file_size, + task_info.photo_path, ) LOGGER.debug(f"[Thread] Completed download of {task_info.photo_path}") diff --git a/src/photo_path_utils.py b/src/photo_path_utils.py index dff5bb9c2..3fcfcbff2 100644 --- a/src/photo_path_utils.py +++ b/src/photo_path_utils.py @@ -24,6 +24,10 @@ LOGGER = get_logger() +# The Live Photo paired-video file_size variants. These are QuickTime movies, +# not images, even though the parent asset's filename ends in .HEIC/.JPG. +_LIVE_VIDEO_SIZES = frozenset({"live_video_original", "live_video_medium", "live_video_thumb"}) + def get_photo_name_and_extension(photo, file_size: str) -> tuple[str, str]: """Extract filename and extension from photo. @@ -49,6 +53,16 @@ def get_photo_name_and_extension(photo, file_size: str) -> tuple[str, str]: else: LOGGER.warning(f"Unknown filetype {filetype} for original_alt version of {filename}") + # Handle Live Photo paired-video versions. photo.filename is the STILL + # (e.g. IMG_1234.HEIC), but the live_video_* versions are the QuickTime + # movie half of the Live Photo. Without this the .mov is written with the + # still's extension (IMG_1234__live_video_original__.HEIC), which every + # downstream image tool then rejects as "unsupported image format" because + # it is really a video. Map to the real container extension instead. + elif file_size in _LIVE_VIDEO_SIZES and file_size in photo.versions: + filetype = photo.versions[file_size].get("type") + extension = _get_video_filetype_mapping().get(filetype, "MOV") + return name, extension @@ -116,6 +130,22 @@ def rename_legacy_file_if_exists(old_path: str, new_path: str) -> None: os.rename(old_path, new_path) +def _get_video_filetype_mapping() -> dict: + """Get mapping of Live Photo paired-video Apple UTI types to extensions. + + Live Photo videos are QuickTime movies; iCloud reports the UTI in the + version's ``type`` field. Anything not listed falls back to ``MOV`` (the + only container Apple has ever used for the Live Photo motion component). + + Returns: + Dictionary mapping Apple UTI type strings to file extensions + """ + return { + "com.apple.quicktime-movie": "MOV", + "public.mpeg-4": "MP4", + } + + def _get_original_alt_filetype_mapping() -> dict: """Get mapping of original_alt file types to extensions. diff --git a/tests/test_live_photo_extension.py b/tests/test_live_photo_extension.py new file mode 100644 index 000000000..1e78ac275 --- /dev/null +++ b/tests/test_live_photo_extension.py @@ -0,0 +1,89 @@ +"""Tests for the Live Photo paired-video extension. + +The live_video_* versions are QuickTime/MP4 movies; they must be written with +the real container extension (not the still's .HEIC), and an existing .HEIC +video from an earlier build must be renamed in place rather than re-downloaded. +""" + +import os +import types + +from src.photo_download_manager import generate_photo_path +from src.photo_path_utils import ( + generate_photo_filename_with_metadata, + get_photo_name_and_extension, +) + +_FTYP_QUICKTIME = b"\x00\x00\x00\x18ftypqt \x00\x00\x02\x00qt " + + +def _photo(filename, versions): + """Minimal stand-in for an iCloudPy PhotoAsset.""" + return types.SimpleNamespace(filename=filename, versions=versions, id="ASSET-ID-1") + + +class TestLivePhotoExtension: + def test_live_video_original_maps_to_mov(self): + photo = _photo( + "IMG_1234.HEIC", + { + "original": {"type": "public.heic"}, + "live_video_original": {"type": "com.apple.quicktime-movie"}, + }, + ) + name, ext = get_photo_name_and_extension(photo, "live_video_original") + assert name == "IMG_1234" + assert ext == "MOV" + + def test_live_video_unknown_type_defaults_to_mov(self): + photo = _photo("IMG_1.HEIC", {"live_video_original": {"type": None}}) + assert get_photo_name_and_extension(photo, "live_video_original")[1] == "MOV" + + def test_live_video_mp4_type_maps_to_mp4(self): + photo = _photo("IMG_1.HEIC", {"live_video_original": {"type": "public.mpeg-4"}}) + assert get_photo_name_and_extension(photo, "live_video_original")[1] == "MP4" + + def test_live_video_medium_and_thumb_are_mov(self): + photo = _photo( + "IMG_2.JPG", + { + "live_video_medium": {"type": "com.apple.quicktime-movie"}, + "live_video_thumb": {"type": "com.apple.quicktime-movie"}, + }, + ) + assert get_photo_name_and_extension(photo, "live_video_medium")[1] == "MOV" + assert get_photo_name_and_extension(photo, "live_video_thumb")[1] == "MOV" + + def test_still_original_keeps_its_extension(self): + photo = _photo("IMG_1234.HEIC", {"original": {"type": "public.heic"}}) + assert get_photo_name_and_extension(photo, "original")[1] == "HEIC" + + def test_filename_with_metadata_ends_in_mov(self): + photo = _photo("IMG_9.HEIC", {"live_video_original": {"type": "com.apple.quicktime-movie"}}) + filename = generate_photo_filename_with_metadata(photo, "live_video_original") + assert filename.endswith(".MOV") + assert "__live_video_original__" in filename + + +class TestDownloadSelfHeal: + def test_generate_photo_path_renames_legacy_heic_video(self, tmp_path): + # An existing IMG__live_video_original__.HEIC (from an earlier build) + # must be renamed in place to the corrected .MOV, not left for re-download. + photo = _photo("IMG_1.HEIC", {"live_video_original": {"type": "com.apple.quicktime-movie"}}) + corrected = generate_photo_filename_with_metadata(photo, "live_video_original") + assert corrected.endswith(".MOV") + legacy = os.path.join(str(tmp_path), corrected[: -len(".MOV")] + ".HEIC") + with open(legacy, "wb") as handle: + handle.write(_FTYP_QUICKTIME) + + result = generate_photo_path(photo, "live_video_original", str(tmp_path), None) + + assert result == os.path.join(str(tmp_path), corrected) + assert os.path.exists(result) + assert not os.path.exists(legacy) + + def test_generate_photo_path_no_legacy_is_noop(self, tmp_path): + photo = _photo("IMG_2.HEIC", {"live_video_original": {"type": "com.apple.quicktime-movie"}}) + result = generate_photo_path(photo, "live_video_original", str(tmp_path), None) + assert result.endswith(".MOV") + assert not os.path.exists(result) diff --git a/tests/test_live_photo_pair_download.py b/tests/test_live_photo_pair_download.py new file mode 100644 index 000000000..523542c01 --- /dev/null +++ b/tests/test_live_photo_pair_download.py @@ -0,0 +1,122 @@ +"""Live Photo .mov via explicit ``file_sizes``. + +Users add ``live_video_original`` (or ``live_video_medium`` / ``live_video_thumb``) +to ``photos.filters.file_sizes`` to download the paired ``.mov`` of a Live Photo. +It flows through the normal download loop like any other version; photos that +aren't Live Photos lack the version and are skipped quietly (DEBUG, not WARNING). +""" + +import logging +import unittest +from unittest.mock import MagicMock, patch + +from src import config_parser, photo_download_manager +from src.album_sync_orchestrator import _collect_photo_download_tasks + + +def _fake_photo(filename, versions): + """A photo stub that quacks like icloudpy.services.photos.PhotoAsset.""" + photo = MagicMock() + photo.filename = filename + photo.versions = versions + photo.id = "test-photo-id" + return photo + + +class TestLiveVideoFileSize(unittest.TestCase): + """``live_video_*`` is a valid, explicit file_size.""" + + def test_validate_accepts_live_video_keys(self): + validated = config_parser.validate_file_sizes( + ["original", "live_video_original", "live_video_medium"], + ) + self.assertIn("live_video_original", validated) + self.assertIn("live_video_medium", validated) + + @patch("src.album_sync_orchestrator.is_photo_wanted", return_value=True) + @patch("src.album_sync_orchestrator.collect_download_task") + def test_live_video_original_collected_for_live_photo(self, fake_collect, _wanted): + """A Live Photo with live_video_original requested yields both tasks.""" + photo = _fake_photo( + "IMG_1234.HEIC", + {"original": {"url": "x"}, "live_video_original": {"url": "y"}}, + ) + fake_collect.side_effect = lambda *a, **kw: MagicMock(name=f"task-{a[1]}") + tasks = _collect_photo_download_tasks( + photo, + destination_path="/tmp/dest", + file_sizes=["original", "live_video_original"], + extensions=None, + files=set(), + folder_format=None, + hardlink_registry=None, + ) + self.assertEqual(len(tasks), 2) + called = [c.args[1] for c in fake_collect.call_args_list] + self.assertIn("live_video_original", called) + + @patch("src.album_sync_orchestrator.is_photo_wanted", return_value=True) + @patch("src.album_sync_orchestrator.collect_download_task") + def test_still_photo_skips_absent_live_video(self, fake_collect, _wanted): + """A still (no live_video_original) yields no .mov task; the still still emits.""" + + def side_effect(*args, **kwargs): + return None if args[1] == "live_video_original" else MagicMock(name=f"task-{args[1]}") + + fake_collect.side_effect = side_effect + photo = _fake_photo("IMG_5678.HEIC", {"original": {"url": "x"}}) + tasks = _collect_photo_download_tasks( + photo, + destination_path="/tmp/dest", + file_sizes=["original", "live_video_original"], + extensions=None, + files=set(), + folder_format=None, + hardlink_registry=None, + ) + self.assertEqual(len(tasks), 1) + + +class TestCollectDownloadTaskMissingVersion(unittest.TestCase): + """A version absent from photo.versions is skipped; live_video_* logs DEBUG.""" + + @patch( + "src.photo_download_manager.generate_photo_path", + return_value="/tmp/dest/IMG.HEIC", + ) + def test_missing_live_video_logs_debug_not_warning(self, _gp): + photo = _fake_photo("IMG.HEIC", {"original": {"url": "x"}}) + with self.assertLogs(photo_download_manager.LOGGER, level=logging.DEBUG) as cm: + result = photo_download_manager.collect_download_task( + photo, + "live_video_original", + "/tmp/dest", + set(), + None, + None, + ) + self.assertIsNone(result) + self.assertTrue(any("live_video_original" in line for line in cm.output)) + self.assertFalse(any("WARNING" in line for line in cm.output)) + + @patch( + "src.photo_download_manager.generate_photo_path", + return_value="/tmp/dest/IMG.HEIC", + ) + def test_missing_regular_size_logs_warning(self, _gp): + photo = _fake_photo("IMG.HEIC", {"original": {"url": "x"}}) + with self.assertLogs(photo_download_manager.LOGGER, level=logging.WARNING) as cm: + result = photo_download_manager.collect_download_task( + photo, + "medium", + "/tmp/dest", + set(), + None, + None, + ) + self.assertIsNone(result) + self.assertTrue(any("WARNING" in line for line in cm.output)) + + +if __name__ == "__main__": + unittest.main()