ff7b721acb
Port meetups, discussions, gigs, dating/chat, VIP pay (ZPay/XorPay), MiroTalk live, digital academy, and ebook reader. Add persistent community_store, git-first deploy docs, and env template for production secrets. Co-authored-by: Cursor <cursoragent@cursor.com>
97 lines
3.0 KiB
Python
97 lines
3.0 KiB
Python
"""Payment helpers — ZPay/XorPay from NomadCNA + DEV_AUTO_PAY toggle."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
from urllib.parse import urlencode
|
|
|
|
from app.services.payment_providers import XorPayProvider, ZPayProvider, provider_for
|
|
|
|
DEV_AUTO_PAY = os.getenv("DEV_AUTO_PAY", "false").lower() in ("1", "true", "yes")
|
|
PAYMENT_JOIN_AMOUNT = int(os.getenv("PAYMENT_JOIN_AMOUNT", "10000")) # fen (100 yuan)
|
|
PAYMENT_DEFAULT_AMOUNT = int(os.getenv("PAYMENT_DEFAULT_AMOUNT", "10000"))
|
|
SITE_BASE = os.getenv("SITE_BASE_URL", "https://nomadweb.nomadro.com").rstrip("/")
|
|
PAY_NOTIFY_URL = os.getenv("PAY_NOTIFY_URL", f"{SITE_BASE}/api/v1/pay/zpay_notify")
|
|
|
|
|
|
def amount_yuan(amount_fen: int) -> str:
|
|
return f"{amount_fen / 100:.2f}"
|
|
|
|
|
|
def notify_url_for(provider_name: str) -> str:
|
|
if provider_name == "xorpay":
|
|
return os.getenv("XORPAY_NOTIFY_URL", f"{SITE_BASE}/api/v1/pay/xorpay_notify")
|
|
return os.getenv("ZPAY_NOTIFY_URL", PAY_NOTIFY_URL)
|
|
|
|
|
|
async def create_pay_redirect(
|
|
order_id: str,
|
|
amount: int,
|
|
return_url: str,
|
|
provider: str = "",
|
|
*,
|
|
channel: str = "alipay",
|
|
device: str = "pc",
|
|
user_agent: str = "",
|
|
client_ip: str = "",
|
|
) -> str:
|
|
if DEV_AUTO_PAY:
|
|
sep = "&" if "?" in return_url else "?"
|
|
return f"{return_url}{sep}order_id={order_id}&status=paid"
|
|
|
|
pay_provider = provider_for(device, provider or os.getenv("PAYMENT_PROVIDER", "zpay"), user_agent)
|
|
yuan = amount_yuan(amount)
|
|
name = "nomadro会员"
|
|
notify = notify_url_for(pay_provider.name)
|
|
|
|
if isinstance(pay_provider, XorPayProvider):
|
|
created = pay_provider.create_order(
|
|
order_id=order_id, name=name, amount_yuan=yuan, notify_url=notify,
|
|
)
|
|
return f"{created['pay_url']}?{urlencode(created['params'])}"
|
|
|
|
api_result = await pay_provider.create_order_api(
|
|
order_id=order_id,
|
|
name=name,
|
|
amount_yuan=yuan,
|
|
channel=channel,
|
|
notify_url=notify,
|
|
return_url=return_url,
|
|
client_ip=client_ip,
|
|
device=device,
|
|
)
|
|
if api_result.get("status") == "ok":
|
|
pay_url = api_result.get("payurl2") if device == "h5" and api_result.get("payurl2") else api_result.get("payurl")
|
|
if pay_url:
|
|
return str(pay_url)
|
|
|
|
created = pay_provider.create_order(
|
|
order_id=order_id,
|
|
name=name,
|
|
amount_yuan=yuan,
|
|
channel=channel,
|
|
notify_url=notify,
|
|
return_url=return_url,
|
|
client_ip=client_ip,
|
|
)
|
|
return f"{created['pay_url']}?{urlencode(created['params'])}"
|
|
|
|
|
|
async def query_order_paid(order_id: str) -> bool:
|
|
zpay, xorpay = await ZPayProvider().query_order_status(order_id), await XorPayProvider().query_order_status(order_id)
|
|
return bool(zpay.get("paid") or xorpay.get("paid"))
|
|
|
|
|
|
def verify_notify(provider: str, data: dict) -> bool:
|
|
if DEV_AUTO_PAY:
|
|
return True
|
|
if provider == "xorpay":
|
|
return XorPayProvider().verify_notify(data)
|
|
return ZPayProvider().verify_notify(data)
|
|
|
|
|
|
def parse_notify_order_id(provider: str, data: dict) -> str:
|
|
if provider == "xorpay":
|
|
return str(data.get("order_id") or "")
|
|
return str(data.get("out_trade_no") or data.get("order_id") or "")
|