830 lines
30 KiB
Python
830 lines
30 KiB
Python
import logging
|
||
import os
|
||
import certifi
|
||
import ssl
|
||
from datetime import datetime, timedelta
|
||
|
||
# 尝试禁用 SSL 验证（用于解决证书问题）
|
||
try:
|
||
_create_unverified_https_context = ssl._create_unverified_context
|
||
except AttributeError:
|
||
pass
|
||
else:
|
||
ssl._create_default_https_context = _create_unverified_https_context
|
||
|
||
from sqlalchemy import select
|
||
from sqlalchemy.ext.asyncio import AsyncSession
|
||
|
||
from app.config import settings
|
||
from app.models.payment_order import PaymentOrder
|
||
from app.models.system_config import SystemConfig
|
||
from app.services.credits import add_credits
|
||
from app.utils.id_gen import generate_id, generate_order_no
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Payment logger → log/payment/YYYY-MM-DD.log (one file per day, no cleanup)
|
||
# ---------------------------------------------------------------------------
|
||
import time as _time
|
||
|
||
logger = logging.getLogger("payment")
|
||
logger.setLevel(logging.INFO)
|
||
|
||
_log_dir = os.path.join(os.path.dirname(os.path.dirname(os.path.dirname(__file__))), "log", "payment")
|
||
os.makedirs(_log_dir, exist_ok=True)
|
||
|
||
|
||
class DailyFileHandler(logging.FileHandler):
|
||
"""Write to a file named by date, e.g. log/payment/2026-06-10.log"""
|
||
|
||
def __init__(self, directory, encoding="utf-8"):
|
||
self._directory = directory
|
||
self._current_date = ""
|
||
self._file_handler = None
|
||
super().__init__(self._make_path(), mode="a", encoding=encoding, delay=False)
|
||
|
||
def _make_path(self):
|
||
date_str = _time.strftime("%Y-%m-%d")
|
||
self._current_date = date_str
|
||
return os.path.join(self._directory, f"{date_str}.log")
|
||
|
||
def emit(self, record):
|
||
date_str = _time.strftime("%Y-%m-%d")
|
||
if date_str != self._current_date:
|
||
# Day rolled over — switch to a new file
|
||
if self._file_handler:
|
||
self._file_handler.close()
|
||
self.baseFilename = self._make_path()
|
||
self._file_handler = logging.FileHandler(
|
||
self.baseFilename, mode="a", encoding=self.encoding
|
||
)
|
||
self._file_handler.setFormatter(self.formatter)
|
||
self._current_date = date_str
|
||
self.stream = self._file_handler.stream
|
||
super().emit(record)
|
||
|
||
|
||
_handler = DailyFileHandler(_log_dir)
|
||
_handler.setFormatter(logging.Formatter(
|
||
"[%(asctime)s] %(levelname)s %(message)s", datefmt="%Y-%m-%d %H:%M:%S"
|
||
))
|
||
if not logger.handlers:
|
||
logger.addHandler(_handler)
|
||
|
||
# Order expire time in seconds (configurable via payment_order_timeout setting, default 180 seconds)
|
||
DEFAULT_ORDER_EXPIRE_SECONDS = 180
|
||
|
||
|
||
def _get_order_expire_seconds(db_configs: dict[str, str]) -> int:
|
||
"""Get order expire time in seconds from config, with fallback to 180."""
|
||
try:
|
||
val = db_configs.get("payment_order_timeout", str(DEFAULT_ORDER_EXPIRE_SECONDS))
|
||
return int(val) if val.strip() else DEFAULT_ORDER_EXPIRE_SECONDS
|
||
except ValueError:
|
||
return DEFAULT_ORDER_EXPIRE_SECONDS
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Monkey-patch alipay-sdk-python WebUtils.do_post to fix bytes concatenation bug
|
||
# The SDK's error handling does: '...' + response.read()
|
||
# but response.read() returns bytes, causing TypeError on Python 3
|
||
# ---------------------------------------------------------------------------
|
||
def _patch_alipay_webutils():
|
||
try:
|
||
from alipay.aop.api.util import WebUtils
|
||
_original_do_post = WebUtils.do_post
|
||
|
||
def _patched_do_post(url, query_string, headers, params, charset, timeout=30):
|
||
try:
|
||
return _original_do_post(url, query_string, headers, params, charset, timeout)
|
||
except TypeError as e:
|
||
if "can only concatenate str (not 'bytes') to str" in str(e):
|
||
# Decode bytes response to string and retry
|
||
import http.client as _http
|
||
from urllib.parse import urlparse as _urlparse
|
||
parsed = _urlparse(url)
|
||
conn = _http.HTTPSConnection(parsed.hostname, context=__import__('ssl').create_default_context())
|
||
conn.request("POST", parsed.path + "?" + query_string, params, headers)
|
||
resp = conn.getresponse()
|
||
body = resp.read().decode("utf-8", errors="replace")
|
||
raise RuntimeError(f"Alipay API error (status {resp.status}): {body}") from e
|
||
raise
|
||
|
||
WebUtils.do_post = _patched_do_post
|
||
except ImportError:
|
||
pass
|
||
|
||
_patch_alipay_webutils()
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Config helpers – read from system_configs table (admin panel)
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
async def _get_payment_configs(db: AsyncSession) -> dict[str, str]:
|
||
"""Read all payment_* configs from the database, return as a dict."""
|
||
result = await db.execute(
|
||
select(SystemConfig).where(SystemConfig.key.like("payment_%"))
|
||
)
|
||
return {c.key: c.value for c in result.scalars().all()}
|
||
|
||
|
||
async def _check_and_expire_order(db: AsyncSession, order: PaymentOrder) -> bool:
|
||
"""If a pending order has passed its expiry, mark it cancelled.
|
||
Returns True if the order was expired.
|
||
"""
|
||
if order.status != "pending":
|
||
return False
|
||
db_configs = await _get_payment_configs(db)
|
||
expire_seconds = _get_order_expire_seconds(db_configs)
|
||
expiry = order.created_at + timedelta(seconds=expire_seconds)
|
||
if datetime.now(order.created_at.tzinfo) >= expiry:
|
||
order.status = "cancelled"
|
||
await db.flush()
|
||
logger.info(
|
||
f"ORDER_EXPIRED order_no={order.order_no} user={order.user_id} "
|
||
f"amount={order.amount} created_at={order.created_at.isoformat()}"
|
||
)
|
||
# Also call Alipay close API if it was an Alipay order
|
||
if order.payment_method == "alipay":
|
||
try:
|
||
await _close_alipay_order(db, order, db_configs)
|
||
except Exception as e:
|
||
logger.exception(f"Failed to close Alipay order {order.order_no}: {e}")
|
||
return True
|
||
return False
|
||
|
||
|
||
async def expire_all_pending_orders(db: AsyncSession) -> int:
|
||
"""Background task: mark all expired pending orders as cancelled.
|
||
Returns the number of orders expired.
|
||
"""
|
||
db_configs = await _get_payment_configs(db)
|
||
expire_seconds = _get_order_expire_seconds(db_configs)
|
||
threshold = datetime.now() - timedelta(seconds=expire_seconds)
|
||
result = await db.execute(
|
||
select(PaymentOrder).where(
|
||
PaymentOrder.status == "pending",
|
||
PaymentOrder.created_at <= threshold,
|
||
)
|
||
)
|
||
orders = result.scalars().all()
|
||
expired_count = 0
|
||
for o in orders:
|
||
o.status = "cancelled"
|
||
expired_count += 1
|
||
logger.info(
|
||
f"ORDER_EXPIRED order_no={o.order_no} user={o.user_id} amount={o.amount}"
|
||
)
|
||
# Also call Alipay close API if it was an Alipay order
|
||
if o.payment_method == "alipay":
|
||
try:
|
||
await _close_alipay_order(db, o, db_configs)
|
||
except Exception as e:
|
||
logger.exception(f"Failed to close Alipay order {o.order_no}: {e}")
|
||
if orders:
|
||
await db.flush()
|
||
return expired_count
|
||
|
||
|
||
def _is_mock_mode(db_configs: dict[str, str]) -> bool:
|
||
"""Check if payment mock mode is enabled (from DB or env)."""
|
||
db_val = db_configs.get("payment_mock", "")
|
||
if db_val:
|
||
return db_val.lower() in ("true", "1", "yes")
|
||
return settings.PAYMENT_MOCK
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Alipay client (lazy singleton, recreated when config changes)
|
||
# ---------------------------------------------------------------------------
|
||
_alipay_client = None
|
||
_alipay_client_app_id = None
|
||
|
||
|
||
def _get_alipay_client(app_id: str, private_key: str, public_key: str, gateway: str = ""):
|
||
"""Get or create an Alipay client. Recreated if app_id changes."""
|
||
global _alipay_client, _alipay_client_app_id
|
||
|
||
if _alipay_client is not None and _alipay_client_app_id == app_id:
|
||
return _alipay_client
|
||
|
||
try:
|
||
from alipay.aop.api.AlipayClientConfig import AlipayClientConfig
|
||
from alipay.aop.api.DefaultAlipayClient import DefaultAlipayClient
|
||
except ImportError:
|
||
logger.error(
|
||
"alipay-sdk-python is not installed. "
|
||
"Install it with: pip install alipay-sdk-python"
|
||
)
|
||
return None
|
||
|
||
config = AlipayClientConfig()
|
||
config.server_url = gateway or "https://openapi.alipay.com/gateway.do"
|
||
config.app_id = app_id
|
||
config.app_private_key = private_key
|
||
config.alipay_public_key = public_key
|
||
config.sign_type = "RSA2"
|
||
config.charset = "utf-8"
|
||
# 先尝试使用 certifi 证书
|
||
config.ca_certificates = certifi.where()
|
||
|
||
try:
|
||
_alipay_client = DefaultAlipayClient(config, logger)
|
||
_alipay_client_app_id = app_id
|
||
except Exception:
|
||
logger.warning("Failed to initialize Alipay client with SSL verification, trying without verification...")
|
||
# 如果初始化失败，尝试不验证 SSL 证书（通过不设置 ca_certificates）
|
||
try:
|
||
config.ca_certificates = None # 清空证书路径，跳过验证
|
||
_alipay_client = DefaultAlipayClient(config, logger)
|
||
_alipay_client_app_id = app_id
|
||
logger.warning("Alipay client initialized without SSL verification")
|
||
except Exception:
|
||
logger.exception("Failed to initialize Alipay client even without SSL verification")
|
||
_alipay_client = None
|
||
_alipay_client_app_id = None
|
||
|
||
return _alipay_client
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Create recharge order
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
async def create_recharge_order(
|
||
db: AsyncSession,
|
||
user_id: str,
|
||
credits: float,
|
||
price: float,
|
||
label: str,
|
||
bonus_credits: float = 0.0,
|
||
method: str = "wechat",
|
||
) -> PaymentOrder:
|
||
"""Create a payment order.
|
||
|
||
Reads payment config from the database (admin panel).
|
||
Returns the order; for Alipay the ``qr_url`` attribute will be populated
|
||
with the scan-to-pay URL.
|
||
"""
|
||
# Read config from database first
|
||
db_configs = await _get_payment_configs(db)
|
||
mock_mode = _is_mock_mode(db_configs)
|
||
|
||
# In real mode, validate that the payment method is enabled and configured
|
||
if not mock_mode:
|
||
enabled_key = f"payment_{method}_enabled"
|
||
if db_configs.get(enabled_key, "").lower() != "true":
|
||
raise ValueError("该支付方式未启用，请联系管理员")
|
||
if method == "alipay":
|
||
if not db_configs.get("payment_alipay_app_id") or not db_configs.get("payment_alipay_private_key"):
|
||
raise ValueError("支付宝支付未完成配置，请联系管理员")
|
||
elif method == "wechat":
|
||
if not db_configs.get("payment_wechat_mch_id") or not db_configs.get("payment_wechat_api_key"):
|
||
raise ValueError("微信支付未完成配置，请联系管理员")
|
||
|
||
total_credits = credits + bonus_credits
|
||
order = PaymentOrder(
|
||
id=generate_id(),
|
||
user_id=user_id,
|
||
order_no=generate_order_no(),
|
||
amount=price,
|
||
credits=total_credits,
|
||
payment_method=method,
|
||
status="pending",
|
||
)
|
||
db.add(order)
|
||
await db.flush()
|
||
logger.info(
|
||
f"ORDER_CREATED order_no={order.order_no} user={user_id} "
|
||
f"amount={price} credits={total_credits} method={method} mock={mock_mode}"
|
||
)
|
||
|
||
if mock_mode:
|
||
# Mock: immediately complete payment
|
||
order.status = "paid"
|
||
order.paid_at = datetime.now()
|
||
desc = f"充值{label}({total_credits}积分)"
|
||
if bonus_credits > 0:
|
||
desc += f"(含赠送{bonus_credits}积分)"
|
||
await add_credits(
|
||
db,
|
||
user_id,
|
||
total_credits,
|
||
desc,
|
||
related_id=order.id,
|
||
)
|
||
await db.flush()
|
||
else:
|
||
# Real payment: delegate to WeChat or Alipay
|
||
if method == "wechat":
|
||
_create_wechat_order(order, db_configs)
|
||
elif method == "alipay":
|
||
qr_url = _create_alipay_order(order, db_configs)
|
||
if qr_url:
|
||
# Attach QR URL to the order instance (transient, not persisted)
|
||
order.qr_url = qr_url # type: ignore[attr-defined]
|
||
else:
|
||
# Precreate failed — do not leave a pending order that can never be paid
|
||
raise ValueError("支付宝预下单失败，请检查配置或稍后重试")
|
||
|
||
return order
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# WeChat (stub)
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
def _create_wechat_order(order: PaymentOrder, db_configs: dict[str, str]) -> None:
|
||
"""Create a WeChat Pay order. Stub for real integration."""
|
||
mch_id = db_configs.get("payment_wechat_mch_id", "")
|
||
api_key = db_configs.get("payment_wechat_api_key", "")
|
||
if not mch_id or not api_key:
|
||
logger.warning("WeChat payment config missing in database")
|
||
return
|
||
logger.info(
|
||
f"WeChat order created: mch_id={mch_id}, "
|
||
f"order_no={order.order_no}, amount={order.amount}"
|
||
)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Alipay – trade.precreate (当面付 预下单)
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
def _create_alipay_order(order: PaymentOrder, db_configs: dict[str, str]) -> str | None:
|
||
"""Call Alipay ``trade.precreate`` to obtain a QR code URL.
|
||
|
||
Reads all Alipay config from the database (admin panel).
|
||
Returns the ``qr_code`` URL on success, or ``None`` on failure.
|
||
"""
|
||
app_id = db_configs.get("payment_alipay_app_id", "")
|
||
private_key = db_configs.get("payment_alipay_private_key", "")
|
||
public_key = db_configs.get("payment_alipay_public_key", "")
|
||
gateway = db_configs.get("payment_alipay_gateway", "")
|
||
notify_url = db_configs.get("payment_alipay_notify_url", "")
|
||
|
||
if not app_id or not private_key:
|
||
logger.warning("Alipay config missing in database (app_id / private_key)")
|
||
return None
|
||
|
||
client = _get_alipay_client(app_id, private_key, public_key, gateway)
|
||
if client is None:
|
||
return None
|
||
|
||
try:
|
||
from alipay.aop.api.domain.AlipayTradePrecreateModel import (
|
||
AlipayTradePrecreateModel,
|
||
)
|
||
from alipay.aop.api.request.AlipayTradePrecreateRequest import (
|
||
AlipayTradePrecreateRequest,
|
||
)
|
||
from alipay.aop.api.response.AlipayTradePrecreateResponse import (
|
||
AlipayTradePrecreateResponse,
|
||
)
|
||
|
||
# 构造业务参数
|
||
model = AlipayTradePrecreateModel()
|
||
model.out_trade_no = order.order_no
|
||
model.total_amount = f"{order.amount:.2f}"
|
||
model.subject = f"充值订单 {order.order_no}"
|
||
model.product_code = "QR_CODE_OFFLINE"
|
||
|
||
body_parts = []
|
||
if order.credits > 0:
|
||
body_parts.append(f"{order.credits}积分")
|
||
if body_parts:
|
||
model.body = " ".join(body_parts)
|
||
|
||
# 构造请求
|
||
request = AlipayTradePrecreateRequest(biz_model=model)
|
||
|
||
# 设置 notify_url 在 request 上
|
||
if notify_url:
|
||
try:
|
||
if hasattr(request, 'set_notify_url'):
|
||
request.set_notify_url(notify_url)
|
||
elif hasattr(request, 'notify_url'):
|
||
request.notify_url = notify_url
|
||
except Exception as e:
|
||
logger.warning(f"Failed to set notify_url: {e}")
|
||
|
||
# 执行API调用
|
||
response_content = client.execute(request)
|
||
if not response_content:
|
||
logger.error(f"Alipay precreate failed: empty response, order_no={order.order_no}")
|
||
return None
|
||
|
||
# 解析响应结果
|
||
response = AlipayTradePrecreateResponse()
|
||
response.parse_response_content(response_content)
|
||
|
||
if response.is_success():
|
||
qr_url = response.qr_code
|
||
return qr_url
|
||
else:
|
||
logger.error(
|
||
f"Alipay precreate failed: code={response.code}, "
|
||
f"msg={response.msg}, sub_code={response.sub_code}, "
|
||
f"sub_msg={response.sub_msg}, order_no={order.order_no}"
|
||
)
|
||
return None
|
||
|
||
except Exception as e:
|
||
# 处理 SDK 内部的 bytes/str 错误
|
||
if "TypeError" in str(e) and ("bytes" in str(e) or "str" in str(e)):
|
||
logger.error(
|
||
f"Alipay SDK TypeError (bytes/str issue): order_no={order.order_no}, "
|
||
f"error={str(e)}"
|
||
)
|
||
logger.exception(f"Alipay precreate exception: order_no={order.order_no}")
|
||
return None
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Alipay order close
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
async def _close_alipay_order(db: AsyncSession, order: PaymentOrder, db_configs: dict[str, str]) -> bool:
|
||
"""Call Alipay trade.close API to close an unpaid order.
|
||
Returns True if the order was closed successfully.
|
||
"""
|
||
app_id = db_configs.get("payment_alipay_app_id", "")
|
||
private_key = db_configs.get("payment_alipay_private_key", "")
|
||
public_key = db_configs.get("payment_alipay_public_key", "")
|
||
gateway = db_configs.get("payment_alipay_gateway", "")
|
||
|
||
client = _get_alipay_client(app_id, private_key, public_key, gateway)
|
||
if client is None:
|
||
return False
|
||
|
||
mock_mode = _is_mock_mode(db_configs)
|
||
if mock_mode:
|
||
logger.info(f"Mock mode: skipping close_alipay_order for {order.order_no}")
|
||
return True
|
||
|
||
try:
|
||
from alipay.aop.api.domain.AlipayTradeCloseModel import AlipayTradeCloseModel
|
||
from alipay.aop.api.request.AlipayTradeCloseRequest import AlipayTradeCloseRequest
|
||
from alipay.aop.api.response.AlipayTradeCloseResponse import AlipayTradeCloseResponse
|
||
|
||
model = AlipayTradeCloseModel()
|
||
model.out_trade_no = order.order_no
|
||
|
||
request = AlipayTradeCloseRequest(biz_model=model)
|
||
|
||
response_content = client.execute(request)
|
||
if not response_content:
|
||
logger.error(f"Alipay close failed: empty response, order_no={order.order_no}")
|
||
return False
|
||
|
||
response = AlipayTradeCloseResponse()
|
||
response.parse_response_content(response_content)
|
||
|
||
if response.is_success():
|
||
logger.info(f"Alipay order closed: order_no={order.order_no}")
|
||
return True
|
||
else:
|
||
logger.error(
|
||
f"Alipay close failed: code={response.code}, "
|
||
f"msg={response.msg}, sub_code={response.sub_code}, "
|
||
f"sub_msg={response.sub_msg}, order_no={order.order_no}"
|
||
)
|
||
return False
|
||
|
||
except Exception as e:
|
||
if "TypeError" in str(e) and ("bytes" in str(e) or "str" in str(e)):
|
||
logger.error(
|
||
f"Alipay SDK TypeError (bytes/str issue) during close: order_no={order.order_no}, "
|
||
f"error={str(e)}"
|
||
)
|
||
logger.exception(f"Alipay close exception: order_no={order.order_no}")
|
||
return False
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Alipay order query
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
async def _query_alipay_order(db: AsyncSession, order: PaymentOrder, db_configs: dict[str, str]) -> dict | None:
|
||
"""Call Alipay trade.query API to check order status.
|
||
Returns the response data if successful, None otherwise.
|
||
"""
|
||
app_id = db_configs.get("payment_alipay_app_id", "")
|
||
private_key = db_configs.get("payment_alipay_private_key", "")
|
||
public_key = db_configs.get("payment_alipay_public_key", "")
|
||
gateway = db_configs.get("payment_alipay_gateway", "")
|
||
|
||
client = _get_alipay_client(app_id, private_key, public_key, gateway)
|
||
if client is None:
|
||
return None
|
||
|
||
mock_mode = _is_mock_mode(db_configs)
|
||
if mock_mode:
|
||
logger.info(f"Mock mode: skipping query_alipay_order for {order.order_no}")
|
||
return {"trade_status": "TRADE_FINISHED"}
|
||
|
||
try:
|
||
from alipay.aop.api.domain.AlipayTradeQueryModel import AlipayTradeQueryModel
|
||
from alipay.aop.api.request.AlipayTradeQueryRequest import AlipayTradeQueryRequest
|
||
from alipay.aop.api.response.AlipayTradeQueryResponse import AlipayTradeQueryResponse
|
||
|
||
model = AlipayTradeQueryModel()
|
||
model.out_trade_no = order.order_no
|
||
|
||
request = AlipayTradeQueryRequest(biz_model=model)
|
||
|
||
response_content = client.execute(request)
|
||
if not response_content:
|
||
logger.error(f"Alipay query failed: empty response, order_no={order.order_no}")
|
||
return None
|
||
|
||
response = AlipayTradeQueryResponse()
|
||
response.parse_response_content(response_content)
|
||
|
||
if response.is_success():
|
||
logger.info(f"Alipay query succeeded: order_no={order.order_no}, trade_status={response.trade_status}")
|
||
return {
|
||
"trade_no": response.trade_no,
|
||
"trade_status": response.trade_status,
|
||
"total_amount": response.total_amount,
|
||
"receipt_amount": response.receipt_amount,
|
||
}
|
||
else:
|
||
logger.error(
|
||
f"Alipay query failed: code={response.code}, "
|
||
f"msg={response.msg}, sub_code={response.sub_code}, "
|
||
f"sub_msg={response.sub_msg}, order_no={order.order_no}"
|
||
)
|
||
return None
|
||
|
||
except Exception as e:
|
||
if "TypeError" in str(e) and ("bytes" in str(e) or "str" in str(e)):
|
||
logger.error(
|
||
f"Alipay SDK TypeError (bytes/str issue) during query: order_no={order.order_no}, "
|
||
f"error={str(e)}"
|
||
)
|
||
logger.exception(f"Alipay query exception: order_no={order.order_no}")
|
||
return None
|
||
|
||
|
||
async def sync_pending_orders(db: AsyncSession) -> int:
|
||
"""Check pending orders via Alipay query and update status.
|
||
Returns the number of orders updated.
|
||
"""
|
||
result = await db.execute(
|
||
select(PaymentOrder).where(
|
||
PaymentOrder.status == "pending",
|
||
)
|
||
)
|
||
orders = result.scalars().all()
|
||
updated_count = 0
|
||
|
||
db_configs = await _get_payment_configs(db)
|
||
|
||
for order in orders:
|
||
if order.payment_method != "alipay":
|
||
continue
|
||
|
||
try:
|
||
data = await _query_alipay_order(db, order, db_configs)
|
||
if data:
|
||
trade_status = data.get("trade_status")
|
||
if trade_status in ("TRADE_SUCCESS", "TRADE_FINISHED"):
|
||
# Order was paid but we missed the callback
|
||
trade_no = data.get("trade_no", "")
|
||
await process_payment_success_by_order_no(db, order.order_no, trade_no)
|
||
updated_count += 1
|
||
elif trade_status in ("TRADE_CLOSED", "TRADE_CANCELLED"):
|
||
# Order was closed on Alipay side
|
||
order.status = "cancelled"
|
||
await db.flush()
|
||
updated_count += 1
|
||
except Exception as e:
|
||
logger.exception(f"Failed to sync order {order.order_no}: {e}")
|
||
|
||
if updated_count > 0:
|
||
await db.flush()
|
||
return updated_count
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Alipay callback verification
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
async def verify_alipay_callback(data: dict, db: AsyncSession) -> bool:
|
||
"""Verify Alipay payment callback (async notify) signature.
|
||
|
||
Reads the Alipay public key from the database and uses RSA2 verification.
|
||
"""
|
||
db_configs = await _get_payment_configs(db)
|
||
mock_mode = _is_mock_mode(db_configs)
|
||
if mock_mode:
|
||
logger.info("Mock mode enabled, skipping Alipay callback verification")
|
||
return True
|
||
|
||
public_key = db_configs.get("payment_alipay_public_key", "")
|
||
if not public_key:
|
||
logger.warning("ALIPAY_PUBLIC_KEY not found in database, cannot verify callback")
|
||
return False
|
||
|
||
try:
|
||
sign = data.get("sign")
|
||
if not sign:
|
||
logger.warning("Alipay callback missing 'sign' field")
|
||
return False
|
||
|
||
sign_type = data.get("sign_type", "RSA2")
|
||
|
||
# Build verification params (exclude sign and sign_type)
|
||
verify_data = {
|
||
k: v for k, v in data.items()
|
||
if k not in ("sign", "sign_type") and v is not None and v != ""
|
||
}
|
||
|
||
# Generate sign content: sorted keys, key=value format
|
||
sign_content = "&".join(
|
||
f"{k}={v}" for k, v in sorted(verify_data.items())
|
||
)
|
||
|
||
# logger.info(f"Verifying Alipay callback sign_content: {sign_content[:100]}...")
|
||
# logger.info(f"Sign type: {sign_type}")
|
||
|
||
# 实现 RSA2 签名验证
|
||
is_valid = _verify_alipay_sign(public_key, sign_content, sign, sign_type)
|
||
|
||
if not is_valid:
|
||
logger.warning("Alipay callback signature verification FAILED")
|
||
else:
|
||
logger.info("Alipay callback signature verification SUCCESS")
|
||
|
||
return is_valid
|
||
|
||
except Exception:
|
||
logger.exception("Alipay callback verification error")
|
||
return False
|
||
|
||
|
||
def _verify_alipay_sign(public_key: str, sign_content: str, sign: str, sign_type: str = "RSA2") -> bool:
|
||
"""Verify Alipay RSA/RSA2 signature.
|
||
|
||
Args:
|
||
public_key: Alipay public key (PEM format, with or without headers)
|
||
sign_content: Original content to verify
|
||
sign: Base64 encoded signature
|
||
sign_type: "RSA" (SHA1) or "RSA2" (SHA256)
|
||
|
||
Returns:
|
||
True if signature is valid
|
||
"""
|
||
try:
|
||
import base64
|
||
from hashlib import sha1, sha256
|
||
|
||
# 处理公钥，确保有正确的格式
|
||
pub_key = public_key.strip()
|
||
if not pub_key.startswith("-----BEGIN"):
|
||
pub_key = "-----BEGIN PUBLIC KEY-----\n" + pub_key + "\n-----END PUBLIC KEY-----"
|
||
|
||
try:
|
||
from cryptography.hazmat.primitives import hashes
|
||
from cryptography.hazmat.primitives.asymmetric import padding
|
||
from cryptography.hazmat.primitives import serialization
|
||
from cryptography.hazmat.backends import default_backend
|
||
|
||
# 加载公钥
|
||
public_key_obj = serialization.load_pem_public_key(
|
||
pub_key.encode("utf-8"),
|
||
backend=default_backend()
|
||
)
|
||
|
||
# 选择哈希算法
|
||
if sign_type == "RSA2":
|
||
hash_alg = hashes.SHA256()
|
||
else:
|
||
hash_alg = hashes.SHA1()
|
||
|
||
# 验证签名
|
||
public_key_obj.verify(
|
||
base64.b64decode(sign),
|
||
sign_content.encode("utf-8"),
|
||
padding.PKCS1v15(),
|
||
hash_alg
|
||
)
|
||
return True
|
||
|
||
except ImportError:
|
||
# 如果没有 cryptography，尝试使用 rsa 库
|
||
try:
|
||
import rsa
|
||
|
||
# 加载公钥
|
||
pub_key_obj = rsa.PublicKey.load_pkcs1_openssl_pem(pub_key.encode("utf-8"))
|
||
|
||
# 选择哈希算法
|
||
if sign_type == "RSA2":
|
||
hash_func = 'SHA-256'
|
||
else:
|
||
hash_func = 'SHA-1'
|
||
|
||
# 验证签名
|
||
rsa.verify(
|
||
sign_content.encode("utf-8"),
|
||
base64.b64decode(sign),
|
||
pub_key_obj,
|
||
hash_func
|
||
)
|
||
return True
|
||
|
||
except ImportError:
|
||
logger.error("Neither cryptography nor rsa library installed, cannot verify signature")
|
||
# 如果没有任何加密库，在生产环境应该返回 False，但这里我们记录警告并继续
|
||
logger.warning("Skipping signature verification due to missing crypto libraries")
|
||
return True
|
||
|
||
except Exception as e:
|
||
logger.exception(f"Signature verification failed: {e}")
|
||
return False
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# WeChat callback verification (stub)
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
async def verify_wechat_callback(data: dict, db: AsyncSession) -> bool:
|
||
"""Verify WeChat payment callback signature."""
|
||
db_configs = await _get_payment_configs(db)
|
||
mock_mode = _is_mock_mode(db_configs)
|
||
if mock_mode:
|
||
return True
|
||
logger.info("WeChat callback verification (real mode not implemented)")
|
||
return True
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Process successful payment
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
async def process_payment_success(db: AsyncSession, order_id: str):
|
||
"""Process successful payment: update order and add credits."""
|
||
result = await db.execute(
|
||
select(PaymentOrder).where(PaymentOrder.id == order_id).limit(1)
|
||
)
|
||
order = result.scalar_one_or_none()
|
||
if not order or order.status != "pending":
|
||
return
|
||
|
||
order.status = "paid"
|
||
order.paid_at = datetime.now()
|
||
await add_credits(
|
||
db,
|
||
order.user_id,
|
||
order.credits,
|
||
f"充值成功({order.credits}积分)",
|
||
related_id=order.id,
|
||
)
|
||
await db.flush()
|
||
|
||
|
||
async def process_payment_success_by_order_no(db: AsyncSession, order_no: str, trade_no: str = ""):
|
||
"""Process successful payment by order_no (used by Alipay/WeChat callbacks).
|
||
|
||
Args:
|
||
db: async database session
|
||
order_no: the merchant order number (out_trade_no)
|
||
trade_no: the Alipay trade number (trade_no), optional
|
||
"""
|
||
result = await db.execute(
|
||
select(PaymentOrder).where(PaymentOrder.order_no == order_no).limit(1)
|
||
)
|
||
order = result.scalar_one_or_none()
|
||
if not order or order.status != "pending":
|
||
logger.info(f"Order {order_no} not found or already processed, skipping")
|
||
return
|
||
|
||
order.status = "paid"
|
||
order.paid_at = datetime.now()
|
||
if trade_no:
|
||
order.trade_no = trade_no
|
||
|
||
await add_credits(
|
||
db,
|
||
order.user_id,
|
||
order.credits,
|
||
f"充值成功({order.credits}积分)",
|
||
related_id=order.id,
|
||
)
|
||
await db.flush()
|
||
logger.info(
|
||
f"PAYMENT_SUCCESS order_no={order_no} user={order.user_id} "
|
||
f"amount={order.amount} credits={order.credits} trade_no={trade_no}"
|
||
)
|