feat: Live Photo .mov via explicit file_sizes (live_video_original)

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

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"

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:

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.
 

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,16 +106,33 @@ 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)
 
     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}")
 

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__<id>.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.
 

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__<id>.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)