最初にどうゆうBotを作ろうかと考えた時、いつも勉強会の参加時にお世話になっているATNDの新着イベントをつぶやく@atnderが頭に浮かんだ。ATND以外に「こくちーず」というサイトでもちょっとだけIT系の勉強会の告知/募集が行われている。では、@atnderの「こくちーず」版を作ってみたらどうだろうか。というわけで、Botの大枠は決まった。
次に、Botの要件をつめていこう。ATNDはAPIが公開されていて、開催イベントについて色々な情報の取得が容易になっている。一方こくちーずの方は、新着情報のRSSがある程度。まずは「新着イベントをRSSから取得し、イベント名と告知ページのURLをTweetするBot」とし、「URLはBit.lyの短縮URLに変換する」とした。
しばらくローカル環境でごにょごにょ試した結果。以下の環境に落ち着いた。
- Google App Engine Python
- simplejson 2.0.9
- Universal feed parser 4.1
- Twython 1.0
- Bit.ly API アカウントページにてAPI Keyを確認しておく
あと、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 minutesCronの間隔は更新頻度と負荷を考えて、とりあえず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」(ソース)』にも上げてあります