Python による Sina Weibo API プロジェクトの呼び出しの実践

最近、Sina Weibo オープン インターフェイスと呼ばれるプロジェクトに出会ったので、Python を使用して Weibo API を呼び出してみようと思いました。

SDK のダウンロード アドレス: http://open.weibo.com/wiki/SDK コードは 10 キロバイト以下で、完全に理解できます。

Weibo アカウントをお持ちの場合は、新しい APP を作成し、APP が OAuth2.0 認証を取得するために必要なアプリ キーとアプリ シークレットを取得できます。

OAuth2 について学ぶには、Sina Weibo へのリンク手順を参照してください。アプリ キーとアプリ シークレットに加えて、OAuth2 認証パラメーターには Web サイトのコールバック アドレス redirect_uri も必要ですが、このコールバック アドレスはローカル エリア ネットワーク内に存在することは許可されていません (Shenma localhost、127.0.0.1 は機能しないようです)。これは本当に長い間私を不安にさせました。私が使用した API は Web サイト呼び出しではなかったので、いろいろ調べました。このアドレス https://api.weibo.com/oauth2/default.html を代わりに使用できると書いている人を見かけました。試してみたらうまくいきました。これは diaosi にとって朗報です。

簡単なプログラムを実行して、感覚を掴んでみましょう:

次のパラメータを設定します

import sys
import weibo
import webbrowser

APP_KEY = ''
MY_APP_SECRET = ''
REDIRECT_URL = 'https://api.weibo.com/oauth2/default.html'

2 行目などの Weibo 認証 URL を取得します。デフォルトのブラウザで開くと、以下に示すように、認証が必要なアカウントで Weibo にログインするよう求められます。

api = weibo.APIClient(app_key=APP_KEY,app_secret=MY_APP_SECRET,redirect_uri=REDIRECT_URL)
authorize_url = api.get_authorize_url()
print(authorize_url)
webbrowser.open_new(authorize_url)

ログイン後、接続 https://api.weibo.com/oauth2/default.html?code=92cc6accecfb5b2176adf58f4c

にリダイレクトされます。

キーはコード値であり、認証の鍵となります。コード値を手動で入力して認証をシミュレートします

request = api.request_access_token(code, REDIRECT_URL)
access_token = request.access_token
expires_in = request.expires_in
api.set_access_token(access_token, expires_in)
api.statuses.update.post(status=u'Test OAuth 2.0 Send a Weibo!')

access_token は取得したトークン、expires_in は認証の有効期限 (UNIX 時間)

set_access_token を使用して認証を保存します。下に行くと、Weiboインターフェースに電話をかけることができます。 Weibo にテスト投稿しました

しかし、このコードを手動で入力する方法は、プログラムを呼び出すのには適していません。何度も検索と参照を行うと、プログラムは自動的に改善されます。コードを取得して保存すると便利です。

accessWeibo

# -*- coding: utf-8 -*- 
#/usr/bin/env python 

#access to SinaWeibo By sinaweibopy 
#实现微博自动登录，token自动生成，保存及更新 
#适合于后端服务调用 


from weibo import APIClient 
import pymongo 
import sys, os, urllib, urllib2 
from http_helper import * 
from retry import * 
try: 
import json 
except ImportError: 
import simplejson as json 

# setting sys encoding to utf-8 
default_encoding = 'utf-8' 
if sys.getdefaultencoding() != default_encoding: 
reload(sys) 
sys.setdefaultencoding(default_encoding) 

# weibo api访问配置 
APP_KEY = '' # app key 
APP_SECRET = '' # app secret 
REDIRECT_URL = 'https://api.weibo.com/oauth2/default.html' # callback url 授权回调页,与OAuth2.0 授权设置的一致 
USERID = '' # 登陆的微博用户名，必须是OAuth2.0 设置的测试账号 
USERPASSWD = '' # 用户密码 


client = APIClient(app_key=APP_KEY, app_secret=APP_SECRET, redirect_uri=REDIRECT_URL) 

def make_access_token(): 
#请求access token 
params = urllib.urlencode({
'action':'submit',
'withOfficalFlag':'0',
'ticket':'',
'isLoginSina':'', 
'response_type':'code',
'regCallback':'',
'redirect_uri':REDIRECT_URL,
'client_id':APP_KEY,
'state':'',
'from':'',
'userId':USERID,
'passwd':USERPASSWD,
}) 

login_url = 'https://api.weibo.com/oauth2/authorize' 

url = client.get_authorize_url() 
content = urllib2.urlopen(url) 
if content: 
headers = { 'Referer' : url } 
request = urllib2.Request(login_url, params, headers) 
opener = get_opener(False) 
urllib2.install_opener(opener) 
try: 
f = opener.open(request) 
return_redirect_uri = f.url 
except urllib2.HTTPError, e: 
return_redirect_uri = e.geturl() 
# 取到返回的code 
code = return_redirect_uri.split('=')[1] 
#得到token 
token = client.request_access_token(code,REDIRECT_URL) 
save_access_token(token) 

def save_access_token(token): 
#将access token保存到MongoDB数据库
mongoCon=pymongo.Connection(host="127.0.0.1",port=27017)
db= mongoCon.weibo

t={
"access_token":token['access_token'],
"expires_in":str(token['expires_in']),
"date":time.strftime('%Y-%m-%d %H:%M:%S',time.localtime(time.time()))
}
db.token.insert(t,safe=True) 

#Decorator 目的是当调用make_access_token()后再执行一次apply_access_token()
@retry(1) 
def apply_access_token(): 
#从MongoDB读取及设置access token 
try: 

mongoCon=pymongo.Connection(host="127.0.0.1",port=27017)
db= mongoCon.weibo
if db.token.count()>0:
tokenInfos=db.token.find().sort([("_id",pymongo.DESCENDING)]).limit(1)
else: 
make_access_token() 
return False 

for tokenInfo in tokenInfos:
access_token=tokenInfo["access_token"]
expires_in=tokenInfo["expires_in"]

try: 
client.set_access_token(access_token, expires_in) 
except StandardError, e: 
if hasattr(e, 'error'): 
if e.error == 'expired_token': 
# token过期重新生成 
make_access_token()
return False 
else: 
pass 
except: 
make_access_token()
return False 

return True 

if __name__ == "__main__": 
apply_access_token() 

# 以下为访问微博api的应用逻辑 
# 以发布文字微博接口为例
client.statuses.update.post(status='Test OAuth 2.0 Send a Weibo!')

retry.py

import math
import time

# Retry decorator with exponential backoff
def retry(tries, delay=1, backoff=2):
"""Retries a function or method until it returns True.

delay sets the initial delay, and backoff sets how much the delay should
lengthen after each failure. backoff must be greater than 1, or else it
isn't really a backoff. tries must be at least 0, and delay greater than
0."""

if backoff <= 1:
raise ValueError("backoff must be greater than 1")

tries = math.floor(tries)
if tries < 0:
raise ValueError("tries must be 0 or greater")

if delay <= 0:
raise ValueError("delay must be greater than 0")

def deco_retry(f):
def f_retry(*args, **kwargs):
mtries, mdelay = tries, delay # make mutable

rv = f(*args, **kwargs) # first attempt
while mtries > 0:
if rv == True or type(rv) == str: # Done on success ..
return rv

mtries -= 1 # consume an attempt
time.sleep(mdelay) # wait...
mdelay *= backoff # make future wait longer

rv = f(*args, **kwargs) # Try again

return False # Ran out of tries :-(

return f_retry # true decorator -> decorated function
return deco_retry # @retry(arg[, ...]) -> true decorator

http_helper.py

# -*- coding: utf-8 -*-
#/usr/bin/env python

import urllib2,cookielib

class SmartRedirectHandler(urllib2.HTTPRedirectHandler):
def http_error_301(cls, req, fp, code, msg, headers):
result = urllib2.HTTPRedirectHandler.http_error_301(cls, req, fp, code, msg, headers)
result.status = code
print headers
return result

def http_error_302(cls, req, fp, code, msg, headers):
result = urllib2.HTTPRedirectHandler.http_error_302(cls, req, fp, code, msg, headers)
result.status = code
print headers
return result

def get_cookie():
cookies = cookielib.CookieJar()
return urllib2.HTTPCookieProcessor(cookies)

def get_opener(proxy=False):
rv=urllib2.build_opener(get_cookie(), SmartRedirectHandler())
rv.addheaders = [('User-agent', 'Mozilla/4.0 (compatible; MSIE 6.0; Windows NT 5.1; SV1)')]
return rv