Simple Twitter poster written in Python

[ permalink ] [ download ]
#!/usr/bin/python
# -*- coding: utf-8 -*-

import urllib
import urllib2
import httplib
import feedparser
import getopt
import sys
import getpass
import signal

class TwitterRequest:

    def __init__(self, username=None, password=None):
        self._username = username
        self._password = password

    def doAuthRequest(self, url, data=None):
        try:
            password_mgr = urllib2.HTTPPasswordMgrWithDefaultRealm()
            password_mgr.add_password(None, "http://twitter.com/", self._username, self._password)
            handler = urllib2.HTTPBasicAuthHandler(password_mgr)
            opener = urllib2.build_opener(handler)
            opener.addheaders = [ ( 'User-agent', 'Mozilla/5.0' ) ]
            req = urllib2.Request(url, data)
            return opener.open(req)
        except urllib2.HTTPError, err:
            print str(err)
            sys.exit(1)

    def doRequest(self, url):
        try:
            opener = urllib2.build_opener()
            opener.addheaders = [ ( 'User-agent', 'Mozilla/5.0' ) ]
            req = urllib2.Request(url)
            return opener.open(req)
        except urllib2.HTTPError, err:
            print str(err)
            sys.exit(1)
 
############################################################################################################

class SimpleTwitter:

    def __init__(self, username, password):
        self._username = username
        self._password = password
        self.msglen = 140

    def _truncate(self, msg):
        if len(msg) > self.msglen:
            print "Message truncated. Maximum allowed message length is " + str(self.msglen) + " characters."
            return msg[:self.msglen]
        else:
            return msg
    
    def getUsername(self):
        return self._username

    def getPassword(self):
        return self._password

    def getFriendsTimeline(self):
        req = TwitterRequest(self.getUsername(), self.getPassword())
        f = req.doAuthRequest('http://twitter.com/statuses/friends_timeline.rss')
        d = feedparser.parse(f.read())

        for elem in reversed(d['entries']):
            print elem.updated + " -- " + elem.title

    def getUserTimeline(self):
        req = TwitterRequest(self.getUsername(), self.getPassword())
        data = urllib.urlencode({'id' : self.getUsername()})
        f = req.doAuthRequest('http://twitter.com/statuses/user_timeline.rss', data)
        d = feedparser.parse(f.read())

        for elem in reversed(d['entries']):
            print elem.updated + " -- " + elem.title

    def getPublicTimeline(self):
        req = TwitterRequest()
        f = req.doRequest('http://twitter.com/statuses/public_timeline.rss')
        d = feedparser.parse(f.read())

        for elem in reversed(d['entries']):
            print elem.updated + " -- " + elem.title

    def update(self, msg):
        data = urllib.urlencode({'status' : self._truncate(msg)})
        req = TwitterRequest(self.getUsername(), self.getPassword())
        f = req.doAuthRequest('http://twitter.com/statuses/update.xml', data)
        if f:
            print "Update OK!"
        else:
            print "Error updating..."

    def directMsg(self, user, msg):
        data = urllib.urlencode({'user' : str(user), 'text' : self._truncate(msg)})
        req = TwitterRequest(self.getUsername(), self.getPassword())
        f = req.doAuthRequest('http://twitter.com/direct_messages/new.xml', data)
        if f:
            print "Direct message sent OK!"
        else:
            print "Error sending Direct Message..."

    def follow(self, user):
        req = TwitterRequest(self.getUsername(), self.getPassword())
        f = req.doAuthRequest('http://twitter.com/friendships/create/'+str(user)+'.xml')
        if f:
            print "OK, now following user " + str(user)
        else:
            print "Error trying to follow user..."

############################################################################################################

def usage():
    print """
    SimpleTwitter usage:
    ===================

    ./twitter.py [options] <message>
    
    Options:
    -d              Delete last post. (Not yet implemented)
    -u              Show user timeline.
    -f              Show friends timeline.
    -p              Show public timeline.
    -F <name>       Follow user.
    -D <name>       Send direct message to user.

    """

def get_user_data():
    user = raw_input("Enter your username: ")
    password = getpass.getpass("Enter your password: ")
    return SimpleTwitter(user, password)

def signal_handler(signal, frame):
            print '\nYou pressed Ctrl+C!\n'
            sys.exit(0)

def main():
    signal.signal(signal.SIGINT, signal_handler)
    try:
        opts, args = getopt.getopt(sys.argv[1:], "dufpF:D:")
        if len(opts) == 0:
            if len(args) != 1:
                usage()
                sys.exit(1)
            else:
                twitt = get_user_data()
                twitt.update(str(args[0]))
                sys.exit(0)
        else:
            twitt = get_user_data()
            for o, a in opts:
                if o == "-d":
                    print "Delete is not yet implemented."
                    sys.exit(0)
                elif o == "-u":
                    print "Getting user timeline... "
                    twitt.getUserTimeline()
                    sys.exit(0)
                elif o == "-f":
                    print "Getting friends timeline... "
                    twitt.getFriendsTimeline()
                    sys.exit(0)
                elif o == "-p":
                    print "Getting public timeline... "
                    twitt.getPublicTimeline()
                    sys.exit(0)
                elif o == "-F":
                    print "Request to follow user " + a
                    twitt.follow(a)
                    sys.exit(0)
                elif o == "-D":
                    print "Direct message to user " + a
                    twitt.directMsg(a, str(args[0]))
                    sys.exit(0)

    except getopt.GetoptError, err:
        print str(err)
        usage()
        sys.exit(1)


if __name__ == "__main__":
    main()
hits counter