diff --git a/agenda/geomob.py b/agenda/geomob.py index 20dc12e..0a6da7f 100644 --- a/agenda/geomob.py +++ b/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 diff --git a/agenda/thespacedevs.py b/agenda/thespacedevs.py index c609e4e..36993e9 100644 --- a/agenda/thespacedevs.py +++ b/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( diff --git a/agenda/utils.py b/agenda/utils.py index 2c4acfd..27e3ad0 100644 --- a/agenda/utils.py +++ b/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])