ゼロと無限の間に

フリーでオープンソースなJavaScriptとかPHPとか。

ユーザ用ツール

サイト用ツール


python-box:appengine-twitter

文書の過去の版を表示しています。


Google App Engineで手軽にTwitterアプリを作成!(OAuthにも対応!) - AppEngine-Twitter

Google App EngineでTwitter APIを操作するライブラリを作ってみた。
車輪の再発明ではなく、趣味の車輪作り。

ライセンス

デモ(動作サンプル)

その1. RT(ReTweet)の多いつぶやきを報告するBot

その2. ユーザー名・パスワードを入力してTwitter APIの主な操作を一通りするデモ

ユーザー名・パスワードを入力してTwitter APIの主な操作を一通りするデモ

ソースコード(※Google App EngineのRequestHandlerを拡張+ちょっと便利な関数群 - AppEngine-BaseHandlerを使っている)

#!/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.1247322679.txt.gz · 最終更新: 2009/07/11 23:31 by dgbadmin

Donate Powered by PHP Valid HTML5 Valid CSS Driven by DokuWiki