manyi/library/wechatpy/oauth.py

# -*- coding: utf-8 -*-
"""
    wechatpy.oauth
    ~~~~~~~~~~~~~~~

    This module provides OAuth2 library for WeChat

    :copyright: (c) 2014 by messense.
    :license: MIT, see LICENSE for more details.
"""
from __future__ import absolute_import, unicode_literals

import requests
import six
from six.moves.urllib.parse import quote

from library import to_binary, to_text
from library.wechatbase.exceptions import WeChatOAuthException
from library.wechatpy.utils import json


class WeChatOAuth(object):
    """ 微信公众平台 OAuth 网页授权

    详情请参考
    https://open.weixin.qq.com/cgi-bin/showdocument?action=dir_list&t=resource/res_list&verify=1&id=open1419316505
    """

    API_BASE_URL = 'https://api.weixin.qq.com/'
    OAUTH_BASE_URL = 'https://open.weixin.qq.com/connect/'

    def __str__(self):
        _repr = '{kclass}(appid: {appid})'.format(
            kclass = self.__class__.__name__,
            appid = self.app_id)

        if six.PY2:
            return to_binary(_repr)
        else:
            return to_text(_repr)

    def __repr__(self):
        return str(self)

    def __init__(self, app_id, secret, scope = 'snsapi_base', state = ''):
        """

        :param app_id: 微信公众号 app_id
        :param secret: 微信公众号 secret
        :param scope: 可选，微信公众号 OAuth2 scope，默认为 ``snsapi_base``
        :param state: 可选，微信公众号 OAuth2 state
        """
        self.app_id = app_id
        self.secret = secret
        self.scope = scope
        self.state = state

    def _request(self, method, url_or_endpoint, **kwargs):
        if not url_or_endpoint.startswith(('http://', 'https://')):
            url = '{base}{endpoint}'.format(
                base = self.API_BASE_URL,
                endpoint = url_or_endpoint
            )
        else:
            url = url_or_endpoint

        if isinstance(kwargs.get('data', ''), dict):
            body = json.dumps(kwargs['data'], ensure_ascii = False)
            body = body.encode('utf-8')
            kwargs['data'] = body

        kwargs['timeout'] = kwargs.get('timeout', 15)

        with requests.sessions.Session() as session:
            res = session.request(
                method = method,
                url = url,
                **kwargs
            )
            try:
                res.raise_for_status()
            except requests.RequestException as reqe:
                raise WeChatOAuthException(
                    errCode = 'HTTP{}'.format(res.status_code),
                    errMsg = reqe.message,
                    client = self,
                    request = reqe.request,
                    response = reqe.response
                )
            result = json.loads(res.content.decode('utf-8', 'ignore'), strict = False)

            if 'errcode' in result and result['errcode'] != 0:
                errcode = result['errcode']
                errmsg = result['errmsg']
                raise WeChatOAuthException(
                    errCode = errcode,
                    errMsg = errmsg,
                    client = self,
                    request = res.request,
                    response = res
                )

            return result

    def _get(self, url, **kwargs):
        return self._request(
            method = 'get',
            url_or_endpoint = url,
            **kwargs
        )

    def authorize_url(self, redirect_uri):
        """获取授权跳转地址

        :return: URL 地址
        """
        redirect_uri = quote(redirect_uri, safe = b'')
        url_list = [
            self.OAUTH_BASE_URL,
            'oauth2/authorize?appid=',
            self.app_id,
            '&redirect_uri=',
            redirect_uri,
            '&response_type=code&scope=',
            self.scope
        ]

        if self.state:
            url_list.extend(['&state=', self.state])
        else:
            url_list.extend(['&state=', ''])

        url_list.extend(['&connect_redirect=', '1'])
        url_list.append('#wechat_redirect')
        return ''.join(url_list)

    def qrconnect_url(self, redirect_uri):
        """生成扫码登录地址

        :return: URL 地址
        """
        redirect_uri = quote(redirect_uri, safe = b'')
        url_list = [
            self.OAUTH_BASE_URL,
            'qrconnect?appid=',
            self.app_id,
            '&redirect_uri=',
            redirect_uri,
            '&response_type=code&scope=',
            'snsapi_login'  # scope
        ]
        if self.state:
            url_list.extend(['&state=', self.state])
        url_list.append('#wechat_redirect')
        return ''.join(url_list)

    def get_oauth_token(self, auth_code):
        """获取网页授权access token
        :param auth_code: 授权完成跳转回来后 URL 中的 code 参数
        :return: JSON 数据包
        """
        res = self._get(
            'sns/oauth2/access_token',
            params = {
                'appid': self.app_id,
                'secret': self.secret,
                'code': auth_code,
                'grant_type': 'authorization_code'
            }
        )
        return res

    def refresh_access_token(self, refresh_token):
        """刷新 access token

        :param refresh_token: OAuth2 refresh token
        :return: JSON 数据包
        """
        res = self._get(
            'sns/oauth2/refresh_token',
            params = {
                'appid': self.app_id,
                'grant_type': 'refresh_token',
                'refresh_token': refresh_token
            }
        )

        return res

    def get_user_info(self, openid, access_token, lang = 'zh_CN'):
        """获取用户信息

        :param openid: 微信 openid，默认获取当前授权用户信息
        :param access_token: 网页授权access token
        :param lang: 可选，语言偏好, 默认为 ``zh_CN``
        :return: JSON 数据包
        """
        return self._get(
            'sns/userinfo',
            params = {
                'access_token': access_token,
                'openid': openid,
                'lang': lang
            }
        )

    def check_access_token(self, openid, access_token):
        """检查 access_token 有效性

        :param openid: 微信 openid，默认获取当前授权用户信息
        :param access_token: 网页授权access token
        :return: 有效返回 True，否则 False
        """
        res = self._get(
            'sns/auth',
            params = {
                'access_token': access_token,
                'openid': openid
            }
        )
        if res['errcode'] == 0:
            return True
        return False

    def jscode2session(self, js_code):
        """
        小程序获取 session_key 和 openid
        """
        url = "https://api.weixin.qq.com/sns/jscode2session"
        args = dict()
        args.setdefault("appid", self.app_id)
        args.setdefault("secret", self.secret)
        args.setdefault("js_code", js_code)
        args.setdefault("grant_type", "authorization_code")
        return self._get('sns/jscode2session', params = args)