add skip-created-before param #1111 #466 (#1145)

Author: AndreyNikiforov

CHANGELOG.md

@@ -2,6 +2,8 @@
 
 ## Unreleased
 
+- feat: `--skip-created-before` to limit assets by creation date [#466](https://github.com/icloud-photos-downloader/icloud_photos_downloader/issues/466) [#1111](https://github.com/icloud-photos-downloader/icloud_photos_downloader/issues/1111)
+
 ## 1.27.5 (2025-05-08)
 
 - fix: HEIF file extension [#1133](https://github.com/icloud-photos-downloader/icloud_photos_downloader/issues/1133)

docs/reference.md

@@ -301,3 +301,14 @@ This is a list of all options available for command line interface (CLI) of the
 
 :   Script to be executed for notification on expired MFA
 
+(skip-created-before-parameter)=
+`--skip-created-before`
+
+:   Does not process assets created before specified timestamp. Timestamp is in ISO format, e.g 2025-06-01, or as interval from now, e.g. 5d. If timezone is not specified for ISO format, then local timezone is used.
+
+    ```{versionadded} 1.28.0
+    ```
+
+    ```{note}
+    The date is when asset was created, not added to the iCloud.
+    ```

src/icloudpd/base.py

@@ -1,6 +1,7 @@
 #!/usr/bin/env python
 """Main script that uses Click to parse command-line arguments"""
 
+import re
 from multiprocessing import freeze_support
 
 import foundation
@@ -32,6 +33,7 @@
     Sequence,
     Tuple,
     TypeVar,
+    Union,
     cast,
 )
 
@@ -260,6 +262,28 @@ def file_match_policy_generator(
         raise ValueError(f"policy was provided with unsupported value of '{policy}'")
 
 
+def skip_created_before_generator(
+    _ctx: click.Context, _param: click.Parameter, formatted: str
+) -> Optional[Union[datetime.datetime, datetime.timedelta]]:
+    if formatted is None:
+        return None
+    # can be timestamp or timedelta
+    m = re.match(r"(\d+)([dD]{1})", formatted)
+    if m is not None and m.lastindex is not None and m.lastindex == 2:
+        return datetime.timedelta(days=float(m.group(1)))
+
+    # try timestamp
+    try:
+        dt = datetime.datetime.fromisoformat(formatted)
+        if dt.tzinfo is None:
+            dt = dt.astimezone(get_localzone())
+        return dt
+    except Exception as e:
+        raise ValueError(
+            f"Timestamp {formatted} for --skip-created-before parameter did not parse from ISO format successfully: {e}"
+        ) from e
+
+
 def locale_setter(_ctx: click.Context, _param: click.Parameter, use_os_locale: bool) -> bool:
     # set locale
     if use_os_locale:
@@ -570,6 +594,11 @@ def report_version(ctx: click.Context, _param: click.Parameter, value: bool) ->
     is_eager=True,
     callback=locale_setter,
 )
+@click.option(
+    "--skip-created-before",
+    help="Do not process assets created before specified timestamp in ISO format (2025-01-02) or interval from now (20d)",
+    callback=skip_created_before_generator,
+)
 @click.option(
     "--version",
     help="Show the version, commit hash and timestamp",
@@ -625,6 +654,7 @@ def main(
     file_match_policy: FileMatchPolicy,
     mfa_provider: MFAProvider,
     use_os_locale: bool,
+    skip_created_before: Optional[Union[datetime.datetime, datetime.timedelta]],
 ) -> NoReturn:
     """Download all iCloud photos to a local directory"""
 
@@ -778,6 +808,7 @@ def main(
                 dry_run,
                 file_match_policy,
                 xmp_sidecar,
+                skip_created_before,
             )
             if directory is not None
             else (lambda _s: lambda _c, _p: False),
@@ -836,6 +867,7 @@ def download_builder(
     dry_run: bool,
     file_match_policy: FileMatchPolicy,
     xmp_sidecar: bool,
+    skip_created_before: Optional[Union[datetime.datetime, datetime.timedelta]],
 ) -> Callable[[PyiCloudService], Callable[[Counter, PhotoAsset], bool]]:
     """factory for downloader"""
 
@@ -866,6 +898,23 @@ def download_photo_(counter: Counter, photo: PhotoAsset) -> bool:
                 )
                 created_date = photo.created
 
+            if skip_created_before is not None:
+                if isinstance(skip_created_before, datetime.timedelta):
+                    temp_created_before = (
+                        datetime.datetime.now(get_localzone()) - skip_created_before
+                    )
+                elif isinstance(skip_created_before, datetime.datetime):
+                    temp_created_before = skip_created_before
+                else:
+                    raise ValueError(
+                        f"skip-created-before is of unsupported type {type(skip_created_before)}"
+                    )
+                if created_date < temp_created_before:
+                    logger.debug(
+                        f"Skipping {photo.filename}, as it was created {created_date}, before {temp_created_before}."
+                    )
+                    return False
+
             if folder_structure.lower() == "none":
                 date_path = ""
             else:

src/icloudpd/email_notifications.py

@@ -44,10 +44,7 @@ def send_2sa_notification(
 {username}'s two-step authentication has expired for the icloud_photos_downloader script.
 Please log in to your server and run the script manually to update two-step authentication."""
 
-    msg = (
-        f"From: {from_addr}\n"
-        + f"To: {to_addr}\nSubject: {subj}\nDate: {date}\n\n{message_text}"
-    )
+    msg = f"From: {from_addr}\n" + f"To: {to_addr}\nSubject: {subj}\nDate: {date}\n\n{message_text}"
 
     smtp.sendmail(from_addr, to_addr, msg)
     smtp.quit()

tests/test_download_photos.py

@@ -2389,3 +2389,88 @@ def test_download_from_shared_library(self) -> None:
             self._caplog.text,
         )
         self.assertIn("INFO     All photos have been downloaded", self._caplog.text)
+
+    def test_download_and_skip_old(self) -> None:
+        base_dir = os.path.join(self.fixtures_path, inspect.stack()[0][3])
+
+        files_to_create: List[Tuple[str, str, int]] = [
+            # ("2018/07/30", "IMG_7408.JPG", 1151066),
+            # ("2018/07/30", "IMG_7407.JPG", 656257),
+        ]
+
+        files_to_download = [("2018/07/31", "IMG_7409.JPG")]
+
+        data_dir, result = run_icloudpd_test(
+            self.assertEqual,
+            self.root_path,
+            base_dir,
+            "listing_photos.yml",
+            files_to_create,
+            files_to_download,
+            [
+                "--username",
+                "jdoe@gmail.com",
+                "--password",
+                "password1",
+                "--recent",
+                "5",
+                "--skip-videos",
+                "--skip-live-photos",
+                "--set-exif-datetime",
+                "--no-progress-bar",
+                "--skip-created-before",
+                "2018-07-31",
+            ],
+        )
+
+        assert result.exit_code == 0
+
+        self.assertIn("DEBUG    Looking up all photos...", self._caplog.text)
+        self.assertIn(
+            f"INFO     Downloading 5 original photos to {data_dir} ...",
+            self._caplog.text,
+        )
+        for dir_name, file_name in files_to_download:
+            file_path = os.path.normpath(os.path.join(dir_name, file_name))
+            self.assertIn(
+                f"DEBUG    Downloading {os.path.join(data_dir, file_path)}",
+                self._caplog.text,
+            )
+        self.assertNotIn(
+            "IMG_7409.MOV",
+            self._caplog.text,
+        )
+        for dir_name, file_name in [
+            (dir_name, file_name) for (dir_name, file_name, _) in files_to_create
+        ]:
+            file_path = os.path.normpath(os.path.join(dir_name, file_name))
+            self.assertIn(
+                f"DEBUG    {os.path.join(data_dir, file_path)} already exists",
+                self._caplog.text,
+            )
+
+        self.assertIn(
+            "DEBUG    Skipping IMG_7405.MOV, only downloading photos.",
+            self._caplog.text,
+        )
+        self.assertIn(
+            "DEBUG    Skipping IMG_7404.MOV, only downloading photos.",
+            self._caplog.text,
+        )
+        self.assertIn(
+            "DEBUG    Skipping IMG_7407.JPG, as it was created 2018-07-30 11:44:05.108000+00:00, before 2018-07-31 00:00:00+00:00.",
+            self._caplog.text,
+        )
+        self.assertIn(
+            "DEBUG    Skipping IMG_7408.JPG, as it was created 2018-07-30 11:44:10.176000+00:00, before 2018-07-31 00:00:00+00:00.",
+            self._caplog.text,
+        )
+        self.assertIn("INFO     All photos have been downloaded", self._caplog.text)
+
+        # Check that file was downloaded
+        # Check that mtime was updated to the photo creation date
+        photo_mtime = os.path.getmtime(
+            os.path.join(data_dir, os.path.normpath("2018/07/31/IMG_7409.JPG"))
+        )
+        photo_modified_time = datetime.datetime.fromtimestamp(photo_mtime, datetime.timezone.utc)
+        self.assertEqual("2018-07-31 07:22:24", photo_modified_time.strftime("%Y-%m-%d %H:%M:%S"))