88 lines
2.8 KiB
Python
88 lines
2.8 KiB
Python
import asyncio
|
|
import logging
|
|
import os
|
|
from parser import parse_logs
|
|
from templates import decorate_message
|
|
from slixmpp import ClientXMPP
|
|
from dotenv import load_dotenv
|
|
from typing import Set
|
|
|
|
|
|
class MumbleNotifierBot(ClientXMPP):
|
|
def __init__(self, jid: str, password: str, room: str, nick: str = 'Mumble'):
|
|
ClientXMPP.__init__(self, jid, password)
|
|
|
|
self.room = room
|
|
self.nick = nick
|
|
|
|
self.add_event_handler('session_start', self.session_start)
|
|
|
|
# If you wanted more functionality, here's how to register plugins:
|
|
# self.register_plugin('xep_0030') # Service Discovery
|
|
# self.register_plugin('xep_0199') # XMPP Ping
|
|
self.register_plugin('xep_0045')
|
|
|
|
# Here's how to access plugins once you've registered them:
|
|
# self['xep_0030'].add_feature('echo_demo')
|
|
|
|
async def session_start(self, event):
|
|
logging.info('Session started')
|
|
await self.get_roster()
|
|
self.send_presence()
|
|
await self.plugin['xep_0045'].join_muc_wait(self.room, self.nick)
|
|
logging.info('Joined MUC {} as {}'.format(self.room, self.nick))
|
|
|
|
# Most get_*/set_* methods from plugins use Iq stanzas, which
|
|
# are sent asynchronously. You can almost always provide a
|
|
# callback that will be executed when the reply is received.
|
|
|
|
def notify_muc(self, new_mumble_users: Set[str]):
|
|
logging.info('Notifying MUC of newly joined Mumble users: {}'.format(new_mumble_users))
|
|
self.send_message(mto=self.room,
|
|
mbody=decorate_message(new_mumble_users),
|
|
mtype='groupchat')
|
|
|
|
|
|
async def check_murmur_new_users():
|
|
last_users = set()
|
|
first_check = True
|
|
|
|
while True:
|
|
logging.debug('Checking Murmur logs for connected users')
|
|
with open(os.getenv('MUMBLE_BOT_MURMUR_LOG', 'r')) as logfile:
|
|
logs = logfile.read()
|
|
current_users = parse_logs(logs)
|
|
|
|
if first_check:
|
|
last_users = current_users
|
|
first_check = False
|
|
else:
|
|
newly_joined = current_users - last_users
|
|
|
|
if newly_joined:
|
|
unique_newly_joined = set(newly_joined)
|
|
xmpp.notify_muc(unique_newly_joined)
|
|
|
|
last_users = current_users
|
|
|
|
await asyncio.sleep(10)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
# Ideally use optparse or argparse to get JID,
|
|
# password, and log level.
|
|
load_dotenv()
|
|
|
|
logging.basicConfig(level=logging.DEBUG,
|
|
format='%(asctime)s %(levelname)-8s %(message)s')
|
|
|
|
xmpp = MumbleNotifierBot(
|
|
os.getenv('MUMBLE_BOT_JID'),
|
|
os.getenv('MUMBLE_BOT_PASSWORD'),
|
|
os.getenv('MUMBLE_BOT_MUC')
|
|
)
|
|
xmpp.connect()
|
|
|
|
xmpp.loop.create_task(check_murmur_new_users())
|
|
xmpp.loop.run_forever()
|