diff --git a/pyproject.toml b/pyproject.toml index a815ef9a..d630278d 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -40,20 +40,21 @@ build-backend = "poetry.core.masonry.api" [tool.pyright] strict = ["**"] exclude = [ + # large and/or annoying multiclass files "**/advent.py", - "**/bot.py", - "**/error_handler.py", "**/events.py", - "**/gaming.py", - "**/haiku.py", - "**/member_counter.py", "**/remindme.py", - "**/snailrace.py", "**/starboard.py", - "**/uptime.py", "**/whatsdue.py", - "**/working_on.py", + + # waiting for external stubs + "**/bot.py", + + # only used by another blocked file + "**/error_handler.py", "**/utils/command_utils.py", - "**/utils/snailrace_utils.py", - "**/utils/uq_course_utils.py" + "**/utils/uq_course_utils.py", + + # isaac's job + "**/haiku.py" ] \ No newline at end of file diff --git a/uqcsbot/advent.py b/uqcsbot/advent.py index d37f6563..13d666ed 100644 --- a/uqcsbot/advent.py +++ b/uqcsbot/advent.py @@ -159,10 +159,6 @@ def sort_key(sort: SortMode) -> Callable[["Member"], Any]: class Advent(commands.Cog): CHANNEL_NAME = "contests" - # Session cookie (will expire in approx 30 days). - # See: https://github.com/UQComputingSociety/uqcsbot-discord/wiki/Tokens-and-Environment-Variables#aoc_session_id - SESSION_ID: str = "" - def __init__(self, bot: UQCSBot): self.bot = bot self.bot.schedule_task( @@ -183,8 +179,10 @@ def __init__(self, bot: UQCSBot): month=12, ) - if os.environ.get("AOC_SESSION_ID") is not None: - SESSION_ID = os.environ.get("AOC_SESSION_ID") + # Session cookie (will expire in approx 30 days). + # See: https://github.com/UQComputingSociety/uqcsbot-discord/wiki/Tokens-and-Environment-Variables#aoc_session_id + if (session := os.environ.get("AOC_SESSION_ID")) is not None: + self.SESSION_ID = session else: raise FatalErrorWithLog( bot, "Unable to find AoC session ID. Not loading advent cog." diff --git a/uqcsbot/bot.py b/uqcsbot/bot.py index e2b045cf..3ba88071 100644 --- a/uqcsbot/bot.py +++ b/uqcsbot/bot.py @@ -13,7 +13,7 @@ """ TODO: TYPE ISSUES IN THIS FILE: - - apscheduler has no stubs. They're planned for the 4.0 release... in the future. + - apscheduler has no stubs. They're planned for the 4.0 release... sometime. - aiohttp handler witchery """ @@ -29,6 +29,8 @@ def __init__(self, *args: Any, **kwargs: Any): # Important channel names & constants go here self.ADMIN_ALERTS_CNAME = "admin-alerts" self.GENERAL_CNAME = "general" + self.BOT_CNAME = "bot-testing" + self.STARBOARD_CNAME = "starboard" self.BOT_TIMEZONE = timezone("Australia/Brisbane") self.uqcs_server: discord.Guild diff --git a/uqcsbot/gaming.py b/uqcsbot/gaming.py index 9793f6e1..1168218b 100644 --- a/uqcsbot/gaming.py +++ b/uqcsbot/gaming.py @@ -1,6 +1,6 @@ from difflib import SequenceMatcher from json import loads -from typing import Optional, Dict, Any +from typing import Optional, Dict, Any, Set, TypedDict, Union from urllib.error import HTTPError from urllib.request import urlopen from xml.etree.ElementTree import fromstring @@ -13,6 +13,23 @@ from uqcsbot.bot import UQCSBot +class Parameters(TypedDict): + categories: Set[str] + mechanics: Set[str] + subranks: Dict[Any, Any] + identity: str + min_players: int + max_players: int + name: str + score: Union[str, None] + users: Union[str, None] + rank: str + description: Union[str, None] + image: Union[str, None] + min_time: str + max_time: str + + class Gaming(commands.Cog): """ Various gaming related commands @@ -36,18 +53,21 @@ def get_bgg_id(self, search_name: str) -> Optional[str]: return None # filters for the closest name match - match = {} + match: Dict[str, float] = {} for item in results: - if item.get("id") is None: + item_id = item.get("id") + if item_id is None: continue for element in item: if element.tag == "name": - match[item.get("id")] = SequenceMatcher( - None, search_name, element.get("value") + match[item_id] = SequenceMatcher( + None, + search_name, + (x if (x := element.get("value")) is not None else ""), ).ratio() - return max(match, key=match.get) + return max(match, key=lambda x: match.get(x, 0)) - def get_board_game_parameters(self, identity: str) -> Optional[Dict[str, str]]: + def get_board_game_parameters(self, identity: str) -> Optional[Parameters]: """ returns the various parameters of a board game from bgg """ @@ -57,28 +77,39 @@ def get_board_game_parameters(self, identity: str) -> Optional[Dict[str, str]]: if query.status_code != 200: return None result = fromstring(query.text)[0] - parameters: Dict[str, Any] = {} - parameters["categories"] = set() - parameters["mechanics"] = set() - parameters["subranks"] = {} - parameters["identity"] = identity + parameters: Parameters = { + "categories": set(), + "mechanics": set(), + "subranks": {}, + "identity": identity, + "min_players": -1, + "max_players": -1, + "name": "", + "score": None, + "users": None, + "rank": "", + "description": None, + "image": None, + "min_time": "", + "max_time": "", + } for element in result: tag = element.tag tag_name = element.attrib.get("name") - tag_value = element.attrib.get("value") + tag_value = element.attrib.get("value", "") tag_type = element.attrib.get("type") tag_text = element.text # sets the range of players if tag == "poll" and tag_name == "suggested_numplayers": - players = set() + players: Set[int] = set() for option in element: - numplayers = option.attrib.get("numplayers") + numplayers = option.attrib.get("numplayers", "0") votes = 0 for result in option: - numvotes = int(result.attrib.get("numvotes")) + numvotes = int(result.attrib.get("numvotes", "0")) direction = ( -1 if result.attrib.get("value") == "Not Recommended" else 1 ) @@ -95,12 +126,12 @@ def get_board_game_parameters(self, identity: str) -> Optional[Dict[str, str]]: parameters["max_players"] = max(players) # sets the backup min players - if tag == "minplayers": - parameters.setdefault("min_players", int(tag_value)) + if tag == "minplayers" and parameters["min_players"] == -1: + parameters["min_players"] = int(tag_value) # sets the backup max players - if tag == "maxplayers": - parameters.setdefault("max_players", int(tag_value)) + if tag == "maxplayers" and parameters["max_players"] == -1: + parameters["max_players"] = int(tag_value) # sets the name of the board game elif tag == "name" and tag_type == "primary": @@ -118,7 +149,7 @@ def get_board_game_parameters(self, identity: str) -> Optional[Dict[str, str]]: elif tag == "statistics": for statistic in element[0]: stat_tag = statistic.tag - stat_value = statistic.attrib.get("value") + stat_value = statistic.attrib.get("value", "") if stat_tag == "average": try: parameters["score"] = str(round(float(stat_value), 2)) @@ -129,7 +160,7 @@ def get_board_game_parameters(self, identity: str) -> Optional[Dict[str, str]]: if stat_tag == "ranks": for genre in statistic: genre_name = genre.attrib.get("name") - genre_value = genre.attrib.get("value") + genre_value = genre.attrib.get("value", "") if genre_name == "boardgame" and genre_value.isnumeric(): position = int(genre_value) # gets the ordinal suffix @@ -141,7 +172,7 @@ def get_board_game_parameters(self, identity: str) -> Optional[Dict[str, str]]: ] parameters["rank"] = f"{position:d}{suffix:s}" elif genre_value.isnumeric(): - friendlyname = genre.attrib.get("friendlyname") + friendlyname = genre.attrib.get("friendlyname", "") # removes "game" as last word friendlyname = " ".join(friendlyname.split(" ")[:-1]) position = int(genre_value) @@ -170,7 +201,7 @@ def get_board_game_parameters(self, identity: str) -> Optional[Dict[str, str]]: return parameters - def format_board_game_parameters(self, parameters: Dict[str, str]) -> discord.Embed: + def format_board_game_parameters(self, parameters: Parameters) -> discord.Embed: embed = discord.Embed(title=parameters.get("name", ":question:")) embed.add_field( name="Summary", @@ -198,12 +229,14 @@ def format_board_game_parameters(self, parameters: Dict[str, str]) -> discord.Em f"• Ranked {value:s} in the {key:s} genre.\n" for key, value in parameters.get("subranks", {}).items() ) - + f"Categories: {', '.join(parameters.get('categories', set())):s}\n" - f"Mechanics: {', '.join(parameters.get('mechanics', set())):s}\n" + + f"Categories: {', '.join(parameters['categories']):s}\n" + + f"Mechanics: {', '.join(parameters['mechanics']):s}\n" ), ) max_message_length = 1000 - description = parameters.get("description", ":question:") + description: str = ( + x if (x := parameters["description"]) is not None else ":question:" + ) if len(description) > max_message_length: description = description[:max_message_length] + "\u2026" embed.add_field(name="Description", inline=False, value=description) diff --git a/uqcsbot/member_counter.py b/uqcsbot/member_counter.py index a6cda54d..b59a6056 100644 --- a/uqcsbot/member_counter.py +++ b/uqcsbot/member_counter.py @@ -7,23 +7,27 @@ from discord import app_commands from discord.ext import commands +from uqcsbot.bot import UQCSBot +from typing import List + class MemberCounter(commands.Cog): MEMBER_COUNT_PREFIX = "Member Count: " RATE_LIMIT = timedelta(minutes=5) NEW_MEMBER_TIME = timedelta(days=7) - def __init__(self, bot: commands.Bot): + def __init__(self, bot: UQCSBot): self.bot = bot self.last_rename_time = datetime.now() self.waiting_for_rename = False @commands.Cog.listener() async def on_ready(self): - member_count_channels = [ + member_count_channels: List[discord.VoiceChannel] = [ channel for channel in self.bot.uqcs_server.channels if channel.name.startswith(self.MEMBER_COUNT_PREFIX) + and isinstance(channel, discord.VoiceChannel) ] if len(member_count_channels) == 0: logging.warning( @@ -43,7 +47,14 @@ async def on_ready(self): ) return - bot_member = self.bot.uqcs_server.get_member(self.bot.user.id) + if ( + bot_member := self.bot.uqcs_server.get_member(self.bot.safe_user.id) + ) is None: + logging.warning( + f"Unable to determine bot permissions for managing #Member Count channel." + ) + return + permissions = self.member_count_channel.permissions_for(bot_member) if not permissions.manage_channels: logging.warning( @@ -55,10 +66,16 @@ async def on_ready(self): @app_commands.command(name="membercount") async def member_count(self, interaction: discord.Interaction, force: bool = False): """Display the number of members""" + if interaction.guild is None or not isinstance( + interaction.user, discord.Member + ): + return + new_members = [ member for member in interaction.guild.members - if member.joined_at + if member.joined_at is not None + and member.joined_at > datetime.now(tz=ZoneInfo("Australia/Brisbane")) - self.NEW_MEMBER_TIME ] await interaction.response.send_message( @@ -107,5 +124,5 @@ async def _update_member_count_channel_name(self): self.waiting_for_rename = False -async def setup(bot: commands.Bot): +async def setup(bot: UQCSBot): await bot.add_cog(MemberCounter(bot)) diff --git a/uqcsbot/snailrace.py b/uqcsbot/snailrace.py index fee8b1fb..dd7e9383 100644 --- a/uqcsbot/snailrace.py +++ b/uqcsbot/snailrace.py @@ -1,10 +1,13 @@ -import discord, asyncio +import discord from discord import app_commands, ui from discord.ext import commands from uqcsbot.bot import UQCSBot import uqcsbot.utils.snailrace_utils as snail +from typing_extensions import TypeVar + +V = TypeVar("V", bound="SnailRaceView", covariant=True) # Trying out Discord buttons for Snail Race Interactions @@ -18,8 +21,12 @@ async def on_timeout(self): Called when the view times out. This will deactivate the buttons and begine the race. """ + if self.raceState.open_interaction is None: + return + for child in self.children: - child.disabled = True + if isinstance(child, discord.ui.Button): + child.disabled = True await self.raceState.open_interaction.edit_original_response( content=snail.SNAILRACE_ENTRY_CLOSE, view=self ) @@ -27,8 +34,11 @@ async def on_timeout(self): @ui.button(label="Enter Race", style=discord.ButtonStyle.primary) async def button_callback( - self, interaction: discord.Interaction, button: discord.ui.Button + self, interaction: discord.Interaction, button: discord.ui.Button[V] ): + if not isinstance(interaction.user, discord.Member): + return + action = self.raceState.add_racer(interaction.user) if action == snail.SnailRaceJoinAdded: diff --git a/uqcsbot/starboard.py b/uqcsbot/starboard.py index ed727d18..923114ce 100644 --- a/uqcsbot/starboard.py +++ b/uqcsbot/starboard.py @@ -1,6 +1,6 @@ import os, time from threading import Timer -from typing import Tuple, List +from typing import Tuple, List, Union from zoneinfo import ZoneInfo import discord @@ -9,6 +9,7 @@ from sqlalchemy.sql.expression import and_ from uqcsbot import models +from uqcsbot.bot import UQCSBot from uqcsbot.utils.err_log_utils import FatalErrorWithLog @@ -23,16 +24,20 @@ class Starboard(commands.Cog): MODLOG_CHANNEL_NAME = "admin-alerts" BRISBANE_TZ = ZoneInfo("Australia/Brisbane") - def __init__(self, bot: commands.Bot): + def __init__(self, bot: UQCSBot): self.bot = bot - self.base_threshold = int(os.environ.get("SB_BASE_THRESHOLD")) - self.big_threshold = int(os.environ.get("SB_BIG_THRESHOLD")) - self.ratelimit = int(os.environ.get("SB_RATELIMIT")) + + if (base := os.environ.get("SB_BASE_THRESHOLD")) is not None: + self.base_threshold = int(base) + if (big := os.environ.get("SB_BIG_THRESHOLD")) is not None: + self.big_threshold = int(big) + if (limit := os.environ.get("SB_RATELIMIT")) is not None: + self.ratelimit = limit # messages that are temp blocked from being resent to the starboard - self.base_blocked_messages = [] + self.base_blocked_messages: List[int] = [] # messages that are temp blocked from being repinned in the starboard - self.big_blocked_messages = [] + self.big_blocked_messages: List[int] = [] self.unblacklist_menu = app_commands.ContextMenu( name="Starboard Unblacklist", @@ -53,13 +58,22 @@ async def on_ready(self): N.B. this does assume the server only has one channel called "starboard" and one emoji called "starhaj". If this assumption stops holding, we may need to move back to IDs (cringe) """ - self.starboard_emoji = discord.utils.get(self.bot.emojis, name=self.EMOJI_NAME) - self.starboard_channel = discord.utils.get( - self.bot.get_all_channels(), name=self.SB_CHANNEL_NAME - ) - self.modlog = discord.utils.get( - self.bot.get_all_channels(), name=self.MODLOG_CHANNEL_NAME - ) + if ( + emoji := discord.utils.get(self.bot.emojis, name=self.EMOJI_NAME) + ) is not None: + self.starboard_emoji = emoji + if ( + channel := discord.utils.get( + self.bot.get_all_channels(), name=self.bot.STARBOARD_CNAME + ) + ) is not None and isinstance(channel, discord.TextChannel): + self.starboard_channel = channel + if ( + log := discord.utils.get( + self.bot.get_all_channels(), name=self.bot.ADMIN_ALERTS_CNAME + ) + ) is not None and isinstance(log, discord.TextChannel): + self.modlog = log @app_commands.command() @app_commands.default_permissions(manage_guild=True) @@ -92,17 +106,19 @@ async def cleanup_starboard(self, interaction: discord.Interaction): .filter(models.Starboard.sent == message.id) .one_or_none() ) - if query is None and message.author.id == self.bot.user.id: + if query is None and message.author.id == self.bot.safe_user.id: # only delete messages that uqcsbot itself sent await message.delete() - elif message.author.id == self.bot.user.id: + elif message.author.id == self.bot.safe_user.id: + recieved_msg: Union[discord.Message, None] = None + starboard_msg: Union[discord.Message, None] = None + try: recieved_msg, starboard_msg = await self._lookup_from_id( self.starboard_channel.id, message.id ) except BlacklistedMessageError: - if starboard_msg is not None: - await starboard_msg.delete() + pass new_reaction_count = await self._count_num_reacts( (recieved_msg, starboard_msg) @@ -118,6 +134,9 @@ async def _blacklist_log( self, message: discord.Message, user: discord.Member, blacklist: bool ): """Logs the use of a blacklist/unblacklist command to the modlog.""" + if not isinstance(message.author, discord.Member): + return + state = "blacklisted" if blacklist else "unblacklisted" embed = discord.Embed( @@ -318,7 +337,7 @@ async def _lookup_from_id( # This is Isaac's initial-starboard message. I know, IDs are bad. BUT # consider that this doesn't cause any of the usual ID-related issues # like breaking lookups in other servers. - return + return (None, None) raise FatalErrorWithLog( client=self.bot, @@ -598,5 +617,5 @@ async def on_raw_reaction_clear_emoji( await self._process_sb_updates(new_reaction_count, recieved_msg, starboard_msg) -async def setup(bot: commands.Bot): +async def setup(bot: UQCSBot): await bot.add_cog(Starboard(bot)) diff --git a/uqcsbot/uptime.py b/uqcsbot/uptime.py index ad38d7c5..a2a8f81e 100644 --- a/uqcsbot/uptime.py +++ b/uqcsbot/uptime.py @@ -11,24 +11,22 @@ class UpTime(commands.Cog): - CHANNEL_NAME = "bot-testing" - def __init__(self, bot: UQCSBot): self.bot = bot @commands.Cog.listener() async def on_ready(self): channel = discord.utils.get( - self.bot.uqcs_server.channels, name=self.CHANNEL_NAME + self.bot.uqcs_server.channels, name=self.bot.BOT_CNAME ) - if channel is not None: + if channel is not None and isinstance(channel, discord.TextChannel): if random.randint(1, 100) == 1: await channel.send("Oopsie, I webooted uwu >_<") else: await channel.send("I have rebooted!") else: - logging.warning(f"Could not find required channel #{self.CHANNEL_NAME}") + logging.warning(f"Could not find required channel #{self.bot.BOT_CNAME}") @app_commands.command() async def uptime(self, interaction: discord.Interaction): diff --git a/uqcsbot/utils/snailrace_utils.py b/uqcsbot/utils/snailrace_utils.py index 50276d2a..fb8df100 100644 --- a/uqcsbot/utils/snailrace_utils.py +++ b/uqcsbot/utils/snailrace_utils.py @@ -1,4 +1,5 @@ import discord, random, datetime, asyncio +from typing import Literal, List, Union # Racer Icon SNAILRACE_SNAIL_EMOJI = "🐌" @@ -35,10 +36,10 @@ SNAILRACE_WINNER = "The race has finished! %s has won!" SNAILRACE_WINNER_TIE = "The race has finished! It's a tie between %s!" -SnailRaceJoinType = 0 | 1 | 2 -SnailRaceJoinAdded = 0 -SnailRaceJoinAlreadyJoined = 1 -SnailRaceJoinRaceFull = 2 +SnailRaceJoinType = Literal[0, 1, 2] +SnailRaceJoinAdded: SnailRaceJoinType = 0 +SnailRaceJoinAlreadyJoined: SnailRaceJoinType = 1 +SnailRaceJoinRaceFull: SnailRaceJoinType = 2 class SnailRacer: @@ -85,7 +86,7 @@ def __init__(self): self.race_start_time = datetime.datetime.now() self.racing = False - self.racers = [] + self.racers: List[SnailRacer] = [] self.open_interaction = None def is_racing(self) -> bool: @@ -127,10 +128,14 @@ def add_racer(self, racer: discord.Member) -> SnailRaceJoinType: async def race_start(self): await self._start_racing(self.open_interaction) - async def _start_racing(self, interaction: discord.Interaction): + async def _start_racing(self, interaction: Union[discord.Interaction, None]): """ Start the race loop, this will be triggered after the entry has closed. """ + if interaction is None or not isinstance( + interaction.channel, discord.TextChannel + ): + return if not len(self.racers) > 0: await interaction.channel.send(SNAILRACE_NO_START) diff --git a/uqcsbot/utils/uq_course_utils.py b/uqcsbot/utils/uq_course_utils.py index b5f79927..8833e0dd 100644 --- a/uqcsbot/utils/uq_course_utils.py +++ b/uqcsbot/utils/uq_course_utils.py @@ -1,19 +1,21 @@ -import requests +import requests, logging from requests.exceptions import RequestException from datetime import datetime from dateutil import parser -from bs4 import BeautifulSoup +from bs4 import BeautifulSoup, Tag from functools import partial from binascii import hexlify -from typing import List, Dict, Optional +from typing import List, Dict, Optional, Tuple, Any -BASE_COURSE_URL = "https://my.uq.edu.au/programs-courses/course.html?course_code=" -BASE_ASSESSMENT_URL = ( +BASE_COURSE_URL: str = "https://my.uq.edu.au/programs-courses/course.html?course_code=" +BASE_ASSESSMENT_URL: str = ( "https://www.courses.uq.edu.au/" "student_section_report.php?report=assessment&profileIds=" ) -BASE_CALENDAR_URL = "http://www.uq.edu.au/events/calendar_view.php?category_id=16&year=" -OFFERING_PARAMETER = "offer" +BASE_CALENDAR_URL: str = ( + "http://www.uq.edu.au/events/calendar_view.php?category_id=16&year=" +) +OFFERING_PARAMETER: str = "offer" class DateSyntaxException(Exception): @@ -21,7 +23,7 @@ class DateSyntaxException(Exception): Raised when an unparsable date syntax is encountered. """ - def __init__(self, date, course_name): + def __init__(self, date: str, course_name: str): self.message = f"Could not parse date '{date}' for course '{course_name}'." self.date = date self.course_name = course_name @@ -33,7 +35,7 @@ class CourseNotFoundException(Exception): Raised when a given course cannot be found for UQ. """ - def __init__(self, course_name): + def __init__(self, course_name: str): self.message = f"Could not find course '{course_name}'." self.course_name = course_name super().__init__(self.message, self.course_name) @@ -44,7 +46,7 @@ class ProfileNotFoundException(Exception): Raised when a profile cannot be found for a given course. """ - def __init__(self, course_name): + def __init__(self, course_name: str): self.message = f"Could not find profile for course '{course_name}'." self.course_name = course_name super().__init__(self.message, self.course_name) @@ -56,14 +58,16 @@ class HttpException(Exception): unsuccessful (i.e. not 200 OK) status code. """ - def __init__(self, url, status_code): + def __init__(self, url: str, status_code: int): self.message = f"Received status code {status_code} from '{url}'." self.url = url self.status_code = status_code super().__init__(self.message, self.url, self.status_code) -def get_offering_code(semester=None, campus="STLUC", is_internal=True): +def get_offering_code( + semester: int = -1, campus: str = "STLUC", is_internal: bool = True +): """ Returns the hex encoded offering string for the given semester and campus. @@ -73,7 +77,8 @@ def get_offering_code(semester=None, campus="STLUC", is_internal=True): is_internal {bool} -- True for internal, false for external. """ # TODO: Codes for other campuses. - if semester is None: + # TODO: Summer Semesters? + if semester == -1: semester = 1 if datetime.today().month <= 6 else 2 location = "IN" if is_internal else "EX" return hexlify(f"{campus}{semester}{location}".encode("utf-8")).decode("utf-8") @@ -98,7 +103,7 @@ def get_uq_request( raise HttpException(message, 500) -def get_course_profile_url(course_name): +def get_course_profile_url(course_name: str) -> str: """ Returns the URL to the latest course profile for the given course. """ @@ -114,10 +119,13 @@ def get_course_profile_url(course_name): profile = html.find("a", class_="profile-available") if profile is None: raise ProfileNotFoundException(course_name) - return profile.get("href") + if type(profile) is Tag: + return str(profile.get("href")) + # TODO: Error Handling - Profile is NavigableString + return "" -def get_course_profile_id(course_name): +def get_course_profile_id(course_name: str): """ Returns the ID to the latest course profile for the given course. """ @@ -153,7 +161,9 @@ def get_current_exam_period(): return start_datetime, end_datetime -def get_parsed_assessment_due_date(assessment_item): +def get_parsed_assessment_due_date( + assessment_item: Tuple[str, Any, str, Any] +) -> Tuple[datetime, datetime]: """ Returns the parsed due date for the given assessment item as a datetime object. If the date cannot be parsed, a DateSyntaxException is raised. @@ -176,7 +186,9 @@ def get_parsed_assessment_due_date(assessment_item): raise DateSyntaxException(due_date, course_name) -def is_assessment_after_cutoff(assessment, cutoff): +def is_assessment_after_cutoff( + assessment: Tuple[str, str, str, str], cutoff: datetime +) -> bool: """ Returns whether the assessment occurs after the given cutoff. """ @@ -185,6 +197,7 @@ def is_assessment_after_cutoff(assessment, cutoff): except DateSyntaxException as e: # TODO bot.logger.error(e.message) # If we can't parse a date, we're better off keeping it just in case. + logging.error(e) # TODO(mitch): Keep track of these instances to attempt to accurately # parse them in future. Will require manual detection + parsing. return True @@ -196,11 +209,15 @@ def get_course_assessment_page(course_names: List[str]) -> str: Determines the course ids from the course names and returns the url to the assessment table for the provided courses """ - profile_ids = map(get_course_profile_id, course_names) + profile_ids: map[str] = map(get_course_profile_id, course_names) return BASE_ASSESSMENT_URL + ",".join(profile_ids) -def get_course_assessment(course_names, cutoff=None, assessment_url=None): +def get_course_assessment( + course_names: List[str], + cutoff: datetime | None = None, + assessment_url: str | None = None, +): """ Returns all the course assessment for the given courses that occur after the given cutoff. @@ -209,14 +226,20 @@ def get_course_assessment(course_names, cutoff=None, assessment_url=None): joined_assessment_url = get_course_assessment_page(course_names) else: joined_assessment_url = assessment_url - http_response = get_uq_request(joined_assessment_url) + http_response = get_uq_request(joined_assessment_url) if http_response.status_code != requests.codes.ok: raise HttpException(joined_assessment_url, http_response.status_code) html = BeautifulSoup(http_response.content, "html.parser") assessment_table = html.find("table", class_="tblborder") + if type(assessment_table) is not Tag: + # TODO: Error Handling here + return # Start from 1st index to skip over the row containing column names. assessment = assessment_table.findAll("tr")[1:] - parsed_assessment = map(get_parsed_assessment_item, assessment) + + parsed_assessment: map[Tuple[str, str, str, str]] = map( + get_parsed_assessment_item, assessment + ) # If no cutoff is specified, set cutoff to UNIX epoch (i.e. filter nothing). cutoff = cutoff or datetime.min assessment_filter = partial(is_assessment_after_cutoff, cutoff=cutoff) @@ -224,14 +247,14 @@ def get_course_assessment(course_names, cutoff=None, assessment_url=None): return list(filtered_assessment) -def get_element_inner_html(dom_element): +def get_element_inner_html(dom_element: BeautifulSoup) -> str: """ Returns the inner html for the given element. """ return dom_element.decode_contents(formatter="html") -def get_parsed_assessment_item(assessment_item): +def get_parsed_assessment_item(assessment_item: Tag) -> Tuple[str, str, str, str]: """ Returns the parsed assessment details for the given assessment item table row element. diff --git a/uqcsbot/working_on.py b/uqcsbot/working_on.py index e3af632c..fac09488 100644 --- a/uqcsbot/working_on.py +++ b/uqcsbot/working_on.py @@ -5,8 +5,7 @@ from discord.ext import commands from uqcsbot.bot import UQCSBot - -GENERAL_CHANNEL = "general" +from typing import List class WorkingOn(commands.Cog): @@ -19,7 +18,7 @@ def __init__(self, bot: UQCSBot): async def workingon(self): """5pm ping for 2 lucky server members to share what they have been working on.""" members = list(self.bot.get_all_members()) - message = [] + message: List[str] = [] while len(message) < 2: chosen = choice(members) @@ -30,10 +29,12 @@ async def workingon(self): ) general_channel = discord.utils.get( - self.bot.uqcs_server.channels, name=GENERAL_CHANNEL + self.bot.uqcs_server.channels, name=self.bot.GENERAL_CNAME ) - if general_channel is not None: + if general_channel is not None and isinstance( + general_channel, discord.TextChannel + ): await general_channel.send( "\n".join(message), allowed_mentions=discord.AllowedMentions( @@ -41,7 +42,9 @@ async def workingon(self): ), ) else: - logging.warning(f"Could not find required channel #{GENERAL_CHANNEL}") + logging.warning( + f"Could not find required channel #{self.bot.GENERAL_CNAME}" + ) async def setup(bot: UQCSBot):