diff --git a/README.md b/README.md index 0e13cc193..95a657042 100644 --- a/README.md +++ b/README.md @@ -144,6 +144,13 @@ photos: all_albums: false # Optional, default false. If true preserve album structure. If same photo is in multiple albums creates duplicates on filesystem use_hardlinks: false # Optional, default false. If true and all_albums is true, create hard links for duplicate photos instead of separate copies. Saves storage space. folder_format: "%Y/%m" # optional, if set put photos in subfolders according to format. Format cheatsheet - https://strftime.org + filename_format: metadata # optional, default "metadata". "metadata" = name__filesize__base64id.ext (legacy). "simple" = plain name.ext (boredazfcuk/Apple-style; lets you migrate without re-downloading) + # file_format: optional single template applied to ALL versions (overrides filename_format). Tokens: + # ${photo.filename} ${photo.ext} ${photo.id} ${photo.file_size} ${photo.year} ${photo.month} ${photo.day} + # ${photo.variant} (empty for original/full, else version) and ${photo.variant_suffix} (separator+variant, only when present). + # Example (keeps plain Apple/boredazfcuk names for originals, so existing files are not re-downloaded): + # file_format: "${photo.filename}${photo.variant_suffix}.${photo.ext}" + # variant_separator: "_" # separator before the variant in ${photo.variant_suffix} (default "_") filters: # List of libraries to download. If omitted (default), photos from all libraries (own and shared) are downloaded. If included, photos only # from the listed libraries are downloaded. diff --git a/config.yaml b/config.yaml index 8a3c0e40d..6b4dc69ac 100644 --- a/config.yaml +++ b/config.yaml @@ -80,6 +80,12 @@ photos: all_albums: false # Optional, default false. If true preserve album structure. If same photo is in multiple albums creates duplicates on filesystem use_hardlinks: false # Optional, default false. If true and all_albums is true, create hard links for duplicate photos instead of separate copies. Saves storage space. # folder_format: "%Y/%m" # optional, if set put photos in subfolders according to format. Format cheatsheet - https://strftime.org + # filename_format: metadata # optional, default "metadata". "metadata" = name__filesize__base64id.ext (legacy). "simple" = plain name.ext (boredazfcuk/Apple-style; lets you migrate without re-downloading) + # file_format: # optional, single template applied to ALL versions (overrides filename_format). + # Tokens: ${photo.filename} ${photo.ext} ${photo.id} ${photo.file_size} ${photo.year} ${photo.month} ${photo.day} + # ${photo.variant} (empty for original/full, else version) and ${photo.variant_suffix} (separator+variant, only when present). + # Example (keeps plain Apple/boredazfcuk names for originals -> no re-download): "${photo.filename}${photo.variant_suffix}.${photo.ext}" + # variant_separator: "_" # optional, separator before the variant in ${photo.variant_suffix} (default "_") filters: # List of libraries to download. If omitted (default), photos from all libraries (own and shared) are downloaded. If included, photos only # from the listed libraries are downloaded. diff --git a/src/config_parser.py b/src/config_parser.py index 4099b0766..ddd8400ec 100644 --- a/src/config_parser.py +++ b/src/config_parser.py @@ -599,6 +599,70 @@ def validate_file_sizes(file_sizes: list[str]) -> list[str]: return validated_sizes if validated_sizes else ["original"] +def get_photos_filename_format(config: dict) -> str: + """Filename naming convention for downloaded photos. + + - ``"metadata"`` (default, backward-compatible): ``name__filesize__base64id.ext`` + — mandarons' historical format. Encodes the CloudKit asset ID into the + filename so the same source photo can be unambiguously round-tripped to + its iCloud record from the local file alone. + - ``"simple"``: ``name.ext`` — the boredazfcuk/docker-icloudpd convention + (and Apple's own download-from-icloud.com convention). Drops the + metadata suffix entirely. Trade-off: rename collisions are possible + (two distinct iCloud photos sharing the same human filename land on + the same path on disk; ``collect_download_task`` detects this and + falls back to the metadata-suffix name for the colliding photo so + both files coexist). + Useful for users migrating from boredazfcuk who want to point + mandarons at their existing photo tree without triggering a full + re-download — ``check_photo_exists`` compares by file size, not by + filename suffix, so matching simple-format files are correctly + treated as already-present. + + Returns: + Either ``"metadata"`` (default) or ``"simple"``. + """ + config_path = ["photos", "filename_format"] + value = get_config_value_or_none(config=config, config_path=config_path) + if value is None: + return "metadata" + value = str(value).lower().strip() + if value not in ("metadata", "simple"): + log_config_not_found_warning( + config_path, f"unknown value {value!r}; falling back to 'metadata'", + ) + return "metadata" + return value + + +def get_photos_file_format(config: dict) -> str | None: + """Single filename template applied to all versions (``photos.file_format``). + + Mirrors ``folder_format`` but for the filename. Uses ``${photo.*}`` tokens + (see ``photo_path_utils.render_filename_template``). When set it overrides + ``filename_format``. Returns ``None`` (use ``filename_format``) when unset. + """ + config_path = ["photos", "file_format"] + value = get_config_value_or_none(config=config, config_path=config_path) + if value is None: + return None + if not isinstance(value, str) or not value.strip(): + log_config_not_found_warning(config_path, "must be a non-empty template string; ignoring") + return None + return value + + +def get_photos_variant_separator(config: dict) -> str: + """Separator inserted before the variant in ``${photo.variant_suffix}``. + + Defaults to ``"_"``. Only appears when a version is non-primary (not + original/full), so primaries stay un-suffixed. + """ + config_path = ["photos", "variant_separator"] + value = get_config_value_or_none(config=config, config_path=config_path) + return str(value) if value is not None else "_" + + def get_photos_libraries_filter(config: dict, base_config_path: list[str]) -> list[str] | None: """Get libraries filter from photos config. diff --git a/src/photo_download_manager.py b/src/photo_download_manager.py index e497bc7d7..2f89e73e2 100644 --- a/src/photo_download_manager.py +++ b/src/photo_download_manager.py @@ -16,6 +16,8 @@ from src.photo_path_utils import ( create_folder_path_if_needed, generate_photo_filename_with_metadata, + get_default_filename_format, + get_file_format, normalize_file_path, rename_legacy_file_if_exists, ) @@ -29,9 +31,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 +67,9 @@ 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 @@ -80,7 +88,9 @@ def generate_photo_path(photo, file_size: str, destination_path: str, filename_with_metadata = generate_photo_filename_with_metadata(photo, file_size) # Create folder path if needed - final_destination = create_folder_path_if_needed(destination_path, folder_format, photo) + final_destination = create_folder_path_if_needed( + destination_path, folder_format, photo, + ) # Generate paths for legacy file format handling filename = photo.filename @@ -90,7 +100,11 @@ 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 @@ -108,9 +122,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: @@ -126,23 +145,74 @@ 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} ...") + 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} ...", + ) return None # Generate photo path photo_path = generate_photo_path(photo, file_size, destination_path, folder_format) - # Thread-safe file set update - if files is not None: - with files_lock: - files.add(photo_path) - # 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): + if files is not None: + with files_lock: + files.add(photo_path) return None + # Filename-collision fallback (``simple`` mode only): a plain + # ``IMG_1234.HEIC`` path may already be claimed by a DIFFERENT iCloud + # photo that happens to share the human filename. Two collision sources: + # + # 1. On-disk collision: ``check_photo_exists`` returned False but the + # path is occupied -- size mismatch with a photo from a prior sync. + # 2. In-flight collision: an earlier call in this same + # ``_collect_album_download_tasks`` pass already claimed this plain + # path. Without this check the later parallel download silently + # overwrites the earlier one and we lose data. ``collect_download_task`` + # runs sequentially during collection so a plain ``in files`` membership + # test under ``files_lock`` is sufficient; we hold the lock only long + # enough to read. + # + # In either case we route this photo to the metadata-suffix filename so + # both files coexist and both round-trip stably on future syncs. We use + # the ``get_default_filename_format()`` accessor (rather than importing + # the module-level constant) so ``set_default_filename_format`` updates + # are observed live on every call. + # Non-unique naming (simple mode, or a photos.file_format template that may + # omit a unique component) can collide; the metadata format cannot. + is_non_unique = get_default_filename_format() == "simple" or get_file_format() is not None + in_flight_collision = False + if is_non_unique and files is not None: + with files_lock: + in_flight_collision = photo_path in files + if is_non_unique and (os.path.isfile(photo_path) or in_flight_collision): + suffix_folder = create_folder_path_if_needed( + destination_path, folder_format, photo, + ) + suffix_basename = generate_photo_filename_with_metadata( + photo, file_size, "metadata", + ) + photo_path = normalize_file_path(os.path.join(suffix_folder, suffix_basename)) + LOGGER.info( + f"Filename collision for {photo.filename} (id={photo.id}); " + f"using suffix path {photo_path} to preserve both photos.", + ) + if check_photo_exists(photo, file_size, photo_path): + if files is not None: + with files_lock: + files.add(photo_path) + return None + + if files is not None: + with files_lock: + files.add(photo_path) + # Check for existing hardlink source hardlink_source = None if hardlink_registry is not None: @@ -176,14 +246,20 @@ def execute_download_task(task_info: DownloadTaskInfo) -> bool: return True else: # Fallback to download if hard link creation fails - LOGGER.warning(f"Hard link creation failed, downloading {task_info.photo_path} instead") + LOGGER.warning( + f"Hard link creation failed, downloading {task_info.photo_path} instead", + ) # Download the photo - result = download_photo_from_server(task_info.photo, task_info.file_size, task_info.photo_path) + result = download_photo_from_server( + task_info.photo, task_info.file_size, task_info.photo_path, + ) 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}") @@ -194,7 +270,9 @@ def execute_download_task(task_info: DownloadTaskInfo) -> bool: return False -def execute_parallel_downloads(download_tasks: list[DownloadTaskInfo], config) -> tuple[int, int]: +def execute_parallel_downloads( + download_tasks: list[DownloadTaskInfo], config, +) -> tuple[int, int]: """Execute download tasks in parallel using thread pool. Args: @@ -228,7 +306,10 @@ def execute_parallel_downloads(download_tasks: list[DownloadTaskInfo], config) - with ThreadPoolExecutor(max_workers=max_threads) as executor: # Submit all download tasks - future_to_task = {executor.submit(execute_download_task, task): task for task in download_tasks} + future_to_task = { + executor.submit(execute_download_task, task): task + for task in download_tasks + } # Process completed downloads for future in as_completed(future_to_task): @@ -242,5 +323,7 @@ def execute_parallel_downloads(download_tasks: list[DownloadTaskInfo], config) - LOGGER.error(f"Unexpected error during photo download: {e!s}") failed_downloads += 1 - LOGGER.info(f"Photo processing complete: {successful_downloads} successful, {failed_downloads} failed") + LOGGER.info( + f"Photo processing complete: {successful_downloads} successful, {failed_downloads} failed", + ) return successful_downloads, failed_downloads diff --git a/src/photo_path_utils.py b/src/photo_path_utils.py index dff5bb9c2..db419497c 100644 --- a/src/photo_path_utils.py +++ b/src/photo_path_utils.py @@ -17,6 +17,7 @@ import base64 import os +import re import unicodedata from urllib.parse import unquote @@ -24,6 +25,9 @@ LOGGER = get_logger() +# ``${photo.*}`` tokens for photos.file_format; unknown tokens are left literal. +_TEMPLATE_TOKEN_RE = re.compile(r"\$\{([^}]+)\}") + def get_photo_name_and_extension(photo, file_size: str) -> tuple[str, str]: """Extract filename and extension from photo. @@ -47,31 +51,150 @@ def get_photo_name_and_extension(photo, file_size: str) -> tuple[str, str]: if filetype in _get_original_alt_filetype_mapping(): extension = _get_original_alt_filetype_mapping()[filetype] else: - LOGGER.warning(f"Unknown filetype {filetype} for original_alt version of {filename}") + LOGGER.warning( + f"Unknown filetype {filetype} for original_alt version of {filename}", + ) return name, extension -def generate_photo_filename_with_metadata(photo, file_size: str) -> str: - """Generate filename with file size and photo ID metadata. +# Module-level default filename format. ``sync_photos`` sets this once at +# the start of a sync run via ``set_default_filename_format`` so the value +# threads through ``generate_photo_path`` -> ``collect_download_task`` without +# requiring a config argument on every downstream signature. +_DEFAULT_FILENAME_FORMAT = "metadata" + + +def set_default_filename_format(filename_format: str) -> None: + """Set the module-level default filename format. See get_photos_filename_format.""" + global _DEFAULT_FILENAME_FORMAT + if filename_format in ("metadata", "simple"): + _DEFAULT_FILENAME_FORMAT = filename_format + + +def get_default_filename_format() -> str: + """Read the current module-level default filename format. + + Public accessor so callers in other modules can observe live updates + from ``set_default_filename_format`` without reaching into the + underscore-prefixed module global (which violates SLF001). + """ + return _DEFAULT_FILENAME_FORMAT + + +# photos.file_format: a single template applied to EVERY downloaded version +# (mirrors folder_format). None means "no template; use filename_format". +_FILE_FORMAT: str | None = None +_VARIANT_SEPARATOR = "_" +# Versions that are the "primary" copy and get no variant tag in ${photo.variant*}. +_PRIMARY_VERSIONS = frozenset({"original", "full"}) + + +def set_file_format(template: str | None, variant_separator: str = "_") -> None: + """Set the single filename template + variant separator (photos.file_format).""" + global _FILE_FORMAT, _VARIANT_SEPARATOR + _FILE_FORMAT = template or None + _VARIANT_SEPARATOR = variant_separator if variant_separator is not None else "_" + + +def get_file_format() -> str | None: + """Read the single filename template (public accessor; avoids SLF001).""" + return _FILE_FORMAT + + +def render_filename_template(template: str, photo, file_size: str) -> str: + """Render photos.file_format for one photo version. + + Tokens (unknown ones left literal): ``${photo.filename}`` ``${photo.ext}`` + ``${photo.id}`` (base64url asset id) ``${photo.file_size}`` and the + created-date parts ``${photo.year}`` ``${photo.month}`` ``${photo.day}``. + + Variant tokens are empty for the primary versions (original/full) and the + version name otherwise: ``${photo.variant}`` (bare, e.g. ``medium``) and + ``${photo.variant_suffix}`` (separator + variant, e.g. ``_medium``) so the + separator only appears when there actually is a variant. + """ + name, extension = get_photo_name_and_extension(photo, file_size) + created = getattr(photo, "created", None) + is_primary = file_size in _PRIMARY_VERSIONS + variant = "" if is_primary else file_size + variant_suffix = "" if is_primary else f"{_VARIANT_SEPARATOR}{file_size}" + values = { + "photo.filename": name, + "photo.ext": extension, + "photo.id": base64.urlsafe_b64encode(photo.id.encode()).decode(), + "photo.file_size": file_size, + "photo.variant": variant, + "photo.variant_suffix": variant_suffix, + "photo.year": f"{created.year:04d}" if created else "", + "photo.month": f"{created.month:02d}" if created else "", + "photo.day": f"{created.day:02d}" if created else "", + } + + def _sub(match: re.Match) -> str: + token = match.group(1).strip() + replacement = values.get(token) + return replacement if replacement is not None else match.group(0) + + return _TEMPLATE_TOKEN_RE.sub(_sub, template) + + +def generate_photo_filename_with_metadata( + photo, file_size: str, filename_format: str | None = None, +) -> str: + """Generate filename for a photo asset. + + Two conventions supported (controlled by ``filename_format`` or the + module-level default set by ``set_default_filename_format``): + + - ``"metadata"`` (default): ``name__filesize__base64id.extension`` — + mandarons' historical format, encodes CloudKit asset id into the filename. + - ``"simple"``: ``name.extension`` — boredazfcuk/Apple convention. Lets + users migrate from boredazfcuk-format trees without re-downloading. + ``collect_download_task`` detects collisions and falls back to the + metadata-suffix path for the colliding photo so both files coexist. Args: photo: Photo object from iCloudPy file_size: File size variant (original, medium, thumb, etc.) + filename_format: ``"metadata"`` or ``"simple"``, or ``None`` to + use the module-level default. Returns: - Filename string with format: name__filesize__base64id.extension + Filename string in the chosen format. """ + forced = filename_format # explicit value; None means a normal (non-fallback) call + if filename_format is None: + filename_format = _DEFAULT_FILENAME_FORMAT name, extension = get_photo_name_and_extension(photo, file_size) - photo_id_encoded = base64.urlsafe_b64encode(photo.id.encode()).decode() + # photos.file_format wins for normal calls; the collision fallback passes + # filename_format="metadata" explicitly to force the always-unique name. + if forced != "metadata" and _FILE_FORMAT is not None: + rendered = render_filename_template(_FILE_FORMAT, photo, file_size) + if rendered.strip(): + return rendered + # Defensive: a template that renders empty for this version (e.g. a bare + # ${photo.variant} on an original) would otherwise yield a path pointing + # at the directory. Fall back to filename_format instead. + LOGGER.warning( + f"photos.file_format rendered an empty name for {file_size}; " + f"falling back to {filename_format} naming.", + ) + + if filename_format == "simple": + return name if extension == "" else f"{name}.{extension}" + + photo_id_encoded = base64.urlsafe_b64encode(photo.id.encode()).decode() if extension == "": return f"{'__'.join([name, file_size, photo_id_encoded])}" else: return f"{'__'.join([name, file_size, photo_id_encoded])}.{extension}" -def create_folder_path_if_needed(destination_path: str, folder_format: str | None, photo) -> str: +def create_folder_path_if_needed( + destination_path: str, folder_format: str | None, photo, +) -> str: """Create folder path based on folder format and photo creation date. Args: diff --git a/src/sync_photos.py b/src/sync_photos.py index abedc0108..b72197b12 100644 --- a/src/sync_photos.py +++ b/src/sync_photos.py @@ -385,6 +385,15 @@ def sync_photos(config, photos): """ # Parse configuration using centralized config parser destination_path = config_parser.prepare_photos_destination(config=config) + # Apply the requested filename convention for this sync run. The module-level + # default in photo_path_utils flows down through generate_photo_path -> all + # downloads without requiring a per-call signature change. + from src.photo_path_utils import set_default_filename_format, set_file_format + set_default_filename_format(config_parser.get_photos_filename_format(config=config)) + set_file_format( + config_parser.get_photos_file_format(config=config), + config_parser.get_photos_variant_separator(config=config), + ) filters = config_parser.get_photos_filters(config=config) files = set() download_all = config_parser.get_photos_all_albums(config=config) diff --git a/tests/test_collision_fallback.py b/tests/test_collision_fallback.py new file mode 100644 index 000000000..c90afd72e --- /dev/null +++ b/tests/test_collision_fallback.py @@ -0,0 +1,209 @@ +"""Filename-collision fallback in simple format. + +Added 2026-05-27. When two distinct iCloud photos share a human filename +(e.g. both are ``IMG_1234.HEIC`` because two iPhones reset their counters), +the second one must NOT silently overwrite the first. Falls back to the +metadata-suffix path for the colliding photo so both files coexist. + +Without this, ``filename_format: simple`` would lose data — unacceptable +for a backup tool. With it, plain photos still get plain names; only +collisions get suffixes. +""" + +import os +import tempfile +import unittest +from unittest.mock import MagicMock + + +def _fake_photo(filename, photo_id, original_size): + photo = MagicMock() + photo.filename = filename + photo.id = photo_id + photo.versions = {"original": {"type": "public.heic", "size": original_size}} + photo.created = MagicMock() + return photo + + +class TestCollisionFallback(unittest.TestCase): + """collect_download_task routes colliding photos to suffix names.""" + + def setUp(self): + # Make sure we're in simple mode for these tests. + from src.photo_path_utils import set_default_filename_format + + set_default_filename_format("simple") + self.tmp = tempfile.mkdtemp() + + def tearDown(self): + from src.photo_path_utils import set_default_filename_format + + set_default_filename_format("metadata") + import shutil + + shutil.rmtree(self.tmp, ignore_errors=True) + + def test_no_collision_uses_simple_path(self): + """First-ever download with no existing file → plain name, task returned.""" + from src.photo_download_manager import collect_download_task + + photo = _fake_photo("IMG_AAAA.HEIC", "id-aaaa", 12345) + task = collect_download_task( + photo, + "original", + self.tmp, + set(), + folder_format=None, + hardlink_registry=None, + ) + assert task is not None + assert task.photo_path.endswith("IMG_AAAA.HEIC") + # Suffix form NOT in the path + assert "__original__" not in task.photo_path + + def test_same_photo_resync_skips(self): + """Same photo (matching size at plain path) → no task (skip).""" + from src.photo_download_manager import collect_download_task + + photo = _fake_photo("IMG_BBBB.HEIC", "id-bbbb", 12345) + # Pre-create the file at the plain path with matching size + plain_path = os.path.join(self.tmp, "IMG_BBBB.HEIC") + with open(plain_path, "wb") as f: + f.write(b"x" * 12345) + + task = collect_download_task( + photo, + "original", + self.tmp, + set(), + folder_format=None, + hardlink_registry=None, + ) + assert task is None # already-present photo, skipped + + def test_collision_falls_back_to_suffix(self): + """Plain path occupied by a DIFFERENT photo → suffix path used. + + Simulates the iCloud-collision scenario: two distinct iCloud photos + named ``IMG_CCCC.HEIC`` with different sizes. The first lives at the + plain path; the second must NOT overwrite it. + """ + from src.photo_download_manager import collect_download_task + + # Pre-create the "first" photo at the plain path (e.g. from a prior + # boredazfcuk sync) — 9999 bytes. + plain_path = os.path.join(self.tmp, "IMG_CCCC.HEIC") + with open(plain_path, "wb") as f: + f.write(b"x" * 9999) + + # Current photo is a DIFFERENT iCloud asset with the same human name + # but a different size (12345 bytes). + photo = _fake_photo("IMG_CCCC.HEIC", "different-id-cccc", 12345) + + task = collect_download_task( + photo, + "original", + self.tmp, + set(), + folder_format=None, + hardlink_registry=None, + ) + # Task returned — colliding photo gets the suffix path. + assert task is not None + assert task.photo_path != plain_path # NOT overwriting the first file + assert "__original__" in task.photo_path + assert task.photo_path.endswith(".HEIC") + # First-photo file at plain path is untouched (still 9999 bytes) + assert os.path.getsize(plain_path) == 9999 + + def test_collision_suffix_already_downloaded_skips(self): + """If the suffix path also already exists with matching size → skip. + + This is the second-and-later-sync of a previously-collided photo: + plain path belongs to photo A, suffix path belongs to photo B, both + are stable. + """ + from src.photo_download_manager import collect_download_task + + # Photo A occupies the plain path + plain_path = os.path.join(self.tmp, "IMG_DDDD.HEIC") + with open(plain_path, "wb") as f: + f.write(b"a" * 100) + + # Photo B (different photo, same name) — must go to suffix path + photo_b = _fake_photo("IMG_DDDD.HEIC", "photo-b-id", 200) + + # Pre-create the suffix path with photo B's size + import base64 as _b64 + + b64id = _b64.urlsafe_b64encode(b"photo-b-id").decode() + suffix_path = os.path.join(self.tmp, f"IMG_DDDD__original__{b64id}.HEIC") + with open(suffix_path, "wb") as f: + f.write(b"b" * 200) + + task = collect_download_task( + photo_b, + "original", + self.tmp, + set(), + folder_format=None, + hardlink_registry=None, + ) + # Both photos are already on disk — no work needed. + assert task is None + + def test_in_flight_collision_within_single_sync(self): + """The same-sync, both-new race: two distinct iCloud photos with + identical human filename, neither on disk yet. Without the in-flight + check, both ``collect_download_task`` calls return tasks pointing at + the SAME plain path; the later parallel download silently overwrites + the earlier one and we lose data (CWE-200-ish for a backup tool). + + Models the scenario by passing the FIRST task's plain path as a + pre-populated entry in the shared ``files`` set (which is what the + real collection loop does between sequential calls), then asserting + the SECOND call sees the in-flight collision and routes to a suffix. + """ + from src.photo_download_manager import collect_download_task + + # First photo's plain path is already in the in-flight set from a + # prior collect_download_task call earlier in this same sync. + plain_path = os.path.join(self.tmp, "IMG_INFLIGHT.HEIC") + in_flight = {plain_path} + + # Second photo, same human name, different iCloud id — must NOT + # produce a task pointing at the same plain path. + photo_b = _fake_photo("IMG_INFLIGHT.HEIC", "second-photo-id", 9876) + task = collect_download_task( + photo_b, + "original", + self.tmp, + in_flight, + folder_format=None, + hardlink_registry=None, + ) + assert task is not None + assert task.photo_path != plain_path # NOT overwriting the first + assert "__original__" in task.photo_path + + def test_metadata_mode_no_collision_logic(self): + """In metadata (default) mode, collision logic is bypassed — already- + unique suffix names cannot collide.""" + from src.photo_path_utils import set_default_filename_format + + set_default_filename_format("metadata") + + from src.photo_download_manager import collect_download_task + + photo = _fake_photo("IMG_EEEE.HEIC", "id-eeee", 12345) + task = collect_download_task( + photo, + "original", + self.tmp, + set(), + folder_format=None, + hardlink_registry=None, + ) + assert task is not None + # In metadata mode the path includes the suffix from the start + assert "__original__" in task.photo_path diff --git a/tests/test_file_formats.py b/tests/test_file_formats.py new file mode 100644 index 000000000..034bb12e8 --- /dev/null +++ b/tests/test_file_formats.py @@ -0,0 +1,227 @@ +"""photos.file_format — single filename template applied to all versions.""" + +import base64 +import datetime +import os +import shutil +import tempfile +import unittest +from unittest.mock import MagicMock + +from src import config_parser, photo_path_utils + + +def _photo( + filename="IMG_1234.HEIC", + pid="ABC/123", + created: datetime.datetime | None = datetime.datetime(2021, 3, 7), +): + photo = MagicMock() + photo.filename = filename + photo.id = pid + photo.created = created + photo.versions = {"original": {"type": "public.heic"}} + return photo + + +class TestRenderFilenameTemplate(unittest.TestCase): + """render_filename_template token substitution + variant handling.""" + + def tearDown(self): + photo_path_utils.set_file_format(None) # reset separator/template globals + + def test_primary_version_has_empty_variant(self): + tmpl = "${photo.year}.${photo.month}.${photo.day} ${photo.filename}${photo.variant_suffix}.${photo.ext}" + self.assertEqual( + photo_path_utils.render_filename_template(tmpl, _photo(), "original"), + "2021.03.07 IMG_1234.HEIC", + ) + + def test_full_is_also_primary(self): + out = photo_path_utils.render_filename_template( + "${photo.filename}${photo.variant_suffix}", + _photo(), + "full", + ) + self.assertEqual(out, "IMG_1234") + + def test_variant_version_is_suffixed(self): + out = photo_path_utils.render_filename_template( + "${photo.filename}${photo.variant_suffix}.${photo.ext}", + _photo(), + "medium", + ) + self.assertEqual(out, "IMG_1234_medium.HEIC") + + def test_bare_variant_token(self): + out = photo_path_utils.render_filename_template( + "${photo.filename}--${photo.variant}", + _photo(), + "thumb", + ) + self.assertEqual(out, "IMG_1234--thumb") + + def test_configurable_separator(self): + photo_path_utils.set_file_format("x", variant_separator="-") + out = photo_path_utils.render_filename_template( + "${photo.filename}${photo.variant_suffix}", + _photo(), + "medium", + ) + self.assertEqual(out, "IMG_1234-medium") + + def test_id_and_file_size_tokens(self): + out = photo_path_utils.render_filename_template( + "${photo.id}__${photo.file_size}", + _photo(), + "original", + ) + self.assertEqual( + out, + f"{base64.urlsafe_b64encode(b'ABC/123').decode()}__original", + ) + + def test_unknown_token_left_literal(self): + out = photo_path_utils.render_filename_template( + "${photo.bogus}-${photo.filename}", + _photo(), + "original", + ) + self.assertEqual(out, "${photo.bogus}-IMG_1234") + + def test_missing_created_renders_blank(self): + out = photo_path_utils.render_filename_template( + "${photo.year}-${photo.filename}", + _photo(created=None), + "original", + ) + self.assertEqual(out, "-IMG_1234") + + +class TestSetFileFormat(unittest.TestCase): + def tearDown(self): + photo_path_utils.set_file_format(None) + + def test_set_and_get(self): + photo_path_utils.set_file_format("tmpl") + self.assertEqual(photo_path_utils.get_file_format(), "tmpl") + + def test_empty_clears(self): + photo_path_utils.set_file_format("tmpl") + photo_path_utils.set_file_format(None) + self.assertIsNone(photo_path_utils.get_file_format()) + + def test_none_separator_defaults_to_underscore(self): + photo_path_utils.set_file_format( + "${photo.filename}${photo.variant_suffix}", + variant_separator=None, + ) + out = photo_path_utils.render_filename_template( + "${photo.filename}${photo.variant_suffix}", + _photo(), + "medium", + ) + self.assertEqual(out, "IMG_1234_medium") + + +class TestGenerateWithFileFormat(unittest.TestCase): + def tearDown(self): + photo_path_utils.set_file_format(None) + + def test_file_format_applies(self): + photo_path_utils.set_file_format( + "${photo.filename}${photo.variant_suffix}.${photo.ext}", + ) + self.assertEqual( + photo_path_utils.generate_photo_filename_with_metadata( + _photo(), + "original", + ), + "IMG_1234.HEIC", + ) + + def test_forced_metadata_bypasses_file_format(self): + photo_path_utils.set_file_format("${photo.filename}.${photo.ext}") + out = photo_path_utils.generate_photo_filename_with_metadata( + _photo(), + "original", + "metadata", + ) + self.assertIn("__original__", out) + + def test_empty_render_falls_back(self): + # A template that renders empty for this version must not yield an empty + # name; it falls back to filename_format (metadata default here). + photo_path_utils.set_file_format("${photo.variant}") + out = photo_path_utils.generate_photo_filename_with_metadata(_photo(), "original") + self.assertIn("__original__", out) + + +class TestConfigGetters(unittest.TestCase): + def test_file_format_absent(self): + self.assertIsNone(config_parser.get_photos_file_format({"photos": {}})) + + def test_file_format_blank_or_nonstring(self): + self.assertIsNone( + config_parser.get_photos_file_format({"photos": {"file_format": " "}}), + ) + self.assertIsNone( + config_parser.get_photos_file_format({"photos": {"file_format": 123}}), + ) + + def test_file_format_valid(self): + self.assertEqual( + config_parser.get_photos_file_format( + {"photos": {"file_format": "${photo.filename}"}}, + ), + "${photo.filename}", + ) + + def test_variant_separator_default_and_custom(self): + self.assertEqual( + config_parser.get_photos_variant_separator({"photos": {}}), + "_", + ) + self.assertEqual( + config_parser.get_photos_variant_separator( + {"photos": {"variant_separator": "-"}}, + ), + "-", + ) + + +class TestFileFormatCollisionFallback(unittest.TestCase): + """A templated name that collides falls back to the unique metadata name.""" + + def setUp(self): + photo_path_utils.set_file_format("${photo.filename}.${photo.ext}") + self.tmp = tempfile.mkdtemp() + + def tearDown(self): + photo_path_utils.set_file_format(None) + shutil.rmtree(self.tmp, ignore_errors=True) + + def test_collision_routes_to_metadata(self): + from src.photo_download_manager import collect_download_task + + occupied = os.path.join(self.tmp, "IMG_EEEE.HEIC") + with open(occupied, "wb") as f: + f.write(b"x" * 9999) + photo = _photo(filename="IMG_EEEE.HEIC", pid="different-id") + photo.versions = {"original": {"type": "public.heic", "size": 12345}} + + task = collect_download_task( + photo, + "original", + self.tmp, + set(), + folder_format=None, + hardlink_registry=None, + ) + self.assertIsNotNone(task) + self.assertIn("__original__", task.photo_path) + self.assertEqual(os.path.getsize(occupied), 9999) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_filename_format.py b/tests/test_filename_format.py new file mode 100644 index 000000000..6ef297051 --- /dev/null +++ b/tests/test_filename_format.py @@ -0,0 +1,179 @@ +"""Tests for the optional ``photos.filename_format`` config knob. + +Added 2026-05-27. Lets users migrating from boredazfcuk/docker-icloudpd +(which uses plain ``IMG_1234.HEIC`` naming) point this container at their +existing photo tree without triggering a full re-download. +""" + +import unittest +from unittest.mock import MagicMock + +from src import config_parser +from src.photo_path_utils import ( + _DEFAULT_FILENAME_FORMAT, # noqa: F401 (re-imported below to assert default) + generate_photo_filename_with_metadata, + set_default_filename_format, +) + + +def _fake_photo(filename="IMG_1234.HEIC", photo_id="some-cloudkit-id"): + photo = MagicMock() + photo.filename = filename + photo.id = photo_id + photo.versions = {"original": {"type": "public.heic"}} + return photo + + +class TestGetPhotosFilenameFormat(unittest.TestCase): + def test_default_is_metadata(self): + assert config_parser.get_photos_filename_format({}) == "metadata" + + def test_simple_is_accepted(self): + config = {"photos": {"filename_format": "simple"}} + assert config_parser.get_photos_filename_format(config) == "simple" + + def test_uppercase_is_normalised(self): + config = {"photos": {"filename_format": "SIMPLE"}} + assert config_parser.get_photos_filename_format(config) == "simple" + + def test_unknown_value_falls_back_to_metadata(self): + config = {"photos": {"filename_format": "weird-thing"}} + assert config_parser.get_photos_filename_format(config) == "metadata" + + +class TestGeneratePhotoFilename(unittest.TestCase): + def setUp(self): + # Reset module-level default before each test so leakage between + # tests doesn't matter. + set_default_filename_format("metadata") + + def test_metadata_format_produces_suffixed_name(self): + photo = _fake_photo("IMG_1234.HEIC", "abc") + name = generate_photo_filename_with_metadata(photo, "original", "metadata") + # ``name__filesize__base64id.ext`` + assert name.startswith("IMG_1234__original__") + assert name.endswith(".HEIC") + + def test_simple_format_produces_plain_name(self): + photo = _fake_photo("IMG_1234.HEIC", "abc") + name = generate_photo_filename_with_metadata(photo, "original", "simple") + assert name == "IMG_1234.HEIC" + + def test_simple_format_handles_extensionless_filename(self): + photo = _fake_photo("noextension", "abc") + name = generate_photo_filename_with_metadata(photo, "original", "simple") + assert name == "noextension" + + def test_explicit_none_uses_module_default(self): + photo = _fake_photo("IMG_1234.HEIC", "abc") + set_default_filename_format("simple") + name = generate_photo_filename_with_metadata(photo, "original", None) + assert name == "IMG_1234.HEIC" + + def test_no_argument_uses_module_default(self): + photo = _fake_photo("IMG_1234.HEIC", "abc") + set_default_filename_format("simple") + name = generate_photo_filename_with_metadata(photo, "original") + assert name == "IMG_1234.HEIC" + + def test_default_format_remains_metadata_for_backward_compat(self): + photo = _fake_photo("IMG_1234.HEIC", "abc") + # Module-level default not changed → metadata-suffix format + name = generate_photo_filename_with_metadata(photo, "original") + assert "__original__" in name + + +class TestSetDefaultFilenameFormat(unittest.TestCase): + def setUp(self): + set_default_filename_format("metadata") + + def test_set_to_simple(self): + set_default_filename_format("simple") + from src import photo_path_utils + + assert photo_path_utils._DEFAULT_FILENAME_FORMAT == "simple" # noqa: SLF001 + + def test_set_to_metadata(self): + set_default_filename_format("simple") + set_default_filename_format("metadata") + from src import photo_path_utils + + assert photo_path_utils._DEFAULT_FILENAME_FORMAT == "metadata" # noqa: SLF001 + + def test_unknown_value_is_ignored(self): + set_default_filename_format("simple") # baseline + set_default_filename_format("invalid") + from src import photo_path_utils + + # Stays at simple — invalid value silently rejected + assert photo_path_utils._DEFAULT_FILENAME_FORMAT == "simple" # noqa: SLF001 + + + +class TestFilenameFormatEndToEnd(unittest.TestCase): + """Regression test for CRITICAL-2 from the 2026-05-27 pre-submission review. + + The module-level default set by ``set_default_filename_format`` must + propagate all the way through to the path that ``collect_download_task`` + generates. Previously, ``generate_photo_path`` had a ``filename_format`` + parameter defaulting to ``"metadata"`` that silently overrode the module + default — so ``filename_format: simple`` in config had zero effect at the + download path level. This test exercises the full path. + """ + + def setUp(self): + set_default_filename_format("metadata") + + def tearDown(self): + set_default_filename_format("metadata") + + def test_generate_photo_path_uses_module_default(self): + """When `set_default_filename_format("simple")` has been called, + ``generate_photo_path`` (the public-ish function called by + collect_download_task) must produce a plain filename — not the + metadata-suffix form.""" + from src.photo_download_manager import generate_photo_path + + photo = _fake_photo("IMG_9999.HEIC", "cloudkit-id-xyz") + # icloudpy's PhotoAsset.versions exposes `size` per version — fake one + photo.versions = {"original": {"type": "public.heic", "size": 12345}} + photo.created = MagicMock() + # asset_date attribute not used when folder_format is None + + set_default_filename_format("simple") + path = generate_photo_path( + photo, + file_size="original", + destination_path="/tmp/dest", + folder_format=None, + ) + # With simple format, path basename is plain `IMG_9999.HEIC` + # (NOT IMG_9999__original__.HEIC) + import os + basename = os.path.basename(path) + assert basename == "IMG_9999.HEIC", ( + f"Expected plain `IMG_9999.HEIC` (simple format), got `{basename}`. " + "This means filename_format: simple is not threading through to " + "the download path — the boredazfcuk migration would re-download." + ) + + def test_generate_photo_path_metadata_format_still_works(self): + """Backward compat: when default stays `metadata`, format is unchanged.""" + from src.photo_download_manager import generate_photo_path + + photo = _fake_photo("IMG_8888.HEIC", "id-abc") + photo.versions = {"original": {"type": "public.heic", "size": 12345}} + photo.created = MagicMock() + + # Default is "metadata"; no set call + path = generate_photo_path( + photo, + file_size="original", + destination_path="/tmp/dest", + folder_format=None, + ) + import os + basename = os.path.basename(path) + # metadata format: IMG_8888__original__.HEIC + assert basename.startswith("IMG_8888__original__") + assert basename.endswith(".HEIC")