# -*- coding: utf-8 -*- # 標準ライブラリ(install不要) import urllib, urllib2, time, datetime, httplib, cgi, sys import time, random # easy_installを使って install.(from pypi) import simplejson from BeautifulSoup import BeautifulSoup from oauth import oauth # 個人情報保持用obj class MyProfile: def __init__( self, user, password, consumer_key, secret_key ): self.user = user # your twitter id/account self.password = password # your twitter password self.consumer_key = consumer_key # 'Consumer key' self.secret_key = secret_key # 'Consumer secret' return profile = MyProfile( 'your twitter id/account', 'your password', 'consumer key', 'secret key' ) try: import myprofile myprofile.set_myprofile( profile ) except: pass # access先URLたち。API documentとか OAuth登録の返事メールなどに表記あり。 request_token_url = 'https://twitter.com/oauth/request_token' # 'Request token URL' access_token_url = 'https://twitter.com/oauth/access_token' # 'Access token URL' authorize_url = 'http://twitter.com/oauth/authorize' # 'Authorize URL' home_timeline_url = 'http://twitter.com/statuses/home_timeline.json' # 自分のTLを取得するURL。拡張子は .xml or .json update_url = 'http://twitter.com/statuses/update.json' # 発言(update)するURL。拡張子は .xml or .json import locale locale.setlocale(locale.LC_ALL, '') locale.setlocale(locale.LC_TIME, 'C') #----- # 指定した urlからHTTP GETを使って responseを取得する。 # # @param url 取得アクセス先URL。 # @param retry_count HTTPエラーが起きたときにリトライする回数 # @param retry_wait HTTPエラーが起きてリトライする時に待つ秒数 # # @return 成功したときは response object。失敗したときは None #----- def http_get( url, retry_count=10, retry_wait=5 ): req = urllib2.Request( url ) for i in range( retry_count ): try: resp = urllib2.urlopen( req ) return resp except: print '...open error' pass time.sleep( retry_wait ) return None #----- # 指定した urlからHTTP GETを使って responseを取得する。 # # @param url 取得アクセス先URL。 # @param post postする文字列データ # @param retry_count HTTPエラーが起きたときにリトライする回数 # @param retry_wait HTTPエラーが起きてリトライする時に待つ秒数 # # @return 成功したときは response object。失敗したときは None #----- def http_post( url, postdata, retry_count=10, retry_wait=5 ): for i in range( retry_count ): try: resp = urllib2.urlopen( url, postdata ) return resp except: pass time.sleep( retry_wait ) return None #---- # 'Request token URL'に、Consumer keyと Secret keyで signしたリクエストをなげて、tokenをもらう # # @param consumer OAuth用consumer # @param signature_method signするための method. twitterだと HMAC_SHA1。 # # @return 成功したら得られた token。失敗したら None。 #---- def get_unauthorized_request_token( consumer, signature_method ): oauth_request = oauth.OAuthRequest.from_consumer_and_token( consumer, http_url=request_token_url ) oauth_request.sign_request( signature_method, consumer, None ) url = oauth_request.to_url() print "\nURL(get_unauthorized_request_token):", url try: resp = http_get( url ).read() print '\nResult(get_unauthorized_request_token):', cgi.parse_qs( resp ) except: return None token = oauth.OAuthToken.from_string( resp ) return token #----- # tokenを使って、認証用の urlを作成する。 #----- def get_authorization_url( consumer, token, signature_method ): oauth_request = oauth.OAuthRequest.from_consumer_and_token(consumer, token=token, http_url=authorize_url) oauth_request.sign_request(signature_method, consumer, token) return oauth_request.to_url() #----- # twitterを update(発言を post)する # 通信は POSTで行う。 def update( consumer, signature_method, pin, token2, status ): oauth_request = oauth.OAuthRequest.from_consumer_and_token( consumer, http_url=update_url, verifier=pin, token=token2 ) oauth_request.set_parameter( 'status', status ) oauth_request.http_method = 'POST' oauth_request.sign_request(signature_method, consumer, token2) postdata = oauth_request.to_postdata() return http_post( update_url, postdata ) #----- # 自分の home timelineを取得する。 # 通信は GETで行う。 #----- def home_timeline( consumer, signature_method, pin, token2 ): oauth_request = oauth.OAuthRequest.from_consumer_and_token( consumer, http_url=home_timeline_url, verifier=pin, token=token2 ) oauth_request.sign_request(signature_method, consumer, token2) url = oauth_request.to_url() print 'URL(home_timeline):', url return http_get( url ) #----- # twitterに loginする。 # # @param url ログイン用ページの url # @param user twitterアカウント名 # @param password twitterパスワード # # @return 成功した時はログイン後ページの response object。失敗したときはNone #---- def login_twitter( url, user, password ): # ログインページの htmlを取得 page = http_get( url ) # ログインページを分解して自動ログイン。 soup = BeautifulSoup( page ) try: form = soup.findAll( 'form', id='login_form' )[0] except: print >>sys.stderr, 'Error:cannot find login_form' exit() inputs = {} for v in form.findAll( 'input' ): if v.has_key('name'): vname = v['name'].encode('utf-8') try: inputs[vname] = v['value'].encode('utf-8') except: inputs[vname] = None inputs['session[password]'] = password inputs['session[username_or_email]'] = user del inputs['cancel'] return http_post( form['action'], urllib.urlencode( inputs ) ) #----- # main #----- if __name__ == "__main__": consumer = oauth.OAuthConsumer( profile.consumer_key, profile.secret_key ) signature_method = oauth.OAuthSignatureMethod_HMAC_SHA1() # 認証をもらうための tokenをもらう token = get_unauthorized_request_token( consumer, signature_method ) if token is None: print >>sys.stderr, 'cannot use get_unauth' exit() print '\nToken(authorize):', token # もらったtokenを使って、認証用の urlを作成する。 aurl = get_authorization_url( consumer, token, signature_method ) print '\nURL(get_authorization_url):', aurl # 認証用URLでログイン(APIでログイン不要にした場合はいらない) page = login_twitter( aurl, profile.user, profile.password ) # desktop用で APIを使用する場合、PINコードの入力が求められるのでそれをページ内容から自動取得。 soup = BeautifulSoup( page ) try: pin = int( soup.find( 'div', id="oauth_pin" ).string.strip() ) print '\npin:', pin except: pass # 得られた PIN(OAuth的には verifier)と、認証をもらうために最初もらった tokenから # アクセス用の新tokenをもらうためのURLを作る。 oauth_request = oauth.OAuthRequest.from_consumer_and_token( consumer, http_url=access_token_url, verifier=pin, token=token ) oauth_request.sign_request(signature_method, consumer, token) url = oauth_request.to_url() print '\nURL(access_token):', url resp = http_get( url ).read() if resp is None: print >>sys.stderr, "Error: cannot get access token." exit() # 実際の通信につかう、アクセス用tokenを取得。 token2 = oauth.OAuthToken.from_string( resp ) print '\nToken(access):', token2 print cgi.parse_qs( resp ) print "\n\n------------" # pin(verifier)とアクセス用tokenを使って、実際の発言やアクセスを行う。 resp = home_timeline( consumer, signature_method, pin, token2 ) resp_dict = simplejson.loads( resp.read() ) for line in resp_dict: print "[%s] %s %s (%s) by %s" % ( line['user']['name'], line['text'], line['created_at'], line['id'], line['source']) # resp = update( consumer, signature_method, pin, token2, str( datetime.datetime.now() ) ) # print resp, resp.read()