新浪|新浪 python sdk 适配 python3.+ 新浪pythonsdk适配python3.+

昨天使用廖雪峰老师的新浪 python sdk，发现该sdk一直没更新，主要是部分模块还是基于python2.+，因此自己尝试修改了下。

#!/usr/bin/env python # -*- coding: utf-8 -*-__version__ = '2.0' __author__ = 'Liao Xuefeng (askxuefeng@gmail.com)'''' Python client SDK for sina weibo API using OAuth 2. '''import json import time import requests import loggingdef _obj_hook(pairs): ''' convert json object to python object. ''' o = JsonObject() for k, v in pairs. items(): o[str(k)] = v return oclass APIError(Exception): ''' raise APIError if got failed json message. '''def __init__(self, error_code, error, request): self.error_code = error_code self.error = error self.request = request Exception.__init__(self, error)def __str__(self): return 'APIError: %s: %s, request: %s' % (self.error_code, self.error, self.request)class JsonObject(dict): ''' general json object that can bind any fields but also act as a dict. '''def __getattr__(self, attr): if attr in self: return self[attr] else: passdef __setattr__(self, attr, value): self[attr] = valuedef _encode_params(**kw): ''' Encode parameters. ''' args = [] for k, v in kw.items(): qv = v.encode('utf-8') if isinstance(v, str) else str(v) args.append('%s=%s' % (k, requests.utils.quote(qv))) return '&'.join(args)def _encode_multipart(**kw): ''' Build a multipart/form-data body with generated random boundary. ''' boundary = '----------%s' % hex(int(time.time() * 1000)) data = https://www.it610.com/article/[] for k, v in kw.items(): data.append('--%s' % boundary) if hasattr(v, 'read'): # file-like object: ext = '' filename = getattr(v, 'name', '') n = filename.rfind('.') if n != (-1): ext = filename[n:].lower() content = v.read() data.append('Content-Disposition: form-data; name="%s"; filename="hidden"' % k) data.append('Content-Length: %d' % len(content)) data.append('Content-Type: %s\r\n' % _guess_content_type(ext)) data.append(content) else: data.append('Content-Disposition: form-data; name="%s"\r\n' % k) data.append(v.encode('utf-8') if isinstance(v, unicode) else v) data.append('--%s--\r\n' % boundary) return '\r\n'.join(data), boundary_CONTENT_TYPES = {'.png': 'image/png', '.gif': 'image/gif', '.jpg': 'image/jpeg', '.jpeg': 'image/jpeg', '.jpe': 'image/jpeg'}def _guess_content_type(ext): return _CONTENT_TYPES.get(ext, 'application/octet-stream')_HTTP_GET = 0 _HTTP_POST = 1 _HTTP_UPLOAD = 2def _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_upload(url, authorization=None, **kw): logging.info('MULTIPART POST %s' % url) return _http_call(url, _HTTP_UPLOAD, authorization, **kw)def _http_call(url, method, authorization, **kw): ''' send an http request and expect to return a json object if no error. ''' params = None boundary = None if method == _HTTP_UPLOAD: params, boundary = _encode_multipart(**kw) else: params = kw http_url = '%s?%s' % (url, params) if method == _HTTP_GET else url http_body = None if method == _HTTP_GET else params header = {} if authorization: header['Authorization'] = 'OAuth2 %s' % authorization if boundary: header['Content-Type'] = 'multipart/form-data; boundary=%s' % boundaryreq = requests.post(http_url, data=https://www.it610.com/article/params, json=header) body = req.text r = json.loads(body, object_hook=_obj_hook) if'error_code' in r: raise APIError(r.error_code, getattr(r, 'error', ''), getattr(r, 'request', '')) return rclass HttpObject(object):def __init__(self, client, method): self.client = client self.method = methoddef __getattr__(self, attr): def wrap(**kw): if self.client.is_expires(): raise APIError('21327', 'expired_token', attr) return _http_call('%s%s.json' % (self.client.api_url, attr.replace('__', '/')), self.method, self.client.access_token, **kw) return wrapclass APIClient(object): ''' API client using synchronized invocation. '''def __init__(self, app_key, app_secret, redirect_uri=None, response_type='code', domain='api.weibo.com', version='2'): self.client_id = app_key self.client_secret = app_secret self.redirect_uri = redirect_uri self.response_type = response_type self.auth_url = 'https://%s/oauth2/' % domain self.api_url = 'https://%s/%s/' % (domain, version) self.access_token = None self.expires = 0.0 self.get = HttpObject(self, _HTTP_GET) self.post = HttpObject(self, _HTTP_POST) self.upload = HttpObject(self, _HTTP_UPLOAD)def set_access_token(self, access_token, expires_in): self.access_token = str(access_token) self.expires = float(expires_in)def get_authorize_url(self, redirect_uri=None, display='default'): ''' return the authroize url that should be redirect. ''' redirect = redirect_uri if redirect_uri else self.redirect_uri if not redirect: raise APIError('21305', 'Parameter absent: redirect_uri', 'OAuth2 request') return '%s%s?%s' % (self.auth_url, 'authorize', _encode_params(client_id=self.client_id, response_type='code', display=display, redirect_uri=redirect))def request_access_token(self, code, redirect_uri=None): ''' return access token as object: {"access_token":"your-access-token","expires_in":12345678}, expires_in is standard unix-epoch-time ''' redirect = redirect_uri if redirect_uri else self.redirect_uri if not redirect: raise APIError('21305', 'Parameter absent: redirect_uri', 'OAuth2 request') r = _http_post('%s%s' % (self.auth_url, 'access_token'), client_id=self.client_id, client_secret=self.client_secret, redirect_uri=redirect, code=code, grant_type='authorization_code') # print(r.expires_in) r.expires_in += int(time.time()) return rdef is_expires(self): return not self.access_token or time.time() > self.expiresdef __getattr__(self, attr): return getattr(self.get, attr)

【新浪|新浪 python sdk 适配 python3.+】 这里附上测试案例

from weibo import APIClient# 需要先在新浪微博开发者里创建应用，创建后得到app_key，app_sercret(应用未审核不影响一般使用) app_key = ""# app_secret = "" CALLBACK_URL = '' #回调网址需要自己在应用里设置 client = APIClient(app_key=app_key, app_secret=app_secret, redirect_uri=CALLBACK_URL)url = client.get_authorize_url()# 获取认证链接 print(url)# code = "f95e0757a38db8356db8d08c266a0728" # 从回调网址里获取code值，code使用一次后失效 # r = client.request_access_token(code) # uid = r.uid#新浪用户uid # expires_in = r.expires_in#access_token生命期 # access_token = r.access_token # print(expires_in,access_token,uid)# 根据api文档获取新浪用户个人信息:https://api.weibo.com/2/users/show.json? # 提供access_token、uid、appkey参数，get请求方式 get_url = 'https://api.weibo.com/2/users/show.json?access_token=xxxx&uid=xxxxx&appkey=xxxx' req = requests.get(get_url)# 返回json数据

这里附上新浪api文档：https://open.weibo.com/wiki/%E5%BE%AE%E5%8D%9AAPI