| " | ⇒ | " |
| " | ⇒ | " |
| & | ⇒ | & |
| & | ⇒ | & |
| ' | ⇒ | ' |
| ' | ⇒ | ' |
| < | ⇒ | < |
| < | ⇒ | < |
| > | ⇒ | > |
| > | ⇒ | > |
application: アプリ名 version: 1 runtime: python api_version: 1 handlers: - url: /main script: get_timeline.py - url: /send script: send_tweet.py - url: /db/delete script: db_delete.py - url: .* script: default.py
cron: - description: Get Timeline job url: /main schedule: every 10 minutes - description: Send Tweet job url: /send schedule: every 15 minutes - description: DB Delete job url: /db/delete schedule: every 60 minutes
# -*- coding: utf-8 -*-
import sys, os, re, urllib, urllib2
import simplejson
import tweepy
from datetime import *
from google.appengine.ext import webapp
from google.appengine.ext.webapp.util import run_wsgi_app
from google.appengine.ext import db
#OAuth
CONSUMER_KEY = 'CONSUMER_KEY' #自分で取得した値を入れる
CONSUMER_SECRET = 'CONSUMER_SECRET' #自分で取得した値を入れる
ACCESS_TOKEN = 'ACCESS_TOKEN' #自分で取得した値を入れる
ACCESS_TOKEN_SECRET = 'ACCESS_TOKEN_SECRET' #自分で取得した値を入れる
#データモデルの定義
class TimelineData(db.Model):
icon_url = db.StringProperty() #アイコン画像URL
scr_name = db.StringProperty() #スクリーン名
usr_name = db.StringProperty() #ユーザ名
tweet = db.StringProperty(multiline=True) #ツイート
status_id = db.StringProperty() #ステータスID
reply_id = db.StringProperty() #リプライID
created_at = db.StringProperty() #ツイートされた日時
ds_created = db.DateTimeProperty() #DataStoreに登録された日時
sent = db.IntegerProperty() #送信済みフラグ
schema_ver = db.IntegerProperty() #スキーマ・バージョン
# メイン処理
class GetTimeline(webapp.RequestHandler):
def get(self):
#HomeTimelineの取得
auth = tweepy.OAuthHandler(CONSUMER_KEY, CONSUMER_SECRET)
auth.set_access_token(ACCESS_TOKEN, ACCESS_TOKEN_SECRET)
api = tweepy.API(auth)
home_timeline = api.home_timeline(count=50)
#TimeLineの並び順を逆順に変更
home_timeline.reverse()
for tw in home_timeline:
try:
#同じステータスIDが登録されているかCheck
registed_check_query = TimelineData().all().order('-ds_created').filter('status_id =', str(tw.id).encode("UTF-8")).get()
print registed_check_query
#登録がなければ、登録
if None == registed_check_query:
#データストアに登録
td = TimelineData()
td.icon_url = tw.user.profile_image_url #アイコン画像URL
td.scr_name = tw.user.screen_name #スクリーン名
td.usr_name = tw.user.name #ユーザ名
td.tweet = tw.text #ツイート
td.status_id = str(tw.id).encode("UTF-8") #ステータスID
td.reply_id = str(tw.in_reply_to_status_id).encode("UTF-8") #リプライID
td.created_at = str(tw.created_at).encode("UTF-8") #ツイートされた日時
td.ds_created = datetime.utcnow() + timedelta(hours=9) #DataStoreに登録された日時
td.sent = 0 #送信済みフラグ、0は未送信
td.schema_ver = 2 #スキーマ・バージョン
td.put()
print "▼Check Status"
print td.icon_url.encode("UTF-8")
print td.scr_name.encode("UTF-8")
print td.usr_name.encode("UTF-8")
print td.tweet.encode("UTF-8")
print td.status_id.encode("UTF-8")
print td.reply_id.encode("UTF-8")
print td.created_at.encode("UTF-8")
print str(td.ds_created).encode("UTF-8")
print str(td.sent).encode("UTF-8")
print str(td.schema_ver).encode("UTF-8")
print "△Save DataStore\n"
else:
print "△Skip DataStore\n"
except:
print "△NG DataHandling\n"
application = webapp.WSGIApplication(
[
('/main', GetTimeline),
],
debug=True)
def main():
run_wsgi_app(application)
if __name__ == "__main__":
main()
# -*- coding: utf-8 -*-
import sys, os, re, urllib, urllib2
from google.appengine.ext import webapp
from google.appengine.ext.webapp.util import run_wsgi_app
from google.appengine.ext import db
import base64
from google.appengine.api import mail
from dateutil import parser
from dateutil import relativedelta
#データモデルの定義
class TimelineData(db.Model):
icon_url = db.StringProperty() #アイコン画像URL
scr_name = db.StringProperty() #スクリーン名
usr_name = db.StringProperty() #ユーザ名
tweet = db.StringProperty(multiline=True) #ツイート
status_id = db.StringProperty() #ステータスID
reply_id = db.StringProperty() #リプライID
created_at = db.StringProperty() #ツイートされた日時
ds_created = db.DateTimeProperty() #DataStoreに登録された日時
sent = db.IntegerProperty() #送信済みフラグ
schema_ver = db.IntegerProperty() #スキーマ・バージョン
#メイン処理
class SendTweet(webapp.RequestHandler):
def get(self):
not_send = TimelineData().all().order('ds_created').filter('sent =', 0).fetch(limit=200)
#print not_send
count = 0 #ツイート数
start = "" #最初のツイート
body = "" #ツイートの要約
#"http://twitter.com/#!/"
if not_send:
for ns in not_send:
#時刻を変換
d = parser.parse(ns.created_at)
jst = d + relativedelta.relativedelta(hours=+9)
#print jst
try:
body += '' + "\n"
body += '![]()
' + "\n"
body += '' + ns.scr_name.encode("UTF-8") + ' (' + ns.usr_name.encode("UTF-8") + ')' + '' + "\n"
body += '' + str(jst).encode("UTF-8") + '' + "\n"
body += ''
body += '' + ns.tweet.encode("UTF-8") + '' + "\n"
body += 'http://mobile.twitter.com/' + ns.scr_name.encode("UTF-8") + '/status/' + ns.status_id.encode("UTF-8") + '' + "\n"
if 'None' != ns.reply_id:
body += ''
body += '' + "\n"
body += '
' + "\n"
body += "\n"
count += 1
if count == 1 :
#start = ns.created_at.encode("UTF-8")
start = str(jst).encode("UTF-8")
ns.sent = 1 #送信済みフラグの変更
ns.put()
print str(ns.sent).encode("UTF-8")
except:
print count
print ns.sent
#print body
subject = start + " から " + str(count).encode("UTF-8") + "件";
print subject
fromAddr = "from: メールアドレス";
toAddr = "to: メールアドレス";
html_h = "\n\n"
html_f = "" + "\n"
html_body = ""
html_body += html_h
html_body += body
html_body += html_f
#print html_body
mail.send_mail(fromAddr, toAddr, subject, body=" ", html=html_body)
else:
print "No Data"
application = webapp.WSGIApplication(
[
('/send', SendTweet),
],
debug=True)
def main():
run_wsgi_app(application)
if __name__ == "__main__":
main()
# -*- coding: utf-8 -*-
import sys, os, re, urllib, urllib2
from google.appengine.ext import webapp
from google.appengine.ext.webapp.util import run_wsgi_app
from google.appengine.ext import db
#データモデルの定義
class TimelineData(db.Model):
icon_url = db.StringProperty() #アイコン画像URL
scr_name = db.StringProperty() #スクリーン名
usr_name = db.StringProperty() #ユーザ名
tweet = db.StringProperty(multiline=True) #ツイート
status_id = db.StringProperty() #ステータスID
reply_id = db.StringProperty() #リプライID
created_at = db.StringProperty() #ツイートされた日時
ds_created = db.DateTimeProperty() #DataStoreに登録された日時
sent = db.IntegerProperty() #送信済みフラグ
schema_ver = db.IntegerProperty() #スキーマ・バージョン
#メイン処理
class DeleteDS(webapp.RequestHandler):
def get(self):
# DataStoreの削除
deleteCheck = TimelineData.all().filter('sent =', 1).fetch(limit=500)
if deleteCheck :
for dc in deleteCheck:
try:
db.delete(dc)
print "deleted"
except:
print "no delete"
else:
print "no delete data"
application = webapp.WSGIApplication(
[
('/db/delete', DeleteDS),
],
debug=True)
def main():
run_wsgi_app(application)
if __name__ == "__main__":
main()
print "Content-Type: text/plain" print "" print "TwitMail"
kokucheese_bot ├app.yaml …自動生成されたファイルを編集 ├cron.yaml …自動生成されたファイルを編集 ├index.yaml …自動生成されたファイルを編集 ├default.py …新規作成 ├get_feed.py …新規作成 ├tweet.py …新規作成 ├feedparser.py …DLしたファイル ├simplejson/ …DLしたフォルダ └twython/ …DLしたフォルダ
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: - description: get_feed job url: /get_feed schedule: every 30 minutes - description: tweet job url: /tweet schedule: every 30 minutesCronの間隔は更新頻度と負荷を考えて、とりあえず30分間隔に。これで最大1時間以内にはTweetされることになるはず。
print "Content-Type: text/plain" print "" print "kokucheese_bot"※AppEngineのURLに / でアクセスしたり、存在しないURLでアクセスしたりすると表示される。
# -*- 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への登録# -*- 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
投稿
すべてのコメント
