Going with username-based Kickstarter linking. Ruff linter cleanup.

main.py

@@ -13,11 +13,11 @@ if sys.version_info < MIN_PYTHON_VERSION:
 	raise RuntimeError(f'rocketbot requires Python {MIN_PYTHON_VERSION[0]}.{MIN_PYTHON_VERSION[1]}.{MIN_PYTHON_VERSION[2]} '
 		f'or greater. Detected {ver[0]}.{ver[1]}.{ver[2]}.')
 
-import asyncio  # noqa: E402
+import asyncio
 
-from config import CONFIG  # noqa: E402
-from rocketbot.bot import start_bot  # noqa: E402
-from rocketbot.utils import bot_log  # noqa: E402
+from config import CONFIG
+from rocketbot.bot import start_bot
+from rocketbot.utils import bot_log
 
 CURRENT_CONFIG_VERSION = 5
 if (CONFIG.get('__config_version') or 0) < CURRENT_CONFIG_VERSION:

rocketbot/bot.py

@@ -1,7 +1,7 @@
 import traceback
-from typing import Optional
 
 from discord import Intents
+from discord.errors import DiscordException
 from discord.ext import commands
 
 from config import CONFIG
@@ -69,13 +69,13 @@ class Rocketbot(commands.Bot):
 					CogSetting.set_up_all(bcog, self, bcog.settings)
 		try:
 			synced_commands = await self.tree.sync()
-			for command in synced_commands:
+			for command in sorted(synced_commands, key=lambda c: c.name):
 				bot_log(None, None, f'Synced command: /{command.name}')
-		except Exception as e:
+		except DiscordException as e:
 			dump_stacktrace(e)
 
 # Current active bot instance
-rocketbot: Optional[Rocketbot] = None
+rocketbot: Rocketbot | None = None
 
 def __create_bot():
 	global rocketbot

rocketbot/botmessage.py

@@ -3,7 +3,7 @@ Classes for crafting messages from the bot. Content can change as information
 changes, and mods can perform actions on the message via emoji reactions.
 """
 from datetime import datetime
-from typing import Any, Optional, Union
+from typing import Any
 
 from discord import Guild, Message, PartialEmoji, TextChannel
 
@@ -42,11 +42,11 @@ class BotMessageReaction:
 
 	@classmethod
 	def standard_set(cls,
-			did_delete: bool = None,
+			did_delete: bool | None = None,
 			message_count: int = 1,
-			did_kick: bool = None,
-			did_ban: bool = None,
-			user_count: int = 1) -> list: # list[BotMessageReaction]:
+			did_kick: bool | None = None,
+			did_ban: bool | None = None,
+			user_count: int = 1) -> list['BotMessageReaction']:
 		"""
 		Convenience factory for generating any of the three most common
 		commands: delete message(s), kick user(s), and ban user(s). All
@@ -125,8 +125,8 @@ class BotMessage:
 			guild: Guild,
 			text: str,
 			type: int = TYPE_DEFAULT, # pylint: disable=redefined-builtin
-			context: Optional[Any] = None,
-			reply_to: Optional[Message] = None,
+			context: Any | None = None,
+			reply_to: Message | None = None,
 			suppress_embeds: bool = False):
 		"""
 		Creates a bot message.
@@ -140,13 +140,13 @@ class BotMessage:
 		self.guild: Guild = guild
 		self.text: str = text
 		self.type: int = type
-		self.context: Optional[Any] = context
-		self.quote: Optional[str] = None
+		self.context: Any | None = context
+		self.quote: str | None = None
 		self.source_cog = None  # Set by `BaseCog.post_message()`
 		self.__posted_text: list[str] = []  # last text posted, to test for changes
 		self.__posted_emoji: set[str] = set()  # last emoji list posted
 		self.__messages: list[Message] = []  # set once the message has been posted
-		self.__reply_to: Optional[Message] = reply_to
+		self.__reply_to: Message | None = reply_to
 		self.__suppress_embeds = suppress_embeds
 		self.__reactions: list[BotMessageReaction] = []
 
@@ -163,7 +163,7 @@ class BotMessage:
 		broken into multiple Discord messages."""
 		return [ m.id for m in self.__messages ]
 
-	def message_sent_at(self) -> Optional[datetime]:
+	def message_sent_at(self) -> datetime | None:
 		"""Returns when the message was sent or None if not sent."""
 		return norm_datetime(self.__messages[0].created_at) if len(self.__messages) > 0 else None
 
@@ -217,7 +217,7 @@ class BotMessage:
 			self.__reactions.append(reaction)
 		await self.update_if_sent()
 
-	async def remove_reaction(self, reaction_or_emoji: Union[BotMessageReaction, str]) -> None:
+	async def remove_reaction(self, reaction_or_emoji: BotMessageReaction | str) -> None:
 		"""
 		Removes a reaction. Can pass either a BotMessageReaction or just the
 		emoji string. If the message has been sent, it will be updated.
@@ -230,7 +230,7 @@ class BotMessage:
 				await self.update_if_sent()
 				return
 
-	def reaction_for_emoji(self, emoji) -> Optional[BotMessageReaction]:
+	def reaction_for_emoji(self, emoji) -> BotMessageReaction | None:
 		"""
 		Finds the BotMessageReaction for the given emoji or None if not found.
 		Accepts either a PartialEmoji or str.
@@ -272,7 +272,7 @@ class BotMessage:
 					self.__messages.append(message)
 				self.__posted_text = message_bodies
 		else: # No messages posted yet
-			channel: Optional[TextChannel] = None
+			channel: TextChannel | None = None
 			for index, body in enumerate(message_bodies):
 				if index == 0 and self.__reply_to:
 					message = await self.__reply_to.reply(content=body, mention_author=False)

rocketbot/cogs/bangcommandcog.py

@@ -11,6 +11,7 @@ from discord import (
 	TextStyle,
 )
 from discord.app_commands import Choice, Group, autocomplete
+from discord.errors import DiscordException
 from discord.ext.commands import Cog
 from discord.ui import Label, Modal, Select, TextInput
 
@@ -67,7 +68,7 @@ class BangCommandCog(BaseCog, name='Bang Commands'):
 	def get_saved_commands(self, guild: Guild) -> dict[str, BangCommand]:
 		return self.get_guild_setting(guild, BangCommandCog.SETTING_COMMANDS)
 
-	def get_saved_command(self, guild: Guild, name: str) -> Optional[BangCommand]:
+	def get_saved_command(self, guild: Guild, name: str) -> BangCommand | None:
 		cmds = self.get_saved_commands(guild)
 		name = BangCommandCog._normalize_name(name)
 		return cmds.get(name, None)
@@ -91,7 +92,7 @@ class BangCommandCog(BaseCog, name='Bang Commands'):
 		}
 	)
 	@autocomplete(name=command_autocomplete)
-	async def define_command(self, interaction: Interaction, name: str, definition: Optional[str] = None, mod_only: bool = False) -> None:
+	async def define_command(self, interaction: Interaction, name: str, definition: str | None = None, mod_only: bool = False) -> None:
 		"""
 		Defines or redefines a bang command.
 
@@ -223,9 +224,8 @@ class BangCommandCog(BaseCog, name='Bang Commands'):
 		if len(content) < 1 or len(content) > 2000:
 			raise ValueError(f'Content must be between 1 and {_MAX_CONTENT_LENGTH} characters.')
 		cmds = self.get_saved_commands(guild)
-		if check_exists:
-			if cmds.get(name, None) is not None:
-				raise ValueError(f'Command with name "{name}" already exists.')
+		if check_exists and cmds.get(name, None) is not None:
+			raise ValueError(f'Command with name "{name}" already exists.')
 		cmds[name] = {
 			'content': content,
 			'mod_only': mod_only,
@@ -262,17 +262,14 @@ class BangCommandCog(BaseCog, name='Bang Commands'):
 
 	@staticmethod
 	def _normalize_name(name: str) -> str:
-		name = name.lower().strip()
-		if name.startswith('!'):
-			name = name[1:]
-		return name
+		return name.lower().strip().removeprefix('!')
 
 	@staticmethod
-	def _is_valid_name(name: Optional[str]) -> bool:
+	def _is_valid_name(name: str | None) -> bool:
 		return name is not None and re.match(r'^!?([a-z]+)([_-][a-z]+)*$', name) is not None
 
 	@staticmethod
-	def _name_from_command_message(name: Optional[str]) -> Optional[str]:
+	def _name_from_command_message(name: str | None) -> str | None:
 		if name is None:
 			return None
 		match = re.match(r'^!((?:[a-z]+)(?:[_-][a-z]+)*)\b.*$', name)
@@ -316,7 +313,11 @@ class _EditModal(Modal, title='Edit Command'):
 		)
 	)
 
-	def __init__(self, name: Optional[str] = None, content: Optional[str] = None, mod_only: Optional[bool] = None, exists: bool = False):
+	def __init__(self,
+			name: str | None = None,
+			content: str | None = None,
+			mod_only: bool | None = None,
+			exists: bool = False):
 		super().__init__()
 		self.exists = exists
 		# noinspection PyTypeChecker
@@ -360,5 +361,5 @@ class _EditModal(Modal, title='Edit Command'):
 				f'{CONFIG["failure_emoji"]} Save failed',
 				ephemeral=True,
 			)
-		except BaseException:
-			pass
+		except DiscordException as e:
+			dump_stacktrace(e)

rocketbot/cogs/basecog.py

@@ -2,7 +2,6 @@
 Base cog class and helper classes.
 """
 from datetime import datetime, timedelta, timezone
-from typing import Optional
 
 import discord
 from discord import (
@@ -17,6 +16,7 @@ from discord import (
 from discord.abc import GuildChannel
 from discord.app_commands import AppCommandError
 from discord.app_commands.errors import CommandInvokeError
+from discord.errors import DiscordException
 from discord.ext.commands import Cog
 
 from config import CONFIG
@@ -43,9 +43,9 @@ class BaseCog(Cog):
 	def __init__(
 			self,
 			bot: Rocketbot,
-			config_prefix: Optional[str],
+			config_prefix: str | None,
 			short_description: str,
-			long_description: Optional[str] = None,
+			long_description: str | None = None,
 	):
 		"""
 		Parameters
@@ -60,7 +60,7 @@ class BaseCog(Cog):
 		self.bot: Rocketbot = bot
 		self.are_settings_setup: bool = False
 		self.settings: list[CogSetting] = []
-		self.config_prefix: Optional[str] = config_prefix
+		self.config_prefix: str | None = config_prefix
 		self.short_description: str = short_description
 		self.long_description: str = long_description
 
@@ -78,8 +78,7 @@ class BaseCog(Cog):
 		except discord.InteractionResponded:
 			try:
 				await interaction.followup.send(f"An error occurred: {message}", ephemeral=True)
-			except BaseException:
-				interaction.channel.send
+			except DiscordException:
 				bot_log(interaction.guild, None, message)
 
 	@property
@@ -124,7 +123,7 @@ class BaseCog(Cog):
 
 	@classmethod
 	def get_guild_setting(cls,
-			guild: Optional[Guild],
+			guild: Guild | None,
 			setting: CogSetting,
 			use_cog_default_if_not_set: bool = True):
 		"""
@@ -303,7 +302,7 @@ class BaseCog(Cog):
 	# Helpers
 
 	@classmethod
-	def log(cls, guild: Optional[Guild], message) -> None:
+	def log(cls, guild: Guild | None, message) -> None:
 		"""
 		Writes a message to the console. Intended for significant events only.
 		"""

rocketbot/cogs/configcog.py

@@ -1,7 +1,6 @@
 """
 Cog handling general configuration for a guild.
 """
-from typing import Optional, Union
 
 from discord import Guild, Interaction, Role, TextChannel, User
 from discord.app_commands import Group
@@ -85,7 +84,7 @@ class ConfigCog(BaseCog, name='Configuration'):
 	)
 	async def set_warning_mention(self,
 			interaction: Interaction,
-			mention: Optional[Union[User, Role]] = None) -> None:
+			mention: User | Role | None = None) -> None:
 		"""
 		Sets a user/role to mention in warning messages.
 

rocketbot/cogs/crosspostcog.py

@@ -3,7 +3,6 @@ Cog for detecting spam messages posted in multiple channels.
 """
 import re
 from datetime import datetime, timedelta, timezone
-from typing import Optional
 
 from discord import Member, Message, TextChannel
 from discord import utils as discordutils
@@ -23,7 +22,7 @@ class SpamContext:
 	def __init__(self, member: Member) -> None:
 		self.member: Member = member
 		self.age: datetime = datetime.now(timezone.utc)
-		self.bot_message: Optional[BotMessage] = None
+		self.bot_message: BotMessage | None = None
 		self.is_kicked: bool = False
 		self.is_banned: bool = False
 		self.is_autobanned: bool = False
@@ -252,7 +251,7 @@ class CrossPostCog(BaseCog, name='Crosspost Detection'):
 		await self.__update_message_from_context(context)
 
 	async def __update_message_from_context(self, context: SpamContext) -> None:
-		first_spam_message: Message = sorted(list(context.spam_messages), key=lambda m: m.created_at)[0]
+		first_spam_message: Message = min(context.spam_messages, key=lambda m: m.created_at)
 		spam_count = len(context.spam_messages)
 		channel_count = len(context.unique_channels)
 		deleted_count = len(context.deleted_messages)
@@ -284,7 +283,7 @@ class CrossPostCog(BaseCog, name='Crosspost Detection'):
 				body += f'messages in {channel_count} channels within {max_age_str} ' + \
 						f'({duplicate_count} are identical, showing first one).'
 			max_links = 10
-			for msg in sorted(list(context.spam_messages), key=lambda m: m.created_at)[:max_links]:
+			for msg in sorted(context.spam_messages, key=lambda m: m.created_at)[:max_links]:
 				body += f'\n- {msg.jump_url}'
 			if len(context.spam_messages) > max_links:
 				body += f'\n- ...{len(context.spam_messages) - max_links} more...'

rocketbot/cogs/generalcog.py

@@ -32,7 +32,7 @@ class GeneralCog(BaseCog, name='General'):
 		)
 		self.is_connected = False
 		self.is_first_connect = True
-		self.last_disconnect_time: Optional[datetime] = None
+		self.last_disconnect_time: datetime | None = None
 		self.noteworthy_disconnect_duration = timedelta(seconds=5)
 		GeneralCog.shared = self
 

rocketbot/cogs/helpcog.py

@@ -1,7 +1,7 @@
 """Provides help commands for getting info on using other commands and configuration."""
 import re
 import time
-from typing import Optional, TypedDict, Union
+from typing import Optional, TypedDict
 
 from discord import AppCommandType, Interaction, Permissions
 from discord.app_commands import (
@@ -12,6 +12,7 @@ from discord.app_commands import (
 	command,
 	guild_only,
 )
+from discord.errors import DiscordException
 
 from config import CONFIG
 from rocketbot.bot import Rocketbot
@@ -19,7 +20,7 @@ from rocketbot.cogs.basecog import BaseCog
 from rocketbot.ui.pagedcontent import paginate, update_paged_content
 from rocketbot.utils import MOD_PERMISSIONS, dump_stacktrace
 
-HelpTopic = Union[Command, Group, BaseCog]
+HelpTopic = Command | Group | BaseCog
 class HelpMeta(TypedDict):
 	id: str
 	text: str
@@ -52,7 +53,7 @@ async def search_autocomplete(interaction: Interaction, current: str) -> list[Ch
 			choice_from_topic(topic, include_full_command=True)
 			for topic in HelpCog.shared.topics_for_keywords(current, interaction.permissions)
 		][:25]
-	except BaseException as e:
+	except DiscordException as e:
 		dump_stacktrace(e)
 		return []
 
@@ -120,24 +121,24 @@ class HelpCog(BaseCog, name='Help'):
 				text += f' {cog.long_description}'
 			self.topics.append({ 'id': key, 'text': process_text(text), 'topic': cog })
 
-	def topic_for_help_symbol(self, symbol: str) -> Optional[HelpTopic]:
+	def topic_for_help_symbol(self, symbol: str) -> HelpTopic | None:
 		self.__create_help_index()
 		return self.id_to_topic.get(symbol, None)
 
-	def all_commands(self) -> list[Union[Command, Group]]:
+	def all_commands(self) -> list[Command | Group]:
 		# PyCharm not interpreting conditional return type correctly.
 		# noinspection PyTypeChecker
-		cmds: list[Union[Command, Group]] = self.bot.tree.get_commands(type=AppCommandType.chat_input)
+		cmds: list[Command | Group] = self.bot.tree.get_commands(type=AppCommandType.chat_input)
 		return sorted(cmds, key=lambda cmd: cmd.name)
 
-	def all_accessible_commands(self, permissions: Optional[Permissions]) -> list[Union[Command, Group]]:
+	def all_accessible_commands(self, permissions: Permissions | None) -> list[Command | Group]:
 		return [
 			cmd
 			for cmd in self.all_commands()
 			if can_use_command(cmd, permissions)
 		]
 
-	def all_accessible_subcommands(self, permissions: Optional[Permissions]) -> list[Command]:
+	def all_accessible_subcommands(self, permissions: Permissions | None) -> list[Command]:
 		cmds = self.all_accessible_commands(permissions)
 		subcmds: list[Command] = []
 		for cmd in cmds:
@@ -147,14 +148,14 @@ class HelpCog(BaseCog, name='Help'):
 						subcmds.append(subcmd)
 		return subcmds
 
-	def all_accessible_cogs(self, permissions: Optional[Permissions]) -> list[BaseCog]:
+	def all_accessible_cogs(self, permissions: Permissions | None) -> list[BaseCog]:
 		return [
 			cog
 			for cog in self.basecogs
 			if can_use_cog(cog, permissions)
 		]
 
-	def all_accessible_topics(self, permissions: Optional[Permissions], *,
+	def all_accessible_topics(self, permissions: Permissions | None, *,
 							  include_cogs: bool = True,
 							  include_commands: bool = True,
 							  include_subcommands: bool = True) -> list[HelpTopic]:
@@ -167,7 +168,7 @@ class HelpCog(BaseCog, name='Help'):
 			topics += self.all_accessible_subcommands(permissions)
 		return topics
 
-	def topics_for_keywords(self, search: str, permissions: Optional[Permissions]) -> list[HelpTopic]:
+	def topics_for_keywords(self, search: str, permissions: Permissions | None) -> list[HelpTopic]:
 		start_time = time.perf_counter()
 		self.__create_help_index()
 
@@ -189,7 +190,7 @@ class HelpCog(BaseCog, name='Help'):
 		accessible_topics = [
 			topic
 			for topic in matching_topics
-			if ((isinstance(topic, Command) or isinstance(topic, Group)) and can_use_command(topic, permissions)) or \
+			if ((isinstance(topic, (Command, Group))) and can_use_command(topic, permissions)) or \
 			   (isinstance(topic, BaseCog) and can_use_cog(topic, permissions))
 		]
 
@@ -221,7 +222,7 @@ class HelpCog(BaseCog, name='Help'):
 	)
 	@guild_only()
 	@autocomplete(search=search_autocomplete)
-	async def help_command(self, interaction: Interaction, search: Optional[str]) -> None:
+	async def help_command(self, interaction: Interaction, search: str | None) -> None:
 		"""
 		Shows help for using commands and subcommands and configuring modules.
 
@@ -259,10 +260,10 @@ class HelpCog(BaseCog, name='Help'):
 			delete_after=10,
 		)
 
-	def get_command_list(self, permissions: Optional[Permissions] = None) -> dict[str, Union[Command, Group]]:
+	def get_command_list(self, permissions: Permissions | None = None) -> dict[str, Command | Group]:
 		return { cmd.name: cmd for cmd in self.bot.tree.get_commands() if can_use_command(cmd, permissions) }
 
-	def get_subcommand_list(self, cmd: Group, permissions: Optional[Permissions] = None) -> dict[str, Command]:
+	def get_subcommand_list(self, cmd: Group, permissions: Permissions | None = None) -> dict[str, Command]:
 		return {
 			subcmd.name: subcmd
 			for subcmd in cmd.commands
@@ -308,11 +309,11 @@ class HelpCog(BaseCog, name='Help'):
 
 		await self.__send_paged_help(interaction, text)
 
-	async def __send_keyword_help(self, interaction: Interaction, matching_topics: Optional[list[HelpTopic]]) -> None:
+	async def __send_keyword_help(self, interaction: Interaction, matching_topics: list[HelpTopic] | None) -> None:
 		matching_commands = [
 			cmd
 			for cmd in matching_topics or []
-			if isinstance(cmd, Command) or isinstance(cmd, Group)
+			if isinstance(cmd, (Command, Group))
 		]
 		matching_cogs = [
 			cog
@@ -346,7 +347,7 @@ class HelpCog(BaseCog, name='Help'):
 
 		await self.__send_paged_help(interaction, text)
 
-	async def __send_command_help(self, interaction: Interaction, command_or_group: Union[Command, Group], addendum: Optional[str] = None) -> None:
+	async def __send_command_help(self, interaction: Interaction, command_or_group: Command | Group, addendum: str | None = None) -> None:
 		text = ''
 		if addendum is not None:
 			text += addendum + '\n\n'
@@ -437,13 +438,13 @@ trivial_words = {
 	'there', 'them', 'they', "they're", 'this', 'to', 'when', 'with',
 }
 
-def can_use_command(cmd: Union[Group, Command], user_permissions: Optional[Permissions]) -> bool:
+def can_use_command(cmd: Group | Command, user_permissions: Permissions | None) -> bool:
 	if user_permissions is None:
 		return False
 	if cmd.parent and not can_use_command(cmd.parent, user_permissions):
 		return False
 	return cmd.default_permissions is None or cmd.default_permissions.is_subset(user_permissions)
 
-def can_use_cog(cog: BaseCog, user_permissions: Optional[Permissions]) -> bool:
+def can_use_cog(cog: BaseCog, user_permissions: Permissions | None) -> bool:
 	# "Using" a cog for now means configuring it, and only mods can configure cogs.
 	return user_permissions is not None and MOD_PERMISSIONS.is_subset(user_permissions)

rocketbot/cogs/logcog.py

@@ -3,9 +3,9 @@ Cog for detecting large numbers of guild joins in a short period of time.
 """
 import difflib
 import re
-from collections.abc import Sequence
+from collections.abc import Callable, Sequence
 from datetime import datetime, timedelta, timezone
-from typing import Any, Callable, Optional, Union
+from typing import Any
 
 from discord import (
 	AuditLogAction,
@@ -24,6 +24,7 @@ from discord import (
 	User,
 )
 from discord.abc import GuildChannel
+from discord.errors import DiscordException
 from discord.ext import tasks
 from discord.ext.commands import Cog
 from discord.utils import escape_markdown
@@ -34,7 +35,7 @@ from rocketbot.utils import dump_stacktrace, norm_datetime
 
 
 class BufferedMessageEditEvent:
-	def __init__(self, guild: Guild, channel: GuildChannel, before: Optional[Message], after: Message, data = None) -> None:
+	def __init__(self, guild: Guild, channel: GuildChannel, before: Message | None, after: Message, data = None) -> None:
 		self.guild = guild
 		self.channel = channel
 		self.before = before
@@ -42,7 +43,7 @@ class BufferedMessageEditEvent:
 		self.data = data
 
 class BufferedMessageDeleteEvent:
-	def __init__(self, guild: Guild, channel: GuildChannel, message_id: int, message: Optional[Message] = None) -> None:
+	def __init__(self, guild: Guild, channel: GuildChannel, message_id: int, message: Message | None = None) -> None:
 		self.guild = guild
 		self.channel = channel
 		self.message_id = message_id
@@ -401,7 +402,7 @@ class LoggingCog(BaseCog, name='Logging'):
 		await bot_message.update()
 
 	@Cog.listener()
-	async def on_member_ban(self, guild: Guild, user: Union[User, Member]) -> None:
+	async def on_member_ban(self, guild: Guild, user: User | Member) -> None:
 		"""
 		Called when user gets banned from a Guild.
 
@@ -424,7 +425,7 @@ class LoggingCog(BaseCog, name='Logging'):
 		bot_message = BotMessage(guild, text, BotMessage.TYPE_LOG)
 		await bot_message.update()
 
-	async def __find_audit_entry(self, user: Union[User, Member], action: AuditLogAction, max_age: int = 10) -> Optional[AuditLogEntry]:
+	async def __find_audit_entry(self, user: User | Member, action: AuditLogAction, max_age: int = 10) -> AuditLogEntry | None:
 		"""
 		Searches the audit log for the most recent entry of a given type for a
 		given user. Intended for finding the relevant entry for a ban/kick that
@@ -442,7 +443,7 @@ class LoggingCog(BaseCog, name='Logging'):
 		return None
 
 	@Cog.listener()
-	async def on_member_unban(self, guild: Guild, user: Union[User, Member]) -> None:
+	async def on_member_unban(self, guild: Guild, user: User | Member) -> None:
 		"""
 		Called when a User gets unbanned from a Guild.
 
@@ -476,7 +477,7 @@ class LoggingCog(BaseCog, name='Logging'):
 			self.buffered_guilds.clear()
 			for guild in guilds:
 				await self.__flush_buffers_for_guild(guild)
-		except Exception as e:
+		except DiscordException as e:
 			dump_stacktrace(e)
 
 	async def __flush_buffers_for_guild(self, guild: Guild) -> None:
@@ -739,7 +740,7 @@ class LoggingCog(BaseCog, name='Logging'):
 		else:
 			complex_deletes = events
 		if len(complex_deletes) > 0:
-			messages_per_author: dict[Optional[User], list[BufferedMessageDeleteEvent]] = self.__groupby(complex_deletes, lambda e: e.author)
+			messages_per_author: dict[User | None, list[BufferedMessageDeleteEvent]] = self.__groupby(complex_deletes, lambda e: e.author)
 			text = 'Multiple messages deleted' if len(complex_deletes) > 1 else 'Message deleted'
 			row_count = 0
 			for author, messages in messages_per_author.items():
@@ -873,7 +874,7 @@ class LoggingCog(BaseCog, name='Logging'):
 			return '> _<no content>_'
 		return '> ' + escape_markdown(s).replace('\n', '\n> ')
 
-	def __describe_user(self, user: Union[User, Member]) -> str:
+	def __describe_user(self, user: User | Member) -> str:
 		"""
 		Standardized Markdown describing a user or member.
 		"""

rocketbot/cogs/patterncog.py

@@ -9,6 +9,7 @@ from typing import Optional
 from discord import Guild, Intents, Interaction, Member, Message
 from discord import utils as discordutils
 from discord.app_commands import Choice, Group, autocomplete
+from discord.errors import DiscordException
 from discord.ext.commands import Cog
 
 from config import CONFIG
@@ -47,7 +48,7 @@ async def pattern_name_autocomplete(interaction: Interaction, current: str) -> l
 		for name in sorted(patterns.keys()):
 			if len(current_normal) == 0 or current_normal.startswith(name.lower()):
 				choices.append(Choice(name=name, value=name))
-	except BaseException as e:
+	except DiscordException as e:
 		dump_stacktrace(e)
 	return choices
 
@@ -55,7 +56,7 @@ async def action_autocomplete(interaction: Interaction, current: str) -> list[Ch
 	# FIXME: WORK IN PROGRESS
 	print(f'autocomplete action - current = "{current}"')
 	regex = re.compile('^(.*?)([a-zA-Z]+)$')
-	match: Optional[re.Match[str]] = regex.match(current)
+	match: re.Match[str] | None = regex.match(current)
 	initial: str = ''
 	stub: str = current
 	if match:
@@ -163,11 +164,11 @@ class PatternCog(BaseCog, name='Pattern Matching'):
 	def __save_patterns(cls,
 			guild: Guild,
 			patterns: dict[str, PatternStatement]) -> None:
-		to_save: list[dict] = list(map(lambda ps: ps.to_json(), patterns.values()))
+		to_save: list[dict] = [ps.to_json() for ps in patterns.values()]
 		cls.set_guild_setting(guild, cls.SETTING_PATTERNS, to_save)
 
 	@classmethod
-	def __get_last_matched(cls, guild: Guild, name: str) -> Optional[datetime]:
+	def __get_last_matched(cls, guild: Guild, name: str) -> datetime | None:
 		last_matched: dict[str, datetime] = Storage.get_state_value(guild, 'PatternCog.last_matched')
 		if last_matched:
 			return last_matched.get(name)

rocketbot/cogs/urlspamcog.py

@@ -270,15 +270,11 @@ class URLSpamCog(BaseCog, name='URL Spam'):
 					link = 'https://' + link[12:]
 				if link.startswith('http://www.'):
 					link = 'http://' + link[11:]
-				if link.endswith('/'):
-					link = link[:-1]
-				if label.startswith('www.'):
-					label = label[4:]
-				if label.endswith('/'):
-					label = label[:-1]
-				if link.startswith('https://') and 'https://' + label != link:
-					return True
-				elif link.startswith('http://') and 'http://' + label != link:
+				link = link.removesuffix('/')
+				label = label.removeprefix('www.')
+				label = label.removesuffix('/')
+				if (link.startswith('https://') and 'https://' + label != link) or \
+					(link.startswith('http://') and 'http://' + label != link):
 					return True
 		return False
 

rocketbot/cogs/usernamecog.py

@@ -1,7 +1,6 @@
 """
 Cog for detecting username patterns.
 """
-from typing import Optional
 
 from discord import Guild, Intents, Interaction, Member
 from discord.app_commands import Group
@@ -19,9 +18,9 @@ class UsernamePatternContext:
 	"""
 	def __init__(self, member: Member) -> None:
 		self.member: Member = member
-		self.kicked_by: Optional[Member] = None
-		self.banned_by: Optional[Member] = None
-		self.ignored_by: Optional[Member] = None
+		self.kicked_by: Member | None = None
+		self.banned_by: Member | None = None
+		self.ignored_by: Member | None = None
 
 	def reactions(self) -> list[BotMessageReaction]:
 		"""
@@ -187,9 +186,7 @@ class UsernamePatternCog(BaseCog, name='Username Pattern'):
 	async def on_member_join(self, member: Member) -> None:
 		"""Event handler"""
 		for pattern in self.__get_patterns(member.guild):
-			if self.matches(pattern, member.name):
-				await self.handle_match(member, pattern)
-			elif self.matches(pattern, member.display_name):
+			if self.matches(pattern, member.name) or self.matches(pattern, member.display_name):
 				await self.handle_match(member, pattern)
 
 	def matches(self, pattern: str, subject: str) -> bool:
@@ -209,7 +206,7 @@ class UsernamePatternCog(BaseCog, name='Username Pattern'):
 		context = UsernamePatternContext(member)
 		bm = BotMessage(
 			member.guild,
-			f'User {member.mention} ({str(member.id)}, {member.display_name}) has ' +
+			f'User {member.mention} ({member.id}, {member.display_name}) has ' +
 			f'username matching pattern `{pattern}`.',
 			BotMessage.TYPE_INFO if self.was_warned_recently(member) else BotMessage.TYPE_MOD_WARNING,
 			context)

rocketbot/cogs/videopreviewcog.py

@@ -1,10 +1,10 @@
 import asyncio
 import json
 import re
-import subprocess
 from datetime import timedelta
+from typing import Any
 
-from discord import Intents, Message
+from discord import Guild, Intents, Message
 from discord.ext.commands import Cog
 
 from rocketbot.cogs.basecog import BaseCog
@@ -19,15 +19,11 @@ from rocketbot.utils import (
 def filter_video_format(format: dict) -> bool:
 	if format.get('resolution') == 'audio only':
 		return False
-	if format.get('format_note') == 'DASH audio':
-		return False
-	return True
+	return format.get('format_note') != 'DASH audio'
 
 def rank_video_format(format: dict) -> tuple:
 	content = 0
-	if format.get('resolution') == 'audio only':
-		content = 1
-	elif format.get('format_note') == 'DASH audio':
+	if format.get('resolution') == 'audio only' or format.get('format_note') == 'DASH audio':
 		content = 1
 	elif format.get('format_note') == 'DASH video':
 		content = 2
@@ -87,7 +83,7 @@ class VideoPreviewCog(BaseCog, name='Video Link Previews'):
 	REGEX_FACEBOOK_POST = r'(https?:\/\/(?:www\.)?)\w*(facebook\.com(?:\/\w+)+\/\w+\/?)'
 	REGEX_TWITTER_POST = r'(https?:\/\/(?:www\.)?)\w*((?:twitter|x)\.com\/\w+\/status\/[0-9]+)'
 
-	REGEX_SPOILERS = '\|\|.+\|\|'
+	REGEX_SPOILERS = r'\|\|.+\|\|'
 
 	# Best video and best audio, mp4 format with m4a audio
 	FORMATS = 'bv*[ext=mp4]+ba[ext=m4a]/' \
@@ -154,11 +150,11 @@ class VideoPreviewCog(BaseCog, name='Video Link Previews'):
 		delay: timedelta = self.get_guild_setting(message.guild, Self.SETTING_DELAY)
 		await asyncio.sleep(delay.total_seconds())
 		# Look for embeds already showing the video
-		# self.log(message.guild, "Checking message for embeds")
+		self.__trace(message.guild, "Checking message for embeds")
 		for embed in message.embeds:
 			if embed.video.url:
 				# If there's any video, skip downloading any previews
-				# self.log(message.guild, "Message already has a video. Skipping this message.")
+				self.__trace(message.guild, "Message already has a video. Skipping this message.")
 				return
 		await self._fetch_previews(message, links)
 
@@ -169,37 +165,36 @@ class VideoPreviewCog(BaseCog, name='Video Link Previews'):
 		await asyncio.gather(*promises)
 
 	async def _fetch_preview(self, message: Message, link: MessageLink):
-		result = subprocess.run(
-			[
-				'yt-dlp',
-				'--skip-download',
-				'--dump-single-json',
-				link.url,
-			],
-			stdout=subprocess.PIPE,
-			stderr=subprocess.PIPE,
-			universal_newlines=True
+		process = await asyncio.create_subprocess_exec(
+			'yt-dlp',
+			'--skip-download',
+			'--dump-single-json',
+			link.url,
+			stdout=asyncio.subprocess.PIPE,
+			stderr=asyncio.subprocess.PIPE
 		)
-		if result.returncode != 0:
-			# self.log(message.guild, "Fetching link info JSON failed. Skipping preview.")
-			# self.log(message.guild, result.stderr)
+		stdout, stderr = await process.communicate()
+		await process.wait()
+		if process.returncode != 0:
+			self.__trace(message.guild, "Fetching link info JSON failed. Skipping preview.")
+			self.__trace(message.guild, stderr.decode())
 			return
 		try:
-			info: dict = json.loads(result.stdout)
-		except Exception as e:
-			# self.log(message.guild, f"Error parsing info.json. Skipping preview. {e}")
+			info: dict = json.loads(stdout)
+		except json.JSONDecodeError as e:
+			self.__trace(message.guild, f"Error parsing info.json. Skipping preview. {e}")
 			return
 		description = info.get('description') or ''
 		formats: list[dict] = info.get('formats') or []
-		# self.log(message.guild, f"Found {len(formats)} formats")
+		self.__trace(message.guild, f"Found {len(formats)} formats")
 		formats = list(filter(filter_video_format, formats))
-		# self.log(message.guild, f"Filtered to {len(formats)} formats")
+		self.__trace(message.guild, f"Filtered to {len(formats)} formats")
 		sorted_formats: list[dict] = sorted(formats, key=rank_video_format, reverse=True)
 		if len(sorted_formats) == 0:
-			# self.log(message.guild, f"No eligible formats for URL {link.url}")
+			self.__trace(message.guild, f"No eligible formats for URL {link.url}")
 			return
 		best_format: dict = sorted_formats[0]
-		# self.log(message.guild, f"Best format is id {best_format.get('format_id')}")
+		self.__trace(message.guild, f"Best format is id {best_format.get('format_id')}")
 		video_url: str = best_format.get('url')
 		link_description: str = "video"
 		if (best_format.get('width') or 0) > 0 and (best_format.get('height') or 0) > 0:
@@ -220,6 +215,10 @@ class VideoPreviewCog(BaseCog, name='Video Link Previews'):
 			mention_author=False
 		)
 
+	def __trace(self, guild: Guild, message: Any):
+		# self.log(guild, message)
+		pass
+
 	@classmethod
 	def supports_intents(cls, intents: Intents) -> bool:
 		return intents.message_content

rocketbot/cogsetting.py

@@ -3,11 +3,12 @@ A guild configuration setting available for editing via bot commands.
 """
 
 from datetime import timedelta
-from typing import TYPE_CHECKING, Any, Literal, Optional, Union
+from typing import TYPE_CHECKING, Any, Literal, Union
 
 from discord import Interaction, Permissions
 from discord.app_commands import Range, Transform, describe
 from discord.app_commands.commands import Command, CommandCallback, Group, rename
+from discord.errors import DiscordException
 from discord.ext.commands import Bot
 
 from config import CONFIG
@@ -54,13 +55,13 @@ class CogSetting:
 
 	def __init__(self,
 			name: str,
-			datatype: Optional[type],
+			datatype: type | None,
 			default_value: Any,
-			brief: Optional[str] = None,
-			description: Optional[str] = None,
-			min_value: Optional[Any] = None,
-			max_value: Optional[Any] = None,
-			enum_values: Optional[set[Any]] = None):
+			brief: str | None = None,
+			description: str | None = None,
+			min_value: Any | None = None,
+			max_value: Any | None = None,
+			enum_values: set[Any] | None = None):
 		"""
 		Parameters
 		----------
@@ -88,11 +89,11 @@ class CogSetting:
 		self.name: str = name
 		self.datatype: type = datatype
 		self.default_value = default_value
-		self.brief: Optional[str] = brief
+		self.brief: str | None = brief
 		self.description: str = description or ''  # Can't be None
-		self.min_value: Optional[Any] = min_value
-		self.max_value: Optional[Any] = max_value
-		self.enum_values: Optional[set[Any]] = enum_values
+		self.min_value: Any | None = min_value
+		self.max_value: Any | None = max_value
+		self.enum_values: set[Any] | None = enum_values
 		if self.enum_values:
 			value_list = '`' + ('`, `'.join(self.enum_values)) + '`'
 			self.description += f' (Permitted values: {value_list})'
@@ -426,7 +427,7 @@ class CogSetting:
 					text,
 					ephemeral=True,
 				)
-			except BaseException as e:
+			except DiscordException as e:
 				dump_stacktrace(e)
 		show_all_command = Command(
 			name='all',

rocketbot/collections.py

@@ -451,8 +451,8 @@ class AgeBoundList(AbstractMutableList[V], Generic[V, A, D]):
 		if self.is_culling or len(self) <= 1:
 			return
 		self.is_culling = True
-		min_age: Optional[A] = None
-		max_age: Optional[A] = None
+		min_age: A | None = None
+		max_age: A | None = None
 		ages: dict[int, A] = {}
 		for i, elem in enumerate(self):
 			age: A = self.element_age(i, elem)

rocketbot/pattern.py

@@ -5,7 +5,7 @@ to take on them.
 import re
 from abc import ABCMeta, abstractmethod
 from datetime import datetime, timezone
-from typing import Any, Literal, Union
+from typing import Any, ClassVar, Literal
 
 from discord import Message
 from discord import utils as discordutils
@@ -260,7 +260,7 @@ class PatternCompiler:
 	DATATYPE_TEXT: str = 'text'
 	DATATYPE_TIMESPAN: str = 'timespan'
 
-	FIELD_TO_DATATYPE: dict[PatternField, str] = {
+	FIELD_TO_DATATYPE: ClassVar[dict[PatternField, str]] = {
 		PatternSimpleExpression.ALIAS_FIELD_AUTHOR_ID: DATATYPE_MEMBER,
 		PatternSimpleExpression.FIELD_AUTHOR_ID: DATATYPE_ID,
 		PatternSimpleExpression.FIELD_AUTHOR_JOINAGE: DATATYPE_TIMESPAN,
@@ -270,9 +270,9 @@ class PatternCompiler:
 		PatternSimpleExpression.FIELD_CONTENT_PLAIN: DATATYPE_TEXT,
 		PatternSimpleExpression.FIELD_LAST_MATCHED: DATATYPE_TIMESPAN,
 	}
-	DEPRECATED_FIELDS: set[PatternField] = { 'content' }
+	DEPRECATED_FIELDS: ClassVar[set[PatternField]] = { 'content' }
 
-	ACTION_TO_ARGS: dict[PatternActionType, list[str]] = {
+	ACTION_TO_ARGS: ClassVar[dict[PatternActionType, list[str]]] = {
 		PatternAction.TYPE_BAN: [],
 		PatternAction.TYPE_DELETE: [],
 		PatternAction.TYPE_KICK: [],
@@ -281,18 +281,18 @@ class PatternCompiler:
 		PatternAction.TYPE_REPLY: [ DATATYPE_TEXT ],
 	}
 
-	OPERATORS_IDENTITY: set[PatternComparisonOperator] = {
+	OPERATORS_IDENTITY: ClassVar[set[PatternComparisonOperator]] = {
 		PatternSimpleExpression.OP_EQUALS,
 		PatternSimpleExpression.OP_NOT_EQUALS,
 	}
-	OPERATORS_COMPARISON: set[PatternComparisonOperator] = {
+	OPERATORS_COMPARISON: ClassVar[set[PatternComparisonOperator]] = {
 		PatternSimpleExpression.OP_LESS_THAN,
 		PatternSimpleExpression.OP_GREATER_THAN,
 		PatternSimpleExpression.OP_LESS_THAN_OR_EQUALS,
 		PatternSimpleExpression.OP_GREATER_THAN_OR_EQUALS,
 	}
-	OPERATORS_NUMERIC: set[PatternComparisonOperator] = OPERATORS_IDENTITY | OPERATORS_COMPARISON
-	OPERATORS_TEXT: set[PatternComparisonOperator] = OPERATORS_IDENTITY | {
+	OPERATORS_NUMERIC: ClassVar[set[PatternComparisonOperator]] = OPERATORS_IDENTITY | OPERATORS_COMPARISON
+	OPERATORS_TEXT: ClassVar[set[PatternComparisonOperator]] = OPERATORS_IDENTITY | {
 		PatternSimpleExpression.OP_CONTAINS,
 		PatternSimpleExpression.OP_NOT_CONTAINS,
 		PatternSimpleExpression.OP_CONTAINS_WORD,
@@ -302,7 +302,7 @@ class PatternCompiler:
 	}
 	OPERATORS_ALL: set[str] = OPERATORS_IDENTITY | OPERATORS_COMPARISON | OPERATORS_TEXT
 
-	DATATYPE_TO_OPERATORS: dict[str, set[PatternComparisonOperator]] = {
+	DATATYPE_TO_OPERATORS: ClassVar[dict[str, set[PatternComparisonOperator]]] = {
 		DATATYPE_ID: OPERATORS_IDENTITY,
 		DATATYPE_MEMBER: OPERATORS_IDENTITY,
 		DATATYPE_TEXT: OPERATORS_TEXT,
@@ -356,7 +356,7 @@ class PatternCompiler:
 		Converts a message filter statement into a list of tokens.
 		"""
 		tokens: list[str] = []
-		in_quote: Union[bool, str] = False
+		in_quote: bool | str = False
 		in_escape: bool = False
 		all_token_types: set[str] = { 'sym', 'op', 'val' }
 		possible_token_types: set[str] = set(all_token_types)
@@ -404,11 +404,11 @@ class PatternCompiler:
 					if ch in cls.OP_CHARS:
 						possible_ch_types.add('op')
 					if len(current_token) > 0 and \
-							possible_ch_types.isdisjoint(possible_token_types):
-						if len(current_token) > 0:
-							tokens.append(current_token)
-							current_token = ''
-							possible_token_types |= all_token_types
+							possible_ch_types.isdisjoint(possible_token_types) and \
+							len(current_token) > 0:
+						tokens.append(current_token)
+						current_token = ''
+						possible_token_types |= all_token_types
 					possible_token_types &= possible_ch_types
 					current_token += ch
 		if len(current_token) > 0:
@@ -590,7 +590,7 @@ class PatternCompiler:
 				raise PatternError(f'Operator {op} cannot be used with ' + \
 					f'field "{field}"')
 			raise PatternError(f'Unrecognized operator "{op}" - allowed: ' + \
-				f'{sorted(list(allowed_ops))}')
+				f'{sorted(allowed_ops)}')
 
 		if token_index >= len(tokens):
 			raise PatternError('Expected value, found EOL')
@@ -606,7 +606,7 @@ class PatternCompiler:
 		return exp, token_index
 
 	@classmethod
-	def __parse_value(cls, value: str, datatype: str, op: str = None) -> Any:
+	def __parse_value(cls, value: str, datatype: str, op: str | None = None) -> Any:
 		"""
 		Converts a value token to its Python value. Raises ValueError on failure.
 		"""

rocketbot/storage.py

@@ -4,7 +4,7 @@ Handles storage of persisted and non-persisted data for the bot.
 import json
 from datetime import datetime, timedelta, timezone
 from os.path import exists
-from typing import Any, Optional
+from typing import Any
 
 from discord import Guild
 
@@ -28,7 +28,7 @@ class Storage:
 
 	# -- Transient state management -----------------------------------------
 
-	__guild_id_to_state: dict[int, dict[str, Any]] = {}
+	__guild_id_to_state: dict[int, dict[str, Any]] = {}  # noqa: RUF012
 
 	@classmethod
 	def get_state(cls, guild: Guild) -> dict[str, Any]:
@@ -43,14 +43,14 @@ class Storage:
 		return state
 
 	@classmethod
-	def get_state_value(cls, guild: Guild, key: str) -> Optional[Any]:
+	def get_state_value(cls, guild: Guild, key: str) -> Any | None:
 		"""
 		Returns a state value for the given guild and key, or `None` if not set.
 		"""
 		return cls.get_state(guild).get(key)
 
 	@classmethod
-	def set_state_value(cls, guild: Guild, key: str, value: Optional[Any]) -> None:
+	def set_state_value(cls, guild: Guild, key: str, value: Any | None) -> None:
 		"""
 		Updates a transient value associated with the given guild and key name.
 		A value of `None` removes any previous value for that key.
@@ -58,7 +58,7 @@ class Storage:
 		cls.set_state_values(guild, { key: value })
 
 	@classmethod
-	def set_state_values(cls, guild: Guild, values: Optional[dict[str, Optional[Any]]]) -> None:
+	def set_state_values(cls, guild: Guild, values: dict[str, Any | None] | None) -> None:
 		"""
 		Merges in a set of key-value pairs into the transient state for the
 		given guild. Any pairs with a value of `None` will be removed from the
@@ -78,7 +78,7 @@ class Storage:
 	# -- Persisted configuration management ---------------------------------
 
 	# discord.Guild.id -> dict
-	__guild_id_to_config: dict[int, dict[str, Any]] = {}
+	__guild_id_to_config: dict[int, dict[str, Any]] = {}  # noqa: RUF012
 
 	@classmethod
 	def get_config(cls, guild: Guild) -> dict[str, Any]:
@@ -99,7 +99,7 @@ class Storage:
 		return config
 
 	@classmethod
-	def get_config_value(cls, guild: Guild, key: str) -> Optional[Any]:
+	def get_config_value(cls, guild: Guild, key: str) -> Any | None:
 		"""
 		Returns a persisted guild config value stored under the given key.
 		Returns `None` if not present.
@@ -107,7 +107,7 @@ class Storage:
 		return cls.get_config(guild).get(key)
 
 	@classmethod
-	def set_config_value(cls, guild: Guild, key: str, value: Optional[Any]) -> None:
+	def set_config_value(cls, guild: Guild, key: str, value: Any | None) -> None:
 		"""
 		Adds/updates the given key-value pair to the persisted config for the
 		given Guild. If `value` is `None` the key will be removed from persisted
@@ -116,7 +116,7 @@ class Storage:
 		cls.set_config_values(guild, { key: value })
 
 	@classmethod
-	def set_config_values(cls, guild: Guild, values: Optional[dict[str, Optional[Any]]]) -> None:
+	def set_config_values(cls, guild: Guild, values: dict[str, Any | None] | None) -> None:
 		"""
 		Merges the given `values` dict with the saved config for the given guild
 		and writes it to disk. `values` must be JSON-encodable or a `ValueError`
@@ -164,7 +164,7 @@ class Storage:
 		cls.__trace('State saved')
 
 	@classmethod
-	def __read_guild_config(cls, guild: Guild) -> Optional[dict[str, Any]]:
+	def __read_guild_config(cls, guild: Guild) -> dict[str, Any] | None:
 		"""
 		Loads config for a guild from a JSON file on disk, or `None` if not
 		found.

rocketbot/ui/pagedcontent.py

@@ -4,10 +4,9 @@ source message to insert `PAGE_BREAK` characters at meaningful breaks, preferabl
 at fairly uniform intervals.
 """
 
-from typing import Optional
-
 from discord import Interaction
-from discord.ui import LayoutView, TextDisplay, ActionRow, Button
+from discord.errors import DiscordException
+from discord.ui import ActionRow, Button, LayoutView, TextDisplay
 
 from rocketbot.utils import dump_stacktrace
 
@@ -46,7 +45,7 @@ def paginate(text: str) -> list[str]:
 
 async def update_paged_content(
 		interaction: Interaction,
-		original_interaction: Optional[Interaction],
+		original_interaction: Interaction | None,
 		current_page: int,
 		pages: list[str],
 		**send_args,
@@ -96,7 +95,7 @@ async def update_paged_content(
 				ephemeral=True,
 				**send_args,
 			)
-	except BaseException as e:
+	except DiscordException as e:
 		dump_stacktrace(e)
 
 class _PagingLayoutView(LayoutView):
@@ -104,7 +103,7 @@ class _PagingLayoutView(LayoutView):
 			self,
 			current_page: int,
 			pages: list[str],
-			original_interaction: Optional[Interaction],
+			original_interaction: Interaction | None,
 			**send_args,
 	) -> None:
 		super().__init__()

rocketbot/utils.py

@@ -5,7 +5,7 @@ import re
 import sys
 import traceback
 from datetime import datetime, timedelta, timezone
-from typing import Any, Optional, Union
+from typing import Any
 
 import discord
 from discord import Guild, Interaction, Permissions
@@ -105,7 +105,7 @@ def describe_timedelta(td: timedelta, max_components: int = 2) -> str:
 		components = components[0:max_components]
 	return ' '.join(components)
 
-def _old_first_command_group(cog: Cog) -> Optional[discord.ext.commands.Group]:
+def _old_first_command_group(cog: Cog) -> discord.ext.commands.Group | None:
 	"""Returns the first command Group found in a cog."""
 	for member_name in dir(cog):
 		member = getattr(cog, member_name)
@@ -113,7 +113,7 @@ def _old_first_command_group(cog: Cog) -> Optional[discord.ext.commands.Group]:
 			return member
 	return None
 
-def first_command_group(cog: Cog) -> Optional[discord.app_commands.Group]:
+def first_command_group(cog: Cog) -> discord.app_commands.Group | None:
 	"""Returns the first slash command Group found in a cog."""
 	for member_name in dir(cog):
 		member = getattr(cog, member_name)
@@ -121,9 +121,9 @@ def first_command_group(cog: Cog) -> Optional[discord.app_commands.Group]:
 			return member
 	return None
 
-def bot_log(guild: Optional[Guild], cog_class: Optional[type], message: Any) -> None:
+def bot_log(guild: Guild | None, cog_class: type | None, message: Any) -> None:
 	"""Logs a message to stdout with time, cog, and guild info."""
-	now: datetime = datetime.now() # local
+	now: datetime = datetime.now(tz=None)  # noqa: DTZ005
 	s = f'[{now.strftime("%Y-%m-%dT%H:%M:%S")}|'
 	s += f'{cog_class.__name__}|' if cog_class else '-|'
 	s += f'{guild.name}] ' if guild else '-] '
@@ -169,11 +169,11 @@ def user_id_from_mention(mention: str) -> str:
 		return m.group(1)
 	raise ValueError(f'"{mention}" is not an @ user mention')
 
-def mention_from_user_id(user_id: Union[str, int]) -> str:
+def mention_from_user_id(user_id: str | int) -> str:
 	"""Returns a Markdown user mention from a user id."""
 	return f'<@!{user_id}>'
 
-def mention_from_role_id(role_id: Union[str, int]) -> str:
+def mention_from_role_id(role_id: str | int) -> str:
 	"""Returns a Markdown role mention from a role id."""
 	return f'<@&{role_id}>'
 
@@ -197,8 +197,7 @@ def suppress_markdown_url_previews(markdown: str) -> str:
 
 def format_bytes(size: int) -> str:
 	"""Formats s size in bytes to a human readable description (e.g. "3.2 KiB")"""
-	if size < 0:
-		size = 0
+	size = max(size, 0)
 	kib = 1024
 	mib = kib * kib
 	gib = mib * kib
@@ -223,7 +222,27 @@ def norm_datetime(dt: datetime) -> datetime:
 		return dt
 	return datetime.fromtimestamp(dt.timestamp(), timezone.utc)
 
+def levenshtein(a: str, b: str) -> int:
+	"""Returns the Levenshtein distance between two strings."""
+	# Based on https://en.wikipedia.org/wiki/Levenshtein_distance#Iterative_with_two_matrix_rows
+	m: int = len(a)
+	n: int = len(b)
+	v0: list[int] = [ i for i in range(n + 1) ]
+	v1: list[int] = [ 0 for i in range(n + 1) ]
+	for i in range(m):
+		v1[0] = i + 1
+		for j in range(n):
+			deletion_cost = v0[j + 1] + 1
+			insertion_cost = v1[j] + 1
+			substitution_cost = v0[j] + (0 if a[i] == b[j] else 1)
+			v1[j + 1] = min(deletion_cost, insertion_cost, substitution_cost)
+		h = v0
+		v0 = v1
+		v1 = h
+	return v0[n]
+
 MOD_PERMISSIONS: Permissions = Permissions(Permissions.manage_messages.flag)
+ADMIN_PERMISSIONS: Permissions = Permissions(Permissions.administrator.flag)
 
 class TimeDeltaTransformer(Transformer):
 	async def transform(self, interaction: Interaction, value: Any) -> timedelta:

sql/kickstarter-create.sql

@@ -1,11 +1,3 @@
-CREATE TABLE IF NOT EXISTS kickstarter_emails (
-	pk INTEGER PRIMARY KEY AUTOINCREMENT,
-	guild_id INTEGER NOT NULL,  -- Discord Guild.id
-	email_hash TEXT NOT NULL,  -- base64(sha256(email.lower.trimmed))
-	imported_at INTEGER NOT NULL,  -- unix timestamp
-	UNIQUE(guild_id, email_hash) ON CONFLICT FAIL
-);
-
 CREATE TABLE IF NOT EXISTS kickstarter_discord_users (
 	pk INTEGER PRIMARY KEY AUTOINCREMENT,
 	guild_id INTEGER NOT NULL,  -- Discord Guild.id
@@ -17,13 +9,3 @@ CREATE TABLE IF NOT EXISTS kickstarter_discord_users (
 );
 CREATE INDEX IF NOT EXISTS idx_kickstarter_discord_users_discord_member_id
 	ON kickstarter_discord_users (discord_member_id);
-
-CREATE TABLE IF NOT EXISTS member_links (
-	pk INTEGER PRIMARY KEY AUTOINCREMENT,
-	guild_id INTEGER NOT NULL,  -- Discord Guild.id
-	member_id INTEGER NOT NULL,  -- Discord Member.id
-	email_hash TEXT NOT NULL,  -- base64(sha256(email.lower.trimmed))
-	complete INTEGER NOT NULL DEFAULT 0,  -- link found, role assigned successfully
-	UNIQUE(guild_id, member_id) ON CONFLICT FAIL,
-	UNIQUE(guild_id, email_hash) ON CONFLICT FAIL
-);