342 lines
11 KiB
Python
342 lines
11 KiB
Python
import asyncio
|
||
import re
|
||
from random import randint
|
||
from typing import Any, Dict, Optional
|
||
import urllib
|
||
import poplib
|
||
from appPublic.log import debug, exception
|
||
from appPublic.dictObject import DictObject
|
||
from appPublic.timeUtils import curDateString, timestampstr
|
||
from ahserver.serverenv import ServerEnv
|
||
from email.parser import Parser
|
||
from email.header import decode_header
|
||
from email.utils import parseaddr
|
||
from sqlor.dbpools import DBPools
|
||
from ..core import Gateway
|
||
|
||
|
||
def guess_charset(msg):
|
||
charset = msg.get_charset()
|
||
if charset is None:
|
||
content_type = msg.get('Content-Type', '').lower()
|
||
pos = content_type.find('charset=')
|
||
if pos >= 0:
|
||
charset = content_type[pos + 8:].strip()
|
||
return charset
|
||
|
||
class EmailClient:
|
||
def __init__(self, pop3_server, emailaddress, password):
|
||
self.client = poplib.POP3(pop3_server)
|
||
self.client.user(emailaddress)
|
||
self.client.pass_(password)
|
||
|
||
def stat(self):
|
||
return self.client.stat()
|
||
|
||
def mail_list(self):
|
||
resp, mails, octets = self.client.list()
|
||
|
||
def get_mail(self, index):
|
||
resp, lines, octets = self.client.retr(index)
|
||
debug(f'{resp=}, {octets=}')
|
||
msg_content = b'\r\n'.join(lines).decode('utf-8')
|
||
msg = Parser().parsestr(msg_content)
|
||
mail = {
|
||
"mailfrom": msg.get('From'),
|
||
"mailto": msg.get('To'),
|
||
"subject": msg.get('Subject'),
|
||
"date": msg.get('Date'),
|
||
"body": self.get_body(msg)
|
||
}
|
||
return DictObject(**mail)
|
||
|
||
def get_body(self, msg):
|
||
if msg.is_multipart():
|
||
parts = msg.get_payload()
|
||
content = ''
|
||
for part in parts:
|
||
content += self.get_body(part)
|
||
return content
|
||
else:
|
||
content_type = msg.get_content_type()
|
||
content = msg.get_payload(decode=True)
|
||
charset = guess_charset(msg)
|
||
if charset:
|
||
content = content.decode(charset)
|
||
else:
|
||
content = content.decode('utf-8')
|
||
return content
|
||
|
||
class TransferGateway(Gateway):
|
||
def __init__(self, from_mail="", pop3server="", email="", password="", account_no="", account_name="", bank_name="", bank_branch=""):
|
||
self.from_mail = from_mail
|
||
self.pop3server = pop3server
|
||
self.email = email
|
||
self.password = password
|
||
self.account_no = account_no
|
||
self.account_name = account_name
|
||
self.bank_name = bank_name
|
||
self.bank_branch = bank_branch
|
||
self.running = False
|
||
|
||
async def new_tcode(self, sor):
|
||
tcode = self.gen_mailcode()
|
||
sql = "select * from transfercode where tcode=${tcode}$ and status='0'"
|
||
while True:
|
||
recs = await sor.sqlExe(sql, {'tcode': tcode})
|
||
if len(recs) == 0:
|
||
break
|
||
tcode = self.gen_mailcode()
|
||
return tcode
|
||
|
||
async def create_payment(self, payload: Dict[str, Any]) -> str:
|
||
"""
|
||
返回一个可以在 H5 里直接重定向的支付宝支付 URL
|
||
"""
|
||
ns = {
|
||
"id": payload.out_trade_no,
|
||
"customerid": payload.customerid,
|
||
"amount": payload.amount,
|
||
"curdate": curDateString(),
|
||
"curtime": timestampstr(),
|
||
"status": '0'
|
||
}
|
||
ns = DictObject(**ns)
|
||
ns.account_no = self.account_no
|
||
ns.account_name = self.account_name
|
||
ns.bank_name = self.bank_name
|
||
ns.bank_branch = self.bank_branch
|
||
env = payload.request._run_ns
|
||
db = DBPools()
|
||
dbname = env.get_module_dbname('unipay')
|
||
async with db.sqlorContext(dbname) as sor:
|
||
sql = "select * from transfercode where tcode=${tcode}$ and status='0'"
|
||
ns.tcode = await self.new_tcode(sor)
|
||
await sor.C('transfercode', ns.copy())
|
||
url = env.entire_url('transfer_info.ui')
|
||
query_str = urllib.parse.urlencode(ns)
|
||
return f'{url}?{query_str}'
|
||
return None
|
||
|
||
def get_transfer_data(self, mail):
|
||
debug(f'get_transfer_data: from={mail.mailfrom} to={mail.mailto} expect_to={self.email}')
|
||
if mail.mailfrom != '95555@message.cmbchina.com':
|
||
debug(f'get_transfer_data: wrong sender')
|
||
return None
|
||
if mail.mailto != self.email:
|
||
debug(f'get_transfer_data: mailto mismatch: {mail.mailto} != {self.email}')
|
||
return None
|
||
if not mail.body.startswith('动账业务通知'):
|
||
debug(f'get_transfer_data: body not start with 动账业务通知, head={mail.body[:30]}')
|
||
return None
|
||
ns = DictObject()
|
||
match = re.search(r'交易金额\s*[:：]?\s*(\d+(?:\.\d+)?)', mail.body)
|
||
if match:
|
||
ns.amount = float(match.group(1))
|
||
debug(f'get_transfer_data: amount={ns.amount}')
|
||
match = re.search(r'摘要[:：]\s*(\d{7})(?!\d)', mail.body)
|
||
if match:
|
||
ns.code = match.group(1)
|
||
debug(f'get_transfer_data: code={ns.code}')
|
||
if not hasattr(ns, 'amount'):
|
||
debug(f'get_transfer_data: no amount found, body={mail.body[:200]}')
|
||
return None
|
||
if not hasattr(ns, 'code'):
|
||
debug(f'get_transfer_data: no code found, body={mail.body[:200]}')
|
||
return None
|
||
return ns
|
||
|
||
def gen_mailcode(self):
|
||
c = '1'
|
||
for i in range(6):
|
||
c += str(randint(0,9))
|
||
return c
|
||
|
||
async def run(self, app):
|
||
"""后台轮询邮箱，匹配转账码并完成充值入账"""
|
||
debug(f'mail check loop will starting')
|
||
self.running = True
|
||
return
|
||
while self.running:
|
||
debug(f'mail check loop running')
|
||
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':
|
||
debug(f'ignored mail from {mail.mailfrom=}')
|
||
continue
|
||
if not mail.body.startswith('动账业务通知'):
|
||
debug(f'ignored mail from {mail.mailfrom=} not start with 动账业务通知{mail.body=}')
|
||
continue
|
||
data = self.get_transfer_data(mail)
|
||
if data is None:
|
||
debug(f'ignored mail from {mail.mailfrom=} get data error {mail.body=}')
|
||
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 check_transfer(self, tcode: str, env):
|
||
"""扫描最近2天邮件，查找指定转账码并完成入账。
|
||
返回: (title, message)"""
|
||
from datetime import datetime, timedelta
|
||
cutoff = datetime.now() - timedelta(days=2)
|
||
|
||
ec = EmailClient(self.pop3server, self.email, self.password)
|
||
count, _ = ec.stat()
|
||
debug(f'check_transfer: tcode={tcode}, total {count} mails, cutoff={cutoff}')
|
||
|
||
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 ('已入账', '转账码 ' + tcode + ' 已入账 ' + str(tc.amount) + ' 元')
|
||
|
||
log_lines = []
|
||
matched = None
|
||
scanned = 0
|
||
|
||
for i in range(count, 0, -1):
|
||
try:
|
||
mail = ec.get_mail(i)
|
||
scanned += 1
|
||
idx = str(i)
|
||
|
||
# 日期过滤
|
||
mail_date_str = getattr(mail, 'date', None)
|
||
skip_date = False
|
||
if mail_date_str:
|
||
try:
|
||
from email.utils import parsedate_to_datetime
|
||
mail_date = parsedate_to_datetime(mail_date_str)
|
||
if mail_date < cutoff:
|
||
skip_date = True
|
||
except:
|
||
pass
|
||
if skip_date:
|
||
debug('mail[' + idx + '] date=' + str(mail_date_str) + ' older than 2 days, skip')
|
||
continue
|
||
|
||
from_addr = getattr(mail, 'mailfrom', '?')
|
||
subject = getattr(mail, 'subject', '?')
|
||
body_head = (getattr(mail, 'body', '') or '')[:50].replace('\n', ' ')
|
||
debug('mail[' + idx + '] date=' + str(mail_date_str) + ' from=' + str(from_addr) + ' subj=' + str(subject) + ' body=' + body_head)
|
||
|
||
if from_addr != '95555@message.cmbchina.com':
|
||
debug('mail[' + idx + '] EXIT: wrong sender')
|
||
continue
|
||
if not (getattr(mail, 'body', '') or '').startswith('动账业务通知'):
|
||
debug('mail[' + idx + '] EXIT: not 动账业务通知')
|
||
continue
|
||
|
||
data = self.get_transfer_data(mail)
|
||
if data is None:
|
||
debug('mail[' + idx + '] EXIT: get_transfer_data returned None')
|
||
continue
|
||
|
||
code = str(data.code)
|
||
amount = data.amount
|
||
debug('mail[' + idx + '] code=' + code + ' amount=' + str(amount) + ' looking_for=' + tcode)
|
||
|
||
if code == tcode:
|
||
matched = data
|
||
debug('mail[' + idx + '] *** MATCHED tcode=' + tcode + ' amount=' + str(amount) + ' ***')
|
||
break
|
||
debug('mail[' + idx + '] EXIT: code mismatch ' + code + ' != ' + tcode)
|
||
|
||
except Exception as e:
|
||
debug('mail[' + str(i) + '] ERROR: ' + str(e))
|
||
|
||
debug('check_transfer: scanned ' + str(scanned) + ' mails, matched=' + str(matched is not None))
|
||
|
||
if matched is None:
|
||
return ('尚未到账', '扫描了 ' + str(scanned) + ' 封邮件，暂未收到转账码 ' + tcode + ' 的到账通知')
|
||
|
||
# 执行入账
|
||
async with db.sqlorContext(dbname) as sor:
|
||
recs = await sor.R('transfercode', {'tcode': tcode, 'status': '0'})
|
||
if len(recs) < 1:
|
||
return ('已处理', '转账码 ' + 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 ('金额不匹配', '到账金额 ' + str(matched.amount) + ' 与充值金额 ' + str(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()
|
||
})
|
||
debug('check_transfer: ACCOUNTING DONE for ' + tcode + ' amount=' + str(matched.amount))
|
||
|
||
return ('充值成功', '转账码 ' + tcode + ' 已确认到账 ' + str(matched.amount) + ' 元')
|
||
|
||
async def query(self, out_trade_no: str) -> Dict[str, Any]:
|
||
pass
|
||
|
||
async def refund(self, *, out_trade_no: str, refund_amount: str, out_request_no: str) -> Dict[str, Any]:
|
||
pass
|
||
|
||
async def handle_notify(self, request) -> Dict[str, Any]:
|
||
pass
|