codurr
This page is obsolete. It is being retained for archival purposes. It may document extensions or features that are obsolete and/or no longer supported. Do not rely on the information here being up-to-date. |
codurr was an IRC bot that idled in the #mediawiki support channel on the freenode IRC network.
Functions
edit- echoed CIA output from #mediawiki-codereview to #mediawiki, adding links and formatting
- output code comments to #mediawiki and #mediawiki-codereview
- output code status and tag changes to #mediawiki-codereview
- provided much-needed TLC
Source
edit#! /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()