增加微信支付的逻辑

DEPLOYMENT.md

@@ -22,8 +22,11 @@ video_item/
 | Redis | >= 6 | 可选，推荐用于限流/验证码/Celery |
 | FFmpeg | 任意 | 可选，用于视频封面截帧 |
 | alipay-sdk-python | >=3.7.1160 | 可选，用于支付 |
+| wechatpayv3 | >=2.0.2  | 可选，用于支付 |
+| volcengine-python-sdk | >=1.1.0 | 可选，用于视频生成 |
 | ca-certificates | 任意 | **必须**，HTTPS 请求需要（新服务器/容器常缺） |
 
+
 ---
 
 ## 二、后端部署 (video-gen-api)
@@ -58,8 +61,13 @@ pip install -e ".[pg,redis,celery]"
 #安装阿里支付sdk
 pip install -e ".[pg,redis,celery,alipay]"
 
+# 安装微信支付sdk
+pip install -e ".[pg,redis,celery,alipay,wechatpayv3]"
+
 #安装火山sdk
-pip install -e ".[pg,redis,celery,alipay,volc]"
+pip install -e ".[pg,redis,celery,alipay,wechatpayv3,volc]"
+
+
 ```
 
 ### 2. 配置环境变量

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

@@ -83,12 +83,86 @@ async def recharge(
 
 @router.post("/wechat/callback")
 async def wechat_callback(request: Request, db: AsyncSession = Depends(get_db)):
-    data = await request.json()
+    # 读取微信支付回调数据
-    if not await verify_wechat_callback(data, db):
+    body_bytes = await request.body()
-        raise HTTPException(status_code=400, detail="签名验证失败")
+    body_str = body_bytes.decode("utf-8")
+    
+    # 获取配置
+    from app.services.payment import _get_payment_configs, _is_mock_mode, _get_wechat_client
+    db_configs = await _get_payment_configs(db)
+    
+    # 检查 mock 模式
+    if _is_mock_mode(db_configs):
+        try:
+            import json
+            data = json.loads(body_str) if body_str else {}
             order_no = data.get("out_trade_no")
             if order_no:
                 await process_payment_success_by_order_no(db, order_no)
+            logger.info(f"Mock WeChat callback processed: order_no={order_no}")
+            return {"code": "SUCCESS", "message": "OK"}
+        except Exception as e:
+            logger.exception(f"Mock WeChat callback error: {e}")
+            return {"code": "SUCCESS", "message": "OK"}  # 微信要求即使处理失败也返回成功
+    
+    # 真实模式：使用 wechatpayv3 SDK 验证回调并解析数据
+    try:
+        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", "")
+        gateway = db_configs.get("payment_wechat_gateway", "")
+        
+        client = _get_wechat_client(mch_id, private_key, cert_serial_no, api_v3_key, gateway)
+        if not client:
+            logger.error("WeChat client not initialized for callback")
+            return {"code": "SUCCESS", "message": "OK"}
+        
+        # 从请求头获取必要信息
+        headers = dict(request.headers)
+        timestamp = headers.get("Wechatpay-Timestamp", "")
+        nonce = headers.get("Wechatpay-Nonce", "")
+        signature = headers.get("Wechatpay-Signature", "")
+        serial_no = headers.get("Wechatpay-Serial", "")
+        
+        # 验证签名
+        is_verified = client.verify(
+            timestamp=timestamp,
+            nonce=nonce,
+            body=body_str,
+            signature=signature,
+            serial_no=serial_no
+        )
+        
+        if not is_verified:
+            logger.warning("WeChat callback signature verification failed")
+            raise HTTPException(status_code=400, detail="签名验证失败")
+        
+        # 解密回调数据
+        decrypted_data = client.decrypt(body_str)
+        if not decrypted_data:
+            logger.error("WeChat callback decryption failed")
+            raise HTTPException(status_code=400, detail="数据解密失败")
+        
+        # 处理支付成功回调
+        if decrypted_data.get("event_type") == "TRANSACTION.SUCCESS":
+            resource = decrypted_data.get("resource", {})
+            order_no = resource.get("out_trade_no", "")
+            transaction_id = resource.get("transaction_id", "")
+            amount_info = resource.get("amount", {})
+            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}"
+                )
+        
+        return {"code": "SUCCESS", "message": "OK"}
+    except Exception as e:
+        logger.exception(f"WeChat callback processing error: {e}")
+        # 微信支付要求即使处理失败也返回成功，避免重复回调
         return {"code": "SUCCESS", "message": "OK"}
 
 
@@ -183,13 +257,19 @@ async def cancel_order(
     if order.status != "pending":
         raise HTTPException(status_code=400, detail=f"订单状态为{order.status}，无法取消")
     
-    # If it's an Alipay order, call close API first
+    # If it's an Alipay or WeChat order, call close API first
-    if order.payment_method == "alipay":
     db_configs = await _get_payment_configs(db)
+    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}")
+    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}")
     
     order.status = "cancelled"
     await db.flush()

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

@@ -143,12 +143,17 @@ async def _check_and_expire_order(db: AsyncSession, order: PaymentOrder) -> bool
             f"ORDER_EXPIRED order_no={order.order_no} user={order.user_id} "
             f"amount={order.amount} created_at={order.created_at.isoformat()}"
         )
-        # Also call Alipay close API if it was an Alipay order
+        # 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
 
@@ -174,12 +179,17 @@ async def expire_all_pending_orders(db: AsyncSession) -> int:
         logger.info(
             f"ORDER_EXPIRED order_no={o.order_no} user={o.user_id} amount={o.amount}"
         )
-        # Also call Alipay close API if it was an Alipay order
+        # 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()
     return expired_count
@@ -269,8 +279,15 @@ async def create_recharge_order(
             if not db_configs.get("payment_alipay_app_id") or not db_configs.get("payment_alipay_private_key"):
                 raise ValueError("支付宝支付未完成配置，请联系管理员")
         elif method == "wechat":
-            if not db_configs.get("payment_wechat_mch_id") or not db_configs.get("payment_wechat_api_key"):
+            required_configs = [
-                raise ValueError("微信支付未完成配置，请联系管理员")
+                "payment_wechat_mch_id",
+                "payment_wechat_private_key",
+                "payment_wechat_cert_serial_no",
+                "payment_wechat_api_v3_key"
+            ]
+            missing_configs = [c for c in required_configs if not db_configs.get(c)]
+            if missing_configs:
+                raise ValueError(f"微信支付未完成配置，缺少: {', '.join(missing_configs)}，请联系管理员")
 
     total_credits = credits + bonus_credits
     order = PaymentOrder(
@@ -307,7 +324,13 @@ async def create_recharge_order(
     else:
         # Real payment: delegate to WeChat or Alipay
         if method == "wechat":
-            _create_wechat_order(order, db_configs)
+            qr_code_content = _create_wechat_order(order, db_configs)
+            if qr_code_content:
+                # Attach QR code content to the order instance (transient, not persisted)
+                order.qr_url = qr_code_content  # type: ignore[attr-defined]
+            else:
+                # Precreate failed — do not leave a pending order that can never be paid
+                raise ValueError("微信支付预下单失败，请检查配置或稍后重试")
         elif method == "alipay":
             qr_url = _create_alipay_order(order, db_configs)
             if qr_url:
@@ -321,21 +344,231 @@ async def create_recharge_order(
 
 
 # ---------------------------------------------------------------------------
-# WeChat (stub)
+# WeChat Pay client (lazy singleton, recreated when config changes)
 # ---------------------------------------------------------------------------
+_wechat_client = None
+_wechat_mch_id = None
 
 
-def _create_wechat_order(order: PaymentOrder, db_configs: dict[str, str]) -> None:
+def _get_wechat_client(
-    """Create a WeChat Pay order. Stub for real integration."""
+    mch_id: str,
-    mch_id = db_configs.get("payment_wechat_mch_id", "")
+    private_key: str,
-    api_key = db_configs.get("payment_wechat_api_key", "")
+    cert_serial_no: str,
-    if not mch_id or not api_key:
+    api_v3_key: str,
-        logger.warning("WeChat payment config missing in database")
+    gateway: str = ""
-        return
+):
-    logger.info(
+    """Get or create a WeChat Pay client. Recreated if config changes."""
-        f"WeChat order created: mch_id={mch_id}, "
+    global _wechat_client, _wechat_mch_id
-        f"order_no={order.order_no}, amount={order.amount}"
+
+    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:
+        # 处理私钥格式，确保是正确的 PEM 格式
+        private_key_str = private_key.strip()
+        if not private_key_str.startswith("-----BEGIN"):
+            private_key_str = "-----BEGIN PRIVATE KEY-----\n" + private_key_str + "\n-----END PRIVATE KEY-----"
+        
+        # 初始化微信支付客户端
+        _wechat_client = WeChatPay(
+            wechatpay_type=WeChatPayType.NATIVE,
+            mchid=mch_id,
+            private_key=private_key_str,
+            cert_serial_no=cert_serial_no,
+            apiv3_private_key=api_v3_key,
+            gateway=gateway or "https://api.mch.weixin.qq.com",
+        )
+        _wechat_mch_id = mch_id
+        logger.info("WeChat Pay client initialized successfully")
+        return _wechat_client
+    except Exception as e:
+        logger.exception(f"Failed to initialize WeChat Pay client: {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", "")
+    gateway = db_configs.get("payment_wechat_gateway", "")
+    notify_url = db_configs.get("payment_wechat_notify_url", "")
+
+    if not all([mch_id, private_key, cert_serial_no, api_v3_key]):
+        logger.warning("WeChat payment config missing in database")
+        return None
+
+    client = _get_wechat_client(mch_id, private_key, cert_serial_no, api_v3_key, gateway)
+    if client is None:
+        return None
+
+    try:
+        # 调用微信支付 Native 下单接口
+        result = client.pay(
+            description=f"充值订单 {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",
+            }
+        )
+        
+        if result.get('code_url'):
+            # 注意：微信返回的 code_url 可能需要进一步处理成二维码图片地址
+            logger.info(f"WeChat order created successfully: order_no={order.order_no}")
+            return result.get('code_url')
+        else:
+            logger.error(
+                f"WeChat pay failed: order_no={order.order_no}, "
+                f"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]) -> bool:
+    """Call WeChat Pay close API to close an unpaid order.
+    Returns True if the order was closed successfully.
+    """
+    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", "")
+    gateway = db_configs.get("payment_wechat_gateway", "")
+
+    client = _get_wechat_client(mch_id, private_key, cert_serial_no, api_v3_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_wechat_order for {order.order_no}")
+        return True
+
+    try:
+        result = client.close(out_trade_no=order.order_no)
+        if result:
+            logger.info(f"WeChat order closed: order_no={order.order_no}")
+            return True
+        else:
+            logger.error(f"WeChat close failed: order_no={order.order_no}")
+            return False
+    except Exception as e:
+        logger.exception(f"WeChat close exception: order_no={order.order_no}")
+        return False
+
+
+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", "")
+    gateway = db_configs.get("payment_wechat_gateway", "")
+
+    client = _get_wechat_client(mch_id, private_key, cert_serial_no, api_v3_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_wechat_order for {order.order_no}")
+        return {"trade_state": "SUCCESS"}
+
+    try:
+        result = client.query(out_trade_no=order.order_no)
+        if result.get('trade_state'):
+            logger.info(
+                f"WeChat query succeeded: order_no={order.order_no}, "
+                f"trade_state={result.get('trade_state')}"
+            )
+            return result
+        else:
+            logger.error(
+                f"WeChat query failed: order_no={order.order_no}, "
+                f"result={result}"
+            )
+            return None
+    except Exception as e:
+        logger.exception(f"WeChat query exception: order_no={order.order_no}")
+        return None
+
+
+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."""
+    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", "")
+    gateway = db_configs.get("payment_wechat_gateway", "")
+
+    client = _get_wechat_client(mch_id, private_key, cert_serial_no, api_v3_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 wechat refund for {order.order_no}")
+        return {"success": True}
+
+    try:
+        out_refund_no = f"{order.order_no}_refund_{int(datetime.now().timestamp())}"
+        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
+        )
+        
+        if result.get('status') == 'SUCCESS':
+            logger.info(f"WeChat refund succeeded: order_no={order.order_no}")
+            return {"success": True, "refund_id": result.get('refund_id')}
+        else:
+            logger.error(
+                f"WeChat refund failed: order_no={order.order_no}, "
+                f"result={result}"
+            )
+            return {
+                "success": False, 
+                "message": f"微信退款失败: {result.get('code', '')}"
+            }
+    except Exception as e:
+        logger.exception(f"WeChat refund exception: order_no={order.order_no}")
+        return {"success": False, "message": f"微信退款异常: {str(e)}"}
 
 
 # ---------------------------------------------------------------------------
@@ -562,7 +795,7 @@ async def _query_alipay_order(db: AsyncSession, order: PaymentOrder, db_configs:
 
 
 async def sync_pending_orders(db: AsyncSession) -> int:
-    """Check pending orders via Alipay query and update status.
+    """Check pending orders via Alipay/WeChat query and update status.
     Returns the number of orders updated.
     """
     result = await db.execute(
@@ -576,23 +809,38 @@ async def sync_pending_orders(db: AsyncSession) -> int:
     db_configs = await _get_payment_configs(db)
 
     for order in orders:
-        if order.payment_method != "alipay":
-            continue
-
         try:
+            if order.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", "")
-                    await process_payment_success_by_order_no(db, order.order_no, 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
                     elif trade_status in ("TRADE_CLOSED", "TRADE_CANCELLED"):
                         # Order was closed on Alipay side
                         order.status = "cancelled"
                         await db.flush()
                         updated_count += 1
+            elif order.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
+                    elif trade_state in ("CLOSED", "REVOKED"):
+                        # Order was closed on WeChat side
+                        order.status = "cancelled"
+                        await db.flush()
+                        updated_count += 1
         except Exception as e:
             logger.exception(f"Failed to sync order {order.order_no}: {e}")
 
@@ -747,12 +995,21 @@ def _verify_alipay_sign(public_key: str, sign_content: str, sign: str, sign_type
 
 
 async def verify_wechat_callback(data: dict, db: AsyncSession) -> bool:
-    """Verify WeChat payment callback signature."""
+    """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
-    logger.info("WeChat callback verification (real mode not implemented)")
+    
+    # 对于真实模式，我们在回调路由处理器中直接使用 SDK 验证
+    # 这里我们返回 True 以保持接口一致性
+    logger.info("WeChat callback verification (delegated to SDK in router)")
     return True
 
 
@@ -881,7 +1138,7 @@ async def process_refund(
     if refund_amount > order.amount:
         return {"success": False, "message": "退款金额超过订单金额"}
     
-    # 如果是支付宝订单，调用支付宝退款API
+    # 根据支付方式调用相应的退款API
     db_configs = await _get_payment_configs(db)
     if order.payment_method == "alipay":
         refund_result = await _refund_alipay_order(
@@ -889,6 +1146,12 @@ async def process_refund(
         )
         if not refund_result.get("success"):
             return refund_result
+    elif order.payment_method == "wechat":
+        refund_result = await _refund_wechat_order(
+            db, order, refund_amount, refund_reason, db_configs
+        )
+        if not refund_result.get("success"):
+            return refund_result
     
     # 扣除积分
     try:

video-gen-api/pyproject.toml

@@ -27,6 +27,7 @@ pg = ["asyncpg>=0.30.0"]
 redis = ["redis>=5.2.0"]
 celery = ["celery>=5.4.0", "redis>=5.2.0"]
 alipay = ["alipay-sdk-python>=3.7.1160"]
+wxpay = ["wechatpayv3>=2.0.2"]
 volc = ["volcengine-python-sdk>=1.1.0"]
 dev = [
     "pytest>=8.3.0",