diff --git a/Heidi_User.conf b/Heidi_User.conf new file mode 100644 index 0000000..b765140 --- /dev/null +++ b/Heidi_User.conf @@ -0,0 +1,2 @@ +[ENTRANCE.SOUND] + diff --git a/bot.py b/bot.py index 7a41a23..858e006 100644 --- a/bot.py +++ b/bot.py @@ -1,170 +1,20 @@ # Example: https://github.com/Rapptz/discord.py/blob/master/examples/app_commands/basic.py -import os, re, random, logging, asyncio, discord, configparser -from discord import app_commands, Member, VoiceState, VoiceChannel, Message, Interaction +import random, logging from discord.app_commands import Choice -from functools import reduce -from dotenv import load_dotenv from typing import Dict, List, Optional, Union, Callable, Any - -# We're fancy today from rich.traceback import install +from heidi_client import * + +# Install rich traceback install(show_locals=True) -load_dotenv() - - -# ================================================================================================ # -# ================================================================================================ # -# NOTE: Always set this correctly: -DOCKER = os.getenv("DOCKER") == "True" -# ================================================================================================ # -# ================================================================================================ # # TODO: Only post in heidi-spam channel # TODO: yt-dlp music support # TODO: Somehow upload voicelines more easily (from discord voice message?) -# IDs of the servers Heidi is used on -LINUS_GUILD = discord.Object(id=431154792308408340) -TEST_GUILD = discord.Object(id=821511861178204161) - -CONFIGPATH = "/config" if DOCKER else "." -USERCONFIGNAME = "Heidi_User.conf" - - -class HeidiClient(discord.Client): - def __init__(self, *, intents: discord.Intents): - super().__init__(status="Nur eine kann GNTM werden!", intents=intents) - - # Separate object that keeps all application command state - self.tree = app_commands.CommandTree(self) - - # Handle persistent user configuration - self.user_config = configparser.ConfigParser() - if not os.path.exists(f"{CONFIGPATH}/{USERCONFIGNAME}"): - os.mknod(f"{CONFIGPATH}/{USERCONFIGNAME}") - 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 = { - # 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, - } - - # 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, - } - - # Synchronize commands to guilds - async def setup_hook(self): - 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. - """ - user_config_sections = ["ENTRANCE.SOUND"] - - for section in user_config_sections: - if section not in self.user_config: - print(f"Adding section {section} to {CONFIGPATH}/{USERCONFIGNAME}") - self.user_config[section] = dict() - - 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. - This function is set in on_voice_state_triggers and triggered by the on_voice_state_update event. - """ - soundpath: Union[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) - - -# ------------------------------------------------------------------------------------------------ - # Log to file handler = logging.FileHandler(filename="discord.log", encoding="utf-8", mode="w") @@ -491,69 +341,6 @@ async def insult( ) # with ephemeral = True only the caller can see the answer -# Helpers ---------------------------------------------------------------------------------------- - - -async def play_voice_line( - interaction: Union[Interaction, None], - voice_channel: VoiceChannel, - board: str, - sound: str, -) -> None: - """ - Play a voice line in the specified channel. - """ - try: - open(f"{SOUNDDIR}/{board}/{sound}.mkv") - except IOError: - print("Error: Invalid soundfile!") - if interaction is not None: - await interaction.response.send_message( - f'Heidi sagt: "{board}/{sound}" kanninich finden bruder' - ) - return - - if interaction is not None: - await interaction.response.send_message(f'Heidi sagt: "{board}/{sound}"') - - audio_source = discord.FFmpegPCMAudio( - f"{SOUNDDIR}/{board}/{sound}.mkv" - ) # only works from docker - voice_client = await voice_channel.connect() - voice_client.play(audio_source) - - while voice_client.is_playing(): - await asyncio.sleep(1) - - await voice_client.disconnect() - - -async def play_voice_line_for_member( - interaction: Union[Interaction, None], - member: Member, - board: str, - sound: str, -) -> None: - """ - Play a voice line in the member's current channel. - """ - # Member needs to be in voice channel to hear audio (Heidi needs to know the channel to join) - if ( - member is None - or member.voice is None - or member.voice.channel is None - or not isinstance(member.voice.channel, VoiceChannel) - ): - print("User not in (valid) voice channel!") - if interaction is not None: - await interaction.response.send_message("Heidi sagt: Komm in den Channel!") - return - - voice_channel: VoiceChannel = member.voice.channel - - await play_voice_line(interaction, voice_channel, board, sound) - - # ------------------------------------------------------------------------------------------------ diff --git a/heidi_client.py b/heidi_client.py new file mode 100644 index 0000000..c7d041c --- /dev/null +++ b/heidi_client.py @@ -0,0 +1,134 @@ +import configparser +from discord import app_commands, Message, VoiceState + +from heidi_constants import * +from heidi_helpers import * + + +class HeidiClient(discord.Client): + def __init__(self, *, intents: discord.Intents): + super().__init__(status="Nur eine kann GNTM werden!", intents=intents) + + # Separate object that keeps all application command state + self.tree = app_commands.CommandTree(self) + + # Handle persistent user configuration + 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}") + 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 = { + # 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, + } + + # 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, + } + + # Synchronize commands to guilds + async def setup_hook(self): + 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. + """ + user_config_sections = ["ENTRANCE.SOUND"] + + for section in user_config_sections: + if section not in self.user_config: + print(f"Adding section {section} to {CONFIGPATH}/{USERCONFIGNAME}") + self.user_config[section] = dict() + + 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. + This function is set in on_voice_state_triggers and triggered by the on_voice_state_update event. + """ + soundpath: Union[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) diff --git a/heidi_constants.py b/heidi_constants.py new file mode 100644 index 0000000..bf35061 --- /dev/null +++ b/heidi_constants.py @@ -0,0 +1,22 @@ +import os +import discord +from dotenv import load_dotenv + +# This is run when this file is imported +load_dotenv() + + +# ================================================================================================ # +# ================================================================================================ # +# NOTE: Always set this correctly: +DOCKER = os.getenv("DOCKER") == "True" +# ================================================================================================ # +# ================================================================================================ # + +# Constants +CONFIGPATH = "/config" if DOCKER else "." +USERCONFIGNAME = "Heidi_User.conf" + +# IDs of the servers Heidi is used on +LINUS_GUILD = discord.Object(id=431154792308408340) +TEST_GUILD = discord.Object(id=821511861178204161) diff --git a/heidi_helpers.py b/heidi_helpers.py new file mode 100644 index 0000000..0d4c7d3 --- /dev/null +++ b/heidi_helpers.py @@ -0,0 +1,65 @@ +import asyncio +from typing import Union + +import discord +from discord import Interaction, VoiceChannel, Member + + +async def play_voice_line( + interaction: Union[Interaction, None], + voice_channel: VoiceChannel, + board: str, + sound: str, +) -> None: + """ + Play a voice line in the specified channel. + """ + try: + open(f"{SOUNDDIR}/{board}/{sound}.mkv") + except IOError: + print("Error: Invalid soundfile!") + if interaction is not None: + await interaction.response.send_message( + f'Heidi sagt: "{board}/{sound}" kanninich finden bruder' + ) + return + + if interaction is not None: + await interaction.response.send_message(f'Heidi sagt: "{board}/{sound}"') + + audio_source = discord.FFmpegPCMAudio( + f"{SOUNDDIR}/{board}/{sound}.mkv" + ) # only works from docker + voice_client = await voice_channel.connect() + voice_client.play(audio_source) + + while voice_client.is_playing(): + await asyncio.sleep(1) + + await voice_client.disconnect() + + +async def play_voice_line_for_member( + interaction: Union[Interaction, None], + member: Member, + board: str, + sound: str, +) -> None: + """ + Play a voice line in the member's current channel. + """ + # Member needs to be in voice channel to hear audio (Heidi needs to know the channel to join) + if ( + member is None + or member.voice is None + or member.voice.channel is None + or not isinstance(member.voice.channel, VoiceChannel) + ): + print("User not in (valid) voice channel!") + if interaction is not None: + await interaction.response.send_message("Heidi sagt: Komm in den Channel!") + return + + voice_channel: VoiceChannel = member.voice.channel + + await play_voice_line(interaction, voice_channel, board, sound)