From dc0f1dd9ec7c6160bd6866039e22bb67d1d4ebb1 Mon Sep 17 00:00:00 2001 From: eric Date: Mon, 16 Mar 2026 20:08:32 -0500 Subject: [PATCH] 's' --- app/main.py | 57 +-- app/routers/__init__.py | 24 +- app/routers/_digital_base.py | 100 +++- 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_meetup.py | 182 +++++--- app/routers/wxh5.py | 274 ----------- app/services/__init__.py | 10 +- app/services/payment.py | 44 +- requirements.txt | 4 +- 15 files changed, 273 insertions(+), 1714 deletions(-) delete mode 100644 app/routers/cnomadcna.py delete mode 100644 app/routers/common.py delete mode 100644 app/routers/digital.py delete mode 100644 app/routers/legacy.py delete mode 100644 app/routers/meetup.py delete mode 100644 app/routers/nomadlms.py delete mode 100644 app/routers/nomadvip.py delete mode 100644 app/routers/wxh5.py diff --git a/app/main.py b/app/main.py index 2b5f250..72021a3 100644 --- a/app/main.py +++ b/app/main.py @@ -1,14 +1,10 @@ """ -支付 API 主入口 -支持多支付渠道（XorPay、ZPAY），通过 PAYMENT_PROVIDER 环境变量切换 +salon API 主入口 +支付（ZPAY/XorPay）+ 报名、用户、小红书资料 -路由模块化，按项目拆分： -- 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 +路由： +- /salon/* 支付（payh5、payh5/redirect、order_status、zpay_notify、xorpay_notify） +- /api/salon/* 报名、check-user、ensure-user、complete-order、join、volunteers、xiaohongshu/profile 本地调试：BASE_URL 需为 ZPAY 可访问地址（如 ngrok） 线上部署：BASE_URL=https://api.nomadyt.com（必须为 ZPAY 可公网访问的地址） @@ -22,21 +18,11 @@ 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, -) +from .routers import salon_router, salon_meetup_router app = FastAPI( - title="PayJS API", - description="nomadvip / digital / cnomadcna / nomadlms 支付接口", + title="Salon API", + description="沙龙报名、支付、小红书资料", version="2.0", ) @@ -96,9 +82,7 @@ def _safe_stderr_write(message: str) -> None: def _should_trace_request(path: str) -> bool: - return any( - segment in path for segment in ("/nomadvip", "/digital", "/cnomadcna", "/nomadlms", "/salon") - ) + return "/salon" in path or "/api/salon" in path # 启动时校验 BASE_URL（线上部署必须为公网可访问地址，否则 ZPAY 无法回调） def _check_base_url(): @@ -121,7 +105,7 @@ async def startup_event(): @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} + out = {"status": "ok", "service": "salonapi", "base_url": BASE_URL} if str(check_zpay).lower() in ("1", "true", "yes"): # 1) 检测 zpayz.cn 首页 try: @@ -153,21 +137,14 @@ async def health(check_zpay: str = ""): 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") + _safe_stderr_write(f"[salonapi] >>> {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") + _safe_stderr_write(f"[salonapi] <<< {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) @@ -176,13 +153,9 @@ async def root(): """健康检查""" return { "status": "ok", - "service": "payjsapi", + "service": "salonapi", "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", + "salon": "/salon/payh5, /salon/payh5/redirect, /salon/order_status, /salon/zpay_notify, /salon/xorpay_notify", + "api": "/api/salon/check-user, /api/salon/ensure-user, /api/salon/complete-order, /api/salon/join, /api/salon/join/by-wechat, /api/salon/join/status, /api/salon/join/volunteers, /api/salon/xiaohongshu/profile", }, } diff --git a/app/routers/__init__.py b/app/routers/__init__.py index 3b312b5..b9161c0 100644 --- a/app/routers/__init__.py +++ b/app/routers/__init__.py @@ -1,30 +1,12 @@ """ -项目路由模块 -- nomadvip: /nomadvip/* -- digital: /digital/* -- cnomadcna: /cnomadcna/* -- nomadlms: /nomadlms/* -- salon: /salon/*（克隆 cnomadcna） -- legacy: /payh5, /zpay_notify 等（向后兼容 digital/cnomadcna） +salon 项目路由模块 +- salon: /salon/* 支付（payh5、order_status、zpay_notify、xorpay_notify） +- salon_meetup: /api/salon/* 报名、用户、小红书资料 """ -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 index 90d94d1..b56f84a 100644 --- a/app/routers/_digital_base.py +++ b/app/routers/_digital_base.py @@ -18,7 +18,8 @@ from ..payment import ( resolve_channel, ) from ..payment.zpay import ZPayProvider -from ..services import handle_payment_success, processed_orders +from ..payment.xorpay import XorPayProvider +from ..services import handle_payment_success, try_add_processed_order, remove_processed_order def _safe_log(level: int, message: str, *args, **kwargs) -> None: @@ -28,6 +29,45 @@ def _safe_log(level: int, message: str, *args, **kwargs) -> None: pass +async def _poll_xorpay_until_paid( + order_id: str, + project_name: str, + site_id: str, + max_attempts: int = 45, + interval_sec: float = 2.0, +) -> None: + """创建 XorPay 订单后后台轮询，支付成功即更新 DB，不依赖 notify 回调""" + provider = get_payment_provider_by_name("xorpay") + if not isinstance(provider, XorPayProvider): + return + for attempt in range(max_attempts): + if attempt > 0: + await asyncio.sleep(interval_sec) + try: + result = await asyncio.to_thread(provider.query_order_status, order_id, 6) + if not result.get("paid"): + continue + pay_price = str(result.get("pay_price", "") or result.get("price", "") or "1") + if not try_add_processed_order(order_id): + return + try: + await asyncio.to_thread( + handle_payment_success, + order_id, + pay_price, + f"{project_name}-xorpay-poll", + use_memos=False, + site_id=site_id, + ) + _safe_log(logging.INFO, "%s XorPay 轮询检测到支付 order_id=%s", project_name, order_id) + except Exception as e: + remove_processed_order(order_id) + _safe_log(logging.ERROR, "%s XorPay 轮询更新失败 order_id=%s: %s", project_name, order_id, e) + return + except Exception as e: + _safe_log(logging.WARNING, "%s XorPay 轮询异常 order_id=%s attempt=%s: %s", project_name, order_id, attempt + 1, e) + + def create_digital_router(prefix: str, project_name: str, site_id: str | None = None) -> APIRouter: """ 创建 digital/cnomadcna/nomadlms 的支付路由 @@ -77,6 +117,7 @@ def create_digital_router(prefix: str, project_name: str, site_id: str | None = type_map = { "book": "购买书籍", "meetup": "数字游民社区报名", + "salon": "沙龙报名茶歇费", "video": "视频解锁", "vip": "VIP会员充值", } @@ -101,6 +142,8 @@ def create_digital_router(prefix: str, project_name: str, site_id: str | None = notify_url, return_url or "", ) + if provider.name == "xorpay" and site_id and pay_type in ("salon", "meetup"): + asyncio.create_task(_poll_xorpay_until_paid(order_id, project_name, site_id)) if provider.name == "xorpay": return { "status": "ok", @@ -261,27 +304,39 @@ def create_digital_router(prefix: str, project_name: str, site_id: str | None = @router.post("/xorpay_notify") async def xorpay_notify(request: Request): - """XorPay 异步通知""" + """XorPay 异步通知：先返回 200 避免超时重试，后台异步更新 solanRed/site_vip""" 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 try_add_processed_order(order_id): + return provider.success_response() 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() + + # 先返回 200，后台异步更新 DB，用户再次打开首页时 by-wechat 可直接查到 solanRed + pay_price = parsed.get("pay_price", data.get("pay_price", "")) + provider_name = f"{project_name}-xorpay" + + def _do_update(): + try: + handle_payment_success( + order_id, + pay_price, + provider_name, + use_memos=False, + site_id=site_id, + ) + except Exception as e: + _safe_log(logging.ERROR, "%s XorPay 后台更新失败 order_id=%s: %s", project_name, order_id, e) + remove_processed_order(order_id) + + asyncio.create_task(asyncio.to_thread(_do_update)) return provider.success_response() @router.get("/zpay_notify") @@ -295,7 +350,7 @@ def create_digital_router(prefix: str, project_name: str, site_id: str | None = 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: + if not try_add_processed_order(order_id): return "success" provider = get_payment_provider_by_name("zpay") if not provider.verify_notify(data): @@ -304,13 +359,18 @@ def create_digital_router(prefix: str, project_name: str, site_id: str | None = 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, - ) + try: + await asyncio.to_thread( + handle_payment_success, + order_id, + parsed.get("pay_price", data.get("money", "")), + f"{project_name}-zpay", + use_memos=False, + site_id=site_id, + ) + except Exception: + remove_processed_order(order_id) + raise return provider.success_response() return router diff --git a/app/routers/cnomadcna.py b/app/routers/cnomadcna.py deleted file mode 100644 index f4bbbeb..0000000 --- a/app/routers/cnomadcna.py +++ /dev/null @@ -1,4 +0,0 @@ -"""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 deleted file mode 100644 index 40391bd..0000000 --- a/app/routers/common.py +++ /dev/null @@ -1,65 +0,0 @@ -"""共享接口：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 deleted file mode 100644 index 56de4c0..0000000 --- a/app/routers/digital.py +++ /dev/null @@ -1,4 +0,0 @@ -"""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 deleted file mode 100644 index 5134a6d..0000000 --- a/app/routers/legacy.py +++ /dev/null @@ -1,4 +0,0 @@ -"""向后兼容：/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 deleted file mode 100644 index 6bd2aba..0000000 --- a/app/routers/meetup.py +++ /dev/null @@ -1,328 +0,0 @@ -""" -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 deleted file mode 100644 index 5530aa4..0000000 --- a/app/routers/nomadlms.py +++ /dev/null @@ -1,4 +0,0 @@ -"""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 deleted file mode 100644 index 406b3d8..0000000 --- a/app/routers/nomadvip.py +++ /dev/null @@ -1,883 +0,0 @@ -"""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_meetup.py b/app/routers/salon_meetup.py index 445f2eb..d0e4619 100644 --- a/app/routers/salon_meetup.py +++ b/app/routers/salon_meetup.py @@ -2,6 +2,7 @@ salon 加入流程：check-user、ensure-user、complete-order 克隆 meetup 逻辑，使用 site_id=salon """ +import asyncio import json import logging import re @@ -326,69 +327,130 @@ async def complete_order(item: dict): if not user_id or pay_type != "salon": raise HTTPException(status_code=400, detail="订单格式无效") - max_retries = 4 - for attempt in range(max_retries): + def _query_solan_red_paid() -> dict | None: + """查 solanRed 是否已有 vip 和 amount，有则说明已支付成功，可快速返回""" + esc = user_id.replace("\\", "\\\\").replace('"', '\\"') + r = requests.get( + f"{base}/api/collections/solanRed/records", + params={"filter": f'user_id = "{esc}"', "sort": "-created", "perPage": 1}, + headers={"Authorization": f"Bearer {token}"}, + timeout=5, + ) + if r.status_code != 200: + return None + items = r.json().get("items") or [] + if not items: + return None + rec = items[0] + if rec.get("vip") and (rec.get("amount") or 0) > 0: + return {"wechatId": rec.get("wechatId"), "nickname": rec.get("nickname"), "media": rec.get("media")} + return None + + paid_record = await asyncio.to_thread(_query_solan_red_paid) + if paid_record is not None: + if user_id.startswith("pending_"): + email = _decode_pending_email(user_id) + if not email: + raise HTTPException(status_code=400, detail="订单格式无效") + def _login_pending() -> dict | None: + 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: + return None + return login_r.json() + auth_data = await asyncio.to_thread(_login_pending) + if not auth_data: + raise HTTPException(status_code=500, detail="登录失败，请稍后重试") + auth_rec = auth_data.get("record") or {} + if paid_record.get("wechatId") and not auth_rec.get("wechatId"): + auth_rec = {**auth_rec, "wechatId": paid_record["wechatId"]} + return {"ok": True, "token": auth_data.get("token"), "record": auth_rec, "is_new": True} + return {"ok": True, "token": None, "record": paid_record, "is_new": False} + + def _query_site_vip_count() -> int: 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 + return (r.json().get("totalItems") or 0) if r.status_code == 200 else 0 + + cached_amount_cents: int | None = None # 从 poll 路径获取时缓存，避免兜底重复查 ZPAY + max_retries = 4 + for attempt in range(max_retries): + total = await asyncio.to_thread(_query_site_vip_count) if total > 0: break if attempt < max_retries - 1: paid_via = None pay_price = "" - zpay_result = _query_zpay_order(order_id) + zpay_result, (xorpay_paid, xorpay_price) = await asyncio.gather( + asyncio.to_thread(_query_zpay_order, order_id), + asyncio.to_thread(_query_xorpay_paid, order_id), + ) if zpay_result: paid_via = "zpay" pay_price = str(zpay_result.get("money", "") or zpay_result.get("pay_price", "")) - if not paid_via: - xorpay_paid, xorpay_price = _query_xorpay_paid(order_id) - if xorpay_paid: - paid_via = "xorpay" - pay_price = xorpay_price + if not paid_via and xorpay_paid: + paid_via = "xorpay" + pay_price = xorpay_price if paid_via: - from ..services.payment import handle_payment_success - price = pay_price if pay_price and float(pay_price or 0) > 0 else "1" - try: - handle_payment_success( - order_id, - price, - f"salon-{paid_via}-poll", - use_memos=False, - site_id=SITE_ID, - ) - time.sleep(1) - continue - except Exception as e: - logging.warning(f"salon complete-order 手动触发支付处理失败: {e}") - time.sleep(2) + money_y = float(pay_price or 0) if pay_price else 0 + cached_amount_cents = int(money_y * 100) if money_y > 0 else 100 + from ..services.payment import handle_payment_success, try_add_processed_order, remove_processed_order + if try_add_processed_order(order_id): + price = pay_price if pay_price and float(pay_price or 0) > 0 else "1" + try: + await asyncio.to_thread( + handle_payment_success, + order_id, + price, + f"salon-{paid_via}-poll", + use_memos=False, + site_id=SITE_ID, + ) + await asyncio.sleep(0.3) # 短暂等待 PB 写入完成 + continue + except Exception as e: + remove_processed_order(order_id) + logging.warning(f"salon complete-order 手动触发支付处理失败: {e}") + await asyncio.sleep(1) 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}") + logging.warning(f"salon complete-order 未找到订单: order_id={order_id}, attempt={attempt + 1}") raise HTTPException(status_code=400, detail="订单不存在或未支付成功") # 兜底更新 solanRed：回调可能未到，确保 order_id、amount、vip 正确（按 user_id 查找，pending_ 用户记录存的是 pending_xxx） - zpay_result = _query_zpay_order(order_id) - money_yuan = float(zpay_result.get("money", 0) or 0) if zpay_result else 0 - amount_cents = int(money_yuan * 100) if money_yuan > 0 else 100 + if cached_amount_cents is not None: + amount_cents = cached_amount_cents + else: + zpay_result = await asyncio.to_thread(_query_zpay_order, order_id) + money_yuan = float(zpay_result.get("money", 0) or 0) if zpay_result else 0 + amount_cents = int(money_yuan * 100) if money_yuan > 0 else 100 lookup_user_id = user_id - _update_solan_red_on_payment(token, lookup_user_id, order_id, amount_cents) + await asyncio.to_thread(_update_solan_red_on_payment, token, lookup_user_id, order_id, amount_cents) 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: + + def _login_pending() -> dict | None: + 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: + return None + return login_r.json() + + auth_data = await asyncio.to_thread(_login_pending) + if not auth_data: raise HTTPException(status_code=500, detail="登录失败，请稍后重试") - auth_data = login_r.json() return { "ok": True, "token": auth_data.get("token"), @@ -397,8 +459,7 @@ async def complete_order(item: dict): } # 非 pending 用户：返回 solanRed 记录供前端显示微信号 - record = None - try: + def _fetch_solan_record() -> dict | None: esc = lookup_user_id.replace("\\", "\\\\").replace('"', '\\"') r = requests.get( f"{base}/api/collections/solanRed/records", @@ -410,13 +471,10 @@ async def complete_order(item: dict): items = r.json().get("items") or [] if items: rec = items[0] - record = { - "wechatId": rec.get("wechatId"), - "nickname": rec.get("nickname"), - "media": rec.get("media"), - } - except Exception: - pass + return {"wechatId": rec.get("wechatId"), "nickname": rec.get("nickname"), "media": rec.get("media")} + return None + + record = await asyncio.to_thread(_fetch_solan_record) return {"ok": True, "token": None, "record": record, "is_new": False} @@ -448,11 +506,8 @@ async def join_status(user_id: str = ""): return {"hasRecord": False, "isComplete": False} -@router.get("/join/by-wechat") -async def join_by_wechat(wechat_id: str = ""): - """按 wechatId 查询 solanRed 记录""" - if not wechat_id or not wechat_id.strip(): - return {"hasRecord": False, "record": None, "isComplete": False, "isWechatFriend": False, "isApproved": False, "isRejected": False, "vip": False} +def _query_by_wechat_sync(wechat_id: str) -> dict: + """同步按 wechatId 查询 solanRed，供 asyncio.to_thread 调用""" try: from ..services import get_pb_client pb_client = get_pb_client() @@ -495,12 +550,16 @@ async def join_by_wechat(wechat_id: str = ""): return {"hasRecord": False, "record": None, "isComplete": False, "isWechatFriend": False, "isApproved": False, "isRejected": False, "vip": False} -@router.get("/join/volunteers") -async def join_volunteers(): - """获取志愿者列表 - solanRed 中 is_volunteer=true && is_approved=true""" - token, err = _get_admin_token() - if not token: - return {"volunteer": None} +@router.get("/join/by-wechat") +async def join_by_wechat(wechat_id: str = ""): + """按 wechatId 查询 solanRed 记录""" + if not wechat_id or not wechat_id.strip(): + return {"hasRecord": False, "record": None, "isComplete": False, "isWechatFriend": False, "isApproved": False, "isRejected": False, "vip": False} + return await asyncio.to_thread(_query_by_wechat_sync, wechat_id) + + +def _fetch_volunteers_sync(token: str) -> dict: + """同步获取志愿者，供 asyncio.to_thread 调用避免阻塞事件循环""" base = (PB_URL or "").rstrip("/") if not base: return {"volunteer": None} @@ -536,6 +595,15 @@ async def join_volunteers(): return {"volunteer": None} +@router.get("/join/volunteers") +async def join_volunteers(): + """获取志愿者列表 - solanRed 中 is_volunteer=true && is_approved=true""" + token, _ = _get_admin_token() + if not token: + return {"volunteer": None} + return await asyncio.to_thread(_fetch_volunteers_sync, token) + + class XiaohongshuProfileRequest(BaseModel): url: str diff --git a/app/routers/wxh5.py b/app/routers/wxh5.py deleted file mode 100644 index 08526dd..0000000 --- a/app/routers/wxh5.py +++ /dev/null @@ -1,274 +0,0 @@ -"""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/services/__init__.py b/app/services/__init__.py index 7d62c10..53518fd 100644 --- a/app/services/__init__.py +++ b/app/services/__init__.py @@ -5,7 +5,13 @@ """ from .pb import get_pb_client from .memos import create_memos_user -from .payment import handle_payment_success, parse_order_id, processed_orders +from .payment import ( + handle_payment_success, + parse_order_id, + processed_orders, + try_add_processed_order, + remove_processed_order, +) __all__ = [ "get_pb_client", @@ -13,4 +19,6 @@ __all__ = [ "handle_payment_success", "parse_order_id", "processed_orders", + "try_add_processed_order", + "remove_processed_order", ] diff --git a/app/services/payment.py b/app/services/payment.py index 75c94a3..490b47b 100644 --- a/app/services/payment.py +++ b/app/services/payment.py @@ -1,16 +1,43 @@ """支付成功后的统一业务处理（基于 SDK）""" import logging +import threading +import time import requests +from cachetools import TTLCache 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() +# 已处理订单（防重复），24h TTL，最大 10000 条，避免内存泄漏 +_processed_orders: TTLCache[str, bool] = TTLCache(maxsize=10000, ttl=86400) +_processed_orders_lock = threading.Lock() + +# 兼容旧导入 +processed_orders = _processed_orders + DEFAULT_PASSWORD = "12345678" +# Admin token 缓存（5 分钟），与 meetup/salon 一致 +_ADMIN_TOKEN_CACHE: tuple[str, float] | None = None +_ADMIN_TOKEN_CACHE_TTL = 300 + + +def try_add_processed_order(order_id: str) -> bool: + """原子性标记订单已处理。返回 True 表示首次处理，False 表示已处理过（跳过）。""" + with _processed_orders_lock: + if order_id in _processed_orders: + return False + _processed_orders[order_id] = True + return True + + +def remove_processed_order(order_id: str) -> None: + """处理失败时移除标记，允许支付方重试。""" + with _processed_orders_lock: + _processed_orders.pop(order_id, None) + def parse_order_id(order_id: str) -> tuple[str, str]: """从 order_id 解析 user_id 和 pay_type""" @@ -74,7 +101,11 @@ def _create_user_by_email(email: str) -> str | None: def _get_admin_token() -> tuple[str | None, str | None]: - """获取 PocketBase 管理员 token""" + """获取 PocketBase 管理员 token，带 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, "未配置" base = (PB_URL or "").rstrip("/") @@ -86,7 +117,10 @@ def _get_admin_token() -> tuple[str | None, str | None]: timeout=10, ) if r.status_code == 200: - return r.json().get("token"), None + token = r.json().get("token") + if token: + _ADMIN_TOKEN_CACHE = (token, now + _ADMIN_TOKEN_CACHE_TTL) + return token, None if r.status_code == 404: continue except Exception: @@ -217,7 +251,7 @@ def handle_payment_success( except Exception as e: logging.warning(f"site_vip 写入失败（不影响主流程）: {e}") - processed_orders.add(order_id) + # 由回调方在验签后、调用前通过 try_add_processed_order 原子性标记，此处不再 add except Exception as e: logging.error("PocketBase 操作失败 order_id=%s error=%s", order_id, e) raise diff --git a/requirements.txt b/requirements.txt index 173e7a5..81f5237 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,10 +1,10 @@ fastapi uvicorn python-dotenv -sqlalchemy pydantic python-multipart pocketbase requests minio -playwright \ No newline at end of file +playwright +cachetools \ No newline at end of file