TwitterのOAuthでログインできるようにしておくと色々便利なのはAppEngineで経験済みなので，そのためのライブラリを作成してみる．
基本的にはこちらのページを参考にこしらえた．
pythonでoauth認証
事前準備は，
さくらインターネットでPython2.7を使ってTwitterAPIを叩くを参考のこと．
なお，Twitterとの接続に使うのは，
python-twitter．
まず，
http://syncer.jp/twitter-api-create-application
を参考に，Twitterのアプリを作成しておく．
ただし，アプリ作成の場所が，分かりづらくなったので，直接こちらにアクセスした方がよい．
https://apps.twitter.com/
Twitter社はアプリ作らせる気がないのか？と思うくらい分かりづらい．
このとき，CallBackURLにアプリのCGIのURLを指定しておくことを忘れずに．
後から変えられるけど．
次に，アプリ作成に入る．
まず，sqliteを使うので，oauth.dbというファイルを作っておいて，
$ sqlite3 oauth.db
sqlite> CREATE TABLE oauth (oauth_token text primary key,oauth_token_secret text);
sqlite> CREATE TABLE user (id integer primary key, consumer_key text, consumer_secret text, access_token text, access_token_secret text, screen_name text, image_url text, time datetime);
sqlite> .exit
として，データベースを作成しておく．
次に，TwitterLogin.pyを作成．
ここが今日のハイライト．
#!/usr/bin/env python
# -*- coding: utf-8 -*-
import site
site.addsitedir('/home/username/lib/python')
import os
import cgi
import cgitb
import oauth2 as oauth
import sqlite3
import twitter
import datetime
from Cookie import SimpleCookie
from twitter import TwitterError
request_token_url = 'https://api.twitter.com/oauth/request_token'
access_token_url = 'https://api.twitter.com/oauth/access_token'
authenticate_url = 'https://api.twitter.com/oauth/authorize'
class TwitterLogin:
def __init__(self, db_name, consumer_key, consumer_secret):
self.db_name = db_name
self.consumer_key = consumer_key
self.consumer_secret = consumer_secret
def getOAuth(self):
consumer = oauth.Consumer(key=self.consumer_key, secret=self.consumer_secret)
client = oauth.Client(consumer)
# reqest_token
resp, content = client.request(request_token_url, 'GET')
request_token = dict(self.parse_qsl(content))
#call_back="http://torix.sakura.ne.jp/twitter/signin.cgi"
# request_token
con = sqlite3.connect(self.db_name)
con.execute(u'insert into oauth values (?, ?)', (request_token['oauth_token'], request_token['oauth_token_secret']))
con.commit()
con.close()
#
url = '%s?oauth_token=%s' % (authenticate_url, request_token['oauth_token'])
print 'Content-type: text/html; charset: utf-8\n\n'
print
print '' % url
def getApi(self):
access_token, access_token_secret, screen_name, image_url = self.loadAccessToken()
self.screen_name = screen_name
self.image_url = image_url
if not access_token == None:
# Cookieを使った認証
api = twitter.Api(consumer_key=self.consumer_key,
consumer_secret=self.consumer_secret,
access_token_key=access_token,
access_token_secret=access_token_secret)
return api
elif 'QUERY_STRING' in os.environ:
# CallBackを使った認証
query = cgi.parse_qs(os.environ['QUERY_STRING'])
if len(query) == 0:
return None
# oauth_token_secret を取得
con = sqlite3.connect(self.db_name)
oauth_token_secret = con.execute(
u'select oauth_token_secret from oauth where oauth_token = ?'
, [query['oauth_token'][0]]).fetchone()[0]
con.close()
if oauth_token_secret == None:
return None
# Access_token と access_token_secret を取得
consumer = oauth.Consumer(key=self.consumer_key, secret=self.consumer_secret)
token = oauth.Token(query['oauth_token'][0], query['oauth_verifier'][0])
client = oauth.Client(consumer, token)
resp, content = client.request(
access_token_url, "POST", body="oauth_verifier=%s" % query['oauth_verifier'][0])
access_token = dict(self.parse_qsl(content))
# access_tokenの保存
api = twitter.Api(consumer_key=self.consumer_key,
consumer_secret=self.consumer_secret,
access_token_key=access_token['oauth_token'],
access_token_secret=access_token['oauth_token_secret'])
try:
user = api.VerifyCredentials()
uid = user.GetId()
self.screen_name = user.screen_name
self.image_url = user.profile_image_url
self.saveAccessToken(uid, access_token['oauth_token'], access_token['oauth_token_secret'], self.screen_name, self.image_url)
except TwitterError:
return None
return api
else:
return None
#save access token to db
def saveAccessToken(self, uid, access_token, access_token_secret, screen_name, image_url):
con = sqlite3.connect(self.db_name)
c = con.cursor()
#print datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")
c.execute(u"replace into user (id, consumer_key, consumer_secret, access_token, access_token_secret, screen_name, image_url, time) values (?, ?, ?, ?, ?, ?, ?, datetime(?))", [uid, self.consumer_key, self.consumer_secret, access_token, access_token_secret, screen_name, image_url, datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")])
con.commit()
c.close()
ck = SimpleCookie()
ck.load(os.environ.get("HTTP_COOKIE",""))
ck["id"] = uid
ck["token"] = access_token
now_date = datetime.datetime.now()
expires = now_date
expires = expires + datetime.timedelta(seconds=+(60*60*24))
expires = expires.strftime('%a, %d-%b-%Y %H:%M:%S %Z')
ck["id"]["expires"] = expires
ck["token"]["expires"] = expires
print ck.output(["id", "expires"])
print ck.output(["token", "expires"])
#load data from data base
def loadAccessToken(self):
ck = SimpleCookie()
ck.load(os.environ.get("HTTP_COOKIE",""))
if ck.has_key("id") and ck.has_key("token"):
id = int(ck["id"].value)
token = ck["token"].value
access_token = None
con = sqlite3.connect(self.db_name)
c = con.cursor()
c.execute(u'select access_token, access_token_secret, screen_name, image_url from user where id=%d'% id)
for row in c:
access_token = row[0]
access_token_secret = row[1]
self.screen_name = row[2]
self.image_url = row[3]
con.close()
if not access_token == None and access_token == token:
return access_token, access_token_secret, self.screen_name, self.image_url
else:
return None, None, None, None
else:
return None, None, None, None
def parse_qsl(url):
param = {}
for i in url.split('&'):
_p = i.split('=')
param.update({_p[0]: _p[1]})
return param
def parse_qsl(self, url):
param = {}
for i in url.split('&'):
_p = i.split('=')
param.update({_p[0]: _p[1]})
return param
getOAuthで，OAuth認証を行うページに飛ばして，callbackで戻ってきたときにgetApiで捕まえる．
それ以降は，UserIDとAccessTokenをCookieに保存することでSession管理をすることにする．
本当は独自のsessionIDを発行した方がいいのかもしれない．
次に，実際にログインするところを作ってみよう．
使い方は以下の通り．
なお，consumer_keyとconsumer_secretはTwitterで登録したアプリのものをつかう．
callback urlがhttp://xxx.xx.xx/twitter/login.cgi
そのURLでアクセスできるcgiを以下のように作成する．
#!/usr/bin/env python
# -*- coding: utf-8 -*-
import site
site.addsitedir('/home/username/lib/python')
from TwitterLogin import TwitterLogin
consumer_key = 'consumer_key'
consumer_secret = 'consumer_secret'
if __name__ == '__main__':
twitter = TwitterLogin('oauth.db', consumer_key, consumer_secret)
api = twitter.getApi()
print 'Content-type: text/html; charset: utf-8\n\n'
if api == None:
print 'Fail to login'
else:
user = api.VerifyCredentials()
print user.screen_name
これで，認証したアカウントのスクリーンネームが表示されるはず．
APIの使い方は，以下を参照．
https://python-twitter.googlecode.com/hg/doc/twitter.html
色々問題はあるかもしれないけど，とりあえず動いたので，満足．
投稿
