Rename instantiate_upcoming_schedule

- Add type annotations,
- Cleanup logic,
- utils.parse_time now accepts to time format strings.

program/services.py

@@ -18,6 +18,7 @@
 #
 
 from datetime import datetime, time, timedelta
+from typing import TypedDict
 
 from dateutil.relativedelta import relativedelta
 from dateutil.rrule import rrule
@@ -35,6 +36,19 @@ from steering.settings import (
 )
 
 
+class ScheduleData(TypedDict):
+    add_business_days_only: bool | None
+    add_days_no: int | None
+    by_weekday: int | None
+    default_playlist_id: int | None
+    end_time: str
+    first_date: str
+    is_repetition: bool
+    last_date: str | None
+    rrule: int
+    start_time: str
+
+
 # TODO: add type annotations
 def resolve_conflicts(data, schedule_pk, show_pk):
     """
@@ -49,7 +63,7 @@ def resolve_conflicts(data, schedule_pk, show_pk):
     solutions = data.get("solutions", [])
 
     # Regenerate conflicts
-    schedule = instantiate_upcoming(sdl, show_pk, schedule_pk)
+    schedule = instantiate_upcoming_schedule(sdl, show_pk, schedule_pk)
     show = schedule.show
     conflicts = make_conflicts(sdl, schedule_pk, show_pk)
 
@@ -315,31 +329,31 @@ def resolve_conflicts(data, schedule_pk, show_pk):
     return model_to_dict(schedule)
 
 
-# TODO: add type annotations
-def instantiate_upcoming(sdl, show_pk, pk=None):
+def instantiate_upcoming_schedule(
+    data: ScheduleData, show_pk: int, pk: int | None = None
+) -> Schedule:
     """Returns an upcoming schedule instance for conflict resolution"""
-    pk = int(pk) if pk is not None else None
-    rrule = RRule.objects.get(pk=int(sdl["rrule"]))
-    show = Show.objects.get(pk=int(show_pk))
 
-    is_repetition = True if sdl.get("is_repetition") is True else False
-    default_playlist_id = (
-        int(sdl["default_playlist_id"]) if sdl.get("default_playlist_id") else None
-    )
-    add_days_no = int(sdl["add_days_no"]) if sdl.get("add_days_no") else None
-    add_business_days_only = True if sdl.get("add_business_days_only") is True else False
+    rrule = RRule.objects.get(pk=data["rrule"])
+    show = Show.objects.get(pk=show_pk)
+
+    is_repetition = data["is_repetition"]
 
-    first_date = parse_date(str(sdl["first_date"]))
-    start_time = (
-        sdl["start_time"] + ":00" if len(str(sdl["start_time"])) == 5 else sdl["start_time"]
+    # these may not be present in data
+    add_business_days_only = (
+        data["default_playlist_id"] if "add_business_days_only" in data else False
     )
-    end_time = sdl["end_time"] + ":00" if len(str(sdl["end_time"])) == 5 else sdl["end_time"]
+    add_days_no = data["add_days_no"] if "add_days_no" in data else None
+    by_weekday = data["by_weekday"] if "by_weekday" in data else None
+    default_playlist_id = data["default_playlist_id"] if "default_playlist_id" in data else None
 
-    start_time = parse_time(str(start_time))
-    end_time = parse_time(str(end_time))
+    first_date = parse_date(data["first_date"])
+    start_time = parse_time(data["start_time"])
+    end_time = parse_time(data["end_time"])
 
-    if sdl["last_date"]:
-        last_date = parse_date(str(sdl["last_date"]))
+    # last_date may not be present in data
+    if "last_date" in data:
+        last_date = parse_date(data["last_date"])
     else:
         # If last_date was not set, set it to the end of the year or add x days
         if AUTO_SET_LAST_DATE_TO_END_OF_YEAR:
@@ -348,9 +362,9 @@ def instantiate_upcoming(sdl, show_pk, pk=None):
         else:
             last_date = first_date + timedelta(days=+AUTO_SET_LAST_DATE_TO_DAYS_IN_FUTURE)
 
-    schedule = Schedule(
+    return Schedule(
         pk=pk,
-        by_weekday=sdl["by_weekday"],
+        by_weekday=by_weekday,
         rrule=rrule,
         first_date=first_date,
         start_time=start_time,
@@ -363,8 +377,6 @@ def instantiate_upcoming(sdl, show_pk, pk=None):
         add_business_days_only=add_business_days_only,
     )
 
-    return schedule
-
 
 # TODO: add type annotations
 def make_conflicts(sdl, schedule_pk, show_pk):
@@ -376,7 +388,7 @@ def make_conflicts(sdl, schedule_pk, show_pk):
     """
 
     # Generate schedule to be saved
-    schedule = instantiate_upcoming(sdl, show_pk, schedule_pk)
+    schedule = instantiate_upcoming_schedule(sdl, show_pk, schedule_pk)
 
     # Copy if first_date changes for generating timeslots
     gen_schedule = schedule

program/utils.py

@@ -60,7 +60,11 @@ def parse_time(date_string: str) -> time:
     """
     parse a time string and return a time object
     """
-    return datetime.strptime(date_string, "%H:%M:%S").time()
+
+    if len(date_string) == 5:
+        return datetime.strptime(date_string, "%H:%M").time()
+    else:
+        return datetime.strptime(date_string, "%H:%M:%S").time()
 
 
 def get_audio_url(cba_id: Optional[int]) -> str: