bit.lyでURLを短縮してTwiiterに投稿するPythonコード

ブログとTwitterの棲み分けは色々議論があると思いますが、個人的にはまとまった記事にならないような一言メモなんかはTwitterにポストするようにしています。

はてなブックマークがWeb Hookをリリースしてから色々妄想したところ、ありがちですがはてブとTwitterを連動させたら面白そうだなと思いました。
下手したらうざいリンクスパムになるかもしれませんが。。。


最近はHookHubなるサービスが登場したのでわざわざ自分で作る必要ないかもしれませんが、とりあえず勉強のために自作することに。

TwitterのAPIはURLを短縮してくれないようなので、あわせてbit.lyのAPIも叩きます。

bit.lyのAPIは認証式なので、アカウントを作成してAPI Keyをゲットする必要があるのでご注意を。


両APIのドキュメントはこちら

Twitter API Wiki
bit.ly API Documentation

import urllib, urllib2, logging
import xml.dom.minidom

class Twitter:

    def __init__(self, login, password, bitly_login=None, bitly_apikey=None):
        self.uri = 'http://twitter.com'
        self.end_point = ''
        self.login = login
        self.password = password
        self.bitly_login = bitly_login
        self.bitly_apikey = bitly_apikey

    def bitly_account(self, bitly_login=None, bitly_apikey=None):
        self.bitly_login = bitly_login
        self.bitly_apikey = bitly_apikey

    def post(self, body='', url='', limit=20):
        self.end_point = '/statuses/update.xml'

        # bit.ly
        if len(url) > limit and self.bitly_login and self.bitly_apikey:
            bit = Bitly(self.bitly_login, self.bitly_apikey)
            try: url = bit.shorten(url).get()
            except BitlyError: pass

        message = '%s %s' % (body, url)
        options = {'status': message.encode('utf_8')}
        return self.send_request(options, auth=True)

    def destroy(self, status_id):
        self.end_point = '/statuses/destroy/%s.xml' % status_id
        return self.send_request({}, auth=True)


    def send_request(self, options, auth=False):
        if auth: self.auth()

        try:
            request = urllib2.Request(self.uri + self.end_point, urllib.urlencode(options))
            response = urllib2.urlopen(request)

            if response.code != 200:
                e = 'Twitter response: code > %d, message > %s' % (response.code, response.msg)
                logging.error(e)
                raise TwitterRequestError, e

            return TwitterResponse(response.read())
        finally:
            response.close()

    def auth(self):
        #Basic Auth
        auth_handler = urllib2.HTTPBasicAuthHandler()
        auth_handler.add_password('Twitter API', self.uri, self.login, self.password)

        opener = urllib2.build_opener(auth_handler)
        urllib2.install_opener(opener)

class TwitterResponse():

    def __init__(self, content):
        self.content = content
        self._node = xml.dom.minidom.parseString(content)

    def get(self, name='id'):
        return self._get_data(name)

    def _get_data(self, nodename, node = None):
        if node is None : node = self._node
        elm = node.getElementsByTagName(nodename)
        return elm[0].firstChild.data.rstrip().lstrip() if elm else ''

class TwitterError(Exception):
    """Used to indicate that an error occurred when trying to access Twitter via its API."""

class TwitterRequestError(TwitterError):
    """Used to indicate that an error occurred when trying to access Twitter via its API."""

class TwitterResponseError(TwitterError):
    """Used to indicate that an error occurred when trying to access Twitter via its API."""



class Bitly:

    def __init__(self, login, api_key):
        self.uri = 'http://api.bit.ly'
        self.end_point = ''
        self.version = '2.0.1'
        self.login = login
        self.api_key = api_key

    def shorten(self, url):
        self.end_point = '/shorten'
        options = {'longUrl': url, 'format': 'xml'}
        return self.send_request(options)

    def send_request(self, options={}):
        options['login'] = self.login
        options['apiKey'] = self.api_key
        options['version'] = self.version

        if not options.has_key('history'):
            # history=1というパラメータがないとbit.lyのHistoryで見ることができない
            options['history'] = '1'

        apiurl = self.uri + self.end_point + '?' + urllib.urlencode(options)

        response = urllib2.urlopen(apiurl)

        try:
            if (response.code != 200):
                e = 'bit.ly response: code >> %d, message > %s' % (response.code, response.msg)
                logging.error(e)
                raise BitlyRequestError, e

            return BitlyResponse(response.read())
        finally:
            response.close()

class BitlyResponse():

    def __init__(self, content):
        self.content = content
        self._node = xml.dom.minidom.parseString(content)
        self.status = self._get_data('statusCode')

    def get(self):
        if self.has_error(): raise BitlyResponseError, self.get_error()
        return self._get_data('shortUrl')

    def has_error(self):
        return self.status == 'ERROR'

    def get_error(self):
        return self._get_data('errorMessage')

    def _get_data(self, nodename, node = None):
        if node is None : node = self._node
        elm = node.getElementsByTagName(nodename)
        return elm[0].firstChild.data.rstrip().lstrip() if elm else ''

class BitlyError(Exception):
    """Used to indicate that an error occurred when trying to access bit.ly via its API."""

class BitlyRequestError(BitlyError):
    """Used to indicate that an error occurred when trying to access bit.ly via its API."""

class BitlyResponseError(BitlyError):
    """Used to indicate that an error occurred when trying to access bit.ly via its API."""

使い方はこんな感じです。

twit = Twitter('Your Twitter Login', 'Your Twitter Password')
twit.bitly_account('Your bit.ly Login', 'Your bit.ly API Key')
twit_res = twit.post(body='コメント', url='http://www.example.com')
# body と url にはてブから投げられたコメントとURLを指定すれば、はてブとTwitterを連動できます