bMorgan/MinecraftStatus
1
0
Fork 0
Code Issues Pull requests Releases 1 Wiki Activity Actions

Major refactors
All checks were successful
Build & Deploy MSTBot / Build & push runner image (push) Successful in 34s
Details
Build & Deploy MSTBot / Build, push & deploy (push) Successful in 1m16s
Details
Build & Deploy MSTBot / Create release (push) Has been skipped
Details

Browse source
This commit is contained in:
Benjamin Morgan 2026-10-04 21:41:29 -06:00
parent 00e2bd5ee3
commit 9cb489b4d0
13 changed files with 746 additions and 679 deletions
Download patch file Download diff file Expand all files Collapse all files

6
.dockerignore Normal file
View file

@ -0,0 +1,6 @@
__pycache__/
.venv/
venv/
.git/
.gitignore
.env

0
resources/image.png → .forgejo/resources/image.png
View file

Side by side Swipe Overlay

Before

Width:  |  Height:  |  Size: 9.7 KiB

After

Width:  |  Height:  |  Size: 9.7 KiB

Before After
Before After

6
Dockerfile
View file

@ -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"]

2
README.md
View file

@ -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
```

676
src/mstbot.py
View file

@ -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.")

0
src/mstbot/__init__.py Normal file
View file

23
src/mstbot/__main__.py Normal file
View file

@ -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()

34
src/mstbot/bot.py Normal file
View file

@ -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

375
src/mstbot/cogs/admin.py Normal file
View file

@ -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

58
src/mstbot/cogs/other.py Normal file
View file

@ -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)

0
src/erl.py → src/mstbot/erl.py
View file

115
src/mstbot/sql.py Normal file
View file

@ -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))

130
src/mstbot/utils.py Normal file
View file

@ -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