新浪微博Python客户端接口OAuth2

简介: Keyword: Python Oauth2 微博 sina weibo   #!/usr/bin/env python # -*- coding: utf-8 -*- __version__ = '1.
Keyword: Python Oauth2 微博 sina weibo

 

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

__version__ = '1.04'
__author__ = 'Liao Xuefeng (askxuefeng@gmail.com)'

'''
Python client SDK for sina weibo API using OAuth 2.
'''

try:
    import json
except ImportError:
    import simplejson as json
import time
import urllib
import urllib2
import logging

def _obj_hook(pairs):
    '''
    convert json object to python object.
    '''
    o = JsonObject()
    for k, v in pairs.iteritems():
        o[str(k)] = v
    return o

class APIError(StandardError):
    '''
    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
        StandardError.__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):
        return self[attr]

    def __setattr__(self, attr, value):
        self[attr] = value

def _encode_params(**kw):
    '''
    Encode parameters.
    '''
    args = []
    for k, v in kw.iteritems():
        qv = v.encode('utf-8') if isinstance(v, unicode) else str(v)
        args.append('%s=%s' % (k, urllib.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 = []
    for k, v in kw.iteritems():
        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 = 2

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_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 = _encode_params(**kw)
    http_url = '%s?%s' % (url, params) if method==_HTTP_GET else url
    http_body = None if method==_HTTP_GET else params
    req = urllib2.Request(http_url, data=http_body)
    if authorization:
        req.add_header('Authorization', 'OAuth2 %s' % authorization)
    if boundary:
        req.add_header('Content-Type', 'multipart/form-data; boundary=%s' % boundary)
    resp = urllib2.urlopen(req)
    body = resp.read()
    r = json.loads(body, object_hook=_obj_hook)
    if hasattr(r, 'error_code'):
        raise APIError(r.error_code, getattr(r, 'error', ''), getattr(r, 'request', ''))
    return r

class HttpObject(object):

    def __init__(self, client, method):
        self.client = client
        self.method = method

    def __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 wrap

class 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')
        r.expires_in += int(time.time())
        return r

    def is_expires(self):
        return not self.access_token or time.time() > self.expires

    def __getattr__(self, attr):
        return getattr(self.get, attr)


def main():
    try:
        #step 1 定义 app key,app secret,回调地址:
        APP_KEY = "14752459xx"
        APP_SECRET = "7be6f636faf7b17d048c0cd3c55adaxx"
        CALLBACK_URL = 'http://api.doucube.com/oauth2/default.html'
        #step 2 引导用户到授权地址
        client = APIClient(app_key=APP_KEY, app_secret=APP_SECRET, redirect_uri=CALLBACK_URL)
        print client.get_authorize_url()
        #step 3 换取Access Token
        r = client.request_access_token(raw_input("Input CODE:"))#输入授权地址中获得的CODE
        client.set_access_token(r.access_token, r.expires_in)
        #step 4 使用获得的OAuth2.0 Access Token调用API
        print client.get.account__get_uid()
        print client.post.statuses__update(status=u'测试Python + OAuth 2.0发微博')
        print client.upload.statuses__upload(status=u'测试Python + OAuth 2.0带图片发微博', pic=open('test.jpg'))

    except Exception as pyOauth2Error:
        print pyOauth2Error

if __name__ == '__main__':
    main()

 

目录
相关文章
|
20天前
|
C语言 Python
python 调用c接口
【10月更文挑战第12天】 ctypes是Python的一个外部库,提供和C语言兼容的数据类型,可以很方便地调用C DLL中的函数
39 0
|
3月前
|
测试技术 索引 Python
Python接口自动化测试框架(练习篇)-- 数据类型及控制流程(一)
本文提供了Python接口自动化测试中的编程练习,包括计算器、猜数字、猜拳和九九乘法表等经典问题,涵盖了数据类型、运算、循环、条件控制等基础知识的综合应用。
43 1
|
25天前
|
Python
Socket学习笔记(二):python通过socket实现客户端到服务器端的图片传输
使用Python的socket库实现客户端到服务器端的图片传输,包括客户端和服务器端的代码实现,以及传输结果的展示。
102 3
Socket学习笔记(二):python通过socket实现客户端到服务器端的图片传输
|
25天前
|
JSON 数据格式 Python
Socket学习笔记(一):python通过socket实现客户端到服务器端的文件传输
本文介绍了如何使用Python的socket模块实现客户端到服务器端的文件传输,包括客户端发送文件信息和内容,服务器端接收并保存文件的完整过程。
101 1
Socket学习笔记(一):python通过socket实现客户端到服务器端的文件传输
|
1月前
|
JSON 缓存 API
在 Python 中使用公共类处理接口请求的响应结果
在 Python 中使用公共类处理接口请求的响应结果
26 1
|
2月前
|
关系型数据库 MySQL Python
mysql之python客户端封装类
mysql之python客户端封装类
|
2月前
|
安全 API 开发者
Web 开发新风尚!Python RESTful API 设计与实现,让你的接口更懂开发者心!
在当前的Web开发中,Python因能构建高效简洁的RESTful API而备受青睐,大大提升了开发效率和用户体验。本文将介绍RESTful API的基本原则及其在Python中的实现方法。以Flask为例,演示了如何通过不同的HTTP方法(如GET、POST、PUT、DELETE)来创建、读取、更新和删除用户信息。此示例还包括了基本的路由设置及操作,为开发者提供了清晰的API交互指南。
104 6
|
2月前
|
Java Python
全网最适合入门的面向对象编程教程:50 Python函数方法与接口-接口和抽象基类
【9月更文挑战第18天】在 Python 中,虽无明确的 `interface` 关键字,但可通过约定实现类似功能。接口主要规定了需实现的方法,不提供具体实现。抽象基类(ABC)则通过 `@abstractmethod` 装饰器定义抽象方法,子类必须实现这些方法。使用抽象基类可使继承结构更清晰、规范,并确保子类遵循指定的方法实现。然而,其使用应根据实际需求决定,避免过度设计导致代码复杂。
|
2月前
|
Python
全网最适合入门的面向对象编程教程:Python函数方法与接口-函数与方法的区别和lamda匿名函数
【9月更文挑战第15天】在 Python 中,函数与方法有所区别:函数是独立的代码块,可通过函数名直接调用,不依赖特定类或对象;方法则是与类或对象关联的函数,通常在类内部定义并通过对象调用。Lambda 函数是一种简洁的匿名函数定义方式,常用于简单的操作或作为其他函数的参数。根据需求,可选择使用函数、方法或 lambda 函数来实现代码逻辑。
|
3月前
|
IDE 测试技术 开发工具
Python接口自动化测试框架(基础篇)-- 不只是txt的文件操作
本文介绍了Python中的文件操作方法,包括使用open()打开文件、close()关闭文件、read()读取内容、readline()读取单行、readlines()读取多行、write()写入内容以及writelines()写入多行的方法。同时,探讨了文件操作模式和编码问题,并扩展了上下文管理器with...as的使用,以及对图片和音频文件操作的思考和练习。
28 1
Python接口自动化测试框架(基础篇)-- 不只是txt的文件操作