Refactor

agenda/geomob.py

@@ -23,7 +23,7 @@ class GeomobEvent:
     hashtag: str
 
 
-def extract_geomob_events(
+def extract_events(
     tree: lxml.html.HtmlElement,
 ) -> List[GeomobEvent]:
     """Extract upcoming events from the HTML content."""
@@ -79,21 +79,8 @@ def geomob_email(new_events: list[GeomobEvent], base_url: str) -> tuple[str, str
 
 def get_cached_upcoming_events_list(geomob_dir: str) -> list[GeomobEvent]:
     """Get known geomob events."""
-    existing = [
-        x
-        for x in (
-            agenda.utils.filename_timestamp(f, "html") for f in os.listdir(geomob_dir)
-        )
-        if x
-    ]
-
-    if not existing:
-        return []
-    existing.sort(reverse=True)
-    f = existing[0][1]
-
-    filename = os.path.join(geomob_dir, f)
-    return extract_geomob_events(lxml.html.parse(filename).getroot())
+    filename = agenda.utils.get_most_recent_file(geomob_dir, "html")
+    return extract_events(lxml.html.parse(filename).getroot()) if filename else []
 
 
 def update(config: flask.config.Config) -> None:
@@ -102,7 +89,7 @@ def update(config: flask.config.Config) -> None:
 
     prev_events = get_cached_upcoming_events_list(geomob_dir)
     r = requests.get("https://thegeomob.com/")
-    cur_events = extract_geomob_events(lxml.html.fromstring(r.content))
+    cur_events = extract_events(lxml.html.fromstring(r.content))
 
     if cur_events == prev_events:
         return  # no change

agenda/thespacedevs.py

@@ -8,7 +8,7 @@ from datetime import datetime
 import requests
 
 from .types import StrDict
-from .utils import filename_timestamp
+from .utils import filename_timestamp, get_most_recent_file
 
 Launch = dict[str, typing.Any]
 Summary = dict[str, typing.Any]
@@ -139,23 +139,16 @@ def summarize_launch(launch: Launch) -> Summary:
     }
 
 
-def load_cached_launches(rocket_dir: str) -> StrDict:
+def load_cached_launches(rocket_dir: str) -> StrDict | None:
     """Read the most recent cache of launches."""
-    existing = [
-        x for x in (filename_timestamp(f, "json") for f in os.listdir(rocket_dir)) if x
-    ]
-
-    existing.sort(reverse=True)
-    f = existing[0][1]
-
-    filename = os.path.join(rocket_dir, f)
-    return typing.cast(StrDict, json.load(open(filename)))
+    filename = get_most_recent_file(rocket_dir, "json")
+    return typing.cast(StrDict, json.load(open(filename))) if filename else None
 
 
 def read_cached_launches(rocket_dir: str) -> list[Summary]:
     """Read cached launches."""
     data = load_cached_launches(rocket_dir)
-    return [summarize_launch(launch) for launch in data["results"]]
+    return [summarize_launch(launch) for launch in data["results"]] if data else []
 
 
 def get_launches(

agenda/utils.py

@@ -1,5 +1,6 @@
 """Utility functions."""
 
+import os
 from datetime import date, datetime, timezone
 
 
@@ -65,3 +66,14 @@ def filename_timestamp(filename: str, ext: str) -> tuple[datetime, str] | None:
     except ValueError:
         return None
     return (ts, filename)
+
+
+def get_most_recent_file(directory: str, ext: str) -> str | None:
+    """Get most recent file from directory."""
+    existing = [
+        x for x in (filename_timestamp(f, ext) for f in os.listdir(directory)) if x
+    ]
+    if not existing:
+        return None
+    existing.sort(reverse=True)
+    return os.path.join(directory, existing[0][1])