diff options
| -rw-r--r-- | .gitignore | 2 | ||||
| -rw-r--r-- | classes/functions.py | 2 | ||||
| -rw-r--r-- | classes/strings.py | 9 | ||||
| -rw-r--r-- | classes/xep.py | 87 | ||||
| -rw-r--r-- | main.py | 44 |
5 files changed, 127 insertions, 17 deletions
@@ -60,4 +60,6 @@ target/ # .idea .idea +.etag bot\.cfg +xeplist.xml diff --git a/classes/functions.py b/classes/functions.py index 08d6146..a8ed356 100644 --- a/classes/functions.py +++ b/classes/functions.py @@ -93,7 +93,7 @@ class ContactInfo: # class handeling XMPPError exeptions class HandleError: - def __init__(self, error, msg, key, target): + def __init__(self, error, msg, key, target="target missing"): self.error = error self.message = msg self.key = key diff --git a/classes/strings.py b/classes/strings.py index ade0520..6866a31 100644 --- a/classes/strings.py +++ b/classes/strings.py @@ -12,7 +12,8 @@ class StaticAnswers: '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'} + 'contact': '!contact domain.tld -- receive XMPP server contact address info', + 'xep': '!xep XEP Number -- recieve information about the specified XEP'} self.possible_answers = { '1': 'I heard that, %s.', '2': 'I am sorry for that %s.', @@ -22,8 +23,10 @@ class StaticAnswers: '2': 'not a valid target' } self.keywords = { - "keywords": ["!help", "!uptime", "!version", "!contact"], - "no_arg_keywords": ["!help"] + "keywords": ["!help", "!uptime", "!version", "!contact", "!xep"], + "domain_keywords": ["!uptime", "!version", "!contact"], + "no_arg_keywords": ["!help"], + "number_keywords": ["!xep"] } def keys(self, arg="", keyword='keywords'): diff --git a/classes/xep.py b/classes/xep.py new file mode 100644 index 0000000..9e4f61f --- /dev/null +++ b/classes/xep.py @@ -0,0 +1,87 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- +import requests +import defusedxml.ElementTree as ET + + +class XEPRequest: + def __init__(self, msg, xepnumber): + """ + class which requests the header of the referenced xep + :param xepnumber: number int or str to request the xep for + """ + self.message_type = msg['type'] + self.muc_nick = msg['mucnick'] + + self.reqxep = str(xepnumber) + self.xeplist = None + self.acceptedxeps = list() + + def req_xeplist(self): + """ + query and save the current xep list to reduce network bandwidth + """ + try: + with open(".etag") as file: + local_etag = file.read() + except FileNotFoundError: + local_etag = "" + + with requests.Session() as s: + s.headers.update({'Accept': 'application/xml'}) + head = s.head("https://xmpp.org/extensions/xeplist.xml") + etag = head.headers['etag'] + + if local_etag == etag: + with open("xeplist.xml", "r") as file: + self.xeplist = ET.fromstring(file.read()) + else: + r = s.get("https://xmpp.org/extensions/xeplist.xml") + r.encoding = 'utf-8' + local_etag = head.headers['etag'] + + with open("xeplist.xml", "w") as file: + file.write(r.content.decode()) + self.xeplist = ET.fromstring(r.content.decode()) + + with open('.etag', 'w') as string: + string.write(local_etag) + + # populate xep comparison list + for xep in self.xeplist.findall(".//*[@accepted='true']/number"): + self.acceptedxeps.append(xep.text) + + def get(self): + """ + function to query the xep entry if xepnumber is present in xeplist + :return: nicely formatted xep header information + """ + # check if xeplist is accurate + self.req_xeplist() + + result = list() + # if requested number is inside acceptedxeps continou + if self.reqxep in self.acceptedxeps: + searchstring = ".//*[@accepted='true']/[number='%s']" % self.reqxep + + for item in self.xeplist.findall(searchstring): + for x in range(1,5): + result.append(item[x].tag + " : " + item[x].text) + + else: + if self.message_type == "groupchat": + result.append(self.muc_nick + " : " + "XEP-" + str(self.reqxep) + " : is not available.") + else: + result.append("XEP-" + str(self.reqxep) + " : is not available.") + + return result + + def format(self): + reply = self.get() + if self.message_type == "groupchat": + text = "%s: " % self.muc_nick + reply[0] = text + reply[0] + + text = '\n'.join(reply) + + return text @@ -19,6 +19,7 @@ from slixmpp.exceptions import XMPPError from classes.strings import StaticAnswers from classes.functions import Version, LastActivity, ContactInfo, HandleError +from classes.xep import XEPRequest class QueryBot(slixmpp.ClientXMPP): @@ -36,7 +37,7 @@ class QueryBot(slixmpp.ClientXMPP): def start(self, event): """ - :param str event -- An empty dictionary. The session_start event does not provide any additional data. + :param event -- An empty dictionary. The session_start event does not provide any additional data. """ self.send_presence() self.get_roster() @@ -47,27 +48,41 @@ class QueryBot(slixmpp.ClientXMPP): def validate_domain(self, wordlist, index): """ - validation method to reduce connection attemps to unvalid domains - :param wordlist: words seperated by " " from the message + validation method to reduce malformed querys and unnecessary connection attempts + :param wordlist: words separated 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): + # check if argument is in the argument list + if argument in StaticAnswers().keys(arg='list'): + # if argument uses a domain check for occurence in list and check domain + if argument in StaticAnswers().keys(arg='list', keyword='domain_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 + + # check if number keyword is used if true check if target is assignable + elif argument in StaticAnswers().keys(arg='list', keyword='number_keywords'): + try: + target = wordlist[index + 1] return True - except IndexError: - # except an IndexError if a keywords is the last word in the message + except IndexError: + # except an IndexError if target is not assignable + return False + # check if argument is inside no_arg list + elif argument in StaticAnswers().keys(arg='list', keyword="no_arg_keywords"): + return True + else: return False - elif argument in StaticAnswers().keys(arg='list', keyword="no_arg_keywords"): - return True else: - return + return False def deduplicate(self, reply): """ @@ -136,6 +151,9 @@ class QueryBot(slixmpp.ClientXMPP): contact = yield from self['xep_0030'].get_info(jid=target, cached=False) reply.append(ContactInfo(contact, msg, target).format_contact()) + elif keyword == "!xep": + reply.append(XEPRequest(msg, target).format()) + except XMPPError as error: reply.append(HandleError(error, msg, key, target).build_report()) |