codurr

From MediaWiki.org
Jump to navigation Jump to search

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()

See also[edit]