2010/01/03

Google App Engineで動かすTwitter Botをつくってみた

先日(もう去年ですが)「BOTつくろう会#5」の勉強会に参加させていただいた。@tetsunosukeさんの初心者教室でPythonでのBot作成について教えていただいたので、そのままPythonでBot作りに挑戦してみた。

最初にどうゆうBotを作ろうかと考えた時、いつも勉強会の参加時にお世話になっているATNDの新着イベントをつぶやく@atnderが頭に浮かんだ。ATND以外に「こくちーず」というサイトでもちょっとだけIT系の勉強会の告知/募集が行われている。では、@atnderの「こくちーず」版を作ってみたらどうだろうか。というわけで、Botの大枠は決まった。

次に、Botの要件をつめていこう。ATNDはAPIが公開されていて、開催イベントについて色々な情報の取得が容易になっている。一方こくちーずの方は、新着情報のRSSがある程度。まずは「新着イベントをRSSから取得し、イベント名と告知ページのURLをTweetするBot」とし、「URLはBit.lyの短縮URLに変換する」とした。

しばらくローカル環境でごにょごにょ試した結果。以下の環境に落ち着いた。
※開発はMacでやっている。PythonはMacPortsで2.6.xも入れていたが、Mac OS Xに標準で入っているVer.2.5.1(10.5の場合)でやることにした。

あと、Feedの取得/パースには結構時間がかかるようで、全ての処理をひとまとめにすると30秒制限に引っ掛かることがあったため、「Feedの取得からDataStoreの登録」「DataStoreから情報を取得しTweet」という2つの処理に分け、それぞれにCronを設定する構成にした。

ファイル構成はこんな感じ
kokucheese_bot
├app.yaml      …自動生成されたファイルを編集
├cron.yaml     …自動生成されたファイルを編集
├index.yaml    …自動生成されたファイルを編集
├default.py    …新規作成
├get_feed.py   …新規作成
├tweet.py      …新規作成
├feedparser.py …DLしたファイル
├simplejson/   …DLしたフォルダ
└twython/      …DLしたフォルダ

app.yaml
application: アプリケーション名
version: 1
runtime: python
api_version: 1

handlers:
- url: /
  script: default.py
  
- url: /get_feed
  script: get_feed.py
  
- url: /tweet
  script: tweet.py
  
- url: /.*
  script: default.py
※cronで叩けるようにプログラムとURLを関連付ける。

cron.yaml
cron:
- description: get_feed job
  url: /get_feed
  schedule: every 30 minutes

- description: tweet job
  url: /tweet
  schedule: every 30 minutes
Cronの間隔は更新頻度と負荷を考えて、とりあえず30分間隔に。これで最大1時間以内にはTweetされることになるはず。
※頻繁にFeedを取得しにいくのも悪いかなと思ってしまうのですが、Googleリーダーとかってどのくらいの間隔で取得しにいくものですかねぇ。

default.py
print "Content-Type: text/plain"
print ""
print "kokucheese_bot"
※AppEngineのURLに / でアクセスしたり、存在しないURLでアクセスしたりすると表示される。
※このBotにはWebページ的なものが必要ないので、表示はすべてprint文で手抜きした。

get_feed.py
# -*- coding: utf-8 -*-
import sys, os, re, urllib, urllib2
from datetime import *
import feedparser
import simplejson
from google.appengine.ext import db


#データモデルの定義
class EntryDataModel(db.Model):
    increment = db.IntegerProperty()   #登録数(連番)
    title     = db.StringProperty()    #イベントタイトル
    link      = db.StringProperty()    #告知ページのURL
    date      = db.StringProperty()    #Feedに登録された日時
    short_url = db.StringProperty()    #短縮済みURL
    created   = db.DateTimeProperty()  #DataStoreに登録された日時
    tweeted   = db.IntegerProperty()   #Tweet済みフラグ


#現在の登録数を返す
def get_inc_num():
    if 0 != EntryDataModel.all().count(limit=1):
        inc_num = EntryDataModel.all().order('-increment').get()
        return inc_num.increment
    else:
        return 0


#短縮URL化(bit.ly API)
def get_short_url(url):
    info_url  = "http://api.bit.ly/%s?version=2.0.1&%s=%s&login=bit.lyログイン名&apiKey=APIキー"
    url_data  = urllib2.urlopen(info_url % ("shorten", "longUrl", urllib2.quote(url))).read() #urllib2.quote(url)…パラメータ付URLもあるのでURLエンコードする
    url_info  = simplejson.loads(url_data)
    return url_info["results"][url]["shortUrl"]


#結果を表示
def check():
    request_all_query = EntryDataModel.all()
    result_set = request_all_query.order('-created').fetch(limit=50) #最新の50件
    for rs in result_set:
        print "No. "     + str(rs.increment).encode("UTF-8")
        print "Title: "  + rs.title.encode("UTF-8")
        print "Link: "   + rs.link.encode("UTF-8")
        print "Update: " + rs.date.encode("UTF-8")
        print "Short: "  + rs.short_url.encode("UTF-8")
        print "Create: " + str(rs.created).encode("UTF-8")
        print "Tweet: "  + str(rs.tweeted).encode("UTF-8")
        print ""


#メイン処理
def main():
    print "Content-Type: text/plain"
    print ""
    
    #情報入手元Feed
    feed_urls = [
        "http://kokucheese.com/main/rss/",
        #"",
        #"",
        ]
    
    #Feedから各エントリを取得し処理
    for feed_url in feed_urls:
        feed_data = feedparser.parse(feed_url) #Feedをパース
        
        #エントリの順番を並び替える
        reverse_ent = []
        reverse_ent = feed_data.entries
        reverse_ent.reverse()
        
        #現在の登録数を取得
        num = get_inc_num()
        
        for ent in reverse_ent:
            try:
                #同じタイトル及びURLが登録されているかCheck
                registed_check_query = EntryDataModel.all().filter('title =', ent.title).filter('link =', ent.link).count(limit=100)
                
                #登録がなければ、登録
                if 0 == registed_check_query:
                    t = ent.title
                    u = get_short_url(ent.link)
                    
                    num += 1
                    
                    #データストアに登録
                    feed_source = EntryDataModel()
                    feed_source.increment = num
                    feed_source.title     = t
                    feed_source.link      = ent.link
                    feed_source.date      = ent.updated
                    feed_source.short_url = u
                    feed_source.created   = datetime.utcnow() + timedelta(hours=9)
                    feed_source.tweeted   = 0
                    feed_source.put()
                    
            except:
                print "skip: " + ent.title.encode("UTF-8")
                
    check()


if __name__ == "__main__":
    main()
※Feedの取得からDataStoreへの登録
※参考にしたサイト:bit.lyのAPIを試してURLを短縮してみた
※複数のFeedを登録可能。でも増やし過ぎると30秒で完了しなくなる。そうなってくると Task Queue 使うとかロジックの見直しが必要。
※Feedのパースは Universal feed parser を使えば楽チン。GAE/Jだと ROME が動かないっぽいので、自前パーサを作るとかしないといけないかも。でもJavaでXMLを扱うならGroovyで書く方が良いと思う。
※トランザクションの使用については今後の課題。

tweet.py
# -*- coding: utf-8 -*-
import sys, os, re, urllib, urllib2
import simplejson
import twython
from google.appengine.ext import db


#データモデルの定義
class EntryDataModel(db.Model):
    increment = db.IntegerProperty()   #登録数(連番)
    title     = db.StringProperty()    #イベントタイトル
    link      = db.StringProperty()    #告知ページのURL
    date      = db.StringProperty()    #Feedに登録された日時
    short_url = db.StringProperty()    #短縮済みURL
    created   = db.DateTimeProperty()  #DataStoreに登録された日時
    tweeted   = db.IntegerProperty()   #Tweet済みフラグ


#結果を表示
def check():
    request_all_query = EntryDataModel.all()
    result_set = request_all_query.order('-created').fetch(limit=50) #最新の50件
    for rs in result_set:
        print ""
        print "Title: " + rs.title.encode("UTF-8")
        print "Tweet: " + str(rs.tweeted).encode("UTF-8")


#メイン処理
def main():
    print "Content-Type: text/plain"
    print ""
    
    #未Tweetのエントリを検索
    tweeted_check_query = EntryDataModel.all()
    untweet_set = tweeted_check_query.filter('tweeted =', 0)
    for tw in untweet_set:
        #Tweetする
        message = tw.title + u" が登録されたよ " + tw.short_url #twythonにはunicode文字列を渡す。日本語で140文字OK
        twitter = twython.core.setup(username="Twitterユーザー名", password="Twitterパスワード") #Twython v1.0
        twitter.updateStatus(message)
        
        #Tweet済みフラグを更新
        tw.tweeted = 1
        tw.put()
    
    check()


if __name__ == "__main__":
    main()
※DataStroreから未Tweet分をTweet
※Twython Ver.0.8では日本語140文字分をTweetするにはソースの修正が必要だが、Ver.1.0では必要なし。

以上のファイルをAppEngineにアップロードすればBotとして動くはずです。

RSSの他にもGoogleアラートや以前の投稿Twitterのつぶやきをキーワード検索させる等、Feedとして得られる情報はたくさんあるので、今回のBot以外にも使い方はいろいろありそうです。

今後はatnderみたいにリマインダー機能とか追加していきたいですね。

完成したBotはこちらkokucheese_bot

※そもそも勘違いしている箇所や、よりより書き方があったりする知れませんがご容赦ください。またその場合はご指摘いただければ幸いです。

※現在はTwitter側の仕様変更によりBasic認証が使えなくなりました。上記のコードそのままでは投稿できません。

※GAE+Twitter+OAuth関連のソースは『TwitterとGoogle App Engineの自分用Webアプリケーション「TwitMail」(ソース)』にも上げてあります