# -*- coding: utf-8 -*- # !/usr/bin/env python import logging import time from django.core.handlers.wsgi import WSGIRequest from django.http import HttpResponse from django.utils.module_loading import import_string from typing import TYPE_CHECKING, cast from apilib.systypes import StrEnum, Singleton from apps import lockCache from apps.web.core import PayAppType from apps.web.core.payment import PaymentGateway from apps.web.exceptions import UserServerException if TYPE_CHECKING: from apps.web.dealer.models import DealerRechargeRecord, RefundDealerRechargeRecord from apps.web.user.models import RechargeRecord, RefundMoneyRecord from typing import Union, Optional, Callable, Dict import six RechargeRecordT = Union[DealerRechargeRecord, RechargeRecord] RefundRecordT = Union[RefundDealerRechargeRecord, RefundMoneyRecord] from apps.web.core.payment.type_checking import PaymentGatewayT logger = logging.getLogger(__name__) class PayNotifyAction(StrEnum): NeedHandle = 'needHandle' NoHandle = 'noHandle' Unknown = 'unknown' class OrderCacheMgr(object): KEY = '{}-{}-lock' def __init__(self, record_or_id, cls_name = None): # type: (Union[RechargeRecordT, basestring], Optional[str])->None if isinstance(record_or_id, basestring): assert cls_name, 'cls_name must not be none if record is string' self.key = self.KEY.format(cls_name, record_or_id) else: assert not cls_name, 'cls_name must be none if record is object' self.key = self.KEY.format(record_or_id.__class__.__name__, str(record_or_id.id)) def initial(self): try: return lockCache.set(self.key, '0', 25 * 3600) except Exception as e: logger.exception(e) return 0 def check_and_set_done(self): # type: ()->bool new_value = lockCache.incr(self.key) if int(new_value) >= 2: return True else: return False def has_done(self): # type: ()->bool new_value = lockCache.get(self.key, None) if not new_value: return False if int(new_value) >= 1: return True else: return False class PayRecordPoller(object): def __init__(self, record_id, interval, total_count, record_cls, post_pay): # type: (str, int, int, six.class_types, Callable)->None self.record_id = record_id self.interval = interval self.total_count = total_count self.record_cls = record_cls self.post_pay = post_pay def update_record(self, record, **payload): # type: (RechargeRecordT, Dict)->bool raise NotImplementedError('sub class must implement update_record') def action_of_pay(self, payment_gateway, record): # type: (PaymentGateway, RechargeRecordT)->(PayNotifyAction, str) raise NotImplementedError('sub class must implement action_of_pay') def start(self): logger.debug('poll record pay status.'.format(self.record_id)) count = 0 order_cache_mgr = OrderCacheMgr(self.record_id, self.record_cls.__name__) if order_cache_mgr.has_done(): logger.debug('record has done.'.format(self.record_id)) return record = self.record_cls.objects(id = self.record_id).first() # type: RechargeRecordT if not record: logger.error('record is not exist.'.format(self.record_id)) return if record.is_success: logger.error('record is success.'.format(self.record_id, record.orderNo)) return payment_gateway = PaymentGateway.from_gateway_key( record.my_gateway, record.pay_gateway_key, record.pay_app_type) # type: PaymentGatewayT while count < self.total_count: try: if count != 0 and order_cache_mgr.has_done(): logger.debug('record has done.'.format(self.record_id, record.orderNo)) return action, query_trade_result = self.action_of_pay(payment_gateway, record) logger.debug( 'record trade info: trade result = {}, action = {}.'.format( self.record_id, record.orderNo, str(query_trade_result), action)) if action == PayNotifyAction.NoHandle: return if action == PayNotifyAction.Unknown: continue try: if order_cache_mgr.check_and_set_done(): logger.debug( 'record check has done.'.format(self.record_id, record.orderNo)) return except Exception as e: logger.error( 'record cache key is not exist or exception. exception = {}, key = {}'.format( self.record_id, record.orderNo, str(e), order_cache_mgr.key)) modified = self.update_record(record, **query_trade_result) if not modified: logger.debug( 'record has been modified.'.format(self.record_id, record.orderNo)) return return self.post_pay(record) except Exception as e: logger.exception( 'record poll trade status failure. e = {}'.format(self.record_id, record.orderNo, str(e))) finally: count = count + 1 time.sleep(self.interval) class PayNotifier(object): def __init__(self, request, record_cls_factory): # type: (WSGIRequest, callable)->None self.payload = self.parse_request(request) self.record_cls = record_cls_factory(self.out_trade_no) def parse_request(self, request): # type: (WSGIRequest)->dict raise NotImplementedError('sub class must implement update_record') @property def out_trade_no(self): raise AttributeError('no out_trade_no attr.') def update_record(self, record, **payload): # type: (RechargeRecordT, Dict)->bool raise NotImplementedError('sub class must implement update_record') def do(self, post_pay): # type: (callable)->HttpResponse raise NotImplementedError('sub class must implement do') class PayPullUpFactoryIntf(object): @classmethod def create(cls, payment_gateway, record, **payload): # type: (PaymentGatewayT, RechargeRecordT, dict)->PayPullUp raise NotImplementedError('sub class must implement create') class PayPullUp(object): def __init__(self, openId, payment_gateway, record, **payload): # type: (basestring, PaymentGatewayT, RechargeRecordT, dict)->None self.openId = openId self.record = record self.payment_gateway = payment_gateway self.payload = payload @classmethod def create(cls, factory_cls, payment_gateway, record, **payload): # type: (cast(PayPullUpFactoryIntf, None), PaymentGatewayT, RechargeRecordT, dict)->PayPullUp return factory_cls.create(payment_gateway, record, **payload) def do(self): # type: ()->HttpResponse raise NotImplementedError('sub class must implement do') class PayManager(Singleton): def __init__(self): super(PayManager, self).__init__() self.map_dict = { PayAppType.WECHAT: { 'poller': import_string('apps.web.common.transaction.pay.wechat.WechatPayRecordPoller'), 'notifier': import_string('apps.web.common.transaction.pay.wechat.WechatPayNotifier'), 'pullup': import_string('apps.web.common.transaction.pay.wechat.WechatPullUp'), }, PayAppType.ALIPAY: { 'poller': import_string('apps.web.common.transaction.pay.alipay.AliPayRecordPoller'), 'notifier': import_string('apps.web.common.transaction.pay.alipay.AliPayNotifier'), 'pullup': import_string('apps.web.common.transaction.pay.alipay.AlipayPullUp') } } def get_poller(self, pay_app_type): assert pay_app_type in self.map_dict, 'not register pay app type' return self.map_dict[pay_app_type]['poller'] def get_notifier(self, pay_app_type): assert pay_app_type in self.map_dict, 'not register pay app type' return self.map_dict[pay_app_type]['notifier'] def get_pull_up(self, pay_app_type): if pay_app_type not in self.map_dict: raise UserServerException(u'第三方支付配置错误，请联系平台客服(1002)') return self.map_dict[pay_app_type]['pullup']