public
imalexnord
read
Discord-Hetzner-Bot
No description yet
Languages
Repository composition by tracked source files.
Python
100%
Create file
Wiki Documentation
Clone
https://nobgit.com/user/imalexnord/discord-hetzner-bot.git
ssh://[email protected]:2222/user/imalexnord/discord-hetzner-bot.git
Commit
Refactor game and Discord bot logic, remove samples
Improves GameData initialization with better error handling and optional snapshot/volume support. Refactors Discord bot to use updated discord.py API, improves command parsing, and provides clearer status reporting. Removes unused sample settings and token files.
25636c1
game.py | 125 +++++++++++++++++++++++++++++++----------------------
gameserver.py | 111 ++++++++++++++++++++++++++++-------------------
settings.py.sample | 11 -----
tokens.py.sample | 5 ---
4 files changed, 141 insertions(+), 111 deletions(-)
delete mode 100644 settings.py.sample
delete mode 100644 tokens.py.sample
Diff
diff --git a/game.py b/game.py
index 2eb3448..640a00b 100644
--- a/game.py
+++ b/game.py
@@ -4,83 +4,107 @@ from hcloud.servers.domain import Server
from hcloud.server_types.domain import ServerType
from hcloud.volumes.domain import Volume
from hcloud.locations.domain import Location
-import time
import asyncio
class GameData:
running = False
-
+
def __init__(self, token, name, servertype, snapshot, location, volume):
self.name = name
- self.servertype = servertype
+ self.servertype = servertype.strip() if isinstance(servertype, str) else servertype
self.client = Client(token=token)
-
- # get id from snapshot
- images = self.client.images.get_all()
- for i in images:
- if i.data_model.description == snapshot:
- imageId = i.data_model.id
- self.snapshot = imageId
-
- # get id from location
+
+ # --- snapshot/image (optional) ---
+ # If snapshot is None => we'll fall back to Ubuntu 24.04 in start()
+ self.snapshot = None
+ if snapshot is not None:
+ images = self.client.images.get_all()
+ for img in images:
+ # support matching by description OR name
+ if img.data_model.description == snapshot or img.data_model.name == snapshot:
+ self.snapshot = img.data_model.id
+ break
+ if self.snapshot is None:
+ raise ValueError(f"Snapshot/Image not found: {snapshot}")
+
+ # --- location (required) ---
+ self.location = None
locations = self.client.locations.get_all()
- for i in locations:
- if i.data_model.name == location:
- locationId = i.data_model.id
- self.location = locationId
-
- # get id from volume
- volumes = self.client.volumes.get_all()
- for i in volumes:
- if i.data_model.name == volume:
- if i.data_model.location.data_model.id == self.location:
- volumeId = i.data_model.id
- self.volume = volumeId
-
- # check for running server
+ for loc in locations:
+ if loc.data_model.name == location:
+ self.location = loc.data_model.id
+ break
+ if self.location is None:
+ raise ValueError(f"Location not found: {location}")
+
+ # --- volume (optional) ---
+ self.volume = None
+ if volume is not None:
+ volumes = self.client.volumes.get_all()
+ for vol in volumes:
+ if vol.data_model.name == volume and vol.data_model.location.data_model.id == self.location:
+ self.volume = vol.data_model.id
+ break
+ if self.volume is None:
+ raise ValueError(f"Volume not found in location {location}: {volume}")
+
+ # --- check for existing server by name ---
servers = self.client.servers.get_all()
self.server = None
- for i in servers:
- if i.data_model.name == name:
- self.server = i.data_model
+ for s in servers:
+ if s.data_model.name == name:
+ self.server = s # keep the Server object
self.running = True
print("existing server " + self.server.status)
async def start(self):
- if self.running == False:
+ if not self.running:
print("Starting " + self.name)
+
+ # Choose image:
+ # - if snapshot provided => use that snapshot image id
+ # - else => boot from Ubuntu 24.04 (Hetzner public image name)
+ image = Image(self.snapshot) if self.snapshot is not None else "ubuntu-24.04"
+
+ # Attach volume only if configured
+ volumes = [Volume(self.volume)] if self.volume is not None else None
+
response = self.client.servers.create(
- self.name,
+ name=self.name,
server_type=ServerType(name=self.servertype),
- image=Image(self.snapshot),
+ image=image,
location=Location(self.location),
- volumes=[Volume(self.volume)]
+ volumes=volumes
)
- self.server = response.server;
- #while self.server.status != Server.STATUS_RUNNING:
- # print(self.server.status)
- # time.sleep(2)
- # serv = self.client.servers.get_by_id(self.server.id)
- # self.server = serv
- #print("Server is now running")
+ self.server = response.server
self.running = True
else:
print(self.name + " is already running")
async def stop(self):
- if self.running == True:
+ if self.running:
print("Stopping " + self.name)
+
self.client.servers.shutdown(self.server)
- time.sleep(30)
- while self.server.status != Server.STATUS_OFF:
- print(self.server.status)
- time.sleep(5)
+
+ # Give it time to begin shutdown, then poll status
+ await asyncio.sleep(5)
+
+ while True:
serv = self.client.servers.get_by_id(self.server.id)
self.server = serv
+ print(self.server.status)
+ if self.server.status == Server.STATUS_OFF:
+ break
+ await asyncio.sleep(5)
+
print("Server is now stopped")
- time.sleep(1)
- self.client.volumes.detach(Volume(self.volume))
- time.sleep(1)
+
+ # Detach volume only if configured
+ if self.volume is not None:
+ self.client.volumes.detach(Volume(self.volume))
+ await asyncio.sleep(1)
+
self.client.servers.delete(self.server)
self.running = False
self.server = None
@@ -89,7 +113,7 @@ class GameData:
def status(self):
msg = self.name
- if self.server == None:
+ if self.server is None:
msg += " isn't running"
else:
self.server = self.client.servers.get_by_id(self.server.id)
@@ -102,5 +126,4 @@ class GameData:
return msg
def isRunning(self):
- return self.running
-
+ return self.running
\ No newline at end of file
diff --git a/gameserver.py b/gameserver.py
index b7dd206..0612cd5 100755
--- a/gameserver.py
+++ b/gameserver.py
@@ -1,78 +1,101 @@
-#! /usr/bin/python3
-from tokens import *
-from settings import *
+from tokens import DISCORD_TOKEN, HETZNER_API_TOKEN
from game import *
-from discord import Game
+from settings import BOT_PREFIX, GAMES
import discord
import sys
-client = discord.Client()
+intents = discord.Intents.default()
+intents.message_content = True # REQUIRED for reading commands
+
+client = discord.Client(intents=intents)
+
@client.event
async def on_message(message):
# we do not want the bot to reply to itself
if message.author == client.user:
return
-
+
if message.content.startswith(BOT_PREFIX + "help"):
- msg = ("!help - prints this help\n"
- "!ping - simple online check\n"
- "!start GAME - starts a server for GAME\n"
- "!stop GAME - stops the server for GAME\n"
- "!status [GAME] - prints basic infos about the server for GAME\n"
- "!exit - exit bot script - should restart it when used with systemd service unit"
- ).format(message)
- await client.send_message(message.channel, msg)
+ msg = (
+ "!help - prints this help\n"
+ "!ping - simple online check\n"
+ "!start GAME - starts a server for GAME\n"
+ "!stop GAME - stops the server for GAME\n"
+ "!status [GAME] - prints basic infos about the server for GAME\n"
+ "!exit - exit bot script - should restart it when used with systemd service unit"
+ )
+ await message.channel.send(msg)
+
elif message.content.startswith(BOT_PREFIX + "ping"):
msg = "I'm online {0.author.mention}".format(message)
- await client.send_message(message.channel, msg)
+ await message.channel.send(msg)
+
elif message.content.startswith(BOT_PREFIX + "start"):
cmd = message.content.split("start", 1)[1].lower().split()
if not cmd:
- msg = ("Syntax error:\n"
- "!start GAME"
- )
- else:
- for i in GAMES:
- if i.name.lower() == cmd[0]:
- if i.isRunning() == False:
- await client.change_presence(game=Game(name="starting server..."))
- await i.start()
- await client.change_presence(game=Game(name="ready"))
- msg = i.status()
- await client.send_message(message.channel, msg)
+ msg = "Syntax error:\n!start GAME"
+ await message.channel.send(msg)
+ return
+
+ msg = f"Game '{cmd[0]}' not found."
+ for i in GAMES:
+ if i.name.lower() == cmd[0]:
+ if i.isRunning() is False:
+ await client.change_presence(activity=discord.Game(name="starting server..."))
+ await i.start()
+ await client.change_presence(activity=discord.Game(name="ready"))
+ msg = i.status()
+ break
+
+ await message.channel.send(msg)
+
elif message.content.startswith(BOT_PREFIX + "stop"):
cmd = message.content.split("stop", 1)[1].lower().split()
if not cmd:
- msg = ("Syntax error:\n"
- "!stop GAME"
- )
- else:
- for i in GAMES:
- if i.name.lower() == cmd[0]:
- if i.isRunning() == True:
- await client.change_presence(game=Game(name="stopping server..."))
- await i.stop()
- await client.change_presence(game=Game(name="ready"))
- msg = i.status()
- await client.send_message(message.channel, msg)
+ msg = "Syntax error:\n!stop GAME"
+ await message.channel.send(msg)
+ return
+
+ msg = f"Game '{cmd[0]}' not found."
+ for i in GAMES:
+ if i.name.lower() == cmd[0]:
+ if i.isRunning() is True:
+ await client.change_presence(activity=discord.Game(name="stopping server..."))
+ await i.stop()
+ await client.change_presence(activity=discord.Game(name="ready"))
+ msg = i.status()
+ break
+
+ await message.channel.send(msg)
+
elif message.content.startswith(BOT_PREFIX + "status"):
cmd = message.content.split("status", 1)[1].lower().split()
+
+ # If no game specified, print all statuses in one message
+ if not cmd:
+ msg = "\n".join(i.status() for i in GAMES)
+ await message.channel.send(msg)
+ return
+
+ msg = f"Game '{cmd[0]}' not found."
for i in GAMES:
- if not cmd:
- msg = i.status()
- elif i.name.lower() == cmd[0]:
+ if i.name.lower() == cmd[0]:
msg = i.status()
- await client.send_message(message.channel, msg)
+ break
+ await message.channel.send(msg)
+
elif message.content.startswith(BOT_PREFIX + "exit"):
sys.exit("exit requested")
+
@client.event
async def on_ready():
print("Logged in as")
print(client.user.name)
print(client.user.id)
print("------")
- await client.change_presence(game=Game(name="ready"))
+ await client.change_presence(activity=discord.Game(name="ready"))
+
client.run(DISCORD_TOKEN)
diff --git a/settings.py.sample b/settings.py.sample
deleted file mode 100644
index 1778eb5..0000000
--- a/settings.py.sample
+++ /dev/null
@@ -1,11 +0,0 @@
-from game import *
-from tokens import *
-
-
-BOT_PREFIX = "!"
-
-
-sample = GameData(HETZNER_API_TOKEN, "Name", "ServerType", "Snapshot", "Volume")
-
-
-GAMES = [sample]
diff --git a/tokens.py.sample b/tokens.py.sample
deleted file mode 100644
index d6917d4..0000000
--- a/tokens.py.sample
+++ /dev/null
@@ -1,5 +0,0 @@
-
-DISCORD_TOKEN = ''
-
-HETZNER_API_TOKEN = ''
-