Clean up type hints everywhere, overhaul bot configuration
All checks were successful
Build Heidi Docker image / build-docker (push) Successful in 2m12s

This commit is contained in:
2025-09-23 20:39:11 +02:00
parent 591f4ea191
commit 75fb627361
6 changed files with 332 additions and 184 deletions

View File

@ -1,4 +1,4 @@
FROM python:3.13.7-slim-buster
FROM python:3.13.7-slim-trixie
RUN apt-get update -y && apt-get install -y ffmpeg libopus0

308
bot.py
View File

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

View File

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

19
heidi_config.py Normal file
View File

@ -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: [],
}

View File

@ -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
HEIDI_SPAM_ID: int = 822223476101742682

View File

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