orochi-discord/orochi/bot.py

from functools import partial

import disnake
from disnake import CategoryChannel, PermissionOverwrite, TextChannel
from disnake.ext import commands
import logging

from orochi import http
from orochi.config import Config
from orochi.models import Game, GameState, Player, RoundVote, Vote, Round, RoundRoom, Room

bot = commands.Bot(command_prefix='!')


@bot.event
async def on_ready():
    config: Config = bot.config
    logger = bot.logger

    if config.guild is None:
        config.save()
        logger.error("The guild ID is missing")
        exit(1)

    guild = await bot.fetch_guild(config.guild)

    if not guild:
        logger.error("Unknown guild.")
        exit(1)

    if config.vote_category is None:
        category = await guild.create_category("Votes")
        config.vote_category = category.id
        config.save()

    if config.secret_category is None:
        category = await guild.create_category("Conversation⋅s secrète⋅s")
        config.secret_category = category.id
        config.save()

    vote_category: CategoryChannel = await guild.fetch_channel(config.vote_category)
    if vote_category is None:
        config.vote_category = None
        return await on_ready()

    secret_category: CategoryChannel = await guild.fetch_channel(config.secret_category)
    if secret_category is None:
        config.secret_category = None
        return await on_ready()

    await vote_category.set_permissions(
        guild.default_role, overwrite=PermissionOverwrite(read_message_history=False, read_messages=False)
    )

    await secret_category.set_permissions(
        guild.default_role, overwrite=PermissionOverwrite(read_message_history=False, read_messages=False)
    )

    for i, player in enumerate(Config.PLAYERS):
        player_id = player.lower()
        if player_id not in config.vote_channels:
            channel: TextChannel = await vote_category.create_text_channel(player_id)
            config.vote_channels[player_id] = channel.id
            config.save()

        channel: TextChannel = await guild.fetch_channel(config.vote_channels[player_id])
        if channel is None:
            del config.vote_channels[player_id]
            return await on_ready()

        await channel.edit(name=player_id, category=vote_category, position=i)
        await channel.set_permissions(
            guild.default_role, overwrite=PermissionOverwrite(read_message_history=False, read_messages=False)
        )

        if player_id not in config.player_roles:
            role = await guild.create_role(name=player)
            config.player_roles[player_id] = role.id
            config.save()

        guild = await bot.fetch_guild(guild.id)  # update roles
        role = guild.get_role(config.player_roles[player_id])
        if role is None:
            del config.player_roles[player_id]
            config.save()
            return await on_ready()

        await channel.set_permissions(
            role, overwrite=PermissionOverwrite(read_message_history=True, read_messages=True)
        )

    game = Game.load()
    if not game:
        game = Game()
        for player in config.PLAYERS:
            game.register_player(player, config.vote_channels[player.lower()])
        game.save()

    # Update private channel id if necessary
    for player in list(game.players.values()):
        if player.private_channel_id != config.vote_channels[player.name.lower()]:
            game.register_player(player.name, config.vote_channels[player.name.lower()])
            game.save()

    # Setup first round if not exists
    if not game.rounds:
        game.rounds.append(game.default_first_round())
        game.save()

    if not config.telepathy_channel:
        channel: TextChannel = await secret_category.create_text_channel("bigbrain")
        config.telepathy_channel = channel.id
        config.save()

    telepathy_channel: TextChannel = await guild.fetch_channel(config.telepathy_channel)
    if not telepathy_channel:
        config.telepathy_channel = None
        return await on_ready()

    await telepathy_channel.edit(name="bigbrain", category=secret_category, position=0,
                                 topic="Échanges télépathiques")
    await telepathy_channel.set_permissions(
        guild.default_role, overwrite=PermissionOverwrite(read_message_history=False, read_messages=False)
    )
    delphine = guild.get_role(config.player_roles['delphine'])
    philia = guild.get_role(config.player_roles['philia'])
    await telepathy_channel.set_permissions(
        delphine, overwrite=PermissionOverwrite(read_message_history=True, read_messages=True)
    )
    await telepathy_channel.set_permissions(
        philia, overwrite=PermissionOverwrite(read_message_history=True, read_messages=True)
    )

    if not config.brother_channel:
        channel: TextChannel = await secret_category.create_text_channel("doliprane")
        config.brother_channel = channel.id
        config.save()

    brother_channel: TextChannel = await guild.fetch_channel(config.brother_channel)
    if not brother_channel:
        config.brother_channel = None
        return await on_ready()

    await brother_channel.edit(name="doliprane", category=secret_category, position=1,
                               topic="Des voix dans la tête ...")
    await brother_channel.set_permissions(
        guild.default_role, overwrite=PermissionOverwrite(read_message_history=False, read_messages=False)
    )
    await brother_channel.set_permissions(
        philia, overwrite=PermissionOverwrite(read_message_history=True, read_messages=True)
    )

    brother_channel_webhook = None
    if config.brother_channel_webhook is not None:
        try:
            brother_channel_webhook = await bot.fetch_webhook(config.brother_channel_webhook)
        except disnake.HTTPException | disnake.NotFound | disnake.Forbidden:
            pass
    if brother_channel_webhook is None:
        brother_channel_webhook = await brother_channel.create_webhook(name="???")
        config.brother_channel_webhook = brother_channel_webhook.id
        config.save()

    if not config.backdoor_channel:
        channel: TextChannel = await secret_category.create_text_channel("backdoor")
        config.backdoor_channel = channel.id
        config.save()

    backdoor_channel: TextChannel = await guild.fetch_channel(config.backdoor_channel)
    if not backdoor_channel:
        config.backdoor_channel = None
        return await on_ready()

    await backdoor_channel.edit(name="backdoor", category=secret_category, position=2,
                                topic="Panel d'administrati0n du jeu")
    await backdoor_channel.set_permissions(
        guild.default_role, overwrite=PermissionOverwrite(read_message_history=False, read_messages=False)
    )
    dan = guild.get_role(config.player_roles['dan'])
    await backdoor_channel.set_permissions(
        dan, overwrite=PermissionOverwrite(read_message_history=True, read_messages=True)
    )

    if not config.sisters_channel:
        channel: TextChannel = await secret_category.create_text_channel("sista")
        config.sisters_channel = channel.id
        config.save()

    sisters_channel: TextChannel = await guild.fetch_channel(config.sisters_channel)
    if not sisters_channel:
        config.sisters_channel = None
        return await on_ready()

    await sisters_channel.edit(name="sista", category=secret_category, position=3,
                               topic="Discussions vestimentaires")
    await sisters_channel.set_permissions(
        guild.default_role, overwrite=PermissionOverwrite(read_message_history=False, read_messages=False)
    )
    nona = guild.get_role(config.player_roles['nona'])
    ennea = guild.get_role(config.player_roles['ennea'])
    await sisters_channel.set_permissions(
        nona, overwrite=PermissionOverwrite(read_message_history=True, read_messages=True)
    )
    await sisters_channel.set_permissions(
        ennea, overwrite=PermissionOverwrite(read_message_history=True, read_messages=True)
    )

    config.save()


@bot.command(help="Sauvegarde la partie")
@commands.has_permissions(administrator=True)
async def save(ctx: commands.Context, filename: str | None = None):
    Game.INSTANCE.save(filename)
    await ctx.reply("La partie a été sauvegardée.")


@bot.command(help="Recharger la partie")
@commands.has_permissions(administrator=True)
async def load(ctx: commands.Context, filename: str | None = None):
    game = Game.load(filename)
    if not game:
        return await ctx.reply("Une erreur est survenue : le fichier n'existe pas ?")
    await ctx.reply("La partie a été rechargée.")


@bot.command(help="Envoyer un message en tant qu'Orochi.")
@commands.has_permissions(administrator=True)
async def send(ctx: commands.Context, *, message: str):
    await ctx.message.delete()
    await ctx.send(message)


@bot.command(help="Supprime les derniers messages.")
@commands.has_permissions(administrator=True)
async def clear(ctx: commands.Context, limit: int = 100):
    await ctx.channel.purge(limit=limit)


@bot.command(help="Envoyer un message à Philia par la pensée en tant que Brother.")
@commands.has_permissions(administrator=True)
async def brother(ctx: commands.Context, *, message: str):
    webhook = await bot.fetch_webhook(bot.config.brother_channel_webhook)
    await webhook.send(message)
    await ctx.message.reply("Message envoyé.")


@bot.command(help="Ouvrir les votes")
@commands.has_permissions(administrator=True)
async def open(ctx: commands.Context):
    game: Game = Game.INSTANCE
    current_round = game.rounds[-1]

    if game.state == GameState.VOTING:
        await ctx.reply("Les votes sont déjà ouverts.")
        return
    elif game.state == GameState.RESULTS:
        await ctx.reply("Les votes viennent d'être fermés, merci de démarrer un nouveau tour avec `!prepare`.")
        return

    # Ensure that each room is configured
    for room in current_round.rooms:
        if room is None:
            await ctx.reply("Les salles ne sont pas configurées.")
        if len(list(room.players)) != 3 or not all(player for player in room.players):
            return await ctx.reply(f"La groupe {room.room.value} ne contient pas trois joueurs, "
                                   f"merci de finir sa configuration avec `!setup {room.room.value}`.")

    # Send messages to players
    for room in current_round.rooms:
        votes = list(room.votes)
        for i, vote in enumerate(votes):
            players = list(vote.players)
            other_vote = votes[1 - i]
            for j, player in enumerate(players):
                if player.score <= 0:
                    # Player is dead
                    continue

                other_player = players[1 - j] if len(players) == 2 else None
                view = VoteView(timeout=3600)
                channel_id = player.private_channel_id
                channel = bot.get_channel(channel_id)
                message = "Les votes sont ouverts.\n"
                effective_room = room.room if vote.player2 is not None else room.room.next
                message += f"Vous devez aller voter en salle **{effective_room.value}**.\n"
                if other_player:
                    message += f"Vous êtes allié⋅e avec **{other_player.name}**.\n"
                message += f"Vous affrontez {' et '.join(f'**{adv.name}**' for adv in other_vote.players)}.\n"
                message += "Bonne chance !"
                await channel.send(message)

                await channel.send("Pour voter, utilisez l'un des boutons ci-dessous. Vous pouvez appuyer "
                                   "sur le bouton plusieurs fois, mais seul le premier vote sera enregistré.",
                                   view=view)

    game.state = GameState.VOTING

    await ctx.reply("Les salles de vote sont ouvertes, les joueur⋅se⋅s peuvent désormais voter.")


@bot.command(help="Fermer les votes")
@commands.has_permissions(administrator=True)
async def close(ctx: commands.Context):
    game: Game = Game.INSTANCE
    if game.state != GameState.VOTING:
        await ctx.reply("Les votes ne sont pas ouverts.")
        return

    game.state = GameState.RESULTS

    current_round = game.rounds[-1]
    for room in current_round.rooms:
        for vote in room.votes:
            if vote.vote is None:
                vote.vote = Vote.ALLY
                await ctx.send(f"L'équipe **{' et '.join(player.name for player in vote.players)}** "
                               f"n'a pas voté en salle **{room.room.value}** et s'est alliée par défaut.")
            if vote.swapped:
                vote.vote = Vote.ALLY if vote.vote == Vote.BETRAY else Vote.BETRAY
                await ctx.send(f"L'équipe **{' et '.join(player.name for player in vote.players)}** "
                               f"dans le groupe **{room.room.value}** a vu son vote échangé. "
                               f"Nouveau vote : **{vote.vote.value}**")

    for player in game.players.values():
        channel = bot.get_channel(player.private_channel_id)
        await channel.send("Les votes sont à présent clos ! "
                           "Rendez-vous dans la salle principale pour découvrir les scores.")
        if player.score <= 0:
            if player.name == "Dan":
                await channel.send("Tiens ! Vous êtes m... Vous semblez bien tenir à la piqûre.")
            else:
                await channel.send("Tiens ! Vous êtes morts :)")
        elif player.score >= 9:
            await channel.send("Vous avez plus de 9 points. Vous pouvez désormais passer la porte 9.\n"
                               "Mais ... Attendrez-vous vos camarades ?")

    await ctx.reply("Les votes ont bien été fermés.")
    game.save()


@bot.command(help="Préparation du tour suivant")
@commands.has_permissions(administrator=True)
async def prepare(ctx: commands.Context):
    game: Game = Game.INSTANCE
    if game.state != GameState.RESULTS:
        await ctx.reply("Le tour actuel n'est pas terminé.")
        return

    game.state = GameState.PREPARING
    game.rounds.append(Round(
        round=len(game.rounds) + 1,
        room_a=RoundRoom(
            room=Room.A,
            vote1=RoundVote(),
            vote2=RoundVote(),
        ),
        room_b=RoundRoom(
            room=Room.B,
            vote1=RoundVote(),
            vote2=RoundVote(),
        ),
        room_c=RoundRoom(
            room=Room.C,
            vote1=RoundVote(),
            vote2=RoundVote(),
        ),
    ))
    game.save()

    await ctx.reply("Le tour suivant est en préparation. Utilisez `!setup A|B|C` pour paramétrer les groupes A, B ou C. "
                    "Dan peut faire la même chose.")


@bot.command(help="Prévisualisation des combats d'un tour")
@commands.has_any_role('IA', 'Dan')
async def preview(ctx: commands.Context):
    game: Game = Game.INSTANCE
    current_round = game.rounds[-1]
    for room in current_round.rooms:
        await ctx.send(f"Dans la groupe **{room.room.value}**, s'affronteront :\n"
                       f"- **{' et '.join(str(player or '_personne_') for player in room.vote1.players)}** "
                       f"(salle {room.room.next.value})\n"
                       f"- **{' et '.join(str(player or '_personne_') for player in room.vote2.players)}** "
                       f"(salle {room.room.value})")


@bot.command(help="Préparation d'une salle")
@commands.has_any_role('IA', 'Dan')
async def setup(ctx: commands.Context, room: Room):
    game: Game = Game.INSTANCE
    current_round = game.rounds[-1]

    if game.state != GameState.PREPARING:
        return await ctx.reply("Vous ne pouvez pas préparer le groupe avant le tour suivant.")

    await ctx.reply(f"Préparation du groupe {room.value}.")

    match room:
        case Room.A:
            round_room = current_round.room_a
        case Room.B:
            round_room = current_round.room_b
        case _:
            round_room = current_round.room_c

    view = PrepareRoomView(round_room, 0, timeout=300)
    await ctx.send(f"Veuillez choisir qui s'affrontera seul dans le groupe **{room.value}**.", view=view)

    if round_room.vote1.player1 is not None:
        await ctx.send(f"Attention : **{round_room.vote1.player1.name}** est déjà choisie.")


@bot.command(help="Falsification des votes")
@commands.has_permissions(administrator=True)
async def vote(ctx: commands.Context, player_name: str, vote: Vote | None):
    game: Game = Game.INSTANCE
    if game.state != GameState.VOTING:
        return await ctx.reply("Les votes ne sont pas ouverts.")

    for player in game.players.values():
        if player.name.lower() == player_name.lower():
            current_player = player
            break
    else:
        return await ctx.reply("Le joueur n'existe pas.")

    current_round = game.rounds[-1]
    for room in current_round.rooms:
        for v in room.votes:
            if current_player in v.players:
                v.vote = vote

    game.save()
    await ctx.reply(f"Le vote de **{current_player.name}** a bien été falsifié (nouveau vote : *{vote}**).")


@bot.command(help="Échange de vote")
@commands.has_any_role('IA', 'Dan')
async def swap(ctx: commands.Context, player_name: str):
    """
    Exchange the vote of the given player.
    """
    game: Game = Game.INSTANCE
    current_round = game.rounds[-1]

    if game.state != GameState.VOTING:
        return await ctx.reply("Les votes ne sont pas ouverts.")

    for player in game.players.values():
        if player.name.lower() == player_name.lower():
            current_player = player
            break
    else:
        return await ctx.reply("Le joueur n'existe pas.")

    for room in current_round.rooms:
        for v in room.votes:
            if current_player in v.players:
                v.swapped ^= True

    game.save()
    await ctx.reply(f"Le vote de **{current_player.name}** a bien été inversé. S'il était déjà inversé, "
                    "alors il est de retour à la normale.")


@bot.command(help="Modification du score d'un bracelet")
@commands.has_any_role('IA', 'Dan')
async def override(ctx: commands.Context, player_name: str, target_score: int):
    """
    Override the score of a given player.
    """
    game: Game = Game.INSTANCE

    if target_score <= 0:
        return await ctx.reply("Le score désiré doit valoir au moins 1.")

    for player in game.players.values():
        if player.name.lower() == player_name.lower():
            current_player = player
            break
    else:
        return await ctx.reply("Le joueur n'existe pas.")

    game.score_overrides[current_player] = target_score

    game.save()
    await ctx.reply(f"Le score de **{current_player.name}** a bien été modifié. "
                    f"Il sera maintenant de **{target_score}**.")


@bot.command(help="Reboot")
@commands.has_permissions(administrator=True)
async def reboot(ctx: commands.Context):
    game: Game = Game.INSTANCE
    for player in game.players.values():
        channel = bot.get_channel(player.private_channel_id)
        await channel.send("REB0OT.")

    game.save('game-prereboot.save')

    game = Game()
    for player in bot.config.PLAYERS:
        game.register_player(player, bot.config.vote_channels[player.lower()])
    game.rounds.append(game.default_first_round())
    game.save()

    await ctx.reply("REB0OT.")


class VoteView(disnake.ui.View):
    @disnake.ui.button(label="S'allier", style=disnake.ButtonStyle.green)
    async def ally(self, button: disnake.ui.Button, interaction: disnake.MessageInteraction):
        if Game.INSTANCE.state != GameState.VOTING:
            return await interaction.response.send_message("Les votes ne sont pas ouverts.", ephemeral=True)

        await interaction.response.send_message("Votre vote a bien été pris en compte.", ephemeral=True)

        await self.vote(interaction, Vote.ALLY)

    @disnake.ui.button(label="Trahir", style=disnake.ButtonStyle.red)
    async def betray(self, button: disnake.ui.Button, interaction: disnake.MessageInteraction):
        if Game.INSTANCE.state != GameState.VOTING:
            return await interaction.response.send_message("Les votes ne sont pas ouverts.", ephemeral=True)

        await interaction.response.send_message("Votre vote a bien été pris en compte.", ephemeral=True)

        await self.vote(interaction, Vote.BETRAY)

    async def vote(self, interaction: disnake.MessageInteraction, vote: Vote) -> None:
        game = Game.INSTANCE
        current_round = game.rounds[-1]
        current_player: Player | None = None
        for player in game.players.values():
            if player.private_channel_id == interaction.channel_id:
                current_player = player
                break

        if current_player.score <= 0 and current_player.name != "Dan":
            return await interaction.send("Vous êtes mort⋅e !")

        current_vote: RoundVote | None = None
        for room in current_round.rooms:
            for v in room.votes:
                if current_player in v.players:
                    current_vote = v
                    break
            else:
                continue
            break

        if current_vote.vote is None:
            current_vote.vote = vote
            game.save()


class PrepareRoomView(disnake.ui.View):
    def __init__(self, round_room: RoundRoom, player_id: int, *args, **kwargs):
        super().__init__(*args, **kwargs)
        assert 0 <= player_id < 3
        self.round_room = round_room
        self.player_id = player_id

        for player_name in Config.PLAYERS:
            async def choose_player(self, button: disnake.ui.Button, interaction: disnake.MessageInteraction):
                game: Game = Game.INSTANCE
                player = game.players[button.label]
                current_round = game.rounds[-1]
                for room in current_round.rooms:
                    for vote in room.votes:
                        if player in vote.players:
                            replaced_player = None
                            if room == self.round_room:
                                match self.player_id:
                                    case 0:
                                        replaced_player = room.vote1.player1
                                    case 1:
                                        replaced_player = room.vote2.player1
                                    case 2:
                                        replaced_player = room.vote2.player2

                            if replaced_player != player:
                                await interaction.send(
                                    f"Attention : **{player.name}** était déjà attribué⋅e dans le groupe "
                                    f"**{room.room.value}**. Vous devrez probablement la re-configurer.")

                        if vote.player1 == player:
                            vote.player1 = None
                        elif vote.player2 == player:
                            vote.player2 = None

                self.clear_items()
                await interaction.send("Choix bien pris en compte.")
                await interaction.edit_original_message(view=self)
                self.stop()

                match self.player_id:
                    case 0:
                        self.round_room.vote1.player1 = player

                        view = PrepareRoomView(self.round_room, 1, timeout=300)
                        await interaction.send(
                            f"**{player.name}** se battra seul⋅e. Veuillez désormais choisir contre qui "
                            "il ou elle se battra.", view=view)
                        if self.round_room.vote2.player1 is not None:
                            await interaction.send(
                                f"Attention : **{self.round_room.vote2.player1.name}** est déjà choisi⋅e."
                            )
                    case 1:
                        self.round_room.vote2.player1 = player

                        view = PrepareRoomView(self.round_room, 2, timeout=300)
                        await interaction.send(
                            f"**{player.name}** se battra contre **{self.round_room.vote1.player1.name}**. "
                            "Veuillez désormais choisir son partenaire.",
                            view=view)
                        if self.round_room.vote2.player2 is not None:
                            await interaction.send(
                                f"Attention : **{self.round_room.vote2.player2.name}** est déjà choisi⋅e."
                            )
                    case 2:
                        self.round_room.vote2.player2 = player
                        await interaction.send(
                            f"Dans le groupe **{round_room.room.value}**, **{round_room.vote1.player1.name}** "
                            f"(salle {round_room.room.next.value}) "
                            f"affrontera **{round_room.vote2.player1.name}** et **{round_room.vote2.player2.name}** "
                            f"(salle {round_room.room.value}).")
                        await interaction.send(
                            "Vous pouvez redéfinir le groupe tant que le tour n'est pas lancé. "
                            "Utilisez !preview pour un récapitulatif des tours.")

                game.save()

            game: Game = Game.INSTANCE
            current_round = game.rounds[-1]
            for room in current_round.rooms:
                for player in room.players:
                    if player is not None and player.name == player_name:
                        style = disnake.ButtonStyle.danger
                        break
                else:
                    continue
                break
            else:
                style = disnake.ButtonStyle.primary

            button = disnake.ui.Button(style=style, label=player_name)
            button.callback = partial(choose_player, self, button)
            button._view = self
            self.add_item(button)


def run():
    config = Config.load()

    http.run_web_server(config)

    logger = logging.getLogger('discord')
    logger.setLevel(logging.DEBUG)
    handler = logging.FileHandler(filename='../discord.log', encoding='utf-8', mode='w')
    handler.setFormatter(logging.Formatter('%(asctime)s:%(levelname)s:%(name)s: %(message)s'))
    logger.addHandler(handler)

    bot.config = config
    bot.logger = logger

    bot.run(config.discord_token)