refactor(transfer): move email scan logic to check_transfer method, DSPY only calls provider

unipay/providers/transfer.py

@@ -209,6 +209,74 @@ class TransferGateway(Gateway):
 				exception(f'transfer poll error: {e}')
 			await asyncio.sleep(30)
 
+	async def check_transfer(self, tcode: str, env):
+		"""手动检查指定转账码的邮件并完成入账。
+		返回: (title, message)"""
+		ec = EmailClient(self.pop3server, self.email, self.password)
+		count, _ = ec.stat()
+		debug(f'check_transfer: tcode={tcode}, got {count} mails')
+
+		db = DBPools()
+		dbname = env.get_module_dbname('unipay')
+
+		# 检查是否已入账
+		async with db.sqlorContext(dbname) as sor:
+			recs = await sor.R('transfercode', {'tcode': tcode})
+			if recs and recs[0].status == '1':
+				tc = recs[0]
+				return ('已入账', f'转账码 {tcode} 已入账 {tc.amount} 元')
+
+		# 扫描邮件
+		matched = None
+		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
+				if str(data.code) == tcode:
+					matched = data
+					break
+			except Exception as e:
+				debug(f'check_transfer: skip mail {i}: {e}')
+
+		if matched is None:
+			return ('尚未到账', f'暂未收到转账码 {tcode} 的到账通知')
+
+		# 执行入账
+		async with db.sqlorContext(dbname) as sor:
+			recs = await sor.R('transfercode', {'tcode': tcode, 'status': '0'})
+			if len(recs) < 1:
+				return ('已处理', f'转账码 {tcode} 已处理或不存在')
+
+			tc = recs[0]
+			diff = abs(float(matched.amount) - float(tc.amount))
+			if diff > 0.01:
+				await sor.U('transfercode', {'id': tc.id, 'status': '9',
+					'remark': '金额不匹配: 到账' + str(matched.amount) + ' 期望' + str(tc.amount)})
+				return ('金额不匹配', f'到账金额 {matched.amount} 与充值金额 {tc.amount} 不一致')
+
+			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()
+				})
+
+			return ('充值成功', f'转账码 {tcode} 已确认到账 {matched.amount} 元')
+
 	async def query(self, out_trade_no: str) -> Dict[str, Any]:
 		pass
 	

wwwroot/transfer_check.dspy

@@ -1,91 +1,15 @@
-"""扫描邮箱检查指定转账码是否已到账，如匹配则入账"""
+"""检查指定转账码是否已到账并完成入账"""
 tcode = str(params_kw.tcode)
 env = request._run_ns
-debug(f'transfer_check.dspy: tcode={tcode}')
+debug('transfer_check.dspy: tcode=' + tcode)
 
 try:
     provider = env.PROVIDERS.get('transfer')
     if provider is None:
         return {"widgettype": "Message", "options": {"timeout": 3, "title": "错误", "message": "转账渠道未初始化"}}
 
-    db = DBPools()
+    title, msg = await provider.check_transfer(tcode, env)
-    dbname = get_module_dbname('unipay')
+    return {"widgettype": "Message", "options": {"timeout": 5, "title": title, "message": msg}}
-
-    # 先检查是否已入账
-    async with db.sqlorContext(dbname) as sor:
-        recs = await sor.R('transfercode', {'tcode': tcode})
-        if recs and recs[0].status == '1':
-            tc = recs[0]
-            return {"widgettype": "Message", "options": {
-                "timeout": 5, "title": "已入账",
-                "message": f"转账码 {tcode} 已入账 {tc.amount} 元"
-            }}
-
-    # 扫描邮件匹配转账码
-    ec = EmailClient(provider.pop3server, provider.email, provider.password)
-    count, _ = ec.stat()
-    debug(f'transfer_check: got {count} mails, searching for tcode={tcode}')
-
-    matched = None
-    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 = provider.get_transfer_data(mail)
-            if data is None:
-                continue
-            if str(data.code) == tcode:
-                matched = data
-                break
-        except Exception as e:
-            debug(f'transfer_check: skip mail {i}: {e}')
-
-    if matched is None:
-        return {"widgettype": "Message", "options": {
-            "timeout": 5, "title": "尚未到账",
-            "message": f"暂未收到转账码 {tcode} 的到账通知，请确认已转账并稍后重试"
-        }}
-
-    # 匹配到邮件，执行入账
-    async with db.sqlorContext(dbname) as sor:
-        recs = await sor.R('transfercode', {'tcode': tcode, 'status': '0'})
-        if len(recs) < 1:
-            return {"widgettype": "Message", "options": {
-                "timeout": 3, "title": "已处理", "message": f"转账码 {tcode} 已处理或不存在"
-            }}
-
-        tc = recs[0]
-        diff = abs(float(matched.amount) - float(tc.amount))
-        if diff > 0.01:
-            await sor.U('transfercode', {'id': tc.id, 'status': '9',
-                'remark': f'金额不匹配: 到账{matched.amount} 期望{tc.amount}'})
-            return {"widgettype": "Message", "options": {
-                "timeout": 5, "title": "金额不匹配",
-                "message": f"到账金额 {matched.amount} 与充值金额 {tc.amount} 不一致，请联系客服"
-            }}
-
-        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()
-            })
-
-        return {"widgettype": "Message", "options": {
-            "timeout": 5, "title": "充值成功",
-            "message": f"转账码 {tcode} 已确认到账 {matched.amount} 元，充值完成"
-        }}
 
 except Exception as e:
     exception('transfer_check.dspy error: ' + str(e))