config.sample.py

 # Copy this file to config.py and fill in "<REQUIRED>" values
 CONFIG = {
-	'__config_version': 4,
+	'__config_version': 5,
 
     # -----------------------------------------------------------------------
     # General
 	'max_members_per_message': 20,
 	# Minimum seconds between warnings about the same member.
 	'squelch_warning_seconds': 300,
+	# Whether the Discord app has Message Content intent enabled. Enables more
+	# features but requires applying for access for bots serving 10k or more users.
+	'has_message_content_intent': False,
+	# Whether the Discord app has Server Members intent enabled. Enables more
+	# features but requires applying for access for bots serving 10k or more users.
+	'has_members_intent': False,
 
     # -----------------------------------------------------------------------
     # Emojis used for performing actions on flagged activities. The bot will

main.py

 	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 = 4
+CURRENT_CONFIG_VERSION = 5
 if (CONFIG.get('__config_version') or 0) < CURRENT_CONFIG_VERSION:
 	# If you're getting this error, it means something changed in config.py's
 	# format. Consult config.sample.py and compare it to your own config.py.

requirements.txt

-discord.py == 2.6.4
+discord.py == 2.7.1
 yt-dlp == 2026.07.04

rocketbot/bot.py

 import traceback
-from typing import Optional
 
 from discord import Intents
+from discord.errors import DiscordException
 from discord.ext import commands
 
 from config import CONFIG
 					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
 		return
 	bot_log(None, None, 'Creating bot...')
 	intents = Intents.default()
-	intents.message_content = True
-	intents.members = True
+	# Set privileged intents. Must agree with configuration in Discord app.
+	intents.message_content = CONFIG.get('has_message_content_intent', False)
+	intents.members = CONFIG.get('has_members_intent', False)
 	intents.presences = False
 	rocketbot = Rocketbot(intents=intents)
 __create_bot()
 	from rocketbot.cogs.generalcog import GeneralCog
 	from rocketbot.cogs.helpcog import HelpCog
 	from rocketbot.cogs.joinraidcog import JoinRaidCog
+	from rocketbot.cogs.kickstartercog import KickstarterCog
 	from rocketbot.cogs.logcog import LoggingCog
 	from rocketbot.cogs.patterncog import PatternCog
 	from rocketbot.cogs.urlspamcog import URLSpamCog
 	await rocketbot.add_cog(ConfigCog(rocketbot))
 
 	# Optional
+	if AutoKickCog.supports_intents(rocketbot.intents):
 		await rocketbot.add_cog(AutoKickCog(rocketbot))
+	if BangCommandCog.supports_intents(rocketbot.intents):
 		await rocketbot.add_cog(BangCommandCog(rocketbot))
+	if CrossPostCog.supports_intents(rocketbot.intents):
 		await rocketbot.add_cog(CrossPostCog(rocketbot))
+	if GamesCog.supports_intents(rocketbot.intents):
 		await rocketbot.add_cog(GamesCog(rocketbot))
+	if HelpCog.supports_intents(rocketbot.intents):
 		await rocketbot.add_cog(HelpCog(rocketbot))
+	if JoinRaidCog.supports_intents(rocketbot.intents):
 		await rocketbot.add_cog(JoinRaidCog(rocketbot))
+	if LoggingCog.supports_intents(rocketbot.intents):
 		await rocketbot.add_cog(LoggingCog(rocketbot))
+	if PatternCog.supports_intents(rocketbot.intents):
 		await rocketbot.add_cog(PatternCog(rocketbot))
+	if URLSpamCog.supports_intents(rocketbot.intents):
 		await rocketbot.add_cog(URLSpamCog(rocketbot))
+	if UsernamePatternCog.supports_intents(rocketbot.intents):
 		await rocketbot.add_cog(UsernamePatternCog(rocketbot))
+	if VideoPreviewCog.supports_intents(rocketbot.intents):
 		await rocketbot.add_cog(VideoPreviewCog(rocketbot))
+	if KickstarterCog.supports_intents(rocketbot.intents):
+		await rocketbot.add_cog(KickstarterCog(rocketbot))
 
 	await rocketbot.start(CONFIG['client_token'], reconnect=True)
 	print('\nBot aborted')

rocketbot/botmessage.py

 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
 
 
 	@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
 			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.
 		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] = []
 
 		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
 
 			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.
 				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.
 					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/autokickcog.py

 from datetime import datetime, timedelta, timezone
 
-from discord import Guild, Member, Status
+from discord import Guild, Intents, Member, Status
 from discord.ext import tasks
 from discord.ext.commands import Cog
 from discord.ext.tasks import Loop
 			if val % 10 == 3:
 				return f'{val}rd'
 		return f'{val}th'
+
+	@classmethod
+	def supports_intents(cls, intents: Intents) -> bool:
+		return intents.members

rocketbot/cogs/bangcommandcog.py

 import re
 from typing import Optional, TypedDict
 
-from discord import Guild, Interaction, Message, SelectOption, TextChannel, TextStyle
+from discord import (
+	Guild,
+	Intents,
+	Interaction,
+	Message,
+	SelectOption,
+	TextChannel,
+	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
 
 	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)
 		}
 	)
 	@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.
 
 		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:
+		if check_exists and cmds.get(name, None) is not None:
 			raise ValueError(f'Command with name "{name}" already exists.')
 		cmds[name] = {
 			'content': content,
 
 	@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)
 		return BangCommandCog._normalize_name(match.group(1)) if match else None
 
+	@classmethod
+	def supports_intents(cls, intents: Intents) -> bool:
+		return intents.message_content
+
 class _EditModal(Modal, title='Edit Command'):
 	name_label = Label(
 		text='Command name',
 		)
 	)
 
-	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
 				f'{CONFIG["failure_emoji"]} Save failed',
 				ephemeral=True,
 			)
-		except BaseException:
-			pass
+		except DiscordException as e:
+			dump_stacktrace(e)

rocketbot/cogs/basecog.py

 Base cog class and helper classes.
 """
 from datetime import datetime, timedelta, timezone
-from typing import Optional
 
 import discord
 from discord import (
 	Guild,
+	Intents,
 	Interaction,
 	Member,
 	Message,
 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
 	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
 		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
 
 	async def cog_app_command_error(self, interaction: Interaction, error: AppCommandError) -> None:
 		if isinstance(error, CommandInvokeError):
 			error = error.original
+		bot_log(interaction.guild, type(self), f"Error in {interaction.command.qualified_name if interaction.command else 'unknown command'}")
 		dump_stacktrace(error)
 		message = f"\nException: {error.__class__.__name__}, "\
 				  f"Command: {interaction.command.qualified_name if interaction.command else None}, "\
 		try:
 			await interaction.response.send_message(f"An error occurred: {message}", ephemeral=True)
 		except discord.InteractionResponded:
+			try:
 				await interaction.followup.send(f"An error occurred: {message}", ephemeral=True)
+			except DiscordException:
+				bot_log(interaction.guild, None, message)
 
 	@property
 	def basecogs(self) -> list['BaseCog']:
 
 	@classmethod
 	def get_guild_setting(cls,
-			guild: Optional[Guild],
+			guild: Guild | None,
 			setting: CogSetting,
 			use_cog_default_if_not_set: bool = True):
 		"""
 	# 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.
 		"""
 		bot_log(guild, cls, message)
+
+	# Meta
+
+	@classmethod
+	def supports_intents(cls, intents: Intents) -> bool:
+		return True

rocketbot/cogs/configcog.py

 """
 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
 	)
 	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

 """
 import re
 from datetime import datetime, timedelta, timezone
-from typing import Optional
 
 from discord import Member, Message, TextChannel
 from discord import utils as discordutils
 	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
 			self.__trace(f"User {message.author.name} exempt from crosspost checks")
 			return
 		def compute_message_hash(m: Message) -> int:
+			if self.bot.intents.message_content:
 				to_hash = m.content
 				# URLs sometimes differ per spam message, so simplify them
 				url_regex = r'http[s]?://(?:[a-zA-Z]|[0-9]|[$-_@.&+]|[!*\(\),]|(?:%[0-9a-fA-F][0-9a-fA-F]))+'
 				# Add attachment metadata
 				for attachment in m.attachments:
 					to_hash += f'\n[[ATT: ct={attachment.content_type} s={attachment.size} w={attachment.width} h={attachment.height}]]'
+			else:
+				# Without content, treat every message as unique
+				to_hash = str(m.id)
 			h = hash(to_hash)
 			self.__trace(f"Hash for message #{m.id} by {m.author.name} is {h}\n\thash content: \"{to_hash}\"")
 			return h
 		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)
 				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

 		)
 		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

 """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 (
 	command,
 	guild_only,
 )
+from discord.errors import DiscordException
 
 from config import CONFIG
 from rocketbot.bot import Rocketbot
 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
 			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 []
 
 				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:
 						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]:
 			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()
 
 		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))
 		]
 
 	)
 	@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.
 
 			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
 
 		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
 
 		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'
 	'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/joinraidcog.py

 from datetime import datetime, timedelta
 from typing import Optional
 
-from discord import Guild, Member
+from discord import Guild, Intents, Member
 from discord.ext.commands import Cog
 
 from config import CONFIG
 			did_kick=kick_count >= member_count,
 			did_ban=ban_count >= member_count,
 			user_count=member_count))
+
+	@classmethod
+	def supports_intents(cls, intents: Intents) -> bool:
+		return intents.members

rocketbot/cogs/kickstartercog.py

+import heapq
+from asyncio import sleep
+from collections.abc import Awaitable, Callable, Iterable
+from enum import IntEnum
+from sqlite3 import Connection, connect
+from time import time as now_timestamp
+from typing import Optional, TypeVar
+
+from discord import Attachment, Guild, Interaction, Member, Role
+from discord.app_commands import Group, default_permissions
+from discord.errors import DiscordException, HTTPException
+from discord.ext.commands import Cog
+from discord.ui import FileUpload, Label, Modal
+
+from config import CONFIG
+from rocketbot.bot import Rocketbot
+from rocketbot.cogs.basecog import BaseCog
+from rocketbot.cogsetting import CogSetting
+from rocketbot.utils import (
+	ADMIN_PERMISSIONS,
+	MOD_PERMISSIONS,
+	dump_stacktrace,
+	is_discord_username,
+	levenshtein,
+)
+
+_T = TypeVar('_T')
+class KickstarterCog(BaseCog):
+	"""
+	Assigns Discord users a role if their username appears in an imported list,
+	generally the results of a Kickstarter survey.
+
+	Discord does not currently have a native Kickstarter integration, and doing
+	a full API integration is a bit ambitious, so this is a stopgap solution.
+	"""
+
+	shared: Optional['KickstarterCog'] = None
+
+	SETTING_ENABLED = CogSetting(
+		name='enabled',
+		datatype=bool,
+		default_value=False,
+		brief='Kickstarter user linking',
+		description='Whether this module is enabled for a guild.',
+	)
+	SETTING_ROLE = CogSetting(
+		name='backer_role',
+		datatype=int,
+		default_value=0,
+		brief='role to assign to Kickstarter backers',
+		description=''
+	)
+
+	def __init__(self, bot: Rocketbot):
+		super().__init__(
+			bot,
+			config_prefix='kickstarter',
+			short_description='For linking Kickstarter backers to their Discord handles.',
+		)
+		Self = KickstarterCog
+		self.add_setting(Self.SETTING_ENABLED)
+		# SETTING_ROLE managed manually
+		self.con: Connection = connect('kickstarter.sqlite3')
+		Self.shared = self
+
+	def __is_enabled(self, guild: Guild) -> bool:
+		Self = KickstarterCog
+		return self.get_guild_setting(guild, Self.SETTING_ENABLED)
+
+	def __get_backer_role_id(self, guild: Guild) -> int | None:
+		Self = KickstarterCog
+		role_id = self.get_guild_setting(guild, Self.SETTING_ROLE)
+		return role_id if role_id != 0 else None
+
+	async def __fetch_backer_role(self, guild: Guild) -> Role:
+		role_id = self.__get_backer_role_id(guild)
+		if role_id is None:
+			raise _NoBackerRoleException()
+		ret_val = guild.get_role(role_id) or await guild.fetch_role(role_id)
+		if ret_val is None:
+			self.log(guild, f"Backer role with id {role_id} could not be retrieved. Role removed?")
+			raise _NoBackerRoleException()
+		return ret_val
+
+	def __get_cached_backer_role(self, guild: Guild) -> Role | None:
+		"""Synchronously gets the configured backer role from the cache if possible."""
+		role_id = self.__get_backer_role_id(guild)
+		return guild.get_role(role_id) if role_id is not None else None
+
+	def __set_backer_role(self, role: Role | None):
+		Self = KickstarterCog
+		self.set_guild_setting(role.guild, Self.SETTING_ROLE, role.id if role else None)
+
+	async def __check_disabled(self, interaction: Interaction) -> bool:
+		"""Checks if this feature is enabled, and if not, sends an error
+		response. Return value is whether caller should bail out."""
+		if not self.__is_enabled(interaction.guild):
+			text = f"{CONFIG['failure_emoji']} Kickstarter feature not enabled"
+			await interaction.response.send_message(text, ephemeral=True)
+			return True
+		return False
+
+	async def __check_configured(self, interaction: Interaction) -> bool:
+		"""Checks if the current guild is configured properly for this feature,
+		and if not, sends an error response. Return value is whether caller
+		should bail out."""
+		if self.__get_backer_role_id(interaction.guild) is None:
+			text = f"{CONFIG['failure_emoji']} Backer role is not configured"
+			await interaction.response.send_message(text, ephemeral=True)
+			return True
+		return False
+
+	# -- Admin commands -----
+
+	kickstarter = Group(
+		name='kickstarter',
+		description='Manages roles for users identified as Kickstarter backers.',
+		guild_only=True,
+		default_permissions=MOD_PERMISSIONS
+	)
+
+	@kickstarter.command(
+		description='Configures which role to give to Kickstarter backers.'
+	)
+	@default_permissions(ADMIN_PERMISSIONS)
+	async def set_role(self, interaction: Interaction, role: Role):
+		self.__set_backer_role(role)
+		text = f"{CONFIG['info_emoji']} Backer role set to {role.name}"
+		await interaction.response.send_message(text, ephemeral=True)
+
+	@kickstarter.command(
+		description='Assigns backer role to imported backer usernames.'
+	)
+	async def sync(self, interaction: Interaction):
+		if await self.__check_disabled(interaction): return
+		if await self.__check_configured(interaction): return
+
+		guild = interaction.guild
+		await interaction.response.defer(ephemeral=True, thinking=True)
+		sync_result: _SyncUsernamesResult = await self.__sync_by_username(guild)
+		text = f"{CONFIG['success_emoji']} Sync complete.\n" + \
+			sync_result.summary_markdown()
+		await interaction.followup.send(text, ephemeral=True)
+
+	@kickstarter.command(
+		description='Shows info about Kickstarter linked Discord members.'
+	)
+	async def info(self, interaction: Interaction):
+		if await self.__check_disabled(interaction): return
+		if await self.__check_configured(interaction): return
+
+		await interaction.response.defer(ephemeral=True, thinking=True)
+		guild = interaction.guild
+		backer_role = await self.__fetch_backer_role(guild)
+		stats: _Stats = self.__fetch_stats(guild.id)
+
+		lines: list[str] = []
+		if backer_role is None:
+			lines.append("- No backer role configured yet (use " \
+				f"`/{KickstarterCog.kickstarter.name} {KickstarterCog.set_role.name}`)")
+		else:
+			lines.append(f"- Backer role configured as `{backer_role.name}`")
+		if stats.username_count > 0:
+			lines.append(f"- **{stats.username_count:,}** Discord usernames imported")
+			lines.append("- Last import with new records on " \
+				f"<t:{stats.username_last_imported_at}:f>")
+			lines.append(f"- **{stats.username_found_count:,}** members linked successfully")
+			if stats.username_not_found_count > 0:
+				lines.append(f"- **{stats.username_not_found_count:,}** imported " \
+					"usernames not yet linked to Discord members")
+			if stats.username_unprocessed_count > 0:
+				lines.append(f"- **{stats.username_unprocessed_count:,}** " \
+					"usernames added since last sync")
+		else:
+			lines.append("- No Discord usernames imported yet. Use " \
+				f"`/{KickstarterCog.kickstarter.name} {KickstarterCog.upload.name}`")
+
+		text = f"{CONFIG['info_emoji']} Kickstarter import stats\n\n"
+		text += "\n".join(lines)
+		await interaction.followup.send(text, ephemeral=True)
+
+	@kickstarter.command(
+		description='Shows Kickstarter link details about a specific Discord member.'
+	)
+	async def find_member(self, interaction: Interaction, member: Member):
+		if await self.__check_disabled(interaction): return
+		if await self.__check_configured(interaction): return
+
+		await interaction.response.defer(ephemeral=True, thinking=True)
+		guild = interaction.guild
+		try:
+			backer_role = await self.__fetch_backer_role(guild)
+		except _NoBackerRoleException:
+			text = f"{CONFIG['failure_emoji']} Backer role not yet configured!"
+			await interaction.followup.send(text, ephemeral=True)
+			return
+
+		if backer_role in member.roles:
+			text = f"{CONFIG['info_emoji']} Member `@{member.name}` has " \
+				f"`{backer_role.name}` role."
+			await interaction.followup.send(text, ephemeral=True)
+			return
+
+		if member.bot:
+			text = f"{CONFIG['info_emoji']} Member `@{member.name}` is a robut! " \
+				"Robuts cannot back Kickstarters!"
+			await interaction.followup.send(text, ephemeral=True)
+			return
+
+		username_record = self.__fetch_kickstarter_username(guild.id, username=member.name)
+		if username_record is not None:
+			await member.add_roles(backer_role)
+			username_record.lookup_status = _LookupStatus.member_found
+			username_record.discord_member_id = member.id
+			self.__update_kickstarter_username(username_record)
+			text = f"{CONFIG['success_emoji']} Member's username found in " \
+				f"Kickstarter export. The `{backer_role.name}` role has now " \
+				"been assigned to them."
+			await interaction.followup.send(text, ephemeral=True)
+			return
+
+		closest_matches = self.__find_nearest_usernames(guild.id, member.name,
+			limit=8, max_distance=3)
+		stats = self.__fetch_stats(guild.id)
+		text = f"{CONFIG['info_emoji']} The username `@{member.name}` is not in " \
+			"the latest Kickstarter import." \
+			"\n" \
+			"\n- Did they complete their survey yet?" \
+			"\n- Did they fill in their Discord username in the survey?" \
+			"\n- Did they complete their survey after the last import " \
+				f"(<t:{stats.username_last_imported_at}:f>)? They might be in the next one."
+		if len(closest_matches) > 0:
+			text += "\n- Did they misspell their username in the survey? " \
+				"Here are some similar ones."
+			for similar in closest_matches:
+				text += f"\n   - `{similar.discord_username}`"
+				if similar.lookup_status == _LookupStatus.misspelled_username:
+					text += f" (manually linked to <@{similar.discord_member_id}> by mod)"
+			text += f"\n-# Hint: Use `/{KickstarterCog.kickstarter.name} {KickstarterCog.link.name} " \
+				f"@{member.name} surveyusername` to link the member to their " \
+				"misspelled survey username"
+		await interaction.followup.send(text, ephemeral=True)
+
+	@kickstarter.command(
+		description='Manually links a member to a misspelled survey username.'
+	)
+	async def link(self, interaction: Interaction, member: Member, username: str):
+		if await self.__check_disabled(interaction): return
+		if await self.__check_configured(interaction): return
+
+		await interaction.response.defer(ephemeral=True, thinking=True)
+		guild = interaction.guild
+		try:
+			backer_role = await self.__fetch_backer_role(guild)
+		except _NoBackerRoleException:
+			text = f"{CONFIG['failure_emoji']} Backer role not configured!"
+			await interaction.followup.send(text, ephemeral=True)
+			return
+
+		normal_username = KickstarterCog.__normalize_discord_username(username)
+		username = self.__fetch_kickstarter_username(guild.id, username=normal_username)
+		if username is None:
+			closest_matches = self.__find_nearest_usernames(guild.id, normal_username, 1)
+			text = f"{CONFIG['failure_emoji']} No survey username found for `{username}`."
+			if len(closest_matches) > 0:
+				text += f" Did you mean `/{KickstarterCog.kickstarter.name} " \
+					f"{KickstarterCog.link.name} @{member.name} " \
+					f"{closest_matches[0].discord_username}`?"
+			await interaction.followup.send(text, ephemeral=True)
+			return
+
+		if username.lookup_status == _LookupStatus.member_found:
+			text = f"{CONFIG['failure_emoji']} That survey username is already " \
+				f"linked to <@{username.discord_member_id}>. Cannot be linked to " \
+				"another member.\n" \
+				f"- `/{KickstarterCog.kickstarter.name} {KickstarterCog.reset.name} " \
+				f"{username}` will unlink that Discord record\n" \
+				"- If all else fails, you can just manually give them the backer role."
+			await interaction.followup.send(text, ephemeral=True)
+			return
+		elif username.lookup_status == _LookupStatus.misspelled_username:
+			text = f"{CONFIG['failure_emoji']} That survey username was already " \
+				f"assigned to <@{username.discord_member_id}> by a mod using this " \
+				"command.\n" \
+				f"- `/{KickstarterCog.kickstarter.name} {KickstarterCog.reset.name} " \
+				f"{username}` will unlink that Discord record\n" \
+				"- If all else fails, you can just manually give them the backer role."
+			await interaction.followup.send(text, ephemeral=True)
+			return
+
+		if backer_role not in member.roles:
+			await member.add_roles(backer_role)
+		username.discord_member_id = member.id
+		username.lookup_status = _LookupStatus.misspelled_username
+		self.__update_kickstarter_username(username)
+
+		text = f"{CONFIG['success_emoji']} Member @{member.name} linked to " \
+			f"survey username @{username.discord_username} and given " \
+			f"{backer_role.name} role!"
+		await interaction.followup.send(text, ephemeral=True)
+
+	@kickstarter.command(
+		description='Resets an imported Kickstarter record.'
+	)
+	async def reset(self, interaction: Interaction, username: str):
+		if await self.__check_disabled(interaction): return
+
+		guild = interaction.guild
+		normal_username = KickstarterCog.__normalize_discord_username(username)
+		record = self.__fetch_kickstarter_username(guild.id, username=normal_username)
+		if record is None:
+			text = f"{CONFIG['failure_emoji']} No record found for `@{normal_username}`."
+			similar = self.__find_nearest_usernames(guild.id, normal_username, limit=5, max_distance=3)
+			if len(similar) > 0:
+				text += " Did you mean one of these?"
+				for s in similar:
+					text += f"\n- `@{s.discord_username}`"
+			await interaction.response.send_message(text, ephemeral=True)
+			return
+		old_member_id = record.discord_member_id
+		old_lookup_status = record.lookup_status
+		record.discord_member_id = None
+		record.lookup_status = _LookupStatus.unprocessed
+		self.__update_kickstarter_username(record)
+		text = f"{CONFIG['success_emoji']} Record for `@{normal_username}` reset. " \
+			"(No roles removed from member.)"
+		if old_member_id is not None:
+			text += f"\n- `discord_member_id` changed from `{old_member_id}` to `{record.discord_member_id}`"
+		if old_lookup_status is not None:
+			text += f"\n- `lookup_status` changed from `{_describe_lookup_status(old_lookup_status)}`" \
+				f" to `{_describe_lookup_status(record.lookup_status)}`"
+		await interaction.response.send_message(text, ephemeral=True)
+
+	@kickstarter.command(
+		description='Uploads an export of Kickstarter users.'
+	)
+	async def upload(self, interaction: Interaction):
+		if await self.__check_disabled(interaction): return
+		if await self.__check_configured(interaction): return
+
+		await interaction.response.send_modal(_UploadUsernamesModal())
+
+	# @override
+	async def interaction_check(self, interaction: Interaction) -> bool:
+		if interaction.command is not None:
+			self.__trace(interaction.guild, f"@{interaction.user.name} used " \
+				f"/{interaction.command.qualified_name}")
+		return True
+
+
+	# -- Events --
+
+	@Cog.listener()
+	async def on_member_join(self, member: Member) -> None:
+		guild = member.guild
+		if not self.__is_enabled(guild): return
+		if self.__get_backer_role_id(guild) is None: return
+
+		try:
+			backer_role = await self.__fetch_backer_role(guild)
+		except _NoBackerRoleException:
+			self.log(guild, "Backer role configured but can't be retrieved")
+			return  # Bad id? Role removed?
+		if backer_role in member.roles:
+			return  # Already has role
+		username = self.__fetch_kickstarter_username(guild.id, username=member.name)
+		if username is None:
+			return  # Not on list
+		if username.lookup_status == _LookupStatus.misspelled_username:
+			self.log(guild, f"\u0007Member @{member.name} joined but their " \
+				"username was already manually attached to user id " \
+				f"{username.discord_member_id}")
+			return
+
+		self.__trace(guild, f"Member @{member.name} joined and is a backer. " \
+			"Granting backer role.")
+		await member.add_roles(backer_role)
+		username.lookup_status = _LookupStatus.member_found
+		username.discord_member_id = member.id
+		self.__update_kickstarter_username(username)
+
+	# -- UI callbacks -----
+
+	async def on_username_upload_submit(self,
+		interaction: Interaction,
+		attachment: Attachment
+	):
+		"""Callback for username upload modal."""
+		await interaction.response.defer(ephemeral=True, thinking=True)
+		guild = interaction.guild
+		import_result: _ImportResult = await self.__import_usernames(guild, attachment)
+		try:
+			sync_result: _SyncUsernamesResult = await self.__sync_by_username(guild)
+			text = f"{CONFIG['success_emoji']} Username import complete.\n" + \
+				import_result.summary_markdown() + "\n" + \
+				sync_result.summary_markdown()
+		except _NoBackerRoleException:
+			text = f"{CONFIG['failure_emoji']} Backer role must be configured " \
+				"before importing."
+		await interaction.followup.send(text, ephemeral=True)
+
+
+	# -- Operations -----
+
+	async def __import_usernames(self,
+		guild: Guild,
+		attachment: Attachment
+	) -> '_ImportResult':
+		"""Imports Discord usernames from an upload attachment."""
+		self.__trace(guild, f"Download - start - {attachment.filename} " \
+			f"({attachment.size:,} bytes, {attachment.content_type})")
+		file_bytes = await attachment.read()
+		file_str = file_bytes.decode('utf-8')
+		self.__trace(guild, "Download - complete")
+
+		self.__trace(guild, "Parse - start")
+		lines = file_str.splitlines(keepends=False)
+		usernames = []
+		malformed_count = 0
+		malformed_usernames = []
+		for line in lines:
+			username = self.__normalize_discord_username(line)
+			if (username.startswith('"') and username.endswith('"')) or \
+				(username.startswith("'") and username.endswith("'")):
+				# Remove quotes
+				username = username[1:-1].strip()
+			if username == '':
+				continue
+			if is_discord_username(username):
+				usernames.append(username)
+			else:
+				self.__trace(guild, f"Not a Discord username: \"{line}\"")
+				malformed_usernames.append(line)
+				malformed_count += 1
+		self.__trace(guild, f"Parse - complete - {len(lines):,} lines, {len(usernames):,} " \
+			f"valid usernames, {malformed_count:,} malformed usernames")
+
+		self.__trace(guild, f"Storing - start - {len(usernames):,} usernames")
+		new_username_count = self.__store_kickstarter_usernames(guild.id, usernames)
+		self.__trace(guild, f"Storing - complete - {new_username_count:,} unique " \
+			"usernames stored")
+
+		return _ImportResult(
+			attachment.filename,
+			attachment.size,
+			len(lines),
+			len(usernames),
+			malformed_count,
+			new_username_count,
+			malformed_usernames
+		)
+
+	async def __sync_by_username(self, guild: Guild) -> '_SyncUsernamesResult':
+		backer_role = await self.__fetch_backer_role(guild)
+		complete_count = 0
+		not_found_count = 0
+		self.__trace(guild, "Fetch usernames - start")
+		usernames: list[_KickstarterDiscordUser] = \
+			self.__fetch_incomplete_kickstarter_usernames(guild.id,
+				{
+					_LookupStatus.unprocessed,
+					_LookupStatus.username_not_found
+				})
+		self.__trace(guild, f"Fetch usernames - complete - found {len(usernames):,}")
+		username_to_member_id: dict[str, int] = {}
+		if self.bot.intents.members:
+			self.__trace(guild, "Fetching guild members from API - start")
+			async for member in guild.fetch_members(limit=None):
+				username_to_member_id[member.name] = member.id
+			self.__trace(guild, "Fetching guild members from API - complete - " \
+				f"got {len(username_to_member_id):,}")
+		async def username_loop_handler(username: _KickstarterDiscordUser):
+			nonlocal complete_count
+			nonlocal not_found_count
+			member_id = username_to_member_id.get(username.discord_username)
+			if member_id is not None:
+				member = guild.get_member(member_id) or \
+					await guild.fetch_member(member_id)
+			else:
+				member = guild.get_member_named(username.discord_username)
+			if member is None:
+				not_found_count += 1
+				if username.lookup_status != _LookupStatus.username_not_found:
+					username.lookup_status = _LookupStatus.username_not_found
+					self.__update_kickstarter_username(username)
+				return
+			username.lookup_status = _LookupStatus.member_found
+			username.discord_member_id = member.id
+			self.__update_kickstarter_username(username)
+			if backer_role not in member.roles:
+				await member.add_roles(backer_role)
+				complete_count += 1
+		self.__trace(guild, "Sync loop - start")
+		failure_count = await self.__throttled_loop(guild, usernames,
+			username_loop_handler, update_seconds=10.0)
+		self.__trace(guild, f"Sync loop - complete - {complete_count:,} completed, " \
+			f"{not_found_count:,} not found")
+		return _SyncUsernamesResult(complete_count, not_found_count, failure_count)
+
+	async def __throttled_loop(self,
+		guild: Guild,
+		iter: Iterable[_T],
+		callback: Callable[[_T], Awaitable[None]],
+		update_seconds: float | None = None
+	) -> int:
+		"""Iterates a loop with automatically adjusting sleeps based on
+		throttling exceptions."""
+		failure_count = 0
+		sleep_length = 0.0
+		start_time = now_timestamp()
+		last_update_time = start_time
+		for iter_count, elem in enumerate(iter):
+			complete = False
+			for _ in range(5):
+				try:
+					await sleep(sleep_length)
+					await callback(elem)
+					complete = True
+					break
+				except HTTPException as ex:
+					if ex.status == 429:  # rate limited
+						retry_header_value = ex.response.headers.get('retry_after')
+						retry_after_millis = float(retry_header_value or '1000')
+						self.__trace(guild, "Rate limited while processing. " \
+							f"retry_after={retry_header_value}")
+						await sleep(retry_after_millis / 1000.0)
+						sleep_length = 1.0 if sleep_length == 0.0 else sleep_length * 2.0
+						self.__trace(guild, f"Sleep increased to {sleep_length:,}s " \
+							"due to rate limiting")
+					else:
+						dump_stacktrace(ex)
+				except DiscordException as ex:
+					dump_stacktrace(ex)
+			if not complete:
+				failure_count += 1
+			if update_seconds is not None and now_timestamp() - last_update_time >= update_seconds:
+				self.__trace(guild, f"Completed {iter_count + 1:,} iterations")
+				last_update_time = now_timestamp()
+		return failure_count
+
+
+	# -- Database functions -----
+
+	def __store_kickstarter_username(self, record: '_KickstarterDiscordUser'):
+		cur = self.con.cursor()
+		cur.execute("""
+			INSERT OR IGNORE INTO kickstarter_discord_users (
+				guild_id,
+				discord_username,
+				discord_member_id,
+				lookup_status,
+				imported_at
+			) VALUES (
+				:guild_id,
+				:discord_username,
+				:discord_member_id,
+				:lookup_status,
+				:imported_at
+			)
+			""", {
+				'guild_id': record.guild_id,
+				'discord_username': self.__normalize_discord_username(record.discord_username),
+				'discord_member_id': record.discord_member_id,
+				'lookup_status': record.lookup_status,
+				'imported_at': record.imported_at
+			})
+		row_id = cur.lastrowid
+		self.con.commit()
+		cur.close()
+		if row_id is not None and row_id != 0:
+			record.pk = row_id
+
+	def __store_kickstarter_usernames(self, guild_id: int, usernames: list[str]) -> int:
+		imported_at: int = int(now_timestamp())
+		cur = self.con.cursor()
+		for username in usernames:
+			cur.execute("""
+				INSERT OR IGNORE INTO kickstarter_discord_users (
+					guild_id,
+					discord_username,
+					imported_at
+				) VALUES (
+					:guild_id,
+					:discord_username,
+					:imported_at
+				)
+				""", {
+					'guild_id': guild_id,
+					'discord_username': self.__normalize_discord_username(username),
+					'imported_at': imported_at
+				})
+		cur.execute("""
+			SELECT COUNT(1)
+			FROM kickstarter_discord_users
+			WHERE imported_at = ?
+			""", (imported_at, ))
+		imported_count = cur.fetchone()[0]
+		self.con.commit()
+		cur.close()
+		return imported_count
+
+	def __fetch_kickstarter_username(self,
+		guild_id: int,
+		*,
+		member_id: int | None = None,
+		username: str | None = None
+	) -> Optional['_KickstarterDiscordUser']:
+		"""Fetches an imported Discord username by EITHER member id or username
+		(must provide exactly one)"""
+		cur = self.con.cursor()
+		cur.row_factory = lambda c, r: _KickstarterDiscordUser(*r)
+		if member_id is not None:
+			cur.execute("""
+				SELECT *
+				FROM kickstarter_discord_users
+				WHERE
+					guild_id = :guild_id
+					AND discord_member_id = :member_id
+				""", { 'guild_id': guild_id, 'member_id': member_id })
+		elif username is not None:
+			cur.execute("""
+				SELECT *
+				FROM kickstarter_discord_users
+				WHERE
+					guild_id = :guild_id
+					AND discord_username = :username
+				""", { 'guild_id': guild_id, 'username': username })
+		ret_val = cur.fetchone()
+		cur.close()
+		return ret_val
+
+	def __fetch_incomplete_kickstarter_usernames(self,
+		guild_id: int,
+		statuses: set['_LookupStatus']
+	) -> list['_KickstarterDiscordUser']:
+		cur = self.con.cursor()
+		cur.row_factory = lambda c, r: _KickstarterDiscordUser(*r)
+		params = [ guild_id ] + list(statuses)
+		status_placeholders = ', '.join('?' * len(statuses))
+		sql = f"""
+			SELECT *
+			FROM kickstarter_discord_users
+			WHERE
+				guild_id = ?
+				AND lookup_status IN ({status_placeholders})
+			"""
+		cur.execute(sql, params)
+		ret_val = cur.fetchall()
+		cur.close()
+		return ret_val
+
+	def __update_kickstarter_username(self, user: '_KickstarterDiscordUser'):
+		cur = self.con.cursor()
+		cur.execute("""
+			UPDATE kickstarter_discord_users
+			SET discord_member_id = :discord_member_id,
+				lookup_status = :lookup_status
+			WHERE pk = :pk
+			""", {
+				'discord_member_id': user.discord_member_id,
+				'lookup_status': user.lookup_status,
+				'pk': user.pk
+			})
+		self.con.commit()
+		cur.close()
+
+	def __fetch_stats(self, guild_id: int) -> '_Stats':
+		"""Returns member link stats."""
+		cur = self.con.cursor()
+
+		cur.execute("""
+			SELECT
+				COUNT(1) AS total,
+				SUM(IIF(lookup_status = 0, 1, 0)) AS unprocessed_count,
+				SUM(IIF(lookup_status = 1, 1, 0)) AS not_found_count,
+				SUM(IIF(lookup_status = 2, 1, 0)) AS found_count,
+				MAX(imported_at) AS last_import
+			FROM kickstarter_discord_users
+			""")
+		(
+			username_count,
+			username_unprocessed_count,
+			username_not_found_count,
+			username_found_count,
+			username_last_imported_at
+		) = cur.fetchone()
+
+		cur.close()
+		return _Stats(
+			username_count,
+			username_unprocessed_count,
+			username_not_found_count,
+			username_found_count,
+			username_last_imported_at
+		)
+
+	def __find_nearest_usernames(self,
+		guild_id: int,
+		username: str,
+		limit: int = 10,
+		max_distance: int = 999
+	) -> list['_KickstarterDiscordUser']:
+		cur = self.con.cursor()
+		cur.row_factory = lambda c, r: _KickstarterDiscordUser(*r)
+		cur.execute("""
+			SELECT *
+			FROM kickstarter_discord_users
+			WHERE guild_id = ?
+				AND lookup_status IN (?, ?, ?)
+		""", (
+			guild_id,
+			_LookupStatus.unprocessed,
+			_LookupStatus.username_not_found,
+			_LookupStatus.misspelled_username,
+		))
+		records: list[_KickstarterDiscordUser] = cur.fetchall()
+		cur.close()
+
+		normal_username = KickstarterCog.__normalize_discord_username(username)
+		in_bounds_count = 0
+		def compute(r: _KickstarterDiscordUser) -> int:
+			nonlocal in_bounds_count
+			score = levenshtein(normal_username, r.discord_username)
+			if score <= max_distance:
+				in_bounds_count += 1
+			return score
+		closest = heapq.nsmallest(limit, records, key=compute)
+		if len(closest) > in_bounds_count:
+			closest = closest[:in_bounds_count]
+		return closest
+
+
+	# -- Utils -----
+
+	def __trace(self, guild: Guild, message: str):
+		self.log(guild, message)
+
+	@staticmethod
+	def __normalize_discord_username(username: str) -> str:
+		norm = username.lower().strip()
+		if norm.startswith('@'):
+			norm = norm[1:].strip()
+		return norm
+
+class _LookupStatus(IntEnum):
+	# Username imported but has not been looked up yet
+	unprocessed = 0
+	# Username was searched for in guild but not found (user might not have joined yet)
+	username_not_found = 1
+	# Username found in guild and member id stored
+	member_found = 2
+	# Mod manually linked this imported username to a Discord member to correct a username typo
+	misspelled_username = 3
+
+def _describe_lookup_status(status: _LookupStatus) -> str:
+	if status == _LookupStatus.unprocessed:
+		return 'unprocessed'
+	if status == _LookupStatus.username_not_found:
+		return 'member not in server'
+	if status == _LookupStatus.member_found:
+		return 'member found'
+	if status == _LookupStatus.misspelled_username:
+		return 'manually linked by mod'
+	return '-unknown-'
+
+class _KickstarterDiscordUser:
+	"""kickstarter_discord_users table row"""
+	def __init__(self,
+		pk: int,
+		guild_id: int,
+		discord_username: str,
+		discord_member_id: int | None,
+		lookup_status: _LookupStatus,
+		imported_at: int
+	):
+		self.pk: int = pk
+		self.guild_id: int = guild_id
+		self.discord_username: str = discord_username
+		"""Discord username provided in Kickstarter survey."""
+		self.discord_member_id: int | None = discord_member_id
+		"""Discord member ID when synced successfully or None when not yet linked."""
+		self.lookup_status: _LookupStatus = lookup_status
+		self.imported_at: int = imported_at
+		"""Unix timestamp when this record was first imported."""
+
+class _Stats:
+	def __init__(self,
+		username_count: int,
+		username_unprocessed_count: int,
+		username_not_found_count: int,
+		username_found_count: int,
+		username_last_imported_at: int
+	):
+		self.username_count: int = username_count
+		self.username_unprocessed_count: int = username_unprocessed_count
+		self.username_not_found_count: int = username_not_found_count
+		self.username_found_count: int = username_found_count
+		self.username_last_imported_at: int = username_last_imported_at
+
+class _NoBackerRoleException(BaseException):
+	pass
+
+class _ImportResult:
+	def __init__(self,
+		filename: str,
+		file_bytes: int,
+		line_count: int,
+		valid_record_count: int,
+		malformed_record_count: int,
+		new_record_count: int,
+		invalid_records: list[str]
+	):
+		self.filename: str = filename
+		"""Filename of the uploaded file."""
+		self.file_bytes: int = file_bytes
+		"""Size of uploaded file in bytes."""
+		self.line_count: int = line_count
+		"""Number of lines in the uploaded text file."""
+		self.valid_record_count: int = valid_record_count
+		"""How many records were valid."""
+		self.malformed_record_count: int = malformed_record_count
+		"""How many records were skipped because they were malformed."""
+		self.new_record_count: int = new_record_count
+		"""How many records were imported. May be less than valid_record_count
+		if some were already imported."""
+		self.invalid_records: list[str] = invalid_records
+		"""List of records that could not be imported. (May be partial if lots of failures.)"""
+
+	def summary_markdown(self) -> str:
+		lines: list[str] = []
+		if self.valid_record_count > 0:
+			lines.append(f"- Read {self.valid_record_count:,} valid records")
+		else:
+			lines.append("- Upload contained **no valid records**")
+		if self.malformed_record_count > 0:
+			lines.append(f"- Read **{self.malformed_record_count:,} malformed records**")
+		if self.new_record_count > 0:
+			lines.append(f"- Imported {self.new_record_count:,} new unique records")
+		else:
+			lines.append("- No new unique records (all previously imported)")
+		return "\n".join(lines)
+
+class _SyncUsernamesResult:
+	def __init__(self,
+		complete_count: int,
+		not_found_count: int,
+		failure_count: int
+	):
+		self.complete_count: int = complete_count
+		self.not_found_count: int = not_found_count
+		self.failure_count: int = failure_count
+
+
+	def summary_markdown(self) -> str:
+		lines: list[str] = []
+		if self.complete_count > 0:
+			lines.append(f"- Gave backer role to {self.complete_count:,} new members")
+		else:
+			lines.append("- No new members found")
+		if self.not_found_count > 0:
+			lines.append(f"- {self.not_found_count:,} members could not be found " \
+				"by the provided username")
+		if self.failure_count > 0:
+			lines.append(f"- Failed to link {self.failure_count:,} members")
+		return "\n".join(lines)
+
+class _UploadUsernamesModal(Modal):
+	upload_label = Label(
+		text='Discord user export',
+		description='Upload a plain text file containing one backer Discord username per line.',
+		component=FileUpload(
+			required=True,
+			min_values=1, max_values=1
+		)
+	)
+
+	def __init__(self):
+		super().__init__(title='Upload Discord Usernames', timeout=None)
+
+	# @override
+	async def on_submit(self, interaction: Interaction) -> None:
+		# noinspection PyTypeChecker
+		upload_input: FileUpload = self.upload_label.component
+		if len(upload_input.values) < 1:
+			text = f"{CONFIG['failure_emoji']} No export file included"
+			await interaction.response.send(text, ephemeral=True)
+			return
+		attachment = upload_input.values[0]
+		await KickstarterCog.shared.on_username_upload_submit(interaction, attachment)
+
+	# @override
+	async def on_error(self, interaction: Interaction, error: Exception) -> None:
+		dump_stacktrace(error)
+		try:
+			await interaction.response.send_message(
+				f'{CONFIG["failure_emoji"]} Upload failed :(',
+				ephemeral=True,
+			)
+		except DiscordException as e:
+			dump_stacktrace(e)

rocketbot/cogs/logcog.py

 """
 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,
 	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
 
 
 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
 		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
 		Called when a Member joins a Guild.
 
 		https://discordpy.readthedocs.io/en/stable/api.html#discord.on_member_join
+
+		Requires Members privileged intent.
 		"""
 		guild = member.guild
 		if not self.get_guild_setting(guild, self.SETTING_ENABLED):
 		will not be called, you may use on_raw_member_remove() instead.
 
 		https://discordpy.readthedocs.io/en/stable/api.html#discord.on_member_remove
+
+		Requires Members privileged intent.
 		"""
 		guild = member.guild
 		if not self.get_guild_setting(guild, self.SETTING_ENABLED):
 		* flags
 
 		https://discordpy.readthedocs.io/en/stable/api.html#discord.on_member_update
+
+		Requires Members privileged intent.
 		"""
 		guild = after.guild
 		if not self.get_guild_setting(guild, self.SETTING_ENABLED):
 		* discriminator
 
 		https://discordpy.readthedocs.io/en/stable/api.html#discord.on_user_update
+
+		Requires Members privileged intent.
 		"""
 		if hasattr(after, 'guild'):
 			guild = after.guild
 		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.
 
 		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
 		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.
 
 			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:
 		Called when a Message is created and sent.
 
 		https://discordpy.readthedocs.io/en/stable/api.html#discord.on_message
+
+		Content fields blank without MessageContent privileged intent.
 		"""
 		# print(f"on_message:"
 		# 	f"\n\tid: {message.id}"
 		* A call message has received an update to its participants or ending time.
 
 		https://discordpy.readthedocs.io/en/stable/api.html#discord.on_message_edit
+
+		Content fields blank without MessageContent privileged intent.
 		"""
 		guild = after.guild
 		if not self.get_guild_setting(guild, self.SETTING_ENABLED):
 		Discord embed server.
 
 		https://discordpy.readthedocs.io/en/stable/api.html#discord.on_raw_message_edit
+
+		Content fields blank without MessageContent privileged intent.
 		"""
 		if payload.cached_message:
 			return  # already handled by on_message_edit
 		RawMessageDeleteEvent.cached_message
 
 		https://discordpy.readthedocs.io/en/stable/api.html#discord.on_raw_message_delete
+
+		Content fields blank without MessageContent privileged intent.
 		"""
 		message = payload.cached_message
 		if message and message.author.id == self.bot.user.id:
 		RawBulkMessageDeleteEvent.cached_messages
 
 		https://discordpy.readthedocs.io/en/stable/api.html#discord.on_raw_bulk_message_delete
+
+		Content fields blank without MessageContent privileged intent.
 		"""
 		guild = self.bot.get_guild(payload.guild_id) or await self.bot.fetch_guild(payload.guild_id)
 		if not guild:
 		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():
 			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

 from datetime import datetime
 from typing import Optional
 
-from discord import Guild, Interaction, Member, Message
+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
 		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
 
 	# 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:
 	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)
 				f'updated to `{priority}`.',
 			ephemeral=True,
 		)
+
+	@classmethod
+	def supports_intents(cls, intents: Intents) -> bool:
+		return intents.message_content

rocketbot/cogs/urlspamcog.py

 from datetime import timedelta
 from typing import Literal
 
-from discord import Guild, Member, Message
+from discord import Guild, Intents, Member, Message
 from discord import utils as discordutils
 from discord.ext.commands import Cog
 from discord.utils import escape_markdown
 					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
 
 	def __find_urls(cls, text: str) -> list[str]:
 		p = re.compile(r'https?://\S+')
 		return re.findall(p, text)
+
+	@classmethod
+	def supports_intents(cls, intents: Intents) -> bool:
+		return intents.message_content

rocketbot/cogs/usernamecog.py

 """
 Cog for detecting username patterns.
 """
-from typing import Optional
 
-from discord import Guild, Interaction, Member
+from discord import Guild, Intents, Interaction, Member
 from discord.app_commands import Group
 from discord.ext.commands import Cog
 
 	"""
 	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]:
 		"""
 	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:
 		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)
 			context.ignored_by = reacted_by
 			self.log(context.member.guild, f'Warning ignored by {reacted_by.name}')
 			await bot_message.set_reactions(context.reactions())
+
+	@classmethod
+	def supports_intents(cls, intents: Intents) -> bool:
+		return intents.members

rocketbot/cogs/videopreviewcog.py

 import asyncio
 import json
 import re
-import subprocess
 from datetime import timedelta
+from typing import Any
 
-from discord import Message
+from discord import Guild, Intents, Message
 from discord.ext.commands import Cog
 
 from rocketbot.cogs.basecog import BaseCog
 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
 	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]/' \
 		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)
 
 		await asyncio.gather(*promises)
 
 	async def _fetch_preview(self, message: Message, link: MessageLink):
-		result = subprocess.run(
-			[
+		process = await asyncio.create_subprocess_exec(
 			'yt-dlp',
 			'--skip-download',
 			'--dump-single-json',
 			link.url,
-			],
-			stdout=subprocess.PIPE,
-			stderr=subprocess.PIPE,
-			universal_newlines=True
+			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:
 			content,
 			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

 """
 
 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
 
 	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
 		----------
 		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})'
 					text,
 					ephemeral=True,
 				)
-			except BaseException as e:
+			except DiscordException as e:
 				dump_stacktrace(e)
 		show_all_command = Command(
 			name='all',

rocketbot/collections.py

 		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

 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
 	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,
 		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: [],
 		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,
 	}
 	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,
 		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)
 					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:
+							possible_ch_types.isdisjoint(possible_token_types) and \
+							len(current_token) > 0:
 						tokens.append(current_token)
 						current_token = ''
 						possible_token_types |= all_token_types
 				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')
 		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

 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
 
 
 	# -- 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]:
 		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.
 		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
 	# -- 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]:
 		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.
 		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
 		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`
 		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

 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
 
 
 async def update_paged_content(
 		interaction: Interaction,
-		original_interaction: Optional[Interaction],
+		original_interaction: Interaction | None,
 		current_page: int,
 		pages: list[str],
 		**send_args,
 				ephemeral=True,
 				**send_args,
 			)
-	except BaseException as e:
+	except DiscordException as e:
 		dump_stacktrace(e)
 
 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

 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
 		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)
 			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)
 			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 '-] '
 __MENTION_REGEX: re.Pattern = re.compile('^<@[!&]([0-9]{17,20})>$')
 __USER_MENTION_REGEX: re.Pattern = re.compile('^<@!([0-9]{17,20})>$')
 __ROLE_MENTION_REGEX: re.Pattern = re.compile('^<@&([0-9]{17,20})>$')
+__EMAIL_REGEX: re.Pattern = re.compile(r'^(?:(?:[^<>()\[\]\\.,;:\s@"]+(?:\.[^<>()\[\]\\.,;:\s@"]+)*)|(?:".+"))@(?:(?:\[[0-9]{1,3}\.[0-9]{1,3}\.[0-9]{1,3}\.[0-9]{1,3}])|(?:(?:[a-zA-Z\-0-9]+\.)+[a-zA-Z]{2,}))$')
+__USERNAME_REGEX: re.Pattern = re.compile(r'[a-z0-9\._]{2,32}')
 
 def is_user_id(val: str) -> bool:
 	"""Tests if a string is in user/role ID format."""
 	"""Tests if a string is a user mention."""
 	return __USER_MENTION_REGEX.match(val) is not None
 
+def is_email_address(val: str) -> bool:
+	"""Tests if a string is a well-formed email address."""
+	return __EMAIL_REGEX.match(val) is not None
+
+def is_discord_username(val: str) -> bool:
+	"""Tests if a string is a properly formatted Discord username."""
+	return __USERNAME_REGEX.match(val.lower())
+
 def user_id_from_mention(mention: str) -> str:
 	"""Extracts the user ID from a mention. Raises a ValueError if malformed."""
 	m = __USER_MENTION_REGEX.match(mention)
 		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}>'
 
 
 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
 		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

+CREATE TABLE IF NOT EXISTS kickstarter_discord_users (
+	pk INTEGER PRIMARY KEY AUTOINCREMENT,
+	guild_id INTEGER NOT NULL,  -- Discord Guild.id
+	discord_username TEXT NOT NULL,  -- Discord Member.name
+	discord_member_id INTEGER DEFAULT NULL,  -- resolved member id
+	lookup_status INTEGER NOT NULL DEFAULT 0,  -- enum defined in code
+	imported_at INTEGER NOT NULL,  -- unix timestamp
+	UNIQUE(guild_id, discord_username COLLATE NOCASE) ON CONFLICT FAIL
+);
+CREATE INDEX IF NOT EXISTS idx_kickstarter_discord_users_discord_member_id
+	ON kickstarter_discord_users (discord_member_id);