kevin/life-echo
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactor: 优化后端支付与微信支付

Browse Source 
- 优化payment/config.py、wechat_pay.py
- 优化routers/payment.py、plans.py、quota.py、websocket.py
- 更新main.py、.env.production

Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
iammm0
2026-02-11 16:06:15 +08:00
parent 240a184da8
commit 44b405d647
8 changed files with 274 additions and 38 deletions
Download Patch File Download Diff File Expand all files Collapse all files

31
api/payment/config.py
View File

@@ -20,6 +20,10 @@ class WeChatPayConfig:
private_key: str = "" # 商户 API 私钥内容（PEM 字符串，可放环境变量，与 private_key_path 二选一）
cert_serial_no: str = "" # 商户证书序列号
notify_url: str = "" # 支付结果回调地址
# 平台公钥模式（二选一）：当环境无法访问 api.mch.weixin.qq.com 时使用，无需拉取平台证书
platform_public_key: str = "" # 微信支付平台公钥 PEM 内容（与 platform_public_key_path 二选一）
platform_public_key_path: str = "" # 平台公钥文件路径
platform_public_key_id: str = "" # 微信支付平台公钥 ID（与 platform_public_key 同时配置）
@property
def is_configured(self) -> bool:
@@ -29,6 +33,12 @@ class WeChatPayConfig:
has_key, self.cert_serial_no, self.notify_url
])
@property
def use_platform_public_key(self) -> bool:
"""是否使用平台公钥模式（无需访问微信拉取证书）"""
has_pub = bool(self.platform_public_key.strip()) or bool(self.platform_public_key_path.strip())
return has_pub and bool(self.platform_public_key_id.strip())
@dataclass
class AlipayConfig:
@@ -58,19 +68,28 @@ class PaymentConfig:
@classmethod
def from_env(cls) -> "PaymentConfig":
"""从环境变量加载配置"""
# 微信私钥：优先使用 WECHAT_PAY_PRIVATE_KEY（PEM 内容），否则用 WECHAT_PAY_PRIVATE_KEY_PATH（文件路径）
# 微信私钥：优先使用 WECHAT_PAY_PRIVATE_KEY_PATH（文件），避免 .env 中长 PEM 的转义问题；否则用 WECHAT_PAY_PRIVATE_KEY
wechat_private_key_path = os.getenv("WECHAT_PAY_PRIVATE_KEY_PATH", "").strip()
wechat_private_key = os.getenv("WECHAT_PAY_PRIVATE_KEY", "").strip()
if wechat_private_key and "\\n" in wechat_private_key:
# 去除可能被解析进来的引号与 BOM
if wechat_private_key:
wechat_private_key = wechat_private_key.strip('"').strip("'").lstrip("\ufeff")
wechat_private_key = wechat_private_key.replace("\\n", "\n")
wechat_platform_pub = os.getenv("WECHAT_PAY_PLATFORM_PUBLIC_KEY", "").strip()
if wechat_platform_pub and "\\n" in wechat_platform_pub:
wechat_platform_pub = wechat_platform_pub.replace("\\n", "\n")
config = cls(
wechat=WeChatPayConfig(
app_id=os.getenv("WECHAT_PAY_APP_ID", ""),
mch_id=os.getenv("WECHAT_PAY_MCH_ID", ""),
api_v3_key=os.getenv("WECHAT_PAY_API_V3_KEY", ""),
private_key_path=os.getenv("WECHAT_PAY_PRIVATE_KEY_PATH", ""),
private_key=wechat_private_key,
private_key_path=wechat_private_key_path,
private_key=wechat_private_key if not wechat_private_key_path else "", # 有路径时不再用 env 内容
cert_serial_no=os.getenv("WECHAT_PAY_CERT_SERIAL_NO", ""),
notify_url=os.getenv("WECHAT_PAY_NOTIFY_URL", ""),
platform_public_key=wechat_platform_pub,
platform_public_key_path=os.getenv("WECHAT_PAY_PLATFORM_PUBLIC_KEY_PATH", "").strip(),
platform_public_key_id=os.getenv("WECHAT_PAY_PLATFORM_PUBLIC_KEY_ID", "").strip(),
),
alipay=AlipayConfig(
app_id=os.getenv("ALIPAY_APP_ID", ""),
@@ -84,7 +103,9 @@ class PaymentConfig:
# 日志输出配置状态
if config.wechat.is_configured:
logger.info(f"微信支付配置已加载: APP_ID={config.wechat.app_id}, MCH_ID={config.wechat.mch_id}")
mode = "平台公钥模式" if config.wechat.use_platform_public_key else "平台证书模式"
key_src = "WECHAT_PAY_PRIVATE_KEY(环境变量)" if config.wechat.private_key.strip() else "WECHAT_PAY_PRIVATE_KEY_PATH(文件)"
logger.info(f"微信支付配置已加载: APP_ID={config.wechat.app_id}, MCH_ID={config.wechat.mch_id}, 模式={mode}, 商户私钥={key_src}")
else:
logger.warning("微信支付配置不完整，微信支付将不可用")

108
api/payment/wechat_pay.py
View File

@@ -8,6 +8,15 @@ import os
import time
from typing import Optional, Dict, Any
# #region agent log
def _debug_log(location: str, message: str, data: dict, hypothesis_id: str):
try:
with open("/Users/zuckxu/PycharmProjects/life-echo/.cursor/debug.log", "a", encoding="utf-8") as f:
f.write(json.dumps({"id": f"log_{int(time.time()*1000)}", "timestamp": int(time.time() * 1000), "location": location, "message": message, "data": data, "hypothesisId": hypothesis_id}, ensure_ascii=False) + "\n")
except Exception:
pass
# #endregion
from .config import WeChatPayConfig
from .schemas import PaymentResult, NotifyResult, PaymentStatus
from .exceptions import PaymentConfigError, PaymentCreateError, PaymentNotifyError, PaymentQueryError
@@ -17,18 +26,40 @@ logger = logging.getLogger(__name__)
def _normalize_pem_key(key: str) -> str:
"""
规范化 PEM 私钥字符串，便于 wechatpayv3 正确解析。
规范化 PEM 私钥/公钥字符串，便于 wechatpayv3 正确解析。
- 去除 BOM、首尾引号
- 环境变量中常将换行写成字面量 \\n，需转为真实换行
- 统一为 \\n 换行，去除 \\r
"""
if not key or not key.strip():
return key
key = key.strip()
key = key.strip().lstrip("\ufeff").strip('"').strip("'")
key = key.replace("\\n", "\n")
key = key.replace("\r\n", "\n").replace("\r", "\n")
return key
def _resolve_key_path(key_path: str) -> str:
"""解析私钥/公钥文件路径：相对路径时优先相对于 api 目录，避免受当前工作目录影响。"""
if not key_path or not key_path.strip():
return key_path
key_path = key_path.strip()
if os.path.isabs(key_path):
return key_path
# 相对路径：先相对于当前工作目录，若文件不存在则相对于 api 目录（本包所在目录的上一级）
abs_cwd = os.path.abspath(key_path)
if os.path.isfile(abs_cwd):
return abs_cwd
try:
api_dir = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
abs_api = os.path.normpath(os.path.join(api_dir, key_path))
if os.path.isfile(abs_api):
return abs_api
except Exception:
pass
return abs_cwd
class WeChatPayClient:
"""微信支付客户端"""
@@ -36,6 +67,10 @@ class WeChatPayClient:
self._config = config
self._client = None
def ensure_client(self):
"""供外部在后台线程中提前触发初始化，避免首次下单时在预支付超时窗口内做耗时 init。"""
self._ensure_client()
def _ensure_client(self):
"""确保微信支付客户端已初始化"""
if not self._config.is_configured:
@@ -43,20 +78,20 @@ class WeChatPayClient:
if self._client is None:
try:
# #region agent log
_debug_log("payment/wechat_pay.py:_ensure_client", "wechat_ensure_client_enter", {}, "H_init")
# #endregion
from wechatpayv3 import WeChatPay, WeChatPayType
# 商户私钥：优先用配置中的 PEM 字符串，否则从文件读取
if self._config.private_key and self._config.private_key.strip():
private_key = self._config.private_key.strip()
else:
key_path = self._config.private_key_path.strip()
if not key_path:
raise PaymentConfigError("未配置微信支付商户私钥：请设置 WECHAT_PAY_PRIVATE_KEY 或 WECHAT_PAY_PRIVATE_KEY_PATH")
# 若为相对路径，相对于当前工作目录解析（建议使用绝对路径或与 main.py 同级的路径）
if not os.path.isabs(key_path):
key_path = os.path.abspath(key_path)
# 商户私钥：优先用 WECHAT_PAY_PRIVATE_KEY_PATH 从文件读取（推荐），否则用 WECHAT_PAY_PRIVATE_KEY
if self._config.private_key_path and self._config.private_key_path.strip():
key_path = _resolve_key_path(self._config.private_key_path)
with open(key_path, "r", encoding="utf-8") as f:
private_key = f.read()
elif self._config.private_key and self._config.private_key.strip():
private_key = self._config.private_key.strip()
else:
raise PaymentConfigError("未配置微信支付商户私钥：请设置 WECHAT_PAY_PRIVATE_KEY_PATH（推荐）或 WECHAT_PAY_PRIVATE_KEY")
# 统一 PEM 格式：换行符规范化、字面量 \n 转真实换行（环境变量或部分编辑器会写成 \n）
private_key = _normalize_pem_key(private_key)
@@ -65,8 +100,12 @@ class WeChatPayClient:
"微信支付商户私钥格式错误：需为 PEM 格式（以 -----BEGIN PRIVATE KEY----- 开头）。"
"若使用 WECHAT_PAY_PRIVATE_KEY_PATH，请确认文件内容正确；若用 WECHAT_PAY_PRIVATE_KEY，请确认环境变量为完整 PEM。"
)
# #region agent log
_debug_log("payment/wechat_pay.py:_ensure_client", "wechat_ensure_client_key_done", {}, "H_init")
# #endregion
self._client = WeChatPay(
# 平台公钥模式：无需访问 api.mch.weixin.qq.com 拉取证书，适合无法直连微信的环境
kwargs = dict(
wechatpay_type=WeChatPayType.APP,
mchid=self._config.mch_id,
private_key=private_key,
@@ -75,14 +114,42 @@ class WeChatPayClient:
apiv3_key=self._config.api_v3_key,
notify_url=self._config.notify_url,
)
if self._config.use_platform_public_key:
if self._config.platform_public_key_path and self._config.platform_public_key_path.strip():
key_path = _resolve_key_path(self._config.platform_public_key_path)
with open(key_path, "r", encoding="utf-8") as f:
platform_pub_key = _normalize_pem_key(f.read())
elif self._config.platform_public_key and self._config.platform_public_key.strip():
platform_pub_key = _normalize_pem_key(self._config.platform_public_key.strip())
else:
raise PaymentConfigError("平台公钥模式需设置 WECHAT_PAY_PLATFORM_PUBLIC_KEY 或 WECHAT_PAY_PLATFORM_PUBLIC_KEY_PATH")
if not platform_pub_key or "-----BEGIN" not in platform_pub_key:
raise PaymentConfigError("微信支付平台公钥格式错误，需为 PEM 格式")
kwargs["public_key"] = platform_pub_key
kwargs["public_key_id"] = self._config.platform_public_key_id.strip()
else:
kwargs["timeout"] = (10, 25) # 证书模式：拉取平台证书时的超时
self._client = WeChatPay(**kwargs)
# #region agent log
_debug_log("payment/wechat_pay.py:_ensure_client", "wechat_ensure_client_sdk_done", {}, "H_init")
# #endregion
logger.info("微信支付客户端初始化成功")
except FileNotFoundError:
key_path = _resolve_key_path(self._config.private_key_path) if self._config.private_key_path else self._config.private_key_path
raise PaymentConfigError(
f"微信支付商户私钥文件不存在: {self._config.private_key_path}（当前工作目录: {os.getcwd()}）"
f"微信支付商户私钥文件不存在: {key_path}（已尝试相对路径与 api 目录，当前工作目录: {os.getcwd()}）"
)
except PaymentConfigError:
raise
except Exception as e:
msg = str(e).strip()
if "No wechatpay platform certificate" in msg or "platform certificate" in msg.lower():
raise PaymentConfigError(
"无法获取微信支付平台证书（当前环境可能无法访问 api.mch.weixin.qq.com）。"
"请改用平台公钥模式：在商户平台「API安全」中获取微信支付公钥与公钥ID，并设置环境变量 "
"WECHAT_PAY_PLATFORM_PUBLIC_KEY（或 WECHAT_PAY_PLATFORM_PUBLIC_KEY_PATH）与 "
"WECHAT_PAY_PLATFORM_PUBLIC_KEY_ID。"
)
raise PaymentConfigError(
f"微信支付客户端初始化失败: {e}。"
"若使用文件路径，请确认文件为 PEM 格式且路径正确；或改用 WECHAT_PAY_PRIVATE_KEY 直接配置私钥内容。"
@@ -105,15 +172,26 @@ class WeChatPayClient:
Returns:
PaymentResult 包含调起支付所需的参数
"""
# #region agent log
_debug_log("payment/wechat_pay.py:create_app_order", "wechat_create_app_order_enter", {"out_trade_no": out_trade_no}, "H3")
# #endregion
self._ensure_client()
# #region agent log
_debug_log("payment/wechat_pay.py:create_app_order", "wechat_client_ready", {"out_trade_no": out_trade_no}, "H3")
# #endregion
try:
# #region agent log
_debug_log("payment/wechat_pay.py:create_app_order", "wechat_pay_request_start", {"out_trade_no": out_trade_no}, "H1,H4")
# #endregion
code, message = self._client.pay(
description=description,
out_trade_no=out_trade_no,
amount={"total": total_amount, "currency": "CNY"},
pay_type=self._get_pay_type(),
)
# #region agent log
_debug_log("payment/wechat_pay.py:create_app_order", "wechat_pay_request_done", {"out_trade_no": out_trade_no, "code": code}, "H1,H4")
# #endregion
result = json.loads(message)