'env'

app/payment/zpay.py

@@ -5,12 +5,27 @@ ZPAY 支付渠道实现
 """
 import hashlib
 import json
+import logging
 from typing import Any, Dict, Optional
 
 import requests
 
 from .base import PaymentProvider
 
+logger = logging.getLogger(__name__)
+
+
+def _is_internal_ip(ip: Optional[str]) -> bool:
+    """判断是否为内网 IP，ZPAY 可能拒收导致 Ubuntu 部署失败"""
+    if not ip or not ip.strip():
+        return True
+    ip = ip.strip()
+    if ip in ("127.0.0.1", "localhost", "::1"):
+        return True
+    if ip.startswith("10.") or ip.startswith("192.168.") or ip.startswith("172.16."):
+        return True
+    return False
+
 
 class ZPayProvider(PaymentProvider):
     """ZPAY 支付（易支付兼容接口）"""
@@ -64,8 +79,10 @@ class ZPayProvider(PaymentProvider):
             "money": total_fee_yuan,
             "return_url": return_url or notify_url,
         }
-        if client_ip:
+        if client_ip and not _is_internal_ip(client_ip):
             params["clientip"] = client_ip
+        elif client_ip and _is_internal_ip(client_ip):
+            logger.warning("ZPAY create_order 跳过内网 client_ip=%s", client_ip)
         if extra and extra.get("param"):
             params["param"] = str(extra["param"])
         params["sign_type"] = "MD5"
@@ -98,6 +115,14 @@ class ZPayProvider(PaymentProvider):
         适用于需要二维码或 H5 跳转的场景
         """
         channel_type = "wxpay" if pay_type.lower() in ("wx", "wechat", "wxpay") else "alipay"
+        effective_ip = (client_ip or "").strip() or "127.0.0.1"
+        # 内网 IP（127.0.0.1/10.x/192.168.x）可能导致 ZPAY 风控拒单，Ubuntu 部署时常见
+        if _is_internal_ip(effective_ip):
+            logger.warning(
+                "ZPAY create_order_api client_ip=%s 为内网地址，已省略 clientip 参数（避免风控拒单）",
+                effective_ip,
+            )
+            effective_ip = ""  # 不传 clientip，由 ZPAY 从连接获取
         params = {
             "pid": self.pid,
             "type": channel_type,
@@ -105,9 +130,10 @@ class ZPayProvider(PaymentProvider):
             "notify_url": notify_url,
             "name": name,
             "money": total_fee_yuan,
-            "clientip": client_ip or "127.0.0.1",
             "device": device,
         }
+        if effective_ip:
+            params["clientip"] = effective_ip
         if return_url:
             params["return_url"] = return_url
         if extra and extra.get("param"):
@@ -117,7 +143,15 @@ class ZPayProvider(PaymentProvider):
 
         try:
             resp = requests.post(self.MAPI_URL, data=params, timeout=15)
+        except requests.exceptions.SSLError as e:
+            logger.error("ZPAY SSL 错误（Ubuntu 可能 CA 证书不全）: %s", e, exc_info=True)
+            return {
+                "status": "error",
+                "provider": self.name,
+                "msg": f"连接 zpayz.cn SSL 失败: {e}。Ubuntu 请安装 ca-certificates: apt install ca-certificates",
+            }
         except requests.RequestException as e:
+            logger.error("ZPAY 连接失败: %s", e, exc_info=True)
             return {
                 "status": "error",
                 "provider": self.name,
@@ -160,10 +194,18 @@ class ZPayProvider(PaymentProvider):
                 "O_id": result.get("O_id"),
                 "trade_no": result.get("trade_no"),
             }
+        err_msg = result.get("msg", resp.text)
+        logger.error(
+            "ZPAY create_order_api 失败 status=%s code=%s msg=%s resp_text=%s",
+            resp.status_code,
+            code,
+            err_msg,
+            resp.text[:500] if resp.text else "",
+        )
         return {
             "status": "error",
             "provider": self.name,
-            "msg": result.get("msg", resp.text),
+            "msg": err_msg,
         }
 
     def verify_notify(self, data: Dict[str, Any]) -> bool: