Adding /kickstarter reset command

rocketbot/cogs/kickstartercog.py

 import heapq
 from asyncio import sleep
-from collections.abc import Iterable
+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
+from typing import Optional, TypeVar
 
 from discord import Attachment, Guild, Interaction, Member, Role
 from discord.app_commands import Group, default_permissions
 	levenshtein,
 )
 
-_PERMITTED_GUILD_IDS = [
-	405011810937339905,  # prod server
-	900805482825007104,  # test server
-]
+_T = TypeVar('_T')
 class KickstarterCog(BaseCog):
 	"""
 	Assigns Discord users a role if their username appears in an imported list,
 		name='kickstarter',
 		description='Manages roles for users identified as Kickstarter backers.',
 		guild_only=True,
-		# guild_ids=_PERMITTED_GUILD_IDS,
 		default_permissions=MOD_PERMISSIONS
 	)
 
 			await interaction.followup.send(text, ephemeral=True)
 			return
 
-		closest_matches = self.__find_nearest_usernames(guild.id, member.name, limit=8, max_distance=3)
+		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." \
 				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. If one looks likely, you can fix it " \
-				f"with `/kickstarter link @{member.name} <username>`"
+				"Here are some similar ones."
 			for similar in closest_matches:
 				text += f"\n   - `{similar.discord_username}`"
 				if similar.lookup_status == _LookupStatus.misspelled_username:
 			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} {KickstarterCog.link.name} @{member.name} {closest_matches[0].discord_username}`?"
+				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
 
 			text = f"{CONFIG['failure_emoji']} That survey username is already " \
 				f"linked to <@{username.discord_member_id}>. Cannot be linked to " \
 				"another member.\n" \
-				"-# If all else fails, you can just manually give them the backer role."
+				f"- `/{KickstarterCog.kickstarter.name} {KickstarterCog.reset.name} " \
+				f"{username}` will unlink that Discord record" \
+				"- 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:
 		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- member_id: `{old_member_id}` --> `{record.discord_member_id}`"
+		if old_lookup_status is not None:
+			text += f"\n- lookup_status: `{_describe_lookup_status(old_lookup_status)}`" \
+				f" --> `{_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):
 
 	async def interaction_check(self, interaction: Interaction) -> bool:
 		if interaction.command is not None:
-			self.__trace(interaction.guild, f"@{interaction.user.name} used /{interaction.command.qualified_name}")
+			self.__trace(interaction.guild, f"@{interaction.user.name} used " \
+				f"/{interaction.command.qualified_name}")
 		return True
 
 
 		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.")
 
 	# -- UI callbacks -----
 
-	async def on_username_upload_submit(self, interaction: Interaction, attachment: Attachment):
+	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
 
 	# -- Operations -----
 
-	async def __import_usernames(self, guild: Guild, attachment: Attachment) -> '_ImportResult':
+	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})")
 			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, f"Fetching guild members from API - complete - got {len(username_to_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
 				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, 100)
-		self.__trace(guild, f"Sync loop - complete - {complete_count} completed, {not_found_count} not found")
+		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, callback, update_seconds: float | None = None) -> int:
+	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
 							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")
+						self.__trace(guild, f"Sleep increased to {sleep_length}s " \
+							"due to rate limiting")
 					else:
 						dump_stacktrace(ex)
 				except DiscordException as ex:
 
 	def __fetch_kickstarter_username(self,
 		guild_id: int,
+		*,
 		member_id: int | None = None,
 		username: str | None = None
 	) -> Optional['_KickstarterDiscordUser']:
 		cur.close()
 		return ret_val
 
-	def __fetch_incomplete_kickstarter_usernames(self, guild_id: int, statuses: set['_LookupStatus']) -> list['_KickstarterDiscordUser']:
+	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)
 			username_last_imported_at
 		)
 
-	def __find_nearest_usernames(self, guild_id: int, username: str, limit: int = 10, max_distance: int = 999) -> list['_KickstarterDiscordUser']:
+	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("""
 	# 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,