Twitter to XMPP (Jabber)

The other day I noticed that irssi (the IRC client I use) had a XMPP plugin. It occurred to me that since I’m on IRC a lot it would be really cool to pull microblog feeds into irssi via this plugin.

For identi.ca this was a doddle, since laconica supports XMPP natively. Twitter however was a problem. I believe twitter used to support XMPP but they pulled that feature for who knows what reason.

There are a couple of apps that will provided a twitter->XMPP gateway for you. I looked at twitterspy, but it sounded like a pain to install. So at first I tried tweet.im, which worked after a fashion. It got the tweets into irssi, but it did it in some weird way so that irssi didn’t realise new messages had come in and hence it didn’t indicate there was activity when I was in a different irssi window.

At this point the little voice in my head that is the bane of my life started saying “You know, you could probably write your own gateway app. How hard could it be?”. Like a fool I listened to this voice, but for once it didn’t turn out disastrously. A few hours later I had a functioning (one-way) twitter->XMPP app, written in python.

I’ve since learned that twitterspy has been rewritten in python and would have been a doddle to install. Not only that but there is a public instance of it I could have used. Still, I’m not immune to NIH syndrome and having my own app to do it is nice. If you want a twitter->XMPP gateway of your own though I suggest you try twitterspy first as it is much more capable (and mature) than my noddy little script.

But if you are interested, and because I like to boast about my tiny little achievements, here is twixm (I’ll give it its own page when I get round to it, but till then it is going to live in this blog post).

#!/usr/bin/python -u
#
# 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, see <http://www.gnu.org/licenses/>.
#

__author__ = "Ghworg"
__version__ = "1.0"
__description__ = "Twitter to XMPP gateway"

import logging, os, signal, sys, threading, time, urllib

import simplejson

try:
    from iniparse.compat import ConfigParser
except ImportError:
    from ConfigParser import ConfigParser

from pyxmpp.jid import JID
from pyxmpp.jabber.simple import send_message

class WebApiOpener(urllib.FancyURLopener):
    """
    Provides a way for HTTP Basic authentication to take place without
    prompting the user for a username and password like FancyURLopener
    would.
    "
""
    def __init__(self, username, password):
        urllib.FancyURLopener.__init__(self,{})
        self.username = username
        self.password = password
   
    def prompt_user_passwd(self, host, realm):
        """Handle Basic Auth automatically."""
        return (self.username, self.password)

class CaseSensitiveConfigParser(ConfigParser):
    """Case Sensitive version of ConfigParser"""
    def __init__(self):
        ConfigParser.__init__(self)
       
    def optionxform(self, option):
        """Make keys case-sensitive"""
        return str(option)
   
    def getAndType(self, section, option):
        """Try to guess variable type and convert to type before returning"""
        rawVal = self.get(section, option)
        isInt = True
        isFloat = True
        for c in rawVal:
            if not c.isdigit():
                isInt = False
                if c != ‘.’:
                    isFloat = False
        if isInt:
            return int(rawVal)
        elif isFloat:
            return float(rawVal)
        else:
            return rawVal

def msgToXMPP(msg):
    """Send a message to Jabber (XMPP)"""
    jid = JID(config[‘jid’])
    if not jid.resource:
        jid = JID(jid.node, jid.domain, ‘send_message’)
    recpt = JID(config[‘jid’])
    logging.debug(‘Sending to XMPP  %s’ % msg)
    send_message(jid, config[‘XMPPpass’], recpt, msg)

def getTweets(site, maxTweets=1):
    """Read in a list of tweets made since the last check"""
    if site[0] in cache:
        url = ‘%s?since_id=%d’ % (site[1], cache[site[0]])
    else:
        url = ‘%s?count=%d’ % (site[1], maxTweets)
   
    opener = WebApiOpener(site[2], site[3])        
    try:
        logging.debug(‘Requesting %s’ % url)
        webPage = opener.open(url)        
        jsonData = simplejson.load(webPage)
        if (‘error’ in jsonData) and (jsonData[‘error’] != ):
            logging.warning(‘Error from server %s: %s’ % (url, jsonData[‘error’]))
            jsonData = None
    except IOError, e:
        logging.error(e)
        logging.error(‘URL = %s’ % src)
    except ValueError, e:
        logging.error(e)
        logging.error(‘URL = %s’ % src)
   
    maxID = 0
    for tweet in jsonData:
        if tweet[u‘id’] > maxID:
            maxID = tweet[u‘id’]
        msg = ‘%s (%s): %s’ % (tweet[u‘user’][u‘screen_name’], site[0], tweet[u‘text’])
        logging.debug(‘Read tweet  %s’ % msg)
        msgToXMPP(msg)
    return maxID

def run():
    """Thread"""
    while True:
        startSignal.wait()
        if quitSignal.isSet():
            return
        startSignal.clear()
       
        for site in config[‘site’]:
            maxID = getTweets(site)
            if (site[0] not in cache) or (maxID > cache[site[0]]):
                cache[site[0]] = maxID

def readConfig():
    """Load config settings"""
    config = {}
   
    cfgfile = os.path.join(os.path.expanduser(‘~’), ‘.twixmrc’)
    parser = CaseSensitiveConfigParser()
    parser.read(cfgfile)
    settings = {}
    for section in parser.sections():
        if ‘XMPP’ == section:
            for optname in parser.options(section):
                config[optname] = parser.getAndType(section, optname)
        else:
            siteName = section
            siteUrl = parser.getAndType(section, ‘URL’)
            siteUser = parser.getAndType(section, ‘Username’)
            sitePass = parser.getAndType(section, ‘Password’)
            site = (siteName, siteUrl, siteUser, sitePass)
            if ‘site’ not in config:
                config[‘site’] = []
            config[‘site’].append(site)
   
    config[‘cycleTime’] = 60.0
   
    # !!! Check for missing config settings and bail out
   
    logging.debug(‘XMPP ID = %s’ % config[‘jid’])
    logging.debug(‘XMPP Pass = %s’ % config[‘XMPPpass’])
    logging.debug(‘Cycle Time = %s’ % config[‘cycleTime’])
    for site in config[‘site’]:
        logging.debug(‘Site: %s’ % site[0])
        logging.debug(\tURL: %s’ % site[1])
        logging.debug(\tUser: %s’ % site[2])
        logging.debug(\tPass: %s’ % site[3])
   
    return config
       

def setupLogging():
    """Setup python logging to stdout"""
    log = logging.getLogger()
    logformat = ‘%(asctime)s %(levelname)s %(message)s’
    dateformat = ‘%Y-%m-%d %H:%M:%S’
    frmttr = logging.Formatter(logformat, dateformat)
    #if options.logfile:
    #    fhdlr = logging.FileHandler(options.logfile)
    #    fhdlr.setFormatter(frmttr)
    #    log.addHandler(fhdlr)
    if sys.stdin.isatty():
        shdlr = logging.StreamHandler(sys.stdout)
        shdlr.setFormatter(frmttr)
        log.addHandler(shdlr)
    #log.setLevel(logging.DEBUG)
    log.setLevel(logging.INFO)

def handler(signum, frame):
    """Handle various termination signals gracefully"""
    logging.warning(‘Signal handler called with signal %d’ % signum)
    quitSignal.set()
    startSignal.set()

###############################################################################
# Main
###############################################################################

# Set the signal handlers
signal.signal(signal.SIGTERM, handler)
signal.signal(signal.SIGQUIT, handler)
signal.signal(signal.SIGINT, handler)

setupLogging()

config = readConfig()
cache = {}

quitSignal = threading.Event()
startSignal = threading.Event()

runThread = threading.Thread(target=run)
runThread.start()

while False == quitSignal.isSet():
    startSignal.set()
    count = 0.0
    while count < config[‘cycleTime’]:
        quitSignal.wait(0.5)
        count += 0.5
        if quitSignal.isSet():
            break
logging.info(‘Exiting’)

 

Update 2009-11-25: I have created a page for twixm now. Go there for the latest code and info.