在上一篇里面，通過Python簡(jiǎn)單的爬蟲在網(wǎng)上獲取了天氣實(shí)況數(shù)據(jù)朴上，這次想辦法用Python把數(shù)據(jù)發(fā)布在微博上面片习，微博提供了這樣的接口，我們所做的就是用Python實(shí)現(xiàn)和微博公共接口的對(duì)接式矫。
在這之前我們需要訪問微博 微博開放平臺(tái) 來注冊(cè)一個(gè)應(yīng)用，這個(gè)完全是免費(fèi)的役耕。注冊(cè)成功之后獲得AppKey和App Secret采转，這是非常重要的數(shù)據(jù)，在后面會(huì)用到瞬痘。
注冊(cè)應(yīng)用.png
通過上面的
key和Secret故慈，我們可以從新浪微博服務(wù)器獲取一個(gè)最最關(guān)鍵的參數(shù)，就是access_token框全，這個(gè)參數(shù)表明我們通過了服務(wù)器認(rèn)證察绷，也就是說微博認(rèn)為我們是擁有發(fā)布微博的權(quán)限的。獲取的方法大概是：
1津辩、使用get方法訪問新浪的OAuth認(rèn)證頁(yè)拆撼，需要傳遞的參數(shù)是AppKey和重定向地址容劳，在通過驗(yàn)證之后，將重定向到我們提供的重定向地址闸度，并提供給我們一個(gè)CODE參數(shù)的值鸭蛙。
2、發(fā)送post請(qǐng)求給微博服務(wù)器筋岛，包含以下參數(shù)：
fields = {'code': code, 'redirect_uri': self.redirect_uri, 'client_id': self.client_id,
'client_secret': self.client_secret, 'grant_type': 'authorization_code'}
收到返回值，一個(gè)就是access_token的值晒哄，另外一個(gè)expires是指access_token的超時(shí)時(shí)間睁宰，超時(shí)后需重新獲取access_token。
之后的工作就是調(diào)用接口發(fā)送數(shù)據(jù)了寝凌，當(dāng)然柒傻，微博在這接口里面做出了很多限制，以前限制很少较木，現(xiàn)在用起來不那么舒服了：
限制.png
簡(jiǎn)易的代碼如下：兩個(gè)文件：
weiboSDK.py
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
__version__ = '1.0.0'
__author__ = 'aeropig@163.com'
'''
Python3 client SDK for sina weibo API using OAuth 2.
'''
# import gzip
import time
import requests
# import json
# import hmac
# import base64
# import hashlib
import logging
# import mimetypes
# import collections
# from io import StringIO
import webbrowser
default_redirect_uri = 'https://api.weibo.com/oauth2/default.html'
_HTTP_GET = 0
_HTTP_POST = 1
_HTTP_POST_UPLOAD = 2
def _encode_params(**kw):
args = []
for (k, v) in kw.items():
args.append('%s=%s' % (str(k), str(v)))
return '&'.join(args)
class APIError(BaseException):
'''
raise APIError
'''
def __init__(self, error_code, error, request):
self.error_code = error_code
self.error = error
self.request = request
BaseException.__init__(self, error)
def __str__(self):
return 'APIError: %s: %s, request: %s' % (self.error_code, self.error, self.request)
def _http_get(url, authorization=None, **kw):
logging.info('GET %s' % url)
return _http_call(url, _HTTP_GET, authorization, **kw)
def _http_post(url, authorization=None, **kw):
logging.info('POST %s' % url)
return _http_call(url, _HTTP_POST, authorization, **kw)
def _http_post_upload(url, authorization=None, **kw):
logging.info('MULTIPART POST %s' % url)
return _http_call(url, _HTTP_POST_UPLOAD, authorization, **kw)
def _http_call(the_url, method, authorization, **kw):
'''
send an http request and return a json object if no error occurred.
'''
if authorization:
kw['access_token'] = authorization
if method == _HTTP_GET:
r = requests.get(the_url, params=kw)
elif method == _HTTP_POST:
r = requests.post(the_url, data=kw)
elif method == _HTTP_POST_UPLOAD:
_files = {'pic': open(kw['pic'], 'rb')}
kw.pop('pic')
r = requests.post(the_url, data=kw, files=_files)
else:
pass
return r.json()
class APIClient(object):
'''
API client using synchronized invocation.
'''
def __init__(self, app_key, app_secret, redirect_uri=default_redirect_uri, domain='api.weibo.com', version='2'):
self.client_id = str(app_key)
self.client_secret = str(app_secret)
self.redirect_uri = redirect_uri
self.auth_url = 'https://%s/oauth2/' % domain
self.api_url = 'https://%s/%s/' % (domain, version)
def authorize(self):
if isinstance(self.access_token, str):
return
authorize_url = '%s%s?' % (self.auth_url, 'authorize')
fields = {'client_id': self.client_id,
'redirect_uri': self.redirect_uri}
params = _encode_params(**fields)
webbrowser.open(authorize_url + params)
def set_access_token(self, **token_dict):
self.access_token = token_dict['access_token']
self.uid = token_dict.get('uid', 0)
current = int(time.time())
self.expires = token_dict['expires_in'] + current
return (self.access_token, self.expires)
def request_access_token(self, code):
fields = {'code': code, 'redirect_uri': self.redirect_uri, 'client_id': self.client_id,
'client_secret': self.client_secret, 'grant_type': 'authorization_code'}
r = _http_post('%s%s' % (self.auth_url, 'access_token'), **fields)
return self.set_access_token(**r)
# def refresh_token(self, refresh_token):
# r = _http_post(req_str,
# client_id=self.client_id,
# client_secret=self.client_secret,
# refresh_token=refresh_token,
# grant_type='refresh_token')
# return self._parse_access_token(r)
def is_expires(self):
return not self.access_token or time.time() > self.expires
def __getattr__(self, attr):
if '__' in attr:
return getattr(self.get, attr)
return _Callable(self, attr)
class _Executable(object):
def __init__(self, client, method, path):
self._client = client
self._method = method
self._path = path
def __call__(self, **kw):
if self._method == _HTTP_POST and 'pic' in kw:
self._method = _HTTP_POST_UPLOAD
return _http_call('%s%s.json' % (self._client.api_url, self._path), self._method, self._client.access_token, **kw)
def __str__(self):
return '_Executable (%s %s)' % (self._method, self._path)
__repr__ = __str__
class _Callable(object):
def __init__(self, client, name):
self._client = client
self._name = name
def __getattr__(self, attr):
if attr == 'get':
return _Executable(self._client, _HTTP_GET, self._name)
if attr == 'post':
return _Executable(self._client, _HTTP_POST, self._name)
name = '%s/%s' % (self._name, attr)
return _Callable(self._client, name)
def __str__(self):
return '_Callable (%s)' % self._name
__repr__ = __str__
if __name__ == '__main__':
pass
第二個(gè)文件：weiboAPP.py
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
import weiboSDK
uid = 'xxxxxxxxxxx'
client_id = 'xxxxxxxxxxxxxx'
client_secret = 'xxxxxxxxxxxxxxxxxxxxxxxxxxxxx'
access_token = 'xxxxxxxxxxxxxxxxxxxxxxxxxx'
expires = 1672970161
if __name__ == '__main__':
weibo = weiboSDK.APIClient(client_id, client_secret)
if access_token is None:
weibo.authorize()
code = input("輸入從瀏覽器獲取的CODE：")
accessTuple = weibo.request_access_token(code)
print(accessTuple)
access_token = accessTuple[0]
expires = accessTuple[1]
weibo.set_access_token(access_token=access_token, expires_in=expires)
weibo.statuses.share.post(status='這條微博只是一個(gè)測(cè)試[doge]... http://www.reibang.com/p/d55b71e85bd0 ')
執(zhí)行效果如下：
發(fā)布成功.png
