Krautspace/hackbot
2
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity

initial commit for a jabberbot based on slixmpp

Browse source
This commit is contained in:
+++ 2020-07-26 14:27:05 +02:00
commit 43ff022969
12 changed files with 1299 additions and 0 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

114
common.py Normal file
View file

@ -0,0 +1,114 @@
#!/usr/bin/python3
# -*- coding: utf-8 -*-
# file: common.py
# date: 24.07.2020
# desc: common functions related to hackbot.py
import slixmpp
def get_type_from_stanza(stanza):
'''
Returns the type of a stanza.
param 1: stanza object
returns: string or false
'''
if isinstance(stanza, slixmpp.Message):
return stanza.get_type()
return False
def get_body_from_stanza(stanza):
'''
Extracts the body from the given stanza.
returns: string or false
'''
message_type = get_type_from_stanza(stanza)
if message_type is not False:
if message_type == 'groupchat':
return stanza['body']
elif message_type in ('chat', 'normal', 'error', 'headline'):
print('\tTyp not supported yet: {}'.format(message_type))
return False
return False
def get_command_from_body(stanza):
'''
Checks if the given stanzas body starts with a command. Returns command
or false.
param 1: stanza object
returns: string or false
'''
body = get_body_from_stanza(stanza)
if body is not False:
if body.lstrip().startswith('!'):
word = body.split()[0]
if word[1:].isalpha():
return word[1:].lower()
return False
def get_arguments_from_body(stanza):
'''
Grabs all from body behind the leading word.
param 1: stanza object
returns: list or false
'''
body = get_body_from_stanza(stanza)
if body is not False:
behind = body.split()[1:]
if len(behind) != 0:
return behind
return False
def get_nick_from_stanza(stanza):
'''
'''
if isinstance(stanza, slixmpp.Message):
message_type = stanza.get_type()
if message_type == 'groupchat':
return stanza.get_mucnick()
elif isinstance(stanza, slixmpp.Presence):
jid = stanza.getFrom()
return jid.resource
else: print('Unhandled message: {}'.format(str(stanza)))
# Elements and functions at a message object:
# print('\nfrom: {}'.format(msg.get_from()))
# print('bare: {}'.format(msg.get_from().bare))
# print('node: {}'.format(msg.get_from().node))
# print('domain: {}'.format(msg.get_from().domain))
# print('resource: {}'.format(msg.get_from().resource))
# print('lang: {}'.format(msg.get_lang()))
# print('muc nick: {}'.format(msg.get_mucnick()))
# print('muc room: {}'.format(msg.get_mucroom()))
# print('parent thread: {}'.format(msg.get_parent_thread()))
# print('payload: {}'.format(msg.get_payload()))
# print('values: {}'.format(msg.get_stanza_values()))
# print('to: {}'.format(msg.get_to()))
# print('type: {}'.format(msg.get_type()))
# print('\npayload:')
# for i in msg.get_payload():
# print('keys: {}'.format(i.keys()))
# print('items: {}'.format(i.items()))
# print('tag: {}'.format(i.tag))
# print('text: {}'.format(i.text))
# print('\nvalues:')
# for i in msg.get_stanza_values():
# print('{}: {}'.format(i, msg[i]))
# print('\n:')
#
# mlang = msg['lang']
# mnick = msg['mucnick']
# mbody = msg['body']
# mroom = msg['mucroom']
# mfrom = msg['from']
# mtype = msg['type']
# mid = msg['id']
# mto = msg['to']

24
constants.py Normal file
View file

@ -0,0 +1,24 @@
#!/usr/bin/python3
# -*- coding: utf-8 -*-
# file: constants.py
# date: 26.07.2020
# desc: provides a class with a read only variable (constante). idea found at
# https://stackoverflow.com/questions/2682745/how-do-i-create-a-constant-in-python
# Modul is used to provide hackbots start time. As hackbot starts plugin manager
# import all modules to grab command and description. If uptime is imported
# itself imports constants and creates the starttime.
import time
class Const:
'''
Class to provide a readonly constant.
'''
__slots__ = ()
birth = time.strftime('%Y-%m-%dT%H:%M:%S', time.gmtime())

9
hackbot.conf.templ Normal file
View file

@ -0,0 +1,9 @@
[settings]
loglevel = info
plugindir = ./plugins
[jabber]
jid = nick@jabber.example.com
password = strong-secure-password
room = room@chat.jabber.example.com
nick = mynick

146
hackbot.py Normal file
View file

@ -0,0 +1,146 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# file: hackbot.py
# date: 24.07.2020
# desc: class to deal with presence and messages in the given muc.
import sys
import logging
import common
from idlebot import IdleBot
from manager import PluginManager
class HackBot(IdleBot):
'''
Deals with the messages and presences from the muc.
'''
def __init__(self, jid, password, room, nick, plugin_dir):
IdleBot.__init__(self, jid, password, room, nick)
self.add_event_handler("groupchat_message", self.muc_message)
self.plugin_manager = PluginManager(plugin_dir)
self.plugin_store = self.plugin_manager.collect_plugins()
def muc_message(self, msg):
"""
Process incoming message stanzas from any chat room. Be aware
that if you also have any handlers for the 'message' event,
message stanzas may be processed by both handlers, so check
the 'type' attribute when using a 'message' event handler.
Arguments:
msg -- The received message stanza. See the documentation
for stanza objects and the Message stanza to see
how it may be used.
"""
# dont answer myself ... prevent self flooding
if msg['mucnick'] == self.nick:
logging.debug('Message from myself ... ignored')
return
# check for command
command = common.get_command_from_body(msg)
if command is not False:
logging.debug('Command received: {}'.format(command))
# if command is help (needs a better argument handling)
if command == 'help':
arguments = common.get_arguments_from_body(msg)
logging.debug('Arguments: {}'.format(arguments))
if arguments is False:
msg.reply(self.help()).send()
else: msg.reply(self.help(arguments[0].strip())).send()
# command refernces a plugin
elif command in self.plugin_store.keys():
logging.debug('Valid comand: {}'.format(command))
self.run_plugin(command, msg, self.answer_muc)
# command is unknown
else:
logging.warning('Unknown command: {}'.format(command))
message = ': '.join((msg['mucnick'],
'{} is not a valid command'.format(msg)))
self.answer_muc(message)
# only for debugging
else: logging.debug('No command found')
def run_plugin(self, command, stanza, callback):
'''
Creates a instance from the module is stored in plugin store under
the given command key and runs it.
Arguments:
command -- The command is received in MUC and matched a key in
the plugin store (string).
stanza -- The message object caused the call.
callback -- Function to post the response from plugin.
'''
instance = self.plugin_store[command].Plugin(self.answer_muc)
instance.run(stanza)
def answer_muc(self, message, room=None):
'''
Sends a message to the given room.
Arguments:
message -- The message to send (string).
room -- The room where to send (string).
'''
if room is None:
room = self.room
self.send_message(mto = room,
mbody = message,
mtype = 'groupchat')
def help(self, *command):
'''
Checks if arguments is false or not. Depends on this result it
calles long or short help.
Arguments:
command -- The command for which help requests. Optional.
Returns:
help -- string
'''
if not command or command is False:
logging.debug('Empty help request. Send all commands.')
return self.help_overview()
else: return self.help_command(command)
def help_overview(self):
'''
Grabs short desciptions from all available plugins an deliver it to
MUC.
Returns:
helpstring -- string
'''
commands = []
helpstring = 'Available commands:'
for key in self.plugin_store.keys():
if key == 'help':
continue
commands.append(key)
commands.sort()
for key in commands:
description = self.plugin_store[key].Plugin.get_description()
line = '{0:10s}: {1}'.format(key, description)
helpstring = '\n'.join((helpstring, line))
return helpstring
def help_command(self, command):
'''
param 1: tuple (with one element)
'''
for i in command:
if i not in self.plugin_store.keys():
msg = '"{}" is not a valid argument for help'.format(i)
return msg
instance = self.plugin_store[i].Plugin()
return instance.help()

148
idlebot.py Normal file
View file

@ -0,0 +1,148 @@
#!/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]:
sel.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))
self.disconnect()
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()

96
manager.py Normal file
View file

@ -0,0 +1,96 @@
#!/usr/bin/python3
# -*- coding: utf-8 -*-
import sys
import logging
logging = logging.getLogger()
import os.path
import importlib
from os import listdir
class Plugin(object):
'''
Proto type of plugin class. Only subclasses are valid plugins.
'''
__command = ''
@staticmethod
def get_module():
return Plugin.__module
@staticmethod
def get_command():
return Plugin.__command
@staticmethod
def get_description():
return Plugin.__description
def __init__(self, callback):
self.callback = callback
def help(self):
return 'Sorry ... Help is unavailable at the moment'
def run(self, msg):
self.callback('Sorry ... Run is unavailable at the moment')
class PluginManager():
'''
Handles the plugins. Optional becomes a directory for search plugins.
'''
def __init__(self, plug_dir=None):
'''
Initialize callback function and plugin directory. If no callback is
given it returns imediality. If no plugin directory is given it uses
actual directory.
'''
if plug_dir is None:
self.plugin_dir = './plugins'
else: self.plugin_dir = plug_dir
def collect_plugins(self):
'''
Find all files in plugin directory and grabs filename, provided
command and short description.
'''
self.plugins = {}
search_dir = os.path.realpath(self.plugin_dir)
sys.path.insert(0, search_dir)
logging.debug('Search plugins in {}'.format(search_dir))
files = [x[:-3] for x in os.listdir(search_dir) if x.endswith('.py')]
for filename in files:
plugin = self.import_plugin(filename)
if plugin is False:
continue
command = plugin.Plugin.get_command()
self.plugins[command] = plugin
return self.plugins
def import_plugin(self, filename):
'''
Imports or reimports a module depending its known or not. It's
tested by the command provided by the plugin. (Not best praxis but
the easyst way. And i dont know, what a trouble is causes, if a
module ist moved.)
param 1: string
'''
try:
plugin = importlib.import_module(filename)
except Exception as e:
logging.error('Cant import module: {}'.format(filename))
logging.error('Exception: {}'.format(e))
return False
if not issubclass(plugin.Plugin, Plugin):
logging.error('Not a valid plugin: {}'.format(filename))
return False
logging.debug('Found plugin {}.'.format(filename))
return plugin

159
plugins/dsa.py Normal file
View file

@ -0,0 +1,159 @@
#!/usr/bin/python
# -*- coding: utf-8 -*-
# file: dsa.py
# date: 26.07.2020
# desc: Serves debians security alerts
import logging
import urllib3
import threading
from lxml import etree
from manager import Plugin
import common
logging = logging.getLogger()
class Plugin(Plugin):
'''
Fetchs debians security alerts, grabs title and links and sends it to
the muc.
'''
__module = __name__
__command = 'dsa'
__description = 'Serves debians security alerts'
@staticmethod
def get_module():
return Plugin.__module
@staticmethod
def get_command():
return Plugin.__command
@staticmethod
def get_description():
return Plugin.__description
def __init__(self, callback=None):
self.callback = callback
def help(self):
return ('!dsa serves the actual debian security alerts. A given '
' number reduces the count of alerts displayed to number. '
'Not implemented at the moment.'
'\nSyntax: !dsa <number>')
def run(self, stanza):
'''
Starts a thread to grab debians security alerts returns
immediately.
param 1: stanza object
'''
call_msg = 'Call "!help {}"'.format(self.get_command())
no_count_msg = ' '.join(('Not a valid count: "{}"!', call_msg))
count = False
muc_nick = common.get_nick_from_stanza(stanza)
arguments = common.get_arguments_from_body(stanza)
if arguments is not False:
count = self.get_count(arguments[0])
if count is False:
self.callback(': '.join((muc_nick, no_count_msg)))
return
logging.debug('Start thread for debian security alerts')
dsa_thread = DsaThread(self.callback, count)
dsa_thread.run()
logging.debug('DSA Thread started')
def get_count(self, item):
'''
Try to convert a string into integer.
param 1: string
retuns: integer or false
'''
try:
value = int(item.strip())
return value
except Exception as e:
logging.warning('Invalid value for count: {}'.format(item))
logging.warning('Exception: {}'.format(e))
return False
class DsaThread(threading.Thread):
'''
The thread who fetched and returns the wp search.
'''
def __init__(self, callback, count):
threading.Thread.__init__(self)
self.callback = callback
self.count = count
def run(self):
'''
Starts the thread.
'''
dsa_response = self.get_file()
if dsa_response == False:
self.callback('Error while fetching DSA')
else:
status = dsa_response.status
logging.debug("Server returns {}".format(status))
if status != 200:
self.callback('Server returns {}'.format(status))
xmldoc = etree.fromstring(dsa_response.data)
message = self.string_factory(xmldoc)
self.callback(message)
def string_factory(self, xmldoc):
'''
Extracts interested things from the given dsa xml document and
creates a string to post im muc.
param 1: xml object
'''
message = 'Debian Security Alerts:'
nsmap = {
"purl": "http://purl.org/rss/1.0/",
"rdf": "http://www.w3.org/1999/02/22-rdf-syntax-ns#",
}
about_list = xmldoc.xpath('//purl:item/@rdf:about', namespaces=nsmap)
for about in reversed(about_list):
dsa_id = self.get_id_from_about(about)
title = xmldoc.xpath(
'//purl:item[@rdf:about="{}"]/purl:title/text()'.format(
about), namespaces=nsmap)[0]
message = '\n'.join((message, title))
return message
def get_file(self):
'''
Fetchs the security alerts from debian.org
param 1: string
returns: request object or false
'''
url = 'https://www.debian.org/security/dsa-long'
logging.debug('Try to fetch {}'.format(url))
http = urllib3.PoolManager()
try:
dsa_response = http.request('Get', url)
return dsa_response
except:
logging.debug('{}: failed to fetch'.format(url))
return False
def get_id_from_about(self, about):
'''
Extracts the dsa id from tehe given string.
param 1: string
'''
return int(about.split('/')[-1].split('-')[1])

131
plugins/status.py Normal file
View file

@ -0,0 +1,131 @@
#!/usr/bin/python3
# -*- coding: utf-8 -*-
# file: ./plugins/status.py
# date: 23.07.2020
# desc: status plugin ... grabs the krautspaces door status from
# https://status.kraut.space/api and returns the result.
import urllib3
import codecs
import json
import time
import threading
import logging
logging = logging.getLogger()
from manager import Plugin
lock = threading.Lock()
class Plugin(Plugin):
__module = 'status'
__command = 'status'
__description = 'Deliver krautspace door status'
@staticmethod
def get_module():
return Plugin.__module
@staticmethod
def get_command():
return Plugin.__command
@staticmethod
def get_description():
return Plugin.__description
def __init__(self, callback=None):
self.callback = callback
def help(self):
return ('!status grabs the doorstatus from https://status.kraut.space '
'and delivers him to MUC.\nSyntax: !status')
def run(self, stanza):
'''
Starts a thread to grab krautspaces door status und returns
immediately.
'''
logging.debug('Start thread for status')
api_thread = ApiThread(self.callback)
api_thread.run()
logging.debug('Status thread started')
class ApiThread(threading.Thread):
'''
The thread who fetched, parsed and returns the door status.
'''
def __init__(self, callback):
threading.Thread.__init__(self)
self.callback = callback
def run(self):
'''
Runs the thread.
'''
api_page = self.get_file()
if api_page == None:
self.go_back('Error while connecting API')
else:
status = api_page.status
logging.debug("Page returns {}".format(status))
if status == 200:
message = self.parse_api(api_page.data)
self.go_back(message)
else:
self.go_back('Error while fetching API')
def get_file(self):
'''
Grabs the API file, parse it and returns a json ... otherwise none.
returns: json or none
'''
url = 'https://status.kraut.space/api'
logging.debug('Try to fetch {}'.format(url))
http = urllib3.PoolManager()
try:
api_page = http.request('Get', url)
logging.debug('{}: successfull fetched'.format(url))
except:
logging.debug('{}: failed to fetch'.format(url))
return None
return api_page
def parse_api(self, page_data):
'''
Extracts needed data from given json and create the message.
param 1: json
returns: string
'''
timestamp = None
status = None
message = None
json_string = page_data.decode('utf-8')
json_dict = json.loads(json_string)
status = json_dict['state']['open']
unixtime = json_dict['state']['lastchange']
timestamp = time.strftime('%d.%m.%Y %H:%M', time.localtime(unixtime))
logging.debug('Open: {}; Time: {}; Last Change: {}'.format(
status, timestamp, unixtime))
if status is True:
message = 'Space is open since {}.'.format(timestamp)
elif status is False:
message = 'Space is closed since {}.'.format(timestamp)
else:
message = 'Invalid status: "{}"'.format(status)
return message
def go_back(self, message):
'''
param 1: string
'''
lock.acquire()
try:
self.callback(message)
finally:
lock.release()

151
plugins/timer.py Normal file
View file

@ -0,0 +1,151 @@
#!/usr/bin/python3
# -*- coding: utf-8 -*-
# file: timer.py
# date: 25.07.2020
# desc: Starts a time for max. 24 hours and sends a message after it has
# expied.
import common
import logging
logging = logging.getLogger()
from manager import Plugin
from threading import Timer, Lock
lock = Lock()
class Plugin(Plugin):
'''
Timer starts a timer and sends a message to muc after it has expired.
'''
__module = 'timer'
__command = 'timer'
__description = 'Starts a timer.'
@staticmethod
def get_module():
return Plugin.__module
@staticmethod
def get_command():
return Plugin.__command
@staticmethod
def get_description():
return Plugin.__description
def __init__(self, callback=None):
self.callback = callback
def help(self):
return ('!timer sets a timer that sends a completion message when it '
'has expired. The function expects a time value and '
'optionally a unit. Possible units are seconds, seconde, '
'sekunden, sec, sek, s, minute, minutes, minuten, min, '
'hour, hours, stunde, stunden or h. If no unit is given it '
'uses seconds as default. The timer runs for a maximum of 24 '
'hours.'
'\nSyntax: !timer <value> <unit>')
def run(self, stanza):
'''
'''
self.units = ('seconde', 'seconds', 'sekunde', 'sekunden',
'sec', 'sek', 's',
'minute', 'minutes', 'minuten', 'min', 'm',
'hour', 'hours', 'h', 'stunde', 'stunden')
call_msg = 'Call "!help {}"'.format(self.get_command())
no_args_msg = ' '.join(('Timer without time!', call_msg))
no_valu_msg = ' '.join(('Not a valid value: "{}"!', call_msg))
no_unit_msg = ' '.join(('Not a valid unit: "{}"!', call_msg))
to_long_msg = 'Sorry ... but i dont want stay here for this time'
value = None
unit = None
muc_nick = common.get_nick_from_stanza(stanza)
arguments = common.get_arguments_from_body(stanza)
logging.debug('Arguments: {}'.format(arguments))
if arguments is False:
logging.warning('No arguments for timer. Abort.')
self.callback(': '.join((muc_nick, no_args_msg)))
return
else:
try:
value = self.get_timer_value(arguments[0])
unit = self.get_timer_unit(arguments[1])
except IndexError:
pass
except Exception as e:
logging.warning('Error while creating timer')
logging.warning('Exception: {}'.format(e))
if value in (None, False):
msg = ': '.join((muc_nick, no_valu_msg.format(arguments[0])))
self.callback(msg)
return
if unit is False:
msg = ': '.join((muc_nick, no_unit_msg.format(arguments[1])))
self.callback(msg)
return
# timer starten
elif unit is None:
self.start_timer(value, self.callback, muc_nick)
else:
value = value * unit
if value > 5184000:
logging.warning('Timer value to hight: {}'.format(value))
self.callback(': '.join((muc_nick, to_long_msg)))
self.start_timer(value, self.callback, muc_nick)
def get_timer_value(self, item):
'''
Try to convert a string into integer.
param 1: string
retuns: integer or false
'''
try:
value = int(item.strip())
return value
except Exception as e:
logging.warning('Invalid value for timer: {}'.format(item))
logging.warning('Exception: {}'.format(e))
return False
def get_timer_unit(self, item):
'''
param 1: string
returns: integer
'''
if item.strip() in self.units[0:7]:
logging.debug('Timer unit: seconds')
factor = 1
elif item.strip() in self.units[7:12]:
logging.debug('Timer unit: minutes')
factor = 60
elif item.strip() in self.units[13:17]:
logging.debug('Timer unit: hours')
factor = 60 * 60
else:
logging.warning('Invalid unit for timer: {}'.format(item.strip()))
factor = False
return factor
def start_timer(self, value, callback, muc_nick):
'''
Starts the timer. Arguments are the duration in seconds and the
chatters nick who called timer.
param 1: integer
param 2: string
'''
timer_start_msg = 'Timer started'
timer_ends_msg = '{}: Timer finished!'.format(muc_nick)
t = Timer(value, callback, [timer_ends_msg])
t.start()
logging.debug('Timer started for {} seconds'.format(value))
self.callback(': '.join((muc_nick, timer_start_msg)))

84
plugins/uptime.py Normal file
View file

@ -0,0 +1,84 @@
#!/usr/bin/python
# -*- coding: utf-8 -*-
# file: uptime.py
# date: 25.07.2020
# desc: Returns bots uptime
import time
import datetime
import logging
logging = logging.getLogger()
from manager import Plugin
from constants import Const
class Plugin(Plugin):
'''
Returns the hackbots uptime. The variable birth from the globalvar
module is used to become the starttime.
'''
__module = __name__
__command = 'uptime'
__description = 'Returns hackbots uptime'
@staticmethod
def get_module():
return Plugin.__module
@staticmethod
def get_command():
return Plugin.__command
@staticmethod
def get_description():
return Plugin.__description
def __init__(self, callback=None):
self.callback = callback
def help(self):
return ('!uptime returns hackbots uptime. There are no options. '
'\nSyntax: !uptime')
def run(self, stanza):
'''
Starts the plugin.
'''
const = Const()
birth = const.birth
uptime = self.calculate_uptime(birth)
if uptime == False:
logging.warning('Error while calculating uptime')
self.callback('Error while calculating uptime')
else:
self.callback('My uptime is {}'.format(uptime))
def calculate_uptime(self, birth):
logging.debug('Calculate uptime since {} UTC'.format(birth))
start = self.birth2time(birth)
now = datetime.datetime.utcnow()
print('*** Start: {} :: Now: {}'.format(start, now))
try:
uptime = now - start
print('Uptime: {}'.format(uptime))
return uptime
except Exception as exc:
logging.warning('ERROR: {}'.format(exc))
return False
def birth2time(self, birth):
time_object = None
datum, zeit = birth.split('T')
jahr, monat, tag = datum.split('-')
stunde, minute, sekunde = zeit.split(':')
time_object = datetime.datetime(int(jahr), int(monat), int(tag), \
int(stunde), int(minute), int(sekunde), 0, None)
return time_object

155
plugins/wpsearch.py Normal file
View file

@ -0,0 +1,155 @@
#!/usr/bin/python
# -*- coding: utf-8 -*-
# file: generic.plgin
# date: 23.07.2020
# desc: generic plugin
import json
import common
import logging
import requests
import threading
from manager import Plugin
logging = logging.getLogger()
lock = threading.Lock()
class Plugin(Plugin):
'''
'''
__module = __name__
__command = 'wpsearch'
__description = 'Seach wikipedia'
@staticmethod
def get_module():
return Plugin.__module
@staticmethod
def get_command():
return Plugin.__command
@staticmethod
def get_description():
return Plugin.__description
def __init__(self, callback=None):
self.callback = callback
def help(self):
return ('!wp-search searches in wikipedia for a given search term. '
'Arguments are language shortcut and search pattern. If no '
'language shortcut is given, en is used as default. '
'\nSyntax: !wp-search <lang> <pattern>')
def run(self, stanza):
'''
'''
lang = 'en'
pattern = None
call_msg = 'Call "!help {}"'.format(self.get_command())
no_args_msg = ' '.join(('No search pattern!', call_msg))
muc_nick = common.get_nick_from_stanza(stanza)
arguments = common.get_arguments_from_body(stanza)
logging.debug('Arguments: {}'.format(arguments))
if arguments is False:
logging.warning('No arguments for wp search. Abort.')
self.callback(': '.join((muc_nick, no_args_msg)))
return
else:
if len(arguments) == 1:
pattern = arguments[0]
elif len(arguments) > 1:
logging.warning('Not implemented yet.')
self.callback('Not implemented yet')
return
logging.debug('Start thread for wp search')
self.callback('Search started')
api_thread = ApiThread(self.callback)
api_thread.run(lang, pattern, muc_nick)
class ApiThread(threading.Thread):
'''
The thread who fetched and returns the wp search.
'''
def __init__(self, callback):
threading.Thread.__init__(self)
self.callback = callback
def run(self, lang, pattern, muc_nick):
'''
Starts the thread.
'''
data = self.get_file(lang, pattern)
if data == False:
self.callback('Error while connecting WP')
else:
# TODO: check if its a json needed !
logging.debug(data)
msg = self.string_factory(data)
self.callback(': '.join((muc_nick, msg)))
def get_file(self, lang, pattern):
'''
Grabs the API file, parse it and returns a json ... otherwise none.
param 1: string
param 2: string
param 3: string
returns: json or none
'''
api_url = 'https://{}.wikipedia.org/w/api.php'.format(lang)
api_params = {
'action': 'query',
'prop': 'extracts|info',
'explaintext': '',
'redirects': '',
'exchars': 200,
'continue': '',
'format': 'json',
'titles': pattern,
'inprop': 'url',
'formatversion': 2
}
try:
logging.debug('Try to fetch {}'.format(api_url))
session = requests.Session()
response = session.get(url = api_url, params = api_params)
data = response.json()
logging.debug('{}: successfull fetched'.format(api_url))
except:
logging.debug('{}: failed to fetch'.format(api_url))
return False
return data
def string_factory(self, data):
'''
param 1: json
returns: string
'''
msg = ''
if 'redirects' in data['query'].keys():
for i in data['query']['redirects']:
rfr = i['from']
rto = i['to']
msg = '\n'.join((msg, 'Redirect from {} to {}'.format( \
rfr, rto)))
logging.debug('Message: {}'.format(msg))
pages = data['query']['pages']
for i in pages:
if 'extract' in i.keys():
msg = '\n'.join((msg, 'Summary:', i['extract']))
logging.debug('Message: {}'.format(msg))
else: msg = '\n'.join((msg, 'Nothing found'))
msg = '\n'.join((msg, 'URL:', i['fullurl']))
logging.debug('Message: {}'.format(msg))
return msg

82
runhackbot.py Executable file
View file

@ -0,0 +1,82 @@
#!/usr/bin/python3
# -*- coding: utf-8 -*-
import os
import sys
import logging
from configparser import ConfigParser
from hackbot import HackBot
# Setup logging.
format_string = '%(asctime)s: %(levelname)-8s %(message)s'
logging.basicConfig(level=logging.WARNING, format=format_string)
def setup_config(default_config, config_file, config):
'''
param 1: dictionary
param 1: string
param 2: configparser object
returns: configparser object oder false
'''
config.read_dict(default_config)
if not config.read(config_file):
logging.error('Config file {} not found or not readable'.format\
(config_file))
return False
return config
def display_setup(config):
'''
param 1: configparser object
'''
for section in config.sections():
print('Section: {}'.format(section))
for key, value in config.items(section):
print('\t{}: {}'.format(key, value))
print('\n', end='')
def run_hackbot():
'''
Configure and starts logging and hackbot.
'''
log_levels = ('critical', 'error', 'warning', 'info', 'debug')
config_file = './hackbot.conf'
default_config = {
'settings': {
'loglevel': 'warning',
'plugindir': '.'
},
'jabber': {
'jid': '',
'password': '',
'room': '',
'nick': ''
}
}
config = ConfigParser()
config = setup_config(default_config, config_file, config)
if config is False:
sys.exit()
logger = logging.getLogger()
if not config['settings']['loglevel'] in log_levels:
logging.warning('Invalid loglevel given: {} Use default level: {}'.\
format(config['settings']['loglevel'],
default_config['settings']['loglevel']))
config.set('settings', 'loglevel', \
default_config['settings']['loglevel'])
logger.setLevel(config['settings']['loglevel'].upper())
xmpp = HackBot(config['jabber']['jid'],
config['jabber']['password'],
config['jabber']['room'],
config['jabber']['nick'],
config['settings']['plugindir'])
xmpp.run()
if __name__ == '__main__':
run_hackbot()