a XMPP logging bot, saving its log as html page
Du kannst nicht mehr als 25 Themen auswählen Themen müssen entweder mit einem Buchstaben oder einer Ziffer beginnen. Sie können Bindestriche („-“) enthalten und bis zu 35 Zeichen lang sein.

218 Zeilen
7.2 KiB

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# To run this bot:
# $ python3 logbot.py
# The output folder of this bot currently is: /var/www/logs/digital-autonomy
import logging
from getpass import getpass
from argparse import ArgumentParser
import slixmpp
import ssl, os, requests, urllib
class MUCBot(slixmpp.ClientXMPP):
"""
A simple Slixmpp bot that will save images
and messages that are marked with @bot to a folder.
"""
def __init__(self, jid, password, room, nick, output):
slixmpp.ClientXMPP.__init__(self, jid, password)
self.room = room
self.nick = nick
self.output = output
# The session_start event will be triggered when
# the bot establishes its connection with the server
# and the XML logs are ready for use. We want to
# listen for this event so that we we can initialize
# our roster.
self.add_event_handler("session_start", self.start)
# The groupchat_message event is triggered whenever a message
# stanza is received from any chat room. If you also also
# register a handler for the 'message' event, MUC messages
# will be processed by both handlers.
self.add_event_handler("groupchat_message", self.muc_message)
def start(self, event):
self.get_roster()
self.send_presence()
# https://xmpp.org/extensions/xep-0045.html
self.plugin['xep_0045'].join_muc(self.room,
self.nick,
# If a room password is needed, use:
# password=the_room_password,
wait=True)
# Send a message to the room
self.send_message(mto=self.room, mbody='Hello! Logbot here. I\'m here to log all the images that are send to this group. You can also log text messages, by including @bot in your message. Happy logging! PS. you can access the logs at https://vvvvvvaria.org/logs/', mtype='groupchat')
def muc_message(self, msg):
# Some inspection commands
#print('Message: {}'.format(msg))
# Always check that a message is not the bot itself, otherwise you will create an infinite loop responding to your own messages.
if msg['mucnick'] != self.nick:
# Check if output folder exists
if not os.path.exists(self.output):
os.mkdir(self.output)
# Check if an OOB URL is included in the stanza (which is how an image is sent)
# (OOB object - https://xmpp.org/extensions/xep-0066.html#x-oob)
if len(msg['oob']['url']) > 0:
# Send a reply
self.send_message(mto=self.room,
mbody="Super, our log is growing. Your image is added!",
mtype='groupchat')
# Save the image to the output folder
url = msg['oob']['url'] # grep the url in the message
filename = os.path.basename(url) # grep the filename in the url
output_path = os.path.join(self.output, filename)
u = urllib.request.urlopen(url) # read the image data
f = open(output_path, 'wb') # open the output file
f.write(u.read()) # write image to file
f.close() # close the output file
# Add the image to the log
img = '<img class="image" src="{}">'.format(filename)
log = 'log.html'
log_path = os.path.join(self.output, log)
f = open(log_path, 'a+')
f.write(img+'\n')
f.close()
# Include messages in the log (only when '@bot' is used in the message)
if '@bot' in msg['body']:
# reply from the bot
self.send_message(mto=self.room,
mbody="Noted! And added to the log. Thanks {}!".format(msg['mucnick']),
mtype='groupchat')
# Add the message to the log!
message = '<p class="message">{}</p>'.format(msg['body'].replace('@bot',''))
log = 'log.html'
log_path = os.path.join(self.output, log)
f = open(log_path, 'a+')
f.write(message+'\n')
f.close()
if '/book' in msg['body']: # Check if this is a book ...
self.send_message(mto=self.room,
mbody="Oh a book, that's cool! Thanks {}!".format(msg['mucnick']),
mtype='groupchat')
# Start of book feature
from bs4 import BeautifulSoup
import re
book = msg['body'].replace('@bot', '').replace('/book', '')
book = re.sub(' +', ' ', book) # remove double spaces
book = book.lstrip().rstrip() # remove spaces at the beginning and at the end
book = book.replace(' ', '+').lower() # turn space into + and lowercase
page_link = 'https://www.worldcat.org/search?q={}&qt=results_page'.format(book)
page_response = requests.get(page_link, timeout=5)
page_content = BeautifulSoup(page_response.content, "html.parser")
try:
book_title = page_content.findAll("div", {"class": "name"})[0].text
book_author = page_content.findAll("div", {"class": "author"})[0].text
book_publisher = page_content.findAll("div", {"class": "publisher"})[0].text
response = '<b>BOOK</b>: ' + book_title + ' ' + book_author + ' ' + book_publisher
book_found = True
except IndexError:
book_found = False
if book_found:
# Add message to log
message = '<b>BOOK</b>: ' + book_title + ' ' + book_author + ' ' + book_publisher
log = 'log.html'
log_path = os.path.join(self.output, log)
f = open(log_path, 'a+')
f.write(message+'\n')
f.close()
self.send_message(mto=self.room, mbody='Hope this was the book you were looking for: ' + book_title + ' ' + book_author + ' ' + book_publisher, mtype='groupchat')
else:
self.send_message(mto=self.room, mbody='Sorry, no book found!', mtype='groupchat')
if __name__ == '__main__':
# Setup the command line arguments.
parser = ArgumentParser()
# output verbosity options.
parser.add_argument("-q", "--quiet", help="set logging to ERROR",
action="store_const", dest="loglevel",
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)
# JID and password options.
parser.add_argument("-j", "--jid", dest="jid",
help="JID to use")
parser.add_argument("-p", "--password", dest="password",
help="password to use")
parser.add_argument("-r", "--room", dest="room",
help="MUC room to join")
parser.add_argument("-n", "--nick", dest="nick",
help="MUC nickname")
# output folder for images
parser.add_argument("-o", "--output", dest="output",
help="output folder, this is where the files are stored",
type=str)
args = parser.parse_args()
# Setup logging.
logging.basicConfig(level=args.loglevel,
format='%(levelname)-8s %(message)s')
if args.jid is None:
args.jid = input("XMPP address: ")
if args.password is None:
args.password = getpass("Password: ")
if args.room is None:
args.room = input("MUC room: ")
if args.nick is None:
args.nick = input("MUC nickname: ")
if args.output is None:
args.output = input("Output folder: ")
# Setup the MUCBot and register plugins. Note that while plugins may
# have interdependencies, the order in which you register them does
# not matter.
xmpp = MUCBot(args.jid, args.password, args.room, args.nick, args.output)
xmpp.register_plugin('xep_0030') # Service Discovery
xmpp.register_plugin('xep_0045') # Multi-User Chat
xmpp.register_plugin('xep_0199') # XMPP Ping
xmpp.register_plugin('xep_0066') # Process URI's (files, images)
# Connect to the XMPP server and start processing XMPP stanzas.
xmpp.connect()
xmpp.process()