微博自动发布气象数据(二)Python发布微博

在上一篇里面，通过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