mirror of
https://github.com/opus-tango/minecraft-server-discord-bot.git
synced 2026-03-20 12:05:22 +00:00
244 lines
8.2 KiB
Python
244 lines
8.2 KiB
Python
import json
|
|
import os
|
|
import discord
|
|
from discord.ext import commands
|
|
import subprocess
|
|
from dotenv import load_dotenv
|
|
import time
|
|
|
|
load_dotenv()
|
|
|
|
TOKEN = os.getenv('DISCORD_TOKEN')
|
|
BASE_PATH = os.getenv('BASE_PATH', './')
|
|
print(TOKEN)
|
|
|
|
chat_channel = None
|
|
command_channel = None
|
|
command_prefix = None
|
|
with open('config.json', 'r') as f:
|
|
config = json.load(f)
|
|
chat_channel = config['config']['chat_channel']
|
|
command_channel = config['config']['command_channel']
|
|
command_prefix = config['config']['command_prefix']
|
|
|
|
intents = discord.Intents.default()
|
|
intents.message_content = True
|
|
bot = commands.Bot(command_prefix=command_prefix, intents=intents)
|
|
|
|
active_server = None
|
|
|
|
'''
|
|
Handle events (messages mostly)
|
|
'''
|
|
@bot.event
|
|
async def on_ready():
|
|
print(f'Logged in as {bot.user})')
|
|
|
|
@bot.event
|
|
async def on_message(message):
|
|
print(f'Message from {message.author}: {message.content}')
|
|
if message.author == bot.user:
|
|
return
|
|
if message.channel.name == chat_channel and not message.content.startswith(command_prefix):
|
|
if not active_server:
|
|
await message.channel.send("No server is currently active.")
|
|
return
|
|
await send_message(message)
|
|
if message.channel.name == command_channel and message.content.startswith(command_prefix):
|
|
await bot.process_commands(message)
|
|
|
|
async def send_message(message):
|
|
global active_server
|
|
if not active_server:
|
|
await message.channel.send("No server is currently active.")
|
|
return
|
|
minecraft_message = "[" + message.author.name + "] " + message.content
|
|
process = subprocess.Popen(f"screen -S {active_server} -p 0 -X stuff '{minecraft_message}\n'", shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, universal_newlines=True)
|
|
stdout, stderr = process.communicate()
|
|
|
|
'''
|
|
Handle commands
|
|
'''
|
|
@bot.command()
|
|
async def startserver(ctx, *args):
|
|
'''
|
|
Start a server
|
|
args:
|
|
servername: The name of the server
|
|
|
|
'''
|
|
global active_server
|
|
# Check if the user is an admin
|
|
if not ctx.author.guild_permissions.administrator:
|
|
await ctx.send("You don't have permission to run this command.")
|
|
return
|
|
# Validate and get the server name
|
|
if len(args) == 0:
|
|
await ctx.send("Please provide a server name.")
|
|
return
|
|
if len(args) > 1:
|
|
await ctx.send("Please provide only one server name.")
|
|
return
|
|
servername = args[0]
|
|
|
|
# Check if the server is already running
|
|
if active_server is not None:
|
|
await ctx.send("A server is already running. If you think this is an error, run !list to refresh the list of servers and try again.")
|
|
return
|
|
|
|
# Load server data from json
|
|
with open('servers.json', 'r') as f:
|
|
servers = json.load(f)
|
|
servers = servers['servers']
|
|
# Find the server in the json
|
|
server = None
|
|
for s in servers:
|
|
if s['servername'] == servername:
|
|
server = s
|
|
break
|
|
if server is None:
|
|
await ctx.send("Server not found.")
|
|
return
|
|
|
|
# Start the server
|
|
await ctx.send("Starting server, please wait...")
|
|
path = BASE_PATH + server['path'] if BASE_PATH else server['path']
|
|
command = f"cd {path} && screen -dmS {servername} {server['startcommand']}"
|
|
print(f"Running command: {command}")
|
|
process = subprocess.Popen(command, shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, universal_newlines=True)
|
|
# Wait for the server to start then check if the screen instance is still live
|
|
stdout, stderr = process.communicate()
|
|
time.sleep(3)
|
|
process = subprocess.Popen(f"screen -ls | grep {servername}", shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, universal_newlines=True)
|
|
stdout, stderr = process.communicate()
|
|
print(stdout)
|
|
if len(stdout) == 0:
|
|
await ctx.send("Server failed to start. Minecraft screen instance terminated within 3 seconds.")
|
|
return
|
|
await ctx.send("Server started.")
|
|
await ctx.send(stdout)
|
|
active_server = servername
|
|
|
|
@bot.command()
|
|
async def stopserver(ctx):
|
|
'''
|
|
Stop the server by sending 'stop' to the server
|
|
|
|
'''
|
|
global active_server
|
|
# Check if the user is an admin
|
|
if not ctx.author.guild_permissions.administrator:
|
|
await ctx.send("You don't have permission to run this command.")
|
|
return
|
|
# Stop the server
|
|
await ctx.send("Stopping server...")
|
|
process = subprocess.Popen(f"screen -S {active_server} -p 0 -X stuff 'stop\n'", shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, universal_newlines=True)
|
|
stdout, stderr = process.communicate()
|
|
await ctx.send(stdout)
|
|
await ctx.send("Server stopped.")
|
|
active_server = None
|
|
|
|
@bot.command()
|
|
async def command(ctx, *args):
|
|
'''
|
|
Send a command to the server
|
|
args:
|
|
command: The command to send
|
|
|
|
'''
|
|
global active_server
|
|
# Check if the user is an admin
|
|
if not ctx.author.guild_permissions.administrator:
|
|
await ctx.send("You don't have permission to run this command.")
|
|
return
|
|
# Send the command to the server
|
|
await ctx.send("Sending command...")
|
|
command = " ".join(args)
|
|
subprocess.Popen(f"screen -S {active_server} -p 0 -X stuff '{command}\n'", shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, universal_newlines=True)
|
|
|
|
@bot.command()
|
|
async def list(ctx):
|
|
'''
|
|
Lists currently running server and available servers
|
|
|
|
'''
|
|
global active_server
|
|
# Load server data from json
|
|
servers = None
|
|
with open('servers.json', 'r') as f:
|
|
servers = json.load(f)
|
|
servers = servers['servers']
|
|
if servers is None:
|
|
await ctx.send("No servers found in the servers.json file.")
|
|
return
|
|
|
|
# Get the currently running screens
|
|
process = subprocess.Popen(f"screen -ls", shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, universal_newlines=True)
|
|
stdout, stderr = process.communicate()
|
|
|
|
# Get the currently active server
|
|
currentservers = []
|
|
for s in servers:
|
|
if s['servername'] in stdout:
|
|
currentservers.append(s)
|
|
if len(currentservers) == 0:
|
|
await ctx.send("No server is currently active.")
|
|
active_server = None
|
|
elif len(currentservers) > 1:
|
|
await ctx.send("Multiple servers are currently active. This should not happen. Please report this to the server owner.")
|
|
await ctx.send(f"Currently active servers: {', '.join([s['servername'] for s in currentservers])}")
|
|
active_server = None
|
|
else:
|
|
await ctx.send(f"Server {currentservers[0]['servername']} is currently active.")
|
|
if len(currentservers) > 0 and currentservers[0]['servername'] != active_server:
|
|
active_server = currentservers[0]['servername']
|
|
|
|
# List the available servers
|
|
listofservers = 'Available servers: '+ "\n".join([s["servername"] for s in servers])
|
|
await ctx.send(listofservers)
|
|
|
|
@bot.command()
|
|
async def forcestopserver(ctx):
|
|
'''
|
|
Force stop the server by sending keyboard interrupt
|
|
|
|
'''
|
|
global active_server
|
|
# Check if the user is an admin
|
|
if not ctx.author.guild_permissions.administrator:
|
|
await ctx.send("You don't have permission to run this command.")
|
|
return
|
|
# Stop the server
|
|
await ctx.send("Stopping server...")
|
|
process = subprocess.Popen(f"screen -S {active_server} -p 0 -X stuff $'\003'", shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, universal_newlines=True)
|
|
process = subprocess.Popen(f"screen -S {active_server} -p 0 -X stuff $'\003'", shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, universal_newlines=True)
|
|
process = subprocess.Popen(f"screen -S {active_server} -p 0 -X stuff $'\003'", shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, universal_newlines=True)
|
|
stdout, stderr = process.communicate()
|
|
await ctx.send(stdout)
|
|
await ctx.send("Server stopped.")
|
|
active_server = None
|
|
|
|
@bot.command()
|
|
async def ping(ctx):
|
|
'''
|
|
Ping the bot
|
|
|
|
'''
|
|
await ctx.send('Pong!')
|
|
|
|
# @bot.command()
|
|
# async def help(ctx):
|
|
# string = "Available commands:\n"
|
|
# for command in bot.commands:
|
|
# string += f"{command.name}: {command.description}\n"
|
|
# await ctx.send(string)
|
|
|
|
@bot.command()
|
|
async def botstop(ctx):
|
|
'''
|
|
Stop the bot
|
|
|
|
'''
|
|
await bot.close()
|
|
|
|
bot.run(TOKEN) |