#!/usr/bin/python # # PyBorg IRC module # # Copyright (c) 2000, 2006 Tom Morton, Sebastien Dailly # # # 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 2 # 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, write to the Free Software # Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA 02111-1307, USA. # import sys try: from ircbot import * from irclib import * except: print "ERROR !!!!\nircbot.py and irclib.py not found, please install them\n( http://python-irclib.sourceforge.net/ )" sys.exit(1) #overide irclib function def my_remove_connection(self, connection): if self.fn_to_remove_socket: self.fn_to_remove_socket(connection._get_socket()) IRC._remove_connection = my_remove_connection import os import pyborg import cfgfile from random import * import time import traceback import thread def get_time(): """ Return time as a nice yummy string """ return time.strftime("%H:%M:%S", time.localtime(time.time())) def user_match_masks(user, masks): for mask in masks: if mask_matches(user, mask): return True return False def adjust_eventtype(e): et = e.eventtype() if et == "ctcp": if is_channel(e.target()): et = "pubmsg" else: et = "privmsg" return et wordrestr = r"(?:^|(?<=\W))%s(?=\W|$)" nicksre = re.compile(wordrestr % "#nick's") nickre = re.compile(wordrestr % "#nick") restripcolor = re.compile(r"\x03[0-9]{1,2}(?:,[0-9]{1,2})?") def word_find(haystack, word, casesens=False): if not casesens: flags = re.I else: flags = 0 result = re.compile(wordrestr % re.escape(word), flags).search(haystack) if result == None: return -1 else: return result.start() def word_replace(haystack, word, repl, casesens=False): if not casesens: flags = re.I else: flags = 0 result = re.compile(wordrestr % re.escape(word), flags).sub(repl, haystack) return result class ModIRC(SingleServerIRCBot): """ Module to interface IRC input and output with the PyBorg learn and reply modules. """ # The bot recieves a standard message on join. The standard part # message is only used if the user doesn't have a part message. join_msg = "%s"# is here" part_msg = "%s"# has left" # For security the owner's host mask is stored # DON'T CHANGE THIS owner_mask = [] # laid out by channel > user # channel is a list containing: # [0]: dict of users # [1]: number of lines seen # user is a list containing # [0]: last line seen # [1]: number of lines seen # [2]: time seen activity = {} # Command list for this module commandlist = "IRC Module Commands:\n!chans, !ignore, \ !join, !nick, !part, !quit, !quitmsg, !reply2ignored, !replyrate, !shutup, \ !stealth, !unignore, !wakeup, !talk, !owner" # Detailed command description dictionary commanddict = { "shutup": "Owner command. Usage: !shutup\nStop the bot talking", "wakeup": "Owner command. Usage: !wakeup\nAllow the bot to talk", "join": "Owner command. Usage: !join #chan1 [#chan2 [...]]\nJoin one or more channels", "part": "Owner command. Usage: !part #chan1 [#chan2 [...]]\nLeave one or more channels", "chans": "Owner command. Usage: !chans\nList channels currently on", "nick": "Owner command. Usage: !nick nickname\nChange nickname", "ignore": "Owner command. Usage: !ignore [nick1 [nick2 [...]]]\nIgnore one or more nicknames. Without arguments it lists ignored nicknames", "unignore": "Owner command. Usage: !unignore nick1 [nick2 [...]]\nUnignores one or more nicknames", "replyrate": "Owner command. Usage: !replyrate [rate%]\nSet rate of bot replies to rate%. Without arguments (not an owner-only command) shows the current reply rate", "reply2ignored": "Owner command. Usage: !reply2ignored [on|off]\nAllow/disallow replying to ignored users. Without arguments shows the current setting", "stealth": "Owner command. Usage: !stealth [on|off]\nTurn stealth mode on or off (disable non-owner commands and don't return CTCP VERSION). Without arguments shows the current setting", "quitmsg": "Owner command. Usage: !quitmsg [message]\nSet the quit message. Without arguments show the current quit message", "talk": "Owner commande. Usage !talk nick message\nmake the bot send the sentence 'message' to 'nick'", "quit": "Owner command. Usage: !quit\nMake the bot quit IRC", "owner": "Usage: !owner password\nAllow to become owner of the bot", "abandon": "Owner command. Usage: !abandon\nMake the bot quit IRC without saving" } def __init__(self, my_pyborg, args): """ Args will be sys.argv (command prompt arguments) """ # PyBorg self.pyborg = my_pyborg # load settings self.settings = cfgfile.cfgset() self.settings.load("pyborg-irc.cfg", { "myname": ("The bot's nickname", "PyBorg"), "realname": ("Reported 'real name'", "Pyborg"), "owners": ("Owner(s) nickname", [ "OwnerNick" ]), "servers": ("IRC Server to connect to (server, port [,password])", [("irc.starchat.net", 6667)]), "chans": ("Channels to auto-join", ["#test"]), "speaking": ("Allow the bot to talk on channels", 1), "stealth": ("Hide the fact we are a bot", 0), "ignorelist": ("Ignore these nicknames:", []), "reply2ignored": ("Reply to ignored people", 0), "reply_chance": ("Chance of reply (%) per message", 33), "quitmsg": ("IRC quit message", "Bye :-("), "password": ("password for control the bot (Edit manually !)", ""), "othernames": ("alternate nicknames for the bot", []), "ignore_prefixes": ("prefixes (e.g. command prefixes) to ignore", ['@']), "action_percent": ("percent of the time to use an action, when reasonably appropriate", 10), "learn_action_percent": ("percent of the time to learn an action as an action", 38), #"action_possessive": ("action possessive, like his", "his"), #"action_pronoun": ("action pronoun, like he", "he"), #"action_objective": ("action objective, like him", "him"), } ) # If a mask was specified, put it in the masks. # Otherwise, put it in the owners nick list to match # the previous pyborg <=1.2.0 behavior. self.owners = [] for x in self.settings.owners: if x.find("!") != -1: self.owner_mask.append(x) else: self.owners.append(x) self.chans = self.settings.chans[:] # Parse command prompt parameters for x in xrange(1, len(args)): # Specify servers if args[x] == "-s": self.settings.servers = [] # Read list of servers for y in xrange(x+1, len(args)): if args[y][0] == "-": break server = args[y].split(":") # Default port if none specified if len(server) == 1: server.append("6667") self.settings.servers.append( (server[0], int(server[1])) ) # Channels if args[x] == "-c": self.settings.chans = [] # Read list of channels for y in xrange(x+1, len(args)): if args[y][0] == "-": break self.settings.chans.append("#"+args[y]) # Nickname if args[x] == "-n": try: self.settings.myname = args[x+1] except IndexError: pass def our_start(self): print "Connecting to server..." SingleServerIRCBot.__init__(self, self.settings.servers, self.settings.myname, self.settings.realname, 2) self.start() def on_welcome(self, c, e): print self.chans for i in self.chans: c.join(i) def shutdown(self): try: self.die() # disconnect from server except AttributeError, e: # already disconnected probably (pingout or whatever) pass def get_version(self): if self.settings.stealth: # stealth mode. we shall be a windows luser today return "VERSION mIRC32 v5.6 K.Mardam-Bey" else: return self.pyborg.ver_string def is_owner(self, source): if user_match_masks(source, self.owner_mask): return True else: return False def on_kick(self, c, e): """ Process leaving """ # Parse Nickname!username@host.mask.net to Nickname kicked = e.arguments()[0] kicker = e.source().split("!")[0] target = e.target() #channel if len(e.arguments()) >= 2: reason = e.arguments()[1] else: reason = "" if kicked == self.settings.myname: print "[%s] <-- %s was kicked off %s by %s (%s)" % (get_time(), kicked, target, kicker, reason) def on_privmsg(self, c, e): self.on_msg(c, e, learn = 0) def on_pubmsg(self, c, e): self.on_msg(c, e) def on_ctcp(self, c, e): ctcptype = e.arguments()[0] if ctcptype == "ACTION": self.on_msg(c, e) else: SingleServerIRCBot.on_ctcp(self, c, e) def _on_disconnect(self, c, e): # self.channels = IRCDict() print "deconnection" self.connection.execute_delayed(self.reconnection_interval, self._connected_checker) def selectuser(self, channel, default, bot, users): # if it was within the last two minutes, give it a higher weight, up to 10% # if it was within 12 channel messages, give it a higher weight, up to 20% curtime = time.time() if len(users) == 0: print "Zomg no users?!" return "(see log)" users = list(users) if bot and bot != "": try: users.remove(bot) except ValueError: pass if not self.activity.has_key(channel): return users[randi(0,len(users)-1)] usersactivity = self.activity[channel][0] totalactivity = self.activity[channel][1] weights = [] for u in users: if u == bot: continue if not usersactivity.has_key(u): continue a = usersactivity[u] timesincelast = curtime - a[2] if timesincelast > 600: continue if timesincelast > 300: weight = 0 else: weight = 220 + int(max(0.0, float(120 - timesincelast)/(120.0/130.0))) #35% msgsincelast = totalactivity - a[0] if msgsincelast > 12: weight += 0 #heh else: weight += 390 + int(max(0.0, float(12 - msgsincelast)/(12.0/260.0))) #65% weights.append([u, weight]) if len(weights) == 0: print "No user activity acceptable, using default nick" return default totalweight = 0 for w in weights: totalweight += w[1] print "Fresh weights: %s total %d" % (weights, totalweight) # put into 1000 if totalweight == 0: print "Total weights 0, using default nick" return default adjustment = 1000.0/float(totalweight) for i in xrange(0, len(weights)): weights[i][1] = int(float(weights[i][1]) * adjustment) weights.sort(lambda x, y: x[1]-y[1]) num = 0 for i in xrange(0, len(weights)): oldval = weights[i][1] weights[i][1] = num num += oldval print "Weights stuff: %s" % weights weights.sort(reverse=True) randi = randint(1, 1000) for w in weights: if randi > w[1]: print "%s won" % w return w[0] print "User random selection error" return "[NoUser]" def on_msg(self, c, e, learn = 1): """ Process messages. """ # Parse Nickname!username@host.mask.net to Nickname source = e.source().split("!")[0] target = e.target() isowner = 0 # First message from owner 'locks' the owner host mask # se people can't change to the owner nick and do horrible # stuff like '!unlearn the' :-) if not e.source() in self.owner_mask and source in self.owners: self.owner_mask.append(e.source()) isowner = 1 print "Locked owner as %s" % e.source() if isowner==0: if self.is_owner(e.source()): isowner=1 # Message text if len(e.arguments()) == 1: # Normal message body = e.arguments()[0] else: # A CTCP thing if e.arguments()[0] == "ACTION": randi = randint(1, 100) if randi <= self.settings.learn_action_percent: body = "/me " + e.arguments()[1] else: body = source + " " + e.arguments()[1] else: # Ignore all the other CTCPs return #print "Before color stripping: %s" % body body = restripcolor.sub("", body).strip() #print "After color stripping: %s" % body #remove special irc fonts chars body = body[body.rfind("\x02")+1:] body = body[body.rfind("\xa0")+1:] # WHOOHOOO!! if target == self.settings.myname or source == self.settings.myname: print "[%s] <%s> > %s> %s" % ( get_time(), source, target, body) # Ignore self. if source == c.get_nickname(): return # Ignore specified prefixes, case insensitively for x in self.settings.ignore_prefixes: if re.match(re.escape(x), body, re.I): return if re.match("RESULT:", body, re.I): return # Ignore selected nicks if self.settings.ignorelist.count(source.lower()) > 0 \ and self.settings.reply2ignored == 1: print "Nolearn from %s" % source learn = 0 elif self.settings.ignorelist.count(source.lower()) > 0: print "Ignoring %s" % source return if body.startswith(self.pyborg.settings.command_prefix): if self.irc_commands(body, source, target, c, e): return if isowner: self.pyborg.do_commands(self, body, (body, source, target, c, e), owner=1) return # if not owner, then only process command if stealth mode is off else: if not self.settings.stealth: #start a new thread thread.start_new_thread(self.pyborg.do_commands, (self, body, (body, source, target, c, e))) return else: body = body[len(self.pyborg.settings.command_prefix):] # replace real nickname with expected nickname body = word_replace(body, c.get_nickname(), self.settings.myname) # We want replies reply_chance%, if speaking is on replyrate = self.settings.speaking * self.settings.reply_chance # Always reply to private messages if adjust_eventtype(e) == "privmsg": replyrate = 100 # replace the bot's other names with his main name for x in self.settings.othernames: body = word_replace(body, x, self.settings.myname) # fix case body = word_replace(body, self.settings.myname, self.settings.myname) # double reply chance if the text contains our nickname :-) if word_find(body, self.settings.myname, casesens=True) != -1: replyrate = 100 #replace nicknames by "#nick" if adjust_eventtype(e) == "pubmsg": if not self.activity.has_key(target): self.activity[target] = [ {}, 0 ] if not self.activity[target][0].has_key(source): self.activity[target][0][source] = [ 0, 0, 0 ] self.activity[target][1] += 1 self.activity[target][0][source][0] = self.activity[target][1] self.activity[target][0][source][1] += 1 self.activity[target][0][source][2] = time.time() users = self.channels[target].users() for x in users: reuser = re.compile(wordrestr % re.escape(x), re.I) if reuser.search(body) == None: continue randi = randint(1,100) print "randi: %d" % randi if (randi <= 85) or x.lower() == self.settings.myname.lower(): if randi <= 3: ## change me to 4 print "Selecting user in on_msg" body = reuser.sub(self.selectuser(target, "#nick", c.get_nickname(), users), body) else: body = reuser.sub("#nick", body) else: print "Not replacing nick %s" % x elif adjust_eventtype(e) == "privmsg": body = word_replace(body, self.settings.myname, "#nick") body = word_replace(body, source, "#nick") if body == "": return # Ignore quoted messages if body[0] == "<" or body[0:1] == "\"" or body[0:1] == " <" or \ (body[0] == "[" and body[1:2].isdigit()): print "Ignoring quoted text" return print "Preprocessed body: " + body # Pass message onto pyborg if isowner: learn = 1 self.pyborg.process_msg(self, body, replyrate, learn, (body, source, target, c, e), owner=1) else: #start a new thread thread.start_new_thread(self.pyborg.process_msg, (self, body, replyrate, learn, (body, source, target, c, e))) def irc_commands(self, body, source, target, c, e): """ Special IRC commands. """ msg = "" command_list = body.split() command_list[0] = command_list[0].lower() ### User commands # Query replyrate if command_list[0] == self.pyborg.settings.command_prefix + "replyrate" and len(command_list)==1: msg = "Reply rate is "+`self.settings.reply_chance`+"%." if command_list[0] == self.pyborg.settings.command_prefix + "owner" and len(command_list) > 1 and source not in self.owners: if command_list[1] == self.settings.password: self.owners.append(source) self.output("You've been added to owners list", ("", source, target, c, e), command=True) else: self.output("Try again", ("", source, target, c, e), command=True) ### Owner commands if self.is_owner(e.source()): # Change nick if command_list[0] == self.pyborg.settings.command_prefix + "nick": try: self.connection.nick(command_list[1]) self.settings.myname = command_list[1] except: pass # stealth mode elif command_list[0] == self.pyborg.settings.command_prefix + "stealth": msg = "Stealth mode " if len(command_list) == 1: if self.settings.stealth == 0: msg = msg + "off" else: msg = msg + "on" else: toggle = command_list[1].lower() if toggle == "on": msg = msg + "on" self.settings.stealth = 1 else: msg = msg + "off" self.settings.stealth = 0 # filter mirc colours out? elif command_list[0] == self.pyborg.settings.command_prefix + "nocolor" or command_list[0] == self.pyborg.settings.command_prefix + "nocolour": msg = "obsolete command " # Allow/disallow replying to ignored nicks # (they will never be learnt from) elif command_list[0] == self.pyborg.settings.command_prefix + "reply2ignored": msg = "Replying to ignored users " if len(command_list) == 1: if self.settings.reply2ignored == 0: msg = msg + "off" else: msg = msg + "on" else: toggle = command_list[1] if toggle == "on": msg = msg + "on" self.settings.reply2ignored = 1 else: msg = msg + "off" self.settings.reply2ignored = 0 # Stop talking elif command_list[0] == self.pyborg.settings.command_prefix + "shutup": if self.settings.speaking == 1: msg = "I'll be quiet :-(" self.settings.speaking = 0 else: msg = ":-x" # Wake up again elif command_list[0] == self.pyborg.settings.command_prefix + "wakeup": if self.settings.speaking == 0: self.settings.speaking = 1 msg = "Whoohoo!" else: msg = "But i'm already awake..." # Join a channel or list of channels elif command_list[0] == self.pyborg.settings.command_prefix + "join": for x in xrange(1, len(command_list)): if not command_list[x] in self.chans: msg = "Attempting to join channel %s" % command_list[x] self.chans.append(command_list[x]) c.join(command_list[x]) # Part a channel or list of channels elif command_list[0] == self.pyborg.settings.command_prefix + "part": for x in xrange(1, len(command_list)): if command_list[x] in self.chans: msg = "Leaving channel %s" % command_list[x] self.chans.remove(command_list[x]) c.part(command_list[x]) # List channels currently on elif command_list[0] == self.pyborg.settings.command_prefix + "chans": if len(self.channels.keys())==0: msg = "I'm currently on no channels" else: msg = "I'm currently on " channels = self.channels.keys() for x in xrange(0, len(channels)): msg = msg+channels[x]+" " # add someone to the ignore list elif command_list[0] == self.pyborg.settings.command_prefix + "ignore": # if no arguments are given say who we are # ignoring if len(command_list) == 1: msg = "I'm ignoring " if len(self.settings.ignorelist) == 0: msg = msg + "nobody" else: for x in xrange(0, len(self.settings.ignorelist)): msg = msg + self.settings.ignorelist[x] + " " # Add everyone listed to the ignore list # eg !ignore tom dick harry else: for x in xrange(1, len(command_list)): self.settings.ignorelist.append(command_list[x].lower()) msg = "done" # remove someone from the ignore list elif command_list[0] == self.pyborg.settings.command_prefix + "unignore": # Remove everyone listed from the ignore list # eg !unignore tom dick harry for x in xrange(1, len(command_list)): try: self.settings.ignorelist.remove(command_list[x].lower()) msg = "done" except: pass # set the quit message elif command_list[0] == self.pyborg.settings.command_prefix + "quitmsg": if len(command_list) > 1: self.settings.quitmsg = body.split(" ", 1)[1] msg = "New quit message is \"%s\"" % self.settings.quitmsg else: msg = "Quit message is \"%s\"" % self.settings.quitmsg # make the pyborg quit elif command_list[0] == self.pyborg.settings.command_prefix + "quit": sys.exit() elif command_list[0] == self.pyborg.settings.command_prefix + "abandon": self.pyborg.should_save = False self.pyborg.settings.num_contexts = self.pyborg.num_contexts_at_load self.pyborg.settings.save() sys.exit() # Change reply rate elif command_list[0] == self.pyborg.settings.command_prefix + "replyrate": try: self.settings.reply_chance = int(command_list[1]) msg = "Now replying to %d%% of messages." % int(command_list[1]) except: msg = "Reply rate is %d%%." % self.settings.reply_chance #make the bot talk elif command_list[0] == self.pyborg.settings.command_prefix + "talk": if len(command_list) >= 2: phrase="" for x in xrange (2, len (command_list)): phrase = phrase + str(command_list[x]) + " " self.output(phrase, ("", command_list[1], "", c, e)) # Save changes self.pyborg.settings.save() self.settings.save() if msg == "": return 0 else: self.output(msg, ("", source, target, c, e), command=True) return 1 def output(self, message, args, command=False): """ Output a line of text. args = (body, source, target, c, e) """ print "Pyborg output: %s" % message if not self.connection.is_connected(): print "Can't send reply : not connected to server" return # Unwrap arguments body, source, target, c, e = args pubmsg = (adjust_eventtype(e) == "pubmsg") action = 0 users = [] # replace by the good nickname if not command: if pubmsg: users = self.channels[e.target()].users() for u in users: reuser = re.compile(wordrestr % re.escape(u.lower())) message = reuser.sub(u, message) # decide if we should do a ctcp action randi = randint(1,100) if message.startswith("#nick ") and len(message) > 6 and message[6] != ":" and randi <= self.settings.action_percent: action = 1 message = message[6:] #message = nicksre.sub(self.settings.action_possessive, message) #message = nickre.sub(self.settings.action_pronoun, message) elif message.startswith("/me "): action = 1 message = message[4:] elif message.startswith("me "): randi = randint(1, 100) if randi <= 5: action = 0 if randi <= 2: if pubmsg: message = self.selectuser(target, source, c.get_nickname(), users) + message[3:] else: message = "you " + message[3:] else: action = 1 message = message[3:] message = word_replace(message, "/me", "me") if pubmsg: p = 0 while True: m = nickre.search(message, p) if not m: break if p == 0: if randint(1,100) <= 33: print "Selecting user in output, first nick" user = self.selectuser(target, source, c.get_nickname(), users) else: user = source else: #randi = randint(0, len(users)-1) #user = users[randi] #if user == c.get_nickname(): # user = self.settings.myname randi = randint(1, 100) if randi <= 4: user = self.settings.myname else: print "Selecting user in output, other nick" user = self.selectuser(target, source, "", users) lendiff = len(user)-5 # 5 for "#nick" p = m.end() + lendiff message = message[:m.start()] + user + message[m.end():] else: message = nickre.sub(source, message) if action == 0: message = word_replace(message, self.settings.myname + "'s", "my") rename = re.compile(wordrestr % re.escape(self.settings.myname), re.I) #m = rename.search(message) message = rename.sub("me", message) #if message.startswith("me:"): # message = "note to self:" + message[3:] else: # it is a command; command results need to be identifiable message = "RESULT: " + message # Joins replies and public messages if e.eventtype() == "join" or e.eventtype() == "quit" or e.eventtype() == "part" or adjust_eventtype(e) == "pubmsg": if action == 0: print "[%s] <%s> > %s> %s" % ( get_time(), self.settings.myname, target, message) c.privmsg(target, message) else: print "[%s] <%s> > %s> /me %s" % ( get_time(), self.settings.myname, target, message) c.action(target, message) # Private messages elif adjust_eventtype(e) == "privmsg": # normal private msg if action == 0: print "[%s] <%s> > %s> %s" % ( get_time(), self.settings.myname, source, message) c.privmsg(source, message) # send copy to owner #if not source in self.owners: # c.privmsg(','.join(self.owners), "(From "+source+") "+body) # c.privmsg(','.join(self.owners), "(To "+source+") "+message) # ctcp action priv msg else: print "[%s] <%s> > %s> /me %s" % ( get_time(), self.settings.myname, target, message) c.action(source, message) # send copy to owner #if not source in self.owners: # map ( ( lambda x: c.action(x, "(From "+source+") "+body) ), self.owners) # map ( ( lambda x: c.action(x, "(To "+source+") "+message) ), self.owners) def main(): if "--help" in sys.argv: print "Pyborg irc bot. Usage:" print " pyborg-irc.py [options]" print " -s server:port" print " -c channel" print " -n nickname" print "Defaults stored in pyborg-irc.cfg" print sys.exit(0) # start the pyborg my_pyborg = pyborg.pyborg() disconnected = False bot = ModIRC(my_pyborg, sys.argv) try: bot.our_start() except KeyboardInterrupt, e: pass except SystemExit, e: pass except: bot.disconnect("Crashed!") disconnected = True traceback.print_exc() c = raw_input("Ooops! It looks like Pyborg has crashed. Would you like to save its dictionary? (y/n) ") if c.lower()[:1] == 'n': sys.exit(0) if not disconnected: bot.disconnect(bot.settings.quitmsg) if my_pyborg.should_save: my_pyborg.save_all() del my_pyborg if __name__ == "__main__": main()