Skip to content

Instantly share code, notes, and snippets.

@uint0
Created February 25, 2021 00:02
Show Gist options
  • Save uint0/c5c3ca24e1a4213a6a60ecc3c429828e to your computer and use it in GitHub Desktop.
Save uint0/c5c3ca24e1a4213a6a60ecc3c429828e to your computer and use it in GitHub Desktop.
Notify a discord channel when someone logs into a minecraft server. Run it on the same machine as it needs to be able to read logs files
import os
import discord
import asyncio
import event_source
client = discord.Client()
async def activity_messenger_task():
await client.wait_until_ready()
print('Client is ready, activity task will begin running...')
channel = client.get_channel(int(os.environ.get('DISBOARD_CHANNEL_ID', None)))
for evt in event_source.get_activity_events():
print(evt)
if evt['event'] == event_source.EVENT_LOGIN:
msg = "{player} joined the game".format(player=evt['data']['player'])
elif evt['event'] == event_source.EVENT_LOGOUT:
msg = "{player} left the game".format(player=evt['data']['player'])
if evt['data']['reason'] != "Disconnected":
msg += "\n@administrator Reason was not Disconnected! [reason={reason}]".format(reason=evt['data']['reason'])
else:
continue
await channel.send(msg)
@client.event
async def on_ready():
print('We have logged in as {0.user}'.format(client))
@client.event
async def on_message(message):
if message.author == client.user:
return
if __name__ == '__main__':
client.loop.create_task(activity_messenger_task())
client.run(os.environ.get('DISCORD_BOT_TOKEN', None))
import os
import time
import datetime as dt
"""
Constants
"""
LOG_FILE = "/path/to/mc/server/logs/latest.log"
POLL_SLEEP = 5
EVENT_LOGIN = 0
EVENT_LOGOUT = 1
"""
Globals
"""
last_poll = dt.datetime.now()
"""
Utils
"""
def follow(name):
current = open(name, "r")
curino = os.fstat(current.fileno()).st_ino
while True:
while True:
line = current.readline()
if not line:
break
yield line
try:
if os.stat(name).st_ino != curino:
new = open(name, "r")
current.close()
current = new
curino = os.fstat(current.fileno()).st_ino
continue
except IOError:
pass
time.sleep(POLL_SLEEP)
def parse_line(line):
line_part = line.split(']', 1)
if len(line_part) == 1:
return None, line
line_date, line_info = line_part
line_date = line_date.lstrip(' \t[')
try:
line_date = dt.datetime.strptime(line_date, '%d%b%Y %H:%M:%S.%f')
except ValueError:
return None, line
return line_date, line_info
def get_activity_events():
global last_poll
log_file = follow(LOG_FILE)
for line in log_file:
tm, info = parse_line(line)
if tm is None or tm <= last_poll:
continue
last_poll = tm
if "net.minecraft.server.management.PlayerList" in info and "logged in" in info:
yield {
'time': tm,
'raw': info,
'event': EVENT_LOGIN,
'data': {
'player': info.split(':')[1].split('[')[0].strip()
}
}
elif "net.minecraft.network.play.ServerPlayNetHandler" in info and "lost connection" in info:
yield {
'time': tm,
'raw': info,
'event': EVENT_LOGOUT,
'data': {
'player': info.split(':')[1].strip().split(' ')[0].strip(),
'reason': info.split(':')[-1].strip()
}
}
if __name__ == '__main__':
for event in get_activity_events():
print(event)
aiohttp==3.7.3
async-timeout==3.0.1
attrs==20.3.0
chardet==3.0.4
discord.py==1.6.0
idna==3.1
idna-ssl==1.1.0
multidict==5.1.0
pkg-resources==0.0.0
typing-extensions==3.7.4.3
yarl==1.6.3
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment