diff --git a/.env.template b/.env.template index 86e75db..a004f3b 100644 --- a/.env.template +++ b/.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= diff --git a/src/config.py b/src/config.py index 4ff172c..9c02091 100644 --- a/src/config.py +++ b/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") diff --git a/src/core/announcement_queue.py b/src/core/announcement_queue.py new file mode 100644 index 0000000..1756454 --- /dev/null +++ b/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" The Technical Seminar {title} will be happening in the {loc} in {MINUTES_BEFORE_EVENT_PING} minutes!", + time, + ) + else: + queue_announcement( + uid, + rec_id, + f" 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" 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" 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" The {title} directorship will be happening in the {loc} in {MINUTES_BEFORE_EVENT_PING} minutes!", + time, + ) + else: + queue_announcement( + uid, + rec_id, + f" 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" testing in the {loc}!", time +# ) +# else: +# queue_announcement( +# uid, rec_id, f" testing!", time +# ) +# ) + +# if TECHNICAL_SEMINAR_KEYWORD.lower() in description: +# taskmanager.create_background_task(create_announcement_worker( +# uid, rec_id, f" reminder: meeting starting soon", time +# )) diff --git a/src/core/cshcalendar.py b/src/core/cshcalendar.py index 2e16cae..777dbc5 100644 --- a/src/core/cshcalendar.py +++ b/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 = '
' 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]: diff --git a/src/core/slack.py b/src/core/slack.py index e5ad40c..56d589b 100644 --- a/src/core/slack.py +++ b/src/core/slack.py @@ -12,6 +12,7 @@ from modules import taskmanager from config import ( + SLACK_ANNOUNCEMENT_CHANNEL, SLACK_API_TOKEN, SLACK_JUMPSTART_MESSAGE, SLACK_DM_TEMPLATE, @@ -319,6 +320,21 @@ async def process_slack_message_actions(payload: str): return ({"status": "success"}, 200) +async def send_announcement_message(msg_text: str) -> None: + """ + Sends a message to the given announcements channel + + Args: + msg_text (str): The text for the message + + """ + if not client: + logger.warning("Client has not been initalized") + return + + await client.chat_postMessage(channel=SLACK_ANNOUNCEMENT_CHANNEL, text=msg_text) + + def convert_user_response_to_bool(message_data: dict) -> bool: """ Converts a Slack message action response to a boolean indicating whether the user approved the announcement. diff --git a/src/main.py b/src/main.py index 8b94c94..8c81b78 100644 --- a/src/main.py +++ b/src/main.py @@ -6,7 +6,6 @@ """ import os -import asyncio from logging import getLogger, Logger @@ -20,6 +19,7 @@ from api import endpoints from core import wikithoughts, cshcalendar +from modules import taskmanager logger: Logger = getLogger(__name__) @@ -27,7 +27,8 @@ @asynccontextmanager async def lifespan(app: FastAPI): logger.info("Starting up the Jumpstart application!") - asyncio.create_task(cshcalendar.rebuild_calendar()) + taskmanager.create_background_task(cshcalendar.rebuild_calendar()) + taskmanager.create_background_task(taskmanager.calendar_worker()) await wikithoughts.auth_bot() yield diff --git a/src/modules/taskmanager.py b/src/modules/taskmanager.py index 2dd58b0..a0c5610 100644 --- a/src/modules/taskmanager.py +++ b/src/modules/taskmanager.py @@ -1,5 +1,6 @@ import asyncio +from core import cshcalendar from logging import getLogger, Logger from collections.abc import Coroutine @@ -7,6 +8,8 @@ running_background_tasks: set[asyncio.Task] = set() +TWENTY_MINUTES = 60 * 20 + def handle_task_exception(task: asyncio.Task) -> None: """ @@ -41,3 +44,12 @@ def create_background_task(coroutine: Coroutine) -> asyncio.Task: task.add_done_callback(running_background_tasks.discard) task.add_done_callback(handle_task_exception) return task + + +async def calendar_worker(): + """ + Loop to force rebuild the calendar every 20 minutes to check for event updates + """ + while True: + await asyncio.sleep(TWENTY_MINUTES) + await cshcalendar.rebuild_calendar()