123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209 |
- # -*- coding: utf-8 -*-
- # !/usr/bin/env python
- from __future__ import unicode_literals
- import hashlib
- import time
- from collections import OrderedDict
- import requests
- import simplejson as json
- import six
- from typing import Dict
- from library import to_binary, to_text
- from .exceptions import DlbPayException
- class DlbPay(object):
- PAY_HOST_DEV = 'https://openapi.duolabao.com'
- PAY_HOST = 'https://openapi.duolabao.com'
- def __str__(self):
- _repr = '{kclass}(customerNum: {customerNum}, shopNum: {shopNum})'.format(
- kclass = self.__class__.__name__,
- customerNum = self.customerNum,
- shopNum = self.shopNum)
- if six.PY2:
- return to_binary(_repr)
- else:
- return to_text(_repr)
- def __repr__(self):
- return str(self)
- def __init__(self, customerNum, shopNum, accessKey, secretKey, machineNum = None, debug = False):
- self.customerNum = customerNum
- self.shopNum = shopNum
- self.machineNum = machineNum
- self.secretKey = secretKey
- self.accessKey = accessKey
- self.debug = debug
- if self.debug:
- self.host_url = DlbPay.PAY_HOST_DEV
- else:
- self.host_url = DlbPay.PAY_HOST
- def token(self, data):
- # type: (OrderedDict)->str
- s = '&'.join([u'{}={}'.format(k, v) for k, v in data.iteritems()])
- s = s.encode('utf-8')
- return hashlib.sha1(s).hexdigest().upper()
- def get(self, endpoint, data = None, **kwargs):
- def make_token(data, uri, timestamp):
- sign_data = OrderedDict()
- sign_data['secretKey'] = self.secretKey
- sign_data['timestamp'] = timestamp
- sign_data['path'] = uri
- return self.token(sign_data)
- timestamp = str(int(time.time()))
- headers = {
- 'accessKey': self.accessKey,
- 'timestamp': timestamp,
- 'token': make_token(data, endpoint, timestamp)
- }
- return self._request(
- method = 'get',
- endpoint = endpoint,
- data = data,
- headers = headers,
- **kwargs)
- def post(self, endpoint, data = None, **kwargs):
- def make_token(data, uri, timestamp):
- sign_data = OrderedDict()
- sign_data['secretKey'] = self.secretKey
- sign_data['timestamp'] = timestamp
- sign_data['path'] = uri
- sign_data['body'] = json.dumps(data, ensure_ascii = False).encode('utf-8')
- return self.token(sign_data)
- timestamp = str(int(time.time()))
- headers = {
- 'accessKey': self.accessKey,
- 'timestamp': timestamp,
- 'token': make_token(data, endpoint, timestamp)
- }
- return self._request(
- method = 'post',
- endpoint = endpoint,
- data = data,
- headers = headers,
- **kwargs
- )
- def _decode_result(self, res):
- try:
- result = json.loads(res.content.decode('utf-8', 'ignore'), strict = False)
- except (TypeError, ValueError):
- return res
- return result
- def _handle_result(self, res, method = None, url = None,
- result_processor = None, **kwargs):
- if not isinstance(res, dict):
- result = self._decode_result(res)
- else:
- result = res
- if result['result'] == 'fail' or result['result'] == 'error':
- raise DlbPayException(
- errorCode = result['error']['errorCode'] if 'error' in result and 'errorCode' in result[
- 'error'] else '',
- errorMsg = result['error']['errorMsg'] if 'error' in result and 'errorMsg' in result[
- 'error'] else u'未知错误',
- client = self,
- request = None,
- response = None)
- return result if not result_processor else result_processor(result)
- def _request(self, method, endpoint, data = None, headers = None, **kwargs):
- url = '{base}{endpoint}'.format(
- base = self.host_url,
- endpoint = endpoint
- )
- if headers:
- headers.update({'Content-Type': 'application/json'})
- else:
- headers = {'Content-Type': 'application/json'}
- if data:
- body = json.dumps(data, ensure_ascii = False)
- body = body.encode('utf-8')
- kwargs['data'] = body
- kwargs['timeout'] = kwargs.get('timeout', 15)
- result_processor = kwargs.pop('result_processor', None)
- with requests.sessions.Session() as session:
- res = session.request(
- url = url,
- method = method,
- headers = headers,
- **kwargs)
- try:
- res.raise_for_status()
- except requests.RequestException as reqe:
- raise DlbPayException(
- errorCode = res.status_code,
- errorMsg = reqe.message,
- client = self,
- request = reqe.request,
- response = reqe.response
- )
- return self._handle_result(
- res, method, url, result_processor, **kwargs
- )
- def create_pay_url(self, requestNum, amount, notify_url, front_url = None, extraInfo = None):
- # type: (str, str, str, Dict)->str
- """
- 创建支付链接
- """
- data = {
- 'customerNum': self.customerNum,
- 'shopNum': self.shopNum,
- 'requestNum': requestNum,
- 'amount': amount,
- 'source': 'API',
- 'callbackUrl': notify_url
- }
- if self.machineNum:
- data.update({'machineNum': self.machineNum})
- if front_url:
- data.update({'completeUrl': front_url})
- if extraInfo:
- data.update({'extraInfo': extraInfo})
- result = self.post(endpoint = '/v1/customer/order/payurl/create',
- data = data)
- return result['data']['url']
- def query_trade_result(self, trade_no = None, out_trade_no = None):
- if out_trade_no:
- endpoint = '/v2/customer/order/payresult/{}/{}/{}'.format(self.customerNum, self.shopNum, out_trade_no)
- elif trade_no:
- endpoint = '/v2/customer/order/payresult/{}/{}/with/{}'.format(self.customerNum, self.shopNum, trade_no)
- else:
- assert False, u'trade_no和out_trade_no至少有一个不能为空'
- result = self.get(endpoint)
- return result['data']
|