Time-based snap filtering

pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "snapmap-archiver"
-version = "2.1.1"
+version = "2.1.2"
 description = "Download all Snap Map content from a specific location."
 readme = "README.md"
 authors = ["Miles Greenwark <[email protected]>"]

setup.py

@@ -3,7 +3,7 @@
 
 from setuptools import setup
 
-version = "2.1.1"
+version = "2.1.2"
 with open("README.md", "r") as f:
     long_descr = f.read()
 

snapmap_archiver/SnapmapArchiver.py

@@ -2,14 +2,15 @@
 import os
 import re
 import sys
+import typing as t
 from datetime import datetime
 from time import sleep
-from typing import Any, Iterable
 
 import requests
 
 from snapmap_archiver.coordinates import Coordinates
 from snapmap_archiver.snap import Snap, SnapJSONEncoder
+from snapmap_archiver.time import since_epoch
 
 DEFAULT_RADIUS = 10_000
 MAX_RADIUS = 85_000
@@ -24,7 +25,8 @@ def __init__(
         self,
         *args: str,
         output_dir: str,
-        input_file: str | None = None,
+        input_file: t.Optional[str] = None,
+        since_time: t.Optional[str] = None,
         locations: list[str] = [],
         radius: int = DEFAULT_RADIUS,
         write_json: bool = False,
@@ -35,6 +37,11 @@ def __init__(
                 "Python 3.10 or above is required to use snapmap-archiver!"
             )
 
+        self.since_time = None
+        if since_time:
+            self.since_time = since_epoch(since_time.lower())
+            print(f"Skipping Snaps older than [{self.since_time}].")
+
         self.input_file = input_file
         self.arg_snaps = args
 
@@ -54,7 +61,7 @@ def __init__(
         self.radius = MAX_RADIUS if radius > MAX_RADIUS else radius
         self.coords_list = [Coordinates(latlon) for latlon in locations]
 
-    def download_snaps(self, group: Iterable[Snap]):
+    def download_snaps(self, group: t.Iterable[Snap]):
         for snap in group:
             fpath = os.path.join(self.output_dir, f"{snap.snap_id}.{snap.file_type}")
 
@@ -67,7 +74,7 @@ def download_snaps(self, group: Iterable[Snap]):
 
             print(f" - Downloaded [{fpath}].")
 
-    def query_snaps(self, snaps: Iterable[str]) -> list[Snap]:
+    def query_snaps(self, snaps: t.Iterable[str]) -> list[Snap]:
         to_query: list[str] = []
         for snap_id in snaps:
             rgx_match = re.search(
@@ -196,7 +203,7 @@ def main(self):
     def _parse_snap(
         self,
         snap: dict[
-            str, Any
+            str, t.Any
         ],  # I don't like the Any type but this dict is so dynamic there isn't much point hinting it accurately.
     ) -> Snap | None:
         if self.all_snaps.get(snap["id"]):
@@ -215,8 +222,16 @@ def _parse_snap(
             print(f'Media URL for snap [{snap["id"]}] could not be determined.')
             return None
 
+        create_time = round(int(snap["timestamp"]) * 10**-3, 3)
+
+        if (self.since_time) and (create_time < self.since_time):
+            print(
+                f" - [{snap['id']}] is older than the specified time of [{self.since_time}]. Snap timestamp: [{int(create_time)}]. Skipping."
+            )
+            return None
+
         s = Snap(
-            create_time=round(int(snap["timestamp"]) * 10**-3, 3),  # type: ignore
+            create_time=create_time,  # type: ignore
             snap_id=snap["id"],
             url=url,
             file_type=file_type,

snapmap_archiver/__init__.py

@@ -51,6 +51,14 @@ def main():
         type=str,
         help="File containing line-separated Snap URLs or IDs",
     )
+    parser.add_argument(
+        "-t",
+        "--since-time",
+        dest="since_time",
+        type=str,
+        help="Remove any Snaps older than the passed time. Either a 10 digit UTC Unix timestamp or [n = number of][m = minutes | h = hours | d = days] (e.g., 1d, 15h, 30m).",
+        default=None,
+    )
     args, unknown = parser.parse_known_args()
 
     sm_archiver = SnapmapArchiver(
@@ -61,5 +69,6 @@ def main():
         zoom_depth=args.zoom_depth,
         write_json=args.write_json,
         input_file=args.input_file,
+        since_time=args.since_time,
     )
     sm_archiver.main()

snapmap_archiver/time.py

@@ -0,0 +1,15 @@
+from datetime import datetime
+
+
+def since_epoch(since_time: str) -> int:
+    if since_time.isdigit():
+        return int(since_time)
+
+    num = int(since_time[:-1])
+    if since_time[-1] == "m":
+        return int(datetime.now().timestamp()) - num * 60
+    if since_time[-1] == "h":
+        return int(datetime.now().timestamp()) - num * 60 * 60
+    if since_time[-1] == "d":
+        return int(datetime.now().timestamp()) - num * 60 * 60 * 24
+    raise ValueError(f"Invalid time filter: [{since_time}]")