diff --git a/bot.py b/bot.py index 2968a2b..6d162f2 100644 --- a/bot.py +++ b/bot.py @@ -1,26 +1,35 @@ # Example: https://github.com/Rapptz/discord.py/blob/master/examples/app_commands/basic.py -from ast import Call -import random, logging -from discord import DMChannel +import os +import random +import logging +from types import CoroutineType +from typing import Any, Callable, override +from discord import app_commands, ui from discord.app_commands import Choice -from typing import Awaitable, Dict, List, Optional, Union, Callable, Any -from rich.traceback import install -from heidi_client import * +from discord.client import Client +from discord.components import SelectOption +from discord.enums import ButtonStyle +from discord.flags import Intents +from discord.interactions import Interaction +from discord.member import Member, VoiceState +from discord.message import Message -# Install rich traceback -install(show_locals=True) +from heidi_client import HeidiClient +from heidi_config import ConfigSection, FlagsConfigKey +from heidi_constants import DOCKER, HEIDI_SPAM_ID, SOUNDDIR +from heidi_helpers import enforce_channel, play_voice_line_for_member -# @todo yt-dlp music support +# TODO: yt-dlp music support # Log to file handler = logging.FileHandler(filename="discord.log", encoding="utf-8", mode="w") # Intents specification is no longer optional -intents = discord.Intents.default() +intents: Intents = Intents.default() intents.members = True # Allow to react to member join/leave etc intents.message_content = True # Allow to read message content from arbitrary messages intents.voice_states = True # Allow to process on_voice_state_update @@ -63,9 +72,7 @@ async def on_message(message: Message) -> None: @client.event -async def on_voice_state_update( - member: Member, before: VoiceState, after: VoiceState -) -> None: +async def on_voice_state_update(member: Member, before: VoiceState, after: VoiceState) -> None: """ This event triggers when a member joins/changes/leaves a voice channel or mutes/unmutes. """ @@ -77,7 +84,9 @@ async def on_voice_state_update( # python iterates over the keys of a map for predicate in client.on_voice_state_triggers: if predicate(member, before, after): - action: Callable = client.on_voice_state_triggers[predicate] + action: Callable[[Member, VoiceState, VoiceState], CoroutineType[Any, Any, None]] = ( + client.on_voice_state_triggers[predicate] + ) print(f"on_voice_state_update: calling {action.__name__}") await action(member, before, after) @@ -85,114 +94,194 @@ async def on_voice_state_update( # Config Commands -------------------------------------------------------------------------------- -class EntranceSoundSoundSelect(discord.ui.Select): - def __init__(self, board: str, on_sound_select_callback): - self.board = board - self.on_sound_select_callback = on_sound_select_callback - - options: List[discord.SelectOption] = [ - discord.SelectOption(label=sound.split(".")[0], value=sound) - for sound in os.listdir(f"{SOUNDDIR}/{board}") - ] - - super().__init__( - placeholder="Select Sound", min_values=1, max_values=1, options=options +class FlagValueSelect(ui.Select[ui.View]): + def __init__( + self, flag: str, on_flag_select_callback: Callable[[Interaction, str, str], CoroutineType[Any, Any, None]] + ) -> None: + self.flag: str = flag + self.on_flag_select_callback: Callable[[Interaction, str, str], CoroutineType[Any, Any, None]] = ( + on_flag_select_callback ) - async def callback(self, interaction: Interaction): - await self.on_sound_select_callback(interaction, self.board, self.values[0]) + options: list[SelectOption] = [SelectOption(label=val, value=val) for val in ["True", "False"]] + + super().__init__(placeholder="Select Value", min_values=1, max_values=1, options=options) + + @override + async def callback(self, interaction: Interaction) -> None: + await self.on_flag_select_callback(interaction, self.flag, self.values[0]) -class EntranceSoundSoundView(discord.ui.View): - def __init__(self, board: str, on_sound_select_callback): +class FlagValueView(ui.View): + def __init__( + self, flag: str, on_flag_select_callback: Callable[[Interaction, str, str], CoroutineType[Any, Any, None]] + ) -> None: super().__init__(timeout=600) - self.add_item(EntranceSoundSoundSelect(board, on_sound_select_callback)) + _ = self.add_item(FlagValueSelect(flag, on_flag_select_callback)) -class EntranceSoundBoardSelect(discord.ui.Select): - def __init__(self, on_sound_select_callback): - self.on_sound_select_callback = on_sound_select_callback - - options: List[discord.SelectOption] = [ - discord.SelectOption(label=board, value=board) - for board in os.listdir(f"{SOUNDDIR}") - ] - - super().__init__( - placeholder="Select Board", min_values=1, max_values=1, options=options +class FlagsSelect(ui.Select[ui.View]): + def __init__( + self, on_flag_select_callback: Callable[[Interaction, str, str], CoroutineType[Any, Any, None]] + ) -> None: + self.on_flag_select_callback: Callable[[Interaction, str, str], CoroutineType[Any, Any, None]] = ( + on_flag_select_callback ) - async def callback(self, interaction: Interaction): - await interaction.response.send_message( - f"Welchen sound willst du?", + options: list[SelectOption] = [SelectOption(label=flag.value, value=flag.value) for flag in FlagsConfigKey] + + super().__init__(placeholder="Select Flag", min_values=1, max_values=1, options=options) + + @override + async def callback(self, interaction: Interaction) -> None: + _ = await interaction.response.send_message( + "Welchen Wert willst du setzen?", + view=FlagValueView(self.values[0], self.on_flag_select_callback), + ephemeral=True, + ) + + +class FlagsView(ui.View): + def __init__( + self, on_flag_select_callback: Callable[[Interaction, str, str], CoroutineType[Any, Any, None]] + ) -> None: + super().__init__(timeout=600) + + _ = self.add_item(FlagsSelect(on_flag_select_callback)) + + +class EntranceSoundBoardSelect(ui.Select[ui.View]): + def __init__( + self, on_sound_select_callback: Callable[[Interaction, str, str], CoroutineType[Any, Any, None]] + ) -> None: + self.on_sound_select_callback: Callable[[Interaction, str, str], CoroutineType[Any, Any, None]] = ( + on_sound_select_callback + ) + + options: list[SelectOption] = [SelectOption(label=board, value=board) for board in os.listdir(f"{SOUNDDIR}")] + + super().__init__(placeholder="Select Board", min_values=1, max_values=1, options=options) + + @override + async def callback(self, interaction: Interaction) -> None: + _ = await interaction.response.send_message( + "Welchen sound willst du?", view=EntranceSoundSoundView(self.values[0], self.on_sound_select_callback), ephemeral=True, ) -class EntranceSoundBoardView(discord.ui.View): - def __init__(self, on_sound_select_callback): +class EntranceSoundBoardView(ui.View): + def __init__( + self, on_sound_select_callback: Callable[[Interaction, str, str], CoroutineType[Any, Any, None]] + ) -> None: super().__init__(timeout=600) - self.add_item(EntranceSoundBoardSelect(on_sound_select_callback)) + _ = self.add_item(EntranceSoundBoardSelect(on_sound_select_callback)) -async def user_config_key_autocomplete( - interaction: Interaction, current: str -) -> List[Choice[str]]: +class EntranceSoundSoundSelect(ui.Select[ui.View]): + def __init__( + self, board: str, on_sound_select_callback: Callable[[Interaction, str, str], CoroutineType[Any, Any, None]] + ) -> None: + self.board: str = board + self.on_sound_select_callback: Callable[[Interaction, str, str], CoroutineType[Any, Any, None]] = ( + on_sound_select_callback + ) + + options: list[SelectOption] = [ + SelectOption(label=sound.split(".")[0], value=sound) for sound in os.listdir(f"{SOUNDDIR}/{board}") + ] + + super().__init__(placeholder="Select Sound", min_values=1, max_values=1, options=options) + + @override + async def callback(self, interaction: Interaction) -> None: + await self.on_sound_select_callback(interaction, self.board, self.values[0]) + + +class EntranceSoundSoundView(ui.View): + def __init__( + self, board: str, on_sound_select_callback: Callable[[Interaction, str, str], CoroutineType[Any, Any, None]] + ) -> None: + super().__init__(timeout=600) + + _ = self.add_item(EntranceSoundSoundSelect(board, on_sound_select_callback)) + + +async def user_config_key_autocomplete(interaction: Interaction, current: str) -> list[Choice[str]]: """ Suggest a value from the user config keys (each .conf section is a key). """ return [ - Choice(name=key, value=key) - for key in client.user_config.sections() - if key.lower().startswith(current.lower()) + Choice(name=key, value=key) for key in client.user_config.sections() if key.lower().startswith(current.lower()) ] @client.tree.command( - name="userconfig", + name="heidiconfig", description="User-spezifische Heidi-Einstellungen (Heidi merkt sie sich in ihrem riesigen Gehirn).", ) @app_commands.rename(config_key="option") @app_commands.describe(config_key="Die Option, welche du ändern willst.") @app_commands.autocomplete(config_key=user_config_key_autocomplete) @enforce_channel(HEIDI_SPAM_ID) -async def user_config(interaction: Interaction, config_key: str) -> None: +async def user_config(interaction: Interaction[Client], config_key: str) -> None: """ Set a user config value for the calling user. """ # Only Members can set settings if not isinstance(interaction.user, Member): print("User not a member") - await interaction.response.send_message( - "Heidi sagt: Komm in die Gruppe!", ephemeral=True - ) + _ = await interaction.response.send_message("Heidi sagt: Komm in die Gruppe!", ephemeral=True) return member: Member = interaction.user - async def on_sound_select_callback(interaction, board: str, sound: str): + async def on_flag_select_callback(interaction: Interaction, flag: str, value: str) -> None: """ - This function is called, when an EntrySoundSoundSelect option is selected. + This function is called when an FlagValueSelect option is selected. + """ + client.user_config[ConfigSection.FLAGS.value][flag] = value + client.write_user_config() + + _ = await interaction.response.send_message( + f"Ok, ich schreibe {flag}={value} in mein fettes Gehirn!", + ephemeral=True, + ) + + async def on_sound_select_callback(interaction: Interaction, board: str, sound: str) -> None: + """ + This function is called when an EntrySoundSoundSelect option is selected. """ client.user_config[config_key][member.name] = f"{board}/{sound}" client.write_user_config() - await interaction.response.send_message( + _ = await interaction.response.send_message( f"Ok, ich schreibe {member.name}={board}/{sound} in mein fettes Gehirn!", ephemeral=True, ) # Views for different user config options are defined here - views = {"ENTRANCE.SOUND": (EntranceSoundBoardView, on_sound_select_callback)} + views: dict[str, tuple[type[ui.View], Callable[..., CoroutineType[Any, Any, None]], str]] = { + ConfigSection.FLAGS.value: ( + FlagsView, + on_flag_select_callback, + "Welches Setting möchtest du ändern?", + ), + ConfigSection.ENTRANCE_SOUND.value: ( + EntranceSoundBoardView, + on_sound_select_callback, + "Aus welchem Soundboard soll dein Sound sein?", + ), + } - view, select_callback = views[config_key] + view, select_callback, description = views[config_key] - await interaction.response.send_message( - f"Aus welchem Soundboard soll dein sound sein?", - view=view(select_callback), + _ = await interaction.response.send_message( + description, + view=view(select_callback), # pyright: ignore[reportCallIssue] ephemeral=True, ) @@ -219,7 +308,8 @@ async def heidi_exclaim(interaction: Interaction) -> None: "Models müssen da halt durch!", "Heul doch nicht!", ] - await interaction.response.send_message(random.choice(messages)) + + _ = await interaction.response.send_message(random.choice(messages)) @client.tree.command(name="miesmuschel", description="Was denkt Heidi?") @@ -247,14 +337,13 @@ async def magic_shell(interaction: Interaction, question: str) -> None: "In deinen Träumen.", "Tom sagt Nein", ] + question = question.strip() question_mark = "" if question[-1] == "?" else "?" - await interaction.response.send_message( - f"{question}{question_mark}\nHeidi sagt: {random.choice(choices)}" - ) + _ = await interaction.response.send_message(f"{question}{question_mark}\nHeidi sagt: {random.choice(choices)}") -# @todo Allow , separated varargs, need to parse manually as slash commands don't support varargs +# TODO: Allow separated varargs, need to parse manually as slash commands don't support varargs @client.tree.command(name="wähle", description="Heidi trifft die Wahl!") @app_commands.rename(option_a="entweder") @app_commands.describe(option_a="Ist es vielleicht dies?") @@ -266,7 +355,7 @@ async def choose(interaction: Interaction, option_a: str, option_b: str) -> None Select an answer from two options. """ options = [option_a.strip(), option_b.strip()] - await interaction.response.send_message( + _ = await interaction.response.send_message( f"{options[0]} oder {options[1]}?\nHeidi sagt: {random.choice(options)}" ) @@ -274,40 +363,28 @@ async def choose(interaction: Interaction, option_a: str, option_b: str) -> None # Sounds ----------------------------------------------------------------------------------------- -async def board_autocomplete( - interaction: Interaction, current: str -) -> List[Choice[str]]: +async def board_autocomplete(interaction: Interaction, current: str) -> list[Choice[str]]: """ Suggest a sound board. """ - boards: List[str] = os.listdir(SOUNDDIR) + boards: list[str] = os.listdir(SOUNDDIR) - return [ - Choice(name=board, value=board) - for board in boards - if board.lower().startswith(current.lower()) - ] + return [Choice(name=board, value=board) for board in boards if board.lower().startswith(current.lower())] -async def sound_autocomplete( - interaction: Interaction, current: str -) -> List[Choice[str]]: +async def sound_autocomplete(interaction: Interaction, current: str) -> list[Choice[str]]: """ Suggest a sound from an already selected board. """ board: str = interaction.namespace.board - sounds: List[str] = os.listdir(f"{SOUNDDIR}/{board}/") + sounds: list[str] = os.listdir(f"{SOUNDDIR}/{board}/") return [ - Choice(name=sound.split(".")[0], value=sound) - for sound in sounds - if sound.lower().startswith(current.lower()) + Choice(name=sound.split(".")[0], value=sound) for sound in sounds if sound.lower().startswith(current.lower()) ] -@client.tree.command( - name="sag", description="Heidi drückt den Knopf auf dem Soundboard." -) +@client.tree.command(name="sag", description="Heidi drückt den Knopf auf dem Soundboard.") @app_commands.describe(sound="Was soll Heidi sagen?") @app_commands.autocomplete(board=board_autocomplete) @app_commands.autocomplete(sound=sound_autocomplete) @@ -319,9 +396,7 @@ async def say_voiceline(interaction: Interaction, board: str, sound: str) -> Non # Only Members can access voice channels if not isinstance(interaction.user, Member): print("User not a member") - await interaction.response.send_message( - "Heidi sagt: Komm in die Gruppe!", ephemeral=True - ) + _ = await interaction.response.send_message("Heidi sagt: Komm in die Gruppe!", ephemeral=True) return member: Member = interaction.user @@ -329,47 +404,42 @@ async def say_voiceline(interaction: Interaction, board: str, sound: str) -> Non await play_voice_line_for_member(interaction, member, board, sound) -class InstantButton(discord.ui.Button): - def __init__(self, label: str, board: str, sound: str): - super().__init__(style=discord.ButtonStyle.red, label=label) +class InstantButton(ui.Button[ui.View]): + def __init__(self, label: str, board: str, sound: str) -> None: + super().__init__(style=ButtonStyle.red, label=label) - self.board = board - self.sound = sound + self.board: str = board + self.sound: str = sound - async def callback(self, interaction: Interaction): + @override + async def callback(self, interaction: Interaction) -> None: """ Handle a press of the button. """ if not isinstance(interaction.user, Member): - await interaction.response.send_message( + _ = await interaction.response.send_message( "Heidi mag keine discord.User, nur discord.Member!", ephemeral=True ) return - await play_voice_line_for_member( - interaction, interaction.user, self.board, self.sound - ) + await play_voice_line_for_member(interaction, interaction.user, self.board, self.sound) -class InstantButtonsView(discord.ui.View): - def __init__(self, board: str, timeout=None): +class InstantButtonsView(ui.View): + def __init__(self, board: str, timeout: float | None = None) -> None: super().__init__(timeout=timeout) sounds = os.listdir(f"{SOUNDDIR}/{board}") for sound in sounds: - self.add_item(InstantButton(sound.split(".")[0], board, sound)) + _ = self.add_item(InstantButton(sound.split(".")[0], board, sound)) -@client.tree.command( - name="instantbuttons", description="Heidi malt Knöpfe für Sounds in den Chat." -) +@client.tree.command(name="instantbuttons", description="Heidi malt Knöpfe für Sounds in den Chat.") @app_commands.describe(board="Welches Soundboard soll knöpfe bekommen?") @app_commands.autocomplete(board=board_autocomplete) @enforce_channel(HEIDI_SPAM_ID) async def soundboard_buttons(interaction: Interaction, board: str) -> None: - await interaction.response.send_message( - f"Soundboard: {board.capitalize()}", view=InstantButtonsView(board) - ) + _ = await interaction.response.send_message(f"Soundboard: {board.capitalize()}", view=InstantButtonsView(board)) # Contextmenu ------------------------------------------------------------------------------------ @@ -384,13 +454,11 @@ async def insult( Send an insult to a member via direct message. """ if not member.dm_channel: - await member.create_dm() + _ = await member.create_dm() if not member.dm_channel: print("Error creating DMChannel!") - await interaction.response.send_message( - "Heidi sagt: Gib mal DM Nummer süße*r!", ephemeral=True - ) + _ = await interaction.response.send_message("Heidi sagt: Gib mal DM Nummer süße*r!", ephemeral=True) return insults = [ @@ -409,8 +477,8 @@ async def insult( "Richtiger Gesichtsgünther ey!", ] - await member.dm_channel.send(random.choice(insults)) - await interaction.response.send_message( + _ = await member.dm_channel.send(random.choice(insults)) + _ = await interaction.response.send_message( "Anzeige ist raus!", ephemeral=True ) # with ephemeral = True only the caller can see the answer diff --git a/heidi_client.py b/heidi_client.py index df242e4..3562ba4 100644 --- a/heidi_client.py +++ b/heidi_client.py @@ -1,22 +1,49 @@ -import configparser -from discord import app_commands, Message, VoiceState +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_constants import * -from heidi_helpers import * +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(discord.Client): - def __init__(self, *, intents: discord.Intents): - super().__init__(status="Nur eine kann GNTM werden!", intents=intents) +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 = app_commands.CommandTree(self) + self.tree: CommandTree = CommandTree(self) # Handle persistent user configuration - self.user_config = configparser.ConfigParser() + self.user_config: ConfigParser = ConfigParser() if not os.path.exists(f"{CONFIGPATH}/{USERCONFIGNAME}"): - open(f"{CONFIGPATH}/{USERCONFIGNAME}", "x") - self.user_config.read(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() @@ -24,43 +51,53 @@ class HeidiClient(discord.Client): # 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 = { + 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, - lambda m: "jeremy" in m.author.nick.lower(): self._autoreact_to_jeremy, - lambda m: "kardashian" in m.author.nick.lower() - or "jenner" in m.author.nick.lower(): self._autoreact_to_kardashian, + 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 = { - lambda m, b, a: b.channel != a.channel - and a.channel is not None - and isinstance(a.channel, VoiceChannel): self._play_entrance_sound, + 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 - async def setup_hook(self): + @override + async def setup_hook(self) -> None: self.tree.copy_global_to(guild=LINUS_GUILD) - await self.tree.sync(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) + _ = 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. """ - user_config_sections = ["ENTRANCE.SOUND"] - for section in user_config_sections: + 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: @@ -119,9 +156,13 @@ class HeidiClient(discord.Client): 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 ( - member is not None + 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 @@ -129,9 +170,7 @@ class HeidiClient(discord.Client): print("Not playing entrance sound, as no other members are present") return - soundpath: Union[str, None] = self.user_config["ENTRANCE.SOUND"].get( - member.name, None - ) + 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") diff --git a/heidi_config.py b/heidi_config.py new file mode 100644 index 0000000..4aa32c5 --- /dev/null +++ b/heidi_config.py @@ -0,0 +1,19 @@ +from enum import Enum + + +class ConfigSection(Enum): + FLAGS = "FLAGS" + ENTRANCE_SOUND = "ENTRANCE.SOUND" + + +class FlagsConfigKey(Enum): + DISABLE_JOIN_SOUND_IF_ALONE = "disable_join_sound_if_alone" + + +# NOTE: This is the default configuration scheme +USER_CONFIG_SCHEME: dict[str, list[tuple[str, str]]] = { + ConfigSection.FLAGS.value: [ + (FlagsConfigKey.DISABLE_JOIN_SOUND_IF_ALONE.value, str(True)), + ], + ConfigSection.ENTRANCE_SOUND.value: [], +} diff --git a/heidi_constants.py b/heidi_constants.py index eb67ad7..00e5741 100644 --- a/heidi_constants.py +++ b/heidi_constants.py @@ -1,29 +1,28 @@ import os -import discord +from discord.object import Object from dotenv import load_dotenv # This is run when this file is imported -load_dotenv() +HAS_VARIABLES: bool = load_dotenv() print("Debug: Importing heidi_constants.py") -# ================================================================================================ # -# ================================================================================================ # -# NOTE: Always set this correctly: -DOCKER = os.getenv("DOCKER") == "True" -# ================================================================================================ # -# ================================================================================================ # +# =========================================================================== # +# =========================================================================== # +DOCKER: bool = os.getenv("DOCKER") == str(True) +# =========================================================================== # +# =========================================================================== # # Constants -CONFIGPATH = "/config" if DOCKER else "." -USERCONFIGNAME = "Heidi_User.conf" +CONFIGPATH: str = "/config" if DOCKER else "." +USERCONFIGNAME: str = "Heidi_User.conf" SOUNDDIR: str = "/sounds" if DOCKER else "./heidi-sounds" # IDs of the servers Heidi is used on -LINUS_GUILD = discord.Object(id=431154792308408340) -TEST_GUILD = discord.Object(id=821511861178204161) +LINUS_GUILD: Object = Object(id=431154792308408340) +TEST_GUILD: Object = Object(id=821511861178204161) # Channel IDs -HEIDI_SPAM_ID = 822223476101742682 \ No newline at end of file +HEIDI_SPAM_ID: int = 822223476101742682 diff --git a/heidi_helpers.py b/heidi_helpers.py index b274bd5..eb0257e 100644 --- a/heidi_helpers.py +++ b/heidi_helpers.py @@ -1,11 +1,12 @@ import asyncio import functools -from typing import Union +from types import CoroutineType +from typing import Any, Callable -import discord -from discord import Interaction, VoiceChannel, Member +from discord import Interaction, Message, VoiceChannel, Member, VoiceClient +from discord.player import FFmpegPCMAudio -from heidi_constants import * +from heidi_constants import SOUNDDIR print("Debug: Importing heidi_helpers.py") @@ -15,14 +16,15 @@ print("Debug: Importing heidi_helpers.py") # 1. @enforce_channel(ID) is added to a function, which evaluates to decorate with the channel_id in its closure # 2. The function is passed to decorate(function), -def enforce_channel(channel_id): +def enforce_channel(channel_id: int): """ - Only run a function if called from the correct channel. + Only run a function if called from the specified voice channel. """ - def decorate(function): + + def decorate(function: Callable[..., CoroutineType[Any, Any, None]]): @functools.wraps(function) - async def wrapped(*args, **kwargs): + async def wrapped(*args: *tuple[Interaction, *tuple[Any, ...]], **kwargs: int): """ Sends an interaction response if the interaction is not triggered from the heidi_spam channel. """ @@ -30,7 +32,7 @@ def enforce_channel(channel_id): # Do not call the decorated function if the channel_id doesn't match if not interaction.channel_id == channel_id: - await interaction.response.send_message("Heidi sagt: Geh in heidi_spam du dulli", ephemeral=True) + _ = await interaction.response.send_message("Heidi sagt: Geh in heidi_spam du dulli", ephemeral=True) return await function(*args, **kwargs) @@ -40,12 +42,34 @@ def enforce_channel(channel_id): return decorate +# Reactions -------------------------------------------------------------------------------------- + + +def create_author_based_message_predicate(names: list[str]) -> Callable[[Message], bool]: + """ + Create a predicate that determines if a message was written by a certain author. + For usage with on_message_triggers. + """ + + def handler(message: Message) -> bool: + for name in names: + if ( + isinstance(message.author, Member) + and message.author.nick is not None + and name.lower() in message.author.nick.lower() + ): + return True + + return False + + return handler + + # Sounds ----------------------------------------------------------------------------------------- -# @todo Normalize volume when playing async def play_voice_line( - interaction: Union[Interaction, None], + interaction: Interaction | None, voice_channel: VoiceChannel, board: str, sound: str, @@ -54,23 +78,22 @@ async def play_voice_line( Play a voice line in the specified channel. """ try: - open(f"{SOUNDDIR}/{board}/{sound}") + # Check if the file exists + _ = open(f"{SOUNDDIR}/{board}/{sound}") except IOError: print(f"Error: Invalid soundfile {SOUNDDIR}/{board}/{sound}!") if interaction is not None: - await interaction.response.send_message( - f'Heidi sagt: "{board}/{sound}" kanninich finden bruder', - ephemeral=True + _ = await interaction.response.send_message( + f'Heidi sagt: "{board}/{sound}" kanninich finden bruder', ephemeral=True ) return if interaction is not None: - await interaction.response.send_message(f'Heidi sagt: "{board}/{sound}"', ephemeral=True) + _ = await interaction.response.send_message(f'Heidi sagt: "{board}/{sound}"', ephemeral=True) - audio_source = discord.FFmpegPCMAudio( - f"{SOUNDDIR}/{board}/{sound}" - ) # only works from docker - voice_client = await voice_channel.connect() + # TODO: Normalize volume when playing + audio_source: FFmpegPCMAudio = FFmpegPCMAudio(f"{SOUNDDIR}/{board}/{sound}") # only works from docker + voice_client: VoiceClient = await voice_channel.connect() voice_client.play(audio_source) while voice_client.is_playing(): @@ -80,7 +103,7 @@ async def play_voice_line( async def play_voice_line_for_member( - interaction: Union[Interaction, None], + interaction: Interaction | None, member: Member, board: str, sound: str, @@ -97,7 +120,7 @@ async def play_voice_line_for_member( ): print("User not in (valid) voice channel!") if interaction is not None: - await interaction.response.send_message("Heidi sagt: Komm in den Channel!", ephemeral=True) + _ = await interaction.response.send_message("Heidi sagt: Komm in den Channel!", ephemeral=True) return voice_channel: VoiceChannel = member.voice.channel