discord-heidi/heidi_client.py

import asyncio
from configparser import ConfigParser
import os
from types import CoroutineType
from typing import Any, Callable, override
from discord import Message, VoiceState
from discord.activity import Activity
from discord.app_commands.tree import CommandTree
from discord.client import Client
from discord.enums import ActivityType, Status, StatusDisplayType
from discord.flags import Intents
from discord.member import Member
from discord.state import VoiceChannel

from heidi_config import USER_CONFIG_SCHEME, ConfigSection, FlagsConfigKey
from heidi_constants import CONFIGPATH, LINUS_GUILD, TEST_GUILD, USERCONFIGNAME
from heidi_helpers import create_author_based_message_predicate, play_voice_line_for_member


class HeidiClient(Client):
    def __init__(self, *, intents: Intents):
        client_activity: Activity = Activity(
            name="GNTM",
            url="https://www.joyn.de/serien/germanys-next-topmodel",
            type=ActivityType.competing,
            state="Nur eine kann es werden!",
            # details="Details",
            # platform="Platform",
            status_display_type=StatusDisplayType.state,
        )

        super().__init__(
            activity=client_activity,
            status=Status.online,
            intents=intents,
        )

        # Separate object that keeps all application command state
        self.tree: CommandTree = CommandTree(self)

        # Handle persistent user configuration
        self.user_config: ConfigParser = ConfigParser()
        if not os.path.exists(f"{CONFIGPATH}/{USERCONFIGNAME}"):
            # Create the file
            _ = open(f"{CONFIGPATH}/{USERCONFIGNAME}", "x")
        _ = self.user_config.read(f"{CONFIGPATH}/{USERCONFIGNAME}")
        self.update_to_default_user_config()
        self.print_user_config()

        # automatic actions on all messages
        # on_message_triggers is a map with tuples of two functions: (predicate, action)
        # the predicate receives the message as argument
        # if the predicate is true the action is performed
        self.on_message_triggers: dict[
            Callable[[Message], bool],
            Callable[[Message], CoroutineType[Any, Any, None]],
        ] = {
            # lambda m: m.author.nick.lower() in self.models.get_in_names(): self.autoreact_to_girls,
            create_author_based_message_predicate(["jeremy"]): self._autoreact_to_jeremy,
            create_author_based_message_predicate(["kardashian", "jenner"]): self._autoreact_to_kardashian,
        }

        # automatic actions on voice state changes
        # on_voice_state_triggers is a map with tuples of two functions: (predicate, action)
        # the predicate receives the member, before- and after-state as arguments
        # if the predicate is true, the action is performed
        self.on_voice_state_triggers: dict[
            Callable[[Member, VoiceState, VoiceState], bool],
            Callable[[Member, VoiceState, VoiceState], CoroutineType[Any, Any, None]],
        ] = {
            lambda member, before, after: before.channel != after.channel
            and after.channel is not None
            and isinstance(after.channel, VoiceChannel): self._play_entrance_sound,
        }

    # Synchronize commands to guilds
    @override
    async def setup_hook(self) -> None:
        self.tree.copy_global_to(guild=LINUS_GUILD)
        _ = await self.tree.sync(guild=LINUS_GUILD)

        self.tree.copy_global_to(guild=TEST_GUILD)
        _ = await self.tree.sync(guild=TEST_GUILD)

    def update_to_default_user_config(self) -> None:
        """
        Adds config keys to the config, if they don't exist yet.
        This writes the user config file.
        """

        for section, keys in USER_CONFIG_SCHEME.items():
            if section not in self.user_config:
                print(f"Adding section {section} to {CONFIGPATH}/{USERCONFIGNAME}")
                self.user_config[section] = dict()

            for key, default in keys:
                if key not in self.user_config[section].keys():
                    print(f"Adding key {key} with default value {default} to section {section}")
                    self.user_config[section][key] = default

        self.write_user_config()

    def print_user_config(self) -> None:
        """
        Print the current user config from memory.
        This does not read the user config file.
        """
        print("Heidi User Config:\n")

        for section in self.user_config.sections():
            print(f"[{section}]")
            for key in self.user_config[section]:
                print(f"{key}={self.user_config[section][key]}")

        print("")

    def write_user_config(self) -> None:
        """
        Write the current configuration to disk.
        """
        if not os.path.exists(f"{CONFIGPATH}/{USERCONFIGNAME}"):
            print(f"Error: {CONFIGPATH}/{USERCONFIGNAME} doesn't exist!")
            return

        print(f"Writing {CONFIGPATH}/{USERCONFIGNAME}")

        with open(f"{CONFIGPATH}/{USERCONFIGNAME}", "w") as file:
            self.user_config.write(file)

    # Automatic Actions ------------------------------------------------------------------------------

    @staticmethod
    async def _autoreact_to_jeremy(message: Message) -> None:
        """
        🧀 Jeremy.
        This function is set in on_message_triggers and triggered by the on_message event.
        """
        await message.add_reaction("🧀")

    @staticmethod
    async def _autoreact_to_kardashian(message: Message) -> None:
        """
        💄 Kardashian.
        This function is set in on_message_triggers and triggered by the on_message event.
        """
        await message.add_reaction("💄")

    async def _play_entrance_sound(
        self,
        member: Member,
        before: VoiceState,
        after: VoiceState,
    ) -> None:
        """
        Play a sound when a member joins a voice channel (and another member is present).
        This function is set in on_voice_state_triggers and triggered by the on_voice_state_update event.
        """

        disable_join_sound_if_alone: bool = self.user_config[ConfigSection.FLAGS.value][
            FlagsConfigKey.DISABLE_JOIN_SOUND_IF_ALONE.value
        ] == str(True)

        # Don't play anything when no other users are present
        if (
            disable_join_sound_if_alone
            and member.voice is not None
            and member.voice.channel is not None
            and len(member.voice.channel.members) <= 1
        ):
            print("Not playing entrance sound, as no other members are present")
            return

        soundpath: str | None = self.user_config["ENTRANCE.SOUND"].get(member.name, None)

        if soundpath is None:
            print(f"User {member.name} has not set an entrance sound")
            return

        board, sound = soundpath.split("/")

        # Wait a bit to not have simultaneous joins
        await asyncio.sleep(1)

        await play_voice_line_for_member(None, member, board, sound)