diff --git a/src/ring-scheduler/app.py b/src/ring-scheduler/app.py index 9517d31..48cdfb5 100644 --- a/src/ring-scheduler/app.py +++ b/src/ring-scheduler/app.py @@ -11,21 +11,18 @@ from datetime import datetime from zoneinfo import ZoneInfo from shared.ring_scheduler import update_queue_routing -from shared.schedule import FALLBACK_EXTENSION, ShiftSchedule +from shared.schedule import ( + FALLBACK_EXTENSION, + WEEKEND_DAYS, + ShiftSchedule, + determine_shift_type, +) from shared.secrets import get_secret logger = logging.getLogger() logger.setLevel(logging.INFO) EASTERN = ZoneInfo("America/New_York") -WEEKEND_DAYS = {"Saturday", "Sunday"} - - -def determine_shift_type(now: datetime) -> str: - day_name = now.strftime("%A") - if day_name in WEEKEND_DAYS and 8 <= now.hour < 17: - return "day" - return "night" def handler(event, context): diff --git a/src/shared/python/shared/schedule.py b/src/shared/python/shared/schedule.py index 40c6b83..aba7f2c 100644 --- a/src/shared/python/shared/schedule.py +++ b/src/shared/python/shared/schedule.py @@ -8,14 +8,30 @@ Single-table design: """ import os +from datetime import datetime +from zoneinfo import ZoneInfo import boto3 from boto3.dynamodb.conditions import Key - +EASTERN = ZoneInfo("America/New_York") +WEEKEND_DAYS = {"Saturday", "Sunday"} FALLBACK_EXTENSION = "100" +def determine_shift_type(now: datetime | None = None) -> str: + """Return the currently active shift type: 'day' or 'night'. + + Weekend 8am-5pm is the day shift; all other times are night. + """ + if now is None: + now = datetime.now(EASTERN) + day_name = now.strftime("%A") + if day_name in WEEKEND_DAYS and 8 <= now.hour < 17: + return "day" + return "night" + + class ShiftSchedule: def __init__(self, table_name: str | None = None): dynamodb = boto3.resource("dynamodb") diff --git a/src/slack-bot/app.py b/src/slack-bot/app.py index 707f6aa..8992d62 100644 --- a/src/slack-bot/app.py +++ b/src/slack-bot/app.py @@ -16,7 +16,12 @@ from shared.blocks import ( build_week_schedule, ) from shared.ring_scheduler import update_queue_routing -from shared.schedule import FALLBACK_EXTENSION, ShiftSchedule +from shared.schedule import ( + FALLBACK_EXTENSION, + WEEKEND_DAYS, + ShiftSchedule, + determine_shift_type, +) from shared.secrets import get_secret logger = logging.getLogger(__name__) @@ -88,16 +93,8 @@ def is_today(date_str: str) -> bool: return date_str == datetime.now(EASTERN).strftime("%Y-%m-%d") -WEEKEND_DAYS = {"Saturday", "Sunday"} - - def _is_active_shift_type(shift_type: str) -> bool: - """Check if the given shift type is the currently active one.""" - now = datetime.now(EASTERN) - day_name = now.strftime("%A") - if day_name in WEEKEND_DAYS and 8 <= now.hour < 17: - return shift_type == "day" - return shift_type == "night" + return determine_shift_type() == shift_type def _find_employee_shift(schedule, date_str, day_name, employee_ext):