christoph/discord-heidi
1
Fork 0
Code Issues 9 Actions Packages Releases 3 Activity

Split Heidi into multiple parts
All checks were successful
Build Heidi Docker image / build-docker (push) Successful in 28s
Details

Browse Source
This commit is contained in:
Christoph Urlacher
2023-12-09 17:14:05 +01:00
parent 16822e0212
commit 2e493e404b
5 changed files with 227 additions and 217 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
Heidi_User.conf Normal file
View File

@ -0,0 +1,2 @@
[ENTRANCE.SOUND]

221
bot.py
View File

@ -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)
# ------------------------------------------------------------------------------------------------

134
heidi_client.py Normal file
View File

@ -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)

22
heidi_constants.py Normal file
View File

@ -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)

65
heidi_helpers.py Normal file
View File

@ -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)