最初にどうゆう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」（ソース）』にも上げてあります
投稿
0 件のコメント:
コメントを投稿