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(f"- Last new import at ") 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 not yet attempted to 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"()? 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- 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): if await self.__check_disabled(interaction): return if await self.__check_configured(interaction): return await interaction.response.send_modal(_UploadUsernamesModal()) 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) 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) 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)