restored from original master after accident
This commit is contained in:
Frank Becker
parent
9d4c2227ff
commit
0febd01103
|
|
@ -1,28 +0,0 @@
|
|||
[loggers]
|
||||
keys=root,hemppa
|
||||
|
||||
[handlers]
|
||||
keys=console
|
||||
|
||||
[formatters]
|
||||
keys=hemppa
|
||||
|
||||
[logger_root]
|
||||
level=INFO
|
||||
handlers=console
|
||||
|
||||
[logger_hemppa]
|
||||
level=NOTSET
|
||||
handlers=
|
||||
propagate=1
|
||||
qualname=hemppa
|
||||
|
||||
[handler_console]
|
||||
class=StreamHandler
|
||||
level=NOTSET
|
||||
formatter=hemppa
|
||||
args=(sys.stdout,)
|
||||
|
||||
[formatter_hemppa]
|
||||
format=%(levelname)s - %(name)s - %(message)s
|
||||
class=logging.Formatter
|
||||
|
|
@ -1,123 +0,0 @@
|
|||
import re
|
||||
|
||||
from nio import RoomMessageText
|
||||
|
||||
from modules.common.module import BotModule
|
||||
|
||||
|
||||
class MatterMessage:
|
||||
def __init__(self, protocol, user, message):
|
||||
self.protocol = protocol
|
||||
self.username = user
|
||||
self.message = message
|
||||
|
||||
|
||||
class MatterBridgeParser:
|
||||
|
||||
def __init__(self):
|
||||
self.reg = re.compile(r"\[(.*)\] <(.*)> (.*)")
|
||||
self.match = None
|
||||
|
||||
def validate(self, message):
|
||||
self.match = self.reg.match(message)
|
||||
return self.match is not None
|
||||
|
||||
def parse(self, message):
|
||||
|
||||
if self.validate(message):
|
||||
groups = self.match.group(1, 2, 3)
|
||||
return MatterMessage(groups[0], groups[1], groups[2])
|
||||
else:
|
||||
return None
|
||||
|
||||
|
||||
class MatrixModule(BotModule):
|
||||
"""
|
||||
Everytime a matterbridge message is seen, the original message is delegated to the bot for command processing
|
||||
"""
|
||||
|
||||
def __init__(self, name):
|
||||
super().__init__(name)
|
||||
self.known_bots = []
|
||||
self.bot = None
|
||||
self.parser = MatterBridgeParser()
|
||||
|
||||
async def matrix_message(self, bot, room, event):
|
||||
bot.must_be_admin(room, event)
|
||||
|
||||
# todo: add subcommand to add administrators
|
||||
# todo: add subcommand to add known matterbridge bot
|
||||
args = event.body.split()
|
||||
|
||||
if len(args) == 2:
|
||||
if args[1] == 'list-bots':
|
||||
await self.list_bots(room)
|
||||
if len(args) > 2:
|
||||
if args[1] == 'add-bot':
|
||||
self.add_bot(args[2])
|
||||
elif args[1] == 'del-bot':
|
||||
self.del_bot(args[2])
|
||||
|
||||
# todo: needs a mechanism to hook into admin check of the bot
|
||||
|
||||
def help(self):
|
||||
return "parses matterbridge messages and delegate the parsed message to the bot"
|
||||
|
||||
def matrix_start(self, bot):
|
||||
self.bot = bot
|
||||
super().matrix_start(bot)
|
||||
bot.client.add_event_callback(self.dispatch_cb, RoomMessageText)
|
||||
|
||||
def matrix_stop(self, bot):
|
||||
super().matrix_stop(bot)
|
||||
bot.remove_callback(self.dispatch_cb)
|
||||
|
||||
def get_settings(self):
|
||||
data = super().get_settings()
|
||||
data['known_bots'] = self.known_bots
|
||||
return data
|
||||
|
||||
def set_settings(self, data):
|
||||
super().set_settings(data)
|
||||
if data.get('known_bots'):
|
||||
self.known_bots = data['known_bots']
|
||||
|
||||
async def dispatch_cb(self, room, event):
|
||||
|
||||
if event.sender not in self.known_bots:
|
||||
self.logger.debug(f"{event.sender} is not a known bot. skip processing.")
|
||||
return
|
||||
|
||||
# no content at all?
|
||||
if len(event.body) < 1:
|
||||
return
|
||||
|
||||
matter_message = self.parser.parse(event.body)
|
||||
|
||||
if matter_message is None:
|
||||
return
|
||||
|
||||
self.logger.info(
|
||||
f"room: {room.name} protocol: {matter_message.protocol} user: {matter_message.username} - dispatch matterbridge message to bot")
|
||||
|
||||
# dispatch. changing the body of the event triggers message_cb of the bot
|
||||
event.body = matter_message.message
|
||||
|
||||
def add_bot(self, bot_name):
|
||||
self.logger.info("Add bot %s to known bots", bot_name)
|
||||
self.known_bots.append(bot_name)
|
||||
self.bot.save_settings()
|
||||
|
||||
def del_bot(self, bot_name):
|
||||
if bot_name in self.known_bots:
|
||||
self.logger.info("Delete bot %s from list of known bots", bot_name)
|
||||
self.known_bots.remove(bot_name)
|
||||
self.bot.save_settings()
|
||||
else:
|
||||
self.logger.warning("%s not in list of known bots. skip delete.", bot_name)
|
||||
|
||||
async def list_bots(self, room):
|
||||
|
||||
await self.bot.send_text(room, f'Known Matterbridge Bots: {len(self.known_bots)}')
|
||||
for bot_name in self.known_bots:
|
||||
await self.bot.send_text(room, bot_name)
|
||||
|
|
@ -1,38 +0,0 @@
|
|||
from timeit import default_timer as timer
|
||||
from urllib.request import urlopen
|
||||
|
||||
from modules.common.module import BotModule
|
||||
|
||||
|
||||
class MatrixModule(BotModule):
|
||||
async def matrix_message(self, bot, room, event):
|
||||
args = event.body.split()
|
||||
args.pop(0)
|
||||
|
||||
url = args[0]
|
||||
|
||||
# check url
|
||||
if not (url.startswith('http://') or url.startswith('https://')):
|
||||
self.logger.debug("adding trailing https")
|
||||
url = "https://" + url
|
||||
|
||||
self.logger.info("ping: %s", url)
|
||||
|
||||
start = timer()
|
||||
try:
|
||||
data = urlopen(url)
|
||||
length = format(len(data.read()) / 1024, '.3g') # kB
|
||||
retcode = data.getcode()
|
||||
|
||||
except Exception as e:
|
||||
await bot.send_text(room, "Ping failed: " + str(e))
|
||||
self.logger.error(str(e))
|
||||
return False
|
||||
|
||||
end = timer()
|
||||
|
||||
await bot.send_text(room, url + ": OK (" + str(retcode) + ") / " + "Size: " + str(length) +
|
||||
" kB / Time: " + str(format(end - start, '.3g')) + " sec")
|
||||
|
||||
def help(self):
|
||||
return 'check if IP or URL is accessible'
|
||||
|
|
@ -1,32 +0,0 @@
|
|||
import subprocess
|
||||
|
||||
from modules.common.module import BotModule
|
||||
|
||||
|
||||
class MatrixModule(BotModule):
|
||||
async def matrix_message(self, bot, room, event):
|
||||
args = event.body.split()
|
||||
args.pop(0)
|
||||
encoding = "utf-8"
|
||||
|
||||
allowed_args = ['list', 'add', 'del', 'done', 'undo', 'calc']
|
||||
|
||||
# wrap task
|
||||
if not args:
|
||||
args = ['list']
|
||||
|
||||
if args[0] not in allowed_args:
|
||||
await bot.send_text(room, "command not allowed")
|
||||
return ()
|
||||
|
||||
result = subprocess.check_output(
|
||||
["task",
|
||||
"rc.confirmation:no",
|
||||
"rc.verbose:list",
|
||||
"rc.bulk:0",
|
||||
"rc.recurrence.confirmation:yes"]
|
||||
+ args, stderr=subprocess.DEVNULL)
|
||||
await bot.send_text(room, result.decode(encoding))
|
||||
|
||||
def help(self):
|
||||
return ('taskwarrior')
|
||||
|
|
@ -165,7 +165,3 @@ class MatrixModule(BotModule):
|
|||
|
||||
def help(self):
|
||||
return "If I see a url in a message I will try to get the title from the page and spit it out"
|
||||
|
||||
def dump(self, obj):
|
||||
for attr in dir(obj):
|
||||
self.logger.info("obj.%s = %r", attr, getattr(obj, attr))
|
||||
|
|
@ -1,18 +0,0 @@
|
|||
import subprocess
|
||||
|
||||
from modules.common.module import BotModule
|
||||
|
||||
|
||||
class MatrixModule(BotModule):
|
||||
async def matrix_message(self, bot, room, event):
|
||||
args = event.body.split()
|
||||
args.pop(0)
|
||||
encoding = "utf-8"
|
||||
|
||||
# get weather from ansiweather
|
||||
result = subprocess.check_output(["ansiweather", "-a false", "-l", ' '.join(args)])
|
||||
|
||||
await bot.send_text(room, result.decode(encoding))
|
||||
|
||||
def help(self):
|
||||
return ('How\'s the weather?')
|
||||
|
|
@ -1,29 +0,0 @@
|
|||
import unittest
|
||||
from modules.devlugdispatch import MatterBridgeParser
|
||||
|
||||
|
||||
class DevlugDispatcherTest(unittest.TestCase):
|
||||
|
||||
def test_recognizes_matterbridge_message(self):
|
||||
"""Test that a message was send by a matterbridge bot"""
|
||||
message = "[irc] <ancho> say something"
|
||||
parser = MatterBridgeParser()
|
||||
|
||||
self.assertEqual(parser.validate(message), True)
|
||||
|
||||
def test_parse_returns_None_for_none_matter_bridge_messages(self):
|
||||
"""Test that a normal message gets parsed to None"""
|
||||
message = "a normal message"
|
||||
parser = MatterBridgeParser()
|
||||
|
||||
self.assertEqual(parser.parse(message), None)
|
||||
|
||||
def test_parse_returns_protocol_username_and_message(self):
|
||||
|
||||
message = "[irc] <ancho> say something"
|
||||
parser = MatterBridgeParser()
|
||||
|
||||
matter_msg = parser.parse(message)
|
||||
self.assertEqual(matter_msg.protocol, "irc")
|
||||
self.assertEqual(matter_msg.username, "ancho")
|
||||
self.assertEqual(matter_msg.message, "say something")
|
||||
Loading…
Reference in New Issue