feat: Add flag --keep-icloud-recent-days #1046 (#1040)

CHANGELOG.md

@@ -2,6 +2,8 @@
 
 ## Unreleased
 
+- feature: add `--keep-icloud-recent-days` parameter to keep photos newer than this many days in iCloud. Deletes the rest.
+
 ## 1.25.1 (2024-12-28)
 
 - chore: bump max/default python version 3.12->3.13

docs/reference.md

@@ -187,6 +187,12 @@ This is a list of all options available for command line interface (CLI) of the
     ```{seealso}
     [Modes of operation](mode)
     ```
+
+(keep-icloud-recent-days-parameter)=
+`--keep-icloud-recent-days X`
+
+:   If specified along with `--delete-after-download`, assets downloaded locally will not be deleted in iCloud if they were created within the specified number of days.
+
 (only-print-filenames-parameter)=
 `--only-print-filenames`
 

src/icloudpd/base.py

@@ -487,6 +487,13 @@ def report_version(ctx: click.Context, _param: click.Parameter, value: bool) ->
     + " Therefore, should not combine with --auto-delete option.",
     is_flag=True,
 )
+@click.option(
+    "--keep-icloud-recent-days",
+    help="Keep photos newer than this many days in iCloud. Deletes the rest. "
+    + "If set to 0, all photos will be deleted from iCloud.",
+    type=click.IntRange(0),
+    default=None,
+)
 @click.option(
     "--domain",
     help="What iCloud root domain to use. Use 'cn' for mainland China (default: 'com')",
@@ -606,6 +613,7 @@ def main(
     notification_script: Optional[str],
     threads_num: int,
     delete_after_download: bool,
+    keep_icloud_recent_days: Optional[int],
     domain: str,
     watch_with_interval: Optional[int],
     dry_run: bool,
@@ -650,6 +658,12 @@ def main(
             print("--auto-delete and --delete-after-download are mutually exclusive")
             sys.exit(2)
 
+        if keep_icloud_recent_days and delete_after_download:
+            print(
+                "--keep-icloud-recent-days and --delete-after-download should not be used together."
+            )
+            sys.exit(2)
+
         if watch_with_interval and (list_albums or only_print_filenames):  # pragma: no cover
             print(
                 "--watch_with_interval is not compatible with --list_albums, --only_print_filenames"
@@ -717,6 +731,7 @@ def main(
             notification_script=notification_script,
             threads_num=threads_num,
             delete_after_download=delete_after_download,
+            keep_icloud_recent_days=keep_icloud_recent_days,
             domain=domain,
             watch_with_interval=watch_with_interval,
             dry_run=dry_run,
@@ -792,6 +807,7 @@ def main(
             no_progress_bar,
             notification_script,
             delete_after_download,
+            keep_icloud_recent_days,
             domain,
             logger,
             watch_with_interval,
@@ -1170,6 +1186,7 @@ def core(
     no_progress_bar: bool,
     notification_script: Optional[str],
     delete_after_download: bool,
+    keep_icloud_recent_days: Optional[int],
     domain: str,
     logger: logging.Logger,
     watch_interval: Optional[int],
@@ -1391,6 +1408,40 @@ def should_break(counter: Counter) -> bool:
                     logger, dry_run, library_object, folder_structure, directory, primary_sizes
                 )
 
+            if keep_icloud_recent_days is not None:
+                try:
+                    now = datetime.datetime.now(get_localzone())
+                    created_date = item.created.astimezone(get_localzone())
+                    age_days = (now - created_date).days
+                    if age_days < keep_icloud_recent_days:
+                        logger.debug(
+                            "Skipping deletion of %s as it is within the keep_icloud_recent_days period (%d days old)",
+                            item.filename,
+                            age_days,
+                        )
+                    else:
+                        delete_local = partial(
+                            delete_photo_dry_run if dry_run else delete_photo,
+                            logger,
+                            icloud.photos,
+                            library_object,
+                            item,
+                        )
+
+                        retrier(delete_local, error_handler)
+                        logger.debug(
+                            "Deleted %s as it is older than the keep_icloud_recent_days period (%d days old)",
+                            item.filename,
+                            age_days,
+                        )
+                except (ValueError, OSError):
+                    logger.error(
+                        "Could not convert photo created date to local timezone (%s)",
+                        item.created,
+                    )
+                except Exception as e:
+                    logger.error(f"Error deleting photo: {e}")
+
             if watch_interval:  # pragma: no cover
                 logger.info(f"Waiting for {watch_interval} sec...")
                 interval: Sequence[int] = range(1, watch_interval)

src/icloudpd/config.py

@@ -40,6 +40,7 @@ def __init__(
         notification_script: Optional[str],
         threads_num: int,
         delete_after_download: bool,
+        keep_icloud_recent_days: Optional[int],
         domain: str,
         watch_with_interval: Optional[int],
         dry_run: bool,
@@ -81,6 +82,7 @@ def __init__(
         self.notification_script = notification_script
         self.threads_num = threads_num
         self.delete_after_download = delete_after_download
+        self.keep_icloud_recent_days = keep_icloud_recent_days
         self.domain = domain
         self.watch_with_interval = watch_with_interval
         self.dry_run = dry_run

tests/test_cli.py

@@ -190,3 +190,21 @@ def test_conflict_options_delete_after_download_and_auto_delete(self) -> None:
             ],
         )
         assert result.exit_code == 2
+
+    def test_conflict_options_delete_after_download_and_keep_icloud_recent_days(self) -> None:
+        runner = CliRunner()
+        result = runner.invoke(
+            main,
+            [
+                "--username",
+                "[email protected]",
+                "--password",
+                "password1",
+                "-d",
+                "/tmp",
+                "--delete-after-download",
+                "--keep-icloud-recent-days",
+                "1",
+            ],
+        )
+        assert result.exit_code == 2

tests/test_keep_icloud_mode.py

@@ -0,0 +1,206 @@
+import datetime
+import inspect
+import os
+from unittest import TestCase, mock
+
+import pytest
+from vcr import VCR
+
+from tests.helpers import path_from_project_root, run_icloudpd_test
+
+vcr = VCR(decode_compressed_response=True, record_mode="none")
+
+
+class KeepICloudModeTestCases(TestCase):
+    @pytest.fixture(autouse=True)
+    def inject_fixtures(self, caplog: pytest.LogCaptureFixture) -> None:
+        self._caplog = caplog
+        self.root_path = path_from_project_root(__file__)
+        self.fixtures_path = os.path.join(self.root_path, "fixtures")
+        self.vcr_path = os.path.join(self.root_path, "vcr_cassettes")
+
+    def test_wide_range_keep_icloud_recent_days(self) -> None:
+        base_dir = os.path.join(self.fixtures_path, inspect.stack()[0][3])
+
+        files_to_download = [("2018/07/31", "IMG_7409.JPG")]
+
+        with mock.patch("datetime.datetime", wraps=datetime.datetime) as dt_mock:
+            # 90 days in the future
+            mock_now = datetime.datetime(2018, 7, 31, tzinfo=datetime.timezone.utc)
+            dt_mock.now.return_value = mock_now + datetime.timedelta(days=90)
+            data_dir, result = run_icloudpd_test(
+                self.assertEqual,
+                self.root_path,
+                base_dir,
+                "listing_photos_keep_icloud_recent_days.yml",
+                [],
+                files_to_download,
+                [
+                    "--username",
+                    "[email protected]",
+                    "--password",
+                    "password1",
+                    "--recent",
+                    "1",
+                    "--skip-videos",
+                    "--skip-live-photos",
+                    "--no-progress-bar",
+                    "--threads-num",
+                    "1",
+                    "--keep-icloud-recent-days",
+                    "100",
+                ],
+            )
+
+            self.assertIn(
+                "DEBUG    Looking up all photos from album All Photos...", self._caplog.text
+            )
+            self.assertIn(
+                f"INFO     Downloading the first original photo to {data_dir} ...",
+                self._caplog.text,
+            )
+            self.assertIn(
+                "DEBUG    Skipping deletion of IMG_7409.JPG as it is within the keep_icloud_recent_days period (89 days old)",
+                self._caplog.text,
+            )
+            self.assertIn("INFO     All photos have been downloaded", self._caplog.text)
+            assert result.exit_code == 0
+
+    def test_narrow_range_keep_icloud_recent_days(self) -> None:
+        base_dir = os.path.join(self.fixtures_path, inspect.stack()[0][3])
+
+        files_to_download = [("2018/07/31", "IMG_7409.JPG")]
+
+        with mock.patch("datetime.datetime", wraps=datetime.datetime) as dt_mock:
+            mock_now = datetime.datetime(2018, 8, 1, tzinfo=datetime.timezone.utc)
+            dt_mock.now.return_value = mock_now
+            data_dir, result = run_icloudpd_test(
+                self.assertEqual,
+                self.root_path,
+                base_dir,
+                "listing_photos_keep_icloud_recent_days.yml",
+                [],
+                files_to_download,
+                [
+                    "--username",
+                    "[email protected]",
+                    "--password",
+                    "password1",
+                    "--recent",
+                    "1",
+                    "--skip-videos",
+                    "--skip-live-photos",
+                    "--no-progress-bar",
+                    "--threads-num",
+                    "1",
+                    "--keep-icloud-recent-days",
+                    "1",
+                ],
+            )
+
+            self.assertIn(
+                "DEBUG    Looking up all photos from album All Photos...", self._caplog.text
+            )
+            self.assertIn(
+                f"INFO     Downloading the first original photo to {data_dir} ...",
+                self._caplog.text,
+            )
+            self.assertIn(
+                "DEBUG    Skipping deletion of IMG_7409.JPG as it is within the keep_icloud_recent_days period (0 days old)",
+                self._caplog.text,
+            )
+            self.assertIn("INFO     All photos have been downloaded", self._caplog.text)
+            assert result.exit_code == 0
+
+    def test_keep_icloud_recent_days_delete_all(self) -> None:
+        base_dir = os.path.join(self.fixtures_path, inspect.stack()[0][3])
+
+        files_to_download = [("2018/07/31", "IMG_7409.JPG")]
+
+        with mock.patch("datetime.datetime", wraps=datetime.datetime) as dt_mock:
+            days_old = 10
+            mock_now = datetime.datetime(2018, 7, 31, tzinfo=datetime.timezone.utc)
+            dt_mock.now.return_value = mock_now + datetime.timedelta(days=days_old)
+            data_dir, result = run_icloudpd_test(
+                self.assertEqual,
+                self.root_path,
+                base_dir,
+                "listing_photos_keep_icloud_recent_days.yml",
+                [],
+                files_to_download,
+                [
+                    "--username",
+                    "[email protected]",
+                    "--password",
+                    "password1",
+                    "--recent",
+                    "1",
+                    "--skip-videos",
+                    "--skip-live-photos",
+                    "--no-progress-bar",
+                    "--threads-num",
+                    "1",
+                    "--keep-icloud-recent-days",
+                    "0",
+                ],
+            )
+
+            self.assertIn(
+                "DEBUG    Looking up all photos from album All Photos...", self._caplog.text
+            )
+            self.assertIn(
+                f"INFO     Downloading the first original photo to {data_dir} ...",
+                self._caplog.text,
+            )
+            self.assertIn(
+                f"DEBUG    Deleted IMG_7409.JPG as it is older than the keep_icloud_recent_days period ({days_old - 1} days old)",
+                self._caplog.text,
+            )
+            self.assertIn("INFO     All photos have been downloaded", self._caplog.text)
+            assert result.exit_code == 0
+
+    def test_keep_icloud_recent_days_1_keeps_today(self) -> None:
+        base_dir = os.path.join(self.fixtures_path, inspect.stack()[0][3])
+
+        files_to_download = [("2018/07/31", "IMG_7409.JPG")]
+
+        with mock.patch("datetime.datetime", wraps=datetime.datetime) as dt_mock:
+            mock_now = datetime.datetime(2018, 7, 31, 23, 59, 59, tzinfo=datetime.timezone.utc)
+            dt_mock.now.return_value = mock_now
+            data_dir, result = run_icloudpd_test(
+                self.assertEqual,
+                self.root_path,
+                base_dir,
+                "listing_photos_keep_icloud_recent_days.yml",
+                [],
+                files_to_download,
+                [
+                    "--username",
+                    "[email protected]",
+                    "--password",
+                    "password1",
+                    "--recent",
+                    "1",
+                    "--skip-videos",
+                    "--skip-live-photos",
+                    "--no-progress-bar",
+                    "--threads-num",
+                    "1",
+                    "--keep-icloud-recent-days",
+                    "1",
+                ],
+            )
+
+            self.assertIn(
+                "DEBUG    Looking up all photos from album All Photos...", self._caplog.text
+            )
+            self.assertIn(
+                f"INFO     Downloading the first original photo to {data_dir} ...",
+                self._caplog.text,
+            )
+            self.assertIn(
+                "DEBUG    Skipping deletion of IMG_7409.JPG as it is within the keep_icloud_recent_days period (0 days old)",
+                self._caplog.text,
+            )
+            self.assertIn("INFO     All photos have been downloaded", self._caplog.text)
+            assert result.exit_code == 0