From d44417d10acdecf11e03f742a99d61a3491f3116 Mon Sep 17 00:00:00 2001 From: eric Date: Sun, 15 Mar 2026 19:12:12 -0500 Subject: [PATCH] 'init' --- .env | 23 + .env.example | 35 ++ .env.local | 17 + .gitignore | 13 + README.md | 102 +++- androidh5api.sh | 1 + app/__init__.py | 0 app/config.py | 59 +++ app/main.py | 188 ++++++++ app/payment/__init__.py | 25 + app/payment/base.py | 59 +++ app/payment/factory.py | 79 ++++ app/payment/xorpay.py | 108 +++++ app/payment/zpay.py | 249 ++++++++++ app/routers/__init__.py | 30 ++ app/routers/_digital_base.py | 311 ++++++++++++ app/routers/cnomadcna.py | 4 + app/routers/common.py | 65 +++ app/routers/digital.py | 4 + app/routers/legacy.py | 4 + app/routers/meetup.py | 328 +++++++++++++ app/routers/nomadlms.py | 4 + app/routers/nomadvip.py | 883 +++++++++++++++++++++++++++++++++++ app/routers/salon.py | 4 + app/routers/salon_meetup.py | 371 +++++++++++++++ app/routers/wxh5.py | 274 +++++++++++ app/sdk/__init__.py | 20 + app/sdk/memos.py | 95 ++++ app/sdk/minio.py | 92 ++++ app/sdk/payment/__init__.py | 15 + app/sdk/pocketbase.py | 69 +++ app/services/__init__.py | 16 + app/services/memos.py | 21 + app/services/minio.py | 51 ++ app/services/payment.py | 217 +++++++++ app/services/pb.py | 27 ++ payjs.sh | 1 + requirements.txt | 9 + run-with-logs.bat | 4 + run-with-logs.sh | 4 + run.py | 31 ++ 手机微信支付调试说明.md | 44 ++ 42 files changed, 3953 insertions(+), 3 deletions(-) create mode 100644 .env create mode 100644 .env.example create mode 100644 .env.local create mode 100644 .gitignore create mode 100644 androidh5api.sh create mode 100644 app/__init__.py create mode 100644 app/config.py create mode 100644 app/main.py create mode 100644 app/payment/__init__.py create mode 100644 app/payment/base.py create mode 100644 app/payment/factory.py create mode 100644 app/payment/xorpay.py create mode 100644 app/payment/zpay.py create mode 100644 app/routers/__init__.py create mode 100644 app/routers/_digital_base.py create mode 100644 app/routers/cnomadcna.py create mode 100644 app/routers/common.py create mode 100644 app/routers/digital.py create mode 100644 app/routers/legacy.py create mode 100644 app/routers/meetup.py create mode 100644 app/routers/nomadlms.py create mode 100644 app/routers/nomadvip.py create mode 100644 app/routers/salon.py create mode 100644 app/routers/salon_meetup.py create mode 100644 app/routers/wxh5.py create mode 100644 app/sdk/__init__.py create mode 100644 app/sdk/memos.py create mode 100644 app/sdk/minio.py create mode 100644 app/sdk/payment/__init__.py create mode 100644 app/sdk/pocketbase.py create mode 100644 app/services/__init__.py create mode 100644 app/services/memos.py create mode 100644 app/services/minio.py create mode 100644 app/services/payment.py create mode 100644 app/services/pb.py create mode 100644 payjs.sh create mode 100644 requirements.txt create mode 100644 run-with-logs.bat create mode 100644 run-with-logs.sh create mode 100644 run.py create mode 100644 手机微信支付调试说明.md diff --git a/.env b/.env new file mode 100644 index 0000000..e7c1515 --- /dev/null +++ b/.env @@ -0,0 +1,23 @@ +# ========== 生产环境配置 ========== + +BASE_URL=https://api.hackrobot.cn +PAYMENT_PROVIDER=zpay + +# ZPAY（请填入实际值） +ZPAY_PID=2025121809351743 +ZPAY_KEY=tpEi7wWIWI2kXiYVTpIG6j7it0mjVW89 + +# PocketBase（check-user、ensure-user 必需，否则报「服务配置错误」） +PB_URL=https://pocketbase.hackrobot.cn +PB_ADMIN_EMAIL=xiaoshuang.eric@gmail.com +PB_ADMIN_PASSWORD=Xiao4669805@ + +# Memos（nomadvip 支付成功后创建账号） +MEMOS_API=https://qun.hackrobot.cn/api/v1/users +MEMOS_TOKEN=eyJhbGciOiJIUzI1NiIsImtpZCI6InYxIiwidHlwIjoiSldUIn0.eyJuYW1lIjoiaGFja3JvYm90IiwiaXNzIjoibWVtb3MiLCJzdWIiOiIxIiwiYXVkIjpbInVzZXIuYWNjZXNzLXRva2VuIl0sImlhdCI6MTc1NzQwNTE0OX0.Idn7kBlxE-CoSOXwWuZ1tHGIRKHAIeDyXSafGS5OHsg + +# MinIO（可选） +MINIO_ENDPOINT=minioweb.hackrobot.cn +MINIO_PORT=443 +MINIO_BUCKET=hackrobot +MINIO_UPLOAD_PREFIX=joins diff --git a/.env.example b/.env.example new file mode 100644 index 0000000..8f5c641 --- /dev/null +++ b/.env.example @@ -0,0 +1,35 @@ +# ========== 本地调试 ========== +# PORT: 默认 8007，与前端 PAYMENT_API_URL 一致 +# BASE_URL 需为 ZPAY 可访问的回调地址。本地调试时 ZPAY 无法回调 localhost/内网， +# 可用 ngrok 等隧道：BASE_URL=https://xxx.ngrok.io +# 或直接使用生产地址测试回调（需 payjsapi 已部署到公网） +PORT=8007 +BASE_URL=https://api.hackrobot.cn + +# 支付渠道：zpay | xorpay +PAYMENT_PROVIDER=zpay + +# XorPay +XORPAY_AID=8220 +XORPAY_SECRET=your_secret + +# ZPAY（网关: https://zpayz.cn/） +ZPAY_PID=2025121809351743 +ZPAY_KEY=tpEi7wWIWI2kXiYVTpIG6j7it0mjVW89 + +# PocketBase（check-user、ensure-user、支付回调、nomadvip/digital/cnomadcna 共用） +PB_URL=https://pocketbase.hackrobot.cn +PB_ADMIN_EMAIL=xiaoshuang.eric@gmail.com +PB_ADMIN_PASSWORD=Xiao4669805@ + +# Memos（nomadvip 支付成功后创建账号，与 payjsapi 同源） +MEMOS_API=https://qun.hackrobot.cn/api/v1/users +MEMOS_TOKEN=eyJhbGciOiJIUzI1NiIsImtpZCI6InYxIiwidHlwIjoiSldUIn0.eyJuYW1lIjoiaGFja3JvYm90IiwiaXNzIjoibWVtb3MiLCJzdWIiOiIxIiwiYXVkIjpbInVzZXIuYWNjZXNzLXRva2VuIl0sImlhdCI6MTc1NzQwNTE0OX0.Idn7kBlxE-CoSOXwWuZ1tHGIRKHAIeDyXSafGS5OHsg + +# MinIO 对象存储（可选） +MINIO_ENDPOINT=minioweb.hackrobot.cn +MINIO_PORT=443 +MINIO_ACCESS_KEY= +MINIO_SECRET_KEY= +MINIO_BUCKET=hackrobot +MINIO_UPLOAD_PREFIX=joins diff --git a/.env.local b/.env.local new file mode 100644 index 0000000..16b8160 --- /dev/null +++ b/.env.local @@ -0,0 +1,17 @@ +# 本地调试 - 端口 8007、zpay 渠道 +PORT=8007 +PAYMENT_PROVIDER=zpay +BASE_URL=https://api.hackrobot.cn + +# PocketBase（支付回调、nomadvip/digital/cnomadcna 共用） +PB_URL=https://pocketbase.hackrobot.cn +PB_ADMIN_EMAIL=xiaoshuang.eric@gmail.com +PB_ADMIN_PASSWORD=Xiao4669805@ + +# Memos（nomadvip 支付成功后创建账号） +MEMOS_API=https://qun.hackrobot.cn/api/v1/users +MEMOS_TOKEN=eyJhbGciOiJIUzI1NiIsImtpZCI6InYxIiwidHlwIjoiSldUIn0.eyJuYW1lIjoiaGFja3JvYm90IiwiaXNzIjoibWVtb3MiLCJzdWIiOiIxIiwiYXVkIjpbInVzZXIuYWNjZXNzLXRva2VuIl0sImlhdCI6MTc1NzQwNTE0OX0.Idn7kBlxE-CoSOXwWuZ1tHGIRKHAIeDyXSafGS5OHsg + +# ZPAY +ZPAY_PID=2025121809351743 +ZPAY_KEY=tpEi7wWIWI2kXiYVTpIG6j7it0mjVW89 diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..8dc3230 --- /dev/null +++ b/.gitignore @@ -0,0 +1,13 @@ +# 日志文件（不产生，避免 git 本地/线上不一致） +*.log +payment_notify.log +xorpay_notify.log + +# Python +__pycache__/ +*.pyc +*.py[cod] +venv/ +.venv/ +# .env +# .env.local diff --git a/README.md b/README.md index b42ee17..0bc21ca 100644 --- a/README.md +++ b/README.md @@ -1,6 +1,102 @@ -# salonapi +# PayJS API +支付 API 服务，支持多支付渠道（XorPay、ZPAY），按项目模块化拆分。 +## 路由与项目对应 + +| 项目 | 路由前缀 | 说明 | +|------|----------|------| +| nomadvip | `/nomadvip/*` | 固定 ZPAY，独立实例 | +| digital | `/digital/*` | 使用 PAYMENT_PROVIDER | +| cnomadcna | `/cnomadcna/*` | 使用 PAYMENT_PROVIDER | +| nomadlms | `/nomadlms/*` | 使用 PAYMENT_PROVIDER | +| legacy | `/payh5`、`/zpay_notify` 等 | 向后兼容 | + +各项目业务互不影响，可单独配置。 + +## 环境 + +本地调试：`BASE_URL` 需为 ZPAY 可访问地址（如 ngrok 暴露 8700 端口） +线上部署：`BASE_URL=https://api.hackrobot.cn` + +## 模块结构 + +``` +app/ +├── config.py # 配置（环境变量） +├── main.py # FastAPI 入口，挂载路由 +├── sdk/ # SDK 组件化封装 +│ ├── payment/ # 支付网关（ZPAY/XorPay） +│ ├── pocketbase.py # PocketBase 数据库 +│ ├── minio.py # MinIO 对象存储 +│ └── memos.py # Memos 用户（仅 nomadvip 使用） +├── services/ # 共享业务逻辑（基于 SDK） +│ ├── pb.py # PocketBase 客户端 +│ ├── memos.py # Memos 用户创建 +│ ├── minio.py # MinIO 上传 +│ └── payment.py # 支付成功处理 +├── routers/ # 项目路由 +│ ├── nomadvip.py # /nomadvip/* +│ ├── digital.py # /digital/* +│ ├── cnomadcna.py # /cnomadcna/* +│ ├── nomadlms.py # /nomadlms/* +│ ├── legacy.py # /payh5 等（向后兼容） +│ └── common.py # /submit_meetup_application、/create_user +└── payment/ + ├── base.py # 支付抽象基类 + ├── xorpay.py # XorPay 实现 + ├── zpay.py # ZPAY 实现（易支付兼容） + └── factory.py # 渠道工厂 +``` + +## 支付渠道 + +| 渠道 | 环境变量 | 说明 | +|------|----------|------| +| XorPay | `PAYMENT_PROVIDER=xorpay` | 默认 | +| ZPAY | `PAYMENT_PROVIDER=zpay` | 需配置 `ZPAY_PID`、`ZPAY_KEY` | + +切换渠道：设置环境变量 `PAYMENT_PROVIDER=xorpay` 或 `zpay`，参考 `.env.example`。 + +## 本地调试 + +```bash +# 复制配置 +cp .env.example .env + +# 本地调试时 BASE_URL 需 ZPAY 可访问，可用 ngrok： +# ngrok http 8700 -> 将 BASE_URL 设为 ngrok 提供的 https 地址 + +python run.py +# 或 uvicorn app.main:app --host 0.0.0.0 --port 8700 --reload +``` + +## 线上部署 + +域名：`https://api.hackrobot.cn` + +```bash +# .env 中设置 +BASE_URL=https://api.hackrobot.cn +PAYMENT_PROVIDER=zpay +# 其他 ZPAY、PocketBase、Memos 配置见 .env.example +``` + +## 线上「支付失败」排查 + +本地正常、Ubuntu 部署后 api.hackrobot.cn 调用支付失败，常见原因： + +1. **BASE_URL 配置错误**：必须为 ZPAY 可公网访问的地址。错误示例：`http://127.0.0.1:8700`、`http://localhost`。正确：`https://api.hackrobot.cn`（域名需解析到本机，nginx 反向代理对应端口）。 + +2. **服务器无法访问 zpayz.cn**：防火墙、DNS 或网络限制。访问 `https://api.hackrobot.cn/health?check_zpay=1` 自检： + - `zpayz_reachable: false` → 开放出站或检查网络 + - `zpayz_error` 含 "SSL" → 执行 `apt install ca-certificates` 后重启 + +3. **client_ip 内网导致 ZPAY 风控拒单**：前端（digital/cnomadcna/nomadvip）通过 nginx 代理时，需配置 `proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for`，否则 payjsapi 收到 127.0.0.1 会触发风控。已做兼容：内网 IP 时自动省略 clientip 参数。 + +4. **启动时控制台有 BASE_URL 警告**：说明配置了内网地址，需修改 `.env` 中的 `BASE_URL` 为公网地址后重启。 + +5. **查看详细错误**：支付失败时，payjsapi 会输出 `ZPAY create_order_api 失败` 日志，含 status/code/msg，便于定位。 ## Getting started @@ -15,14 +111,14 @@ Already a pro? Just edit this README.md and make it your own. Want to make it ea ``` cd existing_repo -git remote add origin http://gitlab.yidooplanet.com/gitlab-instance-0a899031/salonapi.git +git remote add origin http://gitlab.yidooplanet.com/gitlab-instance-0a899031/androidh5api.git git branch -M main git push -uf origin main ``` ## Integrate with your tools -- [ ] [Set up project integrations](http://gitlab.yidooplanet.com/gitlab-instance-0a899031/salonapi/-/settings/integrations) +- [ ] [Set up project integrations](http://gitlab.yidooplanet.com/gitlab-instance-0a899031/androidh5api/-/settings/integrations) ## Collaborate with your team diff --git a/androidh5api.sh b/androidh5api.sh new file mode 100644 index 0000000..64d2ef0 --- /dev/null +++ b/androidh5api.sh @@ -0,0 +1 @@ +python3 run.py \ No newline at end of file diff --git a/app/__init__.py b/app/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/app/config.py b/app/config.py new file mode 100644 index 0000000..23c7dcf --- /dev/null +++ b/app/config.py @@ -0,0 +1,59 @@ +""" +应用配置 +支付渠道可通过 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 接口业务保留） +PAYMENT_PROVIDER: PaymentChannel = os.getenv("PAYMENT_PROVIDER", "zpay").lower() + +# XorPay 配置 +XORPAY_AID = os.getenv("XORPAY_AID", "8220") +XORPAY_SECRET = os.getenv("XORPAY_SECRET", "afcacd99570945f88de62624aaa3578e") +XORPAY_CASHIER_URL = os.getenv("XORPAY_CASHIER_URL", "https://xorpay.com/api/cashier") + +# ZPAY 配置（网关: https://zpayz.cn/） +ZPAY_PID = os.getenv("ZPAY_PID", "2025121809351743") +ZPAY_KEY = os.getenv("ZPAY_KEY", "tpEi7wWIWI2kXiYVTpIG6j7it0mjVW89") + +# 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") +MEMOS_TOKEN = os.getenv( + "MEMOS_TOKEN", + "eyJhbGciOiJIUzI1NiIsImtpZCI6InYxIiwidHlwIjoiSldUIn0.eyJuYW1lIjoiaGFja3JvYm90IiwiaXNzIjoibWVtb3MiLCJzdWIiOiIxIiwiYXVkIjpbInVzZXIuYWNjZXNzLXRva2VuIl0sImlhdCI6MTc1NzQwNTE0OX0.Idn7kBlxE-CoSOXwWuZ1tHGIRKHAIeDyXSafGS5OHsg", +) + +# 回调地址（用于生成 notify_url） +BASE_URL = os.getenv("BASE_URL", "https://api.hackrobot.cn") + +# MinIO 对象存储 +MINIO_ENDPOINT = os.getenv("MINIO_ENDPOINT", "minioweb.hackrobot.cn") +MINIO_PORT = int(os.getenv("MINIO_PORT", "443")) +MINIO_ACCESS_KEY = os.getenv("MINIO_ACCESS_KEY", "") +MINIO_SECRET_KEY = os.getenv("MINIO_SECRET_KEY", "") +MINIO_BUCKET = os.getenv("MINIO_BUCKET", "hackrobot") +MINIO_USE_SSL = os.getenv("MINIO_USE_SSL", "true").lower() == "true" +_minio_ep = os.getenv("MINIO_ENDPOINT", "minioweb.hackrobot.cn") +_minio_bucket = os.getenv("MINIO_BUCKET", "hackrobot") +MINIO_PUBLIC_URL = os.getenv("MINIO_PUBLIC_URL", f"https://{_minio_ep}/{_minio_bucket}") +MINIO_UPLOAD_PREFIX = os.getenv("MINIO_UPLOAD_PREFIX", "joins") diff --git a/app/main.py b/app/main.py new file mode 100644 index 0000000..a68a9ac --- /dev/null +++ b/app/main.py @@ -0,0 +1,188 @@ +""" +支付 API 主入口 +支持多支付渠道（XorPay、ZPAY），通过 PAYMENT_PROVIDER 环境变量切换 + +路由模块化，按项目拆分： +- nomadvip: /nomadvip/* - 固定 ZPAY +- digital: /digital/* - 使用 PAYMENT_PROVIDER +- cnomadcna: /cnomadcna/* - 使用 PAYMENT_PROVIDER +- nomadlms: /nomadlms/* - 使用 PAYMENT_PROVIDER +- legacy: /payh5、/zpay_notify 等（向后兼容） +- common: /submit_meetup_application、/create_user + +本地调试：BASE_URL 需为 ZPAY 可访问地址（如 ngrok） +线上部署：BASE_URL=https://api.hackrobot.cn（必须为 ZPAY 可公网访问的地址） +""" +import logging +import sys + +import requests +from fastapi import FastAPI, Request, HTTPException +from fastapi.middleware.cors import CORSMiddleware +from fastapi.responses import JSONResponse + +from .config import BASE_URL +from .routers import ( + nomadvip_router, + digital_router, + cnomadcna_router, + nomadlms_router, + salon_router, + legacy_router, + common_router, + meetup_router, + salon_meetup_router, +) + +app = FastAPI( + title="PayJS API", + description="nomadvip / digital / cnomadcna / nomadlms 支付接口", + version="2.0", +) + +app.add_middleware( + CORSMiddleware, + allow_origins=["*"], + allow_credentials=False, + allow_methods=["*"], + allow_headers=["*"], +) + + +class _SafeStderrProxy: + def write(self, message: str) -> int: + try: + return sys.stderr.write(message) + except (OSError, ValueError): + return 0 + + def flush(self) -> None: + try: + sys.stderr.flush() + except (OSError, ValueError): + pass + + +def _safe_log(level: int, message: str, *args, **kwargs) -> None: + try: + logging.log(level, message, *args, **kwargs) + except Exception: + pass + + +@app.exception_handler(Exception) +async def global_exception_handler(request: Request, exc: Exception): + """捕获未处理异常，记录日志后返回 500（HTTPException 交由 FastAPI 默认处理）""" + if isinstance(exc, HTTPException): + raise exc + _safe_log(logging.ERROR, "未处理异常: %s - %s", request.url.path, exc, exc_info=True) + return JSONResponse(status_code=500, content={"detail": str(exc)}) + +logging.raiseExceptions = False +logging.basicConfig( + level=logging.INFO, + format="%(asctime)s [%(levelname)s] %(name)s: %(message)s", + handlers=[logging.StreamHandler(_SafeStderrProxy())], + force=True, +) + + +def _safe_stderr_write(message: str) -> None: + try: + sys.stderr.write(message) + sys.stderr.flush() + except (OSError, ValueError): + pass + + +def _should_trace_request(path: str) -> bool: + return any( + segment in path for segment in ("/nomadvip", "/digital", "/cnomadcna", "/nomadlms") + ) + +# 启动时校验 BASE_URL（线上部署必须为公网可访问地址，否则 ZPAY 无法回调） +def _check_base_url(): + url = (BASE_URL or "").strip().lower() + if not url: + logging.warning("BASE_URL 未设置，将使用默认值。线上部署请务必设置 BASE_URL 为公网地址") + return + if "localhost" in url or "127.0.0.1" in url or url.startswith("http://192.168.") or url.startswith("http://10."): + logging.warning( + "BASE_URL=%s 疑似内网地址，ZPAY 无法回调！线上部署请设置 BASE_URL 为公网地址，如 https://api.hackrobot.cn", + BASE_URL, + ) + + +@app.on_event("startup") +async def startup_event(): + _check_base_url() + + +@app.get("/health") +async def health(check_zpay: str = ""): + """健康检查。check_zpay=1 时额外检测 zpayz.cn 连通性（含 SSL、mapi 接口）""" + out = {"status": "ok", "service": "payjsapi", "base_url": BASE_URL} + if str(check_zpay).lower() in ("1", "true", "yes"): + # 1) 检测 zpayz.cn 首页 + try: + r = requests.get("https://zpayz.cn", timeout=5) + out["zpayz_reachable"] = r.status_code in (200, 301, 302, 404) + except requests.exceptions.SSLError as e: + out["zpayz_reachable"] = False + out["zpayz_error"] = f"SSL 错误: {e}（Ubuntu 请执行: apt install ca-certificates）" + except requests.RequestException as e: + out["zpayz_reachable"] = False + out["zpayz_error"] = str(e) + # 2) 检测 mapi 接口（实际支付调用地址） + if out.get("zpayz_reachable"): + try: + r2 = requests.post("https://zpayz.cn/mapi.php", data={"pid": "test"}, timeout=5) + out["zpayz_mapi_reachable"] = True + out["zpayz_mapi_status"] = r2.status_code + except requests.exceptions.SSLError as e: + out["zpayz_mapi_reachable"] = False + out["zpayz_mapi_error"] = f"SSL: {e}" + except requests.RequestException as e: + out["zpayz_mapi_reachable"] = False + out["zpayz_mapi_error"] = str(e) + return out + + +# 请求日志中间件：每个请求都打印到 stderr，确保终端可见 +@app.middleware("http") +async def log_requests(request, call_next): + path = request.url.path + if _should_trace_request(path): + _safe_stderr_write(f"[payjsapi] >>> {request.method} {path}\n") + resp = await call_next(request) + if _should_trace_request(path): + _safe_stderr_write(f"[payjsapi] <<< {request.method} {path} {resp.status_code}\n") + return resp + +# 挂载路由 +app.include_router(nomadvip_router) +app.include_router(digital_router) +app.include_router(cnomadcna_router) +app.include_router(nomadlms_router) +app.include_router(salon_router) +app.include_router(legacy_router) +app.include_router(common_router) +app.include_router(meetup_router) +app.include_router(salon_meetup_router) + + +@app.get("/") +async def root(): + """健康检查""" + return { + "status": "ok", + "service": "payjsapi", + "routes": { + "nomadvip": "/nomadvip/payh5, /nomadvip/payh5/redirect, /nomadvip/check_vip, /nomadvip/check_vip_by_email, /nomadvip/order_status, /nomadvip/zpay_order_status, /nomadvip/zpay_notify, /nomadvip/xorpay_notify", + "digital": "/digital/payh5, /digital/payh5/redirect, /digital/zpay_order_status, /digital/zpay_notify, /digital/xorpay_notify", + "cnomadcna": "/cnomadcna/payh5, ...", + "nomadlms": "/nomadlms/payh5, ...", + "legacy": "/payh5, /payh5/redirect, /zpay_notify, /xorpay_notify, /zpay_order_status", + "common": "/submit_meetup_application, /create_user", + }, + } diff --git a/app/payment/__init__.py b/app/payment/__init__.py new file mode 100644 index 0000000..6b6ab1e --- /dev/null +++ b/app/payment/__init__.py @@ -0,0 +1,25 @@ +""" +支付模块 +支持多支付渠道，可通过配置切换 +""" +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, + resolve_provider, + resolve_channel, +) + +__all__ = [ + "PaymentProvider", + "XorPayProvider", + "ZPayProvider", + "get_payment_provider", + "get_payment_provider_by_name", + "reset_provider", + "resolve_provider", + "resolve_channel", +] diff --git a/app/payment/base.py b/app/payment/base.py new file mode 100644 index 0000000..cd0679b --- /dev/null +++ b/app/payment/base.py @@ -0,0 +1,59 @@ +""" +支付渠道抽象基类 +定义统一的支付接口，便于切换不同支付渠道（XorPay、ZPAY 等） +""" +from abc import ABC, abstractmethod +from typing import Any, Dict, Optional + + +class PaymentProvider(ABC): + """支付渠道抽象基类""" + + @property + @abstractmethod + def name(self) -> str: + """支付渠道名称""" + pass + + @abstractmethod + def create_order( + self, + *, + order_id: str, + name: str, + total_fee_yuan: str, + pay_type: str, + notify_url: str, + return_url: Optional[str] = None, + client_ip: Optional[str] = None, + extra: Optional[Dict[str, Any]] = None, + ) -> Dict[str, Any]: + """ + 创建支付订单 + :return: 包含支付跳转信息等，格式由各渠道自定义 + """ + pass + + @abstractmethod + def verify_notify(self, data: Dict[str, Any]) -> bool: + """验证异步通知签名""" + pass + + @abstractmethod + def parse_notify(self, data: Dict[str, Any]) -> Dict[str, Any]: + """ + 解析异步通知，提取统一字段 + :return: { + "order_id": str, # 商户订单号 + "trade_no": str, # 渠道订单号 + "pay_price": str, # 实付金额（元） + "pay_time": str, # 支付时间 + "trade_status": str, # 支付状态 + } + """ + pass + + @abstractmethod + def success_response(self) -> str: + """异步通知成功时需返回的字符串（如 'ok' 或 'success'）""" + pass diff --git a/app/payment/factory.py b/app/payment/factory.py new file mode 100644 index 0000000..40ca263 --- /dev/null +++ b/app/payment/factory.py @@ -0,0 +1,79 @@ +""" +支付渠道工厂 +根据配置返回对应的支付 Provider +默认使用 zpay 替代 xorpay +""" +from ..config import ( + PAYMENT_PROVIDER, + XORPAY_AID, + XORPAY_SECRET, + XORPAY_CASHIER_URL, + ZPAY_PID, + ZPAY_KEY, +) +from .base import PaymentProvider +from .xorpay import XorPayProvider +from .zpay import ZPayProvider + +_provider_instance: PaymentProvider | None = None + + +def get_payment_provider() -> PaymentProvider: + """获取当前配置的支付渠道实例（单例）""" + global _provider_instance + if _provider_instance is not None: + return _provider_instance + if PAYMENT_PROVIDER == "zpay": + _provider_instance = ZPayProvider(pid=ZPAY_PID, key=ZPAY_KEY) + else: + _provider_instance = XorPayProvider( + aid=XORPAY_AID, + secret=XORPAY_SECRET, + cashier_url=XORPAY_CASHIER_URL, + ) + return _provider_instance + + +def get_payment_provider_by_name(name: str) -> PaymentProvider: + """按名称获取 Provider（用于 per-request 覆盖）""" + if name == "xorpay": + return XorPayProvider( + aid=XORPAY_AID, + secret=XORPAY_SECRET, + cashier_url=XORPAY_CASHIER_URL, + ) + return ZPayProvider(pid=ZPAY_PID, key=ZPAY_KEY) + + +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" diff --git a/app/payment/xorpay.py b/app/payment/xorpay.py new file mode 100644 index 0000000..99d2941 --- /dev/null +++ b/app/payment/xorpay.py @@ -0,0 +1,108 @@ +""" +XorPay 支付渠道实现 +文档: https://xorpay.com +""" +import hashlib +from typing import Any, Dict, Optional + +import requests + +from .base import PaymentProvider + + +class XorPayProvider(PaymentProvider): + """XorPay 支付""" + + def __init__(self, aid: str, secret: str, cashier_url: str = "https://xorpay.com/api/cashier"): + self.aid = aid + self.secret = secret + self.cashier_url = cashier_url.rstrip("/") + + @property + def name(self) -> str: + return "xorpay" + + def _sign(self, name: str, pay_type: str, price: str, order_id: str, notify_url: str) -> str: + 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, + *, + order_id: str, + name: str, + total_fee_yuan: str, + pay_type: str, + notify_url: str, + return_url: Optional[str] = None, + client_ip: Optional[str] = None, + extra: Optional[Dict[str, Any]] = None, + ) -> Dict[str, Any]: + # XorPay 使用 jsapi 表示 H5/JSAPI 支付 + channel_pay_type = "jsapi" + params = { + "name": name, + "pay_type": channel_pay_type, + "price": total_fee_yuan, + "order_id": order_id, + "notify_url": notify_url, + } + params["sign"] = self._sign( + params["name"], + params["pay_type"], + params["price"], + params["order_id"], + params["notify_url"], + ) + return { + "status": "ok", + "provider": self.name, + "params": params, + "pay_url": f"{self.cashier_url}/{self.aid}", + } + + def verify_notify(self, data: Dict[str, Any]) -> 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 + self.secret + calculated = hashlib.md5(raw.encode("utf-8")).hexdigest().lower() + return calculated == received_sign + + def parse_notify(self, data: Dict[str, Any]) -> Dict[str, Any]: + return { + "order_id": data.get("order_id", ""), + "trade_no": data.get("aoid", ""), + "pay_price": data.get("pay_price", ""), + "pay_time": data.get("pay_time", ""), + "trade_status": "success", # XorPay 验签通过即视为成功 + } + + 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)} diff --git a/app/payment/zpay.py b/app/payment/zpay.py new file mode 100644 index 0000000..6978457 --- /dev/null +++ b/app/payment/zpay.py @@ -0,0 +1,249 @@ +""" +ZPAY 支付渠道实现 +文档: https://z-pay.cn/doc.html +支持：页面跳转支付、API 接口支付、异步通知验签 +""" +import hashlib +import json +import logging +from typing import Any, Dict, Optional + +import requests + +from .base import PaymentProvider + +logger = logging.getLogger(__name__) + + +def _is_internal_ip(ip: Optional[str]) -> bool: + """判断是否为内网 IP，ZPAY 可能拒收导致 Ubuntu 部署失败""" + if not ip or not ip.strip(): + return True + ip = ip.strip() + if ip in ("127.0.0.1", "localhost", "::1"): + return True + if ip.startswith("10.") or ip.startswith("192.168.") or ip.startswith("172.16."): + return True + return False + + +class ZPayProvider(PaymentProvider): + """ZPAY 支付（易支付兼容接口）""" + + # API 地址 + SUBMIT_URL = "https://zpayz.cn/submit.php" # 页面跳转 + MAPI_URL = "https://zpayz.cn/mapi.php" # API 接口 + + def __init__(self, pid: str, key: str): + self.pid = pid + self.key = key + + @property + def name(self) -> str: + return "zpay" + + def _md5_sign(self, params: Dict[str, Any], exclude: Optional[set] = None) -> str: + """ + ZPAY MD5 签名算法： + 1. 参数按 ASCII 排序，sign/sign_type/空值不参与 + 2. 拼接为 a=b&c=d + 3. sign = md5(拼接串 + KEY)，小写 + """ + exclude = exclude or {"sign", "sign_type"} + filtered = {k: v for k, v in params.items() if k not in exclude and v is not None and v != ""} + sorted_keys = sorted(filtered.keys()) + parts = [f"{k}={filtered[k]}" for k in sorted_keys] + raw = "&".join(parts) + self.key + return hashlib.md5(raw.encode("utf-8")).hexdigest().lower() + + def create_order( + self, + *, + order_id: str, + name: str, + total_fee_yuan: str, + pay_type: str, + notify_url: str, + return_url: Optional[str] = None, + client_ip: Optional[str] = None, + extra: Optional[Dict[str, Any]] = None, + ) -> Dict[str, Any]: + # ZPAY: alipay / wxpay（pay_type 可为业务类型或支付方式，此处按支付方式解析） + channel_type = "wxpay" if str(pay_type).lower() in ("wx", "wechat", "wxpay") else "alipay" + params = { + "pid": self.pid, + "type": channel_type, + "out_trade_no": order_id, + "notify_url": notify_url, + "name": name, + "money": total_fee_yuan, + "return_url": return_url or notify_url, + } + if client_ip and not _is_internal_ip(client_ip): + params["clientip"] = client_ip + elif client_ip and _is_internal_ip(client_ip): + logger.warning("ZPAY create_order 跳过内网 client_ip=%s", client_ip) + if extra and extra.get("param"): + params["param"] = str(extra["param"]) + params["sign_type"] = "MD5" + params["sign"] = self._md5_sign(params) + + # 页面跳转方式：返回 form 提交 URL 和参数 + return { + "status": "ok", + "provider": self.name, + "params": params, + "pay_url": self.SUBMIT_URL, + "submit_method": "POST", + } + + def create_order_api( + self, + *, + order_id: str, + name: str, + total_fee_yuan: str, + pay_type: str, + notify_url: str, + return_url: Optional[str] = None, + client_ip: str = "127.0.0.1", + device: str = "pc", + extra: Optional[Dict[str, Any]] = None, + ) -> Dict[str, Any]: + """ + ZPAY API 接口支付（返回 payurl/qrcode 等） + 适用于需要二维码或 H5 跳转的场景 + """ + channel_type = "wxpay" if pay_type.lower() in ("wx", "wechat", "wxpay") else "alipay" + effective_ip = (client_ip or "").strip() or "127.0.0.1" + # 内网 IP（127.0.0.1/10.x/192.168.x）可能导致 ZPAY 风控拒单，Ubuntu 部署时常见 + if _is_internal_ip(effective_ip): + logger.warning( + "ZPAY create_order_api client_ip=%s 为内网地址，已省略 clientip 参数（避免风控拒单）", + effective_ip, + ) + effective_ip = "" # 不传 clientip，由 ZPAY 从连接获取 + params = { + "pid": self.pid, + "type": channel_type, + "out_trade_no": order_id, + "notify_url": notify_url, + "name": name, + "money": total_fee_yuan, + "device": device, + } + if effective_ip: + params["clientip"] = effective_ip + if return_url: + params["return_url"] = return_url + if extra and extra.get("param"): + params["param"] = str(extra["param"]) + params["sign_type"] = "MD5" + params["sign"] = self._md5_sign(params) + + try: + resp = requests.post(self.MAPI_URL, data=params, timeout=15) + except requests.exceptions.SSLError as e: + logger.error("ZPAY SSL 错误（Ubuntu 可能 CA 证书不全）: %s", e, exc_info=True) + return { + "status": "error", + "provider": self.name, + "msg": f"连接 zpayz.cn SSL 失败: {e}。Ubuntu 请安装 ca-certificates: apt install ca-certificates", + } + except requests.RequestException as e: + logger.error("ZPAY 连接失败: %s", e, exc_info=True) + return { + "status": "error", + "provider": self.name, + "msg": f"连接 zpayz.cn 失败: {e}。请检查服务器网络、防火墙或 DNS。", + } + + try: + result = resp.json() + except Exception: + result = {} + # ZPAY 可能返回二次编码的 JSON（body 为 JSON 字符串），需再次解析 + if isinstance(result, str): + try: + result = json.loads(result) + except Exception: + result = {} + if not isinstance(result, dict): + result = {} + # 部分环境下 ZPAY 将成功数据放在 msg 中（msg 为 JSON 字符串） + msg_val = result.get("msg") + if isinstance(msg_val, str) and msg_val.strip().startswith("{"): + try: + inner = json.loads(msg_val) + if isinstance(inner, dict) and int(float(inner.get("code", 0))) == 1: + result = inner + except Exception: + pass + try: + code = int(float(result.get("code", 0))) + except (TypeError, ValueError): + code = 0 + if code == 1: + return { + "status": "ok", + "provider": self.name, + "payurl": result.get("payurl"), + "payurl2": result.get("payurl2"), # 微信 H5，手机浏览器打开可唤醒微信 + "qrcode": result.get("qrcode"), + "img": result.get("img"), + "O_id": result.get("O_id"), + "trade_no": result.get("trade_no"), + } + err_msg = result.get("msg", resp.text) + logger.error( + "ZPAY create_order_api 失败 status=%s code=%s msg=%s resp_text=%s", + resp.status_code, + code, + err_msg, + resp.text[:500] if resp.text else "", + ) + return { + "status": "error", + "provider": self.name, + "msg": err_msg, + } + + def verify_notify(self, data: Dict[str, Any]) -> bool: + received = data.get("sign", "") + calculated = self._md5_sign(data) + return calculated == received + + def parse_notify(self, data: Dict[str, Any]) -> Dict[str, Any]: + # ZPAY 通知：trade_status 为 TRADE_SUCCESS 表示成功 + status = "success" if data.get("trade_status") == "TRADE_SUCCESS" else "pending" + return { + "order_id": data.get("out_trade_no", ""), + "trade_no": data.get("trade_no", ""), + "pay_price": data.get("money", ""), + "pay_time": "", # ZPAY 通知未明确返回 + "trade_status": status, + } + + def success_response(self) -> str: + return "success" + + def query_order_status(self, out_trade_no: str, timeout: int = 15) -> Dict[str, Any]: + """查询 ZPAY 订单支付状态（供 nomadvip/digital/cnomadcna/nomadlms 共用）""" + url = f"https://zpayz.cn/api.php?act=order&pid={self.pid}&key={self.key}&out_trade_no={out_trade_no}" + try: + resp = requests.get(url, timeout=timeout) + result = resp.json() if resp.ok else {} + except Exception as e: + return {"paid": False, "error": str(e)} + try: + code = result.get("code", 0) + code = int(code) if isinstance(code, (int, float)) or str(code).isdigit() else 0 + except (ValueError, TypeError): + code = 0 + if code != 1: + return {"paid": False, "msg": result.get("msg", "查询失败")} + try: + status = int(result.get("status", 0)) + except (ValueError, TypeError): + status = 0 + return {"paid": status == 1, "status": status} diff --git a/app/routers/__init__.py b/app/routers/__init__.py new file mode 100644 index 0000000..3b312b5 --- /dev/null +++ b/app/routers/__init__.py @@ -0,0 +1,30 @@ +""" +项目路由模块 +- nomadvip: /nomadvip/* +- digital: /digital/* +- cnomadcna: /cnomadcna/* +- nomadlms: /nomadlms/* +- salon: /salon/*（克隆 cnomadcna） +- legacy: /payh5, /zpay_notify 等（向后兼容 digital/cnomadcna） +""" +from .nomadvip import router as nomadvip_router +from .digital import router as digital_router +from .cnomadcna import router as cnomadcna_router +from .nomadlms import router as nomadlms_router +from .salon import router as salon_router +from .legacy import router as legacy_router +from .common import router as common_router +from .meetup import router as meetup_router +from .salon_meetup import router as salon_meetup_router + +__all__ = [ + "nomadvip_router", + "digital_router", + "cnomadcna_router", + "nomadlms_router", + "salon_router", + "legacy_router", + "common_router", + "meetup_router", + "salon_meetup_router", +] diff --git a/app/routers/_digital_base.py b/app/routers/_digital_base.py new file mode 100644 index 0000000..48aa621 --- /dev/null +++ b/app/routers/_digital_base.py @@ -0,0 +1,311 @@ +""" +digital/cnomadcna/nomadlms 共享的支付逻辑 +根据 prefix 生成对应 notify_path，业务互不影响 +""" +import logging +import random +import string +import time + +from fastapi import APIRouter, Request, HTTPException + +from ..config import BASE_URL +from ..payment import ( + get_payment_provider, + get_payment_provider_by_name, + resolve_provider, + resolve_channel, +) +from ..payment.zpay import ZPayProvider +from ..services import handle_payment_success, processed_orders + + +def _safe_log(level: int, message: str, *args, **kwargs) -> None: + try: + logging.log(level, message, *args, **kwargs) + except Exception: + pass + + +def create_digital_router(prefix: str, project_name: str, site_id: str | None = None) -> APIRouter: + """ + 创建 digital/cnomadcna/nomadlms 的支付路由 + prefix: 如 "/digital", "/cnomadcna", "/nomadlms" 或 ""（legacy） + project_name: 用于日志，如 "digital", "cnomadcna", "nomadlms", "legacy" + site_id: 站点标识，用于 site_vip（digital/meetup/vip） + """ + router = APIRouter(prefix=prefix, tags=[project_name]) + + def _notify_path(path: str) -> str: + base = BASE_URL.rstrip("/") + if prefix: + return f"{base}{prefix}{path}" + return f"{base}{path}" + + @router.post("/payh5") + async def payh5(item: dict, request: Request): + """创建 H5 支付订单。默认微信支付，PC/手机用 zpay，微信内用 xorpay，支付宝暂不提供""" + req_provider = (item.get("provider") or "").strip().lower() or None + device = (item.get("device") or "pc").lower() + ua = (request.headers.get("user-agent") or "").strip() + provider = resolve_provider(device, req_provider, ua) + channel = resolve_channel(item.get("channel")) + user_id = item.get("user_id", "unknown") + pay_type = (item.get("type") or "unknown").lower() + _safe_log( + logging.INFO, + "[payjsapi] %s/payh5 user_id=%s type=%s provider=%s channel=%s", + project_name, + user_id, + pay_type, + provider.name, + channel, + ) + try: + total_fee = int(item.get("total_fee", 80)) + except (TypeError, ValueError): + total_fee = 80 + total_fee_yuan = f"{total_fee / 100:.2f}" + random_suffix = "".join(random.choices(string.hexdigits.lower(), k=8)) + timestamp = int(time.time()) + order_id = f"{user_id}_{pay_type}_order_{timestamp}_{random_suffix}" + return_url = item.get("return_url") + if return_url and "order_id=" not in return_url and "out_trade_no=" not in return_url: + sep = "&" if "?" in return_url else "?" + return_url = f"{return_url}{sep}order_id={order_id}" + type_map = { + "book": "购买书籍", + "meetup": "数字游民社区报名", + "video": "视频解锁", + "vip": "VIP会员充值", + } + name = type_map.get(pay_type, "商品支付") + notify_path = "/xorpay_notify" if provider.name == "xorpay" else "/zpay_notify" + notify_url = _notify_path(notify_path) + result = provider.create_order( + order_id=order_id, + name=name, + total_fee_yuan=total_fee_yuan, + pay_type=channel, + notify_url=notify_url, + return_url=return_url, + client_ip=item.get("client_ip"), + ) + _safe_log( + logging.INFO, + "[payjsapi] %s/payh5 created order_id=%s provider=%s notify_url=%s return_url=%s", + project_name, + order_id, + provider.name, + notify_url, + return_url or "", + ) + if provider.name == "xorpay": + return { + "status": "ok", + "order_id": order_id, + "xorpay_params": result.get("params", {}), + "xorpay_url": result.get("pay_url", ""), + "provider": provider.name, + } + params = result.get("params", {}) + params_str = {k: str(v) for k, v in params.items() if v is not None and str(v).strip() != ""} + return { + "status": "ok", + "params": params_str, + "pay_url": result.get("pay_url", ""), + "provider": provider.name, + "submit_method": "POST", + } + + @router.post("/payh5/redirect") + async def payh5_redirect(item: dict, request: Request): + """ZPAY mapi 接口，返回 302 或二维码。PC/手机用 zpay，默认微信支付，支付宝暂不提供""" + from fastapi.responses import RedirectResponse, JSONResponse + + provider = get_payment_provider_by_name("zpay") + channel = resolve_channel(item.get("channel")) + _safe_log( + logging.INFO, + "[payjsapi] %s/payh5/redirect user_id=%s type=%s device=%s channel=%s", + project_name, + item.get("user_id", ""), + item.get("type", ""), + item.get("device", "pc"), + channel, + ) + try: + total_fee = float(item.get("total_fee", 80)) + except (TypeError, ValueError): + total_fee = 80.0 + total_fee_yuan = f"{total_fee / 100:.2f}" + user_id = item.get("user_id", "unknown") + pay_type = item.get("type", "unknown").lower() + return_url = item.get("return_url") + device = item.get("device", "pc") + random_suffix = "".join(random.choices(string.hexdigits.lower(), k=8)) + timestamp = int(time.time()) + order_id = f"{user_id}_{pay_type}_order_{timestamp}_{random_suffix}" + if return_url and "order_id=" not in return_url and "out_trade_no=" not in return_url: + sep = "&" if "?" in return_url else "?" + return_url = f"{return_url}{sep}order_id={order_id}" + client_ip = item.get("client_ip") or (request.client.host if request.client else "127.0.0.1") + forwarded = request.headers.get("x-forwarded-for") + if forwarded: + client_ip = forwarded.split(",")[0].strip() + type_map = { + "book": "购买书籍", + "meetup": "数字游民社区报名", + "video": "视频解锁", + "vip": "VIP会员充值", + } + name = type_map.get(pay_type, "商品支付") + notify_url = _notify_path("/zpay_notify") + result = provider.create_order_api( + order_id=order_id, + name=name, + total_fee_yuan=total_fee_yuan, + pay_type=channel, + notify_url=notify_url, + return_url=return_url, + client_ip=client_ip, + device=device, + ) + if result.get("status") != "ok": + raise HTTPException(status_code=500, detail=result.get("msg", "ZPAY mapi 创建订单失败")) + _safe_log( + logging.INFO, + "[payjsapi] %s/payh5/redirect created order_id=%s device=%s channel=%s", + project_name, + order_id, + device, + channel, + ) + payurl = result.get("payurl") + payurl2 = result.get("payurl2") + img = result.get("img") + if device == "pc" and img: + return JSONResponse(status_code=200, content={ + "status": "ok", "qrcode_mode": True, "order_id": order_id, + "img": img, "payurl": payurl, "return_url": return_url, "channel": channel, + }) + if device == "wechat" and channel.lower() in ("wxpay", "wx", "wechat"): + return JSONResponse(status_code=200, content={"use_form": True, "order_id": order_id}) + headers = {"X-Pay-Order-Id": order_id} + if device == "h5" and channel.lower() in ("wxpay", "wx", "wechat") and payurl2: + return RedirectResponse(url=payurl2, status_code=302, headers=headers) + if payurl: + return RedirectResponse(url=payurl, status_code=302, headers=headers) + raise HTTPException(status_code=500, detail="ZPAY 未返回支付链接") + + @router.get("/zpay_order_status") + async def zpay_order_status(out_trade_no: str): + """查询 ZPAY 订单支付状态""" + provider = get_payment_provider() + if not isinstance(provider, ZPayProvider): + raise HTTPException(status_code=400, detail="仅 zpay 渠道支持") + result = provider.query_order_status(out_trade_no, timeout=15) + if result.get("error"): + _safe_log(logging.WARNING, "%s ZPAY 订单查询失败: %s", project_name, result.get("error")) + return result + + @router.get("/order_status") + async def order_status(order_id: str = "", out_trade_no: str = ""): + order_id = (order_id or out_trade_no or "").strip() + if not order_id: + return {"paid": False, "error": "missing order_id"} + + try: + from ..services.pb import get_pb_client + + pb = get_pb_client() + records = pb.collection("payments").get_list( + 1, 1, {"filter": f'order_id = "{order_id}"'} + ) + if records.items: + return {"paid": True, "source": "pocketbase"} + except Exception as error: + _safe_log(logging.WARNING, "%s order_status PocketBase failed: %s", project_name, error) + + try: + xorpay = get_payment_provider_by_name("xorpay") + xorpay_result = xorpay.query_order_status(order_id, timeout=8) + if xorpay_result.get("paid"): + return { + "paid": True, + "source": "xorpay", + "status": xorpay_result.get("status", ""), + } + except Exception as error: + _safe_log(logging.WARNING, "%s order_status XorPay failed: %s", project_name, error) + + try: + zpay = get_payment_provider_by_name("zpay") + if isinstance(zpay, ZPayProvider): + zpay_result = zpay.query_order_status(order_id, timeout=8) + if zpay_result.get("paid"): + return { + "paid": True, + "source": "zpay", + "status": zpay_result.get("status", ""), + } + except Exception as error: + _safe_log(logging.WARNING, "%s order_status ZPAY failed: %s", project_name, error) + + return {"paid": False} + + @router.post("/xorpay_notify") + async def xorpay_notify(request: Request): + """XorPay 异步通知""" + form_data = await request.form() + data = {k: v for k, v in form_data.items()} + _safe_log(logging.INFO, "%s XorPay 异步通知: %s", project_name, data) + order_id = data.get("order_id", "") + if order_id in processed_orders: + return "ok" + provider = get_payment_provider_by_name("xorpay") + if not provider.verify_notify(data): + _safe_log(logging.ERROR, "%s XorPay 验签失败: %s", project_name, data) + raise HTTPException(status_code=400, detail="sign error") + parsed = provider.parse_notify(data) + if parsed.get("trade_status") != "success": + return "ok" + handle_payment_success( + order_id, + parsed.get("pay_price", data.get("pay_price", "")), + f"{project_name}-xorpay", + use_memos=False, + site_id=site_id, + ) + return provider.success_response() + + @router.get("/zpay_notify") + @router.post("/zpay_notify") + async def zpay_notify(request: Request): + """ZPAY 异步通知""" + if request.method == "GET": + data = dict(request.query_params) + else: + form_data = await request.form() + data = {k: v for k, v in form_data.items()} + _safe_log(logging.INFO, "%s ZPAY 异步通知: %s", project_name, data) + order_id = data.get("out_trade_no", "") + if order_id in processed_orders: + return "success" + provider = get_payment_provider_by_name("zpay") + if not provider.verify_notify(data): + _safe_log(logging.ERROR, "%s ZPAY 验签失败: %s", project_name, data) + raise HTTPException(status_code=400, detail="sign error") + parsed = provider.parse_notify(data) + if parsed.get("trade_status") != "success": + return "success" + handle_payment_success( + order_id, + parsed.get("pay_price", data.get("money", "")), + f"{project_name}-zpay", + use_memos=False, + site_id=site_id, + ) + return provider.success_response() + + return router diff --git a/app/routers/cnomadcna.py b/app/routers/cnomadcna.py new file mode 100644 index 0000000..f4bbbeb --- /dev/null +++ b/app/routers/cnomadcna.py @@ -0,0 +1,4 @@ +"""cnomadcna 项目支付接口：/cnomadcna/*""" +from ._digital_base import create_digital_router + +router = create_digital_router(prefix="/cnomadcna", project_name="cnomadcna", site_id="meetup") diff --git a/app/routers/common.py b/app/routers/common.py new file mode 100644 index 0000000..40391bd --- /dev/null +++ b/app/routers/common.py @@ -0,0 +1,65 @@ +"""共享接口：meetup 申请、create_user 等""" +import logging +import time + +from fastapi import APIRouter, HTTPException +from pydantic import BaseModel + +from ..services import get_pb_client +from ..sdk.memos import MemosSDK +from ..config import MEMOS_API, MEMOS_TOKEN + +router = APIRouter(tags=["common"]) + + +@router.post("/submit_meetup_application") +async def submit_meetup_application(item: dict): + """数字游民社区报名表单""" + logging.info("submit_meetup_application item: %s", item) + user_id = item.get("user_id") + if not user_id: + raise HTTPException(status_code=400, detail="missing user_id") + try: + pb_client = get_pb_client() + solan = pb_client.collection("solan") + pb_data = { + "user_id": user_id, + "nickname": item.get("nickname", ""), + "occupation": item.get("occupation", ""), + "reason": item.get("reason", ""), + "wechatId": item.get("wechatId", ""), + "gender": item.get("gender", ""), + "education": item.get("education", ""), + "gradYear": item.get("gradYear", ""), + "phone": item.get("phone", ""), + "calculated_age": item.get("calculated_age", 0), + "type": "", + "order_id": "", + "amount": 0, + } + record = solan.create(pb_data) + logging.info(f"meetup 申请表单提交成功（待支付） - user_id: {user_id}, record_id: {record.id}") + return {"status": "ok", "record_id": record.id} + except Exception as e: + logging.error("meetup 申请提交失败: %s", e) + logging.error(f"meetup 申请提交失败 - user_id: {user_id}, error: {e}") + raise HTTPException(status_code=500, detail="submit failed") + + +class CreateUserRequest(BaseModel): + username: str | None = None + password: str | None = None + + +@router.post("/create_user") +async def create_user(item: CreateUserRequest): + """创建 Memos 用户（基于 MemosSDK）""" + sdk = MemosSDK(api_url=MEMOS_API, token=MEMOS_TOKEN, enabled=bool(MEMOS_API and MEMOS_TOKEN)) + if not sdk.enabled: + return {"error": "Memos 未配置"} + timenew = str(int(time.time())) + username = item.username or f"user{timenew}" + password = item.password or "12345678" + result = sdk.create_user_by_username(username=username, password=password) + logging.info("memos_result- %s", result) + return result diff --git a/app/routers/digital.py b/app/routers/digital.py new file mode 100644 index 0000000..56de4c0 --- /dev/null +++ b/app/routers/digital.py @@ -0,0 +1,4 @@ +"""digital 项目支付接口：/digital/*""" +from ._digital_base import create_digital_router + +router = create_digital_router(prefix="/digital", project_name="digital", site_id="digital") diff --git a/app/routers/legacy.py b/app/routers/legacy.py new file mode 100644 index 0000000..5134a6d --- /dev/null +++ b/app/routers/legacy.py @@ -0,0 +1,4 @@ +"""向后兼容：/payh5、/zpay_notify 等（digital/cnomadcna/nomadlms 旧版调用）""" +from ._digital_base import create_digital_router + +router = create_digital_router(prefix="", project_name="legacy") diff --git a/app/routers/meetup.py b/app/routers/meetup.py new file mode 100644 index 0000000..6bd2aba --- /dev/null +++ b/app/routers/meetup.py @@ -0,0 +1,328 @@ +""" +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 位 + +# Admin token 缓存，避免每次请求都向 PocketBase 认证（PB 远程时易超时） +_ADMIN_TOKEN_CACHE: tuple[str, float] | None = None +_TOKEN_CACHE_TTL = 300 # 5 分钟 + + +def _get_admin_token() -> tuple[str | None, str | None]: + """返回 (token, error_detail)。token 为 None 时 error_detail 为失败原因。带 5 分钟缓存。""" + global _ADMIN_TOKEN_CACHE + now = time.time() + if _ADMIN_TOKEN_CACHE and _ADMIN_TOKEN_CACHE[1] > now: + return _ADMIN_TOKEN_CACHE[0], None + + 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=4, + ) + if r.status_code == 200: + data = r.json() + token = data.get("token") + if token: + _ADMIN_TOKEN_CACHE = (token, now + _TOKEN_CACHE_TTL) + return 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", token: str | None = None) -> bool: + """检查用户是否有有效 site_vip（未过期）。可传入 token 避免重复认证。""" + if not token: + 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=4, + ) + 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=4, + ) + 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", token) 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} diff --git a/app/routers/nomadlms.py b/app/routers/nomadlms.py new file mode 100644 index 0000000..5530aa4 --- /dev/null +++ b/app/routers/nomadlms.py @@ -0,0 +1,4 @@ +"""nomadlms 项目支付接口：/nomadlms/*""" +from ._digital_base import create_digital_router + +router = create_digital_router(prefix="/nomadlms", project_name="nomadlms") diff --git a/app/routers/nomadvip.py b/app/routers/nomadvip.py new file mode 100644 index 0000000..406b3d8 --- /dev/null +++ b/app/routers/nomadvip.py @@ -0,0 +1,883 @@ +"""nomadvip 项目支付接口：/nomadvip/*，支持 xorpay / zpay。""" + +import logging +import random +import string +import time + +import requests +from fastapi import APIRouter, HTTPException, Request +from fastapi.responses import JSONResponse, RedirectResponse + +from ..config import BASE_URL, PB_ADMIN_EMAIL, PB_ADMIN_PASSWORD, PB_URL +from ..payment import ( + get_payment_provider, + get_payment_provider_by_name, + resolve_provider, + resolve_channel, +) +from ..payment.zpay import ZPayProvider +from ..services import handle_payment_success, processed_orders + +router = APIRouter(prefix="/nomadvip", tags=["nomadvip"]) + + +def _get_notify_path(provider_name: str) -> str: + if provider_name == "xorpay": + return f"{BASE_URL.rstrip('/')}/nomadvip/xorpay_notify" + return f"{BASE_URL.rstrip('/')}/nomadvip/zpay_notify" + + +@router.post("/payh5") +async def nomadvip_payh5(item: dict, request: Request): + """创建支付订单。微信内用 xorpay，PC/手机用 zpay，默认微信支付，支付宝暂不提供""" + req_provider = (item.get("provider") or "").strip().lower() or None + device = (item.get("device") or "pc").lower() + ua = (request.headers.get("user-agent") or "").strip() + provider = resolve_provider(device, req_provider, ua) + channel = resolve_channel(item.get("channel")) + logging.info( + "[payjsapi] payh5 user_id=%s type=%s provider=%s channel=%s", + item.get("user_id", ""), + item.get("type", ""), + provider.name, + channel, + ) + + total_fee = int(item.get("total_fee", 880)) + total_fee_yuan = f"{total_fee / 100:.2f}" + user_id = item.get("user_id", "unknown") + pay_type = (item.get("type") or "vip").lower() + random_suffix = "".join(random.choices(string.hexdigits.lower(), k=8)) + timestamp = int(time.time()) + order_id = f"{user_id}_{pay_type}_order_{timestamp}_{random_suffix}" + + type_map = { + "book": "购买书籍", + "meetup": "数字游民社区报名", + "video": "视频解锁", + "vip": "VIP会员充值", + } + name = type_map.get(pay_type, "VIP会员充值") + notify_url = _get_notify_path(provider.name) + result = provider.create_order( + order_id=order_id, + name=name, + total_fee_yuan=total_fee_yuan, + pay_type=channel, + notify_url=notify_url, + return_url=item.get("return_url"), + client_ip=item.get("client_ip"), + ) + logging.info( + "[payjsapi] payh5 created order_id=%s provider=%s notify_url=%s return_url=%s", + order_id, + provider.name, + notify_url, + item.get("return_url") or "", + ) + + if provider.name == "xorpay": + return { + "status": "ok", + "order_id": order_id, + "xorpay_params": result.get("params", {}), + "xorpay_url": result.get("pay_url", ""), + "provider": provider.name, + } + + params = result.get("params", {}) + params_str = { + k: str(v) for k, v in params.items() if v is not None and str(v).strip() != "" + } + return { + "status": "ok", + "params": params_str, + "pay_url": result.get("pay_url", ""), + "provider": provider.name, + "submit_method": "POST", + } + + +@router.post("/payh5/redirect") +async def nomadvip_payh5_redirect(item: dict, request: Request): + """ZPAY mapi 接口，PC/手机用 zpay，默认微信支付，支付宝暂不提供""" + provider = get_payment_provider_by_name("zpay") + channel = resolve_channel(item.get("channel")) + + logging.info("nomadvip payh5/redirect request: item=%s", item) + try: + total_fee = float(item.get("total_fee", 880)) + except (TypeError, ValueError): + logging.warning( + "nomadvip payh5/redirect invalid total_fee=%s", item.get("total_fee") + ) + total_fee = 880.0 + + total_fee_yuan = f"{total_fee / 100:.2f}" + user_id = item.get("user_id", "unknown") + pay_type = (item.get("type") or "vip").lower() + return_url = item.get("return_url") + device = item.get("device", "pc") + client_ip = item.get("client_ip") or ( + request.client.host if request.client else "127.0.0.1" + ) + forwarded = request.headers.get("x-forwarded-for") + if forwarded: + client_ip = forwarded.split(",")[0].strip() + + random_suffix = "".join(random.choices(string.hexdigits.lower(), k=8)) + timestamp = int(time.time()) + order_id = f"{user_id}_{pay_type}_order_{timestamp}_{random_suffix}" + + # 追加 order_id，供 paid 页在 cookie 丢失时恢复 user_id。 + if return_url and "order_id=" not in return_url and "out_trade_no=" not in return_url: + sep = "&" if "?" in return_url else "?" + return_url = f"{return_url}{sep}order_id={order_id}" + + type_map = { + "book": "购买书籍", + "meetup": "数字游民社区报名", + "video": "视频解锁", + "vip": "VIP会员充值", + } + name = type_map.get(pay_type, "VIP会员充值") + notify_url = f"{BASE_URL.rstrip('/')}/nomadvip/zpay_notify" + + result = provider.create_order_api( + order_id=order_id, + name=name, + total_fee_yuan=total_fee_yuan, + pay_type=channel, + notify_url=notify_url, + return_url=return_url, + client_ip=client_ip, + device=device, + ) + if result.get("status") != "ok": + err_msg = result.get("msg", "ZPAY mapi 创建订单失败") + logging.error( + "nomadvip payh5/redirect ZPAY create failed: %s result=%s", + err_msg, + result, + ) + raise HTTPException(status_code=500, detail=err_msg) + + payurl = result.get("payurl") + payurl2 = result.get("payurl2") + img = result.get("img") + if device == "pc" and img: + return JSONResponse( + status_code=200, + content={ + "status": "ok", + "qrcode_mode": True, + "order_id": order_id, + "img": img, + "payurl": payurl, + "return_url": return_url, + "channel": channel, + }, + ) + if device == "wechat" and channel.lower() in ("wxpay", "wx", "wechat"): + return JSONResponse( + status_code=200, content={"use_form": True, "order_id": order_id} + ) + + headers = {"X-Pay-Order-Id": order_id} + if device == "h5" and channel.lower() in ("wxpay", "wx", "wechat") and payurl2: + return RedirectResponse(url=payurl2, status_code=302, headers=headers) + if payurl: + return RedirectResponse(url=payurl, status_code=302, headers=headers) + + logging.error( + "nomadvip payh5/redirect ZPAY missing pay url device=%s channel=%s keys=%s", + device, + channel, + list(result.keys()), + ) + raise HTTPException(status_code=500, detail="ZPAY 未返回支付链接") + + +@router.get("/zpay_order_status") +async def nomadvip_zpay_order_status(out_trade_no: str): + """查询 ZPAY 订单支付状态。""" + provider = get_payment_provider() + if not isinstance(provider, ZPayProvider): + raise HTTPException(status_code=400, detail="zpay_order_status 仅支持 zpay") + result = provider.query_order_status(out_trade_no, timeout=15) + if result.get("error"): + logging.warning( + "nomadvip ZPAY order status failed: %s", result.get("error") + ) + return result + + +def _get_linked_email_for_user(pb, user_id: str) -> str | None: + """按 user_id 查询 vip_email_link 关联邮箱。""" + try: + link_rec = pb.collection("vip_email_link").get_list( + 1, 1, {"filter": f'anonymous_user_id = "{user_id}"'} + ) + if link_rec.items: + return link_rec.items[0].get("email") or "" + + if user_id and user_id.startswith("user") and user_id[4:].isdigit(): + link_rec = pb.collection("vip_email_link").get_list( + 1, 1, {"filter": f'user_id = "{user_id}"'} + ) + if link_rec.items: + return link_rec.items[0].get("email") or "" + except Exception: + pass + return None + + +def _uid_has_vip(pb, uid: str, site_id: str = "vip") -> bool: + if not uid: + return False + + sv = pb.collection("site_vip").get_list( + 1, 1, {"filter": f'user_id = "{uid}" && site_id = "{site_id}"'} + ) + if sv.items: + rec = sv.items[0] + expires = rec.get("expires_at") or "" + if not expires: + return True + + from datetime import datetime + + try: + exp_dt = datetime.fromisoformat(expires.replace("Z", "+00:00")) + now = datetime.now(exp_dt.tzinfo) if exp_dt.tzinfo else datetime.now() + return exp_dt.timestamp() >= now.timestamp() + except Exception: + return True + + pm = pb.collection("payments").get_list( + 1, + 1, + {"filter": f'user_id = "{uid}" && type = "vip"', "sort": "-created"}, + ) + return bool(pm.items) + + +@router.get("/check_vip") +async def nomadvip_check_vip(user_id: str = ""): + """Return vip status for a user_id.""" + user_id = (user_id or "").strip() + logging.info("[payjsapi] check_vip user_id=%s", user_id or "(empty)") + if not user_id: + return {"vip": False, "error": "missing user_id"} + + try: + from ..services.pb import get_pb_client + + pb = get_pb_client() + if not _uid_has_vip(pb, user_id): + logging.info("[payjsapi] check_vip miss user_id=%s vip=False", user_id) + return {"vip": False} + + email = _get_linked_email_for_user(pb, user_id) + if not email: + admin_token = _get_pb_admin_token() + if admin_token and user_id.startswith("user") and user_id[4:].isdigit(): + linked_member = _pb_find_linked_member_from_order(admin_token, user_id) + if linked_member: + linked_user_id, linked_email = linked_member + email = linked_email + _pb_upsert_email_link(admin_token, linked_email, linked_user_id, user_id) + + logging.info( + "[payjsapi] check_vip hit user_id=%s vip=True email=%s", + user_id, + email or "(none)", + ) + return {"vip": True, "email": email} + except Exception as error: + logging.warning("[payjsapi] check_vip failed: %s", error) + return {"vip": False, "error": str(error)} + +def _get_pb_admin_token() -> 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: + response = requests.post( + f"{base}{path}", + json={"identity": PB_ADMIN_EMAIL, "password": PB_ADMIN_PASSWORD}, + timeout=10, + ) + if response.status_code == 200: + return response.json().get("token") + except Exception: + pass + return None + + +def _pb_escape(value: str) -> str: + return str(value).replace("\\", "\\\\").replace('"', '\\"') + + +def _pb_list_email_links(admin_token: str, email: str) -> list[dict]: + base = (PB_URL or "").rstrip("/") + headers = {"Authorization": f"Bearer {admin_token}"} + + for candidate in [email, email.lower()]: + safe_email = _pb_escape(candidate) + response = requests.get( + f"{base}/api/collections/vip_email_link/records", + params={"filter": f'email = "{safe_email}"', "perPage": 1}, + headers=headers, + timeout=10, + ) + if response.status_code != 200: + continue + + items = (response.json() or {}).get("items") or [] + if items: + return items + + return [] + + +def _pb_upsert_email_link( + admin_token: str, + email: str, + pb_user_id: str, + anonymous_user_id: str | None = None, +): + base = (PB_URL or "").rstrip("/") + headers = { + "Authorization": f"Bearer {admin_token}", + "Content-Type": "application/json", + } + items = _pb_list_email_links(admin_token, email) + payload = {"email": email.lower(), "user_id": pb_user_id} + if anonymous_user_id: + payload["anonymous_user_id"] = anonymous_user_id + + if items: + record_id = items[0].get("id") + if record_id: + requests.patch( + f"{base}/api/collections/vip_email_link/records/{record_id}", + json=payload, + headers=headers, + timeout=10, + ) + return + + requests.post( + f"{base}/api/collections/vip_email_link/records", + json=payload, + headers=headers, + timeout=10, + ) + + +def _pb_move_site_vip( + admin_token: str, + from_user_id: str, + to_user_id: str, + site_id: str = "vip", +): + if not from_user_id or not to_user_id or from_user_id == to_user_id: + return + + base = (PB_URL or "").rstrip("/") + safe_from_user_id = _pb_escape(from_user_id) + safe_site_id = _pb_escape(site_id) + response = requests.get( + f"{base}/api/collections/site_vip/records", + params={ + "filter": f'user_id = "{safe_from_user_id}" && site_id = "{safe_site_id}"', + "perPage": 500, + }, + headers={"Authorization": f"Bearer {admin_token}"}, + timeout=10, + ) + if response.status_code != 200: + return + + headers = { + "Authorization": f"Bearer {admin_token}", + "Content-Type": "application/json", + } + items = (response.json() or {}).get("items") or [] + for item in items: + record_id = item.get("id") + if not record_id: + continue + requests.patch( + f"{base}/api/collections/site_vip/records/{record_id}", + json={"user_id": to_user_id}, + headers=headers, + timeout=10, + ) + + +def _pb_latest_vip_payment(admin_token: str, user_id: str) -> dict | None: + base = (PB_URL or "").rstrip("/") + safe_user_id = _pb_escape(user_id) + response = requests.get( + f"{base}/api/collections/payments/records", + params={ + "filter": f'user_id = "{safe_user_id}" && type = "vip"', + "perPage": 1, + "sort": "-created", + }, + headers={"Authorization": f"Bearer {admin_token}"}, + timeout=10, + ) + if response.status_code != 200: + return None + + items = (response.json() or {}).get("items") or [] + return items[0] if items else None + + +def _pb_list_site_vip_by_order( + admin_token: str, + order_id: str, + site_id: str = "vip", +) -> list[dict]: + base = (PB_URL or "").rstrip("/") + safe_order_id = _pb_escape(order_id) + safe_site_id = _pb_escape(site_id) + response = requests.get( + f"{base}/api/collections/site_vip/records", + params={ + "filter": f'order_id = "{safe_order_id}" && site_id = "{safe_site_id}"', + "perPage": 500, + "sort": "-created", + }, + headers={"Authorization": f"Bearer {admin_token}"}, + timeout=10, + ) + if response.status_code != 200: + return [] + + return (response.json() or {}).get("items") or [] + + +def _pb_get_user_email(admin_token: str, user_id: str) -> str | None: + base = (PB_URL or "").rstrip("/") + response = requests.get( + f"{base}/api/collections/users/records/{user_id}", + headers={"Authorization": f"Bearer {admin_token}"}, + timeout=10, + ) + if response.status_code != 200: + return None + + email = (response.json() or {}).get("email") or "" + return email.strip().lower() or None + + +def _pb_find_linked_member_from_order( + admin_token: str, + anonymous_user_id: str, +) -> tuple[str, str] | None: + base = (PB_URL or "").rstrip("/") + safe_user_id = _pb_escape(anonymous_user_id) + response = requests.get( + f"{base}/api/collections/payments/records", + params={ + "filter": f'user_id = "{safe_user_id}" && type = "vip"', + "perPage": 20, + "sort": "-created", + }, + headers={"Authorization": f"Bearer {admin_token}"}, + timeout=10, + ) + if response.status_code != 200: + return None + + items = (response.json() or {}).get("items") or [] + seen_orders = set() + for item in items: + order_id = (item.get("order_id") or "").strip() + if not order_id or order_id in seen_orders: + continue + seen_orders.add(order_id) + + for site_vip in _pb_list_site_vip_by_order(admin_token, order_id): + linked_user_id = (site_vip.get("user_id") or "").strip() + if ( + not linked_user_id + or linked_user_id == anonymous_user_id + or (linked_user_id.startswith("user") and linked_user_id[4:].isdigit()) + ): + continue + + email = _pb_get_user_email(admin_token, linked_user_id) + if email: + return linked_user_id, email + + return None + + +def _pb_ensure_site_vip( + admin_token: str, + from_user_id: str, + to_user_id: str, + site_id: str = "vip", +): + if not from_user_id or not to_user_id or from_user_id == to_user_id: + return + + base = (PB_URL or "").rstrip("/") + safe_from_user_id = _pb_escape(from_user_id) + safe_to_user_id = _pb_escape(to_user_id) + safe_site_id = _pb_escape(site_id) + source_records = requests.get( + f"{base}/api/collections/site_vip/records", + params={ + "filter": f'user_id = "{safe_from_user_id}" && site_id = "{safe_site_id}"', + "perPage": 500, + "sort": "-created", + }, + headers={"Authorization": f"Bearer {admin_token}"}, + timeout=10, + ) + to_records = requests.get( + f"{base}/api/collections/site_vip/records", + params={ + "filter": f'user_id = "{safe_to_user_id}" && site_id = "{safe_site_id}"', + "perPage": 500, + "sort": "-created", + }, + headers={"Authorization": f"Bearer {admin_token}"}, + timeout=10, + ) + payment = _pb_latest_vip_payment(admin_token, from_user_id) + order_id = (payment or {}).get("order_id") or "" + order_records = ( + _pb_list_site_vip_by_order(admin_token, order_id, site_id=site_id) + if order_id + else [] + ) + source_items = (source_records.json() or {}).get("items") or [] if source_records.status_code == 200 else [] + target_items = (to_records.json() or {}).get("items") or [] if to_records.status_code == 200 else [] + canonical = (target_items or source_items or order_records or [None])[0] + headers = { + "Authorization": f"Bearer {admin_token}", + "Content-Type": "application/json", + } + + if canonical and canonical.get("id"): + patch_body = {"user_id": to_user_id, "site_id": site_id} + next_order_id = order_id or (canonical.get("order_id") or "").strip() + if next_order_id: + patch_body["order_id"] = next_order_id + requests.patch( + f"{base}/api/collections/site_vip/records/{canonical['id']}", + json=patch_body, + headers=headers, + timeout=10, + ) + + duplicate_ids = set() + for record in [*target_items, *source_items, *order_records]: + record_id = record.get("id") + if record_id and record_id != canonical["id"]: + duplicate_ids.add(record_id) + for record_id in duplicate_ids: + requests.delete( + f"{base}/api/collections/site_vip/records/{record_id}", + headers={"Authorization": f"Bearer {admin_token}"}, + timeout=10, + ) + return + + if not order_id: + return + + requests.post( + f"{base}/api/collections/site_vip/records", + json={"user_id": to_user_id, "site_id": site_id, "order_id": order_id}, + headers=headers, + timeout=10, + ) + + +@router.post("/check_vip_by_email") +async def nomadvip_check_vip_by_email(item: dict): + """Recover VIP by email/password, optionally with anonymous user_id.""" + email = (item.get("email") or "").strip().lower() + password = (item.get("password") or "").strip() + anonymous_user_id = ( + item.get("user_id") or item.get("anonymousUserId") or "" + ).strip() + logging.info( + "[payjsapi] check_vip_by_email email=%s user_id=%s", + email or "(empty)", + anonymous_user_id or "(empty)", + ) + + if not email or not password: + return {"vip": False, "error": "missing email or password"} + + base = (PB_URL or "").rstrip("/") + login_res = requests.post( + f"{base}/api/collections/users/auth-with-password", + json={"identity": email, "password": password}, + timeout=10, + ) + if login_res.status_code != 200: + try: + error = login_res.json() or {} + except Exception: + error = {} + logging.info( + "[payjsapi] check_vip_by_email login failed status=%s message=%s", + login_res.status_code, + error.get("message", ""), + ) + return { + "vip": False, + "error": error.get("message") or "invalid email or password", + } + + login_data = login_res.json() or {} + token = login_data.get("token") + record = login_data.get("record") or {} + pb_user_id = (record.get("id") or "").strip() + if not token or not pb_user_id: + return {"vip": False, "error": "login failed"} + + admin_token = _get_pb_admin_token() + if not admin_token: + return {"vip": False, "error": "PocketBase admin auth failed"} + + try: + from ..services.pb import get_pb_client + + pb = get_pb_client() + matched_user_id = None + recovered_anonymous_user_id = ( + anonymous_user_id + if anonymous_user_id.startswith("user") and anonymous_user_id[4:].isdigit() + else None + ) + recovery_source = None + + if recovered_anonymous_user_id and _uid_has_vip(pb, recovered_anonymous_user_id): + matched_user_id = recovered_anonymous_user_id + recovery_source = "request_user_id" + else: + items = _pb_list_email_links(admin_token, email) + logging.info( + "[payjsapi] check_vip_by_email vip_email_link email=%s count=%s", + email, + len(items), + ) + for link_record in items: + linked_ids = [] + linked_pb_user_id = (link_record.get("user_id") or "").strip() + linked_anonymous_user_id = ( + link_record.get("anonymous_user_id") or "" + ).strip() + if linked_pb_user_id: + linked_ids.append(linked_pb_user_id) + if linked_anonymous_user_id and linked_anonymous_user_id not in linked_ids: + linked_ids.append(linked_anonymous_user_id) + + for linked_uid in linked_ids: + if not _uid_has_vip(pb, linked_uid): + continue + matched_user_id = linked_uid + if linked_anonymous_user_id.startswith("user") and linked_anonymous_user_id[4:].isdigit(): + recovered_anonymous_user_id = linked_anonymous_user_id + elif linked_uid.startswith("user") and linked_uid[4:].isdigit(): + recovered_anonymous_user_id = linked_uid + recovery_source = "vip_email_link" + break + + if matched_user_id: + break + + if not matched_user_id and _uid_has_vip(pb, pb_user_id): + matched_user_id = pb_user_id + recovered_anonymous_user_id = None + recovery_source = "pb_user_id" + + if not matched_user_id: + logging.info("[payjsapi] check_vip_by_email no vip match for email=%s", email) + return { + "vip": False, + "error": "no linked vip record found; provide the paid user_id if this is a historical anonymous order", + } + + if recovered_anonymous_user_id: + _pb_ensure_site_vip(admin_token, recovered_anonymous_user_id, pb_user_id) + _pb_upsert_email_link( + admin_token, + email, + pb_user_id, + recovered_anonymous_user_id, + ) + else: + _pb_upsert_email_link(admin_token, email, pb_user_id) + + logging.info( + "[payjsapi] check_vip_by_email recovered by %s matched_user_id=%s effectiveUserId=%s anonymousUserId=%s", + recovery_source or "unknown", + matched_user_id, + pb_user_id, + recovered_anonymous_user_id or "(none)", + ) + return { + "vip": True, + "token": token, + "record": record, + "effectiveUserId": pb_user_id, + "anonymousUserId": recovered_anonymous_user_id, + } + except Exception as error: + logging.warning("[payjsapi] check_vip_by_email failed: %s", error) + return {"vip": False, "error": str(error)} + +@router.get("/order_status") +async def nomadvip_order_status(order_id: str = "", out_trade_no: str = ""): + """查询订单支付状态，优先查 ZPAY，失败时回退 PocketBase。""" + order_id = (order_id or out_trade_no or "").strip() + logging.info("[payjsapi] order_status order_id=%s", order_id or "(空)") + if not order_id: + return {"paid": False, "error": "缺少 order_id 或 out_trade_no"} + + try: + from ..services.pb import get_pb_client + + pb = get_pb_client() + records = pb.collection("payments").get_list( + 1, 1, {"filter": f'order_id = "{order_id}"'} + ) + paid = len(records.items) > 0 + logging.info("[payjsapi] order_status PocketBase paid=%s", paid) + if paid: + return {"paid": True, "source": "pocketbase"} + except Exception as error: + logging.warning("[payjsapi] order_status PocketBase failed: %s", error) + + xorpay = get_payment_provider_by_name("xorpay") + xorpay_result = xorpay.query_order_status(order_id, timeout=8) + if xorpay_result.get("error"): + logging.warning( + "[payjsapi] order_status XorPay query failed order_id=%s error=%s", + order_id, + xorpay_result.get("error"), + ) + else: + logging.info( + "[payjsapi] order_status XorPay status=%s paid=%s", + xorpay_result.get("status", ""), + xorpay_result.get("paid", False), + ) + if xorpay_result.get("paid"): + return { + "paid": True, + "source": "xorpay", + "status": xorpay_result.get("status"), + "pay_price": xorpay_result.get("pay_price") or "", + } + + zpay = get_payment_provider_by_name("zpay") + if isinstance(zpay, ZPayProvider): + result = zpay.query_order_status(order_id, timeout=8) + if not result.get("error") and result.get("paid"): + logging.info("[payjsapi] order_status ZPAY paid=True") + return result + + return {"paid": False} + + +@router.post("/xorpay_notify") +async def nomadvip_xorpay_notify(request: Request): + """XorPay 异步通知。""" + form_data = await request.form() + data = {k: v for k, v in form_data.items()} + logging.info("nomadvip XorPay notify: %s", data) + order_id = data.get("order_id", "") + if order_id in processed_orders: + logging.info("nomadvip order already processed, skip: %s", order_id) + return "ok" + + provider = get_payment_provider_by_name("xorpay") + if not provider.verify_notify(data): + logging.error("nomadvip XorPay verify failed: %s", data) + raise HTTPException(status_code=400, detail="sign error") + + parsed = provider.parse_notify(data) + if parsed.get("trade_status") != "success": + return "ok" + + pay_price = parsed.get("pay_price", data.get("pay_price", "")) + try: + handle_payment_success( + order_id, + pay_price, + "nomadvip-xorpay", + use_memos=True, + site_id="vip", + ) + logging.info("nomadvip business done: order_id=%s", order_id) + except Exception as error: + logging.error( + "nomadvip PocketBase/Memos failed: %s", error, exc_info=True + ) + raise + return provider.success_response() + + +@router.get("/zpay_notify") +@router.post("/zpay_notify") +async def nomadvip_zpay_notify(request: Request): + """ZPAY 异步通知。""" + provider = get_payment_provider_by_name("zpay") + if request.method == "GET": + data = dict(request.query_params) + else: + form_data = await request.form() + data = {k: v for k, v in form_data.items()} + + logging.info("nomadvip ZPAY notify: %s", data) + order_id = data.get("out_trade_no", "") + if order_id in processed_orders: + logging.info("nomadvip order already processed, skip: %s", order_id) + return "success" + + if not provider.verify_notify(data): + logging.error("nomadvip ZPAY verify failed: %s", data) + raise HTTPException(status_code=400, detail="sign error") + + parsed = provider.parse_notify(data) + trade_status = parsed.get("trade_status") or data.get("trade_status", "") + if trade_status not in ("success", "TRADE_SUCCESS"): + logging.info("nomadvip non-success trade status, skip: %s", trade_status) + return "success" + + pay_price = parsed.get("pay_price") or data.get("money", "") + try: + handle_payment_success( + order_id, + pay_price, + "nomadvip-zpay", + use_memos=True, + site_id="vip", + ) + logging.info("nomadvip business done: order_id=%s", order_id) + except Exception as error: + logging.error( + "nomadvip PocketBase/Memos failed: %s", error, exc_info=True + ) + raise + return provider.success_response() diff --git a/app/routers/salon.py b/app/routers/salon.py new file mode 100644 index 0000000..2c7d95e --- /dev/null +++ b/app/routers/salon.py @@ -0,0 +1,4 @@ +"""salon 项目接口：克隆 cnomadcna 的支付与 meetup 流程，使用 site_id=salon""" +from ._digital_base import create_digital_router + +router = create_digital_router(prefix="/salon", project_name="salon", site_id="salon") diff --git a/app/routers/salon_meetup.py b/app/routers/salon_meetup.py new file mode 100644 index 0000000..22c3d62 --- /dev/null +++ b/app/routers/salon_meetup.py @@ -0,0 +1,371 @@ +""" +salon 加入流程：check-user、ensure-user、complete-order +克隆 meetup 逻辑，使用 site_id=salon +""" +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 + +SITE_ID = "salon" +router = APIRouter(prefix="/api/salon", tags=["salon"]) +DEFAULT_PASSWORD = "12345678" + +_SALON_ADMIN_TOKEN_CACHE: tuple[str, float] | None = None +_TOKEN_CACHE_TTL = 300 + + +def _get_admin_token() -> tuple[str | None, str | None]: + global _SALON_ADMIN_TOKEN_CACHE + now = time.time() + if _SALON_ADMIN_TOKEN_CACHE and _SALON_ADMIN_TOKEN_CACHE[1] > now: + return _SALON_ADMIN_TOKEN_CACHE[0], None + + if not PB_ADMIN_EMAIL or not PB_ADMIN_PASSWORD: + return None, "PB_ADMIN_EMAIL 或 PB_ADMIN_PASSWORD 未配置" + 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=4, + ) + if r.status_code == 200: + data = r.json() + token = data.get("token") + if token: + _SALON_ADMIN_TOKEN_CACHE = (token, now + _TOKEN_CACHE_TTL) + return 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, token: str | None = None) -> bool: + if not token: + token, _ = _get_admin_token() + if not token: + return False + try: + 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=4, + ) + 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): + 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=4, + ) + 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, token) 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"salon check-user error: {e}") + raise HTTPException(status_code=500, detail="操作失败") + + +@router.post("/ensure-user") +async def ensure_user(item: EnsureUserRequest): + 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"salon ensure-user create failed: status={create_r.status_code}, msg={msg}") + 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"salon 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: + 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"salon complete-order zpay query failed: {e}") + return False + + +@router.post("/complete-order") +async def complete_order(item: dict): + 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 != "salon": + raise HTTPException(status_code=400, detail="订单格式无效") + + 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}" && site_id = "{SITE_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"salon complete-order 未找到订单: order_id={order_id}, attempt={attempt + 1}, pb_resp={resp_preview}") + raise HTTPException(status_code=400, detail="订单不存在或未支付成功") + + 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, + } + + return {"ok": True, "token": None, "record": None, "is_new": False} + + +@router.get("/vip/check") +async def vip_check(user_id: str = ""): + """检查用户是否为 salon VIP""" + if not user_id: + return {"vip": False} + token, _ = _get_admin_token() + return {"vip": _check_site_vip(user_id, token)} + + +@router.get("/join/status") +async def join_status(user_id: str = ""): + """检查用户是否已提交 salon 加入申请""" + if not user_id: + return {"hasRecord": False, "isComplete": False} + try: + from ..services import get_pb_client + pb_client = get_pb_client() + solan = pb_client.collection("solanRed") + existing = solan.get_list(1, 1, {"filter": f'user_id = "{user_id}"', "sort": "-created"}) + has_record = len(existing.items) > 0 + rec = existing.items[0] if existing.items else None + is_complete = bool(rec and rec.get("order_id") and rec.get("amount", 0) > 0) + return {"hasRecord": has_record, "isComplete": is_complete} + except Exception: + return {"hasRecord": False, "isComplete": False} + + +@router.post("/join") +async def submit_join(item: dict): + """salon 加入社区报名表单，写入 solanRed 集合""" + user_id = item.get("user_id") + if not user_id: + raise HTTPException(status_code=400, detail="missing user_id") + + try: + from ..services import get_pb_client + pb_client = get_pb_client() + solan = pb_client.collection("solanRed") + pb_data = { + "user_id": user_id, + "nickname": item.get("nickname", ""), + "occupation": item.get("occupation", ""), + "reason": item.get("reason", ""), + "wechatId": item.get("wechatId", ""), + "gender": item.get("gender", ""), + "education": item.get("education", ""), + "gradYear": item.get("gradYear", ""), + "phone": item.get("phone", ""), + "single": item.get("single", ""), + "media": item.get("media", ""), + "calculated_age": item.get("calculated_age", 0), + "type": "", + "order_id": "", + "amount": 0, + } + record = solan.create(pb_data) + logging.info(f"salon 申请表单提交成功 - user_id: {user_id}, record_id: {record.id}") + return {"ok": True, "user_id": user_id, "record_id": record.id} + except Exception as e: + logging.error(f"salon join 提交失败: {e}") + raise HTTPException(status_code=500, detail="submit failed") diff --git a/app/routers/wxh5.py b/app/routers/wxh5.py new file mode 100644 index 0000000..08526dd --- /dev/null +++ b/app/routers/wxh5.py @@ -0,0 +1,274 @@ +"""wxH5 项目支付接口：/wxh5/*，支持 xorpay/zpay，支付成功后同步 Supabase wxgzh_vip_users""" +import logging +import random +import string +import time +from datetime import datetime, timedelta + +from fastapi import APIRouter, Request, HTTPException +from fastapi.responses import RedirectResponse, JSONResponse + +from ..config import BASE_URL +from ..payment import ( + get_payment_provider, + get_payment_provider_by_name, + resolve_provider, + resolve_channel, +) +from ..payment.zpay import ZPayProvider +from ..services import handle_payment_success, processed_orders + +router = APIRouter(prefix="/wxh5", tags=["wxh5"]) + +# 套餐金额（分）与有效期映射 +PLAN_MAP = { + "month": (2900, 30), # 月卡 29 元，30 天 + "year": (19900, 365), # 年卡 199 元，365 天 + "lifetime": (39900, 36500), # 终身 399 元，100 年 +} + + +def _infer_plan_from_amount(amount_cents: int) -> str: + """根据支付金额（分）推断套餐""" + if amount_cents >= 39900: + return "lifetime" + if amount_cents >= 19900: + return "year" + return "month" + + +def _get_notify_path(provider_name: str) -> str: + base = BASE_URL.rstrip("/") + return f"{base}/wxh5/xorpay_notify" if provider_name == "xorpay" else f"{base}/wxh5/zpay_notify" + + +def _sync_wxh5_vip_to_supabase(user_id: str, plan: str): + """支付成功后同步 VIP 到 Supabase wxgzh_vip_users""" + try: + from supabase import create_client + from ..config import SUPABASE_URL, SUPABASE_SERVICE_KEY + url = SUPABASE_URL + key = SUPABASE_SERVICE_KEY + if not url or not key: + logging.warning("wxh5: SUPABASE_URL/SUPABASE_SERVICE_KEY 未配置，跳过 Supabase 同步") + return + days = PLAN_MAP.get(plan, PLAN_MAP["year"])[1] + expire_at = (datetime.utcnow() + timedelta(days=days)).strftime("%Y-%m-%dT%H:%M:%S+00:00") + client = create_client(url, key) + # upsert: 存在则更新 expire_at，不存在则插入 + client.table("wxgzh_vip_users").upsert( + {"user_id": user_id, "expire_at": expire_at}, + on_conflict="user_id", + ).execute() + logging.info(f"wxh5 Supabase VIP 同步成功: user_id={user_id}, plan={plan}") + except Exception as e: + logging.error(f"wxh5 Supabase 同步失败: {e}", exc_info=True) + + +@router.post("/payh5") +async def wxh5_payh5(item: dict, request: Request): + """wxH5 专用：创建支付订单。微信内用 xorpay，PC/手机用 zpay，默认微信支付，支付宝暂不提供""" + req_provider = (item.get("provider") or "").strip().lower() or None + device = (item.get("device") or "pc").lower() + ua = (request.headers.get("user-agent") or "").strip() + provider = resolve_provider(device, req_provider, ua) + channel = resolve_channel(item.get("channel")) + plan = (item.get("plan") or "year").lower() + total_fee = PLAN_MAP.get(plan, PLAN_MAP["year"])[0] + total_fee = int(item.get("total_fee", total_fee)) + total_fee_yuan = f"{total_fee / 100:.2f}" + user_id = item.get("user_id", "unknown") + pay_type = (item.get("type") or "vip").lower() + random_suffix = "".join(random.choices(string.hexdigits.lower(), k=8)) + timestamp = int(time.time()) + order_id = f"{user_id}_{pay_type}_order_{timestamp}_{random_suffix}" + return_url = item.get("return_url", "") + if return_url and "order_id=" not in return_url: + sep = "&" if "?" in return_url else "?" + return_url = f"{return_url}{sep}order_id={order_id}" + type_map = {"book": "购买书籍", "meetup": "社区报名", "video": "视频解锁", "vip": "VIP会员充值"} + name = type_map.get(pay_type, "VIP会员充值") + notify_url = _get_notify_path(provider.name) + result = provider.create_order( + order_id=order_id, + name=name, + total_fee_yuan=total_fee_yuan, + pay_type=channel, + notify_url=notify_url, + return_url=return_url, + client_ip=item.get("client_ip"), + ) + if provider.name == "xorpay": + return { + "status": "ok", + "xorpay_params": result.get("params", {}), + "xorpay_url": result.get("pay_url", ""), + "provider": provider.name, + "order_id": order_id, + } + params = result.get("params", {}) + params_str = {k: str(v) for k, v in params.items() if v is not None and str(v).strip() != ""} + return { + "status": "ok", + "params": params_str, + "pay_url": result.get("pay_url", ""), + "provider": provider.name, + "submit_method": "POST", + "order_id": order_id, + } + + +@router.post("/payh5/redirect") +async def wxh5_payh5_redirect(item: dict, request: Request): + """wxH5 专用：ZPAY mapi 接口，PC/手机用 zpay，默认微信支付，支付宝暂不提供""" + provider = get_payment_provider_by_name("zpay") + channel = resolve_channel(item.get("channel")) + plan = (item.get("plan") or "year").lower() + total_fee = PLAN_MAP.get(plan, PLAN_MAP["year"])[0] + try: + total_fee = int(item.get("total_fee", total_fee)) + except (TypeError, ValueError): + total_fee = PLAN_MAP["year"][0] + total_fee_yuan = f"{total_fee / 100:.2f}" + user_id = item.get("user_id", "unknown") + pay_type = (item.get("type") or "vip").lower() + device = item.get("device", "pc") + client_ip = item.get("client_ip") or (request.client.host if request.client else "127.0.0.1") + forwarded = request.headers.get("x-forwarded-for") + if forwarded: + client_ip = forwarded.split(",")[0].strip() + random_suffix = "".join(random.choices(string.hexdigits.lower(), k=8)) + order_id = f"{user_id}_{pay_type}_order_{int(time.time())}_{random_suffix}" + return_url_raw = item.get("return_url", "") + if return_url_raw and "order_id=" not in return_url_raw: + sep = "&" if "?" in return_url_raw else "?" + return_url = f"{return_url_raw}{sep}order_id={order_id}" + else: + return_url = return_url_raw + type_map = {"book": "购买书籍", "meetup": "社区报名", "video": "视频解锁", "vip": "VIP会员充值"} + name = type_map.get(pay_type, "VIP会员充值") + notify_url = f"{BASE_URL.rstrip('/')}/wxh5/zpay_notify" + result = provider.create_order_api( + order_id=order_id, + name=name, + total_fee_yuan=total_fee_yuan, + pay_type=channel, + notify_url=notify_url, + return_url=return_url, + client_ip=client_ip, + device=device, + ) + if result.get("status") != "ok": + err_msg = result.get("msg", "ZPAY mapi 创建订单失败") + raise HTTPException(status_code=500, detail=err_msg) + payurl = result.get("payurl") + payurl2 = result.get("payurl2") + img = result.get("img") + if device == "pc" and img: + return JSONResponse(status_code=200, content={ + "status": "ok", "qrcode_mode": True, "order_id": order_id, + "img": img, "payurl": payurl, "return_url": return_url, "channel": channel, + }) + if device == "wechat" and channel.lower() in ("wxpay", "wx", "wechat"): + return JSONResponse(status_code=200, content={"use_form": True, "order_id": order_id}) + headers = {"X-Pay-Order-Id": order_id} + prefer_json = str(item.get("prefer_json", "")).lower() in ("1", "true", "yes") + if prefer_json: + target = payurl2 if (device == "h5" and channel.lower() in ("wxpay", "wx", "wechat")) else payurl + if target: + return JSONResponse(status_code=200, content={ + "status": "ok", "redirect_url": target, "order_id": order_id, "provider": "zpay", + }) + if device == "h5" and channel.lower() in ("wxpay", "wx", "wechat") and payurl2: + return RedirectResponse(url=payurl2, status_code=302, headers=headers) + if payurl: + return RedirectResponse(url=payurl, status_code=302, headers=headers) + raise HTTPException(status_code=500, detail="ZPAY 未返回支付链接") + + +def _parse_wxh5_user_id(order_id: str) -> str: + """从 order_id 解析 user_id，格式: user_id_vip_order_ts""" + if "_order_" not in order_id: + return "" + prefix = order_id.split("_order_")[0] + if "_" in prefix: + return prefix.rsplit("_", 1)[0] + return prefix + + +def _handle_wxh5_payment_success(order_id: str, pay_price: str, provider_name: str): + """wxH5 支付成功：PocketBase + Supabase""" + user_id = _parse_wxh5_user_id(order_id) + if not user_id: + return + handle_payment_success(order_id, pay_price, provider_name, use_memos=False) + amount_cents = int(float(pay_price or 0) * 100) + plan = _infer_plan_from_amount(amount_cents) + _sync_wxh5_vip_to_supabase(user_id, plan) + + +@router.get("/order_status") +async def wxh5_order_status(order_id: str): + """wxH5 通用：查询订单支付状态""" + zpay = get_payment_provider_by_name("zpay") + if isinstance(zpay, ZPayProvider): + result = zpay.query_order_status(order_id, timeout=15) + if not result.get("error") and result.get("paid"): + return result + try: + from ..services.pb import get_pb_client + pb = get_pb_client() + records = pb.collection("payments").get_list(1, 1, {"filter": f'order_id = "{order_id}"'}) + return {"paid": len(records.items) > 0} + except Exception as e: + logging.warning(f"wxh5 PocketBase 查询失败: {e}") + return {"paid": False, "error": str(e)} + + +@router.post("/xorpay_notify") +async def wxh5_xorpay_notify(request: Request): + """wxH5 XorPay 异步通知""" + form_data = await request.form() + data = {k: v for k, v in form_data.items()} + order_id = data.get("order_id", "") + if order_id in processed_orders: + return "ok" + provider = get_payment_provider_by_name("xorpay") + if not provider.verify_notify(data): + raise HTTPException(status_code=400, detail="sign error") + parsed = provider.parse_notify(data) + if parsed.get("trade_status") != "success": + return "ok" + try: + _handle_wxh5_payment_success(order_id, parsed.get("pay_price", ""), "wxh5-xorpay") + except Exception as e: + logging.error(f"wxh5 业务处理异常: {e}", exc_info=True) + raise + return provider.success_response() + + +@router.get("/zpay_notify") +@router.post("/zpay_notify") +async def wxh5_zpay_notify(request: Request): + """wxH5 ZPAY 异步通知""" + provider = get_payment_provider_by_name("zpay") + if request.method == "GET": + data = dict(request.query_params) + else: + form_data = await request.form() + data = {k: v for k, v in form_data.items()} + order_id = data.get("out_trade_no", "") + if order_id in processed_orders: + return "success" + if not provider.verify_notify(data): + raise HTTPException(status_code=400, detail="sign error") + parsed = provider.parse_notify(data) + trade_status = parsed.get("trade_status") or data.get("trade_status", "") + if trade_status not in ("success", "TRADE_SUCCESS"): + return "success" + try: + _handle_wxh5_payment_success(order_id, parsed.get("pay_price") or data.get("money", ""), "wxh5-zpay") + except Exception as e: + logging.error(f"wxh5 业务处理异常: {e}", exc_info=True) + raise + return provider.success_response() diff --git a/app/sdk/__init__.py b/app/sdk/__init__.py new file mode 100644 index 0000000..bf9625d --- /dev/null +++ b/app/sdk/__init__.py @@ -0,0 +1,20 @@ +""" +PayJS API SDK 模块 +- payment: 支付网关（ZPAY/XorPay） +- pocketbase: PocketBase 数据库 +- minio: MinIO 对象存储 +- memos: Memos 用户（仅 nomadvip 使用） +""" +from .payment import get_payment_provider, ZPayProvider, XorPayProvider +from .pocketbase import PocketBaseSDK +from .minio import MinioSDK +from .memos import MemosSDK + +__all__ = [ + "get_payment_provider", + "ZPayProvider", + "XorPayProvider", + "PocketBaseSDK", + "MinioSDK", + "MemosSDK", +] diff --git a/app/sdk/memos.py b/app/sdk/memos.py new file mode 100644 index 0000000..5498b66 --- /dev/null +++ b/app/sdk/memos.py @@ -0,0 +1,95 @@ +""" +Memos SDK +用户创建等操作，仅 nomadvip 使用 +""" +import logging +from typing import Optional + +import requests + + +class MemosSDK: + """Memos 用户管理 SDK""" + + def __init__( + self, + api_url: str, + token: str, + enabled: bool = True, + ): + self.api_url = api_url.rstrip("/") + self.token = token + self.enabled = enabled and bool(api_url and token) + + def create_user( + self, + user_id: str, + password: str = "12345678", + ) -> dict: + """ + 创建 Memos 用户 + :param user_id: PocketBase 用户 ID，如 user1766043770158 + :return: {"success": bool, "data"?: dict, "error"?: str} + """ + if not self.enabled: + return {"success": False, "error": "Memos 未启用"} + + try: + if user_id.startswith("user_"): + cleaned = "user" + user_id.split("user_", 1)[1] + elif "_" in user_id: + cleaned = user_id.replace("_", "") + else: + cleaned = user_id + + username = cleaned + data = { + "username": username, + "password": password, + "role": "USER", + "state": "NORMAL", + } + headers = { + "Content-Type": "application/json", + "Authorization": f"Bearer {self.token}", + } + resp = requests.post( + self.api_url, + json=data, + headers=headers, + timeout=10, + ) + if resp.status_code in (200, 201): + logging.info(f"memos 用户创建成功: {username}") + return {"success": True, "data": resp.json()} + else: + logging.error(f"memos 创建失败 - username: {username}, status: {resp.status_code}, response: {resp.text}") + return {"success": False, "error": resp.text} + except Exception as e: + logging.error(f"创建 memos 用户异常 - user_id: {user_id}, error: {e}") + return {"success": False, "error": str(e)} + + def create_user_by_username( + self, + username: str, + password: str = "12345678", + ) -> dict: + """按用户名直接创建（用于 /create_user 接口）""" + if not self.enabled: + return {"success": False, "error": "Memos 未启用"} + try: + data = { + "username": username, + "password": password, + "role": "USER", + "state": "NORMAL", + } + headers = { + "Content-Type": "application/json", + "Authorization": f"Bearer {self.token}", + } + resp = requests.post(self.api_url, json=data, headers=headers, timeout=10) + return resp.json() + except Exception as e: + logging.error(f"创建 memos 用户异常 - username: {username}, error: {e}") + return {"error": str(e)} diff --git a/app/sdk/minio.py b/app/sdk/minio.py new file mode 100644 index 0000000..35ab51e --- /dev/null +++ b/app/sdk/minio.py @@ -0,0 +1,92 @@ +""" +MinIO SDK +对象存储封装，支持配置注入 +""" +import os +import time +import random +import string +from typing import Optional + +try: + from minio import Minio + MINIO_AVAILABLE = True +except ImportError: + MINIO_AVAILABLE = False + Minio = None + + +class MinioSDK: + """MinIO 对象存储 SDK""" + + def __init__( + self, + end_point: str, + access_key: str, + secret_key: str, + bucket: str = "hackrobot", + use_ssl: bool = True, + port: Optional[int] = None, + region: str = "china", + public_url: Optional[str] = None, + upload_prefix: str = "joins", + ): + if not MINIO_AVAILABLE: + raise ImportError("minio 未安装，请执行: pip install minio") + self.end_point = end_point + self.port = port or (443 if use_ssl else 80) + self.access_key = access_key + self.secret_key = secret_key + self.bucket = bucket + self.use_ssl = use_ssl + self.region = region + self.public_url = public_url or f"https://{end_point}/{bucket}" + self.upload_prefix = upload_prefix.rstrip("/") + self._client: Optional[Minio] = None + + def get_client(self) -> "Minio": + """获取 MinIO 客户端""" + if self._client is None: + endpoint = f"{self.end_point}:{self.port}" if self.port not in (80, 443) else self.end_point + self._client = Minio( + endpoint, + access_key=self.access_key, + secret_key=self.secret_key, + secure=self.use_ssl, + region=self.region, + ) + return self._client + + def generate_object_key(self, ext: str) -> str: + """生成上传用的对象键""" + ext = ext.lstrip(".") + suffix = "".join(random.choices(string.ascii_lowercase + string.digits, k=8)) + return f"{self.upload_prefix}/{int(time.time() * 1000)}_{suffix}.{ext}" + + def upload_buffer( + self, + data: bytes, + object_key: str, + content_type: str = "application/octet-stream", + ) -> dict: + """ + 上传文件到 MinIO + :return: {"url": str, "object_key": str} + """ + client = self.get_client() + from io import BytesIO + stream = BytesIO(data) + client.put_object( + self.bucket, + object_key, + stream, + len(data), + content_type=content_type, + ) + url = f"{self.public_url.rstrip('/')}/{object_key}" + return {"url": url, "object_key": object_key} + + @property + def enabled(self) -> bool: + """是否启用（需配置 access_key/secret_key）""" + return bool(self.end_point and self.access_key and self.secret_key) diff --git a/app/sdk/payment/__init__.py b/app/sdk/payment/__init__.py new file mode 100644 index 0000000..8f06589 --- /dev/null +++ b/app/sdk/payment/__init__.py @@ -0,0 +1,15 @@ +""" +支付网关 SDK +支持 ZPAY、XorPay，通过配置切换 +""" +from app.payment.base import PaymentProvider +from app.payment.zpay import ZPayProvider +from app.payment.xorpay import XorPayProvider +from app.payment.factory import get_payment_provider + +__all__ = [ + "PaymentProvider", + "ZPayProvider", + "XorPayProvider", + "get_payment_provider", +] diff --git a/app/sdk/pocketbase.py b/app/sdk/pocketbase.py new file mode 100644 index 0000000..deb8ef0 --- /dev/null +++ b/app/sdk/pocketbase.py @@ -0,0 +1,69 @@ +""" +PocketBase SDK +数据库操作封装，支持配置注入 +""" +from typing import Optional + +from pocketbase import PocketBase + + +class PocketBaseSDK: + """PocketBase 客户端 SDK""" + + def __init__( + self, + url: str, + admin_email: str, + admin_password: str, + ): + self.url = url + self.admin_email = admin_email + self.admin_password = admin_password + self._client: Optional[PocketBase] = None + + def get_client(self) -> PocketBase: + """获取已认证的 PocketBase 客户端""" + if self._client is None: + self._client = PocketBase(self.url) + self._client.admins.auth_with_password(self.admin_email, self.admin_password) + return self._client + + def collection(self, name: str): + """获取集合""" + return self.get_client().collection(name) + + def create_payment(self, user_id: str, pay_type: str, order_id: str, amount: int) -> dict: + """创建支付记录""" + return self.collection("payments").create({ + "user_id": user_id, + "type": pay_type, + "order_id": order_id, + "amount": amount, + }) + + def update_meetup_payment( + self, + user_id: str, + order_id: str, + amount: int, + ) -> None: + """更新或创建 meetup 支付记录""" + solan = self.collection("solan") + existing = solan.get_list(1, 1, { + "filter": f'user_id = "{user_id}"', + "sort": "-created", + }) + if existing.items: + record_id = existing.items[0].id + solan.update(record_id, { + "type": "meetup", + "order_id": order_id, + "amount": amount, + }) + else: + solan.create({ + "user_id": user_id, + "type": "meetup", + "order_id": order_id, + "amount": amount, + }) diff --git a/app/services/__init__.py b/app/services/__init__.py new file mode 100644 index 0000000..7d62c10 --- /dev/null +++ b/app/services/__init__.py @@ -0,0 +1,16 @@ +""" +共享业务逻辑，基于 SDK 封装 +- PocketBase、Memos 通过 app.sdk 使用 +- 仅 nomadvip 涉及 Memos +""" +from .pb import get_pb_client +from .memos import create_memos_user +from .payment import handle_payment_success, parse_order_id, processed_orders + +__all__ = [ + "get_pb_client", + "create_memos_user", + "handle_payment_success", + "parse_order_id", + "processed_orders", +] diff --git a/app/services/memos.py b/app/services/memos.py new file mode 100644 index 0000000..b5ac157 --- /dev/null +++ b/app/services/memos.py @@ -0,0 +1,21 @@ +"""Memos 用户创建（基于 SDK，仅 nomadvip 使用）""" +from app.config import MEMOS_API, MEMOS_TOKEN +from app.sdk.memos import MemosSDK + +_memos_sdk: MemosSDK | None = None + + +def _get_memos_sdk() -> MemosSDK: + global _memos_sdk + if _memos_sdk is None: + _memos_sdk = MemosSDK( + api_url=MEMOS_API, + token=MEMOS_TOKEN, + enabled=bool(MEMOS_API and MEMOS_TOKEN), + ) + return _memos_sdk + + +def create_memos_user(user_id: str, password: str = "12345678"): + """创建 Memos 用户""" + return _get_memos_sdk().create_user(user_id=user_id, password=password) diff --git a/app/services/minio.py b/app/services/minio.py new file mode 100644 index 0000000..f66ecaa --- /dev/null +++ b/app/services/minio.py @@ -0,0 +1,51 @@ +"""MinIO 对象存储（基于 SDK）""" +from app.config import ( + MINIO_ENDPOINT, + MINIO_PORT, + MINIO_ACCESS_KEY, + MINIO_SECRET_KEY, + MINIO_BUCKET, + MINIO_USE_SSL, + MINIO_PUBLIC_URL, + MINIO_UPLOAD_PREFIX, +) +from app.sdk.minio import MinioSDK + +_minio_sdk: MinioSDK | None = None + + +def _get_minio_sdk() -> MinioSDK: + global _minio_sdk + if _minio_sdk is None: + _minio_sdk = MinioSDK( + end_point=MINIO_ENDPOINT, + access_key=MINIO_ACCESS_KEY, + secret_key=MINIO_SECRET_KEY, + bucket=MINIO_BUCKET, + use_ssl=MINIO_USE_SSL, + port=MINIO_PORT, + public_url=MINIO_PUBLIC_URL, + upload_prefix=MINIO_UPLOAD_PREFIX, + ) + return _minio_sdk + + +def get_minio_client(): + """获取 MinIO 客户端（未配置时返回 None）""" + sdk = _get_minio_sdk() + if not sdk.enabled: + return None + return sdk.get_client() + + +def upload_buffer(data: bytes, object_key: str, content_type: str = "application/octet-stream"): + """上传文件到 MinIO（未配置时抛出异常）""" + sdk = _get_minio_sdk() + if not sdk.enabled: + raise RuntimeError("MinIO 未配置，请设置 MINIO_ACCESS_KEY、MINIO_SECRET_KEY") + return sdk.upload_buffer(data, object_key, content_type) + + +def generate_object_key(ext: str) -> str: + """生成上传用的对象键""" + return _get_minio_sdk().generate_object_key(ext) diff --git a/app/services/payment.py b/app/services/payment.py new file mode 100644 index 0000000..ba6a4c3 --- /dev/null +++ b/app/services/payment.py @@ -0,0 +1,217 @@ +"""支付成功后的统一业务处理（基于 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]: + """从 order_id 解析 user_id 和 pay_type""" + if "_order_" not in order_id: + return "", "unknown" + prefix = order_id.split("_order_")[0] + if "_" in prefix: + parts = prefix.rsplit("_", 1) + return parts[0], parts[1].lower() + 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, + provider_name: str, + *, + use_memos: bool = False, + site_id: str | None = None, +): + """ + 支付成功后的统一业务处理 + :param use_memos: 是否创建 Memos 用户，仅 nomadvip 为 True + :param site_id: 站点标识（digital/meetup/vip），用于写入 site_vip + """ + user_id, pay_type = parse_order_id(order_id) + if not user_id: + return + + # meetup/salon 新用户：支付成功后才创建用户 + if pay_type in ("meetup", "salon") 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 + + logging.info("[payjsapi] 支付成功 provider=%s user_id=%s pay_type=%s site_id=%s order_id=%s", + provider_name, user_id, pay_type, site_id, order_id) + + if pay_type == "vip" and use_memos: + create_memos_user(user_id, password="12345678") + + if pay_type in ("meetup", "salon"): + logging.info(f"{pay_type} 类型支付成功，不创建 memos 用户 - user_id: {user_id}") + + valid_types = ["vip", "book", "meetup", "video", "salon"] + if pay_type not in valid_types: + pay_type = "vip" + + amount_cents = int(float(pay_price or 0) * 100) + + try: + pb_client = get_pb_client() + if pay_type == "meetup": + solan = pb_client.collection("solan") + solan_collection = "solan" + elif pay_type == "salon": + solan = pb_client.collection("solanRed") + solan_collection = "solanRed" + else: + solan = None + solan_collection = None + + if solan is not None: + existing = solan.get_list(1, 1, { + "filter": f'user_id = "{user_id}"', + "sort": "-created", + }) + if existing.items: + record_id = existing.items[0].id + solan.update(record_id, { + "type": pay_type, + "order_id": order_id, + "amount": amount_cents, + }) + logging.info(f"{solan_collection} 支付字段补齐成功 - user_id: {user_id}, record_id: {record_id}") + else: + solan.create({ + "user_id": user_id, + "type": pay_type, + "order_id": order_id, + "amount": amount_cents, + }) + logging.info(f"{solan_collection} 完整记录创建（兜底） - user_id: {user_id}") + elif pay_type not in ("meetup", "salon"): + existing = pb_client.collection("payments").get_list(1, 1, { + "filter": f'order_id = "{order_id}"', + }) + if not existing.items: + payload = { + "user_id": user_id, + "type": pay_type, + "order_id": order_id, + "amount": amount_cents, + "total_fee": amount_cents, + "channel": "wxpay", + } + pb_client.collection("payments").create(payload) + + # 写入 site_vip（站点独立 VIP）。meetup 写 site_id=meetup，salon 写 site_id=salon + effective_site_id = "meetup" if pay_type == "meetup" else ("salon" if pay_type == "salon" else site_id) + if effective_site_id: + try: + site_vip = pb_client.collection("site_vip") + existing_vip = site_vip.get_list(1, 1, { + "filter": f'user_id = "{user_id}" && site_id = "{effective_site_id}"', + }) + if existing_vip.items: + site_vip.update(existing_vip.items[0].id, { + "order_id": order_id, + "expires_at": "", + }) + logging.info(f"site_vip 更新成功 - user_id: {user_id}, site_id: {effective_site_id}") + else: + site_vip.create({ + "user_id": user_id, + "site_id": effective_site_id, + "order_id": order_id, + "expires_at": "", + }) + logging.info(f"site_vip 创建成功 - user_id: {user_id}, site_id: {effective_site_id}") + except Exception as e: + logging.warning(f"site_vip 写入失败（不影响主流程）: {e}") + + processed_orders.add(order_id) + except Exception as e: + logging.error("PocketBase 操作失败 order_id=%s error=%s", order_id, e) + raise diff --git a/app/services/pb.py b/app/services/pb.py new file mode 100644 index 0000000..0772f0b --- /dev/null +++ b/app/services/pb.py @@ -0,0 +1,27 @@ +"""PocketBase 客户端（基于 SDK）""" +import logging + +from app.config import PB_URL, PB_ADMIN_EMAIL, PB_ADMIN_PASSWORD +from app.sdk.pocketbase import PocketBaseSDK + +logger = logging.getLogger(__name__) +_pb_sdk: PocketBaseSDK | None = None + + +def _get_pb_sdk() -> PocketBaseSDK: + global _pb_sdk + if _pb_sdk is None: + _pb_sdk = PocketBaseSDK( + url=PB_URL, + admin_email=PB_ADMIN_EMAIL, + admin_password=PB_ADMIN_PASSWORD, + ) + return _pb_sdk + + +def get_pb_client(): + """获取已认证的 PocketBase 客户端""" + sdk = _get_pb_sdk() + client = sdk.get_client() + logger.info("PocketBase admin 登录成功") + return client diff --git a/payjs.sh b/payjs.sh new file mode 100644 index 0000000..39989e0 --- /dev/null +++ b/payjs.sh @@ -0,0 +1 @@ +python3 run.py diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..b5436da --- /dev/null +++ b/requirements.txt @@ -0,0 +1,9 @@ +fastapi +uvicorn +python-dotenv +sqlalchemy +pydantic +python-multipart +pocketbase +requests +minio \ No newline at end of file diff --git a/run-with-logs.bat b/run-with-logs.bat new file mode 100644 index 0000000..e460b54 --- /dev/null +++ b/run-with-logs.bat @@ -0,0 +1,4 @@ +@echo off +REM 强制无缓冲输出，确保日志立即显示在终端 +set PYTHONUNBUFFERED=1 +python -u run.py diff --git a/run-with-logs.sh b/run-with-logs.sh new file mode 100644 index 0000000..2d9c563 --- /dev/null +++ b/run-with-logs.sh @@ -0,0 +1,4 @@ +#!/bin/bash +# 强制无缓冲输出，确保日志立即显示在终端 +export PYTHONUNBUFFERED=1 +exec python -u run.py diff --git a/run.py b/run.py new file mode 100644 index 0000000..3dfdd03 --- /dev/null +++ b/run.py @@ -0,0 +1,31 @@ +import os +import sys + +os.environ.setdefault("PYTHONUNBUFFERED", "1") +if hasattr(sys.stdout, "reconfigure"): + sys.stdout.reconfigure(line_buffering=True) +if hasattr(sys.stderr, "reconfigure"): + sys.stderr.reconfigure(line_buffering=True) + +import uvicorn +from dotenv import load_dotenv + +load_dotenv() + +if __name__ == "__main__": + port = int(os.getenv("PORT", "8007")) + reload_enabled = os.getenv("UVICORN_RELOAD", "0").strip().lower() in ("1", "true", "yes") + try: + sys.stderr.write(f"\n[payjsapi] startup http://0.0.0.0:{port}\n") + sys.stderr.flush() + except (OSError, ValueError): + pass + + uvicorn.run( + "app.main:app", + host="0.0.0.0", + port=port, + reload=reload_enabled, + log_level="info", + ) + diff --git a/手机微信支付调试说明.md b/手机微信支付调试说明.md new file mode 100644 index 0000000..c99c778 --- /dev/null +++ b/手机微信支付调试说明.md @@ -0,0 +1,44 @@ +# 手机微信支付调试说明 + +前端（vipgroup 等）与 payjsapi 同机时，支付请求由服务端发起，应使用 `127.0.0.1:8700`。 + +## 1. 同机部署（vipgroup + payjsapi 都在本机） + +**无需配置**。默认使用 `http://127.0.0.1:8700`。 + +若 `.env.local` 中有 `PAYMENT_API_URL=http://0.0.0.0:8700`，请删除或改为： +```bash +PAYMENT_API_URL=http://127.0.0.1:8700 +``` +（0.0.0.0 无法作为连接目标，会导致立即失败） + +## 2. 确认 payjsapi 已启动 + +```bash +cd C:\project\payjsapi +python run.py +``` + +应监听 `0.0.0.0:8700`，可从局域网访问。 + +## 3. 确认防火墙放行 8700 端口 + +Windows 防火墙可能阻止手机访问 8700。若 vipgroup 与 payjsapi 同机，请求是服务端内部调用，一般无问题。 + +## 4. 查看控制台日志 + +支付失败时，vipgroup 的 Next.js 控制台会输出： + +``` +[pay] 请求失败: ... apiUrl: http://xxx:8700 requestHost: xxx +``` + +- 若 `apiUrl` 为 `http://0.0.0.0:8700` → 在 `.env.local` 中设置 `PAYMENT_API_URL=http://192.168.41.222:8700` +- 若 `apiUrl` 正确但仍失败 → 检查 payjsapi 是否运行、网络是否可达 + +## 5. vipgroup 需包含的配置逻辑 + +若 vipgroup 与 lms/nomadlms 结构不同，需确保： + +1. 调用 `/api/pay` 或直接调用 payjsapi 时，使用正确的 `PAYMENT_API_URL` +2. 本地开发时 `PAYMENT_API_URL` 指向 `http://{你的局域网IP}:8700`