281 lines
9.9 KiB
Python
281 lines
9.9 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
|
||
from fastapi import Request, HTTPException
|
||
import logging
|
||
from pocketbase import PocketBase
|
||
|
||
app = FastAPI()
|
||
|
||
app.add_middleware(
|
||
CORSMiddleware,
|
||
allow_origins=["*"],
|
||
allow_credentials=True,
|
||
allow_methods=["*"],
|
||
allow_headers=["*"],
|
||
)
|
||
|
||
|
||
# ==================== PocketBase 配置 ====================
|
||
PB_URL = "https://pocketbase.hackrobot.cn"
|
||
PB_ADMIN_EMAIL = "xiaoshuang.eric@gmail.com" # 建议用 admin 账号
|
||
PB_ADMIN_PASSWORD = "Xiao4669805@" # 安全起见，建议用环境变量
|
||
# ========================================================
|
||
|
||
# 初始化 PocketBase 客户端（建议只初始化一次）
|
||
# pb_client = PocketBase(PB_URL)
|
||
|
||
|
||
# 登录 admin（启动时登录一次，token 有效期长）
|
||
# try:
|
||
# admin_data = pb_client.admins.auth_with_password(PB_ADMIN_EMAIL, PB_ADMIN_PASSWORD)
|
||
# print("PocketBase admin 登录成功")
|
||
# except Exception as e:
|
||
# print("PocketBase admin 登录失败:", e)
|
||
|
||
|
||
# 动态获取已登录的 PocketBase 客户端
|
||
def get_pb_client():
|
||
client = PocketBase(PB_URL)
|
||
try:
|
||
client.admins.auth_with_password(PB_ADMIN_EMAIL, PB_ADMIN_PASSWORD)
|
||
print("PocketBase admin 登录成功（动态）")
|
||
except Exception as e:
|
||
print("PocketBase admin 登录失败:", e)
|
||
raise e # 如果登录失败，直接抛异常，避免后续写入失败
|
||
return client
|
||
|
||
|
||
# 配置日志（可选，记录到文件方便排查）
|
||
logging.basicConfig(filename='xorpay_notify.log', level=logging.INFO,
|
||
format='%(asctime)s - %(message)s')
|
||
|
||
# 全局已处理订单（测试用，生产建议用 Redis 或数据库）
|
||
processed_orders = set()
|
||
|
||
# ==================== 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}"
|
||
# 新增：获取前端传来的 user_id
|
||
user_id = item.get('user_id', 'unknown')
|
||
pay_type = item.get('type', 'unknown').lower() # ← 新增 .lower()，统一小写
|
||
# 获取 type
|
||
# pay_type = item.get('type', 'unknown')
|
||
|
||
random_suffix = ''.join(random.choices(string.hexdigits.lower(), k=8))
|
||
# order_id = f"order_{int(time.time())}_{random_suffix}"
|
||
timestamp = int(time.time())
|
||
# order_id = f"{user_id}_order_{timestamp}_{random_suffix}"
|
||
# 推荐格式：user_{user_id}_{type}_order_{timestamp}_{random}
|
||
order_id = f"{user_id}_{pay_type}_order_{timestamp}_{random_suffix}"
|
||
|
||
type_map = {
|
||
"book": "购买书籍",
|
||
"meetup": "活动报名",
|
||
"video": "视频解锁",
|
||
"vip": "VIP会员充值",
|
||
}
|
||
name = type_map.get(item.get('type'), "商品支付")
|
||
|
||
# notify_url = "https://www.hackrobot.cn/xorpay_notify" # 建议使用完整路径，避免冲突
|
||
notify_url = "https://api.hackrobot.cnp/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}"
|
||
}
|
||
|
||
def create_memos_user(user_id: str, password: str = "123456"):
|
||
"""
|
||
为指定 user_id 创建 memos 用户
|
||
支持兼容旧格式（带下划线），自动清理
|
||
"""
|
||
try:
|
||
# 新增：清理用户名，确保无下划线，以字母开头
|
||
if user_id.startswith("user_"):
|
||
cleaned_username = "user" + user_id.split("user_", 1)[1] # user_176... → user176...
|
||
elif "_" in user_id:
|
||
cleaned_username = user_id.replace("_", "") # 万一有其他下划线也去掉
|
||
else:
|
||
cleaned_username = user_id # 已经是干净的，直接用
|
||
|
||
username = cleaned_username
|
||
print(f"正在为用户创建 memos 账号: {username}")
|
||
|
||
memos_data = {
|
||
"username": username,
|
||
"password": password,
|
||
"role": "USER",
|
||
"state": "NORMAL"
|
||
}
|
||
memos_headers = {
|
||
"Content-Type": "application/json",
|
||
"Authorization": f"Bearer {MEMOS_TOKEN}"
|
||
}
|
||
|
||
response = requests.post(MEMOS_API, json=memos_data, headers=memos_headers, timeout=10)
|
||
|
||
if response.status_code in (200, 201):
|
||
result = response.json()
|
||
print(f"memos 用户创建成功: {username}")
|
||
logging.info(f"memos 用户创建成功: {username}")
|
||
return {"success": True, "data": result}
|
||
else:
|
||
error_text = response.text
|
||
print(f"memos 创建失败: {response.status_code} {error_text}")
|
||
logging.error(f"memos 创建失败 - username: {username}, status: {response.status_code}, response: {error_text}")
|
||
return {"success": False, "error": error_text}
|
||
|
||
except Exception as e:
|
||
print(f"创建 memos 用户异常: {e}")
|
||
logging.error(f"创建 memos 用户异常 - username: {user_id}, error: {e}")
|
||
return {"success": False, "error": str(e)}
|
||
|
||
|
||
# 异步通知回调
|
||
@app.post("/xorpay_notify")
|
||
async def xorpay_notify(request: Request):
|
||
form_data = await request.form()
|
||
data = {key: value for key, value in form_data.items()}
|
||
|
||
logging.info(f"XorPay 异步通知原始数据: {data}")
|
||
print("XorPay 异步通知:", data)
|
||
|
||
order_id = data.get("order_id", "")
|
||
|
||
# 幂等性检查
|
||
if order_id in processed_orders:
|
||
print(f"订单 {order_id} 已处理过，跳过")
|
||
return "ok"
|
||
|
||
# 1. 验签
|
||
if not verify_xorpay_notify(data):
|
||
print("验签失败")
|
||
logging.error(f"验签失败: {data}")
|
||
raise HTTPException(status_code=400, detail="sign error")
|
||
|
||
# 2. 正确解析 user_id 和 pay_type
|
||
if "_order_" not in order_id:
|
||
raise HTTPException(status_code=400, detail="invalid order_id format")
|
||
|
||
prefix = order_id.split("_order_")[0]
|
||
|
||
if "_" in prefix:
|
||
parts = prefix.rsplit("_", 1) # 从右边拆分一次
|
||
user_id = parts[0]
|
||
pay_type = parts[1].lower() # ← 强制小写
|
||
else:
|
||
user_id = prefix
|
||
pay_type = "unknown"
|
||
|
||
print(f"支付成功，用户ID: {user_id}，类型: {pay_type}，订单: {order_id}")
|
||
|
||
# 在解析完 pay_type 后，只有 vip 才调用
|
||
if pay_type == "vip":
|
||
create_memos_user(user_id) # 直接传 user_id，密码默认 123456
|
||
|
||
# ------------------- 写入 PocketBase -------------------
|
||
try:
|
||
# 动态获取已登录的客户端
|
||
pb_client = get_pb_client()
|
||
|
||
valid_types = ["vip", "book", "meetup", "video"] # 根据你实际允许的类型填
|
||
if pay_type not in valid_types:
|
||
print(f"无效的 pay_type: {pay_type}，强制设为 vip 或拒绝写入")
|
||
pay_type = "vip" # 或者直接 return "ok" 跳过写入
|
||
|
||
pb_data = {
|
||
"user_id": user_id,
|
||
"type": pay_type,
|
||
"order_id": order_id,
|
||
"amount": int(float(data.get("pay_price", 0)) * 100),
|
||
}
|
||
|
||
record = pb_client.collection("payments").create(pb_data)
|
||
print(f"PocketBase 记录创建成功: {record.id}")
|
||
logging.info(f"支付记录写入 PocketBase 成功 - user_id: {user_id}, type: {pay_type}, order_id: {order_id}")
|
||
|
||
processed_orders.add(order_id)
|
||
|
||
except Exception as e:
|
||
print(f"PocketBase 写入失败: {e}")
|
||
logging.error(f"PocketBase 写入失败 - order_id: {order_id}, error: {e}")
|
||
processed_orders.add(order_id) # 防止重试风暴
|
||
|
||
return "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 |