diff --git a/.dockerignore b/.dockerignore new file mode 100644 index 0000000..fcd6639 --- /dev/null +++ b/.dockerignore @@ -0,0 +1,6 @@ +__pycache__/ +.venv/ +venv/ +.git/ +.gitignore +.env \ No newline at end of file diff --git a/resources/image.png b/.forgejo/resources/image.png similarity index 100% rename from resources/image.png rename to .forgejo/resources/image.png diff --git a/Dockerfile b/Dockerfile index 70b1e34..a077360 100644 --- a/Dockerfile +++ b/Dockerfile @@ -8,7 +8,9 @@ COPY schema.sql . ADD src src RUN apt update && apt upgrade -y && apt install -y sqlite3 -RUN pip install --upgrade pip && pip install -r requirements.txt +RUN pip install --upgrade pip && pip install --no-cache-dir -r requirements.txt + +ENV PYTHONPATH=/app/src ENTRYPOINT [ "/app/entrypoint.sh" ] -CMD [ "python3", "-u", "src/mstbot.py" ] +CMD ["python", "-m", "mstbot"] diff --git a/README.md b/README.md index f474182..c12b213 100644 --- a/README.md +++ b/README.md @@ -2,7 +2,7 @@ # MinecraftStatus Minecraft Status discord bot that adds status banners to your guild's sidebar. See below for example. -![Server Status Banners](resources/image.png) +![Server Status Banners](.forgejo/resources/image.png) ### Commands ``` diff --git a/src/mstbot.py b/src/mstbot.py deleted file mode 100644 index 09814e3..0000000 --- a/src/mstbot.py +++ /dev/null @@ -1,676 +0,0 @@ -# TODO: in readme, add badge for guild count, total player count -# TODO: annotate return types everywhere -# TODO: split cogs into separate files, add utils file, remove globals -# TODO: wrap all SQL queries with a func to determine if server has been set up or not. If fetchone returns None, log helpful message to run $setup. -# TODO: save invite link to README and show query port enable instructions -# TODO: Feature: allow discord users to associate their mc usernames, change server nickname? -# TODO: Feature: allow other users to query playtime by username, default to their own. This and above require new columns in names table. -# TODO: If on startup the schema has changed, create new db and attempt to migrate old data into new schema - -import asyncio -import os -import re -import socket -import unicodedata - -from contextlib import contextmanager -from datetime import datetime, timezone -from typing import Union, List - -import sqlite3 -import discord -from discord.ext import commands -from mcstatus import JavaServer - -from erl import EndpointRateLimiter - -async def change_channel_name(channel, name: str): - await channel.edit(name=name) - -limiter = EndpointRateLimiter(301, change_channel_name) - -# Bot Initialization -intents = discord.Intents.default() -intents.messages = True -intents.message_content = True -bot = commands.Bot(command_prefix='$', intents=intents) -servers = [] - -@bot.event -async def on_ready(): - print(f'Logged in as {bot.user}') - - with sqlite_conn() as conn: - rows = conn.execute( - "SELECT id, ip, announce_joins, announce_joins_id FROM servers" - ).fetchall() - - for row in rows: - if get_server_by_id(row[0]) is None: - print(f"Kicked: Bot was removed from {row[0]}. Cleaning up.") - await do_bot_cleanup(row[0]) - continue - - bot.loop.create_task(status_task(row[0])) - servers.append(row[0]) - announcement_status = "Not announcing" if not row[2] else f"Announcing in <#{row[3]}>" - print(f"Now querying {get_server_by_id(row[0]).name}. {announcement_status} for {row[1]}") - - -@bot.event -async def on_command_error(ctx: commands.Context, error: commands.CommandError): - command = ctx.message.content.split()[0][1:] # Extract command name - - if isinstance(error, commands.CommandNotFound): - await safe_send(f'Command "{command}" does not exist. Use `{ctx.prefix}help` for available commands.', ctx) - elif isinstance(error, commands.MissingRequiredArgument): - await log(ctx, f'Command "{command}" is missing required arguments. Use `{ctx.prefix}help {command}` for details.') - elif isinstance(error, commands.MissingPermissions): - await safe_send(f'You lack permissions: {", ".join(error.missing_perms)}.', ctx) - elif isinstance(error, commands.CommandInvokeError) and isinstance(error.original, socket.gaierror): - await log(ctx, "Could not connect, is the server down?") - else: - raise error - - -class Admin(commands.Cog): - """Admin""" - - @commands.command(brief="Sets up new Minecraft server querier", - description="Sets up new Minecraft server querier. Removes previous querier, if applicable. Use a valid domain name or IP address. The default port (25565) is used unless otherwise specified. Specify a channel ID to announce player joins there.") - @commands.has_permissions(administrator=True) - async def setup(self, ctx: discord.ext.commands.context.Context, ip: str, port: int = 25565, query_port: int = 25565, annChanID: Union[int, str, None] = None): - """Sets up a new Minecraft server querier.""" - val_ip, val_port, val_query_port = await validate_ip_port(ctx, ip, port, query_port) - if not val_ip: - return - - server_port_ip = f"{val_ip}:{val_port}" - await log(ctx, f"Attempting to establish connection to {server_port_ip} with query port {val_query_port}...") - mc = JavaServer(val_ip, val_port, query_port=val_query_port) - - status = None - query = None - try: - status = mc.status() - query = mc.query() - names = query.players.list - except asyncio.exceptions.TimeoutError: - await log(ctx, ( - "The setup Query protocal ping failed, but the Minecraft server appears to be accessible.\n" - "Does the server.properties file contain `enable-query=true`? Changes to server.properties do not take effect until the Minecraft server is restarted.\n" - f"Is UDP port {val_query_port} forwarded and open on the firewall?" - )) - if status is None: - await log(ctx, ( - "A Status protocol ping also failed.\n" - "Does the server.properties file contain `enable-status=true`?\n" - f"Is TCP port {val_port} forwarded and open on the firewall?" - )) - except ConnectionRefusedError: - await log(ctx, ( - f"Setup query failed. I could not reach a Minecraft server at {server_port_ip}.\n" - f"Is TCP port {val_port} forwarded and open on the firewall?\n" - f"Can you and a friend (on a different network) connect to and play on {server_port_ip}?" - )) - else: - with sqlite_conn() as conn: - conn.execute(""" - INSERT INTO servers (id, ip, port) - VALUES (?, ?, ?) - ON CONFLICT(id) DO UPDATE SET - ip = excluded.ip, - port = excluded.port - """, (ctx.guild.id, val_ip, val_port)) - - setMCNames(ctx.guild.id, conn, names) - - if annChanID is None: - await setAnn(ctx, False) - else: - await setAnn(ctx, True, annChanID) - - if ctx.guild.id not in servers: - bot.loop.create_task(status_task(ctx.guild.id)) - servers.append(ctx.guild.id) - - await log(ctx, f"Successfully set up a new server querier for {server_port_ip}!") - - @commands.command(brief="Removes querier and status channels", - description="Removes querier and status channels. Deletes your guild's data from my server. Data deletion occurs automatically when I am removed from your guild.") - @commands.has_permissions(administrator=True) - async def cleanup(self, ctx: discord.ext.commands.context.Context): - await do_bot_cleanup(ctx.guild.id, ctx=ctx) - - @commands.command(brief="Turns on player join announcements", - description="Turns on player join announcements in specified channel.") - @commands.has_permissions(administrator=True) - async def announce(self, ctx: discord.ext.commands.context.Context, chanid: Union[int, str]): - await setAnn(ctx, True, chanid) - - @commands.command(brief="Turns off player join announcements", description="Turns off player join announcements.") - @commands.has_permissions(administrator=True) - async def noannounce(self, ctx: discord.ext.commands.context.Context): - await setAnn(ctx, False) - - @commands.command(brief="Turns on total player hour display", description="Turns on total player hour display.") - @commands.has_permissions(administrator=True) - async def showhours(self, ctx: discord.ext.commands.context.Context): - await setShowHours(ctx, True) - - @commands.command(brief="Turns off total player hour display", description="Turns off total player hour display.") - @commands.has_permissions(administrator=True) - async def nohours(self, ctx: discord.ext.commands.context.Context): - await setShowHours(ctx, False) - - -class Other(commands.Cog): - """Other""" - - def __init__(self, bot: discord.ext.commands.Bot): - bot.help_command.cog = self - - @commands.command(brief="Shows server status", description="Shows server status.") - async def status(self, ctx: discord.ext.commands.context.Context): - with sqlite_conn() as conn: - ip, port, query_port = getMCIP(ctx.guild.id, conn) - mc = JavaServer(ip, port, query_port) - - if mc is None: - await safe_send("There is no server query set up. Run the `setup` command to get started.", ctx=ctx) - else: - players, max, _, motd = await getStatus(mc) - - seconds = getPersonSeconds(ctx.guild.id, conn) - seconds = seconds % (24 * 3600) - hour = seconds // 3600 - seconds %= 3600 - minutes = seconds // 60 - seconds %= 60 - - await safe_send("Status:\n " + ip + "\n " + motd.to_plain() + "\n\n Players: " + str(players) + "/" + str(max) + "\n Total Player Time: " + "%d:%02d:%02d" % (hour, minutes, seconds), ctx=ctx, format="```") - - @commands.command(brief="Lists online players", description="Lists online players.") - async def players(self, ctx: discord.ext.commands.context.Context): - with sqlite_conn() as conn: - mc = JavaServer(*getMCIP(ctx.guild.id, conn)) - - if mc is None: - await safe_send("There is no server query set up. Run the `setup` command to get started.", ctx=ctx) - else: - _, _, names, _ = await getStatus(mc) - - pStr = "Online Players:\n" - for name in names: - pStr += " " + name + "\n" - - pStr = pStr[:-1] - await safe_send(pStr, ctx=ctx, format="```") - - @commands.command(brief="Shows last query time", description="Shows last query time. Useful for debugging server connection issues.") - async def lastquery(self, ctx: discord.ext.commands.context.Context): - with sqlite_conn() as conn: - ip, port, query_port = getMCIP(ctx.guild.id, conn) - mc = JavaServer(ip, port, query_port) - last = getMCQueryTime(ctx.guild.id, conn) - - if mc is None: - await safe_send("There is no server query set up. Run the `setup` command to get started.", ctx=ctx) - else: - await safe_send("I last queried " + ip + " at " + str(last) + " UTC", ctx=ctx) - - -async def setAnn(ctx: discord.ext.commands.context.Context, ann: bool, cid: Union[int, str, None] = None): - if cid is not None: - if type(cid) == str: - res = re.match(r"^<#(\d+)>$", cid) - if res is not None: - cid = int(res.group(1)) - - if type(cid) != int: - await log(ctx, f"Announcement channel ID was provided but was malformed.") - return - - if cid is not None and find_channels(serv=ctx.guild, chanid=cid) is None: - await log(ctx, "Channel", str(cid), "does not exist.") - return - - with sqlite_conn() as conn: - ip, _, _ = getMCIP(ctx.guild.id, conn) - - conn.execute(""" - UPDATE servers - SET - announce_joins = ?, - announce_joins_id = ? - WHERE id = ? - """, ( - ann, - cid if ann else None, - ctx.guild.id - )) - - await log(ctx, ( - "Not announcing when a player joins " + ip + ".", - "Announcing when a player joins " + ip + " in <#" + str(cid) + ">.")[ - ann]) - - -async def setShowHours(ctx: discord.ext.commands.context.Context, show_hours: bool): - with sqlite_conn() as conn: - ip, _, _ = getMCIP(ctx.guild.id, conn) - - conn.execute(""" - UPDATE servers - SET show_hours = ? - WHERE id = ? - """, ( - show_hours, - ctx.guild.id - )) - - await log(ctx, ( - "Not displaying total player hours for " + ip + ".", - "Displaying total player hours for " + ip + ".")[ - show_hours] + " Please wait a moment for channel updates to take effect.") - - -async def log(ctx: commands.Context, *msg: str): - """Log messages both in Discord and console.""" - print(ctx.guild.name, "Log:", " ".join(msg)) - await safe_send(" ".join(msg), ctx=ctx) - - -async def validate_ip_port(ctx, ip, port, query_port): - """Validates IP and Port""" - split = ip.split(":") - if len(split) > 1: - ip = split[0] - - try: - port = int(split[1]) - except ValueError: - await log(ctx, str(port), "is not a valid port number, please try again.") - return None, None, None - - if port not in range(65536): - await log(ctx, str(port), "is not a valid port number, please try again.") - return None, None, None - - if query_port not in range(65536): - await log(ctx, str(query_port), "is not a valid port number, please try again.") - return None, None, None - - domain = re.search(r"^([a-z0-9]+(-[a-z0-9]+)*\.)+[a-z]{2,6}$", ip) - addr = re.search( - r"^(([0-9]|[1-9][0-9]|1[0-9]{2}|2[0-4][0-9]|25[0-5])\.){3}([0-9]|[1-9][0-9]|1[0-9]{2}|2[0-4][0-9]|25[0-5])$", - ip) - if domain is None or domain.group(0) != ip: - if addr is None or addr.group(0) != ip: - await log(ctx, ip, "is not a valid domain or IP address, please try again.") - return None, None, None - - if re.search(r"(^127\.)|(^10\.)|(^172\.1[6-9]\.)|(^172\.2[0-9]\.)|(^172\.3[0-1]\.)|(^192\.168\.)", - ip) is not None: - await log(ctx, ip, "is a private IP. I won't be able to query it.") - return None, None, None - - return ip, port, query_port - -def get_server_by_id(sid: int): - for server in bot.guilds: - if server.id == sid: - return server - - -def find_channels(serv: Union[discord.Guild, None] = None, sid: Union[int, None] = None, - chanid: Union[int, None] = None, channame: Union[str, None] = None, channamesearch: str = "equal", - chantype: Union[discord.ChannelType, None] = None): - matches = [] - checkServer = None - - if sid is not None: - checkServer = get_server_by_id(sid) - - if serv is not None: - checkServer = serv - - for c in checkServer.channels: - if chanid is not None and c.id != chanid: - continue - if channame is not None and channamesearch.lower() == "equal" and c.name != channame: - continue - if channame is not None and channamesearch.lower() == "in" and channame not in c.name: - continue - if chantype is not None and c.type != chantype: - continue - - matches.append(c) - - return matches - - -async def getStatus(serv: JavaServer): - status = await asyncio.wait_for(serv.async_query(), timeout=1.0) - p = status.players.online - m = status.players.max - n = status.players.list - d = status.motd - - return p, m, n, d - - -@contextmanager -def sqlite_conn(): - conn = sqlite3.connect('/data/mstbot.db') - try: - # per-connection PRAGMAs - conn.execute("PRAGMA foreign_keys = ON") - conn.execute("PRAGMA busy_timeout = 5000") - yield conn - conn.commit() - except Exception: - conn.rollback() - raise - finally: - # run on close - conn.execute("PRAGMA optimize") - conn.close() - - -def getMCIP(sid: int, conn: sqlite3.Connection): - row = conn.execute( - "SELECT ip, port, query_port FROM servers WHERE id = ?", - (sid,) - ).fetchone() - - [ip, port, query_port] = row - return ip, port, query_port - - -def getMCNames(sid: int, conn: sqlite3.Connection): - rows = conn.execute( - "SELECT name FROM names WHERE id = ?", - (sid,) - ).fetchall() - - return [row[0] for row in rows] - - -def getMCQueryTime(sid: int, conn: sqlite3.Connection): - tstr = conn.execute( - "SELECT last_query FROM servers WHERE id = ?", - (sid,) - ).fetchone()[0] - - if tstr is None: - return tstr - - return datetime.strptime(tstr, '%Y-%m-%d %H:%M:%S') - -def getPersonSeconds(sid: int, conn: sqlite3.Connection): - cols = conn.execute( - f"SELECT SUM(seconds) FROM times WHERE id = ? GROUP BY id", - (sid,) - ).fetchone() - if cols is not None: - return cols[0] - - return 0 - -def getShowHours(sid: int, conn: sqlite3.Connection): - show_hours = conn.execute( - "SELECT show_hours FROM servers WHERE id = ?", - (sid,) - ).fetchone()[0] - - return show_hours - -def getMCJoinAnnounce(sid: int, conn: sqlite3.Connection): - row = conn.execute( - "SELECT announce_joins, announce_joins_id FROM servers WHERE id = ?", - (sid,) - ).fetchone() - ann = row[0] - cid = row[1] - - return ann, cid - - -def setMCNames(sid: int, conn: sqlite3.Connection, names: List[str]): - conn.execute("DELETE FROM names WHERE id = ?", (sid,)) - - if names: - rows = [(sid, name) for name in names] - conn.executemany("INSERT INTO names (id, name) VALUES (?, ?)", rows) - -def setMCQueryTime(sid: int, conn: sqlite3.Connection, dt: datetime): - conn.execute(""" - UPDATE servers - SET last_query = ? - WHERE id = ? - """, ( - dt.strftime('%Y-%m-%d %H:%M:%S'), - sid - )) - - -def incrementPersonSeconds(sid: int, user: str, conn: sqlite3.Connection, seconds: float): - conn.execute(""" - INSERT INTO times (id, name, seconds) - VALUES (?, ?, ?) - ON CONFLICT(id, name) DO UPDATE SET - seconds = times.seconds + excluded.seconds - """, (sid, user, seconds)) - -async def do_bot_cleanup(sid: int, ctx: Union[discord.ext.commands.context.Context, None] = None): - if sid in servers: - servers.remove(sid) - - deleted = False - currServ = get_server_by_id(sid) - if currServ is not None: - iChannels = find_channels(serv=currServ, channame="IP: ", channamesearch="in", chantype=discord.ChannelType.voice) - pChannels = find_channels(serv=currServ, channame="Players: ", channamesearch="in", chantype=discord.ChannelType.voice) - hChannels = find_channels(serv=currServ, channame="Player Hrs: ", channamesearch="in", chantype=discord.ChannelType.voice) - channels = iChannels + pChannels + hChannels - try: - for channel in channels: - await channel.delete() - - deleted = True - except discord.errors.Forbidden: - print(currServ.name, "Error: I don't have permission to delete channels.") - - with sqlite_conn() as conn: - ip, port, query_port = getMCIP(sid, conn) - mc = JavaServer(ip, port, query_port) - - conn.execute("DELETE FROM servers WHERE id = ?", (sid,)) - conn.execute("DELETE FROM names WHERE id = ?", (sid,)) - conn.execute("DELETE FROM times WHERE id = ?", (sid,)) - - if ctx is None: - print("Cleaned up", sid, ".") - else: - if mc is None: - await safe_send("There is no server query set up. Run the `setup` command to get started.", ctx=ctx) - else: - await log(ctx, "Cleaned up! Removed", ip, "querier, deleted", ctx.guild.name + "'s data from my server,", ("but failed to remove my status channels. ", "and removed my status channels.")[deleted]) - -def ascii_clean(s): - return "".join( - c for c in unicodedata.normalize("NFKD", s) - if unicodedata.category(c)[0] != "C" - ) - -async def get_last_channel_name(channel): - last_name = channel.name - latest = await limiter.get_state(channel.id) - if latest: - args, _ = latest - last_name = args[1] - - return last_name - - -async def safe_send(msg: str, ctx: Union[discord.ext.commands.context.Context, None] = None, - chan: Union[discord.TextChannel, None] = None, format: str = ''): - try: - if ctx is not None: - await ctx.send(f"{format}{msg}{format}") - elif chan is not None: - await chan.send(f"{format}{msg}{format}") - except discord.errors.Forbidden: - if ctx is not None: - print(ctx.guild.name, - "Error: I don't have permission to send messages here.") - elif chan is not None: - print(chan.guild.name, - "Error: I don't have permission to send messages here.") - - -async def status_task(sid: int): - first_iter = True - - while True: - # if bot is not a member, clean data, end task - if get_server_by_id(sid) is None: - print("Kicked: Bot must've been kicked from", sid, ". Cleaning up.") - await do_bot_cleanup(sid) - return - - # check to see if specified server is still querying - if sid not in servers: - print(get_server_by_id(sid).name, "is not querying. Ending task.") - return - - with sqlite_conn() as conn: - ip, port, query_port = getMCIP(sid, conn) - mc = JavaServer(ip, port, query_port) - currServ = get_server_by_id(sid) - wait = 10 - - if not ip == "" and currServ is not None: - try: - oldNames = getMCNames(sid, conn) - players, max, names, _ = await getStatus(mc) - lastTime = getMCQueryTime(sid, conn) - lastSeconds = getPersonSeconds(sid, conn) - - currTime = datetime.now(timezone.utc) - if lastTime is None: - lastTime = currTime - else: - lastTime = lastTime.replace(tzinfo=timezone.utc) - - deltaSeconds = (currTime - lastTime).total_seconds() - - - setMCNames(sid, conn, names) - setMCQueryTime(sid, conn, currTime) - if not first_iter: - for name in names: - incrementPersonSeconds(sid, name, conn, deltaSeconds) - - announceJoin, annChanId = getMCJoinAnnounce(sid, conn) - annRole = None - - if announceJoin and annChanId is not None: - jNames = list(set(names) - set(oldNames)) - for name in jNames: - aChannels = find_channels(serv=currServ, chanid=int(annChanId)) - - if len(aChannels): - if annRole is None: - await safe_send(name + " joined the game!", chan=aChannels[0]) - else: - await safe_send("<@&" + str(annRole.id) + "> " + name + " joined the game!", - chan=aChannels[0]) - - print(currServ.name, "Announced player(s) join.") - - except asyncio.exceptions.TimeoutError: - print(currServ.name, "Timeout, server lagging?") - players = -1 - lastSeconds = -1 - - except (ConnectionRefusedError, socket.gaierror): - print(currServ.name, "Cannot connect to server, down?") - players = -1 - lastSeconds = -1 - - iChannels = find_channels(serv=currServ, channame="IP: ", channamesearch="in", chantype=discord.ChannelType.voice) - pChannels = find_channels(serv=currServ, channame="Players: ", channamesearch="in", - chantype=discord.ChannelType.voice) - tChannels = find_channels(serv=currServ, channame="Player Hrs: ", channamesearch="in", - chantype=discord.ChannelType.voice) - - overwrites = { - currServ.default_role: discord.PermissionOverwrite(connect=False, view_channel=True), - currServ.me: discord.PermissionOverwrite(connect=True, view_channel=True, manage_channels=True) - } - - ipStr = f"IP: {ip}" - if len(iChannels): - lastIPName = await get_last_channel_name(iChannels[0]) - - if lastIPName != ipStr: - print(currServ.name, "Update: IP changed!") - await limiter.submit(iChannels[0].id, iChannels[0], ipStr) - else: - await currServ.create_voice_channel(ipStr, overwrites=overwrites) - - p_str_temp = f"Players: {str(players)}/{str(max)}" - if len(pChannels): - lastPName = await get_last_channel_name(pChannels[0]) - - if players == -1: - pStr = lastPName - else: - pStr = p_str_temp - - if lastPName != pStr: - print(currServ.name, "Update: Players changed!") - await limiter.submit(pChannels[0].id, pChannels[0], pStr) - else: - await currServ.create_voice_channel(p_str_temp, - overwrites=overwrites) - - do_show_hours = getShowHours(sid, conn) - if do_show_hours and not first_iter and lastSeconds != -1: - tStr = "Player Hrs: " + str(round(getPersonSeconds(sid, conn)/3600)) - if len(tChannels): - lastTName = await get_last_channel_name(tChannels[0]) - - if lastTName != tStr: - print(currServ.name, "Update: Time changed!") - await limiter.submit(tChannels[0].id, tChannels[0], tStr) - else: - await currServ.create_voice_channel(tStr, - overwrites=overwrites) - elif not do_show_hours: - if len(tChannels): - for channel in tChannels: - await channel.delete() - - timestamp = datetime.now(timezone.utc).strftime("%H:%M:%S") - ip_channel_str = f"IP: {str(await limiter.get_time_remaining(iChannels[0].id)) if len(iChannels) else 'N/A'}" - p_channel_str = f"P: {str(await limiter.get_time_remaining(pChannels[0].id)) if len(pChannels) else 'N/A'}" - t_channel_str = f"T: {str(await limiter.get_time_remaining(tChannels[0].id)) if len(tChannels) else 'N/A'}" - print(currServ.name, "Query:", ip, p_str_temp, timestamp, - str(deltaSeconds), ip_channel_str, p_channel_str, t_channel_str) - - await asyncio.sleep(wait) - first_iter = False - -@bot.event -async def setup_hook(): - await bot.add_cog(Admin()) - await bot.add_cog(Other(bot)) - -token = os.getenv('TOKEN') -if token is not None: - bot.run(token) -else: - print("Token not found, ensure .env file exists and is well-formed.") diff --git a/src/mstbot/__init__.py b/src/mstbot/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/src/mstbot/__main__.py b/src/mstbot/__main__.py new file mode 100644 index 0000000..42b49de --- /dev/null +++ b/src/mstbot/__main__.py @@ -0,0 +1,23 @@ +# TODO: in readme, add badge for guild count, total player count +# TODO: annotate return types everywhere +# TODO: wrap all SQL queries with a func to determine if server has been set up or not. If fetchone returns None, log helpful message to run $setup. +# TODO: Show query port enable instructions +# TODO: Feature: allow discord users to associate their mc usernames, change server nickname? +# TODO: Feature: allow other users to query playtime by username, default to their own. This and above require new columns in names table. +# TODO: If on startup the schema has changed, create new db and attempt to migrate old data into new schema + +import os + +from mstbot.bot import MinecraftStatusBot + +def main(): + bot = MinecraftStatusBot() + + token = os.getenv('TOKEN') + if token is not None: + bot.run(token) + else: + print("Token not found, ensure .env file exists and is well-formed.") + +if __name__ == "__main__": + main() diff --git a/src/mstbot/bot.py b/src/mstbot/bot.py new file mode 100644 index 0000000..f271faa --- /dev/null +++ b/src/mstbot/bot.py @@ -0,0 +1,34 @@ +import discord +import socket + +from mstbot.cogs.admin import Admin +from mstbot.cogs.other import Other +from mstbot.utils import safe_send, log + +class MinecraftStatusBot(discord.ext.commands.Bot): + def __init__(self): + intents = discord.Intents.default() + intents.messages = True + intents.message_content = True + + super().__init__(command_prefix="$", intents=intents) + + + async def setup_hook(self): + await self.add_cog(Admin(self)) + await self.add_cog(Other(self)) + + + async def on_command_error(self, ctx: discord.ext.commands.Context, error: discord.ext.commands.CommandError): + command = ctx.message.content.split()[0][1:] # Extract command name + + if isinstance(error, discord.ext.commands.CommandNotFound): + await safe_send(f'Command "{command}" does not exist. Use `{ctx.prefix}help` for available commands.', ctx) + elif isinstance(error, discord.ext.commands.MissingRequiredArgument): + await log(ctx, f'Command "{command}" is missing required arguments. Use `{ctx.prefix}help {command}` for details.') + elif isinstance(error, discord.ext.commands.MissingPermissions): + await safe_send(f'You lack permissions: {", ".join(error.missing_perms)}.', ctx) + elif isinstance(error, discord.ext.commands.CommandInvokeError) and isinstance(error.original, socket.gaierror): + await log(ctx, "Could not connect, is the server down?") + else: + raise error diff --git a/src/mstbot/cogs/admin.py b/src/mstbot/cogs/admin.py new file mode 100644 index 0000000..c51e942 --- /dev/null +++ b/src/mstbot/cogs/admin.py @@ -0,0 +1,375 @@ +import asyncio +import re +import socket +import unicodedata +from typing import Union, List +from datetime import datetime, timezone + +import discord +from mcstatus import JavaServer + +from mstbot.erl import EndpointRateLimiter +from mstbot.utils import log, validate_ip_port, find_channels, safe_send, getStatus, change_channel_name +from mstbot.sql import sqlite_conn, get_mc_server, setMCNames, getMCNames, getMCQueryTime, getPersonSeconds, setMCQueryTime, incrementPersonSeconds, getMCJoinAnnounce, getShowHours + +class Admin(discord.ext.commands.Cog): + """Admin""" + + def __init__(self, bot: discord.ext.commands.Bot): + self.bot = bot + self.servers = [] + self.limiter = EndpointRateLimiter(301, change_channel_name) + + with sqlite_conn() as conn: + rows = conn.execute( + "SELECT id, ip, announce_joins, announce_joins_id FROM servers" + ).fetchall() + + print(rows) + for row in rows: + server = self.bot.get_guild(row[0]) + if server is None: + print(f"Kicked: Bot was removed from {row[0]}. Cleaning up.") + bot.loop.create_task(cleanup_server_by_id(row[0])) + continue + + bot.loop.create_task(status_task(server, self.servers, self.limiter)) + self.servers.append(server) + announcement_status = "Not announcing" if not row[2] else f"Announcing in <#{row[3]}>" + print(f"Now querying {server.name}. {announcement_status} for {row[1]}") + + + @discord.ext.commands.command(brief="Sets up new Minecraft server querier", + description="Sets up new Minecraft server querier. Removes previous querier, if applicable. Use a valid domain name or IP address. The default port (25565) is used unless otherwise specified. Specify a channel ID to announce player joins there.") + @discord.ext.commands.has_permissions(administrator=True) + async def setup(self, ctx: discord.ext.commands.context.Context, ip: str, port: int = 25565, query_port: int = 25565, annChanID: Union[int, str, None] = None): + """Sets up a new Minecraft server querier.""" + val_ip, val_port, val_query_port = await validate_ip_port(ctx, ip, port, query_port) + if not val_ip: + return + + server_port_ip = f"{val_ip}:{val_port}" + await log(ctx, f"Attempting to establish connection to `{server_port_ip}` with query port `{val_query_port}`...") + mc = JavaServer(val_ip, val_port, query_port=val_query_port) + + status = None + query = None + try: + status = mc.status() + query = mc.query() + names = query.players.list + except asyncio.exceptions.TimeoutError: + await log(ctx, ( + "The setup Query protocal ping failed, but the Minecraft server appears to be accessible.\n" + "Does the server.properties file contain `enable-query=true`? Changes to server.properties do not take effect until the Minecraft server is restarted.\n" + f"Is UDP port `{val_query_port}` forwarded and open on the firewall?" + )) + if status is None: + await log(ctx, ( + "A Status protocol ping also failed.\n" + "Does the server.properties file contain `enable-status=true`?\n" + f"Is TCP port `{val_port}` forwarded and open on the firewall?" + )) + except ConnectionRefusedError: + await log(ctx, ( + f"Setup query failed. I could not reach a Minecraft server at `{server_port_ip}`.\n" + f"Is TCP port `{val_port}` forwarded and open on the firewall?\n" + f"Can you and a friend (on a different network) connect to and play on `{server_port_ip}`?" + )) + else: + with sqlite_conn() as conn: + conn.execute(""" + INSERT INTO servers (id, ip, port) + VALUES (?, ?, ?) + ON CONFLICT(id) DO UPDATE SET + ip = excluded.ip, + port = excluded.port + """, (ctx.guild.id, val_ip, val_port)) + + setMCNames(ctx.guild.id, conn, names) + + if annChanID is None: + await setAnn(ctx, False) + else: + await setAnn(ctx, True, annChanID) + + if ctx.guild not in self.servers: + self.bot.loop.create_task(status_task(ctx.guild, self.servers, self.limiter)) + self.servers.append(ctx.guild) + + await log(ctx, f"Successfully set up a new server querier for `{server_port_ip}`!") + + + @discord.ext.commands.command(brief="Removes querier and status channels", + description="Removes querier and status channels. Deletes your guild's data from my server. Data deletion occurs automatically when I am removed from your guild.") + @discord.ext.commands.has_permissions(administrator=True) + async def cleanup(self, ctx: discord.ext.commands.context.Context): + await do_bot_cleanup(ctx.guild, ctx) + + + @discord.ext.commands.command(brief="Turns on player join announcements", + description="Turns on player join announcements in specified channel.") + @discord.ext.commands.has_permissions(administrator=True) + async def announce(self, ctx: discord.ext.commands.context.Context, chanid: Union[int, str]): + await setAnn(ctx, True, chanid) + + + @discord.ext.commands.command(brief="Turns off player join announcements", description="Turns off player join announcements.") + @discord.ext.commands.has_permissions(administrator=True) + async def noannounce(self, ctx: discord.ext.commands.context.Context): + await setAnn(ctx, False) + + + @discord.ext.commands.command(brief="Turns on total player hour display", description="Turns on total player hour display.") + @discord.ext.commands.has_permissions(administrator=True) + async def showhours(self, ctx: discord.ext.commands.context.Context): + await setShowHours(ctx, True) + + + @discord.ext.commands.command(brief="Turns off total player hour display", description="Turns off total player hour display.") + @discord.ext.commands.has_permissions(administrator=True) + async def nohours(self, ctx: discord.ext.commands.context.Context): + await setShowHours(ctx, False) + + +async def do_bot_cleanup(server: discord.Guild, ctx: Union[discord.ext.commands.context.Context, None]): + await cleanup_channels(server, ctx) + await cleanup_server_by_id(server.id, ctx) + + +async def cleanup_server_by_id(sid: int, ctx: Union[discord.ext.commands.context.Context, None] = None): + mc = None + with sqlite_conn() as conn: + mc = get_mc_server(sid, conn) + if mc is None: + await log(ctx, ctx.guild.name + " has no server queriers set up! Everything has already been cleaned. Run the `setup` command to get started!") + + conn.execute("DELETE FROM servers WHERE id = ?", (sid,)) + conn.execute("DELETE FROM names WHERE id = ?", (sid,)) + conn.execute("DELETE FROM times WHERE id = ?", (sid,)) + + if ctx is None: + print("Cleaned up", sid, ".") + elif mc is not None: + await log(ctx, "Cleaned up! Removed", mc.address.host, "querier, deleted", ctx.guild.name + "'s data from my backend server.") + + +async def cleanup_channels(server: discord.Guild, ctx: Union[discord.ext.commands.context.Context, None]): + iChannels = find_channels(server, channame="IP: ", channamesearch="in", chantype=discord.ChannelType.voice) + pChannels = find_channels(server, channame="Players: ", channamesearch="in", chantype=discord.ChannelType.voice) + hChannels = find_channels(server, channame="Player Hrs: ", channamesearch="in", chantype=discord.ChannelType.voice) + channels = iChannels + pChannels + hChannels + try: + for channel in channels: + await channel.delete() + + await log(ctx, "I successfully deleted my status channels.") + except discord.errors.Forbidden: + log_str = f"I don't have permission to delete channels in {server.name}!" + await log(ctx, log_str) + + +async def setAnn(ctx: discord.ext.commands.context.Context, ann: bool, cid: Union[int, str, None] = None): + if cid is not None: + if type(cid) == str: + res = re.match(r"^<#(\d+)>$", cid) + if res is not None: + cid = int(res.group(1)) + + if type(cid) != int: + await log(ctx, f"Announcement channel ID was provided but was malformed.") + return + + if cid is not None and find_channels(ctx.guild, chanid=cid) is None: + await log(ctx, "Channel", str(cid), "does not exist.") + return + + with sqlite_conn() as conn: + mc = get_mc_server(ctx.guild.id, conn) + if mc is None: + await log(ctx, f"{ctx.guild.name} has no server queriers set up! Use the `setup` command to get started!") + return + + conn.execute(""" + UPDATE servers + SET + announce_joins = ?, + announce_joins_id = ? + WHERE id = ? + """, ( + ann, + cid if ann else None, + ctx.guild.id + )) + + await log(ctx, ( + "Not announcing when a player joins " + mc.address.host + ".", + "Announcing when a player joins " + mc.address.host + " in <#" + str(cid) + ">.")[ + ann]) + + +async def setShowHours(ctx: discord.ext.commands.context.Context, show_hours: bool): + with sqlite_conn() as conn: + mc = get_mc_server(ctx.guild.id, conn) + if mc is None: + await log(ctx, f"{ctx.guild.name} has no server queriers set up! Use the `setup` command to get started!") + return + + conn.execute(""" + UPDATE servers + SET show_hours = ? + WHERE id = ? + """, ( + show_hours, + ctx.guild.id + )) + + await log(ctx, ( + "Not displaying total player hours for " + mc.address.host + ".", + "Displaying total player hours for " + mc.address.host + ".")[ + show_hours] + " Please wait a moment for channel updates to take effect.") + + +def ascii_clean(s): + return "".join( + c for c in unicodedata.normalize("NFKD", s) + if unicodedata.category(c)[0] != "C" + ) + + +async def get_last_channel_name(channel, limiter): + last_name = channel.name + latest = await limiter.get_state(channel.id) + if latest: + args, _ = latest + last_name = args[1] + + return last_name + + + +async def status_task(server: discord.Guild, servers: List[discord.Guild], limiter: EndpointRateLimiter): + first_iter = True + + while True: + # if bot is not a member, clean data, end task + if server not in servers: + print(server.name, "is not querying. Ending task.") + return + + with sqlite_conn() as conn: + mc = get_mc_server(server.id, conn) + wait = 10 + + if mc is not None and server is not None: + try: + oldNames = getMCNames(server.id, conn) + players, max, names, _ = await getStatus(mc) + lastTime = getMCQueryTime(server.id, conn) + lastSeconds = getPersonSeconds(server.id, conn) + + currTime = datetime.now(timezone.utc) + if lastTime is None: + lastTime = currTime + else: + lastTime = lastTime.replace(tzinfo=timezone.utc) + + deltaSeconds = (currTime - lastTime).total_seconds() + + + setMCNames(server.id, conn, names) + setMCQueryTime(server.id, conn, currTime) + if not first_iter: + for name in names: + incrementPersonSeconds(server.id, name, conn, deltaSeconds) + + announceJoin, annChanId = getMCJoinAnnounce(server.id, conn) + annRole = None + + if announceJoin and annChanId is not None: + jNames = list(set(names) - set(oldNames)) + for name in jNames: + aChannels = find_channels(server, chanid=int(annChanId)) + + if len(aChannels): + if annRole is None: + await safe_send(name + " joined the game!", chan=aChannels[0]) + else: + await safe_send("<@&" + str(annRole.id) + "> " + name + " joined the game!", + chan=aChannels[0]) + + print(server.name, "Announced player(s) join.") + + except asyncio.exceptions.TimeoutError: + print(server.name, "Timeout, server lagging?") + players = -1 + lastSeconds = -1 + + except (ConnectionRefusedError, socket.gaierror): + print(server.name, "Cannot connect to server, down?") + players = -1 + lastSeconds = -1 + + iChannels = find_channels(server, channame="IP: ", channamesearch="in", chantype=discord.ChannelType.voice) + pChannels = find_channels(server, channame="Players: ", channamesearch="in", + chantype=discord.ChannelType.voice) + tChannels = find_channels(server, channame="Player Hrs: ", channamesearch="in", + chantype=discord.ChannelType.voice) + + overwrites = { + server.default_role: discord.PermissionOverwrite(connect=False, view_channel=True), + server.me: discord.PermissionOverwrite(connect=True, view_channel=True, manage_channels=True) + } + + ipStr = f"IP: {mc.address.host}" + if len(iChannels): + lastIPName = await get_last_channel_name(iChannels[0], limiter) + + if lastIPName != ipStr: + print(server.name, "Update: IP changed!") + await limiter.submit(iChannels[0].id, iChannels[0], ipStr) + else: + await server.create_voice_channel(ipStr, overwrites=overwrites) + + p_str_temp = f"Players: {str(players)}/{str(max)}" + if len(pChannels): + lastPName = await get_last_channel_name(pChannels[0], limiter) + + if players == -1: + pStr = lastPName + else: + pStr = p_str_temp + + if lastPName != pStr: + print(server.name, "Update: Players changed!") + await limiter.submit(pChannels[0].id, pChannels[0], pStr) + else: + await server.create_voice_channel(p_str_temp, + overwrites=overwrites) + + do_show_hours = getShowHours(server.id, conn) + if do_show_hours and not first_iter and lastSeconds != -1: + tStr = "Player Hrs: " + str(round(getPersonSeconds(server.id, conn)/3600)) + if len(tChannels): + lastTName = await get_last_channel_name(tChannels[0], limiter) + + if lastTName != tStr: + print(server.name, "Update: Time changed!") + await limiter.submit(tChannels[0].id, tChannels[0], tStr) + else: + await server.create_voice_channel(tStr, + overwrites=overwrites) + elif not do_show_hours: + if len(tChannels): + for channel in tChannels: + await channel.delete() + + timestamp = datetime.now(timezone.utc).strftime("%H:%M:%S") + ip_channel_str = f"IP: {str(await limiter.get_time_remaining(iChannels[0].id)) if len(iChannels) else 'N/A'}" + p_channel_str = f"P: {str(await limiter.get_time_remaining(pChannels[0].id)) if len(pChannels) else 'N/A'}" + t_channel_str = f"T: {str(await limiter.get_time_remaining(tChannels[0].id)) if len(tChannels) else 'N/A'}" + print(server.name, "Query:", mc.address.host, p_str_temp, timestamp, + str(deltaSeconds), ip_channel_str, p_channel_str, t_channel_str) + + await asyncio.sleep(wait) + first_iter = False \ No newline at end of file diff --git a/src/mstbot/cogs/other.py b/src/mstbot/cogs/other.py new file mode 100644 index 0000000..b0f4e6e --- /dev/null +++ b/src/mstbot/cogs/other.py @@ -0,0 +1,58 @@ +from discord.ext import commands +from mcstatus.motd import Motd + +from mstbot.utils import safe_send, getStatus, to_discord_ansi +from mstbot.sql import sqlite_conn, get_mc_server, getPersonSeconds, getMCQueryTime + +class Other(commands.Cog): + """Other""" + + def __init__(self, bot: commands.Bot): + bot.help_command.cog = self + + @commands.command(brief="Shows server status", description="Shows server status.") + async def status(self, ctx: commands.context.Context): + with sqlite_conn() as conn: + mc = get_mc_server(ctx.guild.id, conn) + if mc is None: + await safe_send("There is no server query set up. Run the `setup` command to get started.", ctx=ctx) + else: + players, max, _, motd = await getStatus(mc) + motd = Motd.parse(motd.raw.replace("§", "§")) + + seconds = getPersonSeconds(ctx.guild.id, conn) + seconds = seconds % (24 * 3600) + hour = seconds // 3600 + seconds %= 3600 + minutes = seconds // 60 + seconds %= 60 + + await safe_send("ansi\nStatus:\n " + mc.address.host + "\n " + to_discord_ansi(motd.to_ansi()) + "\n\n Players: " + str(players) + "/" + str(max) + "\n Total Player Time: " + "%d:%02d:%02d\n" % (hour, minutes, seconds), ctx=ctx, format="```") + + @commands.command(brief="Lists online players", description="Lists online players.") + async def players(self, ctx: commands.context.Context): + with sqlite_conn() as conn: + mc = get_mc_server(ctx.guild.id, conn) + + if mc is None: + await safe_send("There is no server query set up. Run the `setup` command to get started.", ctx=ctx) + else: + _, _, names, _ = await getStatus(mc) + + pStr = "Online Players:\n" + for name in names: + pStr += " " + name + "\n" + + pStr = pStr[:-1] + await safe_send(pStr, ctx=ctx, format="```") + + @commands.command(brief="Shows last query time", description="Shows last query time. Useful for debugging server connection issues.") + async def lastquery(self, ctx: commands.context.Context): + with sqlite_conn() as conn: + mc = get_mc_server(ctx.guild.id, conn) + last = getMCQueryTime(ctx.guild.id, conn) + + if mc is None: + await safe_send("There is no server query set up. Run the `setup` command to get started.", ctx=ctx) + else: + await safe_send("I last queried " + mc.address.host + " at " + str(last) + " UTC", ctx=ctx) diff --git a/src/erl.py b/src/mstbot/erl.py similarity index 100% rename from src/erl.py rename to src/mstbot/erl.py diff --git a/src/mstbot/sql.py b/src/mstbot/sql.py new file mode 100644 index 0000000..3ed09c6 --- /dev/null +++ b/src/mstbot/sql.py @@ -0,0 +1,115 @@ +import sqlite3 +from contextlib import contextmanager +from datetime import datetime +from typing import List + +from mcstatus import JavaServer + +@contextmanager +def sqlite_conn(): + conn = sqlite3.connect('/data/mstbot.db') + try: + # per-connection PRAGMAs + conn.execute("PRAGMA foreign_keys = ON") + conn.execute("PRAGMA busy_timeout = 5000") + yield conn + conn.commit() + except Exception: + conn.rollback() + raise + finally: + # run on close + conn.execute("PRAGMA optimize") + conn.close() + + +def get_mc_server(sid: int, conn: sqlite3.Connection): + row = conn.execute( + "SELECT ip, port, query_port FROM servers WHERE id = ?", + (sid,) + ).fetchone() + + if row is None: + return None + + return JavaServer(*row) + + +def getMCNames(sid: int, conn: sqlite3.Connection): + rows = conn.execute( + "SELECT name FROM names WHERE id = ?", + (sid,) + ).fetchall() + + return [row[0] for row in rows] + + +def getMCQueryTime(sid: int, conn: sqlite3.Connection): + tstr = conn.execute( + "SELECT last_query FROM servers WHERE id = ?", + (sid,) + ).fetchone()[0] + + if tstr is None: + return tstr + + return datetime.strptime(tstr, '%Y-%m-%d %H:%M:%S') + + +def getPersonSeconds(sid: int, conn: sqlite3.Connection): + cols = conn.execute( + f"SELECT SUM(seconds) FROM times WHERE id = ? GROUP BY id", + (sid,) + ).fetchone() + if cols is not None: + return cols[0] + + return 0 + + +def getShowHours(sid: int, conn: sqlite3.Connection): + show_hours = conn.execute( + "SELECT show_hours FROM servers WHERE id = ?", + (sid,) + ).fetchone()[0] + + return show_hours + + +def getMCJoinAnnounce(sid: int, conn: sqlite3.Connection): + row = conn.execute( + "SELECT announce_joins, announce_joins_id FROM servers WHERE id = ?", + (sid,) + ).fetchone() + ann = row[0] + cid = row[1] + + return ann, cid + + +def setMCNames(sid: int, conn: sqlite3.Connection, names: List[str]): + conn.execute("DELETE FROM names WHERE id = ?", (sid,)) + + if names: + rows = [(sid, name) for name in names] + conn.executemany("INSERT INTO names (id, name) VALUES (?, ?)", rows) + + +def setMCQueryTime(sid: int, conn: sqlite3.Connection, dt: datetime): + conn.execute(""" + UPDATE servers + SET last_query = ? + WHERE id = ? + """, ( + dt.strftime('%Y-%m-%d %H:%M:%S'), + sid + )) + + +def incrementPersonSeconds(sid: int, user: str, conn: sqlite3.Connection, seconds: float): + conn.execute(""" + INSERT INTO times (id, name, seconds) + VALUES (?, ?, ?) + ON CONFLICT(id, name) DO UPDATE SET + seconds = times.seconds + excluded.seconds + """, (sid, user, seconds)) \ No newline at end of file diff --git a/src/mstbot/utils.py b/src/mstbot/utils.py new file mode 100644 index 0000000..5c84e31 --- /dev/null +++ b/src/mstbot/utils.py @@ -0,0 +1,130 @@ +import asyncio +import re +from typing import Union, List + +import discord +from discord.ext import commands +from mcstatus import JavaServer + + +async def change_channel_name(channel, name: str): + await channel.edit(name=name) + + +async def safe_send(msg: str, ctx: Union[commands.context.Context, None] = None, + chan: Union[discord.TextChannel, None] = None, format: str = ''): + try: + if ctx is not None: + await ctx.send(f"{format}{msg}{format}") + elif chan is not None: + await chan.send(f"{format}{msg}{format}") + except discord.errors.Forbidden: + if ctx is not None: + print(ctx.guild.name, + "Error: I don't have permission to send messages here.") + elif chan is not None: + print(chan.guild.name, + "Error: I don't have permission to send messages here.") + + +async def log(ctx: commands.Context, *msg: str): + """Log messages both in Discord and console.""" + print(ctx.guild.name, "Log:", " ".join(msg)) + await safe_send(" ".join(msg), ctx=ctx) + + +async def validate_ip_port(ctx, ip, port, query_port): + """Validates IP and Port""" + split = ip.split(":") + if len(split) > 1: + ip = split[0] + + try: + port = int(split[1]) + except ValueError: + await log(ctx, str(port), "is not a valid port number, please try again.") + return None, None, None + + if port not in range(65536): + await log(ctx, str(port), "is not a valid port number, please try again.") + return None, None, None + + if query_port not in range(65536): + await log(ctx, str(query_port), "is not a valid port number, please try again.") + return None, None, None + + domain = re.search(r"^([a-z0-9]+(-[a-z0-9]+)*\.)+[a-z]{2,6}$", ip) + addr = re.search( + r"^(([0-9]|[1-9][0-9]|1[0-9]{2}|2[0-4][0-9]|25[0-5])\.){3}([0-9]|[1-9][0-9]|1[0-9]{2}|2[0-4][0-9]|25[0-5])$", + ip) + if domain is None or domain.group(0) != ip: + if addr is None or addr.group(0) != ip: + await log(ctx, ip, "is not a valid domain or IP address, please try again.") + return None, None, None + + if re.search(r"(^127\.)|(^10\.)|(^172\.1[6-9]\.)|(^172\.2[0-9]\.)|(^172\.3[0-1]\.)|(^192\.168\.)", + ip) is not None: + await log(ctx, ip, "is a private IP. I won't be able to query it.") + return None, None, None + + return ip, port, query_port + + +def find_channels(server: Union[discord.Guild, None], + chanid: Union[int, None] = None, channame: Union[str, None] = None, channamesearch: str = "equal", + chantype: Union[discord.ChannelType, None] = None): + matches = [] + + for c in server.channels: + if chanid is not None and c.id != chanid: + continue + if channame is not None and channamesearch.lower() == "equal" and c.name != channame: + continue + if channame is not None and channamesearch.lower() == "in" and channame not in c.name: + continue + if chantype is not None and c.type != chantype: + continue + + matches.append(c) + + return matches + + +async def getStatus(serv: JavaServer): + status = await asyncio.wait_for(serv.async_query(), timeout=1.0) + p = status.players.online + m = status.players.max + n = status.players.list + d = status.motd + + return p, m, n, d + +# AI disclosure! +# The below section was written by AI. +DISCORD_RGB = { + (170, 0, 0): "\x1b[0;31m", # dark red -> red + (255, 170, 0): "\x1b[0;33m", # gold/orange -> yellow + (255, 255, 85): "\x1b[0;33m", # bright yellow -> yellow + (0, 170, 0): "\x1b[0;32m", # dark green -> green + (0, 170, 170): "\x1b[0;36m", # aqua -> cyan + (0, 0, 170): "\x1b[0;34m", # dark blue -> blue + (255, 85, 255): "\x1b[0;35m", # bright pink -> pink + (170, 0, 170): "\x1b[0;35m", # purple -> pink + (255, 255, 255): "\x1b[0;37m", # white -> white +} + + +RGB_RE = re.compile(r"\x1b\[38;2;(\d+);(\d+);(\d+)m") + +def to_discord_ansi(ansi: str) -> str: + def replace(match): + rgb = tuple(map(int, match.groups())) + return DISCORD_RGB.get(rgb, "\x1b[0;37m") + + ansi = RGB_RE.sub(replace, ansi) + + # Remove unsupported effects such as blinking. + ansi = ansi.replace("\x1b[5m", "") + ansi = ansi.replace("\x1b[0m", "\x1b[0m") + + return ansi \ No newline at end of file