codurr
From MediaWiki.org
codurr is an IRC bot that idles in the #mediawiki support channel on the freenode IRC network.
[edit] Functions
- echos CIA output from #mediawiki-codereview to #mediawiki, adding links and formatting
- outputs code comments to #mediawiki and #mediawiki-codereview
- outputs code status and tag changes to #mediawiki-codereview
- provides much-needed TLC
[edit] Source
#! /usr/bin/env python
# Public domain; MZMcBride; 2011
import operator
import os
import random
import re
import time
import urllib
import MySQLdb
from twisted.words.protocols import irc
from twisted.internet import reactor, protocol, task
server = 'irc.freenode.net'
trusted = ['wikipedia/MZMcBride',
'wikipedia/TimStarling',
'mediawiki/Catrope']
class snerkBot(irc.IRCClient):
realname = 'codurr'
nickname = 'codurr'
altnick = 'codurrr'
hostmask = 'willow.toolserver.org'
channels = ['#mediawiki', '#mediawiki-codereview']
password = ''
got_Pong = True
all_comments = set()
all_prop_changes = set()
cc_file_contents = set()
base_url = 'http://www.mediawiki.org/wiki/Special:Code/MediaWiki/'
flood = 1
def connectionMade(self):
irc.IRCClient.connectionMade(self)
self.lc = task.LoopingCall(self.sendServerPing)
self.lc.start(60, False)
self.lc2 = task.LoopingCall(self.retrieve_code_props)
self.lc2.start(15, False)
self.lc3 = task.LoopingCall(self.retrieve_cc_file)
self.lc3.start(60, False)
def connectionLost(self, reason):
irc.IRCClient.connectionLost(self, reason)
def signedOn(self):
for i in self.channels:
self.join(i)
def irc_ERR_NICKNAMEINUSE(self, prefix, params):
self.register(self.altnick)
def kill_self(self):
os._exit(0)
def sendServerPing(self):
if not self.got_Pong:
self.kill_self()
self.got_Pong = False
self.sendLine('PING %s' % server)
def retrieve_code_props(self):
try:
# Comments first
# http://www.mediawiki.org/wiki/Special:Code/MediaWiki/comments
temp_comments = set()
conn = MySQLdb.connect(host='sql-s3',
db='mediawikiwiki_p',
read_default_file='/home/project/m/w/b/mwbot/.my.cnf')
cursor = conn.cursor()
cursor.execute('''SELECT
cc_timestamp,
cc_rev_id,
cc_id,
cc_user_text,
cc_text
FROM code_comment
WHERE cc_repo_id = 1
ORDER BY cc_timestamp DESC
LIMIT 50;''')
for row in cursor.fetchall():
cc_text = '%s' % row[4][:100]
cc_text = re.sub(r'\n{1,}', r'\n', cc_text)
cc_text = re.sub(r'\n', r' \\ ', cc_text)
cc_text = re.sub(r'\s{1,}', ' ', cc_text)
entry = ('%s' % row[0], # cc_timestamp
'%s' % row[1], # cc_rev_id
'%s' % row[2], # cc_id
'%s' % row[3], # cc_user_text
'%s' % cc_text)# cc_text
temp_comments.add(entry)
final_comments_set = temp_comments - self.all_comments
self.all_comments |= temp_comments
# Now let's do code prop changes (status changes and tag changes)
# http://www.mediawiki.org/wiki/Special:Code/MediaWiki/statuschanges
temp_prop_changes = set()
cursor.execute('''SELECT
cpc_timestamp,
cpc_rev_id,
cpc_attrib,
cpc_removed,
cpc_added,
cpc_user_text
FROM code_prop_changes
WHERE cpc_repo_id = 1
ORDER BY cpc_timestamp DESC
LIMIT 50;''')
for row in cursor.fetchall():
entry = ('%s' % row[0], # cpc_timestamp
'%s' % row[1], # cpc_rev_id
'%s' % row[2], # cpc_attrib
'%s' % row[3], # cpc_removed
'%s' % row[4], # cpc_added
'%s' % row[5]) # cpc_user_text
temp_prop_changes.add(entry)
final_prop_changes_set = temp_prop_changes - self.all_prop_changes
self.all_prop_changes |= temp_prop_changes
for i in sorted(final_comments_set, key=operator.itemgetter(0)):
if self.flood != 1:
message = 'New code comment: %s; %s; <%s%s#c%s>' % (i[3],
i[4],
self.base_url,
i[1],
i[2])
for i in self.channels:
self.msg(i, message)
for i in sorted(final_prop_changes_set, key=operator.itemgetter(0)):
if self.flood != 1:
if i[2] == 'status':
message = 'Code status change: %s; removed: %s; added: %s; <%s%s>' % (i[5],
i[3],
i[4],
self.base_url,
i[1])
elif i[2] == 'tags':
message = 'Code tags change: %s; removed: %s; added: %s; <%s%s>' % (i[5],
i[3],
i[4],
self.base_url,
i[1])
self.msg(self.channels[1], message)
time.sleep(0.5)
cursor.close()
conn.close()
self.flood += 1
except:
pass
return
def retrieve_cc_file(self):
def clean_cc_file(cc_file):
# Strip preceding or trailing newlines and replace all other newlines with spaces.
cc_file = cc_file.strip('\n').replace('\n',' ')
# Grab usernames and scramble them to avoid pings.
user_name_re = re.compile(r'\b([\w\-]+?)/r(\d+?)\b')
for match in user_name_re.finditer(cc_file):
original_user_name = match.group(1).strip()
lowered_user_name = original_user_name.lower()
# Convert the usernames to all lowercase...
cc_file = re.sub(re.escape(original_user_name), lowered_user_name, cc_file)
revision = match.group(2)
if original_user_name:
while True:
scrambled_user_name = ''.join(random.sample(lowered_user_name, len(lowered_user_name)))
if scrambled_user_name != lowered_user_name:
cc_file = re.sub(re.escape(lowered_user_name), scrambled_user_name, cc_file)
break
return cc_file
try:
cc_file_url = 'http://ci.tesla.usability.wikimedia.org/irc/irc-publish.txt'
cc_file = urllib.urlopen(cc_file_url).read()
if cc_file not in self.cc_file_contents and cc_file != '':
for i in self.channels:
self.msg(i, '%s' % (clean_cc_file(cc_file)))
self.cc_file_contents.add(cc_file)
return
except:
pass
return
def irc_PONG(self, prefix, params):
if params:
self.got_Pong = True
def privmsg(self, sender, channel, msg):
# IRC VARIABLES
user = sender.split('!', 1)[0]
try:
hostmask = sender.split('@', 1)[1]
except:
hostmask = ''
# FIND
hurrfind = re.search(r'(\bh+u+r+)(([,.!?]+)?)(\s|$)+', msg, re.I|re.U)
durrfind = re.search(r'(\bd+u+r+)', msg, re.I|re.U)
sayfind = re.search(r'%s.{0,3}(say|echo) [\'"]{1}(.*)[\'"]{1}\.?' % self.nickname, msg, re.I|re.U)
actfind = re.search(r'%s.{0,3}(do|act) [\'"]{1}(.*)[\'"]{1}\.?' % self.nickname, msg, re.I|re.U)
if hostmask == self.hostmask:
return
elif re.search(r'CIA-\d+', user) and channel.lower() == '#mediawiki-codereview':
if msg.startswith('\x0303'):
# '\x0303nikerabbit\x0f * r\x0269263\x0f \x0310\x0f/trunk/extensions/ (SpecialTranslationStats.php)\x02:\x0f (log message trimmed)'
# new_message = re.sub(r'(\x0303|\x0f|\x02|\x0310)', '', msg)
new_message = re.sub(r'r\x02(?P<rev>\d+)\x0f', '<'+self.base_url+'\g<rev>'+'>', msg, 1)
else:
new_message = msg
self.msg('#mediawiki', new_message)
return
elif sayfind:
self.msg(channel, sayfind.group(2))
return
elif actfind:
self.describe(channel, actfind.group(2))
return
elif channel == self.nickname: # PM'ing with the bot.
try:
hostmask = sender.split('@', 1)[1]
except:
hostmask = ''
if re.search(r'^!r(estart)?\b', msg, re.I|re.U):
self.kill_self()
elif msg.startswith('#'):
target = msg.split(' ')[0]
verb = msg.split(' ')[1]
nmsg = ' '.join(msg.split(' ')[2:])
if verb in ['do', 'act']:
self.describe(target, nmsg)
elif verb in ['say', 'echo']:
self.msg(target, nmsg)
elif hostmask in trusted:
self.sendLine(msg)
return
elif hurrfind and not durrfind:
moarregex = re.match(r'^(h*)(.*)', hurrfind.group(1), re.I)
def equalize_case(str1, str2):
def case(c1, c2):
if c1.islower() != c2.islower():
return c2.swapcase()
return c2
return ''.join(map(case, str1, str2))
smsg = equalize_case(moarregex.group(1), 'd' * len(moarregex.group(1))) + moarregex.group(2)
tmsg = (equalize_case(hurrfind.group(1), smsg))
self.msg(channel, tmsg + str.replace(hurrfind.group(2).strip(','), str(None), ''))
return
elif re.search(r'(\bbastard bot\b)', msg, re.I|re.U):
self.msg(channel, 'Puck you.')
return
def action(self, user, channel, msg):
hostmask = user.split('@', 1)[1]
user = user.split('!', 1)[0]
lovefind = re.search(r'(glomps|hugs|snuggles|snuggleglomps|cuddles|snugglefucks)( a)? %s' % self.nickname, msg, re.I|re.U)
if hostmask == self.hostmask:
return
elif re.search(r'pets %s' % self.nickname, msg, re.I|re.U):
self.describe(channel, 'purrs.')
return
elif lovefind:
self.describe(channel, lovefind.group(1) + ' %s.' % user)
return
elif re.search(r'tickles %s' % self.nickname, msg, re.I|re.U):
self.describe(channel, 'giggles.')
return
elif re.search(r'(stares at|eyes) %s' % self.nickname, msg, re.I|re.U):
self.msg(channel, 'Creep.')
return
elif re.search(r'hugs( a)? %s' % self.nickname, msg, re.I|re.U):
self.describe(channel, 'hugs %s.' % user)
return
elif re.search(r'licks( a)? %s' % self.nickname, msg, re.I|re.U):
self.describe(channel, 'licks %s.' % user)
return
elif re.search(r'farts[.]?$', msg, re.I|re.U) and hostmask not in trusted:
self.msg(channel, 'Ew.')
return
elif re.search(r'mounts %s and rides (him|her|it|%s) around the room' % (self.nickname, self.nickname), msg, re.I|re.U):
self.msg(channel, ':o')
self.msg(channel, '\o/')
reactor.callLater(3, self.me, channel, 'mounts %s and rides him around the room.' % user)
return
elif re.search(r'.*( a)? %s' % self.nickname, msg, re.I|re.U):
self.msg(channel, 'Puck you.')
return
class snerkBotFactory(protocol.ClientFactory):
protocol = snerkBot
def __init__(self):
pass
def clientConnectionLost(self, connector, reason):
connector.connect()
def clientConnectionFailed(self, connector, reason):
reactor.stop()
if __name__ == '__main__':
f = snerkBotFactory()
reactor.connectTCP('%s' % server, 8001, f)
reactor.run()
# Public domain; MZMcBride; 2011
import operator
import os
import random
import re
import time
import urllib
import MySQLdb
from twisted.words.protocols import irc
from twisted.internet import reactor, protocol, task
server = 'irc.freenode.net'
trusted = ['wikipedia/MZMcBride',
'wikipedia/TimStarling',
'mediawiki/Catrope']
class snerkBot(irc.IRCClient):
realname = 'codurr'
nickname = 'codurr'
altnick = 'codurrr'
hostmask = 'willow.toolserver.org'
channels = ['#mediawiki', '#mediawiki-codereview']
password = ''
got_Pong = True
all_comments = set()
all_prop_changes = set()
cc_file_contents = set()
base_url = 'http://www.mediawiki.org/wiki/Special:Code/MediaWiki/'
flood = 1
def connectionMade(self):
irc.IRCClient.connectionMade(self)
self.lc = task.LoopingCall(self.sendServerPing)
self.lc.start(60, False)
self.lc2 = task.LoopingCall(self.retrieve_code_props)
self.lc2.start(15, False)
self.lc3 = task.LoopingCall(self.retrieve_cc_file)
self.lc3.start(60, False)
def connectionLost(self, reason):
irc.IRCClient.connectionLost(self, reason)
def signedOn(self):
for i in self.channels:
self.join(i)
def irc_ERR_NICKNAMEINUSE(self, prefix, params):
self.register(self.altnick)
def kill_self(self):
os._exit(0)
def sendServerPing(self):
if not self.got_Pong:
self.kill_self()
self.got_Pong = False
self.sendLine('PING %s' % server)
def retrieve_code_props(self):
try:
# Comments first
# http://www.mediawiki.org/wiki/Special:Code/MediaWiki/comments
temp_comments = set()
conn = MySQLdb.connect(host='sql-s3',
db='mediawikiwiki_p',
read_default_file='/home/project/m/w/b/mwbot/.my.cnf')
cursor = conn.cursor()
cursor.execute('''SELECT
cc_timestamp,
cc_rev_id,
cc_id,
cc_user_text,
cc_text
FROM code_comment
WHERE cc_repo_id = 1
ORDER BY cc_timestamp DESC
LIMIT 50;''')
for row in cursor.fetchall():
cc_text = '%s' % row[4][:100]
cc_text = re.sub(r'\n{1,}', r'\n', cc_text)
cc_text = re.sub(r'\n', r' \\ ', cc_text)
cc_text = re.sub(r'\s{1,}', ' ', cc_text)
entry = ('%s' % row[0], # cc_timestamp
'%s' % row[1], # cc_rev_id
'%s' % row[2], # cc_id
'%s' % row[3], # cc_user_text
'%s' % cc_text)# cc_text
temp_comments.add(entry)
final_comments_set = temp_comments - self.all_comments
self.all_comments |= temp_comments
# Now let's do code prop changes (status changes and tag changes)
# http://www.mediawiki.org/wiki/Special:Code/MediaWiki/statuschanges
temp_prop_changes = set()
cursor.execute('''SELECT
cpc_timestamp,
cpc_rev_id,
cpc_attrib,
cpc_removed,
cpc_added,
cpc_user_text
FROM code_prop_changes
WHERE cpc_repo_id = 1
ORDER BY cpc_timestamp DESC
LIMIT 50;''')
for row in cursor.fetchall():
entry = ('%s' % row[0], # cpc_timestamp
'%s' % row[1], # cpc_rev_id
'%s' % row[2], # cpc_attrib
'%s' % row[3], # cpc_removed
'%s' % row[4], # cpc_added
'%s' % row[5]) # cpc_user_text
temp_prop_changes.add(entry)
final_prop_changes_set = temp_prop_changes - self.all_prop_changes
self.all_prop_changes |= temp_prop_changes
for i in sorted(final_comments_set, key=operator.itemgetter(0)):
if self.flood != 1:
message = 'New code comment: %s; %s; <%s%s#c%s>' % (i[3],
i[4],
self.base_url,
i[1],
i[2])
for i in self.channels:
self.msg(i, message)
for i in sorted(final_prop_changes_set, key=operator.itemgetter(0)):
if self.flood != 1:
if i[2] == 'status':
message = 'Code status change: %s; removed: %s; added: %s; <%s%s>' % (i[5],
i[3],
i[4],
self.base_url,
i[1])
elif i[2] == 'tags':
message = 'Code tags change: %s; removed: %s; added: %s; <%s%s>' % (i[5],
i[3],
i[4],
self.base_url,
i[1])
self.msg(self.channels[1], message)
time.sleep(0.5)
cursor.close()
conn.close()
self.flood += 1
except:
pass
return
def retrieve_cc_file(self):
def clean_cc_file(cc_file):
# Strip preceding or trailing newlines and replace all other newlines with spaces.
cc_file = cc_file.strip('\n').replace('\n',' ')
# Grab usernames and scramble them to avoid pings.
user_name_re = re.compile(r'\b([\w\-]+?)/r(\d+?)\b')
for match in user_name_re.finditer(cc_file):
original_user_name = match.group(1).strip()
lowered_user_name = original_user_name.lower()
# Convert the usernames to all lowercase...
cc_file = re.sub(re.escape(original_user_name), lowered_user_name, cc_file)
revision = match.group(2)
if original_user_name:
while True:
scrambled_user_name = ''.join(random.sample(lowered_user_name, len(lowered_user_name)))
if scrambled_user_name != lowered_user_name:
cc_file = re.sub(re.escape(lowered_user_name), scrambled_user_name, cc_file)
break
return cc_file
try:
cc_file_url = 'http://ci.tesla.usability.wikimedia.org/irc/irc-publish.txt'
cc_file = urllib.urlopen(cc_file_url).read()
if cc_file not in self.cc_file_contents and cc_file != '':
for i in self.channels:
self.msg(i, '%s' % (clean_cc_file(cc_file)))
self.cc_file_contents.add(cc_file)
return
except:
pass
return
def irc_PONG(self, prefix, params):
if params:
self.got_Pong = True
def privmsg(self, sender, channel, msg):
# IRC VARIABLES
user = sender.split('!', 1)[0]
try:
hostmask = sender.split('@', 1)[1]
except:
hostmask = ''
# FIND
hurrfind = re.search(r'(\bh+u+r+)(([,.!?]+)?)(\s|$)+', msg, re.I|re.U)
durrfind = re.search(r'(\bd+u+r+)', msg, re.I|re.U)
sayfind = re.search(r'%s.{0,3}(say|echo) [\'"]{1}(.*)[\'"]{1}\.?' % self.nickname, msg, re.I|re.U)
actfind = re.search(r'%s.{0,3}(do|act) [\'"]{1}(.*)[\'"]{1}\.?' % self.nickname, msg, re.I|re.U)
if hostmask == self.hostmask:
return
elif re.search(r'CIA-\d+', user) and channel.lower() == '#mediawiki-codereview':
if msg.startswith('\x0303'):
# '\x0303nikerabbit\x0f * r\x0269263\x0f \x0310\x0f/trunk/extensions/ (SpecialTranslationStats.php)\x02:\x0f (log message trimmed)'
# new_message = re.sub(r'(\x0303|\x0f|\x02|\x0310)', '', msg)
new_message = re.sub(r'r\x02(?P<rev>\d+)\x0f', '<'+self.base_url+'\g<rev>'+'>', msg, 1)
else:
new_message = msg
self.msg('#mediawiki', new_message)
return
elif sayfind:
self.msg(channel, sayfind.group(2))
return
elif actfind:
self.describe(channel, actfind.group(2))
return
elif channel == self.nickname: # PM'ing with the bot.
try:
hostmask = sender.split('@', 1)[1]
except:
hostmask = ''
if re.search(r'^!r(estart)?\b', msg, re.I|re.U):
self.kill_self()
elif msg.startswith('#'):
target = msg.split(' ')[0]
verb = msg.split(' ')[1]
nmsg = ' '.join(msg.split(' ')[2:])
if verb in ['do', 'act']:
self.describe(target, nmsg)
elif verb in ['say', 'echo']:
self.msg(target, nmsg)
elif hostmask in trusted:
self.sendLine(msg)
return
elif hurrfind and not durrfind:
moarregex = re.match(r'^(h*)(.*)', hurrfind.group(1), re.I)
def equalize_case(str1, str2):
def case(c1, c2):
if c1.islower() != c2.islower():
return c2.swapcase()
return c2
return ''.join(map(case, str1, str2))
smsg = equalize_case(moarregex.group(1), 'd' * len(moarregex.group(1))) + moarregex.group(2)
tmsg = (equalize_case(hurrfind.group(1), smsg))
self.msg(channel, tmsg + str.replace(hurrfind.group(2).strip(','), str(None), ''))
return
elif re.search(r'(\bbastard bot\b)', msg, re.I|re.U):
self.msg(channel, 'Puck you.')
return
def action(self, user, channel, msg):
hostmask = user.split('@', 1)[1]
user = user.split('!', 1)[0]
lovefind = re.search(r'(glomps|hugs|snuggles|snuggleglomps|cuddles|snugglefucks)( a)? %s' % self.nickname, msg, re.I|re.U)
if hostmask == self.hostmask:
return
elif re.search(r'pets %s' % self.nickname, msg, re.I|re.U):
self.describe(channel, 'purrs.')
return
elif lovefind:
self.describe(channel, lovefind.group(1) + ' %s.' % user)
return
elif re.search(r'tickles %s' % self.nickname, msg, re.I|re.U):
self.describe(channel, 'giggles.')
return
elif re.search(r'(stares at|eyes) %s' % self.nickname, msg, re.I|re.U):
self.msg(channel, 'Creep.')
return
elif re.search(r'hugs( a)? %s' % self.nickname, msg, re.I|re.U):
self.describe(channel, 'hugs %s.' % user)
return
elif re.search(r'licks( a)? %s' % self.nickname, msg, re.I|re.U):
self.describe(channel, 'licks %s.' % user)
return
elif re.search(r'farts[.]?$', msg, re.I|re.U) and hostmask not in trusted:
self.msg(channel, 'Ew.')
return
elif re.search(r'mounts %s and rides (him|her|it|%s) around the room' % (self.nickname, self.nickname), msg, re.I|re.U):
self.msg(channel, ':o')
self.msg(channel, '\o/')
reactor.callLater(3, self.me, channel, 'mounts %s and rides him around the room.' % user)
return
elif re.search(r'.*( a)? %s' % self.nickname, msg, re.I|re.U):
self.msg(channel, 'Puck you.')
return
class snerkBotFactory(protocol.ClientFactory):
protocol = snerkBot
def __init__(self):
pass
def clientConnectionLost(self, connector, reason):
connector.connect()
def clientConnectionFailed(self, connector, reason):
reactor.stop()
if __name__ == '__main__':
f = snerkBotFactory()
reactor.connectTCP('%s' % server, 8001, f)
reactor.run()
