diff --git a/unipay/init.py b/unipay/init.py index f615147..4a97283 100644 --- a/unipay/init.py +++ b/unipay/init.py @@ -1,5 +1,6 @@ # init.py import os +import asyncio import traceback from appPublic.log import debug,exception from ahserver.configuredServer import add_startup @@ -245,6 +246,9 @@ def load_unipay(): PROVIDERS[name] = get_provider(name, conf) if PROVIDERS[name] is not None: print(f"[unipay] {name} 初始化成功") + if name == 'transfer': + asyncio.ensure_future(PROVIDERS[name].run()) + print(f"[unipay] transfer 邮件轮询已启动") else: print(f"[unipay] {name} 初始化返回 None(渠道 disabled)") except Exception as e: diff --git a/unipay/providers/transfer.py b/unipay/providers/transfer.py index 777d2e6..8165f2b 100644 --- a/unipay/providers/transfer.py +++ b/unipay/providers/transfer.py @@ -1,9 +1,10 @@ +import asyncio import re from random import randint from typing import Any, Dict, Optional import urllib import poplib -from appPublic.log import debug +from appPublic.log import debug, exception from appPublic.dictObject import DictObject from appPublic.timeUtils import curDateString, timestampstr from email.parser import Parser @@ -116,19 +117,21 @@ class TransferGateway(Gateway): return None def get_transfer_data(self, mail): - assert mail.mailfrom == '95555@message.cmbchina.com' - assert mail.mailto == self.email - assert mail.body.startswith('动账业务通知') + if mail.mailfrom != '95555@message.cmbchina.com': + return None + if mail.mailto != self.email: + return None + if not mail.body.startswith('动账业务通知'): + return None ns = DictObject() match = re.search(r'交易金额\s*[::]?\s*(\d+(?:\.\d+)?)', mail.body) if match: ns.amount = float(match.group(1)) - p = r'摘要[::]\s*(\d{7})(?!\d)' match = re.search(r'摘要[::]\s*(\d{7})(?!\d)', mail.body) if match: ns.code = match.group(1) - assert ns.amount - assert ns.code + if not hasattr(ns, 'amount') or not hasattr(ns, 'code'): + return None return ns def gen_mailcode(self): @@ -136,13 +139,70 @@ class TransferGateway(Gateway): for i in range(6): c += str(randint(0,9)) return c - - def run(self): + + async def run(self): + """后台轮询邮箱,匹配转账码并完成充值入账""" + from ahserver.serverenv import ServerEnv self.running = True while self.running: - ec = EmailClient(self.pop3server, self.email, self.password) - mail = ec.get_mail(1) - + try: + ec = EmailClient(self.pop3server, self.email, self.password) + count, _ = ec.stat() + if count == 0: + await asyncio.sleep(30) + continue + for i in range(count, 0, -1): + try: + mail = ec.get_mail(i) + if mail.mailfrom != '95555@message.cmbchina.com': + continue + if not mail.body.startswith('动账业务通知'): + continue + data = self.get_transfer_data(mail) + if data is None: + continue + debug(f'transfer mail matched: code={data.code}, amount={data.amount}') + + env = ServerEnv() + db = DBPools() + dbname = env.get_module_dbname('unipay') + async with db.sqlorContext(dbname) as sor: + recs = await sor.R('transfercode', {'tcode': data.code, 'status': '0'}) + if len(recs) < 1: + debug(f'tcode {data.code} not found or already processed') + continue + tc = recs[0] + if abs(float(tc.amount) - data.amount) > 0.01: + debug(f'amount mismatch: tc={tc.amount}, mail={data.amount}') + continue + + biz_date = await env.get_business_date(sor) + await sor.U('transfercode', {'id': tc.id, 'status': '1'}) + + recs2 = await sor.R('payment_log', {'id': tc.id, 'payment_status': '0'}) + if len(recs2) > 0: + plog = recs2[0] + await env.recharge_accounting(sor, + plog.customerid, + 'RECHARGE', + plog.id, + biz_date, + plog.amount_total, + plog.pay_feerate) + await sor.U('payment_log', { + 'id': plog.id, + 'payment_status': '1', + 'payed_timestamp': timestampstr() + }) + debug(f'transfer recharge done: {tc.id}, amount={data.amount}') + ec.client.dele(i) + except Exception as e2: + exception(f'transfer mail[{i}] error: {e2}') + ec.client.quit() + except Exception as e: + exception(f'transfer poll error: {e}') + await asyncio.sleep(30) + async def query(self, out_trade_no: str) -> Dict[str, Any]: pass