""" 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 import markov.lib as markovlib from markov.models import MarkovContext, MarkovState, MarkovTarget 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, ()) 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 = markovlib.get_or_create_target_context(target) markovlib.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 = markovlib.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(markovlib.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(markovlib.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 = markovlib.get_or_create_target_context(target) markovlib.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 = markovlib.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(markovlib.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(markovlib.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')