株式会社ライブキャスト

3月11日、東日本大震災から2年がたちました。

今もあの時のことを思うと、熱くなるものがあります。

その当時書いたブログがこちらです。

今、私にできること。

このとき作ったプログラム、@urayasu_kohobotをTwitter API v1.1対応しました。

とは言っても、もともとXAuthで認証済の状態でAPIをコールするにしていたので、URLを変更するだけで済みました。

●他のユーザのタイムラインを読み込むAPI

url = 'http://api.twitter.com/1/statuses/user_timeline.json'

↓

url = 'http://api.twitter.com/1.1/statuses/user_timeline.json'

●statusを更新するAPI

url = 'http://api.twitter.com/1/statuses/update.json'

↓

url = 'http://api.twitter.com/1.1/statuses/update.json'

Google App Engineの無料枠内で十分稼働するので、ほぼ放置状態でしたが、もうちょっとメンテしないといけなそうです。
※ 翻訳APIはGoogle Translateが有料化されてから、Microsoft Bing APIに変更しています。ブログにはしていませんでしたが。。。

全ソースは以下のようになっています。

import logging
import os
import oauth
import re
import urllib
from google.appengine.ext import db
from google.appengine.ext.webapp import template
from google.appengine.ext.webapp import RequestHandler
from google.appengine.ext.webapp import WSGIApplication
from google.appengine.ext.webapp.util import run_wsgi_app
from google.appengine.api import urlfetch
from google.appengine.api import memcache
from django.utils import simplejson
from xml.etree.ElementTree import *

class TwitterHandler(RequestHandler):
    def __init__(self):
        twit = TwitBot()
        self.action = {
            'timeline'  :twit.timeline,
            'update'    :twit.update
            }

    def get(self):
        action = self.action.get(self.request.get('action'))
        if action:
            logging.debug(action)
            action()
        template_values = {
            'actions'   :sorted(self.action.keys()),
            'done'      :self.request.get('action'),
            'path'      :self.request.path,
            }
        path = os.path.join(os.path.dirname(__file__), 'template.html')
        self.response.out.write(template.render(path, template_values))

class TwitBot:
    def __init__(self):
        self.client = oauth.TwitterClient(
            '********************************',
            '***************************************',
            None)

    def timeline(self):
        url = 'http://api.twitter.com/1.1/statuses/user_timeline.json'
        params = {
            'screen_name'   :'urayasu_koho',
            }
        since_id = IDs.all().filter('name = ', 'since_id').get()
        if (since_id == None):
            IDs(name = 'since_id').put();
            since_id = IDs.all().filter('name = ', 'since_id').get()
        else:
            if since_id.since_id != None :
                params['since_id'] = since_id.since_id
        result = self.client.make_request(
            url,
            token = '************************************************************',
            secret = '****************************************',
            additional_params = params,
            protected = True,
            method = urlfetch.GET)
        logging.debug(result.status_code)
        if result.status_code == 200:
            statuses = simplejson.loads(result.content)
            statuses.reverse()
            for status in statuses:
                self.translate(status = re.sub('#urayasu', '', status['text']))
                since_id.since_id = status['id']
                since_id.put()

    def update(self, status = None):
        if status != None:
            url = 'http://api.twitter.com/1.1/statuses/update.json'
            if len(status) > 140:
                self.write(status)
            else:
                params = {
                    'status'    : status.encode('utf-8'),
                    'count'     :20
                    }
                result = self.client.make_request(
                    url,
                    token = '************************************************************',
                    secret = '****************************************',
                    additional_params = params,
                    protected = True,
                    method = urlfetch.POST)
                logging.debug(result.status_code)
                logging.debug(result.content)

    def translate(self, status = None):
        if status != None:
            logging.debug(status)
            url = 'http://api.microsofttranslator.com/V2/Http.svc/Translate?appId=****************************************&'+urllib.urlencode({'text':status.encode('utf-8')})+'&from=ja&to=en&contentType=text/plain' 
            result = urllib.urlopen(url).read()
            translated = fromstring(result).text
            self.update(status = translated)

    def write(self, post = None):
        if post != None:
            url = 'http://www.tumblr.com/api/write'
            opt = {
                'email'     :'mojamoja@example.com',
                'password'  :'***********',
                'type'      :'regular',
                'title'     :'',
                'body'      :post.encode('utf-8'),
                'generator' :'urayasu_kohobot'
                }
            try:
                res = urllib.urlopen(url, urllib.urlencode(opt))
            except urllib.error.HTTPError, e:
                if e.code == 200 or e.code == 201:
                    return
                else:
                    raise TumblrException(str(e))
            except Exception, e:
                raise TumblrException(str(e))

class IDs(db.Model):
    name = db.StringProperty()
    since_id = db.IntegerProperty()
    
class TumblrException:
    def __init__(self, msg):
        self._msg = msg

    def __str__(self):
        return self._msg
            
def main():
    logging.getLogger().setLevel(logging.DEBUG)
    application = WSGIApplication([
        ('/bot/', TwitterHandler),],
                                    debug = False)
    run_wsgi_app(application)

if __name__ == '__main__':
    main()

短時間でざくっと作ったので粗々にもかかわらず、今もひっそりとですがちゃんと動き続けています。