在上一篇里面，通过Python简单的爬虫在网上获取了天气实况数据，这次想办法用Python把数据发布在微博上面，微博提供了这样的接口，我们所做的就是用Python实现和微博公共接口的对接。
在这之前我们需要访问微博 微博开放平台 来注册一个应用，这个完全是免费的。注册成功之后获得AppKey和App Secret，这是非常重要的数据，在后面会用到。
注册应用.png
通过上面的
key和Secret，我们可以从新浪微博服务器获取一个最最关键的参数，就是access_token，这个参数表明我们通过了服务器认证，也就是说微博认为我们是拥有发布微博的权限的。获取的方法大概是：
1、使用get方法访问新浪的OAuth认证页，需要传递的参数是AppKey和重定向地址，在通过验证之后，将重定向到我们提供的重定向地址，并提供给我们一个CODE参数的值。
2、发送post请求给微博服务器，包含以下参数：
fields = {'code': code, 'redirect_uri': self.redirect_uri, 'client_id': self.client_id,
'client_secret': self.client_secret, 'grant_type': 'authorization_code'}
收到返回值，一个就是access_token的值，另外一个expires是指access_token的超时时间，超时后需重新获取access_token。
之后的工作就是调用接口发送数据了，当然，微博在这接口里面做出了很多限制，以前限制很少，现在用起来不那么舒服了：
限制.png
简易的代码如下：两个文件：
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
第二个文件：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='这条微博只是一个测试[doge]... https://www.jianshu.com/p/d55b71e85bd0 ')
执行效果如下：
发布成功.png
网友评论