ChirpUserStreamをOAuthでPythonから
みなさんChirpUserStreamしてますか!便利ですよね!
最新のTwitter API「ChirpUserStreams」をPythonから使う - ctrlshiftの日記を使ってみたのですが、basic認証の有効期限が残り僅かなのが気になります。たしか六月中に廃止されるはず。
ということで、認証部分をoauthと入れ替えてやればいい。
最初はoauth対応Twitterライブラリのtweepyをハックしようとしてたんですが、すごく独特な記述が行われていたので断念。もっと勉強しないと。
で、Twitter API を OAuth で認証するスクリプトを 0 から書いてみた - trial and errorを参考に認証部分をOAuthに。
ちゃんと動きました。よかった。
動かすには事前にoauthの登録が必要です。
Twitter / アプリケーション
なんか適当に登録してconsumer_keyとcousumer_secret を取得
初回起動時にアクセス用のトークンを取得するためのurlがもらえるので、そこから7桁の番号を取得して打ち込みます。
自分はtweepyを使って取得しました。
tweepyでOAuthを試してみた
import tweepy consumer_token = "xxxxxxxxxxxxxx" consumer_secret = "xxxxxxxxxxxxxxxx" auth = tweepy.OAuthHandler(consumer_token, consumer_secret) redirect_url = auth.get_authorization_url() redirect_url # redirect_urlのURLにブラウザでアクセスして許可すると7桁の数字が表示されるのでそれを入力 verifier = '1234567' auth.get_access_token(verifier) key = auth.access_token.key secret = auth.access_token.secret auth = tweepy.OAuthHandler(consumer_token, consumer_secret) auth.set_access_token(key, secret) oauthapi = tweepy.API(auth) oauthapi.update_status('hello from tweepy(OAuth)')
#/usr/bin/python #-*- encoding:utf-8 -*- import time, random import urllib, urllib2 import hmac, hashlib import cgi import simplejson import time ckey = "" csecret = "" atoken ="" atoken_secret = "" def make_signature(params, url, method, csecret, secret = ""): # Generate Signature Base String plist = [] for i in sorted(params): plist.append("%s=%s" % (i, params[i])) pstr = "&".join(plist) msg = "%s&%s&%s" % (method, urllib.quote(url, ""), urllib.quote(pstr, "")) # Calculate Signature h = hmac.new("%s&%s" % (csecret, secret), msg, hashlib.sha1) sig = h.digest().encode("base64").strip() return sig def init_params(): p = { "oauth_consumer_key": ckey, "oauth_signature_method": "HMAC-SHA1", "oauth_timestamp": str(int(time.time())), "oauth_nonce": str(random.getrandbits(64)), "oauth_version": "1.0" } return p def oauth_header(params): plist = [] for p in params: plist.append('%s="%s"' % (p, urllib.quote(params[p]))) return "OAuth %s" % (", ".join(plist)) # Request Token URL reqt_url = 'http://twitter.com/oauth/request_token' # Authorize URL auth_url = 'http://twitter.com/oauth/authorize' # Access Token URL acct_url = 'http://twitter.com/oauth/access_token' # status update post_url = 'http://twitter.com/statuses/update.xml' # show friends timeline frtl_url = 'http://twitter.com/statuses/friends_timeline.xml' 'chirpstream' chirp_url ='http://chirpstream.twitter.com/2b/user.json' if not atoken and not atoken_secret: # Request Parameters params = init_params() print "Get request token:", # Generate Signature sig = make_signature(params, reqt_url, "GET", csecret) params["oauth_signature"] = sig # Get Token req = urllib2.Request("%s?%s" % (reqt_url, urllib.urlencode(params))) resp = urllib2.urlopen(req) print "\t[OK]" # Parse Token Parameters ret = cgi.parse_qs(resp.read()) token = ret["oauth_token"][0] token_secret = ret["oauth_token_secret"][0] # Get PIN print "* Please access to this URL, and allow." print "> %s?%s=%s" % (auth_url, "oauth_token", token) print "\n* After that, will display 7 digit PIN, input here." print "PIN ->", pin = raw_input() pin = int(pin) print "Get access token:", # Generate Access Token Request params = init_params() params["oauth_verifier"] = pin params["oauth_token"] = token sig = make_signature(params, acct_url, "GET", csecret, token_secret) params["oauth_signature"] = sig # Get Access Token req = urllib2.Request("%s?%s" % (acct_url, urllib.urlencode(params))) resp = urllib2.urlopen(req) print "\t[OK]" # Parse Access Token fin = cgi.parse_qs(resp.read()) atoken = fin["oauth_token"][0] atoken_secret = fin["oauth_token_secret"][0] print "Access Token: %s" % atoken print "Access Token Secret: %s" % atoken_secret print "Your screen_name is '%s'." % fin["screen_name"][0] def update(): # Update Status by OAuth Authorization print "What are you doing?:", post = 'oauth test' params = init_params() params["oauth_token"] = atoken params["status"] = urllib.quote(post, "") sig = make_signature(params, post_url, "POST", csecret, atoken_secret) params["oauth_signature"] = sig del params["status"] req = urllib2.Request(post_url) req.add_data("status=%s" % urllib.quote(post, "")) req.add_header("Authorization", oauth_header(params)) try: resp = urllib2.urlopen(req) except urllib2.HTTPError, e: print "Error: %s" % e print e.read() def getTimeline(): # Get Friends Timeline by OAuth Authorization params = init_params() params["oauth_token"] = atoken sig = make_signature(params, frtl_url, "GET", csecret, atoken_secret) params["oauth_signature"] = sig req = urllib2.Request(frtl_url) req.add_header("Authorization", oauth_header(params)) resp = urllib2.urlopen(req) # Show Timeline print resp.read() print "Done!!" # Get Stream params = init_params() params["oauth_token"] = atoken sig = make_signature(params, chirp_url, "GET", csecret, atoken_secret) params["oauth_signature"] = sig req = urllib2.Request(chirp_url) req.add_header("Authorization", oauth_header(params)) res = urllib2.urlopen(req) res.readline() res.readline() while True: try: j = simplejson.loads(res.readline()) print "[%s] %s".encode('utf-8') % (j['user']['screen_name'], j['text']) except: time.sleep(0.5)