summaryrefslogtreecommitdiffstats
path: root/main.py
diff options
context:
space:
mode:
Diffstat (limited to 'main.py')
-rwxr-xr-x[-rw-r--r--]main.py209
1 files changed, 102 insertions, 107 deletions
diff --git a/main.py b/main.py
index 2625728..50f9b03 100644..100755
--- a/main.py
+++ b/main.py
@@ -1,24 +1,25 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
- Slixmpp: The Slick XMPP Library
- Copyright (C) 2010 Nathanael C. Fritz
- This file is part of Slixmpp.
+ James the MagicXMPP Bot
+ build with Slick XMPP Library
+ Copyright (C) 2018 Nico Wellpott
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
+import common.misc as misc
+from common.strings import StaticAnswers
+from classes.servercontact import ServerContact
+from classes.version import Version
+from classes.uptime import LastActivity
from classes.xep import XEPRequest
@@ -28,11 +29,19 @@ class QueryBot(slixmpp.ClientXMPP):
self.ssl_version = ssl.PROTOCOL_TLSv1_2
self.room = room
self.nick = nick
+ self.use_message_ids = True
+
+ self.functions = {
+ "!uptime": LastActivity(),
+ "!contact": ServerContact(),
+ "!version": Version(),
+ "!xep": XEPRequest()
+ }
# 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
+ # register receive handler for both groupchat and normal message events
self.add_event_handler('message', self.message)
def start(self, event):
@@ -43,125 +52,111 @@ class QueryBot(slixmpp.ClientXMPP):
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 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]
-
- # 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 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
- else:
- return False
-
- 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
+ if self.room:
+ for rooms in self.room.split(sep=","):
+ logging.debug("joining: %s" % rooms)
+ self.plugin['xep_0045'].join_muc(rooms, self.nick, wait=True)
- @asyncio.coroutine
- def message(self, msg):
+ async def message(self, msg):
"""
:param msg: received message stanza
"""
- # init empty reply list
- reply = list()
+ data = {
+ 'words': list(),
+ 'reply': list(),
+ 'queue': 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())
+ data['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]})
+ data = self.build_queue(data, msg)
# queue
- for job in queue:
- for key in job:
- keyword = key
- index = job[key]
+ for job in data['queue']:
+ keys = list(job.keys())
+ keyword = keys[0]
+
+ target = job[keyword][0]
+ opt_arg = job[keyword][1]
+ query = None
- if keyword == '!help':
- reply.append(StaticAnswers().gen_help())
- continue
+ if keyword == '!help':
+ data['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())
+ try:
+ if keyword == "!uptime":
+ query = await self['xep_0012'].get_last_activity(jid=target)
- elif keyword == "!version":
- version = yield from self['xep_0092'].get_version(target)
- reply.append(Version(version, msg, target).format_version())
+ elif keyword == "!version":
+ query = await self['xep_0092'].get_version(jid=target)
- elif keyword == "!contact":
- contact = yield from self['xep_0030'].get_info(jid=target, cached=False)
- reply.append(ContactInfo(contact, msg, target).format_contact())
+ elif keyword == "!contact":
+ query = await self['xep_0030'].get_info(jid=target, cached=False)
- elif keyword == "!xep":
- reply.append(XEPRequest(msg, target).format())
+ except XMPPError as error:
+ logging.info(misc.HandleError(error, keyword, target).report())
+ data['reply'].append(misc.HandleError(error, keyword, target).report())
+ continue
- except XMPPError as error:
- reply.append(HandleError(error, msg, key, target).build_report())
+ data["reply"].append(self.functions[keyword].format(query=query, target=target, opt_arg=opt_arg))
# remove None type from list and send all elements
- if list(filter(None.__ne__, reply)) and reply:
- reply = self.deduplicate(reply)
+ if list(filter(None.__ne__, data['reply'])) and data['reply']:
+
+ # if msg type is groupchat prepend mucnick
+ if msg["type"] == "groupchat":
+ data["reply"][0] = "%s: " % msg["mucnick"] + data["reply"][0]
+
+ # reply = misc.deduplicate(data['reply'])
+ reply = data["reply"]
self.send_message(mto=msg['from'].bare, mbody="\n".join(reply), mtype=msg['type'])
+ def build_queue(self, data, msg):
+ # building the queue
+ # double splitting to exclude whitespaces
+ data['words'] = " ".join(msg['body'].split()).split(sep=" ")
+ wordcount = len(data["words"])
+
+ # check all words in side the message for possible hits
+ for x in enumerate(data['words']):
+ # check for valid keywords
+ index = x[0]
+ keyword = x[1]
+
+ # match all words starting with ! and member of no_arg_keywords
+ if keyword.startswith("!") and keyword in StaticAnswers().keys("no_arg_keywords"):
+ data['queue'].append({keyword: [None, None]})
+
+ # matching all words starting with ! and member of keywords
+ elif keyword.startswith("!") and keyword in StaticAnswers().keys("keywords"):
+ # init variables to circumvent IndexErrors
+ target, opt_arg = None, None
+
+ # compare to wordcount if assignment is possible
+ if index + 1 < wordcount:
+ target = data["words"][index + 1]
+
+ if index + 2 < wordcount:
+ if not data["words"][index + 2].startswith("!"):
+ opt_arg = data["words"][index + 2]
+
+ # only add job to queue if domain is valid
+ if misc.validate(keyword, target):
+ logging.debug("Item added to queue %s" % {str(keyword): [target, opt_arg]})
+ data['queue'].append({str(keyword): [target, opt_arg]})
+
+ # deduplicate queue elements
+ data["queue"] = misc.deduplicate(data["queue"])
+
+ return data
+
if __name__ == '__main__':
# command line arguments.
@@ -170,12 +165,12 @@ if __name__ == '__main__':
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')
+ 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')
+ logging.basicConfig(filename=args.logfile, level=args.loglevel, format='%(levelname)s: %(asctime)s: %(message)s')
logger = logging.getLogger(__name__)
# configfile
generated by cgit v1.2.3-18-g5258 (git 2.32.0) at 2024-04-27 12:06:12 +0200