hackbot/idlebot.py

148 lines
5.1 KiB
Python
Raw Normal View History

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# file: idlebot.py
# date: 23.07.2020
# desc: class to deal with server related jabber events and muc-offline
# presences.
import time
import random
import logging
import slixmpp
logging = logging.getLogger(__name__)
class IdleBot(slixmpp.ClientXMPP):
"""
Connect the given server, logs in and join the given room. If lost
connection to server it tryes to reconnect.
"""
def __init__(self, jid, password, room, nick):
slixmpp.ClientXMPP.__init__(self, jid, password)
self.nick = nick
self.room = room
self.room_roster = {}
self.add_event_handler("session_start", self.start)
self.add_event_handler("session_end", self.reconnect)
self.add_event_handler("disconnected", self.reconnect)
self.add_event_handler("muc::%s::got_online" % self.room,
self.muc_online)
self.add_event_handler("muc::%s::got_offline" % room,
self.muc_offline)
async def start(self, event):
"""
Process the session_start event.
Typical actions for the session_start event are
requesting the roster and broadcasting an initial
presence stanza.
Arguments:
event -- An empty dictionary. The session_start
event does not provide any additional
data.
"""
await self.get_roster()
logging.info('Send presence')
self.send_presence()
self.join_room(self.room)
def join_room(self, room):
'''
Sends a presence stanza for the given chat room.
Arguments:
room -- The room to join.
'''
self.plugin['xep_0045'].join_muc(room,
self.nick,
# If a room password is needed, use:
# password=the_room_password,
wait=True)
logging.info('Joined room {}'.format(room))
self.room_roster[room] = []
def muc_online(self, presence):
"""
Process a presence stanza from a chat room. In this case,
we only add the sers nick to our room roster. Items in
presence['muc'] are 'room', 'nick', 'jid', 'lang', 'role',
'affiliation'. Because 'jid' is (depends on server) possible
empty, we only can add 'nick' to the roster.
Arguments:
presence -- The received presence stanza. See the
documentation for the Presence stanza
to see how else it may be used.
"""
nick = presence['muc']['nick']
room = presence['muc']['room']
if nick not in self.room_roster[room]:
self.room_roster[room].append(nick)
logging.debug('Roster: {}'.format(self.room_roster))
# if bot joins the room great
greeting = ("Hello everybody, my name is {} and i'am the "
"new kid in town. :)".format(self.nick))
if presence['muc']['nick'] == self.nick:
self.send_message(mto = room,
mbody = greeting,
mtype = 'groupchat')
def muc_offline(self, presence):
"""
Process a presence stanza from a chat room. At first we look
for the nick who leaves the room. In case we are the user, we
clear the roster and try to rejoin. Otherwise we remove the nick
from roster.
Arguments:
presence -- The received presence stanza. See the
documentation for the Presence stanza
to see how else it may be used.
"""
nick = presence['muc']['nick']
room = presence['muc']['room']
if nick == self.nick:
self.room_roster[room] = []
logging.info('Receive unavailable from {}'.format(room))
timeout = random.randint(0,10)
logging.debug('Set timeout to {}'.format(timeout))
time.sleep(timeout)
self.join_room(room)
else:
if nick in self.room_roster[room]:
2020-07-26 17:34:07 +02:00
self.room_roster[room].remove(nick)
logging.debug('Roster: {}'.format(self.room_roster))
def reconnect(self, event):
'''
Deals with alls events for disconnections. Tryes to reconnect.
'''
logging.warning('Receive a disconnect event: {}'.format(event))
logging.info('Try to reconnect')
self.connect()
def hangup(self):
'''
Process a disconnect from server. Is only called from
KeyboardInterrup exception to disconnect from server and terminate
the process..
'''
self.disconnect()
def run(self):
'''
Registers needed plugins, connect the server and try to hold this
connection.
'''
self.register_plugin('xep_0045') # Multi-User Chat
self.register_plugin('xep_0012') # Last Activity
self.connect()
try:
self.process(forever=True)
except KeyboardInterrupt:
self.hangup()