Added a clash of code module

requirements.txt

@@ -3,3 +3,4 @@ psycopg2-binary==2.9.3
 python-dotenv==0.20.0
 PyYAML==6.0
 SQLAlchemy==1.4.37
+requests==2.30.0

vesta/__init__.py

@@ -7,7 +7,7 @@
 from .lang import Lang
 from yaml import load, Loader
 with open("vesta/data/lang.yml") as file:
-    lang = Lang(load(file.read(), Loader), session_maker)
+    lang_file = Lang(load(file.read(), Loader), session_maker)
 
 from .client import Vesta
 

vesta/client.py

@@ -6,6 +6,7 @@
 from discord import app_commands
 
 from . import session_maker
+from .services import clash_of_code_helper
 from .tables import CustomCommand, select
 
 logger = logging.getLogger(__name__)
@@ -27,6 +28,8 @@ def __init__(self, *, intents: discord.Intents):
     async def on_ready(self):
         logger.info(f"Logged on as {self.user}")
 
+        clash_of_code_helper.resume_update_loops()
+
         for com in self.tree.get_commands():
             logger.debug(f"Globals {com} name : {com.name}")
 
@@ -56,13 +59,21 @@ async def command(interaction: discord.Interaction):
             if active:
                 await self.tree.sync(guild=guild)
 
-    async def on_member_join(self, member):
+    async def on_member_join(self, member: discord.Member):
         logger.debug(f"Member joined : {member}!")
+        if not (await member.guild.fetch_member(self.user.id)).guild_permissions.manage_nicknames:
+            logger.debug(f"Bot doesn't have manage nicknames permission on {member.guild}")
+            return
+
         if not re.match(regex_name, member.display_name):
             await member.edit(nick=f"{random.choice(names).capitalize()}{random.choice(adjectives).capitalize()}")
 
-    async def on_member_update(self, before, after):
+    async def on_member_update(self, before: discord.Member, after: discord.Member):
         logger.debug(f"Member update : {after}!")
+        if not (await after.guild.fetch_member(self.user.id)).guild_permissions.manage_nicknames:
+            logger.debug(f"Bot doesn't have manage nicknames permission on {after.guild}")
+            return
+
         if not after.guild_permissions.manage_nicknames and not re.match(regex_name, after.display_name):
             await after.edit(nick=f"{random.choice(names).capitalize()}{random.choice(adjectives).capitalize()}")
 

vesta/commands/__init__.py

@@ -2,3 +2,4 @@
 from . import presentation
 from . import custom
 from . import config
+from . import clash_of_code

vesta/commands/clash_of_code.py

@@ -0,0 +1,128 @@
+import logging
+import re
+from typing import Tuple
+
+import discord
+from discord import app_commands
+from sqlalchemy import select
+
+from .. import vesta_client, session_maker, lang_file
+from ..exceptions import CommandRuntimeError
+from ..services import clash_of_code_helper, State
+from ..services.clash_of_code_helper import start_update_loop
+from ..tables import ClashOfCodeGuildGame
+from ..tables import Guild
+
+logger = logging.getLogger(__name__)
+session = session_maker()
+
+regex_clash_of_code_game = r"^(https://|)(www.|)codingame.com/clashofcode/clash/[^/]+(/|)$"
+
+
+@vesta_client.tree.command(name="clash-of-code", description="Invites users with the \"Clash of Code\" role to play")
+@app_commands.describe(link="The link to the Clash of Code game")
+async def clash_of_code(interaction: discord.Interaction, link: str):
+    if not re.match(regex_clash_of_code_game, link):
+        await _send_error(interaction, "coc_invalid_link")
+        return
+
+    try:
+        (game, game_id) = _get_game(link)
+        guild_game = _get_guild_game(interaction)
+        (guild, role, channel) = _get_guild(interaction)
+    except CommandRuntimeError as e:
+        await _send_error(interaction, e.message)
+        return
+
+    view = discord.ui.View()
+    view.add_item(discord.ui.Button(
+        label=lang_file.get("coc_game_join", interaction.guild),
+        url=game.link,
+        emoji="🎮"
+    ))
+
+    embed = game.embed(interaction.guild)
+    embed.set_author(name=interaction.user.name, icon_url=interaction.user.avatar.url)
+
+    announcement_message = await channel.send(
+        content=f"{role.mention} {lang_file.get('coc_game_invite', interaction.guild)}",
+        embed=embed,
+        view=view
+    )
+
+    guild_game.last_clash_id = game_id
+    guild_game.announcement_message_id = announcement_message.id
+    session.commit()
+
+    await interaction.response.send_message(
+        lang_file.get("coc_successfully_invited", interaction.guild),
+        ephemeral=True
+    )
+
+    start_update_loop(message=announcement_message, guild=interaction.guild)
+
+
+def _get_game(link: str):
+    game_id = clash_of_code_helper.game_id_from_link(link)
+    game = clash_of_code_helper.fetch(game_id)
+
+    if not game:
+        raise CommandRuntimeError("coc_invalid_link")
+    if game.state != State.PENDING:
+        raise CommandRuntimeError("coc_game_already_started")
+
+    return game, game_id
+
+
+def _get_guild_game(interaction: discord.Interaction) -> ClashOfCodeGuildGame:
+    r = select(ClashOfCodeGuildGame).where(ClashOfCodeGuildGame.guild_id == interaction.guild.id)
+    guild_game: ClashOfCodeGuildGame = session.scalar(r)
+
+    if guild_game and not guild_game.can_start_new():
+        raise CommandRuntimeError("coc_already_in_progress")
+
+    if not guild_game:
+        logger.debug(f"Creating new guild game for guild {interaction.guild_id}")
+        guild_game = ClashOfCodeGuildGame(guild_id=interaction.guild_id)
+        session.add(guild_game)
+
+    return guild_game
+
+
+def _get_guild(interaction: discord.Interaction) -> Tuple[Guild, discord.Role, discord.TextChannel]:
+    r = select(Guild).where(Guild.id == interaction.guild_id)
+    guild: Guild = session.scalar(r)
+
+    if not guild:
+        logger.debug(f"Add guild {interaction.guild_id} to the database")
+        guild = Guild(id=interaction.guild_id, name=interaction.guild.name)
+        session.add(guild)
+
+    return guild, _get_role(guild, interaction), _get_channel(guild, interaction)
+
+
+def _get_role(guild: Guild, interaction: discord.Interaction) -> discord.Role:
+    if not guild.coc_role:
+        raise CommandRuntimeError("coc_role_not_set")
+
+    role = interaction.guild.get_role(guild.coc_role)
+    if not role:
+        raise CommandRuntimeError("coc_role_not_found")
+
+    return role
+
+
+def _get_channel(guild: Guild, interaction: discord.Interaction) -> discord.TextChannel:
+    if not guild.coc_channel:
+        raise CommandRuntimeError("coc_channel_not_set")
+
+    channel = interaction.guild.get_channel(guild.coc_channel)
+    if not channel:
+        raise CommandRuntimeError("coc_channel_not_found")
+
+    return channel
+
+
+async def _send_error(interaction: discord.Interaction, key: str):
+    msg = lang_file.get(key, interaction.guild)
+    await interaction.response.send_message(f"❌ {msg}", ephemeral=True)

vesta/commands/config.py

@@ -1,9 +1,11 @@
+import typing
+
 import discord
 from discord import app_commands
 import logging
 import traceback
 
-from .. import vesta_client, session_maker, lang
+from .. import vesta_client, session_maker, lang_file
 from ..tables import Guild, select
 
 logger = logging.getLogger(__name__)
@@ -17,14 +19,14 @@ async def on_error(self, interaction: discord.Interaction, error):
         logger.debug(f"Error {error} raised")
         if isinstance(error, app_commands.errors.MissingPermissions):
             await interaction.response.send_message(
-                lang.get("permissions_error", interaction.guild), ephemeral=True)
+                lang_file.get("permissions_error", interaction.guild), ephemeral=True)
         elif isinstance(error, app_commands.errors.BotMissingPermissions):
             await interaction.response.send_message(
-                lang.get("bot_permissions_error", interaction.guild) + f" {', '.join(error.missing_permissions)}",
+                lang_file.get("bot_permissions_error", interaction.guild) + f" {', '.join(error.missing_permissions)}",
                 ephemeral=True)
         else:
             logger.error(traceback.format_exc())
-            await interaction.response.send_message(lang.get("unexpected_error", interaction.guild), ephemeral=True)
+            await interaction.response.send_message(lang_file.get("unexpected_error", interaction.guild), ephemeral=True)
 
 
 config_manager = ConfigManager()
@@ -35,76 +37,79 @@ async def on_error(self, interaction: discord.Interaction, error):
 async def review(interaction: discord.Interaction, channel: discord.TextChannel):
     logger.debug(f"Command /config review {channel} used")
 
-    r = select(Guild).where(Guild.id == interaction.guild_id)
-    guild = session.scalar(r)
-    if not guild:
-        logger.debug(f"Add guild {interaction.guild_id} to the database")
-        guild = Guild(id=interaction.guild_id, name=interaction.guild.name)
-        session.add(guild)
+    def update(g):
+        g.review_channel = channel.id
+    await update_config_element(interaction, update)
 
-    guild.review_channel = channel.id
-
-    try:
-        session.commit()
-    except:
-        session.rollback()
-
-        logger.error(traceback.format_exc())
-        return await interaction.response.send_message(lang.get("unexpected_error", interaction.guild), ephemeral=True)
-
-    await interaction.response.send_message(lang.get("config_review", interaction.guild), ephemeral=True)
+    await interaction.response.send_message(lang_file.get("config_review", interaction.guild), ephemeral=True)
 
 
 @config_manager.command(description="Set Projets Channel")
 @app_commands.describe(channel="The future projects channel")
 async def projects(interaction: discord.Interaction, channel: discord.TextChannel):
     logger.debug(f"Command /config projects {channel} used")
 
-    r = select(Guild).where(Guild.id == interaction.guild_id)
-    guild = session.scalar(r)
-    if not guild:
-        logger.debug(f"Add guild {interaction.guild_id} to the database")
-        guild = Guild(id=interaction.guild_id, name=interaction.guild.name)
-        session.add(guild)
+    def update(g):
+        g.projects_channel = channel.id
+    await update_config_element(interaction, update)
 
-    guild.projects_channel = channel.id
+    await interaction.response.send_message(lang_file.get("config_projects", interaction.guild), ephemeral=True)
 
-    try:
-        session.commit()
-    except:
-        session.rollback()
 
-        logger.error(traceback.format_exc())
-        return await interaction.response.send_message(lang.get("unexpected_error", interaction.guild), ephemeral=True)
+@config_manager.command(name="coc-role", description="Set clash of code ping role")
+@app_commands.rename(coc_role="role")
+@app_commands.describe(coc_role="The role to ping when a game of Clash of Code is launched")
+async def change_clash_of_code_role(interaction: discord.Interaction, coc_role: discord.Role):
+    logger.debug(f"Command /config coc-role {coc_role.name} used")
+
+    def update(g):
+        g.coc_role = coc_role.id
+    await update_config_element(interaction, update)
+
+    await interaction.response.send_message(lang_file.get("config_coc_role", interaction.guild), ephemeral=True)
+
+@config_manager.command(name="coc-channel", description="Set clash of code channel")
+@app_commands.rename(coc_channel="channel")
+@app_commands.describe(coc_channel="The channel to send clash of code messages")
+async def change_clash_of_code_channel(interaction: discord.Interaction, coc_channel: discord.TextChannel):
+    logger.debug(f"Command /config coc-channel {coc_channel.name} used")
 
-    await interaction.response.send_message(lang.get("config_projects", interaction.guild), ephemeral=True)
+    def update(g):
+        g.coc_channel = coc_channel.id
+    await update_config_element(interaction, update)
 
+    await interaction.response.send_message(lang_file.get("config_coc_channel", interaction.guild), ephemeral=True)
 
 @config_manager.command(name="lang", description="Set Guild Lang")
 @app_commands.rename(guild_lang='lang')
 @app_commands.describe(guild_lang="The lang for the bot")
-@app_commands.choices(guild_lang=[app_commands.Choice(name=l, value=l) for l in lang.data])
+@app_commands.choices(guild_lang=[app_commands.Choice(name=l, value=l) for l in lang_file.data])
 async def change_lang(interaction: discord.Interaction, guild_lang: app_commands.Choice[str]):
     logger.debug(f"Command /config lang {guild_lang.value} used")
 
+    def update(g):
+        g.lang = guild_lang.value
+    await update_config_element(interaction, update)
+
+    await interaction.response.send_message(lang_file.get("config_lang", interaction.guild), ephemeral=True)
+
+async def update_config_element(interaction: discord.Interaction, updater: typing.Callable[[Guild], None]):
     r = select(Guild).where(Guild.id == interaction.guild_id)
     guild = session.scalar(r)
     if not guild:
         logger.debug(f"Add guild {interaction.guild_id} to the database")
         guild = Guild(id=interaction.guild_id, name=interaction.guild.name)
         session.add(guild)
 
-    guild.lang = guild_lang.value
+    updater(guild)
 
     try:
         session.commit()
     except:
         session.rollback()
 
         logger.error(traceback.format_exc())
-        return await interaction.response.send_message(lang.get("unexpected_error", interaction.guild), ephemeral=True)
-
-    await interaction.response.send_message(lang.get("config_lang", interaction.guild), ephemeral=True)
-
+        return await interaction.response.send_message(lang_file.get("unexpected_error", interaction.guild), ephemeral=True)
+    pass
 
 vesta_client.tree.add_command(config_manager)