'c1'

app/payment/__init__.py

@@ -5,7 +5,13 @@
 from .base import PaymentProvider
 from .xorpay import XorPayProvider
 from .zpay import ZPayProvider
-from .factory import get_payment_provider, get_payment_provider_by_name, reset_provider
+from .factory import (
+    get_payment_provider,
+    get_payment_provider_by_name,
+    reset_provider,
+    resolve_provider,
+    resolve_channel,
+)
 
 __all__ = [
     "PaymentProvider",
@@ -14,4 +20,6 @@ __all__ = [
     "get_payment_provider",
     "get_payment_provider_by_name",
     "reset_provider",
+    "resolve_provider",
+    "resolve_channel",
 ]

app/payment/factory.py

@@ -49,3 +49,31 @@ def reset_provider():
     """重置 Provider（用于测试或切换渠道）"""
     global _provider_instance
     _provider_instance = None
+
+
+def resolve_provider(
+    device: str | None = None,
+    req_provider: str | None = None,
+    user_agent: str | None = None,
+) -> PaymentProvider:
+    """
+    根据设备解析支付渠道：
+    - 微信内(device=wechat 或 UA 含 MicroMessenger) → xorpay
+    - PC/手机浏览器 → zpay
+    显式传入的 provider 优先。
+    """
+    req = (req_provider or "").strip().lower()
+    if req in ("xorpay", "zpay"):
+        return get_payment_provider_by_name(req)
+    dev = (device or "pc").lower()
+    if dev == "wechat":
+        return get_payment_provider_by_name("xorpay")
+    ua = (user_agent or "").lower()
+    if "micromessenger" in ua:
+        return get_payment_provider_by_name("xorpay")
+    return get_payment_provider_by_name("zpay")
+
+
+def resolve_channel(item_channel: str | None) -> str:
+    """支付方式固定为微信支付，支付宝暂不提供"""
+    return "wxpay"

app/payment/xorpay.py

@@ -5,6 +5,8 @@ XorPay 支付渠道实现
 import hashlib
 from typing import Any, Dict, Optional
 
+import requests
+
 from .base import PaymentProvider
 
 
@@ -24,6 +26,10 @@ class XorPayProvider(PaymentProvider):
         raw = name + pay_type + price + order_id + notify_url + self.secret
         return hashlib.md5(raw.encode("utf-8")).hexdigest().lower()
 
+    def _query2_sign(self, order_id: str) -> str:
+        raw = order_id + self.secret
+        return hashlib.md5(raw.encode("utf-8")).hexdigest().lower()
+
     def create_order(
         self,
         *,
@@ -80,3 +86,23 @@ class XorPayProvider(PaymentProvider):
 
     def success_response(self) -> str:
         return "ok"
+
+    def query_order_status(self, order_id: str, timeout: int = 10) -> Dict[str, Any]:
+        url = f"https://xorpay.com/api/query2/{self.aid}"
+        params = {
+            "order_id": order_id,
+            "sign": self._query2_sign(order_id),
+        }
+        try:
+            response = requests.get(url, params=params, timeout=timeout)
+            response.raise_for_status()
+            data = response.json() or {}
+            status = str(data.get("status", "")).strip().lower()
+            return {
+                "paid": status in ("payed", "success"),
+                "status": status,
+                "pay_price": data.get("pay_price") or data.get("price") or "",
+                "raw": data,
+            }
+        except Exception as error:
+            return {"paid": False, "error": str(error)}