Refactoring to consolidate shared code

lektor/lektordata/scripts/calendar-fetcher-main.py

@@ -2,65 +2,12 @@
 
 import sys
 from pathlib import Path
-from datetime import datetime, date, timedelta, timezone
+from datetime import datetime, date, timezone
 from typing import Optional, NamedTuple
 from dataclasses import dataclass
-from copy import deepcopy
-import requests
 from bs4 import BeautifulSoup
-from icalendar import Calendar, Event
-import recurring_ical_events
 import locale
-
+from calendarstuff import get_events
-
-#This is duplicated code from calendar-fetcher.py
-def split_multiday_events(events):
-    """
-    Split multi-day events into separate daily events.
-    
-    Args:
-        events: Iterator of iCal events
-        
-    Returns:
-        List of events where multi-day events are split into daily events
-    """
-    split_events = []
-    
-    for event in events:
-        # Get start and end dates
-        start = event.get('dtstart').dt
-        end = event.get('dtend').dt if event.get('dtend') else start
-        
-        # Convert datetime to date if needed
-        if isinstance(start, datetime):
-            start = start.date()
-        if isinstance(end, datetime):
-            end = end.date()
-            
-        # If it's a single day event or not a date/datetime, add as is
-        if not isinstance(start, date) or not isinstance(end, date) or start == end:
-            split_events.append(event)
-            continue
-            
-        # For multi-day events, create separate events for each day
-        current_date = start
-        while current_date < end:
-            # Create a copy of the original event
-            daily_event = Event()
-            for key in event:
-                daily_event[key] = deepcopy(event[key])
-
-            
-            # Update the date for this instance
-            daily_event['dtstart'].dt = current_date
-
-            if 'dtend' in daily_event:
-                daily_event['dtend'].dt = current_date + timedelta(days=1)
-                
-            split_events.append(daily_event)
-            current_date += timedelta(days=1)
-    return split_events
-
 
 
 @dataclass
@@ -120,36 +67,13 @@ page.html
         except locale.Error as e:
             print(f"Warning: Failed to set locale: {e}", file=sys.stderr)
 
-    def fetch_calendar(self) -> Optional[Calendar]:
-        try:
-            response = requests.get(self.ics_url, timeout=10)
-            response.raise_for_status()
-            return Calendar.from_ical(response.content)
-        except (requests.RequestException, ValueError) as e:
-            print(f"Error fetching calendar: {e}", file=sys.stderr)
-            return None
-
     def format_event_time(self, start: datetime | date) -> str:
         return "" if isinstance(start, date) and not isinstance(start, datetime) else start.strftime(" %-H:%M")
 
-    def get_next_event(self, calendar: Calendar) -> Optional[EventDetails]:
+    def get_next_event(self) -> Optional[EventDetails]:
         try:
 
-            now=datetime.now(timezone.utc) # + timedelta(days=1)
+            events = get_events(self.ics_url)
-
-            raw_events = split_multiday_events(recurring_ical_events.of(calendar).after(now))
-
-            now_date = now.date()
-            now_datetime = now
-            events = []
-            for raw_event in raw_events:
-                start = raw_event.get('dtstart').dt
-                if isinstance(start, datetime):
-                    my_now = now_datetime
-                else:
-                     my_now = now_date
-                if start >= my_now:
-                    events.append(raw_event)
             
             event = events[0]
             start = event.get('dtstart').dt
@@ -185,11 +109,7 @@ page.html
             return self.fallback_content
             
         if len(source_str) > 0 and source_str in content:
-            calendar = self.fetch_calendar()
+            event = self.get_next_event()
-            if not calendar:
-                return content.replace(source_str, self.fallback_html).rstrip()
-                
-            event = self.get_next_event(calendar)
             if not event:
                 return content.replace(source_str, self.fallback_html).rstrip()
                 

lektor/lektordata/scripts/calendar-fetcher.py

@@ -1,87 +1,15 @@
-from datetime import datetime, date, timedelta, timezone
+from datetime import datetime, date, timezone
-import recurring_ical_events
-import requests
-from copy import deepcopy
-from icalendar import Calendar, Event
 import locale
 import sys
-
+from calendarstuff import get_events
-#This is duplicated code from calendar-fetcher-main.py
-def split_multiday_events(events):
-    """
-    Split multi-day events into separate daily events.
-    
-    Args:
-        events: Iterator of iCal events
-        
-    Returns:
-        List of events where multi-day events are split into daily events
-    """
-    split_events = []
-    
-    for event in events:
-        # Get start and end dates
-        start = event.get('dtstart').dt
-        end = event.get('dtend').dt if event.get('dtend') else start
-        
-        # Convert datetime to date if needed
-        if isinstance(start, datetime):
-            start = start.date()
-        if isinstance(end, datetime):
-            end = end.date()
-            
-        # If it's a single day event or not a date/datetime, add as is
-        if not isinstance(start, date) or not isinstance(end, date) or start == end:
-            split_events.append(event)
-            continue
-            
-        # For multi-day events, create separate events for each day
-        current_date = start
-        while current_date < end:
-            # Create a copy of the original event
-            daily_event = Event()
-            for key in event:
-                daily_event[key] = deepcopy(event[key])
-
-            
-            # Update the date for this instance
-            daily_event['dtstart'].dt = current_date
-
-            if 'dtend' in daily_event:
-                daily_event['dtend'].dt = current_date + timedelta(days=1)
-                
-            split_events.append(daily_event)
-            current_date += timedelta(days=1)
-    return split_events
-
 
 
 def fetch_upcoming_events(ics_url):
-    # Set German locale
-    locale.setlocale(locale.LC_TIME, 'de_DE.UTF-8')
 
-    response = requests.get(ics_url)
+    events = get_events(ics_url)
-    calendar = Calendar.from_ical(response.content)
-
-    now=datetime.now(timezone.utc) # + timedelta(days=1)
-
-    raw_events = split_multiday_events(recurring_ical_events.of(calendar).after(now))
-
-    now_date = now.date()
-    now_datetime = now
-    events = []
-    for raw_event in raw_events:
-        start = raw_event.get('dtstart').dt
-        if isinstance(start, datetime):
-            my_now = now_datetime
-        else:
-             my_now = now_date
-        if start >= my_now:
-            events.append(raw_event)
 
     for event in events:
 
-
         start = event.get('dtstart').dt
         out_summary = event.get('summary')
         location = event.get('location', 'No location specified')

lektor/lektordata/scripts/calendarstuff.py

@@ -0,0 +1,81 @@
+from datetime import datetime, date, timedelta, timezone
+import recurring_ical_events
+import requests
+from copy import deepcopy
+from icalendar import Calendar, Event
+import locale
+import sys
+
+
+#This is duplicated code from calendar-fetcher-main.py
+def split_multiday_events(events):
+    """
+    Split multi-day events into separate daily events.
+    
+    Args:
+        events: Iterator of iCal events
+        
+    Returns:
+        List of events where multi-day events are split into daily events
+    """
+    split_events = []
+    
+    for event in events:
+        # Get start and end dates
+        start = event.get('dtstart').dt
+        end = event.get('dtend').dt if event.get('dtend') else start
+        
+        # Convert datetime to date if needed
+        if isinstance(start, datetime):
+            start = start.date()
+        if isinstance(end, datetime):
+            end = end.date()
+            
+        # If it's a single day event or not a date/datetime, add as is
+        if not isinstance(start, date) or not isinstance(end, date) or start == end:
+            split_events.append(event)
+            continue
+            
+        # For multi-day events, create separate events for each day
+        current_date = start
+        while current_date < end:
+            # Create a copy of the original event
+            daily_event = Event()
+            for key in event:
+                daily_event[key] = deepcopy(event[key])
+
+            
+            # Update the date for this instance
+            daily_event['dtstart'].dt = current_date
+
+            if 'dtend' in daily_event:
+                daily_event['dtend'].dt = current_date + timedelta(days=1)
+                
+            split_events.append(daily_event)
+            current_date += timedelta(days=1)
+    return split_events
+
+
+def get_events(ics_url):
+    # Set German locale
+    locale.setlocale(locale.LC_TIME, 'de_DE.UTF-8')
+    
+    response = requests.get(ics_url)
+    calendar = Calendar.from_ical(response.content)
+
+    now=datetime.now(timezone.utc) # + timedelta(days=1)
+
+    raw_events = split_multiday_events(recurring_ical_events.of(calendar).after(now))
+
+    now_date = now.date()
+    now_datetime = now
+    events = []
+    for raw_event in raw_events:
+        start = raw_event.get('dtstart').dt
+        if isinstance(start, datetime):
+            my_now = now_datetime
+        else:
+             my_now = now_date
+        if start >= my_now:
+            events.append(raw_event)
+    return events