#!/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
nicksre = re.compile(r"(?:^|(?<=\W))#nick's(?=\W|$)")
nickre = re.compile(r"(?:^|(?<=\W))#nick(?=\W|$)")
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(r"(?:^|(?<=\W))%s(?=\W|$)" % 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(r"(?:^|(?<=\W))%s(?=\W|$)" % 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"
should_save = True
# 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)
users.remove(bot)
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 > 120:
weight = 0
else:
weight = 20 + int(min(0, (120 - timesincelast)/(120.0/15.0))) #35%
msgsincelast = totalactivity - a[0]
if msgsincelast > 12:
weight += 0 #heh
else:
weight += 20 + int(min(0, (12 - msgsincelast)/(12.0/45.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]
# put into 1000
if totalweight == 0:
print "Total weights 0, using default nick"
return default
adjustment = 1000.0/totalweight
for i in xrange(0, len(weights)):
weights[i][1] = int(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)
print randi
for w in weights:
if randi > w[1]:
print 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
for x in self.settings.ignore_prefixes:
if body.startswith(x): return
# replace real nickname with expected nickname
body = word_replace(body, c.get_nickname(), self.settings.myname)
# 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):]
# 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:
randi = randint(1,100)
if (randi <= 70) or x == self.settings.myname:
if randi <= 4: ## change me to 4
print "Selecting user in on_msg"
body = word_replace(body, x, self.selectuser(target, "#nick", c.get_nickname(), users))
else:
body = word_replace(body, x, "#nick")
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] == " <":
print "Ignoring quoted text"
return
print 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.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, ("<none>", source, target, c, e))
return 1
def output(self, message, args, command=False):
"""
Output a line of text. args = (body, source, target, c, e)
"""
print "MSG: %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
action = 0
# replace by the good nickname
if not command:
users = self.channels[e.target()].users()
for u in users:
message = (" " + message + " ").replace(u.lower(), u).strip()
# 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:]
if adjust_eventtype(e) == "pubmsg":
p = 0
while True:
m = nickre.search(message, p)
if not m: break
if p == 0:
if randint(1,100) <= 4:
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")
message = word_replace(message, self.settings.myname, "me")
# 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 bot.should_save: my_pyborg.save_all()
del my_pyborg
if __name__ == "__main__":
main()