5  Discord Bots

5.1 Einleitung

Discord ist mit über 563 Millionen Nutzern weltweit einer der bekanntesten Chat und VoIP Plattformen überhaupt. Um die Funktionen der Plattform zu erweitern, bietet Discord eine nutzerfreundliche Schnittstelle an um Bots und weitere Applikationen für die Plattform zu schreiben. In diesem Projekt haben wir uns mit der Erstellung eines solchen Discord Bots beschäftigt.

5.2 Funktionen

5.2.1 Allgemeiner Code

Um mit der API von Discord zu interagieren, bietet Discord ein PyPi Paket an welche die Implementierung vereinfacht.

Zunächst werden die folgenden Pakete importiert:

import os
import logging
import discord
from discord import Intents
from discord.ext import commands
from dotenv import load_dotenv
import requests

Um geheime API Token strikt vom Programmcode zu trennen wird durch die Methode “load_dotenv” aus dem Paket “dotenv” zunächst der Discord Bot Token ins Programm geladen. Diesen Token erhält man, nachdem der Bot über die Entwickler Seite von Discord erstellt wurde. Außerdem wird ein logger initialisiert.

load_dotenv()
TOKEN = os.getenv('DISCORD_BOT_TOKEN')

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger('discord_bot')

Nun muss noch definiert werden, welche Berechtigungen der Bot anfragen soll. Dazu werden sog. “Intents” festgelgt, welche zuvor über das Entwickler Portal von Discord für den entsprechenden Bot freigeschaltet werden müssen. Abschließend wird noch ein Präfix für den Bot festgelegt und das finale Bot Objekt erstellt.

intents = Intents.default()
intents.message_content = True
intents.members = True

bot = commands.Bot(command_prefix='!', intents=intents)

Nun können Events mittels des @bot.event Decorators registriert werden. So kann unser Bot code auf bestimmte Ereignisse reagieren.

@bot.event
async def on_ready():
    logger.info(f'Bot eingeloggt als {bot.user} (ID: {bot.user.id})')

@bot.event
async def on_message(message):
    if message.author == bot.user:
        return

    content = message.content.strip()

    parts = content[len(bot.command_prefix):].split()
    command = parts[0].lower()

    logger.info(f'Empfangenes Kommando: {command} mit args: {message.content} von {message.author}')

    # Nutzer hat dem Bot eine DM gesendet
    if message.channel.type == discord.ChannelType.private and not content.startswith(
        bot.command_prefix
    ):
        logger.info("DM received")
        await llm_answer(message=message)
    else:
        if not content.startswith(bot.command_prefix):
            return
        await bot.process_commands(message)

Das wichtigste Event ist das “on_message” Event. Diese Funktion wird immer dann aufgerufen, wenn der Bot eine Nachricht empfängt (entweder privat oder auf dem Textkanal eines Servers). In unserem Fall werden dem Bot nur im Rahmen der Nutzung des LLMs Direktnachrichten gesendet, deshalb wird in der on_message Funktion abgefragt ob der Bot die Nachricht privat, oder auf einem Server erhalten hat. Wenn ersteres der Fall ist, wird eine LLM Antwort über die llm_answer Methode erstellt und gesendet. Sollte hingegen eine Nachricht auf einem Server empfangen worden sein die nicht vom Bot selbst stammt und mit dem oben definierten Bot Präfix (hier “!”) beginnt, so wird über die bot.process_commands Funktion die Verarbeitung an eine der im Folgenden beschriebenen Funktionen weitergegeben, die verschiedene Funktionalitäten bereitstellen.

5.2.2 Chatbot mit LLM

Dieser Teil des Discord-Bots implementiert die Möglichkeit, mit einem lokal gehosteten Large Language Model über Discord Chat zu kommunizieren. Gehostet wird das Modell über Ollama.

5.2.2.1 Voraussetzungen

Damit dieser Teil des Bots problemlos ausgeführt werden kann, müssen auf dem Gerät, auf dem der Bot ausgeführt wird, verschiedene Voraussetzungen erfüllt sein:

• Ollama muss installiert sein (https://ollama.com/download)
• Das Python Paket für Ollama muss ebenfalls installiert sein (https://pypi.org/project/ollama/)
• Das entsprechende LLM, das ausgeführt werden soll, muss über Ollama installiert sein (https://ollama.com/search)

5.2.2.2 Implementation

5.2.2.2.1 Modell

# Ollama settings
model = "llama3.1:8b"
system_prompt = """
You are a helpful and friendly bot on a Discord Server. People will write you messages and you will answer them.
Since you are writing on Discord, which has a message of 2000 characters, you will need to keep your answers under this character count.
Keep your answers always under the limit of 2000 characters.
"""
options = {"num_predict": 1000, "temperature": 0.3}

# Chat cache for saving conversations with users
chat_cache = {}

In diesen Variablen kann das spezifische genutzte Modell gewählt und konfiguriert werden. Der System Prompt kann hier auch spezifisch verfasst werden und in den options können sowohl die Anzahl der maximal generierten Tokens, als auch die Temperatur des Sprachmodells eingestellt werden.

Da Discord eine maximale Charactezahl von 2000 pro gesendeter Nachricht hat, sollte der System Prompt dies behandeln, da das Modell sonst dazu neigt, bei komplexeren Anfragen zu lange Texte zu verfassen.

Zuletzt wird ein dictionary erstellt, das den Chat Cache darstellt.

5.2.2.2.2 create_message Funktion

def create_message(message, role):
    """
    Returns message in a format that is convenient for Ollama to work with.

    Args:
        message (str): The message to be saved.
        role (str): The role of the message author. Can be either 'user' or 'assistant'.

    Returns:
        dict: A dictionary containing the message and the author.
    """
    return {"role": role, "content": message}

Diese Funktion übernimmt die Aufgabe, eine gesendete oder emfangene Nachricht in ein Format zu übertragen, das als Chat Historie von Ollama gut zu verarbeiten ist.

5.2.2.2.3 llm_answer Funktion

async def llm_answer(message):
    """
    Generates an answer to the received message using Ollama.

    Args:
        message: The message to be answered by the LLM.
    """
    # Load message info
    usr = message.author
    msg = message.content

    # Load context
    try:
        chat_history = chat_cache[usr]
    except KeyError:
        chat_history = [{"role": "system", "content": system_prompt}]

    # Log message
    logger.info(f"Received Message from {usr}: {msg}")
    logger.info(f"Generating Response for {usr}...")

    # Write answer via DM
    channel = await usr.create_dm()
    async with channel.typing():
        usr_msg = create_message(msg, "user")
        chat_history.append(usr_msg)

        response = chat(model=model, messages=chat_history, options=options)

        rsp = response.message.content
        chat_history.append(create_message(rsp, "assistant"))

        chat_cache[usr] = chat_history

        logger.info(f"Response to {usr}: {rsp}")

        await channel.send(rsp)

Diese Funktion übernimmt die Aufgabe, auf die Nachricht eines Benutzers eine entsprechende Antwort von dem eingestellten Sprachmodell generieren zu lassen und diese über Privatnachricht and diesen Benutzer zu senden.

Zunächst wird aus dem Chat Cache der bisherige Nachrichtenverlauf mit dem Benutzer geladen. Sollte kein bisheriger Verlauf existieren, wird der Systemprompt stattdessen geladen.

Darauf wird ein Channel zur Direktbenachrichtigung erstellt, eine Antwort über Ollama generiert und diese Antwort über den Channel versendet, nachdem sie dem Verlauf hinzugefügt wurde.

5.2.2.2.4 on_message Event

@bot.event
async def on_message(message):
    """
    Answer received message via LLM on Ollama.

    Args:
        message: The message to be answered
    """
    if message.author == bot.user:
        return

    content = message.content.strip()

    # Only answer to DM messages
    if message.channel.type == discord.ChannelType.private and not content.startswith(
        bot.command_prefix
    ):
        await llm_answer(message=message)
    # If something was sent to bot-channel, process the command
    elif message.channel.id == CHANNEL_ID:
        await bot.process_commands(message)

Damit der Bot auf Nachrichten reagieren kann, muss das entsprechende Event einer empfangenen Nachricht behandelt werden. Hierzu wird unterschieden, ob die empfangene Nachricht entweder über eine Direktnachricht oder über einen dedizierten Bot Channel empfangen wurde. Im ersten Fall soll eine ensprechende Nachricht vom Sprachmodell generiert werden, indem die llm_answer Funktion aufgerufgen wird. Wird ein Kommando in den eigenen Bot Channel geschickt, soll dieser entsprechend behandelt werden. Andere Kanäle werden nicht beachten.

5.2.2.2.5 chat Command

@bot.command(name="chat")
async def llm_chat(ctx):
    """Request to initiate an LLM chat via DM."""
    logger.info(f"Got LLM chat request from {ctx.author.name}")
    usr = ctx.author
    await ctx.channel.send(f"Hi {usr}! Du willst mit mir sprechen? Klar! Ich schreib dir ne DM. 😄")
    # Send DM to initiate chat with LLM
    channel = await usr.create_dm()
    await channel.send(f"Hey {usr}. Was gibts?")

Falls in den Bot Channel das Kommando !chat gesendet wird, startet der Bot einen privaten Chat, indem er einen festgelegten Begrüßungstext über eine Direktnachricht an den Benutzer sendet, der das Kommando eingegeben hat.

5.2.2.3 Benutzung

Um einen privaten Chat mit dem Bot anzufragen, muss wie bereits genannt, das Kommando !chat in den dedizierten Bot Channel gesendet werden. Daraufhin leitet der Bot einen privaten Chat ein, in dem dann mit dem Large Language Model kommuniziert werden kann.

5.2.2.4 Gesamter Code

import os

import logging

from dotenv import load_dotenv

import discord
from discord.ext import commands

import ollama
from ollama import chat
from ollama import ChatResponse

# Configure logger
logging.basicConfig(
    filename="llm_bot.log",
    filemode="a",
    format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
    datefmt="%Y-%m-%d %H:%M:%S",
    level=logging.INFO,
)
logger = logging.getLogger("llm_test_bot")

# Load Tokens for Discord connection
load_dotenv()
TOKEN = os.getenv("DISCORD_TOKEN")
GUILD = os.getenv("DISCORD_GUILD")
CHANNEL_ID = int(os.getenv("DISCORD_CHANNEL_ID"))

# Configure bot
intents = discord.Intents.default()
intents.message_content = True
intents.members = True
bot = commands.Bot(command_prefix="!", intents=intents)

# Ollama settings
model = "llama3.1:8b"
system_prompt = """
You are a helpful and friendly bot on a Discord Server. People will write you messages and you will answer them.
Since you are writing on Discord, which has a message of 2000 characters, you will need to keep your answers under this character count.
Keep your answers always under the limit of 2000 characters.
"""
options = {"num_predict": 1000, "temperature": 0.3}

# Chat cache for saving conversations with users
chat_cache = {}


def create_message(message, role):
    """
    Returns message in a format that is convenient for Ollama to work with.

    Args:
        message (str): The message to be saved.
        role (str): The role of the message author. Can be either 'user' or 'assistant'.

    Returns:
        dict: A dictionary containing the message and the author.
    """
    return {"role": role, "content": message}


async def llm_answer(message):
    """
    Generates an answer to the received message using Ollama.

    Args:
        message: The message to be answered by the LLM.
    """
    # Load message info
    usr = message.author
    msg = message.content

    # Load context
    try:
        chat_history = chat_cache[usr]
    except KeyError:
        chat_history = [{"role": "system", "content": system_prompt}]

    # Log message
    logger.info(f"Received Message from {usr}: {msg}")
    logger.info(f"Generating Response for {usr}...")

    # Write answer via DM
    channel = await usr.create_dm()
    async with channel.typing():
        usr_msg = create_message(msg, "user")
        chat_history.append(usr_msg)

        response = chat(model=model, messages=chat_history, options=options)

        rsp = response.message.content
        chat_history.append(create_message(rsp, "assistant"))

        chat_cache[usr] = chat_history

        logger.info(f"Response to {usr}: {rsp}")

        await channel.send(rsp)


@bot.event
async def on_ready():
    for guild in bot.guilds:
        if guild.name == GUILD:
            break

    logger.info(
        f"{bot.user} connected to the following guild:\n"
        f"{guild.name}(id: {guild.id})"
    )

    members = "\n - ".join([member.name for member in guild.members])
    logger.debug(f"Guild Members:\n - {members}")


@bot.event
async def on_message(message):
    """
    Answer received message via LLM on Ollama.

    Args:
        message: The message to be answered
    """
    if message.author == bot.user:
        return

    content = message.content.strip()

    # Only answer to DM messages
    if message.channel.type == discord.ChannelType.private and not content.startswith(
        bot.command_prefix
    ):
        await llm_answer(message=message)
    # If something was sent to bot-channel, process the command
    elif message.channel.id == CHANNEL_ID:
        await bot.process_commands(message)


@bot.command(name="chat")
async def llm_chat(ctx):
    """Request to initiate an LLM chat via DM."""
    logger.info(f"Got LLM chat request from {ctx.author.name}")
    usr = ctx.author
    await ctx.channel.send(f"Hi {usr}! Du willst mit mir sprechen? Klar! Ich schreib dir ne DM. 😄")
    # Send DM to initiate chat with LLM
    channel = await usr.create_dm()
    await channel.send(f"Hey {usr}. Was gibts?")

bot.run(TOKEN)

5.2.3 Hang-Man

Dieses Projekt implementiert ein interaktives Hangman-Spiel (Galgenmännchen) in einem Discord-Bot mit Python und der discord.py-Bibliothek. Die Nutzer können mit dem Bot spielen, indem sie per Befehl ein Spiel starten und Buchstaben oder ganze Wörter raten.

---

5.2.3.1 Verwendete Technologien

• Programmiersprache: Python
• Bibliotheken:
  • discord: Für die Interaktion mit Discord
  • discord.ext.commands: Erweiterung für Bot-Kommandos
  • random: Für zufällige Wortauswahl und Begrüßungen

---

5.2.3.2 Projektstruktur

project/
│
├── bot.py           # Hauptcode für den Bot
├── token.txt        # Externe Datei zur sicheren Speicherung des Tokens

5.2.3.3 Spielmechanik

Beim Aufruf von !hangman startet ein neues Spiel mit folgenden Eigenschaften:

• Zufälliges Wort aus einer vordefinierten Liste
• 6 Leben (symbolisch durch Emojis dargestellt)
• Anzeige des Fortschritts mit Unterstrichen
• Eingaben über !guess oder !guess

5.2.3.4 Zentrale Datenstrukturen

games = {} # Spielstände für aktive Nutzer (pro User-ID) stats = {} # Statistik pro Nutzer (Siege & Niederlagen)

5.2.3.5 Wichtige Kommandos

!hangman

Startet ein neues Spiel für den aufrufenden Nutzer:

@bot.command()
async def hangman(ctx):
    ...

!guess <buchstabe|wort>

Nutzer rät einen Buchstaben oder das ganze Wort.

• Bei Erfolg wird der Fortschritt aktualisiert.

• Bei Misserfolg wird ein Leben abgezogen.

• Sieg/Niederlage wird erkannt und das Spiel beendet.

@bot.command()
async def guess(ctx, *, guess: str):
    ...

!stats Zeigt die bisherigen Siege und Niederlagen des Nutzers.

@bot.command()
async def stats(ctx):
    ...

5.2.3.6 Interaktion & Benutzererlebnis

Der Bot enthält witzige und provokante Begrüßungstexte für mehr Persönlichkeit:

greetings = [
    "Los gehts mein kleiner Freund",
    "Ich glaube ja nicht, dass du es schaffst...",
    ...
]

5.2.3.7 Ganzer Code

import discord
from discord.ext import commands
import random

# Lade den Token sicher aus Datei weil ich mega die Probleme anfangs hatte diesen blöden Token richtig einzunringen...
with open('token.txt', 'r') as f:
    TOKEN = f.read().strip()

# Bot-Intents (damit der Hanswurst Nachrichten lesen darf)
intents = discord.Intents.default()
intents.message_content = True

bot = commands.Bot(command_prefix="!", intents=intents)

# Wörterliste, kann wie man lustig ist angepasst werden, evtl lade ich später einfach ein Wörterbuch
words = ["python", "discord", "galgenmaennchen", "programmieren", "hangman", "eier", 
"ridvan", "yomama", "vincent", "marcel", "bruttosozialprodukt",
 "donaudampfschiffahrtsgesellschaftskapitän", 
"grundstücksverkehrsgenehmigungszuständigkeitsübertragungsverordnung",
 "rinderkennzeichnungsfleischetikettierungsüberwachungsaufgabenübertragungsgesetz", 
"bimbam", "blubberwasser", "bimbam", "betthupferl", "fickelfackel", "luftikus", "lulatsch",
"pustekuchen", "ratzefummel", "schwitzkasten", "sternhagelvoll"]

hangman_stages = [
    "💀💀💀💀💀💀", # 0 Leben
    "❤️️💀💀💀💀💀",
    "❤️️❤️️💀💀💀💀",
    "❤️️❤️️❤️️💀💀💀",
    "❤️️❤️️❤️️❤️️💀💀",
    "❤️️❤️️❤️️❤️️❤️️💀",
    "❤️️❤️️❤️️❤️️❤️️❤️️",  # 6 Leben
]


# Spielzustände
games = {}

stats = {}  # user_id: {'wins': int, 'losses': int} noch keine Funktion

greetings = [
    "Los gehts mein kleiner Freund",
    "Ich glaube ja nicht das du es shcaffst aber nun gut",
    "Du gegen den Bot - ganz klar wer hier gewinnt -> ICH 😈",
    "Hallo...Wilslt du dir das wirklich antun?",
    "Hahaha -  Du schon wieder ? Lass knacken mein Großer",
    "Spielt beginnt, bite gib dir dieses mal Mühe"
]

@bot.event
async def on_ready():
    print(f'✅ Eingeloggt als {bot.user.name}')

@bot.command()
async def hangman(ctx):
    word = random.choice(words)
    display = ['_'] * len(word)
    games[ctx.author.id] = {
        'word': word,
        'display': display,
        'guessed': [],
        'tries': 6
    }
    greeting = random.choice(greetings)
    await ctx.send(f"🕹️ {greeting}\n{hangman_stages[6]}\nWort: {' '.join(display)}\nRate mit `!guess <buchstabe oder wort>`")


@bot.command()
async def guess(ctx, *, guess: str):
    game = games.get(ctx.author.id)
    if not game:
        await ctx.send("❌ Du hast noch kein Spiel gestartet. Nutze `!hangman`.")
        return

    guess = guess.lower()

    # Buchstabe raten
    if len(guess) == 1:
        if guess in game['guessed']:
            await ctx.send("⚠️ Buchstabe wurde schon geraten.")
            return

        game['guessed'].append(guess)

        if guess in game['word']:
            for i, c in enumerate(game['word']):
                if c == guess:
                    game['display'][i] = guess
            await ctx.send(f"✅ Richtig! {hangman_stages[game['tries']]}\nWort: {' '.join(game['display'])}")
        else:
            game['tries'] -= 1
            await ctx.send(f"❌ Falsch! {hangman_stages[game['tries']]}\nVersuche übrig: {game['tries']}\nWort: {' '.join(game['display'])}")

    # Ganzes Wort raten
    else:
        if guess == game['word']:
            await ctx.send(f"🎉 Volltreffer! Das Wort war: `{game['word']}`")
            user_stats = stats.setdefault(ctx.author.id, {'wins': 0, 'losses': 0})
            user_stats['wins'] += 1
            del games[ctx.author.id]
            return
        else:
            game['tries'] -= 1
            await ctx.send(f"❌ Falsch geraten! {hangman_stages[game['tries']]}\nVersuche übrig: {game['tries']}\nWort: {' '.join(game['display'])}")

    # Sieg oder Niederlage prüfen
    if '_' not in game['display']:
        await ctx.send(f"🎉 Du hast das Wort erraten: `{game['word']}`! Glückwunsch! 🎊")
        user_stats = stats.setdefault(ctx.author.id, {'wins': 0, 'losses': 0})
        user_stats['wins'] += 1
        del games[ctx.author.id]
    elif game['tries'] <= 0:
        await ctx.send(f"💀 Verloren! Das Wort war: `{game['word']}`\nNächstes Mal besser. 🪦")
        user_stats = stats.setdefault(ctx.author.id, {'wins': 0, 'losses': 0})
        user_stats['losses'] += 1
        del games[ctx.author.id]

# Statistik anzeigen hat aber keine Funktion bis jetzt 
@bot.command()
async def stats(ctx):
    user_stats = stats.get(ctx.author.id, {'wins': 0, 'losses': 0})
    await ctx.send(
        f"📈 Deine Statistik für Hangman:\n🏆 Siege: {user_stats['wins']}\n💀 Niederlagen: {user_stats['losses']}"
    )

# Bot starten
bot.run(TOKEN)

5.2.4 Weather Forcast

Mit diesem Teil des Bots ist es möglich die aktuellen Wetterdaten (die tiefste Temperatur des Tages, die höchste Temperatur des Tages, die Durchschnitstemperatur des Tages und die aktuelle Temperatur) einer Stadt abzurufen.

5.2.4.1 Verwendete Libraries

• openmeteo_requests: Open-Meteo API Interfaces
• requests_cache: Um anfragen an die API zwischen zu speichern
• retry_requests: Damit öfter Anfragen gesendet werden, falls es zu Fehlern kommt

5.2.4.2 Implementation

@bot.command(name='weather')
async def weather(ctx, *, city="augsburg"):
    the_answer = '42'
    forcast = 0
    cities = {
        "augsburg": [48.37, 13.04],
        "berlin": [52.52, 13.40],
        "kiev": [50.45, 30.52],
        "münchen": [48.13, 11.58],
    }

Zunächst wartet der Bot darauf, dass der Befehl !weather gesendet wird. Wenn das passiert, wird die asynchrone Funktion weather ausgeführt, die einen optionalen Parameter entgegen nehmen kann; Als Standartwert wird augsburg übergeben. In der Funktion werden dann ein paar Variablen iniziirt, zum einen forcast, welche später die Antworten der API entgegen nimmt, zum anderen ein Dictionary, welches den Namen einer Stadt in Koordinaten umwandelt.

if '.' in city and ' ' in city:
        coordinates = city.split(None, 1)
        latitude = coordinates[0]
        longitude = coordinates[1]
    
        if ' ' in longitude:
            longitude = longitude.split(None, 1)[0]

        try:
            latitude = float(latitude)
            longitude = float(longitude)
            forcast = weather_forcast(latitude=latitude, longitude=longitude)
        except:
            await ctx.channel.send(f"At least one Value is not a float")

Anschließend wird überprüft ob es sich bei dem Parameter um Koordinaten handelt. Ist das der Fall, werden die Koordinaten aufgeteilt und an die Funktion weather_report geschickt. Das Ergebniss wird in der Variable forcast gespeichert. Handelt es sich nicht um Koordinaten wird eine Fehlernachricht zurückgegeben.

elif city.lower() in cities:
        city = city.lower()
        latitude = cities[city][0]
        longitude = cities[city][1]
        forcast = weather_report(latitude=latitude, longitude=longitude)

elif city == "help":
    await ctx.channel.send(
        f"The main way to use this bot is to parse the latitude and longitude of your deired city like this:    /weather 48.37 13.04\n" 
        f"You can also try the city name but it is probably not included yet"
    )
    return

Entspricht der Parameter nicht den Anforderungen für Koordinaten, so wird geprüft, ob es sich um einen Stadtnamen handelt, der bereits im Dictionary eingetragen ist. Sollte dass der Fall sein, werden die Koordinaten wieder an die Funktion weather_report und das Ergebniss wieder in forcast gespeichert. Sollte das wieder nicht der Fall sein, wird ein letztes mal getestet, ob es sich bei dem Parameter um help handelt.

# Setup the Open-Meteo API client with cache and retry on error
    cache_session = requests_cache.CachedSession('.cache', expire_after = 3600)
    retry_session = retry(cache_session, retries = 5, backoff_factor = 0.2)
    openmeteo = openmeteo_requests.Client(session = retry_session)

In der weather_forcast Funktion werden zuerst Variablen erstellt, die den Cache, die Wiederholungen bei einem fehlerhaften Verbindungsaufbau und die Verbindung zum Client verwalten.

url = "https://api.open-meteo.com/v1/forecast"
params = {
    "latitude": latitude,
    "longitude": longitude,
    "daily": ["temperature_2m_min", "temperature_2m_max", "temperature_2m_mean"],
    "current": "temperature_2m",
    "timezone": "auto",
    "forecast_days": 1
}
responses = openmeteo.weather_api(url, params=params)

Anschließend werden die Parameter in einem Dictionary zusammen getragen und mit der angefragten URL an die Instanz des Clients übergeben. Zurück kommt eine Liste, die unter den Namen responses gespeichert wird.

# Current values. The order of variables needs to be the same as requested.
current = responses[0].Current()
current_temp = str(current.Variables(0).Value())

# Process daily data. The order of variables needs to be the same as requested.
daily = responses[0].Daily()
min_temp = str(daily.Variables(0).ValuesAsNumpy())
max_temp= str(daily.Variables(1).ValuesAsNumpy())
mean_temp = str(daily.Variables(2).ValuesAsNumpy())

temps = [min_temp, max_temp, mean_temp, current_temp]

for i, temp in enumerate(temps):
    temp = temp.replace("[", "")
    temp = temp.replace("]", "")
    temp = temp.split(".", 1)[0]
    temps[i] = temp

return temps

Danach werden die angefragten Werte in Strings Formatiert zusammen gesammelt, ein letztes mal Formatiert und in einer Liste an die Discord-Funktion zurück gegeben.

if forcast != 0:
        await ctx.channel.send(
            f"Lowest Temperature: {forcast[0]}°C\n"
            f"Highest Temperature: {forcast[1]}°C\n"
            f"Mean Temperature: {forcast[2]}°C\n"
            f"Current Temperature: {forcast[3]}°C\n"
        )
else:    
    await ctx.channel.send(f"Either the city doesn't exist or you misspelled it or the coordinates don't exist")

Sollte die Variable forcast bearbeitet worden sein, so werden die Ergebnisse in Discord ausgegeben. Ändert sich in der ganzen Zeit nichts, so wird eine Meldung dazu zurück gegeben.

5.2.4.3 Ganzer Code

import os
import discord
import openmeteo_requests
import requests_cache

from retry_requests import retry
from dotenv import load_dotenv
from discord.ext import commands

load_dotenv()
TOKEN = os.getenv('DISCORD_TOKEN')

intents = discord.Intents.default()
intents.message_content = True

bot = commands.Bot(command_prefix='!', intents=intents)

@bot.command(name='weather')
async def weather(ctx, *, city="augsburg"):
    the_answer = '42'
    forcast = 0
    cities = {
        "augsburg": [48.37, 13.04],
        "berlin": [52.52, 13.40],
        "kiev": [50.45, 30.52],
        "münchen": [48.13, 11.58],
    }

    if '.' in city and ' ' in city:
        coordinates = city.split(None, 1)
        latitude = coordinates[0]
        longitude = coordinates[1]
    
        if ' ' in longitude:
            longitude = longitude.split(None, 1)[0]

        try:
            latitude = float(latitude)
            longitude = float(longitude)
            forcast = weather_report(latitude=latitude, longitude=longitude)
        except:
            await ctx.channel.send(f"At least one Value is not a float")

    elif city.lower() in cities:
        city = city.lower()
        latitude = cities[city][0]
        longitude = cities[city][1]
        forcast = weather_report(latitude=latitude, longitude=longitude)

    elif city == "help":
        await ctx.channel.send(
            f"The main way to use this bot is to parse the latitude and longitude of your deired city like this:    /weather 48.37 13.04\n" 
            f"You can also try the city name but it is probably not included yet"
        )
        return


    if forcast != 0:
        await ctx.channel.send(
            f"Lowest Temperature: {forcast[0]}°C\n"
            f"Highest Temperature: {forcast[1]}°C\n"
            f"Mean Temperature: {forcast[2]}°C\n"
            f"Current Temperature: {forcast[3]}°C\n"
        )
    else:
        await ctx.channel.send(f"Either the city doesn't exist or you misspelled it or the coordinates don't exist")


def weather_report(latitude=48.37, longitude=13.04):
    # Setup the Open-Meteo API client with cache and retry on error
    cache_session = requests_cache.CachedSession('.cache', expire_after = 3600)
    retry_session = retry(cache_session, retries = 5, backoff_factor = 0.2)
    openmeteo = openmeteo_requests.Client(session = retry_session)

    # Make sure all required weather variables are listed here
    # The order of variables in hourly or daily is important to assign them correctly below
    url = "https://api.open-meteo.com/v1/forecast"
    params = {
        "latitude": latitude,
        "longitude": longitude,
        "daily": ["temperature_2m_min", "temperature_2m_max", "temperature_2m_mean"],
        "current": "temperature_2m",
        "timezone": "auto",
        "forecast_days": 1
    }
    responses = openmeteo.weather_api(url, params=params)

    # Current values. The order of variables needs to be the same as requested.
    current = responses[0].Current()
    current_temp = str(current.Variables(0).Value())

    # Process daily data. The order of variables needs to be the same as requested.
    daily = responses[0].Daily()
    min_temp = str(daily.Variables(0).ValuesAsNumpy())
    max_temp= str(daily.Variables(1).ValuesAsNumpy())
    mean_temp = str(daily.Variables(2).ValuesAsNumpy())

    temps = [min_temp, max_temp, mean_temp, current_temp]

    for i, temp in enumerate(temps):
        temp = temp.replace("[", "")
        temp = temp.replace("]", "")
        temp = temp.split(".", 1)[0]
        temps[i] = temp

    return temps


bot.run(TOKEN)

5.2.5 Witzbo(l)t

Dieser Teil des Bots benutzt eine Öffentliche Witze-API (https://v2.jokeapi.dev/) um dem Nutzer Witze zu erzählen.

5.2.5.1 Verwendete Libraries

• requests: Standard-Library für Web-Requests
• discord.py: Für Bot-Funktionalität
• dotenv: Zum sicheren Laden des Bot-Tokens

5.2.5.2 Implementation

Der Witzbot (Witzbo(l)t) verwendet die öffentliche JokeAPI, um zufällige Witze in verschiedenen Kategorien und Sprachen abzurufen. Die Implementierung basiert auf Discord.py, nutzt einen Dekorator-Ansatz zur Befehlsregistrierung und unterscheidet zwischen öffentlichen und privaten Antworten.

5.2.5.2.1 Command-Dispatcher mit Registry

Alle Befehle werden mithilfe eines Decorators in ein zentrales Dictionary (FUNCTION_REGISTRY) eingetragen. Dadurch ist der Bot modular erweiterbar:

FUNCTION_REGISTRY = {}

def register_command(name: str, mode: ResponseMode = ResponseMode.PUBLIC):
    def decorator(func):
        FUNCTION_REGISTRY[name] = (func, mode)
        return func
    return decorator

5.2.5.3 Nachrichtenverarbeitung

Die zentrale Nachrichtenverarbeitung erfolgt im on_message-Event. Hier prüft der Bot, ob die Nachricht mit dem Befehlspräfix (!) beginnt, extrahiert das Kommando sowie Argumente und ruft die zugehörige Funktion aus dem Registry-System auf. Je nach ResponseMode antwortet der Bot öffentlich oder privat:

@bot.event
async def on_message(message):
    if message.author == bot.user:
        return

    content = message.content.strip()
    if not content.startswith(bot.command_prefix):
        return

    parts = content[len(bot.command_prefix):].split()
    command = parts[0].lower()
    args = parts[1:]

    logger.info(f'Empfangenes Kommando: {command} mit args: {args} von {message.author}')

    if command in FUNCTION_REGISTRY:
        func, mode = FUNCTION_REGISTRY[command]
        try:
            result = await func(message, *args)
            if result is not None:
                if mode == ResponseMode.PRIVATE:
                    dm = await message.author.create_dm()
                    await dm.send(result)
                else:
                    await message.channel.send(result)
        except Exception as e:
            logger.exception(f'Fehler beim Ausführen von {command}: {e}')
            err_msg = f"Ein Fehler ist aufgetreten: {e}"
            if mode == ResponseMode.PRIVATE:
                dm = await message.author.create_dm()
                await dm.send(err_msg)
            else:
                await message.channel.send(err_msg)
    else:
        await message.channel.send(f"Unbekanntes Kommando: {command}")

---

5.2.5.4 Witzeaufruf via API

Mit dem Befehl !witz kann der Nutzer einen Witz in einer bestimmten Kategorie und Sprache abrufen. Die JokeAPI liefert entweder einen Einzelwitz oder einen zweiteiligen Witz (Setup + Pointe), der mit Discords Spoiler-Tags (||...||) dargestellt wird:

@register_command('witz', mode=ResponseMode.PUBLIC)
async def cmd_witz(message, *args):
    category = args[0] if args else "Any"
    lang = args[1] if len(args) > 1 else "de"
    api_url = f"https://v2.jokeapi.dev/joke/{category}?lang={lang}"
    headers = {"accept": "application/json"}

    try:
        response = requests.get(api_url, headers=headers)
        data = response.json()
        if response.status_code == 400:
            return f"Fehler beim Abrufen des Witzes: ({data['message']})."
        if not data:
            return "Kein Witz gefunden."

        if data["type"] == "single":
            return data["joke"]
        elif data["type"] == "twopart":
            return data["setup"] + "\n\n||" + data["delivery"] + "||"
        return "Unbekannter Witz-Typ."
    except Exception as e:
        logger.exception("Fehler beim Abrufen des Witzes")
        return f"Ein Fehler ist aufgetreten: {e}"

Argumente: - !witz → Zufallswitz auf Deutsch - !witz Programming en → englischer Programmierwitz

---

5.2.5.5 Hilfebefehl für Kategorien & Sprachen

Mit !witz-hilfe listet der Bot alle verfügbaren Witzkategorien und unterstützten Sprachen auf. Das hilft dem Nutzer, gültige Argumente für den !witz-Befehl zu finden.

@register_command('witz-hilfe', mode=ResponseMode.PUBLIC)
async def cmd_witz_hilfe(message, *args):
    try:
        kategorien = ["Any", "Programming", "Misc", "Dark", "Pun", "Spooky", "Christmas"]
        sprachen = ["de", "en", "es", "cs", "fr", "pt"]
        kategorien_text = "**Verfügbare Kategorien:**\n" + "\n".join(
            f"- `{k}`" for k in kategorien
        ) + "\n\n**Verfügbare Sprachen:**\n" + "\n".join(
            f"- `{k}`" for k in sprachen
        )
        return kategorien_text
    except Exception as e:
        logger.exception("Fehler beim Abrufen der Hilfe")
        return f"Ein Fehler ist aufgetreten: {e}"

Beispielausgabe:

**Verfügbare Kategorien:**
- `Any`
- `Programming`
- `Misc`
- `Dark`
- `Pun`
- `Spooky`
- `Christmas`

**Verfügbare Sprachen:**
- `de`
- `en`
- `es`
- `cs`
- `fr`
- `pt`

5.2.5.6 Ganzer Code

import os
import logging
import requests
from enum import Enum
from discord import Intents, ChannelType
from discord.ext import commands
from dotenv import load_dotenv

load_dotenv()
TOKEN = os.getenv('DISCORD_BOT_TOKEN')

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger('discord_bot')

intents = Intents.default()
intents.message_content = True

bot = commands.Bot(command_prefix='!', intents=intents)

class ResponseMode(Enum):
    PUBLIC = 'public'
    PRIVATE = 'private'

# Registry für Funktionen: name -> (Funktion, response_mode)
FUNCTION_REGISTRY = {}

def register_command(name: str, mode: ResponseMode = ResponseMode.PUBLIC):
    """Decorator zum Registrieren einer Funktion im Bot-Dispatcher mit ResponseMode"""
    def decorator(func):
        FUNCTION_REGISTRY[name] = (func, mode)
        return func
    return decorator

@bot.event
async def on_ready():
    logger.info(f'Bot eingeloggt als {bot.user} (ID: {bot.user.id})')

@bot.event
async def on_message(message):
    if message.author == bot.user:
        return

    content = message.content.strip()
    if not content.startswith(bot.command_prefix):
        return

    parts = content[len(bot.command_prefix):].split()
    command = parts[0].lower()
    args = parts[1:]

    logger.info(f'Empfangenes Kommando: {command} mit args: {args} von {message.author}')

    if command in FUNCTION_REGISTRY:
        func, mode = FUNCTION_REGISTRY[command]
        try:
            result = await func(message, *args)
            if result is not None:
                #privat oder öffentlich antworten
                if mode == ResponseMode.PRIVATE:
                    dm = await message.author.create_dm()
                    await dm.send(result)
                else:
                    await message.channel.send(result)
        except Exception as e:
            logger.exception(f'Fehler beim Ausführen von {command}: {e}')
            err_msg = f"Ein Fehler ist aufgetreten: {e}"
            if mode == ResponseMode.PRIVATE:
                dm = await message.author.create_dm()
                await dm.send(err_msg)
            else:
                await message.channel.send(err_msg)
    else:
        await message.channel.send(f"Unbekanntes Kommando: {command}")

@register_command('witz', mode=ResponseMode.PUBLIC)
async def cmd_witz(message, *args):
    category = args[0] if args else "Any"
    lang = args[1] if len(args) > 1 else "de"
    api_url = f"https://v2.jokeapi.dev/joke/{category}?lang={lang}"
    headers = {"accept": "application/json"}
    try:
        response = requests.get(api_url, headers=headers)
        data = response.json()
        if response.status_code == 400:
            return f"Fehler beim Abrufen des Witzes: ({data["message"]})."
        if not data:
            return "Kein Witz gefunden."
        
        if data["type"] == "single":
            return data["joke"]
        elif data["type"] == "twopart":
            return data["setup"] + "\n\n||" + data["delivery"] + "||"
        return
    except Exception as e:
        logger.exception("Fehler beim Abrufen des Witzes")
        return f"Ein Fehler ist aufgetreten: {e}"
    
@register_command('witz-hilfe', mode=ResponseMode.PUBLIC)
async def cmd_witz_hilfe(message, *args):
    try:
        kategorien = ["Any", "Programming", "Misc", "Dark", "Pun", "Spooky", "Christmas"]
        sprachen = ["de", "en", "es", "cs", "fr", "pt"]
        kategorien_text = "**Verfügbare Kategorien:**\n" + "\n".join(
        f"- `{k}`" for k in kategorien
        ) + "\n\n**Verfügbare Sprachen:**\n" + "\n".join(f"- `{k}`" for k in sprachen
        )
        return kategorien_text
    except Exception as e:
        logger.exception("Fehler beim Abrufen der Hilfe")
        return f"Ein Fehler ist aufgetreten: {e}"

if __name__ == '__main__':
    if not TOKEN:
        logger.error('DISCORD_BOT_TOKEN ist nicht gesetzt!')
        exit(1)
    bot.run(TOKEN)