fix: 实现转账邮件轮询后台任务，自动匹配转账码并完成入账

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:

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'
+		if mail.mailfrom != '95555@message.cmbchina.com':
-		assert mail.mailto == self.email
+			return None
-		assert mail.body.startswith('动账业务通知')
+		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
+		if not hasattr(ns, 'amount') or not hasattr(ns, 'code'):
-		assert ns.code
+			return None
 		return ns
 	
 	def gen_mailcode(self):
@@ -137,11 +140,68 @@ class TransferGateway(Gateway):
 			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)
+			try:
-			mail = ec.get_mail(1)
+				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