manyi/apps/web/core/auth/unionpay.py

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

import logging
import urllib

from apilib.systypes import IterConstant

from apps.web.core.auth.base import AuthBridge
from apps.web.user.conf import USER_AUTH_REDIRECT_URL
from library.unionpay.oauth import UnionPayOAuth
from apilib.utils_url import add_query

logger = logging.getLogger(__name__)


class unionAuthScope(IterConstant):
    # 静默跳转，无需用户点击授权
    AUTH_SCOPE_BASE = 'upapi_base'


class UnionAuthBridge(AuthBridge):
    TOKEN_CACHE_KEY = 'access_token_from_{appid}_{code}'
    
    # 测试网关oauth
    DEV_ALIPAY_OAUTH_URL = \
        'https://open.95516.com/s/open/html/oauth.html?' + \
        'appId={appid}&scope={scope}&redirectUri={encoded_return_uri}&state={state}&responseType=code'

    # 生产环境网关oauth
    #https://open.95516.com/s/open/html/oauth.html?appId=APPID&redirectUri=REDIRECTURI&responseType=code&scope=SCOPE&state=STATE
    
    PRODUCTION_ALIPAY_OAUTH_URL = \
        'https://open.95516.com/s/open/html/oauth.html?' + \
        'appId={appid}&scope={scope}&redirectUri={encoded_return_uri}&state={state}&responseType=code'
        
    def __init__(self, app):
        self.app = app
        if self.debug:
            self._auth_gateway_tmpl = self.DEV_ALIPAY_OAUTH_URL
        else:
            self._auth_gateway_tmpl = self.PRODUCTION_ALIPAY_OAUTH_URL
            
    def __repr__(self):
        return '<unionAuthBridge(appid=%s, secret=******, occupant=%s)>' \
               % (self.app.appid, self.app.occupantId)

    @property
    def appid(self):
        return self.app.appid

    @property
    def secret(self):
        return self.app.secret

    @property
    def occupantId(self):
        return self.app.occupantId
    
    @property
    def enable(self):
        return self.app.alipay_enable
    
    @property
    def client(self):
        return UnionPayOAuth(self.appid, self.secret)

    def authorize(self, auth_code):
        logger.debug('{} authorize enter. code = {}'.format(repr(self), auth_code))

        if auth_code is None:
            logger.error('{} fail to authorize because code is null'.format(repr(self)))
            return None
        try:
            openId = self.client.get_oauth_token(auth_code).get('openid')

            logger.debug('{} success to authorize. open id = {}'.format(repr(self), openId))

            return openId
        except Exception as e:
            logger.error('{} fail to authorize because of exception. code = {}'.format(repr(self), auth_code))
            logger.exception(e)
            return None

    def get_user_info(self, token,code):
        # type:(str)->dict
        logger.debug('{bridge} get user info. code = {code}'.format(bridge = repr(self), code = code))
        return self.client.get_user_info(token = token, code = code)

    def generate_auth_url(self, redirect_uri, payload = '', scope = unionAuthScope.AUTH_SCOPE_BASE):
        """
        生成授权url
        :param redirect_uri:
        :param payload:
        :param scope:
        :return:
        """
        redirect_uri = ''
        return UnionPayOAuth(self.appid, self.secret, scope = scope).authorize_url(redirect_uri = redirect_uri)

    def generate_auth_callback_url(self, payload = None, auth_callback_url = USER_AUTH_REDIRECT_URL.UNIONPAY):
        """
        生成微信跳转url base范围，只能获取openId
        :param payload:
        :return:
        """
        logger.debug('generate_auth_callback_url enter. bridge = {}, callback url = {}; payload = {}'.format(repr(self),
                                                                                                             auth_callback_url,
                                                                                                             payload))

        if payload:
            callback_url = add_query(auth_callback_url, {'payload': payload})

        encoded_return_uri = urllib.quote_plus(callback_url)

        result = self._auth_gateway_tmpl.format(appid = self.appid,
                                              scope = unionAuthScope.AUTH_SCOPE_BASE,
                                              encoded_return_uri = encoded_return_uri,
                                              state = '')
        logger.debug('generate_auth_callback_url success. result = {}'.format(str(result)))
        return result