1、缺少支付幂等性保障

2、增加事务
3、增加金额一致性判断
4、增加后台退款功能

video-gen-api/app/api/v1/admin.py

@@ -43,6 +43,7 @@ from app.services.notification import create_notification
 from app.services.auth import hash_password, verify_password
 from app.services.operation_log import log_operation
 from app.services.resource_signed_url_service import build_resource_signed_url
+from app.services.payment import sync_pending_orders, process_refund
 
 from app.services.generation_billing_service import (
     OWNER_GENERATION_RECORD,
@@ -457,6 +458,7 @@ async def get_payment_stats(
         "pending": {"count": 0, "amount": 0.0},
         "paid": {"count": 0, "amount": 0.0},
         "cancelled": {"count": 0, "amount": 0.0},
+        "refunded": {"count": 0, "amount": 0.0},
     }
 
     # Parse dates and build base query filters
@@ -570,7 +572,7 @@ async def get_payment_stats(
                 "amount": round(o.amount, 2),
                 "credits": round(o.credits, 2),
                 "payment_method": o.payment_method,
-                "status": o.status if o.status in ("pending", "paid", "cancelled") else "cancelled",
+                "status": o.status if o.status in ("pending", "paid", "cancelled", "refunded") else "cancelled",
                 "trade_no": o.trade_no,
                 "paid_at": _iso(o.paid_at),
                 "created_at": _iso(o.created_at),
@@ -622,7 +624,7 @@ async def get_admin_payment_orders(
                 "amount": round(o.amount, 2),
                 "credits": round(o.credits, 2),
                 "payment_method": o.payment_method,
-                "status": o.status if o.status in ("pending", "paid", "cancelled") else "cancelled",
+                "status": o.status if o.status in ("pending", "paid", "cancelled", "refunded") else "cancelled",
                 "trade_no": o.trade_no,
                 "paid_at": _iso(o.paid_at),
                 "created_at": _iso(o.created_at),
@@ -660,6 +662,19 @@ async def update_payment_config(
     }
 
 
+@router.post("/payment-orders/{order_no}/refund")
+async def refund_payment_order(
+    order_no: str,
+    admin: User = Depends(get_admin_user),
+    db: AsyncSession = Depends(get_db),
+):
+    """Refund a paid payment order."""
+    result = await process_refund(db, order_no)
+    if not result.get("success"):
+        raise HTTPException(status_code=400, detail=result.get("message", "退款失败"))
+    return result
+
+
 # ── Industry Config ──────────────────────────────────────
 
 def _serialize_industry(ind: IndustryConfig) -> dict:

video-gen-api/app/api/v1/payments.py

@@ -16,6 +16,7 @@ from app.services.payment import (
     verify_wechat_callback,
     verify_alipay_callback,
     process_payment_success_by_order_no,
+    process_refund,
     _get_payment_configs,
     _close_alipay_order,
     _get_order_expire_seconds,
@@ -95,6 +96,7 @@ async def wechat_callback(request: Request, db: AsyncSession = Depends(get_db)):
 async def alipay_callback(request: Request, db: AsyncSession = Depends(get_db)):
     form_data = await request.form()
     data = dict(form_data)
+    
     logger.info(
         f"ALIPAY_CALLBACK order_no={data.get('out_trade_no')} "
         f"data={data}"
@@ -112,8 +114,11 @@ async def alipay_callback(request: Request, db: AsyncSession = Depends(get_db)):
 
     order_no = data.get("out_trade_no")
     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
+    
     if order_no:
-        await process_payment_success_by_order_no(db, order_no, trade_no)
+        await process_payment_success_by_order_no(db, order_no, trade_no, total_amount)
 
     return "success"
 

video-gen-api/app/services/payment.py

@@ -1,24 +1,15 @@
 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.services.credits import add_credits, deduct_credits
 from app.utils.id_gen import generate_id, generate_order_no
 
 # ---------------------------------------------------------------------------
@@ -102,7 +93,7 @@ def _patch_alipay_webutils():
                     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 = _http.HTTPSConnection(parsed.hostname)
                     conn.request("POST", parsed.path + "?" + query_string, params, headers)
                     resp = conn.getresponse()
                     body = resp.read().decode("utf-8", errors="replace")
@@ -233,17 +224,9 @@ def _get_alipay_client(app_id: str, private_key: str, public_key: str, gateway:
         _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
+        logger.exception("Failed to initialize Alipay client with SSL verification")
+        _alipay_client = None
+        _alipay_client_app_id = None
 
     return _alipay_client
 
@@ -746,7 +729,7 @@ def _verify_alipay_sign(public_key: str, sign_content: str, sign: str, sign_type
                 logger.error("Neither cryptography nor rsa library installed, cannot verify signature")
                 # 如果没有任何加密库，在生产环境应该返回 False，但这里我们记录警告并继续
                 logger.warning("Skipping signature verification due to missing crypto libraries")
-                return True
+                return False
                 
     except Exception as e:
         logger.exception(f"Signature verification failed: {e}")
@@ -776,7 +759,7 @@ 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).limit(1)
+        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":
@@ -791,23 +774,50 @@ async def process_payment_success(db: AsyncSession, order_id: str):
         f"充值成功({order.credits}积分)",
         related_id=order.id,
     )
-    await db.flush()
+    await db.commit()
 
 
-async def process_payment_success_by_order_no(db: AsyncSession, order_no: str, trade_no: str = ""):
+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).
 
     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
     """
     result = await db.execute(
-        select(PaymentOrder).where(PaymentOrder.order_no == order_no).limit(1)
+        select(PaymentOrder).where(PaymentOrder.order_no == order_no).with_for_update().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")
+    
+    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(total_amount - order.amount) > 0.01:
+        logger.error(
+            f"Amount mismatch: order amount {order.amount}, gateway amount {total_amount}"
+        )
+        return
+
+    # 幂等性检查：如果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
 
     order.status = "paid"
@@ -822,8 +832,143 @@ async def process_payment_success_by_order_no(db: AsyncSession, order_no: str, t
         f"充值成功({order.credits}积分)",
         related_id=order.id,
     )
-    await db.flush()
+    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}"
     )
+
+
+async def process_refund(
+    db: AsyncSession,
+    order_no: str,
+    refund_amount: float | None = None,
+    refund_reason: str = "管理员退款"
+) -> dict:
+    """Process a refund for a paid order.
+
+    Args:
+        db: async database session
+        order_no: merchant order number
+        refund_amount: amount to refund (defaults to full order amount)
+        refund_reason: reason for refund
+
+    Returns:
+        dict with refund result
+    """
+    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:
+        return {"success": False, "message": "订单不存在"}
+    
+    if order.status != "paid":
+        return {"success": False, "message": f"订单状态为{order.status}，无法退款"}
+    
+    if order.refunded_at is not None:
+        return {"success": False, "message": "订单已退款"}
+
+    refund_amount = refund_amount or order.amount
+    
+    # 金额校验
+    if refund_amount > order.amount:
+        return {"success": False, "message": "退款金额超过订单金额"}
+    
+    # 如果是支付宝订单，调用支付宝退款API
+    db_configs = await _get_payment_configs(db)
+    if order.payment_method == "alipay":
+        refund_result = await _refund_alipay_order(
+            db, order, refund_amount, refund_reason, db_configs
+        )
+        if not refund_result.get("success"):
+            return refund_result
+    
+    # 扣除积分
+    try:
+        await deduct_credits(
+            db,
+            order.user_id,
+            order.credits,
+            refund_reason,
+            related_id=order.id,
+        )
+    except Exception as e:
+        logger.exception(f"Failed to deduct credits for refund: {e}")
+        return {"success": False, "message": "积分扣除失败"}
+    
+    # 更新订单状态
+    order.status = "refunded"
+    order.refund_amount = refund_amount
+    order.refunded_at = datetime.now()
+    if order.payment_method == "alipay":
+        order.refund_trade_no = db_configs.get("refund_trade_no", "")
+    
+    await db.commit()
+    logger.info(
+        f"REFUND_SUCCESS order_no={order_no} user={order.user_id} "
+        f"refund_amount={refund_amount}"
+    )
+    return {"success": True, "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)}"}