Additional code jam management utilities

bot/exts/events/code_jams/__init__.py

@@ -0,0 +1,8 @@
+from bot.bot import Bot
+
+
+def setup(bot: Bot) -> None:
+    """Load the CodeJams cog."""
+    from bot.exts.events.code_jams._cog import CodeJams
+
+    bot.add_cog(CodeJams(bot))

bot/exts/events/code_jams/_channels.py

@@ -0,0 +1,113 @@
+import logging
+import typing as t
+
+import discord
+
+from bot.constants import Categories, Channels, Roles
+
+log = logging.getLogger(__name__)
+
+MAX_CHANNELS = 50
+CATEGORY_NAME = "Code Jam"
+
+
+async def _get_category(guild: discord.Guild) -> discord.CategoryChannel:
+    """
+    Return a code jam category.
+
+    If all categories are full or none exist, create a new category.
+    """
+    for category in guild.categories:
+        if category.name == CATEGORY_NAME and len(category.channels) < MAX_CHANNELS:
+            return category
+
+    return await _create_category(guild)
+
+
+async def _create_category(guild: discord.Guild) -> discord.CategoryChannel:
+    """Create a new code jam category and return it."""
+    log.info("Creating a new code jam category.")
+
+    category_overwrites = {
+        guild.default_role: discord.PermissionOverwrite(read_messages=False),
+        guild.me: discord.PermissionOverwrite(read_messages=True)
+    }
+
+    category = await guild.create_category_channel(
+        CATEGORY_NAME,
+        overwrites=category_overwrites,
+        reason="It's code jam time!"
+    )
+
+    await _send_status_update(
+        guild, f"Created a new category with the ID {category.id} for this Code Jam's team channels."
+    )
+
+    return category
+
+
+def _get_overwrites(
+        members: list[tuple[discord.Member, bool]],
+        guild: discord.Guild,
+) -> dict[t.Union[discord.Member, discord.Role], discord.PermissionOverwrite]:
+    """Get code jam team channels permission overwrites."""
+    team_channel_overwrites = {
+        guild.default_role: discord.PermissionOverwrite(read_messages=False),
+        guild.get_role(Roles.code_jam_event_team): discord.PermissionOverwrite(read_messages=True)
+    }
+
+    for member, _ in members:
+        team_channel_overwrites[member] = discord.PermissionOverwrite(
+            read_messages=True
+        )
+
+    return team_channel_overwrites
+
+
+async def create_team_channel(
+        guild: discord.Guild,
+        team_name: str,
+        members: list[tuple[discord.Member, bool]],
+        team_leaders: discord.Role
+) -> None:
+    """Create the team's text channel."""
+    await _add_team_leader_roles(members, team_leaders)
+
+    # Get permission overwrites and category
+    team_channel_overwrites = _get_overwrites(members, guild)
+    code_jam_category = await _get_category(guild)
+
+    # Create a text channel for the team
+    await code_jam_category.create_text_channel(
+        team_name,
+        overwrites=team_channel_overwrites,
+    )
+
+
+async def create_team_leader_channel(guild: discord.Guild, team_leaders: discord.Role) -> None:
+    """Create the Team Leader Chat channel for the Code Jam team leaders."""
+    category: discord.CategoryChannel = guild.get_channel(Categories.summer_code_jam)
+
+    team_leaders_chat = await category.create_text_channel(
+        name="team-leaders-chat",
+        overwrites={
+            guild.default_role: discord.PermissionOverwrite(read_messages=False),
+            team_leaders: discord.PermissionOverwrite(read_messages=True)
+        }
+    )
+
+    await _send_status_update(guild, f"Created {team_leaders_chat.mention} in the {category} category.")
+
+
+async def _send_status_update(guild: discord.Guild, message: str) -> None:
+    """Inform the events lead with a status update when the command is ran."""
+    channel: discord.TextChannel = guild.get_channel(Channels.code_jam_planning)
+
+    await channel.send(f"<@&{Roles.events_lead}>\n\n{message}")
+
+
+async def _add_team_leader_roles(members: list[tuple[discord.Member, bool]], team_leaders: discord.Role) -> None:
+    """Assign the team leader role to the team leaders."""
+    for member, is_leader in members:
+        if is_leader:
+            await member.add_roles(team_leaders)

bot/exts/events/code_jams/_cog.py

@@ -0,0 +1,235 @@
+import asyncio
+import csv
+import logging
+import typing as t
+from collections import defaultdict
+
+import discord
+from discord import Colour, Embed, Guild, Member
+from discord.ext import commands
+
+from bot.bot import Bot
+from bot.constants import Emojis, Roles
+from bot.exts.events.code_jams import _channels
+from bot.utils.services import send_to_paste_service
+
+log = logging.getLogger(__name__)
+
+TEAM_LEADERS_COLOUR = 0x11806a
+DELETION_REACTION = "\U0001f4a5"
+
+
+class CodeJams(commands.Cog):
+    """Manages the code-jam related parts of our server."""
+
+    def __init__(self, bot: Bot):
+        self.bot = bot
+
+    @commands.group(aliases=("cj", "jam"))
+    @commands.has_any_role(Roles.admins)
+    async def codejam(self, ctx: commands.Context) -> None:
+        """A Group of commands for managing Code Jams."""
+        if ctx.invoked_subcommand is None:
+            await ctx.send_help(ctx.command)
+
+    @codejam.command()
+    async def create(self, ctx: commands.Context, csv_file: t.Optional[str] = None) -> None:
+        """
+        Create code-jam teams from a CSV file or a link to one, specifying the team names, leaders and members.
+
+        The CSV file must have 3 columns: 'Team Name', 'Team Member Discord ID', and 'Team Leader'.
+
+        This will create the text channels for the teams, and give the team leaders their roles.
+        """
+        async with ctx.typing():
+            if csv_file:
+                async with self.bot.http_session.get(csv_file) as response:
+                    if response.status != 200:
+                        await ctx.send(f"Got a bad response from the URL: {response.status}")
+                        return
+
+                    csv_file = await response.text()
+
+            elif ctx.message.attachments:
+                csv_file = (await ctx.message.attachments[0].read()).decode("utf8")
+            else:
+                raise commands.BadArgument("You must include either a CSV file or a link to one.")
+
+            teams = defaultdict(list)
+            reader = csv.DictReader(csv_file.splitlines())
+
+            for row in reader:
+                member = ctx.guild.get_member(int(row["Team Member Discord ID"]))
+
+                if member is None:
+                    log.trace(f"Got an invalid member ID: {row['Team Member Discord ID']}")
+                    continue
+
+                teams[row["Team Name"]].append((member, row["Team Leader"].upper() == "Y"))
+
+            team_leaders = await ctx.guild.create_role(name="Code Jam Team Leaders", colour=TEAM_LEADERS_COLOUR)
+
+            for team_name, members in teams.items():
+                await _channels.create_team_channel(ctx.guild, team_name, members, team_leaders)
+
+            await _channels.create_team_leader_channel(ctx.guild, team_leaders)
+            await ctx.send(f"{Emojis.check_mark} Created Code Jam with {len(teams)} teams.")
+
+    @codejam.command()
+    @commands.has_any_role(Roles.admins)
+    async def end(self, ctx: commands.Context) -> None:
+        """
+        Delete all code jam channels.
+
+        A confirmation message is displayed with the categories and channels to be deleted.. Pressing the added reaction
+        deletes those channels.
+        """
+        def predicate_deletion_emoji_reaction(reaction: discord.Reaction, user: discord.User) -> bool:
+            """Return True if the reaction :boom: was added by the context message author on this message."""
+            return (
+                reaction.message.id == message.id
+                and user.id == ctx.author.id
+                and str(reaction) == DELETION_REACTION
+            )
+
+        # A copy of the list of channels is stored. This is to make sure that we delete precisely the channels displayed
+        # in the confirmation message.
+        categories = self.jam_categories(ctx.guild)
+        category_channels = {category: category.channels.copy() for category in categories}
+
+        confirmation_message = await self._build_confirmation_message(category_channels)
+        message = await ctx.send(confirmation_message)
+        await message.add_reaction(DELETION_REACTION)
+        try:
+            await self.bot.wait_for(
+                'reaction_add',
+                check=predicate_deletion_emoji_reaction,
+                timeout=10
+            )
+
+        except asyncio.TimeoutError:
+            await message.clear_reaction(DELETION_REACTION)
+            await ctx.send("Command timed out.", reference=message)
+            return
+
+        else:
+            await message.clear_reaction(DELETION_REACTION)
+            for category, channels in category_channels.items():
+                for channel in channels:
+                    await channel.delete(reason="Code jam ended.")
+                await category.delete(reason="Code jam ended.")
+
+            await message.add_reaction(Emojis.check_mark)
+
+    @staticmethod
+    async def _build_confirmation_message(
+        categories: dict[discord.CategoryChannel, list[discord.abc.GuildChannel]]
+    ) -> str:
+        """Sends details of the channels to be deleted to the pasting service, and formats the confirmation message."""
+        def channel_repr(channel: discord.abc.GuildChannel) -> str:
+            """Formats the channel name and ID and a readable format."""
+            return f"{channel.name} ({channel.id})"
+
+        def format_category_info(category: discord.CategoryChannel, channels: list[discord.abc.GuildChannel]) -> str:
+            """Displays the category and the channels within it in a readable format."""
+            return f"{channel_repr(category)}:\n" + "\n".join("  - " + channel_repr(channel) for channel in channels)
+
+        deletion_details = "\n\n".join(
+            format_category_info(category, channels) for category, channels in categories.items()
+        )
+
+        url = await send_to_paste_service(deletion_details)
+        if url is None:
+            url = "**Unable to send deletion details to the pasting service.**"
+
+        return f"Are you sure you want to delete all code jam channels?\n\nThe channels to be deleted: {url}"
+
+    @codejam.command()
+    @commands.has_any_role(Roles.admins, Roles.code_jam_event_team)
+    async def info(self, ctx: commands.Context, member: Member) -> None:
+        """
+        Send an info embed about the member with the team they're in.
+
+        The team is found by searching the permissions of the team channels.
+        """
+        channel = self.team_channel(ctx.guild, member)
+        if not channel:
+            await ctx.send(":x: I can't find the team channel for this member.")
+            return
+
+        embed = Embed(
+            title=str(member),
+            colour=Colour.blurple()
+        )
+        embed.add_field(name="Team", value=self.team_name(channel), inline=True)
+
+        await ctx.send(embed=embed)
+
+    @codejam.command()
+    @commands.has_any_role(Roles.admins)
+    async def move(self, ctx: commands.Context, member: Member, new_team_name: str) -> None:
+        """Move participant from one team to another by changing the user's permissions for the relevant channels."""
+        old_team_channel = self.team_channel(ctx.guild, member)
+        if not old_team_channel:
+            await ctx.send(":x: I can't find the team channel for this member.")
+            return
+
+        if old_team_channel.name == new_team_name or self.team_name(old_team_channel) == new_team_name:
+            await ctx.send(f"`{member}` is already in `{new_team_name}`.")
+            return
+
+        new_team_channel = self.team_channel(ctx.guild, new_team_name)
+        if not new_team_channel:
+            await ctx.send(f":x: I can't find a team channel named `{new_team_name}`.")
+            return
+
+        await old_team_channel.set_permissions(member, overwrite=None, reason=f"Participant moved to {new_team_name}")
+        await new_team_channel.set_permissions(
+            member,
+            overwrite=discord.PermissionOverwrite(read_messages=True),
+            reason=f"Participant moved from {old_team_channel.name}"
+        )
+
+        await ctx.send(
+            f"Participant moved from `{self.team_name(old_team_channel)}` to `{self.team_name(new_team_channel)}`."
+        )
+
+    @codejam.command()
+    @commands.has_any_role(Roles.admins)
+    async def remove(self, ctx: commands.Context, member: Member) -> None:
+        """Remove the participant from their team. Does not remove the participants or leader roles."""
+        channel = self.team_channel(ctx.guild, member)
+        if not channel:
+            await ctx.send(":x: I can't find the team channel for this member.")
+            return
+
+        await channel.set_permissions(
+            member,
+            overwrite=None,
+            reason=f"Participant removed from the team  {self.team_name(channel)}."
+        )
+        await ctx.send(f"Removed the participant from `{self.team_name(channel)}`.")
+
+    @staticmethod
+    def jam_categories(guild: Guild) -> list[discord.CategoryChannel]:
+        """Get all the code jam team categories."""
+        return [category for category in guild.categories if category.name == _channels.CATEGORY_NAME]
+
+    @staticmethod
+    def team_channel(guild: Guild, criterion: t.Union[str, Member]) -> t.Optional[discord.TextChannel]:
+        """Get a team channel through either a participant or the team name."""
+        for category in CodeJams.jam_categories(guild):
+            for channel in category.channels:
+                if isinstance(channel, discord.TextChannel):
+                    if (
+                        # If it's a string.
+                        criterion == channel.name or criterion == CodeJams.team_name(channel)
+                        # If it's a member.
+                        or criterion in channel.overwrites
+                    ):
+                        return channel
+
+    @staticmethod
+    def team_name(channel: discord.TextChannel) -> str:
+        """Retrieves the team name from the given channel."""
+        return channel.name.replace("-", " ").title()

bot/exts/filters/antimalware.py

@@ -7,7 +7,7 @@
 
 from bot.bot import Bot
 from bot.constants import Channels, Filter, URLs
-from bot.exts.utils.jams import CATEGORY_NAME as JAM_CATEGORY_NAME
+from bot.exts.events.code_jams._channels import CATEGORY_NAME as JAM_CATEGORY_NAME
 
 log = logging.getLogger(__name__)
 

bot/exts/filters/antispam.py

@@ -17,8 +17,8 @@
     Guild as GuildConfig, Icons,
 )
 from bot.converters import Duration
+from bot.exts.events.code_jams._channels import CATEGORY_NAME as JAM_CATEGORY_NAME
 from bot.exts.moderation.modlog import ModLog
-from bot.exts.utils.jams import CATEGORY_NAME as JAM_CATEGORY_NAME
 from bot.utils import lock, scheduling
 from bot.utils.messages import format_user, send_attachments
 

bot/exts/filters/filtering.py

@@ -19,8 +19,8 @@
     Channels, Colours, Filter,
     Guild, Icons, URLs
 )
+from bot.exts.events.code_jams._channels import CATEGORY_NAME as JAM_CATEGORY_NAME
 from bot.exts.moderation.modlog import ModLog
-from bot.exts.utils.jams import CATEGORY_NAME as JAM_CATEGORY_NAME
 from bot.utils.messages import format_user
 from bot.utils.regex import INVITE_RE
 from bot.utils.scheduling import Scheduler