Add Auto Announcement Pings for events

.env.template

@@ -4,6 +4,13 @@ CALENDAR_EVENT_MAXIMUM=
 CALENDAR_CACHE_REFRESH=
 CALENDAR_TIMEZONE=
 
+SLACK_ALLOW_ANNOUNCEMENTS=false
+SLACK_ANNOUNCEMENT_CHANNEL=
+SLACK_ACTIVE_GROUP_ID=
+SLACK_MEETINGS_GROUP_ID=
+SLACK_FROSH_GROUP_ID=
+SLACK_TEST_GROUP_ID=
+
 WATCHED_CHANNELS=
 SLACK_API_TOKEN=
 SLACK_SIGNING_SECRET=

src/config.py

@@ -2,13 +2,14 @@
 import json
 import logging
 from dotenv import load_dotenv
+from typing import Any
 
 load_dotenv()
 
 logger: logging.Logger = logging.getLogger(__name__)
 
 
-def _get_env_variable(name: str, default: str | None = None) -> str | None:
+def _get_env_variable(name: str, default: str | None = None) -> str | Any:
 	"""
 	Retrieves an environment variable, with an optional default value.
 
@@ -21,7 +22,7 @@ def _get_env_variable(name: str, default: str | None = None) -> str | None:
 	"""
 
 	try:
-		value: str = os.getenv(name, default)
+		value: str | None = os.getenv(name, default)
 
 		if value in (None, ""):
 			logger.warning(
@@ -37,16 +38,25 @@ def _get_env_variable(name: str, default: str | None = None) -> str | None:
 
 BASE_DIR: str = os.path.dirname(os.path.abspath(__file__))
 
-SLACK_API_TOKEN: str | None = _get_env_variable("SLACK_API_TOKEN", None)
+SLACK_ALLOW_ANNOUNCEMENTS: bool = (
+	_get_env_variable("SLACK_ALLOW_ANNOUNCEMENTS", "false") == "true"
+)
+SLACK_ANNOUNCEMENT_CHANNEL: str = _get_env_variable("SLACK_ANNOUNCEMENT_CHANNEL", "")
+SLACK_ACTIVE_GROUP_ID: str = _get_env_variable("SLACK_ACTIVE_GROUP_ID", "")
+SLACK_MEETINGS_GROUP_ID: str = _get_env_variable("SLACK_MEETINGS_GROUP_ID", "")
+SLACK_FROSH_GROUP_ID: str = _get_env_variable("SLACK_FROSH_GROUP_ID", "")
+SLACK_TEST_GROUP_ID: str = _get_env_variable("SLACK_TEST_GROUP_ID", "")
+
+SLACK_API_TOKEN: str = _get_env_variable("SLACK_API_TOKEN", "")
 SLACK_JUMPSTART_MESSAGE: str = "Would you like to post this message to Jumpstart?"
 SLACK_SIGNING_SECRET: str = _get_env_variable("SLACK_SIGNING_SECRET", None)
 
 WATCHED_CHANNELS: tuple[str] = tuple(
-	_get_env_variable("WATCHED_CHANNELS", "").split(",")
+	_get_env_variable("WATCHED_CHANNELS", "0,1,2").split(",")
 )
 SLACK_DM_TEMPLATE: dict | None = None
 
-CALENDAR_URL: str | None = _get_env_variable("CALENDAR_URL", None)
+CALENDAR_URL: str = _get_env_variable("CALENDAR_URL", "")
 CALENDAR_OUTLOOK_DAYS: int = int(_get_env_variable("CALENDAR_OUTLOOK_DAYS", "7"))
 CALENDAR_EVENT_MAXIMUM: int = int(_get_env_variable("CALENDAR_EVENT_MAXIMUM", "10"))
 CALENDAR_TIMEZONE: str = _get_env_variable("CALENDAR_TIMEZONE", "America/New_York")

src/core/announcement_queue.py

@@ -0,0 +1,197 @@
+from logging import getLogger, Logger
+
+from slack_sdk.web.async_client import AsyncWebClient
+
+from modules import taskmanager
+from core import slack
+
+from datetime import datetime, timedelta
+from zoneinfo import ZoneInfo
+import asyncio
+
+from typing import Any
+from config import (
+	CALENDAR_TIMEZONE,
+	SLACK_ACTIVE_GROUP_ID,
+	SLACK_FROSH_GROUP_ID,
+	SLACK_MEETINGS_GROUP_ID,
+	SLACK_ALLOW_ANNOUNCEMENTS,
+)
+
+logger: Logger = getLogger(__name__)
+client: AsyncWebClient | None = None
+
+event_id_cache: dict[str, str] = {}
+queued_announcement_id_cache: dict[str, asyncio.Task] = {}
+
+TEN_MINUTES = 60 * 10
+TECHNICAL_SEMINAR_KEYWORD: str = "technical"
+STANDARD_SEMINAR_KEYWORD: str = "seminar"
+MEETING_KEYWORD: str = "meeting"
+TEST_KEYWORD: str = "test_gick"
+
+MINUTES_BEFORE_EVENT_PING = 15
+
+
+async def create_announcement_worker(
+	event_uid: str, event_recurrence_id: str, text: str, event_time: datetime
+) -> None:
+	"""
+	Creates a new worker that will send an announcement 15 minutes before the stated event
+
+	Args:
+		event_uid (str): The UID for the recurring event
+		event_recurrence_id (str): The ID for which occuring event it is.
+		text (str): The message to be sent
+		event_time (datetime): The time for the event.
+	"""
+	key: str = f"{event_uid}:{event_recurrence_id}"  # we should use redis instead
+
+	current_time: datetime = datetime.now(ZoneInfo(CALENDAR_TIMEZONE))
+	if current_time < (event_time - timedelta(minutes=MINUTES_BEFORE_EVENT_PING)):
+		wait_time = (
+			event_time - current_time - timedelta(minutes=MINUTES_BEFORE_EVENT_PING)
+		)
+		try:
+			await asyncio.sleep(wait_time.total_seconds())
+			await slack.send_announcement_message(text)
+		except asyncio.CancelledError:
+			logger.info("Announcement worker cancelled: %s", key)
+			raise
+		finally:
+			task = asyncio.current_task()
+
+			if queued_announcement_id_cache.get(key) is task:
+				queued_announcement_id_cache.pop(key, None)
+
+
+def queue_announcement(
+	event_uid: str,
+	event_recurrence_id: str,
+	text: str,
+	event_time: datetime,
+) -> None:
+	key = f"{event_uid}:{event_recurrence_id}"
+
+	existing_task = queued_announcement_id_cache.get(key)
+
+	if existing_task is not None:
+		existing_task.cancel()
+
+	task = taskmanager.create_background_task(
+		create_announcement_worker(
+			event_uid,
+			event_recurrence_id,
+			text,
+			event_time,
+		)
+	)
+
+	queued_announcement_id_cache[key] = task
+
+
+def clear_running_workers() -> None:
+	"""
+	Loops through and removes each running worker event. Used for clearing events that have been deleted
+	"""
+
+	for event in queued_announcement_id_cache.values():
+		event.cancel()
+
+	queued_announcement_id_cache.clear()
+
+
+def check_for_announcement(event: dict[str, Any], time: datetime) -> None:
+	"""
+	Checks to see if a worker needs to be created for an event
+
+	Args:
+		event (dict[str, str]): The information for the event
+		time (datetime): The time for the event
+	"""
+
+	if not SLACK_ALLOW_ANNOUNCEMENTS:
+		return
+
+	description: str = event.get("DESCRIPTION", "")
+	if not description:
+		return
+
+	title: str = event.get("SUMMARY", "")
+	if not title:
+		return
+
+	uid: str = str(event.get("UID", ""))
+	if not uid:
+		return
+
+	recurrence_id = event.get("RECURRENCE-ID", None)
+	if not recurrence_id:
+		return
+
+	rec_id: str = recurrence_id.dt.isoformat()
+	loc = event.get("LOCATION", None)
+
+	description = description.lower().strip()
+	if TECHNICAL_SEMINAR_KEYWORD.lower() in description:
+		if loc:
+			queue_announcement(
+				uid,
+				rec_id,
+				f"<!subteam^{SLACK_ACTIVE_GROUP_ID}> <!subteam^{SLACK_FROSH_GROUP_ID}> The Technical Seminar {title} will be happening in the {loc} in {MINUTES_BEFORE_EVENT_PING} minutes!",
+				time,
+			)
+		else:
+			queue_announcement(
+				uid,
+				rec_id,
+				f"<!subteam^{SLACK_ACTIVE_GROUP_ID}> <!subteam^{SLACK_FROSH_GROUP_ID}> The Technical Seminar {title} will be happening in {MINUTES_BEFORE_EVENT_PING} minutes!",
+				time,
+			)
+	elif STANDARD_SEMINAR_KEYWORD.lower() in description:
+		if loc:
+			queue_announcement(
+				uid,
+				rec_id,
+				f"<!subteam^{SLACK_ACTIVE_GROUP_ID}> <!subteam^{SLACK_FROSH_GROUP_ID}> The Non-Technical Seminar {title} will be happening in the {loc} in {MINUTES_BEFORE_EVENT_PING} minutes!",
+				time,
+			)
+		else:
+			queue_announcement(
+				uid,
+				rec_id,
+				f"<!subteam^{SLACK_ACTIVE_GROUP_ID}> <!subteam^{SLACK_FROSH_GROUP_ID}> The Non-Technical Seminar {title} will be happening in {MINUTES_BEFORE_EVENT_PING} minutes!",
+				time,
+			)
+	elif MEETING_KEYWORD.lower() in description:
+		if loc:
+			queue_announcement(
+				uid,
+				rec_id,
+				f"<!subteam^{SLACK_MEETINGS_GROUP_ID}> The {title} directorship will be happening in the {loc} in {MINUTES_BEFORE_EVENT_PING} minutes!",
+				time,
+			)
+		else:
+			queue_announcement(
+				uid,
+				rec_id,
+				f"<!subteam^{SLACK_MEETINGS_GROUP_ID}> The {title} directorship will be happening in {MINUTES_BEFORE_EVENT_PING} minutes!",
+				time,
+			)
+
+
+# 	elif TEST_KEYWORD.lower() in description:
+# 		if loc:
+# 			queue_announcement(
+# 				uid, rec_id, f"<!subteam^{SLACK_TEST_GROUP_ID}> testing in the {loc}!", time
+# 			)
+# 		else:
+# queue_announcement(
+# 					uid, rec_id, f"<!subteam^{SLACK_TEST_GROUP_ID}> testing!", time
+# 				)
+# 			)
+
+# if TECHNICAL_SEMINAR_KEYWORD.lower() in description:
+# 	taskmanager.create_background_task(create_announcement_worker(
+# 		uid, rec_id, f"<!subteam^{SLACK_ACTIVE_GROUP_ID}> reminder: meeting starting soon", time
+# 	))

src/core/cshcalendar.py

@@ -2,6 +2,8 @@
 from datetime import datetime, date, timedelta, time
 from zoneinfo import ZoneInfo
 
+from core import announcement_queue
+
 from icalendar.cal import Event, Calendar
 import httpx
 import recurring_ical_events
@@ -34,6 +36,7 @@
 
 logger: Logger = getLogger(__name__)
 logger.info("Starting up the calendar service!")
+
 cshcal_client = httpx.AsyncClient()
 
 # Conversion from seconds
@@ -59,6 +62,8 @@
 BORDER_STRING: str = '<hr class="calendar-border">'
 TIME_PATTERN = re.compile(r"%([^%]+)%")
 
+calendar_rebuild_lock: asyncio.Lock = asyncio.Lock()
+
 
 # Automatically format all info into the class
 class CalendarInfo:
@@ -184,51 +189,55 @@ async def rebuild_calendar() -> None:
 
 	global calendar_cache, cal_last_update, cal_constructed_event
 
-	current_time: datetime = datetime.now(ZoneInfo(CALENDAR_TIMEZONE))
-	try:
-		cal_constructed_event.clear()
-		found_events: set[CalendarInfo] = set()
-		response: httpx.Response = await cshcal_client.get(CALENDAR_URL, timeout=10)
-		response.raise_for_status()
-
-		cal: Calendar = Calendar.from_ical(response.content)
-
-		fetched_daily_events: list[Event] = recurring_ical_events.of(cal).between(
-			current_time, current_time + timedelta(days=CALENDAR_OUTLOOK_DAYS)
-		)
+	async with calendar_rebuild_lock:
+		current_time: datetime = datetime.now(ZoneInfo(CALENDAR_TIMEZONE))
+		try:
+			cal_constructed_event.clear()
+			found_events: set[CalendarInfo] = set()
+			response: httpx.Response = await cshcal_client.get(CALENDAR_URL, timeout=20)
+			response.raise_for_status()
 
-		for event in fetched_daily_events:
-			dt = event.get("DTSTART").dt
+			cal: Calendar = Calendar.from_ical(response.content)
 
-			if isinstance(dt, date) and not isinstance(dt, datetime):
-				dt = datetime.combine(dt, time.min, tzinfo=ZoneInfo(CALENDAR_TIMEZONE))
-
-			elif dt.tzinfo is None:
-				dt = dt.replace(tzinfo=ZoneInfo(CALENDAR_TIMEZONE))
-
-			else:
-				dt = dt.astimezone(ZoneInfo(CALENDAR_TIMEZONE))
-
-			new_event: CalendarInfo = CalendarInfo(
-				event.get("SUMMARY"),
-				dt,
-				event.get("LOCATION"),
+			fetched_daily_events: list[Event] = recurring_ical_events.of(cal).between(
+				current_time, current_time + timedelta(days=CALENDAR_OUTLOOK_DAYS)
 			)
 
-			found_events.add(new_event)
-
-		cal = None
-		fetched_daily_events = None
-	except Exception as e:
-		logger.warning("Failed to rebuild calendar cache! Error:")
-		logger.warning(e)
-		cal_constructed_event.set()
-
-	cal_last_update = current_time
-	calendar_cache = sorted(found_events, key=lambda x: x.date)[
-		:CALENDAR_EVENT_MAXIMUM
-	]  # Only cache the first elements of this list
-	cal_constructed_event.set()
+			announcement_queue.clear_running_workers()
+
+			for event in fetched_daily_events:
+				dt = event.get("DTSTART").dt
+
+				if isinstance(dt, date) and not isinstance(dt, datetime):
+					dt = datetime.combine(
+						dt, time.min, tzinfo=ZoneInfo(CALENDAR_TIMEZONE)
+					)
+
+				elif dt.tzinfo is None:
+					dt = dt.replace(tzinfo=ZoneInfo(CALENDAR_TIMEZONE))
+
+				else:
+					dt = dt.astimezone(ZoneInfo(CALENDAR_TIMEZONE))
+
+				new_event: CalendarInfo = CalendarInfo(
+					event.get("SUMMARY"),
+					dt,
+					event.get("LOCATION"),
+				)
+
+				announcement_queue.check_for_announcement(event, dt)
+				found_events.add(new_event)
+
+			cal_last_update = current_time
+			calendar_cache = sorted(found_events, key=lambda x: x.date)[
+				:CALENDAR_EVENT_MAXIMUM
+			]  # Only cache the first elements of this list
+		except Exception as e:
+			logger.warning("Failed to rebuild calendar cache! Error:")
+			logger.warning(e)
+			cal_constructed_event.set()
+		finally:
+			cal_constructed_event.set()
 
 
 async def get_future_events() -> list[CalendarInfo]: