1730 lines
65 KiB
Python
1730 lines
65 KiB
Python
import logging
|
||
import os
|
||
import json
|
||
from datetime import datetime, timedelta
|
||
from enum import Enum
|
||
|
||
# 尝试设置 SSL 证书路径
|
||
try:
|
||
import certifi
|
||
os.environ["SSL_CERT_FILE"] = certifi.where()
|
||
os.environ["REQUESTS_CA_BUNDLE"] = certifi.where()
|
||
except ImportError:
|
||
pass
|
||
|
||
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.credit.product import CreditProduct
|
||
from app.models.user import User
|
||
from app.models.system_config import SystemConfig
|
||
from app.enums.credit_record import (
|
||
CreditRecordAction,
|
||
CreditRecordBillingScene,
|
||
CreditRecordChargeKind,
|
||
CreditRecordOwnerType,
|
||
CreditRecordSourceModule,
|
||
CreditRecordSubject,
|
||
)
|
||
from app.services.credit_record_meta_service import CreditRecordMeta, build_recharge_meta
|
||
from app.services.credits import add_credits, deduct_credits
|
||
from app.enums.common import PaymentOrderSourceEnum
|
||
from app.enums.credit_product import CreditProductType
|
||
from app.models.team import Team
|
||
from app.services.credit.locking import acquire_team_business_lock, acquire_user_credit_lock
|
||
from app.services.credit.product_service import ensure_repeat_purchase_allowed, quote_product, resolve_team_purchase_context
|
||
from app.services.credit.subscription_service import fulfill_payment_product
|
||
from app.services.credit.utils import to_credit_decimal, utc_now
|
||
from app.utils.id_gen import generate_id, generate_order_no
|
||
|
||
|
||
|
||
def _payment_biz_key(order: PaymentOrder, *, charge_kind: str, action: str) -> str:
|
||
return (
|
||
f"{CreditRecordOwnerType.PAYMENT_ORDER.value}:{order.id}:"
|
||
f"attempt:1:{charge_kind}:{action}"
|
||
)
|
||
|
||
|
||
def _payment_recharge_meta(order: PaymentOrder) -> CreditRecordMeta:
|
||
meta = build_recharge_meta(owner_id=order.id)
|
||
meta.attempt_no = 1
|
||
return meta
|
||
|
||
|
||
def _payment_refund_meta(order: PaymentOrder) -> CreditRecordMeta:
|
||
return CreditRecordMeta(
|
||
owner_type=CreditRecordOwnerType.PAYMENT_ORDER.value,
|
||
owner_id=order.id,
|
||
attempt_no=1,
|
||
charge_kind=CreditRecordChargeKind.REFUND.value,
|
||
charge_action=CreditRecordAction.REFUND.value,
|
||
credit_subject=CreditRecordSubject.REFUND.value,
|
||
billing_scene=CreditRecordBillingScene.REFUND.value,
|
||
source_module=CreditRecordSourceModule.PAYMENT.value,
|
||
)
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 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.DEBUG)
|
||
|
||
_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
|
||
|
||
|
||
class PaymentCloseResult(str, Enum):
|
||
"""Result of attempting to close a remote payment order.
|
||
|
||
``PAID`` is intentionally distinct from ``FAILED``: when a gateway
|
||
explicitly says an order is already paid, callers must recover the paid
|
||
transaction instead of cancelling locally or retrying close forever.
|
||
"""
|
||
|
||
CLOSED = "closed"
|
||
PAID = "paid"
|
||
FAILED = "failed"
|
||
|
||
|
||
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)
|
||
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 _close_order_for_cancellation(
|
||
db: AsyncSession,
|
||
order: PaymentOrder,
|
||
db_configs: dict[str, str],
|
||
) -> PaymentCloseResult:
|
||
"""Close a gateway order before changing the local order to cancelled.
|
||
|
||
A gateway may explicitly report that the order has already been paid.
|
||
That is not a close failure: it must be routed into paid-order recovery so
|
||
the normal payment fulfillment pipeline can run.
|
||
"""
|
||
if order.payment_method == "alipay":
|
||
closed = await _close_alipay_order(db, order, db_configs)
|
||
return PaymentCloseResult.CLOSED if closed else PaymentCloseResult.FAILED
|
||
if order.payment_method == "wechat":
|
||
return await _close_wechat_order(db, order, db_configs)
|
||
return PaymentCloseResult.CLOSED
|
||
|
||
|
||
async def _resolve_order_for_cancellation(
|
||
db: AsyncSession,
|
||
order: PaymentOrder,
|
||
db_configs: dict[str, str],
|
||
) -> PaymentCloseResult:
|
||
"""Resolve a cancellation attempt into closed / paid / failed.
|
||
|
||
Normal orders are simply closed. For WeChat ``ORDERPAID``, query only
|
||
this order once and feed the verified result into the existing atomic
|
||
payment-success handler.
|
||
"""
|
||
close_result = await _close_order_for_cancellation(db, order, db_configs)
|
||
if close_result != PaymentCloseResult.PAID:
|
||
return close_result
|
||
|
||
if order.payment_method != "wechat":
|
||
return PaymentCloseResult.FAILED
|
||
|
||
recovered = await _recover_wechat_paid_order(db, order, db_configs)
|
||
return PaymentCloseResult.PAID if recovered else PaymentCloseResult.FAILED
|
||
|
||
|
||
async def _check_and_expire_order(db: AsyncSession, order: PaymentOrder) -> bool:
|
||
"""Expire one pending order after resolving the remote gateway state.
|
||
|
||
If WeChat reports ``ORDERPAID``, the order is queried once and recovered
|
||
through the normal payment-success pipeline instead of being cancelled.
|
||
"""
|
||
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 utc_now() < expiry:
|
||
return False
|
||
|
||
log_order_no = str(order.order_no)
|
||
log_user_id = str(order.user_id)
|
||
log_amount = order.amount
|
||
close_result = await _resolve_order_for_cancellation(db, order, db_configs)
|
||
if close_result == PaymentCloseResult.PAID:
|
||
logger.info(
|
||
f"ORDER_EXPIRE_PAID_RECOVERED order_no={log_order_no} "
|
||
f"user={log_user_id} amount={log_amount}"
|
||
)
|
||
return False
|
||
|
||
if close_result != PaymentCloseResult.CLOSED:
|
||
logger.warning(
|
||
f"ORDER_EXPIRE_CLOSE_PENDING order_no={log_order_no} "
|
||
f"user={log_user_id} amount={log_amount}"
|
||
)
|
||
return False
|
||
|
||
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()}"
|
||
)
|
||
return True
|
||
|
||
|
||
async def expire_all_pending_orders(db: AsyncSession) -> int:
|
||
"""Background task: resolve and expire overdue pending orders.
|
||
|
||
Each order is handled in its own database transaction. This keeps one
|
||
gateway or fulfillment failure from invalidating ORM state for the rest of
|
||
the batch. A WeChat ``ORDERPAID`` result is recovered as a paid order,
|
||
never persisted as cancelled.
|
||
"""
|
||
db_configs = await _get_payment_configs(db)
|
||
expire_seconds = _get_order_expire_seconds(db_configs)
|
||
threshold = utc_now() - timedelta(seconds=expire_seconds)
|
||
|
||
result = await db.execute(
|
||
select(PaymentOrder.order_no).where(
|
||
PaymentOrder.status == "pending",
|
||
PaymentOrder.created_at <= threshold,
|
||
)
|
||
)
|
||
order_nos = [str(row[0]) for row in result.all()]
|
||
|
||
# End the read-only candidate transaction before processing individual
|
||
# orders. Subsequent commit/rollback never leaves us reusing stale ORM
|
||
# instances from this initial scan.
|
||
await db.rollback()
|
||
|
||
expired_count = 0
|
||
|
||
for order_no in order_nos:
|
||
try:
|
||
result = await db.execute(
|
||
select(PaymentOrder)
|
||
.where(
|
||
PaymentOrder.order_no == order_no,
|
||
PaymentOrder.status == "pending",
|
||
)
|
||
.with_for_update()
|
||
.limit(1)
|
||
)
|
||
order = result.scalar_one_or_none()
|
||
if order is None:
|
||
await db.rollback()
|
||
continue
|
||
|
||
close_result = await _resolve_order_for_cancellation(
|
||
db, order, db_configs
|
||
)
|
||
|
||
if close_result == PaymentCloseResult.PAID:
|
||
logger.info(
|
||
f"ORDER_EXPIRE_PAID_RECOVERED order_no={order_no}"
|
||
)
|
||
# Paid recovery commits inside process_payment_success_by_order_no.
|
||
continue
|
||
|
||
if close_result != PaymentCloseResult.CLOSED:
|
||
logger.warning(
|
||
f"ORDER_EXPIRE_CLOSE_PENDING order_no={order_no} "
|
||
f"user={order.user_id} amount={order.amount}"
|
||
)
|
||
await db.rollback()
|
||
continue
|
||
|
||
log_user_id = str(order.user_id)
|
||
log_amount = order.amount
|
||
order.status = "cancelled"
|
||
|
||
await db.commit()
|
||
expired_count += 1
|
||
logger.info(
|
||
f"ORDER_EXPIRED order_no={order_no} user={log_user_id} "
|
||
f"amount={log_amount}"
|
||
)
|
||
except Exception:
|
||
await db.rollback()
|
||
logger.exception(
|
||
f"ORDER_EXPIRE_PROCESS_FAILED order_no={order_no}"
|
||
)
|
||
|
||
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"
|
||
|
||
try:
|
||
_alipay_client = DefaultAlipayClient(config, logger)
|
||
_alipay_client_app_id = app_id
|
||
except Exception:
|
||
logger.exception("Failed to initialize Alipay client")
|
||
_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 = 0.0,
|
||
price: float = 0.0,
|
||
label: str = "积分商品",
|
||
bonus_credits: float = 0.0,
|
||
method: str = "wechat",
|
||
*,
|
||
product_id: str | None = None,
|
||
quantity: int = 1,
|
||
request_time: datetime | None = None,
|
||
) -> PaymentOrder:
|
||
"""创建线上支付订单。
|
||
|
||
商品订单在创建时冻结商品、定价和数量快照；之后商品改价、下架或软删除均不改变订单合同。
|
||
个人/团队订阅存在未完成订单时全局阻止再次创建订阅订单。
|
||
"""
|
||
db_configs = await _get_payment_configs(db)
|
||
mock_mode = _is_mock_mode(db_configs)
|
||
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":
|
||
required_configs = [
|
||
"payment_wechat_appid", "payment_wechat_mch_id", "payment_wechat_private_key",
|
||
"payment_wechat_cert_serial_no", "payment_wechat_api_v3_key",
|
||
]
|
||
missing_configs = [key for key in required_configs if not db_configs.get(key)]
|
||
if missing_configs:
|
||
raise ValueError(f"微信支付未完成配置，缺少: {', '.join(missing_configs)}，请联系管理员")
|
||
else:
|
||
raise ValueError("不支持的线上支付方式")
|
||
|
||
checked_at = request_time or utc_now()
|
||
await acquire_user_credit_lock(db, user_id)
|
||
user_result = await db.execute(select(User).where(User.id == user_id).limit(1))
|
||
user = user_result.scalar_one_or_none()
|
||
if user is None:
|
||
raise ValueError("用户不存在")
|
||
|
||
order_id = generate_id()
|
||
order_no = generate_order_no()
|
||
product: CreditProduct | None = None
|
||
final_price = to_credit_decimal(price)
|
||
total_credits = to_credit_decimal(float(credits or 0) + float(bonus_credits or 0))
|
||
snapshot = None
|
||
product_type = None
|
||
team_id_snapshot = None
|
||
quoted_unit = final_price
|
||
quoted_amount = final_price
|
||
actual_unit = final_price
|
||
purchase_scene = "legacy_recharge"
|
||
price_type = "regular"
|
||
|
||
if product_id:
|
||
product_result = await db.execute(
|
||
select(CreditProduct)
|
||
.where(
|
||
CreditProduct.id == product_id,
|
||
CreditProduct.deleted_at.is_(None),
|
||
CreditProduct.is_active.is_(True),
|
||
)
|
||
.limit(1)
|
||
)
|
||
product = product_result.scalar_one_or_none()
|
||
if product is None:
|
||
raise ValueError("积分商品不存在、已下架或已删除")
|
||
product_type = product.product_type
|
||
if product_type == CreditProductType.TEAM_SUBSCRIPTION.value:
|
||
if not 2 <= int(quantity) <= 20:
|
||
raise ValueError("客户端团队套餐单次购买数量必须在2到20之间")
|
||
team, available, reason = await resolve_team_purchase_context(db, user=user)
|
||
if not available:
|
||
raise ValueError(reason or "当前不能购买团队订阅套餐")
|
||
if team is not None:
|
||
await acquire_team_business_lock(db, team.id)
|
||
team, available, reason = await resolve_team_purchase_context(db, user=user)
|
||
if not available or team is None:
|
||
raise ValueError(reason or "当前不能购买团队订阅套餐")
|
||
team_id_snapshot = team.id if team else None
|
||
first_purchase = bool(team is None or team.first_subscription_paid_at is None)
|
||
else:
|
||
if int(quantity) != 1:
|
||
raise ValueError("个人订阅和积分增值包购买数量固定为1")
|
||
quantity = 1
|
||
first_purchase = user.first_membership_paid_at is None if product_type == CreditProductType.SUBSCRIPTION.value else False
|
||
|
||
if product_type in {CreditProductType.SUBSCRIPTION.value, CreditProductType.TEAM_SUBSCRIPTION.value}:
|
||
ensure_repeat_purchase_allowed(product, first_purchase=first_purchase)
|
||
incomplete_result = await db.execute(
|
||
select(PaymentOrder.id).where(
|
||
PaymentOrder.user_id == user_id,
|
||
PaymentOrder.product_type.in_([
|
||
CreditProductType.SUBSCRIPTION.value,
|
||
CreditProductType.TEAM_SUBSCRIPTION.value,
|
||
]),
|
||
(
|
||
(PaymentOrder.status == "pending")
|
||
| ((PaymentOrder.status == "paid") & (PaymentOrder.fulfillment_status != "fulfilled"))
|
||
),
|
||
).limit(1)
|
||
)
|
||
if incomplete_result.scalar_one_or_none():
|
||
raise ValueError("存在未完成的订阅订单，请先前往订单记录完成支付或处理原订单")
|
||
|
||
quote = quote_product(
|
||
product,
|
||
first_purchase=first_purchase,
|
||
request_time=checked_at,
|
||
quantity=int(quantity),
|
||
)
|
||
quoted_unit = quote.quoted_unit_price
|
||
quoted_amount = quote.quoted_amount
|
||
final_price = quoted_amount
|
||
actual_unit = quoted_unit
|
||
purchase_scene = quote.purchase_scene
|
||
price_type = quote.price_type
|
||
label = product.name
|
||
if product_type == CreditProductType.TEAM_SUBSCRIPTION.value:
|
||
total_credits = to_credit_decimal((product.monthly_grant_credits or 0) * int(quantity))
|
||
elif product_type == CreditProductType.SUBSCRIPTION.value:
|
||
total_credits = to_credit_decimal(product.monthly_grant_credits or 0)
|
||
else:
|
||
total_credits = to_credit_decimal(product.grant_credits or 0)
|
||
snapshot = {
|
||
"id": product.id,
|
||
"product_code": product.product_code,
|
||
"product_type": product.product_type,
|
||
"name": product.name,
|
||
"description": product.description,
|
||
"features": product.features_json or [],
|
||
"tier_code": product.tier_code,
|
||
"tier_rank": product.tier_rank,
|
||
"billing_cycle": product.billing_cycle,
|
||
"monthly_grant_credits": float(product.monthly_grant_credits or 0),
|
||
"first_purchase_price": float(product.first_purchase_price or 0),
|
||
"regular_price": float(product.regular_price or 0),
|
||
"activity_price": float(product.activity_price) if product.activity_price is not None else None,
|
||
"activity_start_at": product.activity_start_at.isoformat() if product.activity_start_at else None,
|
||
"activity_end_at": product.activity_end_at.isoformat() if product.activity_end_at else None,
|
||
"renewal_enabled": bool(product.renewal_enabled),
|
||
"grant_credits": float(product.grant_credits or 0),
|
||
"validity_months": product.validity_months,
|
||
"credit_level": product.credit_level,
|
||
"currency": product.currency,
|
||
}
|
||
|
||
# 固定锁序：user advisory 已获取；团队订单已先获取 Team advisory，最后才锁 User 行。
|
||
locked_user_result = await db.execute(select(User).where(User.id == user_id).limit(1).with_for_update())
|
||
locked_user = locked_user_result.scalar_one_or_none()
|
||
if locked_user is None:
|
||
raise ValueError("用户不存在")
|
||
if locked_user.team_id != user.team_id:
|
||
raise ValueError("用户团队关系已发生变化，请刷新后重试")
|
||
user = locked_user
|
||
|
||
order = PaymentOrder(
|
||
id=order_id,
|
||
user_id=user_id,
|
||
order_no=order_no,
|
||
amount=final_price,
|
||
credits=total_credits,
|
||
payment_method=method,
|
||
order_source=PaymentOrderSourceEnum.ONLINE_PAYMENT.value,
|
||
status="pending",
|
||
product_id=product.id if product else None,
|
||
product_type=product_type,
|
||
purchase_scene=purchase_scene,
|
||
price_type=price_type,
|
||
product_code_snapshot=product.product_code if product else None,
|
||
product_name_snapshot=label,
|
||
product_snapshot_json=snapshot,
|
||
quantity=int(quantity),
|
||
quoted_unit_price_snapshot=quoted_unit,
|
||
quoted_amount_snapshot=quoted_amount,
|
||
actual_unit_price_snapshot=actual_unit,
|
||
team_id_snapshot=team_id_snapshot,
|
||
fulfillment_status="pending" if product else None,
|
||
)
|
||
db.add(order)
|
||
await db.flush()
|
||
logger.info(
|
||
f"ORDER_CREATED order_no={order.order_no} user={user_id} amount={order.amount} "
|
||
f"credits={order.credits} method={method} product_id={product_id} quantity={quantity} mock={mock_mode}"
|
||
)
|
||
|
||
if mock_mode:
|
||
order.status = "paid"
|
||
order.paid_at = checked_at
|
||
desc = f"充值{label}({order.credits}积分)"
|
||
await _fulfill_paid_order(db, order=order, fulfilled_at=checked_at, legacy_description=desc)
|
||
await db.flush()
|
||
elif method == "wechat":
|
||
qr_code_content = _create_wechat_order(order, db_configs)
|
||
if not qr_code_content:
|
||
raise ValueError("微信支付预下单失败，请检查配置或稍后重试")
|
||
order.qr_url = qr_code_content # type: ignore[attr-defined]
|
||
elif method == "alipay":
|
||
qr_url = _create_alipay_order(order, db_configs)
|
||
if not qr_url:
|
||
raise ValueError("支付宝预下单失败，请检查配置或稍后重试")
|
||
order.qr_url = qr_url # type: ignore[attr-defined]
|
||
return order
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# WeChat Pay client (lazy singleton, recreated when config changes)
|
||
# ---------------------------------------------------------------------------
|
||
_wechat_client = None
|
||
_wechat_mch_id = None
|
||
|
||
|
||
def _parse_wechat_result(result) -> dict:
|
||
"""解析微信支付SDK返回的result为字典"""
|
||
data = {}
|
||
if isinstance(result, dict):
|
||
data = result
|
||
elif isinstance(result, str):
|
||
try:
|
||
data = json.loads(result)
|
||
except:
|
||
pass
|
||
elif hasattr(result, 'get'):
|
||
data = result
|
||
elif hasattr(result, '__dict__'):
|
||
data = result.__dict__
|
||
return data
|
||
|
||
|
||
def _get_wechat_client(
|
||
mch_id: str,
|
||
private_key: str,
|
||
cert_serial_no: str,
|
||
api_v3_key: str,
|
||
appid: str,
|
||
notify_url: str = "",
|
||
public_key: str = None,
|
||
public_key_id: str = None
|
||
):
|
||
"""Get or create a WeChat Pay client. Recreated if config changes."""
|
||
global _wechat_client, _wechat_mch_id
|
||
|
||
if _wechat_client is not None and _wechat_mch_id == mch_id:
|
||
return _wechat_client
|
||
|
||
try:
|
||
from wechatpayv3 import WeChatPay, WeChatPayType
|
||
except ImportError:
|
||
logger.error(
|
||
"wechatpayv3 is not installed. "
|
||
"Install it with: pip install wechatpayv3"
|
||
)
|
||
return None
|
||
|
||
try:
|
||
# 初始化微信支付客户端
|
||
wechatpay_args = {
|
||
"wechatpay_type": WeChatPayType.NATIVE,
|
||
"mchid": mch_id,
|
||
"private_key": private_key.strip(),
|
||
"cert_serial_no": cert_serial_no,
|
||
"appid": appid,
|
||
"apiv3_key": api_v3_key,
|
||
"notify_url": notify_url,
|
||
"logger": logger,
|
||
}
|
||
# 如果配置了 public_key 和 public_key_id，就使用它们，否则不设置
|
||
if public_key and public_key_id:
|
||
wechatpay_args["public_key"] = public_key
|
||
wechatpay_args["public_key_id"] = public_key_id
|
||
else:
|
||
logger.info("No platform public key configured, will try to download")
|
||
_wechat_client = WeChatPay(**wechatpay_args)
|
||
_wechat_mch_id = mch_id
|
||
return _wechat_client
|
||
except Exception as e:
|
||
logger.exception(
|
||
"Failed to initialize WeChat Pay client: %s\n"
|
||
"Please double-check your merchant private key configuration.\n"
|
||
"It should be from apiclient_key.pem, NOT apiclient_cert.pem!",
|
||
e
|
||
)
|
||
_wechat_client = None
|
||
_wechat_mch_id = None
|
||
return None
|
||
|
||
|
||
def _create_wechat_order(order: PaymentOrder, db_configs: dict[str, str]) -> str | None:
|
||
"""Create a WeChat Pay Native order. Returns QR code content (code_url).
|
||
|
||
Reads all WeChat config from the database (admin panel).
|
||
Returns the ``code_url`` on success, or ``None`` on failure.
|
||
"""
|
||
mch_id = db_configs.get("payment_wechat_mch_id", "")
|
||
private_key = db_configs.get("payment_wechat_private_key", "")
|
||
cert_serial_no = db_configs.get("payment_wechat_cert_serial_no", "")
|
||
api_v3_key = db_configs.get("payment_wechat_api_v3_key", "")
|
||
appid = db_configs.get("payment_wechat_appid", "")
|
||
notify_url = db_configs.get("payment_wechat_notify_url", "")
|
||
public_key = db_configs.get("payment_wechat_public_key", "")
|
||
public_key_id = db_configs.get("payment_wechat_public_key_id", "")
|
||
if not all([mch_id, private_key, cert_serial_no, api_v3_key, appid]):
|
||
logger.warning("WeChat payment config missing in database")
|
||
return None
|
||
|
||
client = _get_wechat_client(mch_id, private_key, cert_serial_no, api_v3_key, appid, notify_url, public_key, public_key_id)
|
||
if client is None:
|
||
return None
|
||
|
||
try:
|
||
# 调用微信支付 Native 下单接口
|
||
code, result = client.pay(
|
||
description=f"充值积分{order.credits}，订单 {order.order_no}",
|
||
out_trade_no=order.order_no,
|
||
amount={
|
||
"total": int(order.amount * 100), # 微信支付以分为单位
|
||
"currency": "CNY"
|
||
},
|
||
notify_url=notify_url,
|
||
scene_info={
|
||
"payer_client_ip": "127.0.0.1",
|
||
}
|
||
)
|
||
|
||
data = _parse_wechat_result(result)
|
||
|
||
if code == 200 and data.get('code_url'):
|
||
# 注意：微信返回的 code_url 可能需要进一步处理成二维码图片地址
|
||
logger.info(f"WeChat order created successfully: order_no={order.order_no}")
|
||
return data.get('code_url')
|
||
else:
|
||
logger.error(
|
||
f"WeChat pay failed: order_no={order.order_no}, "
|
||
f"code={code}, result={result}"
|
||
)
|
||
return None
|
||
except Exception as e:
|
||
logger.exception(f"WeChat pay exception: order_no={order.order_no}")
|
||
return None
|
||
|
||
|
||
async def _close_wechat_order(
|
||
db: AsyncSession,
|
||
order: PaymentOrder,
|
||
db_configs: dict[str, str],
|
||
) -> PaymentCloseResult:
|
||
"""Call WeChat Pay close API and classify the authoritative result."""
|
||
mch_id = db_configs.get("payment_wechat_mch_id", "")
|
||
private_key = db_configs.get("payment_wechat_private_key", "")
|
||
cert_serial_no = db_configs.get("payment_wechat_cert_serial_no", "")
|
||
api_v3_key = db_configs.get("payment_wechat_api_v3_key", "")
|
||
appid = db_configs.get("payment_wechat_appid", "")
|
||
public_key = db_configs.get("payment_wechat_public_key", "")
|
||
public_key_id = db_configs.get("payment_wechat_public_key_id", "")
|
||
notify_url = db_configs.get("payment_wechat_notify_url", "")
|
||
|
||
client = _get_wechat_client(mch_id, private_key, cert_serial_no, api_v3_key, appid, notify_url, public_key, public_key_id)
|
||
if client is None:
|
||
return PaymentCloseResult.FAILED
|
||
|
||
mock_mode = _is_mock_mode(db_configs)
|
||
if mock_mode:
|
||
logger.info(f"Mock mode: skipping close_wechat_order for {order.order_no}")
|
||
return PaymentCloseResult.CLOSED
|
||
|
||
try:
|
||
code, result = client.close(out_trade_no=order.order_no)
|
||
if code == 204:
|
||
logger.info(f"WeChat order closed: order_no={order.order_no}")
|
||
return PaymentCloseResult.CLOSED
|
||
|
||
data = _parse_wechat_result(result)
|
||
error_code = str(
|
||
data.get("code")
|
||
or data.get("err_code")
|
||
or data.get("error_code")
|
||
or ""
|
||
).upper()
|
||
raw_result = str(result).upper()
|
||
|
||
# 微信明确返回“订单不存在”时，同样视为没有远端订单需要继续关闭。
|
||
# 只兼容明确的订单不存在错误，不吞掉签名、网络、系统等其他异常。
|
||
if error_code in {"ORDER_NOT_EXIST", "ORDERNOTEXIST"} or (
|
||
code == 404
|
||
and ("ORDER_NOT_EXIST" in raw_result or "ORDERNOTEXIST" in raw_result)
|
||
):
|
||
logger.info(
|
||
f"WeChat order not exist, treat as closed for local expiry: "
|
||
f"order_no={order.order_no}, code={code}, error_code={error_code or 'unknown'}"
|
||
)
|
||
return PaymentCloseResult.CLOSED
|
||
|
||
# ORDERPAID means the remote transaction is already paid. Do not
|
||
# cancel locally and do not retry close forever; the caller will query
|
||
# this one order once and recover it through the normal paid pipeline.
|
||
if error_code == "ORDERPAID" or "ORDERPAID" in raw_result:
|
||
logger.info(
|
||
f"WeChat close reports order already paid: "
|
||
f"order_no={order.order_no}, code={code}"
|
||
)
|
||
return PaymentCloseResult.PAID
|
||
|
||
logger.error(
|
||
f"WeChat close failed: order_no={order.order_no}, "
|
||
f"code={code}, result={result}"
|
||
)
|
||
return PaymentCloseResult.FAILED
|
||
except Exception as e:
|
||
logger.exception(f"WeChat close exception: order_no={order.order_no}")
|
||
return PaymentCloseResult.FAILED
|
||
|
||
|
||
async def _query_wechat_order(db: AsyncSession, order: PaymentOrder, db_configs: dict[str, str]) -> dict | None:
|
||
"""Call WeChat Pay query API to check order status.
|
||
Returns the response data if successful, None otherwise.
|
||
"""
|
||
mch_id = db_configs.get("payment_wechat_mch_id", "")
|
||
private_key = db_configs.get("payment_wechat_private_key", "")
|
||
cert_serial_no = db_configs.get("payment_wechat_cert_serial_no", "")
|
||
api_v3_key = db_configs.get("payment_wechat_api_v3_key", "")
|
||
appid = db_configs.get("payment_wechat_appid", "")
|
||
public_key = db_configs.get("payment_wechat_public_key", "")
|
||
public_key_id = db_configs.get("payment_wechat_public_key_id", "")
|
||
notify_url = db_configs.get("payment_wechat_notify_url", "")
|
||
|
||
client = _get_wechat_client(mch_id, private_key, cert_serial_no, api_v3_key, appid, notify_url, public_key, public_key_id)
|
||
if client is None:
|
||
return None
|
||
|
||
mock_mode = _is_mock_mode(db_configs)
|
||
if mock_mode:
|
||
logger.info(f"Mock mode: skipping query_wechat_order for {order.order_no}")
|
||
return {"trade_state": "SUCCESS"}
|
||
|
||
try:
|
||
code, result = client.query(out_trade_no=order.order_no)
|
||
data = _parse_wechat_result(result)
|
||
if code == 200 and data.get('trade_state'):
|
||
logger.info(
|
||
f"WeChat query succeeded: order_no={order.order_no}, "
|
||
f"trade_state={data.get('trade_state')}"
|
||
)
|
||
return data
|
||
else:
|
||
logger.error(
|
||
f"WeChat query failed: order_no={order.order_no}, "
|
||
f"code={code}, result={result}"
|
||
)
|
||
return None
|
||
except Exception as e:
|
||
logger.exception(f"WeChat query exception: order_no={order.order_no}")
|
||
return None
|
||
|
||
|
||
async def _recover_wechat_paid_order(
|
||
db: AsyncSession,
|
||
order: PaymentOrder,
|
||
db_configs: dict[str, str],
|
||
) -> bool:
|
||
"""Recover one WeChat order after close reports ``ORDERPAID``.
|
||
|
||
This is not a periodic query of all pending orders. It runs only after
|
||
WeChat itself explicitly says this specific order has already been paid.
|
||
The query supplies the authoritative transaction id and amount, then the
|
||
existing atomic payment-success handler performs all subscription / upgrade
|
||
/ credit fulfillment.
|
||
"""
|
||
local_order_no = str(order.order_no)
|
||
data = await _query_wechat_order(db, order, db_configs)
|
||
if not data:
|
||
logger.warning(
|
||
f"WECHAT_ORDERPAID_QUERY_FAILED order_no={local_order_no}"
|
||
)
|
||
return False
|
||
|
||
trade_state = str(data.get("trade_state") or "").upper()
|
||
remote_order_no = str(data.get("out_trade_no") or "")
|
||
transaction_id = str(data.get("transaction_id") or "")
|
||
amount_info = data.get("amount") or {}
|
||
currency = str(amount_info.get("currency") or "CNY").upper()
|
||
total_fen = amount_info.get("total")
|
||
|
||
if trade_state != "SUCCESS":
|
||
logger.warning(
|
||
f"WECHAT_ORDERPAID_QUERY_NOT_SUCCESS order_no={local_order_no} "
|
||
f"trade_state={trade_state or 'unknown'}"
|
||
)
|
||
return False
|
||
|
||
if remote_order_no and remote_order_no != local_order_no:
|
||
logger.error(
|
||
f"WECHAT_ORDERPAID_ORDER_MISMATCH local={local_order_no} "
|
||
f"remote={remote_order_no}"
|
||
)
|
||
return False
|
||
|
||
if currency != "CNY":
|
||
logger.error(
|
||
f"WECHAT_ORDERPAID_CURRENCY_MISMATCH order_no={local_order_no} "
|
||
f"currency={currency}"
|
||
)
|
||
return False
|
||
|
||
if total_fen is None:
|
||
logger.error(
|
||
f"WECHAT_ORDERPAID_AMOUNT_MISSING order_no={local_order_no}"
|
||
)
|
||
return False
|
||
|
||
try:
|
||
total_amount = float(to_credit_decimal(total_fen) / to_credit_decimal(100))
|
||
except Exception:
|
||
logger.exception(
|
||
f"WECHAT_ORDERPAID_AMOUNT_INVALID order_no={local_order_no} "
|
||
f"total={total_fen}"
|
||
)
|
||
return False
|
||
|
||
if not transaction_id:
|
||
logger.error(
|
||
f"WECHAT_ORDERPAID_TRANSACTION_ID_MISSING order_no={local_order_no}"
|
||
)
|
||
return False
|
||
|
||
processed = await process_payment_success_by_order_no(
|
||
db,
|
||
local_order_no,
|
||
transaction_id,
|
||
total_amount,
|
||
)
|
||
if processed:
|
||
logger.info(
|
||
f"WECHAT_ORDERPAID_RECOVERED order_no={local_order_no} "
|
||
f"transaction_id={transaction_id} amount={total_amount}"
|
||
)
|
||
else:
|
||
logger.warning(
|
||
f"WECHAT_ORDERPAID_RECOVERY_NOT_PROCESSED order_no={local_order_no} "
|
||
f"transaction_id={transaction_id} amount={total_amount}"
|
||
)
|
||
return processed
|
||
|
||
|
||
async def _refund_wechat_order(
|
||
db: AsyncSession,
|
||
order: PaymentOrder,
|
||
refund_amount: float,
|
||
refund_reason: str,
|
||
db_configs: dict[str, str]
|
||
) -> dict:
|
||
"""Call WeChat Pay refund API.
|
||
|
||
微信退款是异步的，调用后会返回PROCESSING状态，实际退款结果通过回调通知。
|
||
"""
|
||
mch_id = db_configs.get("payment_wechat_mch_id", "")
|
||
private_key = db_configs.get("payment_wechat_private_key", "")
|
||
cert_serial_no = db_configs.get("payment_wechat_cert_serial_no", "")
|
||
api_v3_key = db_configs.get("payment_wechat_api_v3_key", "")
|
||
appid = db_configs.get("payment_wechat_appid", "")
|
||
public_key = db_configs.get("payment_wechat_public_key", "")
|
||
public_key_id = db_configs.get("payment_wechat_public_key_id", "")
|
||
notify_url = db_configs.get("payment_wechat_notify_url", "")
|
||
|
||
client = _get_wechat_client(mch_id, private_key, cert_serial_no, api_v3_key, appid, notify_url, public_key, public_key_id)
|
||
if client is None:
|
||
return {"success": False, "message": "微信支付客户端初始化失败"}
|
||
|
||
mock_mode = _is_mock_mode(db_configs)
|
||
if mock_mode:
|
||
logger.info(f"Mock mode: skipping wechat refund for {order.order_no}")
|
||
return {"success": True}
|
||
|
||
try:
|
||
out_refund_no = f"{order.order_no}_refund_{int(datetime.now().timestamp())}"
|
||
code, result = client.refund(
|
||
out_trade_no=order.order_no,
|
||
out_refund_no=out_refund_no,
|
||
amount={
|
||
"total": int(order.amount * 100), # 订单总金额
|
||
"refund": int(refund_amount * 100), # 退款金额
|
||
"currency": "CNY"
|
||
},
|
||
reason=refund_reason
|
||
)
|
||
|
||
data = _parse_wechat_result(result)
|
||
|
||
# 微信退款是异步的，PROCESSING是正常状态，表示退款已受理
|
||
if code == 200:
|
||
status = data.get('status')
|
||
if status in ('SUCCESS', 'PROCESSING', 'REFUNDCLOSE'):
|
||
logger.info(
|
||
f"WeChat refund initiated: order_no={order.order_no}, "
|
||
f"status={status}, refund_id={data.get('refund_id')}"
|
||
)
|
||
return {
|
||
"success": True,
|
||
"refund_id": data.get('refund_id'),
|
||
"status": status,
|
||
"message": "退款申请已提交，等待微信处理"
|
||
}
|
||
|
||
logger.error(
|
||
f"WeChat refund failed: order_no={order.order_no}, "
|
||
f"code={code}, result={result}"
|
||
)
|
||
return {
|
||
"success": False,
|
||
"message": f"微信退款失败: code={code}, {data.get('message', '')}"
|
||
}
|
||
except Exception as e:
|
||
logger.exception(f"WeChat refund exception: order_no={order.order_no}")
|
||
return {"success": False, "message": f"微信退款异常: {str(e)}"}
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 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.credits}，订单 {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
|
||
|
||
# 支付宝明确返回“交易不存在”时，远端没有可继续关闭的交易。
|
||
# 对本地已经达到超时时间的 pending 订单，可视为关单目标已经满足，
|
||
# 避免 Celery 每分钟对同一订单无限重复调用 trade.close。
|
||
if str(response.sub_code or "").upper() == "ACQ.TRADE_NOT_EXIST":
|
||
logger.info(
|
||
f"Alipay trade not exist, treat as closed for local expiry: "
|
||
f"order_no={order.order_no}, code={response.code}, "
|
||
f"sub_code={response.sub_code}"
|
||
)
|
||
return True
|
||
|
||
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/WeChat query and update status.
|
||
|
||
Candidate rows are collected as primitive values first. A failed payment
|
||
fulfillment may roll back the session; using primitive candidates prevents
|
||
that rollback from expiring ORM objects needed by later iterations.
|
||
"""
|
||
result = await db.execute(
|
||
select(PaymentOrder.order_no, PaymentOrder.payment_method).where(
|
||
PaymentOrder.status == "pending",
|
||
)
|
||
)
|
||
candidates = [(str(row.order_no), str(row.payment_method)) for row in result.all()]
|
||
updated_count = 0
|
||
db_configs = await _get_payment_configs(db)
|
||
|
||
for order_no, payment_method in candidates:
|
||
try:
|
||
order_result = await db.execute(
|
||
select(PaymentOrder).where(
|
||
PaymentOrder.order_no == order_no,
|
||
PaymentOrder.status == "pending",
|
||
).limit(1)
|
||
)
|
||
order = order_result.scalar_one_or_none()
|
||
if order is None:
|
||
continue
|
||
|
||
if payment_method == "alipay":
|
||
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"):
|
||
trade_no = data.get("trade_no", "")
|
||
total_amount_str = data.get("total_amount", "")
|
||
total_amount = float(total_amount_str) if total_amount_str else None
|
||
processed = await process_payment_success_by_order_no(
|
||
db, order_no, trade_no, total_amount
|
||
)
|
||
if processed:
|
||
updated_count += 1
|
||
elif trade_status in ("TRADE_CLOSED", "TRADE_CANCELLED"):
|
||
order.status = "cancelled"
|
||
await db.commit()
|
||
updated_count += 1
|
||
|
||
elif payment_method == "wechat":
|
||
data = await _query_wechat_order(db, order, db_configs)
|
||
if data:
|
||
trade_state = data.get("trade_state")
|
||
if trade_state == "SUCCESS":
|
||
transaction_id = data.get("transaction_id", "")
|
||
total_amount = float(data.get("amount", {}).get("total", 0)) / 100
|
||
processed = await process_payment_success_by_order_no(
|
||
db, order_no, transaction_id, total_amount
|
||
)
|
||
if processed:
|
||
updated_count += 1
|
||
elif trade_state in ("CLOSED", "REVOKED"):
|
||
order.status = "cancelled"
|
||
await db.commit()
|
||
updated_count += 1
|
||
except Exception as e:
|
||
await db.rollback()
|
||
logger.exception(f"Failed to sync order {order_no}: {e}")
|
||
|
||
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 False
|
||
|
||
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.
|
||
|
||
Note: Since we're using wechatpayv3 SDK which handles verification internally,
|
||
in the callback handler we'll verify with the SDK. This function is kept for
|
||
interface consistency and mock mode support.
|
||
"""
|
||
db_configs = await _get_payment_configs(db)
|
||
mock_mode = _is_mock_mode(db_configs)
|
||
if mock_mode:
|
||
logger.info("Mock mode: skipping WeChat callback verification")
|
||
return True
|
||
|
||
# 对于真实模式，我们在回调路由处理器中直接使用 SDK 验证
|
||
# 这里我们返回 True 以保持接口一致性
|
||
logger.info("WeChat callback verification (delegated to SDK in router)")
|
||
return True
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Process successful payment
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
async def _fulfill_paid_order(
|
||
db: AsyncSession,
|
||
*,
|
||
order: PaymentOrder,
|
||
fulfilled_at: datetime,
|
||
legacy_description: str,
|
||
) -> None:
|
||
"""Fulfill the business side of a paid order without committing.
|
||
|
||
Product orders are considered successfully paid locally only when the
|
||
product service leaves the order in the explicit ``fulfilled`` state.
|
||
This turns any internal upgrade/reconciliation failure into an exception so
|
||
the outer payment transaction can roll back the temporary ``paid`` state.
|
||
"""
|
||
if order.product_id:
|
||
await fulfill_payment_product(db, order=order, fulfilled_at=fulfilled_at)
|
||
if order.fulfillment_status != "fulfilled":
|
||
raise RuntimeError(
|
||
f"Payment product fulfillment incomplete: order_no={order.order_no}, "
|
||
f"fulfillment_status={order.fulfillment_status}"
|
||
)
|
||
return
|
||
|
||
await add_credits(
|
||
db,
|
||
order.user_id,
|
||
order.credits,
|
||
legacy_description,
|
||
related_id=order.id,
|
||
biz_key=_payment_biz_key(
|
||
order,
|
||
charge_kind=CreditRecordChargeKind.RECHARGE.value,
|
||
action=CreditRecordAction.CHARGE.value,
|
||
),
|
||
record_meta=_payment_recharge_meta(order),
|
||
payment_order_id=order.id,
|
||
source_id=order.id,
|
||
)
|
||
|
||
|
||
async def process_payment_success(db: AsyncSession, order_id: str) -> bool:
|
||
"""Atomically mark a pending order paid and fulfill its business value."""
|
||
try:
|
||
result = await db.execute(
|
||
select(PaymentOrder)
|
||
.where(PaymentOrder.id == order_id)
|
||
.with_for_update()
|
||
.limit(1)
|
||
)
|
||
order = result.scalar_one_or_none()
|
||
if not order:
|
||
return False
|
||
if order.status == "paid":
|
||
return True
|
||
if order.status != "pending":
|
||
return False
|
||
|
||
paid_at = utc_now()
|
||
order.status = "paid"
|
||
order.paid_at = paid_at
|
||
await _fulfill_paid_order(
|
||
db,
|
||
order=order,
|
||
fulfilled_at=paid_at,
|
||
legacy_description=f"充值成功({order.credits}积分)",
|
||
)
|
||
await db.commit()
|
||
return True
|
||
except Exception:
|
||
await db.rollback()
|
||
raise
|
||
|
||
|
||
async def process_payment_success_by_order_no(
|
||
db: AsyncSession,
|
||
order_no: str,
|
||
trade_no: str = "",
|
||
total_amount: float | None = None,
|
||
) -> bool:
|
||
"""Atomically process an Alipay/WeChat payment success.
|
||
|
||
``paid`` is persisted only together with successful local fulfillment.
|
||
When fulfillment fails, the whole local transaction is rolled back and the
|
||
order remains ``pending``; the existing gateway query loop may retry it.
|
||
"""
|
||
try:
|
||
result = await db.execute(
|
||
select(PaymentOrder)
|
||
.where(PaymentOrder.order_no == order_no)
|
||
.with_for_update()
|
||
.limit(1)
|
||
)
|
||
order = result.scalar_one_or_none()
|
||
|
||
if not order:
|
||
logger.info(f"Order {order_no} not found, skipping")
|
||
return False
|
||
|
||
if order.status == "paid":
|
||
logger.info(f"Order {order_no} already processed, skipping")
|
||
return True
|
||
|
||
if order.status != "pending":
|
||
logger.info(f"Order {order_no} is in {order.status} state, cannot process")
|
||
return False
|
||
|
||
if (
|
||
total_amount is not None
|
||
and abs(
|
||
to_credit_decimal(total_amount) - to_credit_decimal(order.amount)
|
||
) > to_credit_decimal("0.01")
|
||
):
|
||
logger.error(
|
||
f"Amount mismatch: order amount {order.amount}, gateway amount {total_amount}"
|
||
)
|
||
return False
|
||
|
||
# A pending order must never be skipped merely because trade_no is
|
||
# already present. Historical partial writes can otherwise remain
|
||
# pending forever. The row lock + paid state + credit biz_key provide
|
||
# the actual idempotency boundary.
|
||
paid_at = utc_now()
|
||
order.status = "paid"
|
||
order.paid_at = paid_at
|
||
if trade_no:
|
||
order.trade_no = trade_no
|
||
|
||
await _fulfill_paid_order(
|
||
db,
|
||
order=order,
|
||
fulfilled_at=paid_at,
|
||
legacy_description=(
|
||
f"充值成功({order.credits}积分), "
|
||
f"订单号: {order_no}, 金额: {order.amount}"
|
||
),
|
||
)
|
||
|
||
log_user_id = str(order.user_id)
|
||
log_amount = order.amount
|
||
log_credits = order.credits
|
||
await db.commit()
|
||
logger.info(
|
||
f"PAYMENT_SUCCESS order_no={order_no} user={log_user_id} "
|
||
f"amount={log_amount} credits={log_credits} trade_no={trade_no}"
|
||
)
|
||
return True
|
||
except Exception:
|
||
await db.rollback()
|
||
raise
|
||
|
||
|
||
async def process_refund(
|
||
db: AsyncSession,
|
||
order_no: str,
|
||
refund_amount: float | None = None,
|
||
refund_reason: str = "管理员退款",
|
||
) -> dict:
|
||
"""本版本保留退款 Service 入口，但主动订单退款统一关闭。"""
|
||
result = await db.execute(select(PaymentOrder).where(PaymentOrder.order_no == order_no).limit(1))
|
||
order = result.scalar_one_or_none()
|
||
if not order:
|
||
return {"success": False, "message": "订单不存在"}
|
||
logger.warning(
|
||
"REFUND_BLOCKED order_no=%s user=%s requested_amount=%s reason=%s",
|
||
order_no, order.user_id, refund_amount, refund_reason,
|
||
)
|
||
return {"success": False, "message": "当前版本暂未开放订单退款"}
|
||
|
||
|
||
async def _refund_alipay_order(
|
||
db: AsyncSession,
|
||
order: PaymentOrder,
|
||
refund_amount: float,
|
||
refund_reason: str,
|
||
db_configs: dict[str, str]
|
||
) -> dict:
|
||
"""Call Alipay refund API."""
|
||
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 {"success": False, "message": "支付宝客户端初始化失败"}
|
||
|
||
mock_mode = _is_mock_mode(db_configs)
|
||
if mock_mode:
|
||
logger.info(f"Mock mode: skipping alipay refund for {order.order_no}")
|
||
return {"success": True}
|
||
|
||
try:
|
||
from alipay.aop.api.domain.AlipayTradeRefundModel import AlipayTradeRefundModel
|
||
from alipay.aop.api.request.AlipayTradeRefundRequest import AlipayTradeRefundRequest
|
||
from alipay.aop.api.response.AlipayTradeRefundResponse import AlipayTradeRefundResponse
|
||
|
||
model = AlipayTradeRefundModel()
|
||
model.out_trade_no = order.order_no
|
||
model.refund_amount = f"{refund_amount:.2f}"
|
||
model.refund_reason = refund_reason
|
||
model.out_request_no = f"{order.order_no}_refund_{int(datetime.now().timestamp())}"
|
||
|
||
request = AlipayTradeRefundRequest(biz_model=model)
|
||
response_content = client.execute(request)
|
||
|
||
if not response_content:
|
||
logger.error(f"Alipay refund failed: empty response, order_no={order.order_no}")
|
||
return {"success": False, "message": "支付宝退款响应为空"}
|
||
|
||
response = AlipayTradeRefundResponse()
|
||
response.parse_response_content(response_content)
|
||
|
||
if response.is_success():
|
||
logger.info(f"Alipay refund succeeded: order_no={order.order_no}")
|
||
return {"success": True, "trade_no": response.trade_no}
|
||
else:
|
||
logger.error(
|
||
f"Alipay refund 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 {
|
||
"success": False,
|
||
"message": f"支付宝退款失败: {response.sub_msg or response.msg}"
|
||
}
|
||
except Exception as e:
|
||
logger.exception(f"Alipay refund exception: order_no={order.order_no}, {e}")
|
||
return {"success": False, "message": f"支付宝退款异常: {str(e)}"}
|