johnno/lolbot
johnno/lolbot
1
0
Fork 0
Code Issues Pull requests Releases Wiki Activity

Refactor, add very basic support for MoxQuizz questions.

Browse source 
- Add rudimentary classes for reading MoxQuizz questions.
 - Add a very simple ask command. No scores or quiz game-play yet.
 - Refactor SQL models out into a separate file.
 - Make the code more Python 3 friendly.
 - PEP8 and pyflakes.
This commit is contained in:
Jonathan Harker 2014-11-25 01:13:26 +13:00
parent 4f546f18a3
commit 3100081986
3 changed files with 314 additions and 100 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

161
lolbot.py
View file

@ -9,6 +9,8 @@ Useful bot for folks stuck behind censor walls at work
Logs a channel and collects URLs for later.
"""
from __future__ import print_function # unicode_literals
try:
import sys
import os
@ -21,12 +23,12 @@ try:
from twisted.internet import protocol
from twisted.internet import reactor
from datetime import datetime
from mechanize import Browser
from sqlalchemy import MetaData, Table, Column, String, Text, Integer, DateTime, create_engine
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from sqlalchemy.ext.declarative import declarative_base
from pymoxquizz import QuestionBank, Question
from models import Url, Log, Model
except ImportError:
print "Some modules missing: Lolbot relies on Twisted IRC, Mechanize and SQLAlchemy.\n"
print("Some modules missing: Lolbot relies on Twisted IRC, Mechanize and SQLAlchemy.\n")
sys.exit
# Exclamations - wrong input
@ -47,70 +49,6 @@ ponderings = [
"No it's a week night 8pm is past my bedtime.",
]
SqlBase = declarative_base()
class Log(SqlBase):
"""
This class represents an event in the log table and inherits from a SQLAlchemy
convenience ORM class.
"""
__tablename__ = "log"
id = Column(Integer, primary_key=True)
timestamp = Column(DateTime)
nickname = Column(String(20))
text = Column(Text)
def __init__(self, nickname, text, timestamp=None):
"""
Creates an event log for the IRC logger.
"""
if timestamp is None:
timestamp = datetime.now()
self.timestamp = timestamp
self.nickname = nickname
self.text = text
def __repr__(self):
return "(%s) %s: %s" % (self.timestamp.strftime("%Y-%m-%d %H:%M:%S"), self.nickname, self.text)
class Url(SqlBase):
"""
This class represents a saved URL and inherits from a SQLAlchemy convenience
ORM class.
"""
__tablename__ = "url"
id = Column(Integer, primary_key=True)
timestamp = Column(DateTime)
nickname = Column(String(20))
url = Column(String(200), unique=True)
title = Column(Text)
def __init__(self, nickname, url, title=None, timestamp=None):
if timestamp is None:
timestamp = datetime.now()
self.timestamp = timestamp
self.nickname = nickname
self.url = url
self.title = title
# populate the title from the URL if not given.
if title is None:
try:
br = Browser()
br.open(self.url)
self.title = br.title()
except Exception as ex:
self.title = ''
def __repr__(self):
if not self.title:
return "%s: %s" % (self.nickname, self.url)
else:
return "%s: %s - %s" % (self.nickname, self.url, self.title)
class LolBot(irc.IRCClient):
"""
@ -123,10 +61,15 @@ class LolBot(irc.IRCClient):
return connection
def created(self, when):
# load some MoxQuizz questions
self.qb = QuestionBank('/home/johnno/questions.doctorlard.en').questions
random.shuffle(self.qb)
self.question = None
# connect to the database
self.dbengine = create_engine('sqlite+pysqlite://', creator=self._get_connection)
SqlBase.metadata.bind = self.dbengine
SqlBase.metadata.create_all()
Model.metadata.bind = self.dbengine
Model.metadata.create_all()
self.get_session = sessionmaker(bind=self.dbengine)
self.helptext = "Keeps a list of URLs. Commands: list [n|x-y] - prints the last 10 URLs (or n URLs, or x through y); clear - clears the list; lol - say something funny; <url> - adds the URL to the list; help - this message."
@ -144,10 +87,10 @@ class LolBot(irc.IRCClient):
db.commit()
else:
theurl = db.query(Url).filter(Url.url == url).one()
print theurl
print(theurl)
title = theurl.title
except Exception, ex:
print "Exception caught saving URL: %s" % ex
except Exception as ex:
print("Exception caught saving URL: %s" % ex)
return title
def log_event(self, nick, text):
@ -156,9 +99,9 @@ class LolBot(irc.IRCClient):
db = self.get_session()
db.add(entry)
db.commit()
print entry
except Exception, ex:
print "Exception caught logging event: %s" % ex
print(entry)
except Exception as ex:
print("Exception caught logging event: %s" % ex)
def _get_nickname(self):
return self.factory.nickname
@ -174,13 +117,14 @@ class LolBot(irc.IRCClient):
def signedOn(self):
self.join(self.channel)
print "Signed on as %s." % (self.nickname,)
print("Signed on as %s." % (self.nickname,))
def joined(self, channel):
print "Joined %s." % (channel,)
print("Joined %s." % (channel,))
def privmsg(self, user, channel, msg):
user = user.split('!')[0]
if channel == self.nickname:
# Private /msg from a user
self.do_command(msg, user)
@ -188,6 +132,15 @@ class LolBot(irc.IRCClient):
# log it
self.log_event(user, msg)
# deal with MoxQuizz question
answered = False
if isinstance(self.question, Question) and self.question.answer.lower() in msg.lower():
answered = True
print("%s +1" % user)
self.reply('Correct! %s scores one point.' % user)
if answered:
self.question = None
args = string.split(msg, ":", 1)
if len(args) > 1 and args[0] == self.nickname:
self.do_command(string.strip(args[1]))
@ -197,7 +150,7 @@ class LolBot(irc.IRCClient):
for w in words:
if w.startswith('http://') or w.startswith('https://'):
title = self.save_url(user, w)
if title == False:
if title is False:
self.say_public("Sorry, I'm useless at UTF-8.")
else:
self.say_public("URL added. %s" % title)
@ -236,6 +189,10 @@ class LolBot(irc.IRCClient):
if cmd == 'help':
self.reply(self.helptext, target)
elif cmd == 'ask':
self.question = random.choice(self.qb)
self.reply(str(self.question.question))
elif cmd == 'lol':
self.reply(self.ponder(), target)
@ -259,14 +216,14 @@ class LolBot(irc.IRCClient):
y = abs(int(y))
if y < x:
x, y = y, x
except ValueError, ex:
except ValueError as ex:
self.reply("Give me a number or a range of numbers, e.g. list 5 or list 11-20", target)
raise ex
rows = db.query(Url).order_by(Url.timestamp.desc())[x-1:y]
else:
try:
n = abs(int(n))
except ValueError, ex:
except ValueError as ex:
self.reply("Give me a number or a range of numbers, e.g. list 5 or list 11-20", target)
raise ex
rows = db.query(Url).order_by(Url.timestamp.desc())[:n]
@ -277,8 +234,8 @@ class LolBot(irc.IRCClient):
time.sleep(1)
elif cmd.startswith('http:') or cmd.startswith('https:'):
title = self.save_url(from_private, cmd)
if title == False:
title = self.save_url(target, cmd)
if title is False:
self.say_public("Sorry, I'm useless at UTF-8.")
else:
self.reply('URL added. %s' % title, target)
@ -286,12 +243,13 @@ class LolBot(irc.IRCClient):
else:
self.reply(self.exclaim(), target)
except Exception, ex:
print "Exception caught processing command: %s" % ex
print " command was '%s' from %s" % (cmd, target)
except Exception as ex:
print("Exception caught processing command: %s" % ex)
print(" command was '%s' from %s" % (cmd, target))
self.reply("Sorry, I didn't understand: %s" % cmd, target)
self.reply(self.helptext, target)
class LolBotFactory(protocol.ClientFactory):
protocol = LolBot
@ -303,28 +261,30 @@ class LolBotFactory(protocol.ClientFactory):
self.channel = self.config['irc.channel']
def clientConnectionLost(self, connector, reason):
print "Lost connection (%s), reconnecting." % (reason,)
print("Lost connection (%s), reconnecting." % (reason,))
connector.connect()
def clientConnectionFailed(self, connector, reason):
print "Could not connect: %s" % (reason,)
print("Could not connect: %s" % (reason,))
def get_options():
try:
(options, args) = getopt.getopt(sys.argv[1:], "hc:", ["help", "config=", ])
except getopt.GetoptError, err:
print str(err)
except getopt.GetoptError as err:
print(str(err))
usage()
sys.exit(2)
config_path = ""
for option,value in options:
for option, value in options:
if option in ("-h", "--help"):
usage()
sys.exit(2)
if option in ("-c", "--config"):
config_path = value
return { 'config_path': config_path }
return {'config_path': config_path}
def get_config(config_path):
"""
@ -342,7 +302,7 @@ def get_config(config_path):
if not config_path:
config_path = os.path.join(os.path.dirname(os.path.abspath(__file__)), "lolbot.conf")
if not os.path.exists(config_path):
print "Error: configuration file not found. By default lolbot will look for a lolbot.conf file in the same directory as the lolbot script, but you can override this by specifying a path on the command line with the --config option."
print("Error: configuration file not found. By default lolbot will look for a lolbot.conf file in the same directory as the lolbot script, but you can override this by specifying a path on the command line with the --config option.")
usage()
sys.exit(1)
@ -364,7 +324,7 @@ def get_config(config_path):
# validate IRC host
if "irc.server" not in config.keys():
print "Error: the IRC server was not specified. Use --help for more information."
print("Error: the IRC server was not specified. Use --help for more information.")
sys.exit(1)
# validate IRC port
@ -373,12 +333,12 @@ def get_config(config_path):
try:
config["irc.port"] = int(config["irc.port"])
except ValueError:
print "Error: the IRC port must be an integer. If not specified, lolbot will use the default IRC port value 6667. Use --help for more information."
print("Error: the IRC port must be an integer. If not specified, lolbot will use the default IRC port value 6667. Use --help for more information.")
sys.exit(1)
# validate IRC channel
if "irc.channel" not in config.keys() or not config["irc.channel"].startswith("#"):
print "Error: the IRC channel is not specified or incorrect. It must begin with a # - e.g. #mychatchannel. Use --help for more information."
print("Error: the IRC channel is not specified or incorrect. It must begin with a # - e.g. #mychatchannel. Use --help for more information.")
sys.exit(1)
# validate bot nickname
@ -387,8 +347,9 @@ def get_config(config_path):
return config
def usage():
print """Run a lolbot.
print("""Run a lolbot.
-h, --help
This message.
@ -416,7 +377,8 @@ Configuration:
by SQL Alchemy. If not specified, lolbot will attempt to use a
SQLite database called lolbot.db in the same directory as the
script.
"""
""")
if __name__ == "__main__":
args = get_options()
@ -425,5 +387,4 @@ if __name__ == "__main__":
reactor.connectTCP(config['irc.server'], config['irc.port'], LolBotFactory(args['config_path']))
reactor.run()
except KeyboardInterrupt:
print "Shutting down."
print("Shutting down.")

72
models.py Normal file
View file

@ -0,0 +1,72 @@
from mechanize import Browser
from datetime import datetime
from sqlalchemy import (Column, String, Text, Integer, DateTime)
from sqlalchemy.ext.declarative import (declarative_base)
Model = declarative_base()
class Log(Model):
"""
This class represents an event in the log table and inherits from a SQLAlchemy
convenience ORM class.
"""
__tablename__ = "log"
id = Column(Integer, primary_key=True)
timestamp = Column(DateTime)
nickname = Column(String(20))
text = Column(Text)
def __init__(self, nickname, text, timestamp=None):
"""
Creates an event log for the IRC logger.
"""
if timestamp is None:
timestamp = datetime.now()
self.timestamp = timestamp
self.nickname = nickname
self.text = text
def __repr__(self):
return "(%s) %s: %s" % (self.timestamp.strftime("%Y-%m-%d %H:%M:%S"), self.nickname, self.text)
class Url(Model):
"""
This class represents a saved URL and inherits from a SQLAlchemy convenience
ORM class.
"""
__tablename__ = "url"
id = Column(Integer, primary_key=True)
timestamp = Column(DateTime)
nickname = Column(String(20))
url = Column(String(200), unique=True)
title = Column(Text)
def __init__(self, nickname, url, title=None, timestamp=None):
if timestamp is None:
timestamp = datetime.now()
self.timestamp = timestamp
self.nickname = nickname
self.url = url
self.title = title
# populate the title from the URL if not given.
if title is None:
try:
br = Browser()
br.open(self.url)
self.title = br.title()
except Exception:
self.title = ''
def __repr__(self):
if not self.title:
return "%s: %s" % (self.nickname, self.url)
else:
return "%s: %s - %s" % (self.nickname, self.url, self.title)

181
pymoxquizz.py Normal file
View file

@ -0,0 +1,181 @@
#!/usr/bin/env python
## -*- coding: utf-8 -*-
from __future__ import unicode_literals, print_function
from io import open
class Question:
"""
Represents one MoxQuizz question.
"""
category = None
question = None
answer = None
regexp = None
author = None
level = None
comment = None
score = 0
tip = list()
tipcycle = 0
TRIVIAL = 1
EASY = 2
NORMAL = 3
HARD = 4
EXTREME = 5
LEVELS = (TRIVIAL, EASY, NORMAL, HARD, EXTREME)
def __init__(self, attributes_dict):
self.parse(attributes_dict)
def parse(self, attributes_dict):
"""
Populate fields from a dictionary of attributes (from a question bank).
"""
## Valid keys:
# ----------
# Category? (should always be on top!)
# Question (should always stand after Category)
# Answer (will be matched if no regexp is provided)
# Regexp? (use UNIX-style expressions)
# Author? (the brain behind this question)
# Level? [baby|easy|normal|hard|extreme] (difficulty)
# Comment? (comment line)
# Score? [#] (credits for answering this question)
# Tip* (provide one or more hints)
# TipCycle? [#] (Specify number of generated tips)
if 'Question' in attributes_dict.keys():
self.question = attributes_dict['Question']
else:
raise Exception("Cannot instantiate Question: 'Question' attribute required.")
if 'Category' in attributes_dict.keys():
self.category = attributes_dict['Category']
if 'Answer' in attributes_dict.keys():
self.answer = attributes_dict['Answer']
else:
raise Exception("Cannot instantiate Question: 'Answer' attribute required.")
if 'Regexp' in attributes_dict.keys():
self.regexp = attributes_dict['Regexp']
if 'Author' in attributes_dict.keys():
self.category = attributes_dict['Author']
if 'Level' in attributes_dict.keys() and attributes_dict['Level'] in self.LEVELS:
self.level = attributes_dict['level']
if 'Comment' in attributes_dict.keys():
self.comment = attributes_dict['Comment']
if 'Score' in attributes_dict.keys():
self.score = attributes_dict['Score']
if 'Tip' in attributes_dict.keys():
self.tip = attributes_dict['Tip']
if 'Tipcycle' in attributes_dict.keys():
self.tipcycle = attributes_dict['Tipcycle']
class QuestionBank:
"""
Represents a MoxQuizz question bank.
"""
filename = ''
questions = list()
# Case sensitive, to remain backwards-compatible with MoxQuizz.
KEYS = ('Answer',
'Author',
'Category',
'Comment',
'Level',
'Question',
'Regexp',
'Score',
'Tip',
'Tipcycle',
)
def __init__(self, filename):
"""
Construct a question bank from a file.
"""
self.filename = filename
self.questions = self.parse(filename)
def parse(self, filename):
"""
Read a Moxquizz question bank file into a list.
"""
questions = list()
with open(filename) as f:
key = ''
i = 0
# new question
q = dict()
q['Tip'] = list()
for line in f:
line = line.strip()
i += 1
# Ignore comments.
if line.startswith('#'):
continue
# A blank line starts a new question.
if line == '':
# Store the previous question, if valid.
if 'Question' in q.keys() and 'Answer' in q.keys():
question = Question(q)
questions.append(question)
# Start a new question.
q = dict()
q['Tip'] = list()
continue
# Fetch the next parameter.
try:
(key, value) = line.split(':', 1)
except ValueError:
print("Unexpected weirdness in MoxQuizz questionbank '%s', line %s." % (self.filename, i))
continue
# break # TODO: is it appropriate to bail on broken bank files?
# Ignore bad parameters.
if key not in self.KEYS:
print("Unexpected key '%s' in MoxQuizz questionbank '%s', line %s." % (key, self.filename, i))
continue
# Enumerate the Tips.
if key == 'Tip':
q['Tip'].append(value.strip())
else:
q[key] = value.strip()
return questions
# A crappy test.
if __name__ == '__main__':
qb = QuestionBank('questions.doctorlard.en')
for q in qb.questions:
print(q.question)
a = unicode(raw_input('A: '), 'utf8')
#a = input('A: ') # Python 3
if a.lower() == q.answer.lower():
print("Correct!")
else:
print("Incorrect - the answer is '%s'" % q.answer)