diff --git a/.dockerignore b/.dockerignore deleted file mode 100644 index fcd6639..0000000 --- a/.dockerignore +++ /dev/null @@ -1,6 +0,0 @@ -__pycache__/ -.venv/ -venv/ -.git/ -.gitignore -.env \ No newline at end of file diff --git a/Dockerfile b/Dockerfile index a077360..70b1e34 100644 --- a/Dockerfile +++ b/Dockerfile @@ -8,9 +8,7 @@ COPY schema.sql . ADD src src RUN apt update && apt upgrade -y && apt install -y sqlite3 -RUN pip install --upgrade pip && pip install --no-cache-dir -r requirements.txt - -ENV PYTHONPATH=/app/src +RUN pip install --upgrade pip && pip install -r requirements.txt ENTRYPOINT [ "/app/entrypoint.sh" ] -CMD ["python", "-m", "mstbot"] +CMD [ "python3", "-u", "src/mstbot.py" ] diff --git a/README.md b/README.md index c12b213..f474182 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](.forgejo/resources/image.png) +![Server Status Banners](resources/image.png) ### Commands ``` diff --git a/.forgejo/resources/image.png b/resources/image.png similarity index 100% rename from .forgejo/resources/image.png rename to resources/image.png diff --git a/src/mstbot/erl.py b/src/erl.py similarity index 100% rename from src/mstbot/erl.py rename to src/erl.py diff --git a/src/mstbot.py b/src/mstbot.py new file mode 100644 index 0000000..09814e3 --- /dev/null +++ b/src/mstbot.py @@ -0,0 +1,676 @@ +# 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 deleted file mode 100644 index e69de29..0000000 diff --git a/src/mstbot/__main__.py b/src/mstbot/__main__.py deleted file mode 100644 index 42b49de..0000000 --- a/src/mstbot/__main__.py +++ /dev/null @@ -1,23 +0,0 @@ -# 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 deleted file mode 100644 index f271faa..0000000 --- a/src/mstbot/bot.py +++ /dev/null @@ -1,34 +0,0 @@ -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 deleted file mode 100644 index c51e942..0000000 --- a/src/mstbot/cogs/admin.py +++ /dev/null @@ -1,375 +0,0 @@ -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 deleted file mode 100644 index b0f4e6e..0000000 --- a/src/mstbot/cogs/other.py +++ /dev/null @@ -1,58 +0,0 @@ -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/mstbot/sql.py b/src/mstbot/sql.py deleted file mode 100644 index 3ed09c6..0000000 --- a/src/mstbot/sql.py +++ /dev/null @@ -1,115 +0,0 @@ -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 deleted file mode 100644 index 5c84e31..0000000 --- a/src/mstbot/utils.py +++ /dev/null @@ -1,130 +0,0 @@ -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