From 809ff4712011dab741efa5bd383ae770c968e681 Mon Sep 17 00:00:00 2001 From: GinHa <15201596918@163.com> Date: Tue, 11 Aug 2026 11:57:55 +0800 Subject: [PATCH] =?UTF-8?q?=E4=BC=9A=E5=91=98=E7=A7=AF=E5=88=86=E6=94=B9?= =?UTF-8?q?=E7=89=88V6=20=E4=BF=AE=E5=A4=8D=E8=AE=A2=E5=8D=95=E5=BC=82?= =?UTF-8?q?=E6=AD=A5=E6=97=B6=E5=8C=BA=E6=A0=A1=E9=AA=8CBUG?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- video-gen-api/app/api/v1/payments.py | 56 +++- video-gen-api/app/services/payment.py | 401 ++++++++++++++++---------- 2 files changed, 288 insertions(+), 169 deletions(-) diff --git a/video-gen-api/app/api/v1/payments.py b/video-gen-api/app/api/v1/payments.py index fc254473..10d47cfc 100644 --- a/video-gen-api/app/api/v1/payments.py +++ b/video-gen-api/app/api/v1/payments.py @@ -217,11 +217,30 @@ async def wechat_callback(request: Request, db: AsyncSession = Depends(get_db)): total_amount = amount_info.get("total", 0) / 100 # 转换为元 if order_no: - await process_payment_success_by_order_no(db, order_no, transaction_id, total_amount) - logger.info( - f"WeChat callback processed: order_no={order_no}, " - f"transaction_id={transaction_id}, amount={total_amount}" - ) + try: + processed = await process_payment_success_by_order_no( + db, order_no, transaction_id, total_amount + ) + if processed: + logger.info( + f"WeChat callback processed: order_no={order_no}, " + f"transaction_id={transaction_id}, amount={total_amount}" + ) + else: + logger.warning( + f"WeChat payment success not fulfilled locally: order_no={order_no}, " + f"transaction_id={transaction_id}, amount={total_amount}" + ) + except Exception as e: + # 微信已经通过验签、解密并明确通知支付成功。 + # 本地履约异常属于项目内部故障：回滚本地事务，但仍向微信返回 SUCCESS， + # 后续由 pending 主动查单重试或日志人工对账处理。 + await db.rollback() + logger.exception( + f"WeChat payment fulfillment error: order_no={order_no}, " + f"transaction_id={transaction_id}, error={e}" + ) + return {"code": "SUCCESS", "message": "OK"} # 处理退款回调 elif event_type == "REFUND.SUCCESS": @@ -402,19 +421,26 @@ async def cancel_order( if order.status != "pending": raise HTTPException(status_code=400, detail=f"订单状态为{order.status}，无法取消") - # If it's an Alipay or WeChat order, call close API first + # Alipay/WeChat must confirm remote close before local cancellation. + # If close fails or the remote state is uncertain, keep the local order + # pending so a later gateway query can still discover a successful payment. db_configs = await _get_payment_configs(db) + closed = True 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_no}: {e}") + closed = await _close_alipay_order(db, order, db_configs) elif order.payment_method == "wechat": - try: - from app.services.payment import _close_wechat_order - await _close_wechat_order(db, order, db_configs) - except Exception as e: - logger.exception(f"Failed to close WeChat order {order_no}: {e}") + from app.services.payment import _close_wechat_order + closed = await _close_wechat_order(db, order, db_configs) + + if not closed: + logger.warning( + f"ORDER_CANCEL_CLOSE_PENDING order_no={order_no} " + f"user={current_user.id} method={order.payment_method}" + ) + raise HTTPException( + status_code=409, + detail="支付渠道暂未确认订单关闭，请稍后重试", + ) order.status = "cancelled" if order.upgrade_period_ids_json: diff --git a/video-gen-api/app/services/payment.py b/video-gen-api/app/services/payment.py index 3c267a1d..c75eb6b4 100644 --- a/video-gen-api/app/services/payment.py +++ b/video-gen-api/app/services/payment.py @@ -170,46 +170,69 @@ async def _get_payment_configs(db: AsyncSession) -> dict[str, str]: 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], +) -> bool: + """Close a gateway order before changing the local order to cancelled. + + For Alipay/WeChat, local cancellation is only allowed after the gateway + confirms that the order has been closed. Unknown/legacy payment methods + keep the historical local-only cancellation behaviour. + """ + if order.payment_method == "alipay": + return await _close_alipay_order(db, order, db_configs) + if order.payment_method == "wechat": + return await _close_wechat_order(db, order, db_configs) + return True + + 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. + """Expire a pending order only after the payment gateway confirms close. + + Returns True only when the local order was actually changed to cancelled. + If the gateway close result is uncertain/failed, the order remains pending + so that later query/reconciliation can still discover a successful payment. """ 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" - if order.upgrade_period_ids_json: - await release_upgrade_reservation(db, order=order, released_at=utc_now()) - 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()}" + if utc_now() < expiry: + return False + + closed = await _close_order_for_cancellation(db, order, db_configs) + if not closed: + logger.warning( + f"ORDER_EXPIRE_CLOSE_PENDING order_no={order.order_no} " + f"user={order.user_id} amount={order.amount}" ) - # Also call close API if it was an Alipay or WeChat 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}") - elif order.payment_method == "wechat": - try: - await _close_wechat_order(db, order, db_configs) - except Exception as e: - logger.exception(f"Failed to close WeChat order {order.order_no}: {e}") - return True - return False + return False + + order.status = "cancelled" + if order.upgrade_period_ids_json: + await release_upgrade_reservation(db, order=order, released_at=utc_now()) + 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: mark all expired pending orders as cancelled. - Returns the number of orders expired. + """Background task: close and cancel expired pending orders. + + Alipay/WeChat orders remain pending when the remote close cannot be + confirmed. This prevents a real paid order from being persisted locally + as cancelled. """ db_configs = await _get_payment_configs(db) expire_seconds = _get_order_expire_seconds(db_configs) - threshold = datetime.now() - timedelta(seconds=expire_seconds) + threshold = utc_now() - timedelta(seconds=expire_seconds) result = await db.execute( select(PaymentOrder).where( PaymentOrder.status == "pending", @@ -218,27 +241,27 @@ async def expire_all_pending_orders(db: AsyncSession) -> int: ) orders = result.scalars().all() expired_count = 0 - for o in orders: - o.status = "cancelled" - if o.upgrade_period_ids_json: - await release_upgrade_reservation(db, order=o, released_at=utc_now()) + + for order in orders: + closed = await _close_order_for_cancellation(db, order, db_configs) + if not closed: + logger.warning( + f"ORDER_EXPIRE_CLOSE_PENDING order_no={order.order_no} " + f"user={order.user_id} amount={order.amount}" + ) + continue + + order.status = "cancelled" + if order.upgrade_period_ids_json: + await release_upgrade_reservation(db, order=order, released_at=utc_now()) expired_count += 1 logger.info( - f"ORDER_EXPIRED order_no={o.order_no} user={o.user_id} amount={o.amount}" + f"ORDER_EXPIRED order_no={order.order_no} user={order.user_id} " + f"amount={order.amount}" ) - # Also call close API if it was an Alipay or WeChat 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}") - elif o.payment_method == "wechat": - try: - await _close_wechat_order(db, o, db_configs) - except Exception as e: - logger.exception(f"Failed to close WeChat order {o.order_no}: {e}") - if orders: - await db.flush() + + if expired_count > 0: + await db.commit() return expired_count @@ -409,19 +432,15 @@ async def create_recharge_order( if mock_mode: order.status = "paid" order.paid_at = checked_at - if product: - await fulfill_payment_product(db, order=order, fulfilled_at=checked_at) - else: - 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, - 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, - ) + desc = f"充值{label}({total_credits}积分)" + if bonus_credits > 0: + desc += f"(含赠送{bonus_credits}积分)" + await _fulfill_paid_order( + db, + order=order, + fulfilled_at=checked_at, + legacy_description=desc, + ) await db.flush() else: if method == "wechat": @@ -953,60 +972,78 @@ async def _query_alipay_order(db: AsyncSession, order: PaymentOrder, db_configs: async def sync_pending_orders(db: AsyncSession) -> int: """Check pending orders via Alipay/WeChat query and update status. - Returns the number of orders updated. + + 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).where( + select(PaymentOrder.order_no, PaymentOrder.payment_method).where( PaymentOrder.status == "pending", ) ) - orders = result.scalars().all() + 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 in orders: + for order_no, payment_method in candidates: try: - if order.payment_method == "alipay": + 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"): - # Order was paid but we missed the callback 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 - await process_payment_success_by_order_no(db, order.order_no, trade_no, total_amount) - updated_count += 1 + 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 was closed on Alipay side order.status = "cancelled" if order.upgrade_period_ids_json: - await release_upgrade_reservation(db, order=order, released_at=utc_now()) - await db.flush() + await release_upgrade_reservation( + db, order=order, released_at=utc_now() + ) + await db.commit() updated_count += 1 - elif order.payment_method == "wechat": + + 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": - # Order was paid but we missed the callback transaction_id = data.get("transaction_id", "") total_amount = float(data.get("amount", {}).get("total", 0)) / 100 - await process_payment_success_by_order_no(db, order.order_no, transaction_id, total_amount) - updated_count += 1 + 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 was closed on WeChat side order.status = "cancelled" if order.upgrade_period_ids_json: - await release_upgrade_reservation(db, order=order, released_at=utc_now()) - await db.flush() + await release_upgrade_reservation( + db, order=order, released_at=utc_now() + ) + await db.commit() updated_count += 1 except Exception as e: - logger.exception(f"Failed to sync order {order.order_no}: {e}") + await db.rollback() + logger.exception(f"Failed to sync order {order_no}: {e}") - if updated_count > 0: - await db.flush() return updated_count @@ -1179,96 +1216,152 @@ async def verify_wechat_callback(data: dict, db: AsyncSession) -> bool: # --------------------------------------------------------------------------- -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).with_for_update().limit(1) - ) - order = result.scalar_one_or_none() - if not order or order.status != "pending": +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 - order.status = "paid" - order.paid_at = datetime.now() - if order.product_id: - await fulfill_payment_product(db, order=order, fulfilled_at=order.paid_at) - else: - await add_credits( - db, order.user_id, order.credits, f"充值成功({order.credits}积分)", - 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, + 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) ) - await db.commit() + 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 -): - """Process successful payment by order_no (used by Alipay/WeChat callbacks). + total_amount: float | None = None, +) -> bool: + """Atomically process an Alipay/WeChat payment success. - Args: - db: async database session - order_no: the merchant order number (out_trade_no) - trade_no: the Alipay trade number (trade_no), optional - total_amount: the payment amount from the gateway, for consistency check + ``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. """ - 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 - - if order.status == "paid": - logger.info(f"Order {order_no} already processed, skipping") - return - - if order.status != "pending": - logger.info(f"Order {order_no} is in {order.status} state, cannot process") - return - - # 金额一致性校验 - 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}" + try: + result = await db.execute( + select(PaymentOrder) + .where(PaymentOrder.order_no == order_no) + .with_for_update() + .limit(1) ) - return + order = result.scalar_one_or_none() - # 幂等性检查：如果trade_no已存在且相同，则跳过 - if trade_no and order.trade_no and order.trade_no == trade_no: - logger.info(f"Trade no {trade_no} already processed, skipping") - return + if not order: + logger.info(f"Order {order_no} not found, skipping") + return False - order.status = "paid" - order.paid_at = datetime.now() - if trade_no: - order.trade_no = trade_no + if order.status == "paid": + logger.info(f"Order {order_no} already processed, skipping") + return True - if order.product_id: - await fulfill_payment_product(db, order=order, fulfilled_at=order.paid_at) - else: - await add_credits( - db, order.user_id, order.credits, - f"充值成功({order.credits}积分), 订单号: {order_no}, 金额: {order.amount}", - 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, + 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}" + ), ) - await db.commit() - logger.info( - f"PAYMENT_SUCCESS order_no={order_no} user={order.user_id} " - f"amount={order.amount} credits={order.credits} trade_no={trade_no}" - ) + + await db.commit() + logger.info( + f"PAYMENT_SUCCESS order_no={order_no} user={order.user_id} " + f"amount={order.amount} credits={order.credits} trade_no={trade_no}" + ) + return True + except Exception: + await db.rollback() + raise async def process_refund( @@ -1345,7 +1438,7 @@ async def process_refund( # 更新订单状态 order.status = "refunded" order.refund_amount = refund_amount - order.refunded_at = datetime.now() + order.refunded_at = utc_now() if order.payment_method == "alipay": order.refund_trade_no = db_configs.get("refund_trade_no", "")