Various fixes and support for ty

.env.template

@@ -1,3 +1,5 @@
+LOGGING_LEVEL=DEBUG
+
 CALENDAR_URL=
 CALENDAR_OUTLOOK_DAYS=
 CALENDAR_EVENT_MAXIMUM=

docker-compose.yml

@@ -1,12 +1,13 @@
 ---
-
 services:
   jumpstart:
     build: .
     container_name: Jumpstart
     ports:
       - "8000:8000"
     environment:
+      - LOGGING_LEVEL=${LOGGING_LEVEL}
+
       - CALENDAR_URL=${CALENDAR_URL}
       - CALENDAR_OUTLOOK_DAYS=${CALENDAR_OUTLOOK_DAYS}
       - CALENDAR_EVENT_MAXIMUM=${CALENDAR_EVENT_MAXIMUM}

mkdocs.yml

@@ -38,7 +38,7 @@ use_directory_urls: false
 nav:
   - Home: index.md
   - Getting Started: getting-started/getting-started.md
-  - Backend:      
+  - Backend:
       - Calendar: core/csh_calendar.md
       - Slack: core/slack.md
       - Wikithoughts: core/wikithoughts.md

src/api/endpoints.py

@@ -1,13 +1,12 @@
-from logging import getLogger, Logger
-
 import json
-import httpx
+from logging import Logger, getLogger
 
-from fastapi import APIRouter, Request, Form
+import httpx
+from fastapi import APIRouter, Form, Request
 from fastapi.responses import JSONResponse
 
-from core import slack, wikithoughts, cshcalendar
 from config import WATCHED_CHANNELS
+from core import cshcalendar, slack, wikithoughts
 
 logger: Logger = getLogger(__name__)
 router: APIRouter = APIRouter()
@@ -28,10 +27,14 @@ async def get_calendar() -> JSONResponse:
 	events: list[dict[str, str]] = []
 
 	try:
-		get_future_events_ical: list[
-			cshcalendar.CalendarInfo
-		] = await cshcalendar.get_future_events()
-		events = cshcalendar.format_events(get_future_events_ical)
+		get_future_events_ical: (
+			list[cshcalendar.CalendarInfo] | None
+		) = await cshcalendar.get_future_events()
+
+		if get_future_events_ical is None:
+			raise Exception("Gathering future events resulted in None")
+
+		events.extend(cshcalendar.format_events(get_future_events_ical))
 	except Exception as e:
 		logger.error(f"Error fetching calendar events: {e}")
 		return JSONResponse({"status": "error", "message": str(e)}, status_code=500)
@@ -118,13 +121,15 @@ async def message_actions(payload: str = Form(...)) -> JSONResponse:
 		if form_json.get("type") != "block_actions":
 			return JSONResponse({}, status_code=200)
 
+		message: str = DENY_MESSAGE
+
 		if slack.convert_user_response_to_bool(form_json):
 			logger.info(
 				"User approved the announcement, Adding it to the announcement list!"
 			)
 
-			message_object: dict[str, dict] = json.loads(
-				form_json.get("actions", [{}])[0].get("value", '{text:""}')
+			message_object: str | None = json.loads(
+				form_json.get("actions", [{}])[0].get("value", {})
 			).get("text", None)
 
 			user_id = form_json.get("user", {}).get("id")
@@ -133,20 +138,14 @@ async def message_actions(payload: str = Form(...)) -> JSONResponse:
 			username = username[:40]
 
 			slack.add_announcement(message_object, username)
-
-			if response_url:
-				async with httpx.AsyncClient() as client:
-					await client.post(
-						response_url,
-						json={"text": ACCEPT_MESSAGE, "replace_original": True},
-					)
-		else:
-			if response_url:
-				async with httpx.AsyncClient() as client:
-					await client.post(
-						response_url,
-						json={"text": DENY_MESSAGE, "replace_original": True},
-					)
+			message: str = ACCEPT_MESSAGE
+
+		if response_url:
+			async with httpx.AsyncClient() as client:
+				await client.post(
+					response_url,
+					json={"text": message, "replace_original": True},
+				)
 
 	except Exception as e:
 		logger.error(f"Error in message_actions: {e}")

src/config.py

@@ -1,13 +1,23 @@
-import os
 import json
 import logging
+import os
+from typing import overload
+
 from dotenv import load_dotenv
 
 load_dotenv()
 
 logger: logging.Logger = logging.getLogger(__name__)
 
 
+@overload
+def _get_env_variable(name: str, default: None = None) -> str | None: ...
+
+
+@overload
+def _get_env_variable(name: str, default: str) -> str: ...
+
+
 def _get_env_variable(name: str, default: str | None = None) -> str | None:
 	"""
 	Retrieves an environment variable, with an optional default value.
@@ -21,7 +31,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,12 +47,30 @@ def _get_env_variable(name: str, default: str | None = None) -> str | None:
 
 BASE_DIR: str = os.path.dirname(os.path.abspath(__file__))
 
+_raw_logging_level: str = _get_env_variable("LOGGING_LEVEL", "DEBUG")
+LOGGING_LEVEL: int = logging.INFO
+
+match _raw_logging_level:
+	case "DEBUG":
+		LOGGING_LEVEL = logging.DEBUG
+	case "WARN":
+		LOGGING_LEVEL = logging.WARN
+	case "ERROR":
+		LOGGING_LEVEL = logging.ERROR
+	case "FATAL":
+		LOGGING_LEVEL = logging.FATAL
+	case "CRITICAL":
+		LOGGING_LEVEL = logging.CRITICAL
+
 SLACK_API_TOKEN: str | None = _get_env_variable("SLACK_API_TOKEN", None)
 SLACK_JUMPSTART_MESSAGE: str = "Would you like to post this message to Jumpstart?"
-WATCHED_CHANNELS: tuple[str] = tuple(
-	_get_env_variable("WATCHED_CHANNELS", "").split(",")
+RAW_CHANNELS: str = _get_env_variable("WATCHED_CHANNELS", "")
+WATCHED_CHANNELS: tuple[str, ...] = tuple(RAW_CHANNELS.split(","))
+
+SLACK_DM_TEMPLATE_FILEPATH: str = os.path.join(
+	BASE_DIR, "static", "slack", "dm_request_template.json"
 )
-SLACK_DM_TEMPLATE: dict | None = None
+SLACK_DM_TEMPLATE: list | None = None
 
 CALENDAR_URL: str | None = _get_env_variable("CALENDAR_URL", None)
 CALENDAR_OUTLOOK_DAYS: int = int(_get_env_variable("CALENDAR_OUTLOOK_DAYS", "7"))
@@ -51,9 +79,10 @@ def _get_env_variable(name: str, default: str | None = None) -> str | None:
 CALENDAR_CACHE_REFRESH: int = int(_get_env_variable("CALENDAR_CACHE_REFRESH", "10"))
 
 WIKI_API: str | None = _get_env_variable("WIKI_API", None)
-WIKIBOT_USER: str | None = _get_env_variable("WIKIBOT_USER", None)
-WIKIBOT_PASSWORD: str | None = _get_env_variable("WIKIBOT_PASSWORD", None)
+WIKIBOT_USER: str = _get_env_variable("WIKIBOT_USER", "")
+WIKIBOT_PASSWORD: str = _get_env_variable("WIKIBOT_PASSWORD", "")
 WIKI_CATEGORY: str = _get_env_variable("WIKI_CATEGORY", "JobAdvice")
 
-with open(os.path.join(BASE_DIR, "static", "slack", "dm_request_template.json")) as f:
-	SLACK_DM_TEMPLATE = json.load(f)
+if os.path.exists(SLACK_DM_TEMPLATE_FILEPATH):
+	with open(SLACK_DM_TEMPLATE_FILEPATH, mode="r") as f:
+		SLACK_DM_TEMPLATE = json.load(f)

src/core/cshcalendar.py

@@ -1,21 +1,22 @@
-from logging import getLogger, Logger
-from datetime import datetime, date, timedelta, time
+import asyncio
+import re
+from datetime import date, datetime, time, timedelta
+from logging import Logger, getLogger
 from zoneinfo import ZoneInfo
 
-from icalendar.cal import Event, Calendar
+import arrow
 import httpx
 import recurring_ical_events
-import arrow
-import re
+from icalendar.cal import Calendar, Event
 
 from config import (
 	CALENDAR_CACHE_REFRESH,
 	CALENDAR_EVENT_MAXIMUM,
 	CALENDAR_OUTLOOK_DAYS,
 	CALENDAR_TIMEZONE,
 	CALENDAR_URL,
+	LOGGING_LEVEL,
 )
-import asyncio
 
 calendar_cache: list[CalendarInfo] = []  # The current cache of the calendar
 cal_last_update: date | None = (
@@ -31,6 +32,8 @@
 cal_constructed_event.clear()
 
 logger: Logger = getLogger(__name__)
+logger.setLevel(LOGGING_LEVEL)
+
 logger.info("Starting up the calendar service!")
 cshcal_client = httpx.AsyncClient()
 
@@ -45,7 +48,7 @@
 WARNING: PERCENTAGE SIGNS WILL TRIGGER A REGEX OPERATION
 WARNING: FOLLOW INSERTION ORDER
 """
-HUMANIZER_CHECKS: dict[int, str] = {
+HUMANIZER_CHECKS: dict[int | float, str] = {
 	MINUTE: "In 1 Minute",
 	(HOUR - MINUTE): f"In %{MINUTE}% Minutes",
 	(HOUR * 1.5): "In 1 Hour",
@@ -67,7 +70,7 @@ class CalendarInfo:
 
 	def __init__(self, name: str, date_time: date, location: str | None = None):
 		self.name: str = name
-		self.date: arrow.arrow = arrow.get(date_time)  # Arrow has way cooler stuff
+		self.date: arrow.Arrow = arrow.get(date_time)  # Arrow has way cooler stuff
 		self.location: str | None = location
 
 	def __eq__(self, other):
@@ -93,7 +96,7 @@ def ceil_division(num: int, den: int) -> int:
 	return (num + den - 1) // den
 
 
-def time_humanizer(current_time: datetime, event_time: datetime) -> str:
+def time_humanizer(current_time: datetime, event_time: arrow.Arrow) -> str:
 	"""
 	Custom humanizer for text to be displayed
 
@@ -118,7 +121,7 @@ def repl(match: re.Match[str]) -> str:
 		num = int(match.group(1))
 		return str(round(time_before_event / num))
 
-	time_before_event: int = (event_time - current_time).total_seconds()
+	time_before_event: int | float = (event_time - current_time).total_seconds()
 
 	if time_before_event > WEEK:
 		return "Over a Week Away"
@@ -136,6 +139,7 @@ def repl(match: re.Match[str]) -> str:
 
 	return TIME_PATTERN.sub(repl, unformatted_string)
 
+
 def format_events(events: list[CalendarInfo]) -> list[dict[str, str]]:
 	"""
 	Formats a parsed list of CalendarInfos, and returns the HTML required for front end
@@ -150,25 +154,21 @@ def format_events(events: list[CalendarInfo]) -> list[dict[str, str]]:
 	current_date: date = datetime.now(ZoneInfo(CALENDAR_TIMEZONE))
 
 	if not events:
-		return {"data": [{"header": ":(", "content": "No Events on the Calendar"}]}
+		return [{"header": ":(", "content": "No Events on the Calendar"}]
 
 	formatted_list: list[dict[str, str]] = []
 
 	for event in events:
-		content_dict: dict[str, str] = {}
+		content_dict: dict[str, str] = {"content": str(event.name)}
 
-		event_cur_happening: bool = event.date < current_date
-		if event_cur_happening:
-			formatted: str = (
+		if event.date < current_date:
+			content_dict["header"] = (
 				f"Happening in {event.location}!"
 				if event.location
 				else "Happening Now!"
 			)
-			content_dict["header"] = formatted
-			content_dict["content"] = str(event.name)
 		else:
 			content_dict["header"] = time_humanizer(current_date, event.date)
-			content_dict["content"] = str(event.name)
 
 		formatted_list.append(content_dict)
 	return formatted_list
@@ -181,14 +181,17 @@ async def rebuild_calendar() -> None:
 
 	global calendar_cache, cal_last_update, cal_constructed_event
 
+	if CALENDAR_URL is None:
+		raise Exception("Calendar URL is None, cant request.")
+
 	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)
+		cal: Calendar | None = 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)
@@ -199,10 +202,8 @@ async def rebuild_calendar() -> None:
 
 			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))
 
@@ -214,11 +215,10 @@ async def rebuild_calendar() -> None:
 
 			found_events.add(new_event)
 
-		cal = None
-		fetched_daily_events = None
+		fetched_daily_events.clear()
 	except Exception as e:
 		logger.warning("Failed to rebuild calendar cache! Error:")
-		logger.warning(e)
+		logger.error(e)
 		cal_constructed_event.set()
 
 	cal_last_update = current_time
@@ -228,7 +228,7 @@ async def rebuild_calendar() -> None:
 	cal_constructed_event.set()
 
 
-async def get_future_events() -> list[CalendarInfo]:
+async def get_future_events() -> list[CalendarInfo] | None:
 	"""
 	Returns the first events up to event maximum within the the calendar outlook day amount
 	custom object has name, date and the location
@@ -244,6 +244,9 @@ async def get_future_events() -> list[CalendarInfo]:
 		header_none_match, \
 		cal_constructed_event
 
+	if CALENDAR_URL is None:
+		raise Exception("Calendar URL is None, cant request.")
+
 	if not cal_constructed_event.is_set():
 		await cal_constructed_event.wait()
 		return calendar_cache
@@ -261,10 +264,11 @@ async def get_future_events() -> list[CalendarInfo]:
 
 	logger.info("Checking to rebuild CSH Calendar...")
 	try:
-		headers: dict[str, str | None] = {}
+		headers: dict[str, str] = {}
 
 		if header_none_match:
 			headers["If-None-Match"] = header_none_match
+
 		if header_last_modified:
 			headers["If-Modified-Since"] = header_last_modified