refactorization (#1)

* changed all lineendings to lf
* cleaned up main class
* refactor bot
* refactor bot functions
* moved functions.py to classes dir
* code comment changes
* changed code comment style
- removed unnecessary pass statement
+ added missing newline
* simplified function and return statements
This commit is contained in:
nico 2018-10-01 23:17:09 +02:00 committed by GitHub
parent b9a2e5adcd
commit 294a728b0f
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
6 changed files with 460 additions and 351 deletions

View file

@ -28,7 +28,7 @@ Do not opperate this bot on foreign servers.
jid=nick@domain.tld/querybot-0.1 jid=nick@domain.tld/querybot-0.1
password=super_secret_password password=super_secret_password
[MUC] [MUC]
rooms=room_to_connect_to@conference.domain.tld rooms=room_to_connect_to@conference.domain.tld,another_room@conference.domain.tld
nick=mucnickname nick=mucnickname
[ADMIN] [ADMIN]
admins=admins ( ! muc nick and not the jid nickname) admins=admins ( ! muc nick and not the jid nickname)

108
classes/functions.py Normal file
View file

@ -0,0 +1,108 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# XEP-0072: Server Version
class Version:
def __init__(self, version, msg, target):
self.version = version['software_version']['version']
self.os = version['software_version']['os']
self.name = version['software_version']['name']
self.nick = msg['mucnick']
self.message_type = msg['type']
self.target = target
def format_version(self):
if self.message_type == "groupchat":
text = "%s: %s is running %s version %s on %s" % (self.nick, self.target, self.name, self.version, self.os)
else:
text = "%s is running %s version %s on %s" % (self.target, self.name, self.version, self.os)
return text
# XEP-0012: Last Activity
class LastActivity:
""" query the server uptime of the specified domain, defined by XEP-0012 """
def __init__(self, last_activity, msg, target):
self.last_activity = last_activity
self.nick = msg['mucnick']
self.message_type = msg['type']
self.target = target
def format_values(self, granularity=4):
seconds = self.last_activity['last_activity']['seconds']
uptime = []
intervals = (
('years', 31536000), # 60 * 60 * 24 * 365
('weeks', 604800), # 60 * 60 * 24 * 7
('days', 86400), # 60 * 60 * 24
('hours', 3600), # 60 * 60
('minutes', 60),
('seconds', 1)
)
for name, count in intervals:
value = seconds // count
if value:
seconds -= value * count
if value == 1:
name = name.rstrip('s')
uptime.append("{} {}".format(value, name))
result = ' '.join(uptime[:granularity])
if self.message_type == "groupchat":
text = "%s: %s is running since %s" % (self.nick, self.target, result)
else:
text = "%s is running since %s" % (self.target, result)
return text
# XEP-0157: Contact Addresses for XMPP Services
class ContactInfo:
def __init__(self, contact, msg, target):
self.contact = contact
self.message = msg
self.target = target
def format_contact(self):
server_info = []
sep = ' , '
possible_vars = ['abuse-addresses',
'admin-addresses',
'feedback-addresses',
'sales-addresses',
'security-addresses',
'support-addresses']
for field in self.contact['disco_info']['form']:
var = field['var']
if var in possible_vars:
field_value = field.get_value(convert=False)
value = sep.join(field_value) if isinstance(field_value, list) else field_value
server_info.append(' - %s: %s' % (var, value))
if server_info:
text = "contact addresses for %s are" % self.target
for count in range(server_info.__len__()):
text += "\n" + server_info[count]
else:
text = "%s has no contact addresses configured." % self.target
return text
# class handeling XMPPError exeptions
class HandleError:
def __init__(self, error, msg, key, target):
self.error = error
self.message = msg
self.key = key
self.target = target
def build_report(self):
condition = self.error.condition
keyword = self.key[1:]
text = "There was an error requesting " + self.target + '\'s ' + keyword + " : " + condition
return text

51
classes/strings.py Normal file
View file

@ -0,0 +1,51 @@
# -*- coding: utf-8 -*-
from random import randint
class StaticAnswers:
"""
collection of callable static/ semi-static strings
"""
def __init__(self, nick=""):
self.nickname = nick
self.helpfile = {
'help': '!help -- display this text',
'version': '!version domain.tld -- receive XMPP server version',
'uptime': '!uptime domain.tld -- receive XMPP server uptime',
'contact': '!contact domain.tld -- receive XMPP server contact address info'}
self.possible_answers = {
'1': 'I heard that, %s.',
'2': 'I am sorry for that %s.',
'3': '%s did you try turning it off and on again?'}
self.error_messages = {
'1': 'not reachable',
'2': 'not a valid target'
}
self.keywords = {
"keywords": ["!help", "!uptime", "!version", "!contact"],
"no_arg_keywords": ["!help"]
}
def keys(self, arg="", keyword='keywords'):
if arg == 'list':
try:
return self.keywords[keyword]
except KeyError:
return self.keywords['keywords']
else:
return self.keywords
def gen_help(self):
helpdoc = "\n".join(['%s' % value for (_, value) in self.helpfile.items()])
return helpdoc
def gen_answer(self):
possible_answers = self.possible_answers
return possible_answers[str(randint(1, possible_answers.__len__()))] % self.nickname
def error(self,code):
try:
text = self.error_messages[str(code)]
except KeyError:
return 'unknown error'
return text

216
main.py
View file

@ -8,22 +8,23 @@
See the file LICENSE for copying permission. See the file LICENSE for copying permission.
""" """
import asyncio import asyncio
import configparser
import logging
import slixmpp import slixmpp
import ssl import ssl
import validators import validators
import configparser
import logging
from argparse import ArgumentParser from argparse import ArgumentParser
from datetime import datetime, timedelta
from random import randint
from slixmpp.exceptions import XMPPError from slixmpp.exceptions import XMPPError
from classes.strings import StaticAnswers
from classes.functions import Version, LastActivity, ContactInfo, HandleError
class QueryBot(slixmpp.ClientXMPP): class QueryBot(slixmpp.ClientXMPP):
""" A simple Slixmpp bot with some features """
def __init__(self, jid, password, room, nick): def __init__(self, jid, password, room, nick):
slixmpp.ClientXMPP.__init__(self, jid, password) slixmpp.ClientXMPP.__init__(self, jid, password)
self.ssl_version = ssl.PROTOCOL_TLSv1_2
self.room = room self.room = room
self.nick = nick self.nick = nick
@ -35,8 +36,7 @@ class QueryBot(slixmpp.ClientXMPP):
def start(self, event): def start(self, event):
""" """
Arguments: :param str event -- An empty dictionary. The session_start event does not provide any additional data.
event -- An empty dictionary. The session_start event does not provide any additional data.
""" """
self.send_presence() self.send_presence()
self.get_roster() self.get_roster()
@ -45,152 +45,104 @@ class QueryBot(slixmpp.ClientXMPP):
for rooms in self.room.split(sep=","): for rooms in self.room.split(sep=","):
self.plugin['xep_0045'].join_muc(rooms, self.nick, wait=True) self.plugin['xep_0045'].join_muc(rooms, self.nick, wait=True)
def validate_domain(self, wordlist, index):
@staticmethod
def precheck(line):
""" """
pre check function validation method to reduce connection attemps to unvalid domains
- check that keywords are used properly :param wordlist: words seperated by " " from the message
- check that following a keyword a proper jid is following :param index: keyword index inside the message
:param line: line from message body :return: true if valid
:return: true if correct
""" """
keywords = ["!help", "!uptime", "!version", "!contact"] # keyword inside the message
proper_domain, proper_key = False, False 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: try:
# check for valid keyword in position 0 target = wordlist[index + 1]
if line[0] in keywords: if validators.domain(target):
proper_key = True return True
else:
return
# help command is used
if line[0] == "!help":
proper_domain = True
# check if domain is valid
elif validators.domain(line[1]):
proper_domain = True
else:
return
except IndexError: except IndexError:
pass # 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
return proper_key and proper_domain 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 @asyncio.coroutine
def message(self, msg): def message(self, msg):
""" """
Arguments: :param msg: received message stanza
msg -- The received message stanza. See the documentation for stanza objects and the Message stanza to see
how it may be used.
""" """
# init empty reply list
reply = list()
# catch self messages to prevent self flooding # catch self messages to prevent self flooding
if msg['mucnick'] == self.nick: if msg['mucnick'] == self.nick:
return return
elif self.nick in msg['body']:
# add pre predefined text to reply list
reply.append(StaticAnswers(msg['mucnick']).gen_answer())
if self.nick in msg['body']: # building the queue
# answer with predefined text when mucnick is used # double splitting to exclude whitespaces
self.send_message(mto=msg['from'].bare, mbody=notice_answer(msg['mucnick']), mtype=msg['type']) words = " ".join(msg['body'].split()).split(sep=" ")
queue = list()
for line in msg['body'].splitlines(): # check all words in side the message for possible hits
""" split multiline messages into lines to check every line for keywords """ for x in enumerate(words):
line = line.split(sep= " ") # 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]})
if self.precheck(line): # queue
""" true if keyword and domain are valid """ for job in queue:
# Display help for key in job:
if line[0] == '!help': keyword = key
""" display help when keyword !help is recieved """ index = job[key]
self.send_message(mto=msg['from'].bare, mbody=help_doc(), mtype=msg['type'])
if keyword == '!help':
reply.append(StaticAnswers().gen_help())
# XEP-0072: Server Version
if line[0] == '!version':
""" query the server software version of the specified domain, defined by XEP-0092 """
try: try:
version = yield from self['xep_0092'].get_version(line[1]) 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())
if msg['type'] == "groupchat": elif keyword == "!version":
text = "%s: %s is running %s version %s on %s" % (msg['mucnick'], line[1], version[ version = yield from self['xep_0092'].get_version(target)
'software_version']['name'], version['software_version']['version'], version[ reply.append(Version(version, msg, target).format_version())
'software_version']['os'])
else:
text = "%s is running %s version %s on %s" % (line[1], version['software_version'][
'name'], version['software_version']['version'], version['software_version']['os'])
self.send_message(mto=msg['from'].bare, mbody=text, mtype=msg['type']) elif keyword == "!contact":
except NameError: contact = yield from self['xep_0030'].get_info(jid=target, cached=False)
pass reply.append(ContactInfo(contact, msg, target).format_contact())
except XMPPError:
pass
# XEP-0012: Last Activity except XMPPError as error:
if line[0] == '!uptime': reply.append(HandleError(error, msg, key, target).build_report())
""" query the server uptime of the specified domain, defined by XEP-0012 """
try:
# try if domain[0] is set if not just pass
last_activity = yield from self['xep_0012'].get_last_activity(line[1])
uptime = datetime(1, 1, 1) + timedelta(seconds=last_activity['last_activity']['seconds'])
if msg['type'] == "groupchat": # remove None type from list and send all elements
text = "%s: %s is running since %d days %d hours %d minutes" % (msg['mucnick'], line[1], if list(filter(None.__ne__, reply)) and reply:
uptime.day - 1, uptime.hour, reply = self.deduplicate(reply)
uptime.minute) self.send_message(mto=msg['from'].bare, mbody="\n".join(reply), mtype=msg['type'])
else:
text = "%s is running since %d days %d hours %d minutes" % (line[1], uptime.day - 1,
uptime.hour, uptime.minute)
self.send_message(mto=msg['from'].bare, mbody=text, mtype=msg['type'])
except NameError:
pass
except XMPPError:
pass
# XEP-0157: Contact Addresses for XMPP Services
if line[0] == "!contact":
""" query the XEP-0030: Service Discovery and extract contact information """
try:
result = yield from self['xep_0030'].get_info(jid=line[1], cached=False)
server_info = []
for field in result['disco_info']['form']:
var = field['var']
if field['type'] == 'hidden' and var == 'FORM_TYPE':
title = field['value'][0]
continue
sep = ', '
field_value = field.get_value(convert=False)
value = sep.join(field_value) if isinstance(field_value, list) else field_value
server_info.append('%s: %s' % (var, value))
text = "contact addresses for %s are" % (line[1])
for count in range(len(server_info)):
text += "\n" + server_info[count]
self.send_message(mto=msg['from'].bare, mbody=text, mtype=msg['type'])
except NameError:
pass
except XMPPError:
pass
# TODO
# append all results to single message send just once
else:
pass
def help_doc():
helpfile = {'help': '!help -- display this text',
'version': '!version domain.tld -- receive XMPP server version',
'uptime':'!uptime domain.tld -- receive XMPP server uptime',
'contact': '!contact domain.tld -- receive XMPP server contact address info'}
return "".join(['%s\n' % (value) for (_, value) in helpfile.items()])
def notice_answer(nickname):
possible_answers = {'1': 'I heard that, %s.',
'2': 'I am sorry for that %s.',
'3': '%s did you try turning it off and on again?'}
return possible_answers[str(randint(1, len(possible_answers)))] % nickname
if __name__ == '__main__': if __name__ == '__main__':
# command line arguments. # command line arguments.
@ -199,8 +151,8 @@ if __name__ == '__main__':
const=logging.ERROR, default=logging.INFO) const=logging.ERROR, default=logging.INFO)
parser.add_argument('-d', '--debug', help='set logging to DEBUG', action='store_const', dest='loglevel', parser.add_argument('-d', '--debug', help='set logging to DEBUG', action='store_const', dest='loglevel',
const=logging.DEBUG, default=logging.INFO) const=logging.DEBUG, default=logging.INFO)
parser.add_argument('-D', '--dev', help='set logging to console', action='store_const', dest='logfile', parser.add_argument('-D', '--dev', help='set logging to console', action='store_const', dest='logfile', const="",
const="", default='bot.log') default='bot.log')
args = parser.parse_args() args = parser.parse_args()
# logging # logging
@ -214,11 +166,9 @@ if __name__ == '__main__':
args.password = config.get('Account', 'password') args.password = config.get('Account', 'password')
args.room = config.get('MUC', 'rooms') args.room = config.get('MUC', 'rooms')
args.nick = config.get('MUC', 'nick') args.nick = config.get('MUC', 'nick')
args.admins = config.get('ADMIN', 'admins')
# init the bot and register used slixmpp plugins # init the bot and register used slixmpp plugins
xmpp = QueryBot(args.jid, args.password, args.room, args.nick) xmpp = QueryBot(args.jid, args.password, args.room, args.nick)
xmpp.ssl_version = ssl.PROTOCOL_TLSv1_2
xmpp.register_plugin('xep_0012') # Last Activity xmpp.register_plugin('xep_0012') # Last Activity
xmpp.register_plugin('xep_0030') # Service Discovery xmpp.register_plugin('xep_0030') # Service Discovery
xmpp.register_plugin('xep_0045') # Multi-User Chat xmpp.register_plugin('xep_0045') # Multi-User Chat