Initial Upload
This commit is contained in:
commit
d108673aaa
7 changed files with 417 additions and 0 deletions
50
bot.py
Normal file
50
bot.py
Normal file
|
@ -0,0 +1,50 @@
|
|||
import discord
|
||||
from discord.ext import commands
|
||||
from datetime import datetime
|
||||
import logging
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
# Greedily Grabs all Intents.
|
||||
# Not the most secure, but the most bug free lol
|
||||
intents = discord.Intents(
|
||||
voice_states=True,
|
||||
messages=True,
|
||||
reactions=True,
|
||||
message_content=True,
|
||||
guilds=True,
|
||||
members=True,
|
||||
)
|
||||
|
||||
|
||||
# Pretty Simple setup for Bot.
|
||||
# https://github.com/Rapptz/discord.py/tree/master/examples
|
||||
class RoboBot(commands.Bot):
|
||||
def __init__(self, command_prefix, config, **options):
|
||||
self.config = config
|
||||
self.uptime = datetime.utcnow() # used for info :)
|
||||
self.src = config.get("github_url")
|
||||
# Set up the command prefix and intents from commands.Bot
|
||||
super().__init__(command_prefix=command_prefix, intents=intents, **options)
|
||||
|
||||
async def setup_hook(self): # Runs automatically
|
||||
# Load cogs from config. (/cogs/{cog})
|
||||
# https://discordpy.readthedocs.io/en/stable/ext/commands/cogs.html
|
||||
for cog in self.config.get("cogs", []):
|
||||
try:
|
||||
await self.load_extension("cogs." + cog) # f-strings better but.
|
||||
logger.info(f"Successfully loaded cog: {cog}")
|
||||
except Exception as e:
|
||||
logger.error(f"Failed to load cog {cog}: {e}")
|
||||
|
||||
async def on_ready(self): # also runs automatically, when bot is on discord
|
||||
logger.info(f"Logged in as {self.user.name} (ID: {self.user.id})")
|
||||
|
||||
# Create bot invite link
|
||||
if self.user:
|
||||
invite_url = discord.utils.oauth_url(
|
||||
self.user.id, permissions=discord.Permissions(administrator=True)
|
||||
)
|
||||
logger.info(f"Invite the bot to your server: {invite_url}")
|
||||
|
||||
logger.info("Bot is ready and operational.")
|
38
cogs/info.py
Normal file
38
cogs/info.py
Normal file
|
@ -0,0 +1,38 @@
|
|||
from discord.ext import commands
|
||||
from utils import handle_api_request
|
||||
import logging
|
||||
|
||||
from utils import about_me_embed
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
# Setup function for the cog
|
||||
async def setup(bot):
|
||||
await bot.add_cog(Info(bot))
|
||||
|
||||
|
||||
class Info(commands.Cog):
|
||||
def __init__(self, bot):
|
||||
self.bot = bot
|
||||
|
||||
@commands.hybrid_command(
|
||||
name="info_bot",
|
||||
description="Displays information about the bot.",
|
||||
aliases=["about", "bot_info"],
|
||||
) # fun fact, commands can't start with cog_ or bot_
|
||||
async def info_bot(self, ctx):
|
||||
# see utils.py
|
||||
embed = about_me_embed(self.bot)
|
||||
await ctx.send(embed=embed)
|
||||
|
||||
@commands.hybrid_command(
|
||||
name="server_status", description="Fetches the game server status"
|
||||
)
|
||||
async def server_status(self, ctx):
|
||||
result, message = handle_api_request("server_status")
|
||||
if result:
|
||||
status = result.get("status", "Unknown")
|
||||
await ctx.send(f"Server status: {status}")
|
||||
else:
|
||||
await ctx.send(f"Failed to retrieve server status: {message}")
|
110
cogs/robocraft.py
Normal file
110
cogs/robocraft.py
Normal file
|
@ -0,0 +1,110 @@
|
|||
from discord.ext import commands
|
||||
from utils import is_mod, handle_api_request
|
||||
import logging
|
||||
|
||||
# didn't use discord apparently so i didn't import it.
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
# Setup function for the cog, necessary for every cog.
|
||||
async def setup(bot):
|
||||
await bot.add_cog(Robocraft(bot))
|
||||
|
||||
|
||||
# Scary looking but it's actually really simple.
|
||||
# https://discordpy.readthedocs.io/en/stable/ext/commands/cogs.html
|
||||
class Robocraft(commands.Cog):
|
||||
def __init__(self, bot):
|
||||
self.bot = bot
|
||||
|
||||
# # This is what tells Discord.py about your commands.
|
||||
# Every Command requires one of these decorators.
|
||||
# all parameters optional, name defaults to function name
|
||||
@commands.hybrid_command(
|
||||
name="start_server",
|
||||
description="Starts the game server",
|
||||
aliases=["start", "boot", "bootup"],
|
||||
)
|
||||
# @mod_only() for commands.command will do the same as the whole if not is_mod block.
|
||||
# Unfortunately doesn't work for slash commands or hybrid commands.
|
||||
|
||||
async def start_server(self, ctx):
|
||||
# every command needs self and ctx (Context)
|
||||
# use ctx.send() to send a message
|
||||
# ctx.author for author (Member/User), ctx.guild for guild (Guild)
|
||||
# ctx.message for the command message (ctx.message.content for text of command ;) )
|
||||
|
||||
# Check if the user is a mod
|
||||
if not is_mod(ctx.author):
|
||||
logger.warning(
|
||||
f"{ctx.author} tried to start the server without permission."
|
||||
)
|
||||
await ctx.send("You do not have permission to use this command.")
|
||||
return
|
||||
|
||||
# Handle the API request
|
||||
result, message = handle_api_request("start_server")
|
||||
if result:
|
||||
logger.info(f"Server started by {ctx.author}.")
|
||||
await ctx.send("Server started successfully!")
|
||||
else:
|
||||
await ctx.send(f"Failed to start server: {message}")
|
||||
|
||||
@commands.hybrid_command(
|
||||
name="stop_server",
|
||||
description="Stops the game server",
|
||||
aliases=["stop", "shutdown"],
|
||||
)
|
||||
async def stop_server(self, ctx):
|
||||
# Check if the user is a mod
|
||||
if not is_mod(ctx.author):
|
||||
logger.warning(f"{ctx.author} tried to stop the server without permission.")
|
||||
await ctx.send("You do not have permission to use this command.")
|
||||
return
|
||||
|
||||
# Handle the API request
|
||||
result, message = handle_api_request("stop_server")
|
||||
if result:
|
||||
logger.info(f"Server stopped by {ctx.author}.")
|
||||
await ctx.send("Server stopped successfully!")
|
||||
else:
|
||||
await ctx.send(f"Failed to stop server: {message}")
|
||||
|
||||
@commands.hybrid_command(
|
||||
name="change_map",
|
||||
description="Changes the game map",
|
||||
aliases=["map", "new_map"],
|
||||
)
|
||||
async def change_map(self, ctx, *, map_name: str):
|
||||
# takes a str, oooo
|
||||
# , *, means it greedily captures the rest of the string
|
||||
# str could also be int, discord.User, discord.Member, or really anything you want.
|
||||
# https://discordpy.readthedocs.io/en/stable/ext/commands/commands.html#parameters
|
||||
# entire API docsis sacred god bless danny
|
||||
|
||||
# Check if the user is a mod
|
||||
if not is_mod(ctx.author):
|
||||
logger.warning(f"{ctx.author} tried to change the map without permission.")
|
||||
await ctx.send("You do not have permission to use this command.")
|
||||
return
|
||||
|
||||
# Handle the API request with the map name
|
||||
# You can put whatever else you need here, mapping to whatever else the user puts in
|
||||
# For here, it's just a basic dict with key map and value whatever the hell the user sends
|
||||
|
||||
# Ideally, this should contain (or grab) a list of maps from the API in a way the backend understands
|
||||
# But i don't know how you have that set up.
|
||||
|
||||
# To make it random,
|
||||
# from random import choice
|
||||
# map_name = choice(maps)
|
||||
# or whatever
|
||||
|
||||
data = {"map": map_name}
|
||||
|
||||
result, message = handle_api_request("change_map", data)
|
||||
if result:
|
||||
logger.info(f"Map changed to {map_name} by {ctx.author}.")
|
||||
await ctx.send(f"Map changed to {map_name}!")
|
||||
else:
|
||||
await ctx.send(f"Failed to change map: {message}")
|
28
config.yaml
Normal file
28
config.yaml
Normal file
|
@ -0,0 +1,28 @@
|
|||
# Bot configuration
|
||||
prefix: ""
|
||||
token: "Your token here!"
|
||||
github_url: ""
|
||||
|
||||
# A list of cogs to load at startup
|
||||
# These are found in /cogs/
|
||||
cogs:
|
||||
- robocraft
|
||||
- info
|
||||
|
||||
# API configuration
|
||||
api_url: "https://api.whofuckingknows.com/"
|
||||
# Add routes here.
|
||||
routes:
|
||||
start_server: "/server/start"
|
||||
stop_server: "/server/stop"
|
||||
change_map: "/server/change_map"
|
||||
server_status: "/server/status"
|
||||
|
||||
# Mod IDs (for permissions)
|
||||
# Replicate if you want admins, gamemasters, etc.
|
||||
# see utils.py for details
|
||||
mods:
|
||||
- "123456789012345678"
|
||||
- "987654321098765432"
|
||||
|
||||
|
34
main.py
Normal file
34
main.py
Normal file
|
@ -0,0 +1,34 @@
|
|||
import logging
|
||||
from bot import RoboBot
|
||||
from utils import load_config
|
||||
|
||||
# Initialize logging
|
||||
logging.basicConfig(
|
||||
level=logging.INFO,
|
||||
format="%(asctime)s - %(levelname)s - %(message)s",
|
||||
datefmt="%H:%M:%S",
|
||||
)
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
def main():
|
||||
# Load configuration
|
||||
config = load_config()
|
||||
|
||||
# Get bot token and prefix from the config
|
||||
token = config.get("token") # .get() returns None if the key is not found
|
||||
prefix = config.get("prefix", "!")
|
||||
|
||||
if not token:
|
||||
logger.error("Bot token not found in config.yaml")
|
||||
return
|
||||
|
||||
# Create the bot instance
|
||||
bot = RoboBot(command_prefix=prefix, config=config)
|
||||
|
||||
# Run the bot
|
||||
bot.run(token)
|
||||
|
||||
|
||||
if __name__ == "__main__": # always do this lol
|
||||
main()
|
5
requirements.txt
Normal file
5
requirements.txt
Normal file
|
@ -0,0 +1,5 @@
|
|||
discord.py>=2.4.0
|
||||
humanfriendly==10.0
|
||||
psutil==6.0.0
|
||||
PyYAML==6.0.2
|
||||
Requests==2.32.3
|
152
utils.py
Normal file
152
utils.py
Normal file
|
@ -0,0 +1,152 @@
|
|||
import yaml
|
||||
import logging
|
||||
import requests
|
||||
import discord
|
||||
import datetime
|
||||
import psutil
|
||||
import humanfriendly
|
||||
|
||||
# imagine a world where devs imported exactly what they needed
|
||||
# but it's just not possible /s
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
# Basic Loading with some error handling
|
||||
def load_config():
|
||||
try:
|
||||
with open("config.yaml", "r") as file:
|
||||
config = yaml.safe_load(file)
|
||||
logger.info("Configuration file loaded successfully.")
|
||||
return config
|
||||
except FileNotFoundError:
|
||||
logger.error("Config file not found. Please ensure 'config.yaml' exists.")
|
||||
return {}
|
||||
except yaml.YAMLError as exc:
|
||||
logger.error(f"Error parsing YAML config: {exc}")
|
||||
return {}
|
||||
|
||||
|
||||
# Unused Saving
|
||||
def save_config(config):
|
||||
try:
|
||||
with open("config.yaml", "w") as file:
|
||||
yaml.safe_dump(config, file)
|
||||
logger.info("Configuration file saved successfully.")
|
||||
except Exception as e:
|
||||
logger.error(f"Error saving config: {e}")
|
||||
|
||||
|
||||
# is_mod, takes a Member, returns a bool
|
||||
# not the best as DM'ing the bot won't produce a list of roles and thus won't count mods.
|
||||
# i just don't want to rewrite the yaml
|
||||
def is_mod(member):
|
||||
config = load_config()
|
||||
mod_ids = config.get("mods", [])
|
||||
|
||||
if str(member.id) in mod_ids:
|
||||
return True
|
||||
|
||||
for role in member.roles:
|
||||
if str(role.id) in mod_ids:
|
||||
return True
|
||||
|
||||
return False
|
||||
|
||||
|
||||
# Unused Decorator for commands.
|
||||
# You'll come to learn there's a good amount of stuff that's just easier with commands rather than slash commands.
|
||||
# See robocraft.py for some examples
|
||||
def mod_only():
|
||||
def wrapper(func):
|
||||
async def wrapped(self, ctx, *args, **kwargs):
|
||||
if not is_mod(ctx.author):
|
||||
logger.warning(
|
||||
f"{ctx.author} tried to access {ctx.command} without permission."
|
||||
)
|
||||
await ctx.send("You do not have permission to use this command.")
|
||||
return
|
||||
return await func(self, ctx, *args, **kwargs)
|
||||
|
||||
return wrapped
|
||||
|
||||
return wrapper
|
||||
|
||||
|
||||
# Gross function, but it works.
|
||||
def handle_api_request(route_key, data=None):
|
||||
# route_key is your YAML key
|
||||
# Data defaults to None so we don't have to set it
|
||||
|
||||
config = load_config()
|
||||
|
||||
# If you wanted to set up multiple API URLS, you could use them like routes (a list), or just copy how api_url is setup.
|
||||
api_url = config.get("api_url")
|
||||
routes = config.get("routes", {})
|
||||
|
||||
if not api_url: # you fucked up
|
||||
logger.error("API URL is not configured.")
|
||||
return None, "API URL is not configured."
|
||||
|
||||
if route_key not in routes: # you also fucked up
|
||||
logger.warning(f"Invalid route key: {route_key}")
|
||||
return None, "Invalid API route key."
|
||||
|
||||
route = routes[route_key]
|
||||
url = f"{api_url}{route}"
|
||||
|
||||
try: # Very prone to errors. Server shite.
|
||||
if data:
|
||||
response = requests.post(url, json=data)
|
||||
else:
|
||||
response = requests.post(url)
|
||||
|
||||
if response.status_code == 200:
|
||||
logger.info(f"API request to {url} succeeded.")
|
||||
return response.json(), "Success!"
|
||||
else:
|
||||
return None, f"Error {response.status_code}: {response.text}"
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Request failed: {e}")
|
||||
return None, f"Request failed: {str(e)}"
|
||||
|
||||
|
||||
# Baller about me embed, don't worry too much about this
|
||||
# Good insight into how embeds work though.
|
||||
def about_me_embed(bot):
|
||||
cpu = psutil.cpu_percent()
|
||||
memory = psutil.virtual_memory().percent
|
||||
disk = psutil.disk_usage("/")
|
||||
disk = {
|
||||
"used_gb": round(disk.used / (1024.0**3), 2),
|
||||
"available_gb": round(disk.free / (1024.0**3), 2),
|
||||
"total_gb": round(disk.total / (1024.0**3), 2),
|
||||
}
|
||||
uptime = datetime.utcnow() - bot.uptime
|
||||
if bot.src: # remember that?
|
||||
description = f"This bot is [**Open Source!**](<{bot.src}>)"
|
||||
else:
|
||||
description = "A Discord Bot Written in Python!" # idk
|
||||
|
||||
# make the embed
|
||||
embed = discord.Embed(
|
||||
title=f"About: {bot.user.name}",
|
||||
description=description,
|
||||
color=discord.Color.random(),
|
||||
)
|
||||
# add fields
|
||||
embed.add_field(name="CPU: ", value=f"{cpu}% of {psutil.cpu_count()} cores")
|
||||
embed.add_field(
|
||||
name="Memory: ",
|
||||
value=f"{memory}% of {round(psutil.virtual_memory().total / (1024.0 ** 3), 2)} GB",
|
||||
)
|
||||
embed.add_field(
|
||||
name="Disk Usage: ", value=f"{disk['used_gb']} GB used of {disk['total_gb']} GB"
|
||||
)
|
||||
embed.add_field(name="Uptime: ", value=f"{humanfriendly.format_timespan(uptime)}")
|
||||
|
||||
# add avatar and footer
|
||||
embed.set_image(url=bot.user.display_avatar.url)
|
||||
embed.set_footer(text="Made with :3 by Flo ❤️", icon_url=bot.user.display_avatar.url)
|
||||
return embed
|
Loading…
Reference in a new issue