154 lines
5.2 KiB
Python
154 lines
5.2 KiB
Python
from fastapi import FastAPI
|
||
from fastapi.middleware.cors import CORSMiddleware
|
||
from pydantic import BaseModel
|
||
import hashlib
|
||
import time
|
||
import random
|
||
import string
|
||
import requests
|
||
|
||
app = FastAPI()
|
||
|
||
app.add_middleware(
|
||
CORSMiddleware,
|
||
allow_origins=["*"],
|
||
allow_credentials=True,
|
||
allow_methods=["*"],
|
||
allow_headers=["*"],
|
||
)
|
||
|
||
# ==================== XorPay 配置 ====================
|
||
AID = "8220" # 你的 aid
|
||
SECRET = "afcacd99570945f88de62624aaa3578e" # 你的 app secret
|
||
# ====================================================
|
||
|
||
# memos 配置（从原代码恢复）
|
||
MEMOS_API = "https://qun.hackrobot.cn/api/v1/users"
|
||
MEMOS_TOKEN = "eyJhbGciOiJIUzI1NiIsImtpZCI6InYxIiwidHlwIjoiSldUIn0.eyJuYW1lIjoiaGFja3JvYm90IiwiaXNzIjoibWVtb3MiLCJzdWIiOiIxIiwiYXVkIjpbInVzZXIuYWNjZXNzLXRva2VuIl0sImlhdCI6MTc1NzQwNTE0OX0.Idn7kBlxE-CoSOXwWuZ1tHGIRKHAIeDyXSafGS5OHsg"
|
||
|
||
# 官方推荐签名方式（严格无分隔符拼接）
|
||
def xorpay_sign(name: str, pay_type: str, price: str, order_id: str, notify_url: str) -> str:
|
||
raw = name + pay_type + price + order_id + notify_url + SECRET
|
||
return hashlib.md5(raw.encode('utf-8')).hexdigest().lower()
|
||
|
||
# 回调验签函数
|
||
def verify_xorpay_notify(data: dict) -> bool:
|
||
received_sign = data.get("sign", "")
|
||
aoid = data.get("aoid", "")
|
||
order_id = data.get("order_id", "")
|
||
pay_price = data.get("pay_price", "")
|
||
pay_time = data.get("pay_time", "")
|
||
raw = aoid + order_id + pay_price + pay_time + SECRET
|
||
calculated_sign = hashlib.md5(raw.encode('utf-8')).hexdigest().lower()
|
||
return calculated_sign == received_sign
|
||
|
||
@app.post("/payh5")
|
||
async def payh5(item: dict):
|
||
print('payh5 item:', item)
|
||
|
||
total_fee_yuan = f"{item['total_fee'] / 100:.2f}"
|
||
|
||
random_suffix = ''.join(random.choices(string.hexdigits.lower(), k=8))
|
||
order_id = f"order_{int(time.time())}_{random_suffix}"
|
||
|
||
type_map = {
|
||
"book": "购买书籍",
|
||
"meetup": "活动报名",
|
||
"video": "视频解锁",
|
||
"vip": "VIP会员充值",
|
||
}
|
||
name = type_map.get(item.get('type'), "商品支付")
|
||
|
||
notify_url = "https://www.hackrobot.cn/xorpay_notify" # 建议使用完整路径，避免冲突
|
||
|
||
params = {
|
||
"name": name,
|
||
"pay_type": "jsapi",
|
||
"price": total_fee_yuan,
|
||
"order_id": order_id,
|
||
"notify_url": notify_url,
|
||
}
|
||
|
||
params["sign"] = xorpay_sign(
|
||
params["name"],
|
||
params["pay_type"],
|
||
params["price"],
|
||
params["order_id"],
|
||
params["notify_url"]
|
||
)
|
||
|
||
print("生成的参数:", params)
|
||
|
||
# 返回参数字典，前端用 form POST 到 xorpay 收银台
|
||
return {
|
||
"status": "ok",
|
||
"xorpay_params": params,
|
||
"xorpay_url": f"https://xorpay.com/api/cashier/{AID}"
|
||
}
|
||
|
||
# 异步通知回调：在支付成功后自动创建 memos 用户
|
||
@app.post("/xorpay_notify")
|
||
async def xorpay_notify(request: dict):
|
||
print("XorPay 异步通知:", request)
|
||
|
||
# 1. 验签（强烈推荐正式环境开启）
|
||
if not verify_xorpay_notify(request):
|
||
print("验签失败，可能为伪造通知")
|
||
return {"status": "fail", "msg": "sign error"}
|
||
|
||
# 2. 检查支付状态（XorPay 成功通知时 status=1）
|
||
if request.get("status") != "1":
|
||
print("支付未成功")
|
||
return {"status": "ok"} # 仍返回 ok，避免重复通知
|
||
|
||
# # 3. 这里可以根据你的业务逻辑，从 order_id 或其他参数提取用户名/密码等信息
|
||
# # 假设前端在支付时传了自定义参数（如 order_id 中编码了用户名），或使用 return_url 携带信息
|
||
# # 示例：简单生成一个用户（类似原 /create_user）
|
||
# timenew = str(int(time.time()))
|
||
# username = f"user_{timenew}"
|
||
# password = "123456" # 可以随机生成或从业务中获取
|
||
|
||
# memos_data = {
|
||
# "username": username,
|
||
# "password": password,
|
||
# "role": "USER",
|
||
# "state": "NORMAL"
|
||
# }
|
||
# memos_headers = {
|
||
# "Content-Type": "application/json",
|
||
# "Authorization": f"Bearer {MEMOS_TOKEN}"
|
||
# }
|
||
|
||
# try:
|
||
# memos_resp = requests.post(MEMOS_API, json=memos_data, headers=memos_headers)
|
||
# memos_result = memos_resp.json()
|
||
# print('memos 创建结果:', memos_result)
|
||
# # 这里可以记录订单与 memos 用户的关联（数据库等）
|
||
# except Exception as e:
|
||
# print("创建 memos 用户失败:", e)
|
||
|
||
# 返回 ok 表示通知成功接收
|
||
return {"status": "ok"}
|
||
|
||
# 保留原来的手动创建用户接口（如果还需要）
|
||
class CreateUserRequest(BaseModel):
|
||
username: str = None
|
||
password: str = None
|
||
|
||
@app.post("/create_user")
|
||
async def create_user(item: CreateUserRequest):
|
||
timenew = str(int(time.time()))
|
||
memos_data = {
|
||
"username": item.username or f"user{timenew}",
|
||
"password": item.password or "123456",
|
||
"role": "USER",
|
||
"state": "NORMAL"
|
||
}
|
||
memos_headers = {
|
||
"Content-Type": "application/json",
|
||
"Authorization": f"Bearer {MEMOS_TOKEN}"
|
||
}
|
||
memos_resp = requests.post(MEMOS_API, json=memos_data, headers=memos_headers)
|
||
memos_result = memos_resp.json()
|
||
print('memos_result-', memos_result)
|
||
return memos_result |