""" Markov - Chatterbot via Markov chains for IRC Copyright (C) 2010 Brian S. Stephan This program is free software: you can redistribute it and/or modify it under the terms of the GNU General Public License as published by the Free Software Foundation, either version 3 of the License, or (at your option) any later version. This program is distributed in the hope that it will be useful, but WITHOUT ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for more details. You should have received a copy of the GNU General Public License along with this program. If not, see . """ from datetime import datetime import random import re import thread import time from dateutil.relativedelta import relativedelta from markov.models import MarkovContext, MarkovState, MarkovTarget from markov.views import _generate_line, _learn_line from extlib import irclib from Module import Module class Markov(Module): """Create a chatterbot very similar to a MegaHAL, but simpler and implemented in pure Python. Proof of concept code from Ape. Ape wrote: based on this: http://uswaretech.com/blog/2009/06/pseudo-random-text-markov-chains-python/ and this: http://code.activestate.com/recipes/194364-the-markov-chain-algorithm/ """ def __init__(self, irc, config): """Create the Markov chainer, and learn text from a file if available. """ # set up regexes, for replying to specific stuff learnpattern = '^!markov\s+learn\s+(.*)$' replypattern = '^!markov\s+reply(\s+min=(\d+))?(\s+max=(\d+))?(\s+(.*)$|$)' self.learnre = re.compile(learnpattern) self.replyre = re.compile(replypattern) self.shut_up = False self.lines_seen = [] Module.__init__(self, irc, config) self.next_shut_up_check = 0 self.next_chatter_check = 0 thread.start_new_thread(self.thread_do, ()) # TODO: bring this back somehow #irc.xmlrpc_register_function(self._generate_line, # "markov_generate_line") def register_handlers(self): """Handle pubmsg/privmsg, to learn and/or reply to IRC events.""" self.irc.server.add_global_handler('pubmsg', self.on_pub_or_privmsg, self.priority()) self.irc.server.add_global_handler('privmsg', self.on_pub_or_privmsg, self.priority()) self.irc.server.add_global_handler('pubmsg', self.learn_from_irc_event) self.irc.server.add_global_handler('privmsg', self.learn_from_irc_event) def unregister_handlers(self): self.irc.server.remove_global_handler('pubmsg', self.on_pub_or_privmsg) self.irc.server.remove_global_handler('privmsg', self.on_pub_or_privmsg) self.irc.server.remove_global_handler('pubmsg', self.learn_from_irc_event) self.irc.server.remove_global_handler('privmsg', self.learn_from_irc_event) def learn_from_irc_event(self, connection, event): """Learn from IRC events.""" what = ''.join(event.arguments()[0]) my_nick = connection.get_nickname() what = re.sub('^' + my_nick + '[:,]\s+', '', what) target = event.target() nick = irclib.nm_to_n(event.source()) if not irclib.is_channel(target): target = nick self.lines_seen.append((nick, datetime.now())) # don't learn from commands if self.learnre.search(what) or self.replyre.search(what): return if not event._recursing: context = _get_or_create_target_context(target) _learn_line(what, context) def do(self, connection, event, nick, userhost, what, admin_unlocked): """Handle commands and inputs.""" target = event.target() if self.learnre.search(what): return self.irc.reply(event, self.markov_learn(event, nick, userhost, what, admin_unlocked)) elif self.replyre.search(what) and not self.shut_up: return self.irc.reply(event, self.markov_reply(event, nick, userhost, what, admin_unlocked)) if not self.shut_up: # not a command, so see if i'm being mentioned if re.search(connection.get_nickname(), what, re.IGNORECASE) is not None: context = _get_or_create_target_context(target) addressed_pattern = '^' + connection.get_nickname() + '[:,]\s+(.*)' addressed_re = re.compile(addressed_pattern) if addressed_re.match(what): # i was addressed directly, so respond, addressing # the speaker topics = [x for x in addressed_re.match(what).group(1).split(' ') if len(x) >= 3] self.lines_seen.append(('.self.said.', datetime.now())) return self.irc.reply(event, u"{0:s}: {1:s}".format(nick, u" ".join(_generate_line(context, topics=topics, max_sentences=1)))) else: # i wasn't addressed directly, so just respond topics = [x for x in what.split(' ') if len(x) >= 3] self.lines_seen.append(('.self.said.', datetime.now())) return self.irc.reply(event, u"{0:s}".format(u" ".join(_generate_line(context, topics=topics, max_sentences=1)))) def markov_learn(self, event, nick, userhost, what, admin_unlocked): """Learn one line, as provided to the command.""" target = event.target() if not irclib.is_channel(target): target = nick match = self.learnre.search(what) if match: line = match.group(1) context = _get_or_create_target_context(target) _learn_line(line, context) # return what was learned, for weird chaining purposes return line def markov_reply(self, event, nick, userhost, what, admin_unlocked): """Generate a reply to one line, without learning it.""" target = event.target() if not irclib.is_channel(target): target = nick match = self.replyre.search(what) if match: min_size = 15 max_size = 30 context = _get_or_create_target_context(target) if match.group(2): min_size = int(match.group(2)) if match.group(4): max_size = int(match.group(4)) if match.group(5) != '': line = match.group(6) topics = [x for x in line.split(' ') if len(x) >= 3] self.lines_seen.append(('.self.said.', datetime.now())) return u" ".join(_generate_line(context, topics=topics, min_words=min_size, max_words=max_size, max_sentences=1)) else: self.lines_seen.append(('.self.said.', datetime.now())) return u" ".join(_generate_line(context, min_words=min_size, max_words=max_size, max_sentences=1)) def thread_do(self): """Do various things.""" while not self.is_shutdown: self._do_shut_up_checks() self._do_random_chatter_check() time.sleep(1) def _do_random_chatter_check(self): """Randomly say something to a channel.""" # TODO: make this do stuff again return def _do_shut_up_checks(self): """Check to see if we've been talking too much, and shut up if so.""" if self.next_shut_up_check < time.time(): self.shut_up = False self.next_shut_up_check = time.time() + 30 last_30_sec_lines = [] for (nick, then) in self.lines_seen: rdelta = relativedelta(datetime.now(), then) if (rdelta.years == 0 and rdelta.months == 0 and rdelta.days == 0 and rdelta.hours == 0 and rdelta.minutes == 0 and rdelta.seconds <= 29): last_30_sec_lines.append((nick, then)) if len(last_30_sec_lines) >= 8: lines_i_said = len(filter(lambda (a, b): a == '.self.said.', last_30_sec_lines)) if lines_i_said >= 8: self.shut_up = True targets = self._get_chatter_targets() for t in targets: self.sendmsg(t['target'], 'shutting up for 30 seconds due to last 30 seconds of activity') def _get_or_create_target_context(target_name): """Return the context for a provided nick/channel, creating missing ones.""" # find the stuff, or create it try: target = MarkovTarget.objects.get(name=target_name) return target.context except MarkovContext.DoesNotExist: # make a context context = MarkovContext() context.name = target_name context.save() target.context = context target.save() return target.context except MarkovTarget.DoesNotExist: # first we need to make a context for this context = MarkovContext() context.name = target_name context.save() target = MarkovTarget() target.name = target_name target.context = context target.save() return target.context # vi:tabstop=4:expandtab:autoindent