eric/gitlab-instance-0a899031_payjsapi
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

's'

Browse Source
This commit is contained in:
eric
2026-03-12 19:50:50 -05:00
parent ae575f1828
commit 90378502cf
18 changed files with 438 additions and 22 deletions
Download Patch File Download Diff File Expand all files Collapse all files

16
.env.example
View File

@@ -1,20 +1,12 @@
# ========== 本地调试 ==========
# PORT: 默认 8007，与前端 PAYMENT_API_URL 一致
# BASE_URL 需为 ZPAY 可访问的回调地址。本地调试时 ZPAY 无法回调 localhost/内网，
# 可用 ngrok 等隧道：BASE_URL=https://xxx.ngrok.io
# 或直接使用生产地址测试回调（需 payjsapi 已部署到公网）
# nomadvip/digital/cnomadcna/nomadlms 的 zpay_notify 均依赖此地址，未正确配置会导致 PocketBase/Memos 不更新
# 本地调试：ngrok 暴露 8700 后设置 BASE_URL=https://xxx.ngrok.io
# 线上部署：BASE_URL=https://api.hackrobot.cn
# ========== 生产部署（Ubuntu 等）==========
# 线上部署必须正确配置，否则会显示「支付失败」：
# 1. BASE_URL 必须为 payjsapi 公网地址，ZPAY 需能访问此地址做异步回调
# 错误示例：http://127.0.0.1:8700、http://localhost（ZPAY 无法访问）
# 正确示例：https://api.hackrobot.cn（域名需解析到本机并配置 nginx 反向代理）
# 2. 服务器需能访问 zpayz.cn（检查防火墙、DNS），可访问 /health?check_zpay=1 自检
PORT=8007
BASE_URL=https://api.hackrobot.cn
# 支付渠道：zpay（已替代 xorpay）| xorpay
# 支付渠道：zpay | xorpay
PAYMENT_PROVIDER=zpay
# XorPay
@@ -25,7 +17,7 @@ XORPAY_SECRET=your_secret
ZPAY_PID=2025121809351743
ZPAY_KEY=tpEi7wWIWI2kXiYVTpIG6j7it0mjVW89
# PocketBase
# PocketBase（check-user、ensure-user、支付回调必需）
PB_URL=https://pocketbase.hackrobot.cn
PB_ADMIN_EMAIL=your@email.com
PB_ADMIN_PASSWORD=your_password

BIN
app/__pycache__/config.cpython-312.pyc
View File

Binary file not shown.

BIN
app/__pycache__/main.cpython-312.pyc
View File

Binary file not shown.

20
app/config.py
View File

@@ -3,8 +3,16 @@
支付渠道可通过 PAYMENT_PROVIDER 切换：xorpay | zpay
"""
import os
from pathlib import Path
from typing import Literal
from dotenv import load_dotenv
# 强制从 payjsapi 项目根目录加载 .env、.env.local，避免 cwd 不同导致加载失败
_root = Path(__file__).resolve().parent.parent
load_dotenv(_root / ".env")
load_dotenv(_root / ".env.local", override=True)
PaymentChannel = Literal["xorpay", "zpay"]
# 当前使用的支付渠道（zpay 替代 xorpay，payjs2 接口业务保留）
@@ -19,10 +27,14 @@ XORPAY_CASHIER_URL = os.getenv("XORPAY_CASHIER_URL", "https://xorpay.com/api/cas
ZPAY_PID = os.getenv("ZPAY_PID", "2025121809351743")
ZPAY_KEY = os.getenv("ZPAY_KEY", "tpEi7wWIWI2kXiYVTpIG6j7it0mjVW89")
# PocketBase
PB_URL = os.getenv("PB_URL", "https://pocketbase.hackrobot.cn")
PB_ADMIN_EMAIL = os.getenv("PB_ADMIN_EMAIL", "xiaoshuang.eric@gmail.com")
PB_ADMIN_PASSWORD = os.getenv("PB_ADMIN_PASSWORD", "Xiao4669805@")
# PocketBase（支持 PB_ADMIN_* 或 POCKETBASE_*，空值时用默认）
PB_URL = os.getenv("PB_URL") or os.getenv("POCKETBASE_URL") or "https://pocketbase.hackrobot.cn"
PB_ADMIN_EMAIL = (
os.getenv("PB_ADMIN_EMAIL") or os.getenv("POCKETBASE_EMAIL") or "xiaoshuang.eric@gmail.com"
)
PB_ADMIN_PASSWORD = (
os.getenv("PB_ADMIN_PASSWORD") or os.getenv("POCKETBASE_PASSWORD") or "Xiao4669805@"
)
# Memos
MEMOS_API = os.getenv("MEMOS_API", "https://qun.hackrobot.cn/api/v1/users")

2
app/main.py
View File

@@ -28,6 +28,7 @@ from .routers import (
nomadlms_router,
legacy_router,
common_router,
meetup_router,
)
app = FastAPI(
@@ -99,6 +100,7 @@ app.include_router(cnomadcna_router)
app.include_router(nomadlms_router)
app.include_router(legacy_router)
app.include_router(common_router)
app.include_router(meetup_router)
@app.get("/")

2
app/routers/__init__.py
View File

@@ -12,6 +12,7 @@ from .cnomadcna import router as cnomadcna_router
from .nomadlms import router as nomadlms_router
from .legacy import router as legacy_router
from .common import router as common_router
from .meetup import router as meetup_router
__all__ = [
"nomadvip_router",
@@ -20,4 +21,5 @@ __all__ = [
"nomadlms_router",
"legacy_router",
"common_router",
"meetup_router",
]

BIN
app/routers/__pycache__/__init__.cpython-312.pyc
View File

Binary file not shown.

BIN
app/routers/__pycache__/common.cpython-312.pyc
View File

Binary file not shown.

2
app/routers/common.py
View File

@@ -59,7 +59,7 @@ async def create_user(item: CreateUserRequest):
return {"error": "Memos 未配置"}
timenew = str(int(time.time()))
username = item.username or f"user{timenew}"
password = item.password or "123456"
password = item.password or "12345678"
result = sdk.create_user_by_username(username=username, password=password)
print("memos_result-", result)
return result

315
app/routers/meetup.py Normal file
View File

@@ -0,0 +1,315 @@
"""
meetup 加入流程：check-user、ensure-user、complete-order
供 cnomadcna 等前端调用，当 /api/* 代理到 payjsapi 时使用
"""
import logging
import time
import requests
from fastapi import APIRouter, HTTPException
from pydantic import BaseModel
from ..config import PB_URL, PB_ADMIN_EMAIL, PB_ADMIN_PASSWORD
router = APIRouter(prefix="/api/meetup", tags=["meetup"])
DEFAULT_PASSWORD = "12345678" # PocketBase 默认要求至少 8 位
def _get_admin_token() -> tuple[str | None, str | None]:
"""返回 (token, error_detail)。token 为 None 时 error_detail 为失败原因"""
if not PB_ADMIN_EMAIL or not PB_ADMIN_PASSWORD:
return None, "PB_ADMIN_EMAIL 或 PB_ADMIN_PASSWORD 未配置"
base = (PB_URL or "").rstrip("/")
# PocketBase v0.23+ 使用 _superusers，旧版用 admins
for path in ["/api/collections/_superusers/auth-with-password", "/api/admins/auth-with-password"]:
try:
r = requests.post(
f"{base}{path}",
json={"identity": PB_ADMIN_EMAIL, "password": PB_ADMIN_PASSWORD},
timeout=10,
)
if r.status_code == 200:
data = r.json()
return data.get("token"), None
if r.status_code == 404:
continue # 尝试下一个路径
try:
msg = r.json().get("message", r.text[:100])
except Exception:
msg = r.text[:100]
logging.warning(f"PocketBase admin auth failed: status={r.status_code}, msg={msg}")
return None, f"PocketBase 管理员认证失败: {msg}"
except requests.exceptions.ConnectionError as e:
logging.error(f"PocketBase 连接失败: {e}")
return None, f"无法连接 {PB_URL}，请检查网络或 PB_URL"
except Exception as e:
logging.warning(f"PocketBase auth try {path}: {e}")
return None, "PocketBase 管理员认证失败: 请确认 PB_URL 及管理员账号密码正确"
class CheckUserRequest(BaseModel):
email: str
class EnsureUserRequest(BaseModel):
email: str
password: str | None = None
def _check_site_vip(user_id: str, site_id: str = "meetup") -> bool:
"""检查用户是否有有效 site_vip（未过期）"""
token, _ = _get_admin_token()
if not token:
return False
try:
from datetime import datetime
r = requests.get(
f"{PB_URL}/api/collections/site_vip/records",
params={
"filter": f'user_id = "{user_id}" && site_id = "{site_id}"',
"perPage": 1,
"sort": "-expires_at",
},
headers={"Authorization": f"Bearer {token}"},
timeout=10,
)
if r.status_code != 200:
return False
data = r.json()
items = data.get("items") or []
if not items:
return False
rec = items[0]
exp = rec.get("expires_at")
if not exp:
return True
try:
from datetime import datetime as _dt, timezone
exp_dt = _dt.fromisoformat(exp.replace("Z", "+00:00"))
now = _dt.now(timezone.utc)
if exp_dt.tzinfo is None:
exp_dt = exp_dt.replace(tzinfo=timezone.utc)
return exp_dt > now
except Exception:
return True
except Exception:
return False
@router.post("/check-user")
async def check_user(item: CheckUserRequest):
"""检查邮箱是否已在 users 表存在，以及 VIP 状态（新/老用户判断）"""
email = (item.email or "").strip()
if not email:
raise HTTPException(status_code=400, detail="请输入邮箱")
token, err = _get_admin_token()
if not token:
raise HTTPException(status_code=500, detail=f"服务配置错误：{err}")
try:
filter_val = f'email="{email}"'
r = requests.get(
f"{PB_URL}/api/collections/users/records",
params={"filter": filter_val, "perPage": 1},
headers={"Authorization": f"Bearer {token}"},
timeout=10,
)
if r.status_code != 200:
raise HTTPException(status_code=500, detail="查询失败")
data = r.json()
items = data.get("items") or []
exists = len(items) > 0
user_id = items[0].get("id") if items else None
vip = _check_site_vip(user_id, "meetup") if user_id else False
return {"ok": True, "exists": exists, "vip": vip, "user_id": user_id}
except HTTPException:
raise
except Exception as e:
logging.error(f"check-user error: {e}")
raise HTTPException(status_code=500, detail="操作失败")
@router.post("/ensure-user")
async def ensure_user(item: EnsureUserRequest):
"""确保用户存在：老用户验证密码，新用户用默认密码 12345678 创建"""
email = (item.email or "").strip()
if not email:
raise HTTPException(status_code=400, detail="请输入邮箱")
password = (item.password or "").strip() or DEFAULT_PASSWORD
try:
# 尝试用密码登录
login_r = requests.post(
f"{PB_URL}/api/collections/users/auth-with-password",
json={"identity": email, "password": password},
timeout=10,
)
if login_r.status_code == 200:
data = login_r.json()
return {
"ok": True,
"user_id": data.get("record", {}).get("id"),
"token": data.get("token"),
"record": data.get("record"),
"is_new": False,
}
try:
err_data = login_r.json()
except Exception:
err_data = {}
is_not_found = login_r.status_code == 400 and (
"Invalid" in str(err_data.get("message", ""))
or "identity" in str(err_data.get("message", "")).lower()
)
if not is_not_found and item.password:
raise HTTPException(status_code=400, detail="密码错误")
# 新用户：创建
token, err = _get_admin_token()
if not token:
raise HTTPException(status_code=500, detail=f"服务配置错误：{err}")
create_r = requests.post(
f"{PB_URL}/api/collections/users/records",
json={
"email": email,
"password": DEFAULT_PASSWORD,
"passwordConfirm": DEFAULT_PASSWORD,
},
headers={
"Content-Type": "application/json",
"Authorization": f"Bearer {token}",
},
timeout=10,
)
if create_r.status_code != 200:
try:
create_err = create_r.json()
except Exception:
create_err = {}
data = create_err.get("data", {})
if (
create_r.status_code == 400
and "already" in str(data.get("email", {}).get("message", "")).lower()
):
raise HTTPException(status_code=400, detail="该邮箱已注册，请输入密码登录")
msg = create_err.get("message", "创建账号失败")
logging.warning(f"ensure-user create failed: status={create_r.status_code}, msg={msg}, data={data}")
raise HTTPException(status_code=500, detail=msg)
# 登录新用户
final_r = requests.post(
f"{PB_URL}/api/collections/users/auth-with-password",
json={"identity": email, "password": DEFAULT_PASSWORD},
timeout=10,
)
if final_r.status_code != 200:
raise HTTPException(status_code=500, detail="登录失败")
auth_data = final_r.json()
return {
"ok": True,
"user_id": auth_data.get("record", {}).get("id"),
"token": auth_data.get("token"),
"record": auth_data.get("record"),
"is_new": True,
}
except HTTPException:
raise
except Exception as e:
logging.error(f"ensure-user error: {e}")
raise HTTPException(status_code=500, detail="操作失败")
def _decode_pending_email(user_id: str) -> str | None:
if not user_id.startswith("pending_"):
return None
try:
return bytes.fromhex(user_id[8:]).decode("utf-8")
except Exception:
return None
def _query_zpay_paid(order_id: str) -> bool:
"""查询 ZPAY 订单是否已支付"""
try:
from ..payment import get_payment_provider
from ..payment.zpay import ZPayProvider
provider = get_payment_provider()
if not isinstance(provider, ZPayProvider):
return False
result = provider.query_order_status(order_id, timeout=10)
return bool(result.get("paid"))
except Exception as e:
logging.warning(f"complete-order zpay query failed: {e}")
return False
@router.post("/complete-order")
async def complete_order(item: dict):
"""
支付成功后完成订单：验证订单、为新用户同步 session。
支持 pending_ 订单（新用户）和已存在用户订单（user_id 为 PB id）。
异步通知可能晚于用户回跳，故在 site_vip 未找到时轮询重试（ZPAY 已确认支付时）。
"""
order_id = (item.get("order_id") or "").strip()
if not order_id:
raise HTTPException(status_code=400, detail="缺少 order_id")
base = (PB_URL or "").rstrip("/")
token, err = _get_admin_token()
if not token:
raise HTTPException(status_code=500, detail=f"服务配置错误：{err}")
from ..services.payment import parse_order_id
user_id, pay_type = parse_order_id(order_id)
if not user_id or pay_type != "meetup":
raise HTTPException(status_code=400, detail="订单格式无效")
# 验证订单：site_vip 中需存在该 order_id（异步通知可能晚于用户回跳，ZPAY 已支付时轮询重试）
max_retries = 4
for attempt in range(max_retries):
r = requests.get(
f"{base}/api/collections/site_vip/records",
params={"filter": f'order_id = "{order_id}"', "perPage": 1},
headers={"Authorization": f"Bearer {token}"},
timeout=10,
)
total = (r.json().get("totalItems") or 0) if r.status_code == 200 else 0
if total > 0:
break
if attempt < max_retries - 1 and _query_zpay_paid(order_id):
time.sleep(2)
continue
resp_preview = r.text[:200] if r.ok else f"status={r.status_code}"
logging.warning(f"complete-order 未找到订单: order_id={order_id}, attempt={attempt + 1}, pb_resp={resp_preview}")
raise HTTPException(status_code=400, detail="订单不存在或未支付成功")
# pending_ 新用户：用邮箱+默认密码登录返回 token
if user_id.startswith("pending_"):
email = _decode_pending_email(user_id)
if not email:
raise HTTPException(status_code=400, detail="订单格式无效")
login_r = requests.post(
f"{base}/api/collections/users/auth-with-password",
json={"identity": email, "password": DEFAULT_PASSWORD},
timeout=10,
)
if login_r.status_code != 200:
raise HTTPException(status_code=500, detail="登录失败，请稍后重试")
auth_data = login_r.json()
return {
"ok": True,
"token": auth_data.get("token"),
"record": auth_data.get("record"),
"is_new": True,
}
# 已存在用户（PB user_id）：订单已验证，返回 ok，前端已有 session 则无需 token
return {"ok": True, "token": None, "record": None, "is_new": False}

BIN
app/sdk/__pycache__/memos.cpython-312.pyc
View File

Binary file not shown.

4
app/sdk/memos.py
View File

@@ -24,7 +24,7 @@ class MemosSDK:
def create_user(
self,
user_id: str,
password: str = "123456",
password: str = "12345678",
) -> dict:
"""
创建 Memos 用户
@@ -72,7 +72,7 @@ class MemosSDK:
def create_user_by_username(
self,
username: str,
password: str = "123456",
password: str = "12345678",
) -> dict:
"""按用户名直接创建（用于 /create_user 接口）"""
if not self.enabled:

BIN
app/services/__pycache__/memos.cpython-312.pyc
View File

Binary file not shown.

BIN
app/services/__pycache__/payment.cpython-312.pyc
View File

Binary file not shown.

2
app/services/memos.py
View File

@@ -16,6 +16,6 @@ def _get_memos_sdk() -> MemosSDK:
return _memos_sdk
def create_memos_user(user_id: str, password: str = "123456"):
def create_memos_user(user_id: str, password: str = "12345678"):
"""创建 Memos 用户"""
return _get_memos_sdk().create_user(user_id=user_id, password=password)

89
app/services/payment.py
View File

@@ -1,11 +1,15 @@
"""支付成功后的统一业务处理（基于 SDK）"""
import logging
import requests
from ..config import PB_URL, PB_ADMIN_EMAIL, PB_ADMIN_PASSWORD
from .pb import get_pb_client
from .memos import create_memos_user
# 已处理订单（防重复）
processed_orders: set[str] = set()
DEFAULT_PASSWORD = "12345678"
def parse_order_id(order_id: str) -> tuple[str, str]:
@@ -19,6 +23,77 @@ def parse_order_id(order_id: str) -> tuple[str, str]:
return prefix, "unknown"
def _decode_pending_email(user_id: str) -> str | None:
"""从 pending_hex 格式解析邮箱"""
if not user_id.startswith("pending_"):
return None
hex_part = user_id[8:] # 去掉 "pending_"
try:
return bytes.fromhex(hex_part).decode("utf-8")
except Exception:
return None
def _create_user_by_email(email: str) -> str | None:
"""用管理员创建用户，返回 user_id。若已存在则返回已有用户 id"""
token, _ = _get_admin_token()
if not token:
return None
base = (PB_URL or "").rstrip("/")
r = requests.post(
f"{base}/api/collections/users/records",
json={
"email": email,
"password": DEFAULT_PASSWORD,
"passwordConfirm": DEFAULT_PASSWORD,
},
headers={"Content-Type": "application/json", "Authorization": f"Bearer {token}"},
timeout=10,
)
if r.status_code == 200:
return r.json().get("id")
if r.status_code == 400:
try:
err = r.json()
if "already" in str(err.get("data", {}).get("email", {}).get("message", "")).lower():
# 用户已存在，查询获取 id
get_r = requests.get(
f"{base}/api/collections/users/records",
params={"filter": f'email="{email}"', "perPage": 1},
headers={"Authorization": f"Bearer {token}"},
timeout=10,
)
if get_r.status_code == 200:
items = get_r.json().get("items", [])
if items:
return items[0].get("id")
except Exception:
pass
logging.warning(f"create user failed: {r.status_code} {r.text[:200]}")
return None
def _get_admin_token() -> tuple[str | None, str | None]:
"""获取 PocketBase 管理员 token"""
if not PB_ADMIN_EMAIL or not PB_ADMIN_PASSWORD:
return None, "未配置"
base = (PB_URL or "").rstrip("/")
for path in ["/api/collections/_superusers/auth-with-password", "/api/admins/auth-with-password"]:
try:
r = requests.post(
f"{base}{path}",
json={"identity": PB_ADMIN_EMAIL, "password": PB_ADMIN_PASSWORD},
timeout=10,
)
if r.status_code == 200:
return r.json().get("token"), None
if r.status_code == 404:
continue
except Exception:
pass
return None, "认证失败"
def handle_payment_success(
order_id: str,
pay_price: str,
@@ -36,10 +111,22 @@ def handle_payment_success(
if not user_id:
return
# meetup 新用户：支付成功后才创建用户
if pay_type == "meetup" and user_id.startswith("pending_"):
email = _decode_pending_email(user_id)
if email:
real_id = _create_user_by_email(email)
if real_id:
user_id = real_id
logging.info(f"meetup 支付成功，已创建用户: {email}")
else:
logging.error(f"meetup 支付成功但创建用户失败: {email}")
return
print(f"支付成功 [{provider_name}] 用户ID: {user_id}，类型: {pay_type}，站点: {site_id}，订单: {order_id}")
if pay_type == "vip" and use_memos:
create_memos_user(user_id)
create_memos_user(user_id, password="12345678")
if pay_type == "meetup":
logging.info(f"meetup 类型支付成功，不创建 memos 用户 - user_id: {user_id}")

1
requirements.txt
View File

@@ -1,5 +1,6 @@
fastapi
uvicorn
python-dotenv
sqlalchemy
pydantic
python-multipart

7
run.py
View File

@@ -1,6 +1,11 @@
import uvicorn
from dotenv import load_dotenv
load_dotenv()
if __name__ == "__main__":
uvicorn.run("app.main:app", host="0.0.0.0", port=8700, reload=True)
import os
port = int(os.getenv("PORT", "8007"))
uvicorn.run("app.main:app", host="0.0.0.0", port=port, reload=True)
# uvicorn.run("app.main:app", host="0.0.0.0", port=8200, reload=True)