python-box:appengine-twitter
文書の過去の版を表示しています。
目次
Google App Engineで手軽にTwitterアプリを作成!(OAuthにも対応!) - AppEngine-Twitter
更新履歴とコメントはAppEngine-Twitter Archive - ゼロと無限の間のログへどうぞ。
Google App EngineでTwitter APIを操作するライブラリを作ってみた。
車輪の再発明ではなく、趣味の車輪作り。
ライセンス
MITライセンスで。
デモ(動作サンプル)
※OAuthを使っているデモについてはGoogle App Engineで手軽にOAuthアプリを作成!(Twitterとか!) - AppEngine-OAuthを参照
その1. RT(ReTweet)の多いつぶやきを報告するBot
その2. 地名を告げるとその近くにいるTwitterユーザーの名前を教えてくれるBot
その3. ユーザー名・パスワードを入力してTwitter APIの主な操作を一通りするデモ
ユーザー名・パスワードを入力してTwitter APIの主な操作を一通りするデモ
ソースコード
(※ページ表示でGoogle App EngineのRequestHandlerを拡張+ちょっと便利な関数群 - AppEngine-BaseHandlerを使っている。それとOAuth使っていない場合でも、Google App Engineで手軽にOAuthアプリを作成!(Twitterとか!) - AppEngine-OAuthも無いと動かないかも。)
#!/usr/bin/env python # -*- coding: UTF-8 -*- ''' Sample for AppEngine-Twitter on Google App Engine See: http://0-oo.net/sbox/python-box/appengine-twitter ''' import logging import wsgiref.handlers from appengine_twitter import AppEngineTwitter from basehandler import BaseHandler, h from google.appengine.ext import webapp class MainHandler(BaseHandler): def get(self): self.show() def post(self): self.show(self.request.get('name'), self.request.get('pswd')) def show(self, name='', pswd=''): self.simple_header(u'AppEngine-Twitterのデモ') self.p(u'<h1>AppEngine-Twitterのデモ</h1>') self.p(u'<h2>Twitterログイン情報を入力してください</h2>') self.p('<form method="post" action="./">') self.p(u'ユーザー名 ') self.p('<input type="text" name="name" value="' + h(name) + '">') self.p(u' パスワード ') self.p('<input type="password" name="pswd"' + ' value="' + h(pswd) + '">') self.p(' <input type="submit">') self.p('</form>', True) self.p('<br>') if name and pswd: self.test(name, pswd) self.p('<hr>', True) self.p(u'このページは') self.p('<a href="http://0-oo.net/sbox/python-box/appengine-twitter">') self.p('AppEngine-Twitter') self.p(u'</a>のデモです') self.simple_footer() def test(self, name, pswd): twitter = AppEngineTwitter(name, pswd) self.p(u'<h2>実行結果</h2>', True) self.p(u'<b>verify()</b> ユーザー名とパスワードの確認(200ならOK)', True) self.p(twitter.verify(), True) self.p('<br>') self.p(u'<b>search()</b> "腹減った"を検索(3件まで)', True) results = twitter.search(u'腹減った'.encode('utf8'), {'rpp': 3}) for result in results: self.p(result['text'], True) self.p('<br>') self.p(u'<b>is_following()</b> @uresuji_booksをフォローしてる?', True) if twitter.is_following('uresuji_books'): self.p('True') else: self.p('False') self.p('<br>', True) note = u'<b>follow()</b> @uresuji_booksをフォローする' note += u'(200なら成功、403なら既にフォロー済み)' self.p(note, True) self.p(twitter.follow('uresuji_books'), True) self.p('<br>') self.p(u'<b>update()</b> つぶやく(200なら成功。自分のTLを見てみてね)', True) msg = 'AppEngine-Twitter (on Python and Google App Engine) ' msg += u'http://0-oo.appspot.com/twitter/ からこんにちは!' self.p(twitter.update(msg.encode('utf8')), True) self.p('<br>') logging.getLogger() routing = [('/twitter/.*', MainHandler)] application = webapp.WSGIApplication(routing, debug=False) wsgiref.handlers.CGIHandler().run(application)
AppEngine-Twitterのソースコード
#!/usr/bin/env python # -*- coding: utf-8 -*- ''' AppEngine-Twitter Twitter API wrapper for applications on Google App Engine See: http://0-oo.net/sbox/python-box/appengine-twitter License: http://0-oo.net/pryn/MIT_license.txt (The MIT license) See also: http://apiwiki.twitter.com/Twitter-API-Documentation http://code.google.com/intl/ja/appengine/docs/python/urlfetch/ ''' __author__ = 'dgbadmin@gmail.com' __version__ = '0.1.0' import base64 import urllib from appengine_oauth import AppEngineOAuth from django.utils import simplejson from google.appengine.api import urlfetch class AppEngineTwitter(object): def __init__(self, tw_name='', tw_pswd=''): ''' Note: Some actions require password or OAuth. ''' self._api_url = 'https://twitter.com' self._search_url = 'http://search.twitter.com' self.tw_name = tw_name self._oauth = None self._headers = {} if tw_pswd != '': auth = base64.encodestring(tw_name + ':' + tw_pswd)[:-1] self._headers['Authorization'] = 'Basic ' + auth def update(self, message): ''' Post a tweet Sucess => Retrun 200 / Fialed => Return other HTTP status ''' return self._post('/statuses/update.json', {'status': message}) def follow(self, target_name): ''' Sucess => Return 200 / Already following => Return 403 / Fialed => Return other HTTP status ''' return self._post('/friendships/create.json', {'screen_name': target_name}) def is_following(self, target_name): ''' Yes => Return True / No => Return False / Fialed => Return HTTP status except 200 ''' if self.tw_name == '': # With OAuth, screen_name is not required. self.verify() user_info = simplejson.loads(self.last_response.content) self.tw_name = user_info['screen_name'] status = self._get('/friendships/exists.json', {'user_a': self.tw_name, 'user_b': target_name}) if status == 200: return (self.last_response.content == 'true') else: return status def verify(self): ''' Verify user_name and password, and get user info Sucess => Return 200 / Fialed => Return other HTTP status ''' return self._get('/account/verify_credentials.json', {}) def search(self, keyword, params={}): ''' Sucess => Return Array of dict / Fialed => Return HTTP status except 200 ''' params['q'] = keyword return self._search('/search.json', params) # OAuth methods # (See http://0-oo.net/sbox/python-box/appengine-oauth ) def set_oauth(self, key, secret, acs_token='', acs_token_secret=''): ''' Set OAuth parameters ''' self._oauth = AppEngineOAuth(key, secret, acs_token, acs_token_secret) def prepare_oauth_login(self): ''' Get request token, request token secret and login URL ''' dic = self._oauth.prepare_login(self._api_url + '/oauth/request_token/') dic['url'] = self._api_url + '/oauth/authorize?' + dic['params'] return dic def exchange_oauth_tokens(self, req_token, req_token_secret): ''' Exchange request token for access token ''' return self._oauth.exchange_tokens(self._api_url + '/oauth/access_token/', req_token, req_token_secret) # Private methods def _post(self, path, params): url = self._api_url + path if self._oauth != None: params = self._oauth.get_oauth_params(url, params, 'POST') res = urlfetch.fetch(url=url, payload=urllib.urlencode(params), method='POST', headers=self._headers) self.last_response = res return res.status_code def _get(self, path, params): url = self._api_url + path if self._oauth != None: params = self._oauth.get_oauth_params(url, params, 'GET') url += '?' + urllib.urlencode(params) res = urlfetch.fetch(url=url, method='GET', headers=self._headers) self.last_response = res return res.status_code def _search(self, path, params): ''' FYI http://apiwiki.twitter.com/Rate-limiting (Especially 503 error) ''' url = url=self._search_url + path + '?' + urllib.urlencode(params) res = urlfetch.fetch(url=url, method='GET') self.last_response = res if res.status_code == 200: return simplejson.loads(res.content)['results'] elif res.status_code == 503: err_msg = 'Rate Limiting: Retry After ' + res.headers['Retry-After'] else: err_msg = 'Error: HTTP Status is ' + str(res.status_code) raise Exception('Twitter Search API ' + err_msg)
python-box/appengine-twitter.1266038179.txt.gz · 最終更新: 2010/02/13 14:16 by dgbadmin