From 83e8ceee7e48c0eb54bb19dcd61ef8e16b603001 Mon Sep 17 00:00:00 2001 From: wwwwwwwww <526125649@qq.com> Date: Thu, 11 Jun 2026 18:13:03 +0800 Subject: [PATCH] =?UTF-8?q?=E5=A2=9E=E5=8A=A0=E5=BE=AE=E4=BF=A1=E6=94=AF?= =?UTF-8?q?=E4=BB=98=E7=9A=84=E9=80=BB=E8=BE=91?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- DEPLOYMENT.md | 10 +- video-gen-api/app/api/v1/payments.py | 98 +++++++- video-gen-api/app/services/payment.py | 333 +++++++++++++++++++++++--- video-gen-api/pyproject.toml | 1 + 4 files changed, 397 insertions(+), 45 deletions(-) diff --git a/DEPLOYMENT.md b/DEPLOYMENT.md index f33f9823..05376543 100644 --- a/DEPLOYMENT.md +++ b/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. 配置环境变量 diff --git a/video-gen-api/app/api/v1/payments.py b/video-gen-api/app/api/v1/payments.py index 0390ea7c..44edb008 100644 --- a/video-gen-api/app/api/v1/payments.py +++ b/video-gen-api/app/api/v1/payments.py @@ -83,13 +83,87 @@ 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): - raise HTTPException(status_code=400, detail="签名验证失败") - order_no = data.get("out_trade_no") - if order_no: - await process_payment_success_by_order_no(db, order_no) - return {"code": "SUCCESS", "message": "OK"} + # 读取微信支付回调数据 + body_bytes = await request.body() + 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"} @router.post("/alipay/callback") @@ -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 + db_configs = await _get_payment_configs(db) if order.payment_method == "alipay": - db_configs = await _get_payment_configs(db) 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() diff --git a/video-gen-api/app/services/payment.py b/video-gen-api/app/services/payment.py index 57cbca9c..5f0133f0 100644 --- a/video-gen-api/app/services/payment.py +++ b/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"): - raise ValueError("微信支付未完成配置，请联系管理员") + required_configs = [ + "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: - """Create a WeChat Pay order. Stub for real integration.""" +def _get_wechat_client( + mch_id: str, + private_key: str, + cert_serial_no: str, + api_v3_key: str, + gateway: str = "" +): + """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: + # 处理私钥格式，确保是正确的 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", "") - api_key = db_configs.get("payment_wechat_api_key", "") - if not mch_id or not api_key: + 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 - logger.info( - f"WeChat order created: mch_id={mch_id}, " - f"order_no={order.order_no}, amount={order.amount}" - ) + 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: - 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) - 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 + 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", "") + 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: diff --git a/video-gen-api/pyproject.toml b/video-gen-api/pyproject.toml index b7721cdb..68477bd0 100644 --- a/video-gen-api/pyproject.toml +++ b/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",