s

.env.example

@@ -4,7 +4,7 @@
 # 可用 ngrok 等隧道：BASE_URL=https://xxx.ngrok.io
 # 或直接使用生产地址测试回调（需 payjsapi 已部署到公网）
 PORT=8007
-BASE_URL=https://api.nomadyt.com
+BASE_URL=https://salonapi.hackrobot.cn
 
 # 支付渠道：zpay | xorpay
 PAYMENT_PROVIDER=zpay

README.md

@@ -17,7 +17,7 @@
 ## 环境
 
 本地调试：`BASE_URL` 需为 ZPAY 可访问地址（如 ngrok 暴露 8700 端口）  
-线上部署：`BASE_URL=https://api.nomadyt.com`
+线上部署：`BASE_URL=https://salonapi.hackrobot.cn`
 
 ## 模块结构
 
@@ -73,7 +73,7 @@ python run.py
 
 ## 线上部署
 
-域名：`https://api.nomadyt.com`
+域名：`https://salonapi.hackrobot.cn`
 
 ```bash
 # .env 中设置
@@ -86,9 +86,9 @@ PAYMENT_PROVIDER=zpay
 
 本地正常、Ubuntu 部署后 api.hackrobot.cn 调用支付失败，常见原因：
 
-1. **BASE_URL 配置错误**：必须为 ZPAY 可公网访问的地址。错误示例：`http://127.0.0.1:8700`、`http://localhost`。正确：`https://api.nomadyt.com`（域名需解析到本机，nginx 反向代理对应端口）。
+1. **BASE_URL 配置错误**：必须为 ZPAY 可公网访问的地址。错误示例：`http://127.0.0.1:8700`、`http://localhost`。正确：`https://salonapi.hackrobot.cn`（域名需解析到本机，nginx 反向代理对应端口）。
 
-2. **服务器无法访问 zpayz.cn**：防火墙、DNS 或网络限制。访问 `https://api.nomadyt.com/health?check_zpay=1` 自检：
+2. **服务器无法访问 zpayz.cn**：防火墙、DNS 或网络限制。访问 `https://salonapi.hackrobot.cn/health?check_zpay=1` 自检：
    - `zpayz_reachable: false` → 开放出站或检查网络
    - `zpayz_error` 含 "SSL" → 执行 `apt install ca-certificates` 后重启
 

app/config.py

@@ -43,8 +43,8 @@ MEMOS_TOKEN = os.getenv(
     "eyJhbGciOiJIUzI1NiIsImtpZCI6InYxIiwidHlwIjoiSldUIn0.eyJuYW1lIjoiaGFja3JvYm90IiwiaXNzIjoibWVtb3MiLCJzdWIiOiIxIiwiYXVkIjpbInVzZXIuYWNjZXNzLXRva2VuIl0sImlhdCI6MTc1NzQwNTE0OX0.Idn7kBlxE-CoSOXwWuZ1tHGIRKHAIeDyXSafGS5OHsg",
 )
 
-# 回调地址（用于生成 notify_url）。salon 线上部署 api.nomadyt.com
+# 回调地址（用于生成 notify_url）。线上 https://salonapi.hackrobot.cn
-BASE_URL = os.getenv("BASE_URL", "https://api.nomadyt.com")
+BASE_URL = os.getenv("BASE_URL", "https://salonapi.hackrobot.cn")
 
 # MinIO 对象存储
 MINIO_ENDPOINT = os.getenv("MINIO_ENDPOINT", "minioweb.hackrobot.cn")

app/main.py

@@ -1,17 +1,13 @@
 """
-支付 API 主入口
+salon API 主入口
-支持多支付渠道（XorPay、ZPAY），通过 PAYMENT_PROVIDER 环境变量切换
+支付（ZPAY/XorPay）+ 报名、用户、小红书资料
 
-路由模块化，按项目拆分：
+路由：
-- nomadvip: /nomadvip/*     - 固定 ZPAY
+- /salon/*      支付（payh5、payh5/redirect、order_status、zpay_notify、xorpay_notify）
-- digital:  /digital/*      - 使用 PAYMENT_PROVIDER
+- /api/salon/*  报名、check-user、ensure-user、complete-order、join、volunteers、xiaohongshu/profile
-- 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.nomadyt.com（必须为 ZPAY 可公网访问的地址）
+线上部署：BASE_URL=https://salonapi.hackrobot.cn（必须为 ZPAY 可公网访问的地址）
 """
 import logging
 import sys
@@ -22,21 +18,11 @@ from fastapi.middleware.cors import CORSMiddleware
 from fastapi.responses import JSONResponse
 
 from .config import BASE_URL
-from .routers import (
+from .routers import salon_router, salon_meetup_router
-    nomadvip_router,
-    digital_router,
-    cnomadcna_router,
-    nomadlms_router,
-    salon_router,
-    legacy_router,
-    common_router,
-    meetup_router,
-    salon_meetup_router,
-)
 
 app = FastAPI(
-    title="PayJS API",
+    title="Salon API",
-    description="nomadvip / digital / cnomadcna / nomadlms 支付接口",
+    description="沙龙报名、支付、小红书资料",
     version="2.0",
 )
 
@@ -96,9 +82,7 @@ def _safe_stderr_write(message: str) -> None:
 
 
 def _should_trace_request(path: str) -> bool:
-    return any(
+    return "/salon" in path or "/api/salon" in path
-        segment in path for segment in ("/nomadvip", "/digital", "/cnomadcna", "/nomadlms", "/salon")
-    )
 
 # 启动时校验 BASE_URL（线上部署必须为公网可访问地址，否则 ZPAY 无法回调）
 def _check_base_url():
@@ -108,7 +92,7 @@ def _check_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.nomadyt.com",
+            "BASE_URL=%s 疑似内网地址，ZPAY 无法回调！线上部署请设置 BASE_URL 为公网地址，如 https://salonapi.hackrobot.cn",
             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",
+            "salon": "/salon/payh5, /salon/payh5/redirect, /salon/order_status, /salon/zpay_notify, /salon/xorpay_notify",
-            "digital": "/digital/payh5, /digital/payh5/redirect, /digital/zpay_order_status, /digital/zpay_notify, /digital/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",
-            "cnomadcna": "/cnomadcna/payh5, ...",
-            "nomadlms": "/nomadlms/payh5, ...",
-            "legacy": "/payh5, /payh5/redirect, /zpay_notify, /xorpay_notify, /zpay_order_status",
-            "common": "/submit_meetup_application, /create_user",
         },
     }

app/routers/__init__.py

@@ -1,30 +1,12 @@
 """
-项目路由模块
+salon 项目路由模块
-- nomadvip: /nomadvip/*
+- salon:        /salon/* 支付（payh5、order_status、zpay_notify、xorpay_notify）
-- digital:   /digital/*
+- salon_meetup: /api/salon/* 报名、用户、小红书资料
-- 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",
 ]

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"
+            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,
-            parsed.get("pay_price", data.get("pay_price", "")),
+                    pay_price,
-            f"{project_name}-xorpay",
+                    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(
+        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

app/routers/cnomadcna.py

@@ -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")

app/routers/common.py

@@ -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

app/routers/digital.py

@@ -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")

app/routers/legacy.py

@@ -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")

app/routers/meetup.py

@@ -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}

app/routers/nomadlms.py

@@ -1,4 +0,0 @@
-"""nomadlms 项目支付接口：/nomadlms/*"""
-from ._digital_base import create_digital_router
-
-router = create_digital_router(prefix="/nomadlms", project_name="nomadlms")

app/routers/nomadvip.py

@@ -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()

app/routers/salon_meetup.py

@@ -2,10 +2,12 @@
 salon 加入流程：check-user、ensure-user、complete-order
 克隆 meetup 逻辑，使用 site_id=salon
 """
+import asyncio
 import json
 import logging
 import re
 import time
+from datetime import datetime, timezone, timedelta
 from urllib.parse import parse_qs, unquote, urlparse
 
 import requests
@@ -17,11 +19,18 @@ from ..config import PB_URL, PB_ADMIN_EMAIL, PB_ADMIN_PASSWORD
 SITE_ID = "salon"
 router = APIRouter(prefix="/api/salon", tags=["salon"])
 DEFAULT_PASSWORD = "12345678"
+CHINA_TZ = timezone(timedelta(hours=8))
+
+
+def _now_china_str() -> str:
+    """当前中国公历时间，格式 YYYY-MM-DD HH:mm:ss，用于 submitted_at_cn 上传"""
+    return datetime.now(CHINA_TZ).strftime("%Y-%m-%d %H:%M:%S")
 
 _SALON_ADMIN_TOKEN_CACHE: tuple[str, float] | None = None
 _TOKEN_CACHE_TTL = 300
 _XHS_PROFILE_PATTERN = re.compile(r"^https?://(www\.)?(xiaohongshu\.com|xhslink\.com)/user/profile/[a-zA-Z0-9_-]+", re.I)
 _XHS_SHORT_PATTERN = re.compile(r"^https?://xhslink\.com/m/[a-zA-Z0-9_-]+", re.I)
+_XHS_NOTE_PATTERN = re.compile(r"^https?://(xhslink\.com/o/|(www\.)?xiaohongshu\.com/explore/)[a-zA-Z0-9_-]+", re.I)
 _XHS_REQUEST_HEADERS = {
     "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/122.0.0.0 Safari/537.36",
     "Accept-Language": "zh-CN,zh;q=0.9",
@@ -286,7 +295,7 @@ def _update_solan_red_on_payment(token: str, user_id: str, order_id: str, amount
             f"{base}/api/collections/solanRed/records",
             params={"filter": f'user_id = "{user_id}"', "sort": "-created", "perPage": 1},
             headers={"Authorization": f"Bearer {token}"},
-            timeout=10,
+            timeout=(8, 20),
         )
         if r.status_code != 200:
             return
@@ -302,9 +311,11 @@ def _update_solan_red_on_payment(token: str, user_id: str, order_id: str, amount
             f"{base}/api/collections/solanRed/records/{rec_id}",
             json=patch,
             headers={"Content-Type": "application/json", "Authorization": f"Bearer {token}"},
-            timeout=10,
+            timeout=(8, 20),
         )
         logging.info(f"salon complete-order 兜底更新 solanRed - user_id: {user_id}, amount: {amount_cents}")
+    except requests.exceptions.RequestException as e:
+        logging.warning(f"salon complete-order 兜底更新 solanRed 请求超时/失败: {e}")
     except Exception as e:
         logging.warning(f"salon complete-order 兜底更新 solanRed 失败: {e}")
 
@@ -326,69 +337,145 @@ async def complete_order(item: dict):
     if not user_id or pay_type != "salon":
         raise HTTPException(status_code=400, detail="订单格式无效")
 
-    max_retries = 4
+    def _query_solan_red_paid() -> dict | None:
-    for attempt in range(max_retries):
+        """查 solanRed 是否已有 vip 和 amount，有则说明已支付成功，可快速返回"""
+        try:
+            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=(6, 12),
+            )
+            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")}
+        except requests.exceptions.RequestException as e:
+            logging.warning(f"salon complete-order PocketBase solanRed 请求超时/失败: {e}")
+        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:
+                try:
+                    login_r = requests.post(
+                        f"{base}/api/collections/users/auth-with-password",
+                        json={"identity": email, "password": DEFAULT_PASSWORD},
+                        timeout=(6, 15),
+                    )
+                    if login_r.status_code != 200:
+                        return None
+                    return login_r.json()
+                except requests.exceptions.RequestException as e:
+                    logging.warning(f"salon complete-order PocketBase login 请求超时/失败: {e}")
+                    return None
+            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:
+        try:
             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,
+                timeout=(8, 20),
             )
-        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
+        except requests.exceptions.RequestException as e:
+            logging.warning(f"salon complete-order PocketBase site_vip 请求超时/失败: {e}")
+            return 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:
+            if not paid_via and xorpay_paid:
-                xorpay_paid, xorpay_price = _query_xorpay_paid(order_id)
-                if xorpay_paid:
                 paid_via = "xorpay"
                 pay_price = xorpay_price
             if paid_via:
-                from ..services.payment import handle_payment_success
+                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:
-                    handle_payment_success(
+                        await asyncio.to_thread(
+                            handle_payment_success,
                             order_id,
                             price,
                             f"salon-{paid_via}-poll",
                             use_memos=False,
                             site_id=SITE_ID,
                         )
-                    time.sleep(1)
+                        await asyncio.sleep(0.3)  # 短暂等待 PB 写入完成
                         continue
                     except Exception as e:
+                        remove_processed_order(order_id)
                         logging.warning(f"salon complete-order 手动触发支付处理失败: {e}")
-                time.sleep(2)
+                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}")
-        logging.warning(f"salon complete-order 未找到订单: order_id={order_id}, attempt={attempt + 1}, pb_resp={resp_preview}")
         raise HTTPException(status_code=400, detail="订单不存在或未支付成功")
 
     # 兜底更新 solanRed：回调可能未到，确保 order_id、amount、vip 正确（按 user_id 查找，pending_ 用户记录存的是 pending_xxx）
-    zpay_result = _query_zpay_order(order_id)
+    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="订单格式无效")
+
+        def _login_pending() -> dict | None:
+            try:
                 login_r = requests.post(
                     f"{base}/api/collections/users/auth-with-password",
                     json={"identity": email, "password": DEFAULT_PASSWORD},
-            timeout=10,
+                    timeout=(6, 15),
                 )
                 if login_r.status_code != 200:
+                    return None
+                return login_r.json()
+            except requests.exceptions.RequestException as e:
+                logging.warning(f"salon complete-order PocketBase login 请求超时/失败: {e}")
+                return None
+
+        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,26 +484,25 @@ async def complete_order(item: dict):
         }
 
     # 非 pending 用户：返回 solanRed 记录供前端显示微信号
-    record = None
+    def _fetch_solan_record() -> dict | None:
         try:
             esc = lookup_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,
+                timeout=(6, 12),
             )
             if r.status_code == 200:
                 items = r.json().get("items") or []
                 if items:
                     rec = items[0]
-                record = {
+                    return {"wechatId": rec.get("wechatId"), "nickname": rec.get("nickname"), "media": rec.get("media")}
-                    "wechatId": rec.get("wechatId"),
+        except requests.exceptions.RequestException as e:
-                    "nickname": rec.get("nickname"),
+            logging.warning(f"salon complete-order PocketBase fetch_solan_record 请求超时/失败: {e}")
-                    "media": rec.get("media"),
+        return None
-                }
+
-    except Exception:
+    record = await asyncio.to_thread(_fetch_solan_record)
-        pass
     return {"ok": True, "token": None, "record": record, "is_new": False}
 
 
@@ -448,11 +534,8 @@ async def join_status(user_id: str = ""):
         return {"hasRecord": False, "isComplete": False}
 
 
-@router.get("/join/by-wechat")
+def _query_by_wechat_sync(wechat_id: str) -> dict:
-async def join_by_wechat(wechat_id: str = ""):
+    """同步按 wechatId 查询 solanRed，供 asyncio.to_thread 调用"""
-    """按 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}
     try:
         from ..services import get_pb_client
         pb_client = get_pb_client()
@@ -495,12 +578,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")
+@router.get("/join/by-wechat")
-async def join_volunteers():
+async def join_by_wechat(wechat_id: str = ""):
-    """获取志愿者列表 - solanRed 中 is_volunteer=true && is_approved=true"""
+    """按 wechatId 查询 solanRed 记录"""
-    token, err = _get_admin_token()
+    if not wechat_id or not wechat_id.strip():
-    if not token:
+        return {"hasRecord": False, "record": None, "isComplete": False, "isWechatFriend": False, "isApproved": False, "isRejected": False, "vip": False}
-        return {"volunteer": None}
+    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,8 +623,87 @@ 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)
+
+
+def _get_this_week_range_china() -> tuple[str, str]:
+    """中国公历本周一 00:00 至周日 23:59:59 的日期字符串"""
+    now = datetime.now(CHINA_TZ)
+    days_to_monday = now.weekday()  # Mon=0, Sun=6
+    monday = now - timedelta(days=days_to_monday)
+    sunday = monday + timedelta(days=6)
+    start = monday.strftime("%Y-%m-%d 00:00:00")
+    end = sunday.strftime("%Y-%m-%d 23:59:59")
+    return start, end
+
+
+def _get_this_week_range_iso() -> tuple[str, str]:
+    """中国公历本周的 UTC 时间范围，用于 created 过滤。PocketBase 要求 Y-m-d H:i:s.uZ 格式（空格非 T）"""
+    now = datetime.now(CHINA_TZ)
+    days_to_monday = now.weekday()
+    monday = now - timedelta(days=days_to_monday)
+    sunday = monday + timedelta(days=6)
+    monday_start = monday.replace(hour=0, minute=0, second=0, microsecond=0)
+    sunday_end = sunday.replace(hour=23, minute=59, second=59, microsecond=999999)
+    start_utc = monday_start.astimezone(timezone.utc).strftime("%Y-%m-%d %H:%M:%S.000Z")
+    end_utc = sunday_end.astimezone(timezone.utc).strftime("%Y-%m-%d %H:%M:%S.999Z")
+    return start_utc, end_utc
+
+
+def _fetch_stats_sync() -> dict:
+    """同步获取报名统计，按场次返回 count 和 joiners。使用 get_pb_client 与 by-wechat 一致"""
+    try:
+        from ..services import get_pb_client
+        pb_client = get_pb_client()
+        solan = pb_client.collection("solanRed")
+    except Exception as e:
+        logging.warning(f"salon join stats 初始化失败: {e}")
+        return {"ok": True, "stats": {"session-1": {"count": 0, "joiners": []}, "session-2": {"count": 0, "joiners": []}}}
+    sessions = [
+        {"id": "session-1", "day_keyword": "周六"},
+        {"id": "session-2", "day_keyword": "周日"},
+    ]
+    result = {}
+    _get = lambda r, k, d=None: getattr(r, k, d) if hasattr(r, k) else (r.get(k, d) if isinstance(r, dict) else d)
+    for s in sessions:
+        sid = s["id"]
+        day_kw = s["day_keyword"]
+        meetupdate_filter = f'(meetupdate ~ "{day_kw}" || meetupdate = "")' if sid == "session-1" else f'meetupdate ~ "{day_kw}"'
+        try:
+            existing = solan.get_list(1, 6, {"filter": meetupdate_filter, "sort": "-created"})
+            items = list(existing.items) if existing.items else []
+            total = existing.total_items if hasattr(existing, "total_items") else len(items)
+            joiners = []
+            for r in items:
+                nickname = str(_get(r, "xiaohongshu_nickname") or _get(r, "nickname") or "").strip() or "匿名"
+                avatar = str(_get(r, "xiaohongshu_avatar") or _get(r, "media") or "").strip() or None
+                xh_url = str(_get(r, "xiaohongshu_url") or "").strip() or None
+                joiners.append({"nickname": nickname, "avatar": avatar, "xiaohongshu_url": xh_url})
+            result[sid] = {"count": total, "joiners": joiners}
+        except Exception as e:
+            logging.warning(f"salon join stats 查询失败 {sid}: {e}")
+            result[sid] = {"count": 0, "joiners": []}
+    return {"ok": True, "stats": result}
+
+
+@router.get("/join/stats")
+async def join_stats():
+    """按场次返回报名人数和最近报名者"""
+    return await asyncio.to_thread(_fetch_stats_sync)
+
+
 class XiaohongshuProfileRequest(BaseModel):
-    url: str
+    url: str | None = None
+    id: str | None = None
+
+    def get_url_or_id(self) -> str:
+        return (self.url or self.id or "").strip()
 
 
 def _normalize_xiaohongshu_profile_url(url: str, depth: int = 0) -> str:
@@ -566,7 +732,7 @@ def _normalize_xiaohongshu_profile_url(url: str, depth: int = 0) -> str:
 
 
 def _resolve_xhslink(url: str) -> str:
-    """解析 xhslink.com 短链为真实 profile URL"""
+    """解析 xhslink.com 短链为真实 profile URL（m/ 为主页短链）"""
     if not _XHS_SHORT_PATTERN.match(url):
         return _normalize_xiaohongshu_profile_url(url)
     try:
@@ -581,6 +747,54 @@ def _resolve_xhslink(url: str) -> str:
         return _normalize_xiaohongshu_profile_url(url)
 
 
+def _extract_profile_url_from_note_state(state: dict) -> str | None:
+    """从笔记页 __INITIAL_STATE__ 提取作者 profile URL"""
+    if not state or not isinstance(state, dict):
+        return None
+
+    def find_user_id(obj, depth=0):
+        if depth > 18:
+            return None
+        if isinstance(obj, dict):
+            uid = obj.get("userId") or obj.get("user_id") or obj.get("id")
+            if uid and len(str(uid)) > 10:
+                return str(uid).strip()
+            for k in ("note", "noteCard", "user", "author", "creator", "note_card", "basicInfo"):
+                v = obj.get(k)
+                if isinstance(v, dict):
+                    found = find_user_id(v, depth + 1)
+                    if found:
+                        return found
+            for v in obj.values():
+                found = find_user_id(v, depth + 1)
+                if found:
+                    return found
+        elif isinstance(obj, list) and obj:
+            return find_user_id(obj[0], depth + 1)
+        return None
+
+    uid = find_user_id(state)
+    return f"https://www.xiaohongshu.com/user/profile/{uid}" if uid else None
+
+
+def _resolve_note_url_to_profile_url(note_url: str) -> str | None:
+    """从笔记链接解析出作者 profile URL，支持 xhslink.com/o/xxx、xiaohongshu.com/explore/xxx"""
+    if not _XHS_NOTE_PATTERN.match(note_url):
+        return None
+    try:
+        r = requests.get(
+            note_url,
+            allow_redirects=True,
+            headers=_XHS_REQUEST_HEADERS,
+            timeout=12,
+        )
+        state = _extract_state_from_html(r.text)
+        return _extract_profile_url_from_note_state(state) if state else None
+    except Exception as e:
+        logging.debug(f"[xiaohongshu] note resolve failed: {e}")
+        return None
+
+
 def _unwrap_ref(value):
     if isinstance(value, dict) and value.get("__v_isRef"):
         raw = value.get("_rawValue")
@@ -867,9 +1081,26 @@ def _parse_dom_text(text: str) -> dict:
     if not text:
         return out
 
+    # 优先匹配「数字万?+? 关注/粉丝/获赞与收藏」格式，如 10+关注、1万+粉丝、10+ 关注
+    for pattern, key in [
+        (r"(\d+万?\+?)\s*关注", "following"),
+        (r"(\d+万?\+?)\s*粉丝", "followers"),
+        (r"(\d+万?\+?)\s*获赞与收藏", "likesAndCollects"),
+        (r"关注\s*(\d+万?\+?)", "following"),
+        (r"粉丝\s*(\d+万?\+?)", "followers"),
+        (r"获赞与收藏\s*(\d+万?\+?)", "likesAndCollects"),
+    ]:
+        m = re.search(pattern, text)
+        if m:
+            val = m.group(1).strip()
+            if not out.get(key) and not out.get(f"{key}Text"):
+                _set_count_fields(out, key, val)
+
     lines = [line.strip() for line in text.splitlines() if line.strip()]
 
     def pick_count(label: str, key: str) -> None:
+        if out.get(key) or out.get(f"{key}Text"):
+            return
         for idx, line in enumerate(lines):
             if line != label or idx == 0:
                 continue
@@ -883,6 +1114,7 @@ def _parse_dom_text(text: str) -> dict:
     pick_count("\u83b7\u8d5e\u4e0e\u6536\u85cf", "likesAndCollects")
 
     # 格式：0 关注 559 粉丝 536 获赞与收藏。找「关注」之后第一个「粉丝」，避免匹配到推荐区的 10+粉丝
+    if not out.get("followers") and not out.get("followersText"):
         i_guan = text.find("关注")
         i_fen = text.find("粉丝", i_guan) if i_guan >= 0 else text.find("粉丝")
         i_huo = text.find("获赞与收藏", i_fen) if i_fen >= 0 else text.find("获赞与收藏")
@@ -893,9 +1125,9 @@ def _parse_dom_text(text: str) -> dict:
             nums_after = [int(m.group()) for m in re.finditer(r"\d+", after)]
             if nums_before:
                 out["followers"] = max(nums_before) if max(nums_before) > 50 else nums_before[-1]
-        if len(nums_before) >= 2:
+            if len(nums_before) >= 2 and not out.get("following"):
                 out["following"] = nums_before[0]
-        if nums_after:
+            if nums_after and not out.get("likesAndCollects"):
                 out["likesAndCollects"] = nums_after[0]
 
     # 笔记/发帖
@@ -1206,8 +1438,8 @@ async def _fetch_xiaohongshu_profile(url: str) -> dict | None:
                     await route.continue_()
 
             await page.route("**/*", handle_route)
-            await page.goto(resolved_url, wait_until="domcontentloaded", timeout=30000)
+            await page.goto(resolved_url, wait_until="domcontentloaded", timeout=20000)
-            await page.wait_for_timeout(5000)
+            await page.wait_for_timeout(2500)
 
             state = await page.evaluate("""() => {
                 const unwrap = (value) => {
@@ -1228,6 +1460,35 @@ async def _fetch_xiaohongshu_profile(url: str) -> dict | None:
 
             body_text = await page.evaluate("""() => document.body ? document.body.innerText : ''""")
             dom_parsed = _parse_dom_text(body_text or "")
+
+            dom_stats = await page.evaluate("""() => {
+                const body = document.body?.innerText || '';
+                const extractNum = (label) => {
+                    const r = new RegExp('(\\\\d+万?\\\\+?)\\\\s*' + label + '|' + label + '\\\\s*(\\\\d+万?\\\\+?)');
+                    const m = body.match(r);
+                    return m ? (m[1] || m[2] || '').trim() : null;
+                };
+                const toNum = (s) => {
+                    if (!s) return null;
+                    const m = s.match(/([\\d.]+)/);
+                    if (!m) return null;
+                    let n = parseFloat(m[1]);
+                    if (/万/.test(s)) n *= 10000;
+                    return Math.floor(n);
+                };
+                return {
+                    followersText: extractNum('粉丝'),
+                    followingText: extractNum('关注'),
+                    likesAndCollectsText: extractNum('获赞与收藏'),
+                    followers: toNum(extractNum('粉丝')),
+                    following: toNum(extractNum('关注')),
+                    likesAndCollects: toNum(extractNum('获赞与收藏'))
+                };
+            }""")
+            for k, v in (dom_stats or {}).items():
+                if v is not None and v != "":
+                    dom_parsed[k] = v
+
             title = await page.title()
 
             nick_from_title = title.split(" - ")[0].strip() if title and " - " in title else None
@@ -1258,23 +1519,32 @@ async def _fetch_xiaohongshu_profile(url: str) -> dict | None:
 
 @router.post("/xiaohongshu/profile")
 async def xiaohongshu_profile(item: XiaohongshuProfileRequest):
-    """获取小红书用户资料（昵称/粉丝/发帖数/性别），支持 xhslink 短链"""
+    """获取小红书用户资料，支持主页链接、xhslink.com/m/ 短链、笔记链接 xhslink.com/o/"""
     logging.info("[xiaohongshu] profile request received")
-    url = (item.url or "").strip()
+    url_raw = (item.get_url_or_id() or "").strip()
-    if not url:
+    if not url_raw:
         logging.warning("[xiaohongshu] 400: 请输入小红书链接")
-        raise HTTPException(status_code=400, detail="请输入小红书链接")
+        raise HTTPException(status_code=400, detail="请输入小红书主页链接或笔记链接")
 
     xh_profile = re.compile(r"^https?://(www\.)?(xiaohongshu\.com|xhslink\.com)/user/profile/[a-zA-Z0-9_-]+", re.I)
     xh_short = re.compile(r"^https?://xhslink\.com/m/[a-zA-Z0-9_-]+", re.I)
-    if not xh_profile.match(url) and not xh_short.match(url):
-        logging.warning(f"[xiaohongshu] 400: 链接格式异常 url={url[:80]}")
-        raise HTTPException(status_code=400, detail="小红书link异常")
 
-    url = _resolve_xhslink(url)
+    if _XHS_NOTE_PATTERN.match(url_raw):
+        url = _resolve_note_url_to_profile_url(url_raw)
+        if not url:
+            raise HTTPException(
+                status_code=400,
+                detail="无法从笔记链接解析用户，请使用主页链接（xhslink.com/m/xxx 或 xiaohongshu.com/user/profile/xxx）",
+            )
+        logging.info(f"[xiaohongshu] 从笔记链接解析得到 profile URL")
+    elif xh_profile.match(url_raw) or xh_short.match(url_raw):
+        url = _resolve_xhslink(url_raw)
+    else:
+        raise HTTPException(status_code=400, detail="请输入有效的小红书主页链接或笔记链接（xhslink.com/o/xxx）")
+
     if not xh_profile.match(url):
-        logging.warning(f"[xiaohongshu] 400: 短链解析失败 resolved={url[:80]}")
+        logging.warning(f"[xiaohongshu] 400: 解析失败 resolved={url[:80]}")
-        raise HTTPException(status_code=400, detail="短链解析失败")
+        raise HTTPException(status_code=400, detail="链接解析失败，请确认链接有效")
 
     parsed = await _fetch_xiaohongshu_profile(url)
     if not parsed:
@@ -1291,21 +1561,31 @@ async def submit_join(item: dict):
     user_id = item.get("user_id")
     if not user_id:
         raise HTTPException(status_code=400, detail="missing user_id")
+    if not (item.get("xiaohongshu_url") or "").strip():
+        raise HTTPException(status_code=400, detail="请填写小红书主页链接或笔记链接")
+    reason = (item.get("reason") or item.get("intro") or "").strip()
+    if not reason:
+        raise HTTPException(status_code=400, detail="请填写自我介绍")
+    if len(reason) < 20:
+        raise HTTPException(status_code=400, detail="自我介绍不少于20字")
 
     try:
         from ..services import get_pb_client
         pb_client = get_pb_client()
         solan = pb_client.collection("solanRed")
         raw_type = item.get("type", "")
-        # PocketBase type 仅接受 session-1/session-2 等，digital-nomad/volunteer 映射为空
+        # PocketBase type 字段为 select 枚举，可能仅接受 digital-nomad/volunteer 等，session-1/session-2 会报 validation_invalid_value
-        pb_type = raw_type if raw_type in ("session-1", "session-2") else ""
+        # 场次信息存 meetupdate，stats 按 meetupdate 过滤
+        pb_type = ""
 
         qualify = bool(item.get("qualify", False))  # 女性且小红书帖子>10 时自动通过
+        user_info_val = (item.get("userInfo") or "").strip()
+        meetupdate_val = str(item.get("meetupdate", "") or "").strip()
         pb_data = {
             "user_id": user_id,
             "nickname": item.get("nickname", ""),
             "occupation": item.get("occupation", ""),
-            "reason": item.get("reason", ""),
+            "reason": reason,
             "wechatId": item.get("wechatId", ""),
             "gender": item.get("gender", ""),
             "education": item.get("education", ""),
@@ -1322,6 +1602,9 @@ async def submit_join(item: dict):
             "is_approved": qualify,  # 符合条件时自动通过审核
             "age_range": item.get("age_range", ""),
             "first_time": item.get("first_time", ""),
+            "userInfo": user_info_val,
+            "meetupdate": meetupdate_val,
+            "submitted_at_cn": _now_china_str(),
         }
         phone_val = item.get("phone", "")
         if phone_val and str(phone_val).strip():

app/routers/wxh5.py

@@ -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()

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",
 ]

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
 
-# 已处理订单（防重复）
+# 已处理订单（防重复），24h TTL，最大 10000 条，避免内存泄漏
-processed_orders: set[str] = set()
+_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

requirements.txt

@@ -1,10 +1,10 @@
 fastapi
 uvicorn
 python-dotenv
-sqlalchemy
 pydantic
 python-multipart
 pocketbase
 requests
 minio
 playwright
+cachetools

run.py

@@ -13,7 +13,7 @@ from dotenv import load_dotenv
 load_dotenv()
 
 if __name__ == "__main__":
-    port = int(os.getenv("PORT", "8007"))
+    port = int(os.getenv("PORT", "8008"))
     reload_enabled = os.getenv("UVICORN_RELOAD", "0").strip().lower() in ("1", "true", "yes")
     try:
         sys.stderr.write(f"\n[salonapi] startup http://0.0.0.0:{port}\n")