xmpp-chatbot/main.py
nico 83e5b1f0aa IndexError fix
* prevent IndexError when help is called
2018-10-02 00:31:43 +02:00

184 lines
5.8 KiB
Python

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
Slixmpp: The Slick XMPP Library
Copyright (C) 2010 Nathanael C. Fritz
This file is part of Slixmpp.
See the file LICENSE for copying permission.
"""
import asyncio
import slixmpp
import ssl
import validators
import configparser
import logging
from argparse import ArgumentParser
from slixmpp.exceptions import XMPPError
from classes.strings import StaticAnswers
from classes.functions import Version, LastActivity, ContactInfo, HandleError
class QueryBot(slixmpp.ClientXMPP):
def __init__(self, jid, password, room, nick):
slixmpp.ClientXMPP.__init__(self, jid, password)
self.ssl_version = ssl.PROTOCOL_TLSv1_2
self.room = room
self.nick = nick
# session start event, starting point for the presence and roster requests
self.add_event_handler('session_start', self.start)
# register handler to recieve both groupchat and normal message events
self.add_event_handler('message', self.message)
def start(self, event):
"""
:param str event -- An empty dictionary. The session_start event does not provide any additional data.
"""
self.send_presence()
self.get_roster()
# If a room password is needed, use: password=the_room_password
for rooms in self.room.split(sep=","):
self.plugin['xep_0045'].join_muc(rooms, self.nick, wait=True)
def validate_domain(self, wordlist, index):
"""
validation method to reduce connection attemps to unvalid domains
:param wordlist: words seperated by " " from the message
:param index: keyword index inside the message
:return: true if valid
"""
# keyword inside the message
argument = wordlist[index]
# if the argument is not inside the no_arg_keywords target is index + 1
if argument not in StaticAnswers().keys(arg='list', keyword="no_arg_keywords"):
try:
target = wordlist[index + 1]
if validators.domain(target):
return True
except IndexError:
# except an IndexError if a keywords is the last word in the message
return False
elif argument in StaticAnswers().keys(arg='list', keyword="no_arg_keywords"):
return True
else:
return
def deduplicate(self, reply):
"""
deduplication method for the result list
:param list reply: list containing strings
:return: list containing unique strings
"""
reply_dedup = list()
for item in reply:
if item not in reply_dedup:
reply_dedup.append(item)
return reply_dedup
@asyncio.coroutine
def message(self, msg):
"""
:param msg: received message stanza
"""
# init empty reply list
reply = list()
# catch self messages to prevent self flooding
if msg['mucnick'] == self.nick:
return
elif self.nick in msg['body']:
# add pre predefined text to reply list
reply.append(StaticAnswers(msg['mucnick']).gen_answer())
# building the queue
# double splitting to exclude whitespaces
words = " ".join(msg['body'].split()).split(sep=" ")
queue = list()
# check all words in side the message for possible hits
for x in enumerate(words):
# check word for match in keywords list
for y in StaticAnswers().keys(arg='list'):
# if so queue the keyword and the postion in the string
if x[1] == y:
# only add job to queue if domain is valid
if self.validate_domain(words, x[0]):
queue.append({str(y): x[0]})
# queue
for job in queue:
for key in job:
keyword = key
index = job[key]
if keyword == '!help':
reply.append(StaticAnswers().gen_help())
continue
try:
target = words[index + 1]
if keyword == '!uptime':
last_activity = yield from self['xep_0012'].get_last_activity(target)
reply.append(LastActivity(last_activity, msg, target).format_values())
elif keyword == "!version":
version = yield from self['xep_0092'].get_version(target)
reply.append(Version(version, msg, target).format_version())
elif keyword == "!contact":
contact = yield from self['xep_0030'].get_info(jid=target, cached=False)
reply.append(ContactInfo(contact, msg, target).format_contact())
except XMPPError as error:
reply.append(HandleError(error, msg, key, target).build_report())
# remove None type from list and send all elements
if list(filter(None.__ne__, reply)) and reply:
reply = self.deduplicate(reply)
self.send_message(mto=msg['from'].bare, mbody="\n".join(reply), mtype=msg['type'])
if __name__ == '__main__':
# command line arguments.
parser = ArgumentParser()
parser.add_argument('-q', '--quiet', help='set logging to ERROR', action='store_const', dest='loglevel',
const=logging.ERROR, default=logging.INFO)
parser.add_argument('-d', '--debug', help='set logging to DEBUG', action='store_const', dest='loglevel',
const=logging.DEBUG, default=logging.INFO)
parser.add_argument('-D', '--dev', help='set logging to console', action='store_const', dest='logfile', const="",
default='bot.log')
args = parser.parse_args()
# logging
logging.basicConfig(filename=args.logfile, level=args.loglevel, format='%(levelname)-8s %(message)s')
logger = logging.getLogger(__name__)
# configfile
config = configparser.RawConfigParser()
config.read('./bot.cfg')
args.jid = config.get('Account', 'jid')
args.password = config.get('Account', 'password')
args.room = config.get('MUC', 'rooms')
args.nick = config.get('MUC', 'nick')
# init the bot and register used slixmpp plugins
xmpp = QueryBot(args.jid, args.password, args.room, args.nick)
xmpp.register_plugin('xep_0012') # Last Activity
xmpp.register_plugin('xep_0030') # Service Discovery
xmpp.register_plugin('xep_0045') # Multi-User Chat
xmpp.register_plugin('xep_0060') # PubSub
xmpp.register_plugin('xep_0085') # Chat State Notifications
xmpp.register_plugin('xep_0092') # Software Version
xmpp.register_plugin('xep_0128') # Service Discovery Extensions
xmpp.register_plugin('xep_0199') # XMPP Ping
# connect and start receiving stanzas
xmpp.connect()
xmpp.process()