's'

app/routers/_digital_base.py

@@ -2,6 +2,7 @@
 digital/cnomadcna/nomadlms 共享的支付逻辑
 根据 prefix 生成对应 notify_path，业务互不影响
 """
+import asyncio
 import logging
 import random
 import string
@@ -76,6 +77,7 @@ def create_digital_router(prefix: str, project_name: str, site_id: str | None =
         type_map = {
             "book": "购买书籍",
             "meetup": "数字游民社区报名",
+            "salon": "沙龙报名茶歇费",
             "video": "视频解锁",
             "vip": "VIP会员充值",
         }
@@ -156,6 +158,7 @@ def create_digital_router(prefix: str, project_name: str, site_id: str | None =
         type_map = {
             "book": "购买书籍",
             "meetup": "数字游民社区报名",
+            "salon": "沙龙报名茶歇费",
             "video": "视频解锁",
             "vip": "VIP会员充值",
         }
@@ -215,43 +218,46 @@ def create_digital_router(prefix: str, project_name: str, site_id: str | None =
         if not order_id:
             return {"paid": False, "error": "missing order_id"}
 
-        try:
-            from ..services.pb import get_pb_client
+        def _query_pb() -> dict | None:
+            try:
+                from ..services.pb import get_pb_client
+                pb = get_pb_client()
+                records = pb.collection("payments").get_list(1, 1, {"filter": f'order_id = "{order_id}"'})
+                if records.items:
+                    return {"paid": True, "source": "pocketbase"}
+            except Exception as error:
+                _safe_log(logging.WARNING, "%s order_status PocketBase failed: %s", project_name, error)
+            return None
 
-            pb = get_pb_client()
-            records = pb.collection("payments").get_list(
-                1, 1, {"filter": f'order_id = "{order_id}"'}
-            )
-            if records.items:
-                return {"paid": True, "source": "pocketbase"}
-        except Exception as error:
-            _safe_log(logging.WARNING, "%s order_status PocketBase failed: %s", project_name, error)
+        def _query_xorpay() -> dict | None:
+            try:
+                xorpay = get_payment_provider_by_name("xorpay")
+                r = xorpay.query_order_status(order_id, timeout=6)
+                if r.get("paid"):
+                    return {"paid": True, "source": "xorpay", "status": r.get("status", "")}
+            except Exception as error:
+                _safe_log(logging.WARNING, "%s order_status XorPay failed: %s", project_name, error)
+            return None
 
-        try:
-            xorpay = get_payment_provider_by_name("xorpay")
-            xorpay_result = xorpay.query_order_status(order_id, timeout=8)
-            if xorpay_result.get("paid"):
-                return {
-                    "paid": True,
-                    "source": "xorpay",
-                    "status": xorpay_result.get("status", ""),
-                }
-        except Exception as error:
-            _safe_log(logging.WARNING, "%s order_status XorPay failed: %s", project_name, error)
-
-        try:
-            zpay = get_payment_provider_by_name("zpay")
-            if isinstance(zpay, ZPayProvider):
-                zpay_result = zpay.query_order_status(order_id, timeout=8)
-                if zpay_result.get("paid"):
-                    return {
-                        "paid": True,
-                        "source": "zpay",
-                        "status": zpay_result.get("status", ""),
-                    }
-        except Exception as error:
-            _safe_log(logging.WARNING, "%s order_status ZPAY failed: %s", project_name, error)
+        def _query_zpay() -> dict | None:
+            try:
+                zpay = get_payment_provider_by_name("zpay")
+                if isinstance(zpay, ZPayProvider):
+                    r = zpay.query_order_status(order_id, timeout=6)
+                    if r.get("paid"):
+                        return {"paid": True, "source": "zpay", "status": r.get("status", "")}
+            except Exception as error:
+                _safe_log(logging.WARNING, "%s order_status ZPAY failed: %s", project_name, error)
+            return None
 
+        # 并行查询 PB、ZPAY、XorPay，任一返回 paid 即立即返回（PC 扫码用 ZPAY，微信用 XorPay）
+        pb_task = asyncio.to_thread(_query_pb)
+        xorpay_task = asyncio.to_thread(_query_xorpay)
+        zpay_task = asyncio.to_thread(_query_zpay)
+        results = await asyncio.gather(pb_task, xorpay_task, zpay_task)
+        for r in results:
+            if r:
+                return r
         return {"paid": False}
 
     @router.post("/xorpay_notify")

app/routers/meetup.py

@@ -193,6 +193,7 @@ async def ensure_user(item: EnsureUserRequest):
                 "email": email,
                 "password": DEFAULT_PASSWORD,
                 "passwordConfirm": DEFAULT_PASSWORD,
+                "live_allowed": True,
             },
             headers={
                 "Content-Type": "application/json",

app/routers/salon.py

@@ -0,0 +1,4 @@
+"""salon 项目接口：克隆 cnomadcna 的支付与 meetup 流程，使用 site_id=salon"""
+from ._digital_base import create_digital_router
+
+router = create_digital_router(prefix="/salon", project_name="salon", site_id="salon")

app/routers/salon_meetup.py

@@ -0,0 +1,502 @@
+"""
+salon 加入流程：check-user、ensure-user、complete-order
+克隆 meetup 逻辑，使用 site_id=salon
+"""
+import logging
+import time
+
+import requests
+from fastapi import APIRouter, HTTPException
+from pydantic import BaseModel
+
+from ..config import PB_URL, PB_ADMIN_EMAIL, PB_ADMIN_PASSWORD
+
+SITE_ID = "salon"
+router = APIRouter(prefix="/api/salon", tags=["salon"])
+DEFAULT_PASSWORD = "12345678"
+
+_SALON_ADMIN_TOKEN_CACHE: tuple[str, float] | None = None
+_TOKEN_CACHE_TTL = 300
+
+
+def _get_admin_token() -> tuple[str | None, str | None]:
+    global _SALON_ADMIN_TOKEN_CACHE
+    now = time.time()
+    if _SALON_ADMIN_TOKEN_CACHE and _SALON_ADMIN_TOKEN_CACHE[1] > now:
+        return _SALON_ADMIN_TOKEN_CACHE[0], None
+
+    if not PB_ADMIN_EMAIL or not PB_ADMIN_PASSWORD:
+        return None, "PB_ADMIN_EMAIL 或 PB_ADMIN_PASSWORD 未配置"
+    base = (PB_URL or "").rstrip("/")
+    for path in ["/api/collections/_superusers/auth-with-password", "/api/admins/auth-with-password"]:
+        try:
+            r = requests.post(
+                f"{base}{path}",
+                json={"identity": PB_ADMIN_EMAIL, "password": PB_ADMIN_PASSWORD},
+                timeout=4,
+            )
+            if r.status_code == 200:
+                data = r.json()
+                token = data.get("token")
+                if token:
+                    _SALON_ADMIN_TOKEN_CACHE = (token, now + _TOKEN_CACHE_TTL)
+                return token, None
+            if r.status_code == 404:
+                continue
+            try:
+                msg = r.json().get("message", r.text[:100])
+            except Exception:
+                msg = r.text[:100]
+            logging.warning(f"PocketBase admin auth failed: status={r.status_code}, msg={msg}")
+            return None, f"PocketBase 管理员认证失败: {msg}"
+        except requests.exceptions.ConnectionError as e:
+            logging.error(f"PocketBase 连接失败: {e}")
+            return None, f"无法连接 {PB_URL}，请检查网络或 PB_URL"
+        except Exception as e:
+            logging.warning(f"PocketBase auth try {path}: {e}")
+    return None, "PocketBase 管理员认证失败: 请确认 PB_URL 及管理员账号密码正确"
+
+
+class CheckUserRequest(BaseModel):
+    email: str
+
+
+class EnsureUserRequest(BaseModel):
+    email: str
+    password: str | None = None
+
+
+def _check_site_vip(user_id: str, token: str | None = None) -> bool:
+    if not token:
+        token, _ = _get_admin_token()
+    if not token:
+        return False
+    try:
+        r = requests.get(
+            f"{PB_URL}/api/collections/site_vip/records",
+            params={
+                "filter": f'user_id = "{user_id}" && site_id = "{SITE_ID}"',
+                "perPage": 1,
+                "sort": "-expires_at",
+            },
+            headers={"Authorization": f"Bearer {token}"},
+            timeout=4,
+        )
+        if r.status_code != 200:
+            return False
+        data = r.json()
+        items = data.get("items") or []
+        if not items:
+            return False
+        rec = items[0]
+        exp = rec.get("expires_at")
+        if not exp:
+            return True
+        try:
+            from datetime import datetime as _dt, timezone
+            exp_dt = _dt.fromisoformat(exp.replace("Z", "+00:00"))
+            now = _dt.now(timezone.utc)
+            if exp_dt.tzinfo is None:
+                exp_dt = exp_dt.replace(tzinfo=timezone.utc)
+            return exp_dt > now
+        except Exception:
+            return True
+    except Exception:
+        return False
+
+
+@router.post("/check-user")
+async def check_user(item: CheckUserRequest):
+    email = (item.email or "").strip()
+    if not email:
+        raise HTTPException(status_code=400, detail="请输入邮箱")
+
+    token, err = _get_admin_token()
+    if not token:
+        raise HTTPException(status_code=500, detail=f"服务配置错误：{err}")
+
+    try:
+        filter_val = f'email="{email}"'
+        r = requests.get(
+            f"{PB_URL}/api/collections/users/records",
+            params={"filter": filter_val, "perPage": 1},
+            headers={"Authorization": f"Bearer {token}"},
+            timeout=4,
+        )
+        if r.status_code != 200:
+            raise HTTPException(status_code=500, detail="查询失败")
+        data = r.json()
+        items = data.get("items") or []
+        exists = len(items) > 0
+        user_id = items[0].get("id") if items else None
+        vip = _check_site_vip(user_id, token) if user_id else False
+        return {"ok": True, "exists": exists, "vip": vip, "user_id": user_id}
+    except HTTPException:
+        raise
+    except Exception as e:
+        logging.error(f"salon check-user error: {e}")
+        raise HTTPException(status_code=500, detail="操作失败")
+
+
+@router.post("/ensure-user")
+async def ensure_user(item: EnsureUserRequest):
+    email = (item.email or "").strip()
+    if not email:
+        raise HTTPException(status_code=400, detail="请输入邮箱")
+
+    password = (item.password or "").strip() or DEFAULT_PASSWORD
+
+    try:
+        login_r = requests.post(
+            f"{PB_URL}/api/collections/users/auth-with-password",
+            json={"identity": email, "password": password},
+            timeout=10,
+        )
+        if login_r.status_code == 200:
+            data = login_r.json()
+            return {
+                "ok": True,
+                "user_id": data.get("record", {}).get("id"),
+                "token": data.get("token"),
+                "record": data.get("record"),
+                "is_new": False,
+            }
+
+        try:
+            err_data = login_r.json()
+        except Exception:
+            err_data = {}
+        is_not_found = login_r.status_code == 400 and (
+            "Invalid" in str(err_data.get("message", ""))
+            or "identity" in str(err_data.get("message", "")).lower()
+        )
+        if not is_not_found and item.password:
+            raise HTTPException(status_code=400, detail="密码错误")
+
+        token, err = _get_admin_token()
+        if not token:
+            raise HTTPException(status_code=500, detail=f"服务配置错误：{err}")
+
+        create_r = requests.post(
+            f"{PB_URL}/api/collections/users/records",
+            json={
+                "email": email,
+                "password": DEFAULT_PASSWORD,
+                "passwordConfirm": DEFAULT_PASSWORD,
+                "live_allowed": True,
+            },
+            headers={
+                "Content-Type": "application/json",
+                "Authorization": f"Bearer {token}",
+            },
+            timeout=10,
+        )
+        if create_r.status_code != 200:
+            try:
+                create_err = create_r.json()
+            except Exception:
+                create_err = {}
+            data = create_err.get("data", {})
+            if (
+                create_r.status_code == 400
+                and "already" in str(data.get("email", {}).get("message", "")).lower()
+            ):
+                raise HTTPException(status_code=400, detail="该邮箱已注册，请输入密码登录")
+            msg = create_err.get("message", "创建账号失败")
+            logging.warning(f"salon ensure-user create failed: status={create_r.status_code}, msg={msg}")
+            raise HTTPException(status_code=500, detail=msg)
+
+        final_r = requests.post(
+            f"{PB_URL}/api/collections/users/auth-with-password",
+            json={"identity": email, "password": DEFAULT_PASSWORD},
+            timeout=10,
+        )
+        if final_r.status_code != 200:
+            raise HTTPException(status_code=500, detail="登录失败")
+
+        auth_data = final_r.json()
+        return {
+            "ok": True,
+            "user_id": auth_data.get("record", {}).get("id"),
+            "token": auth_data.get("token"),
+            "record": auth_data.get("record"),
+            "is_new": True,
+        }
+    except HTTPException:
+        raise
+    except Exception as e:
+        logging.error(f"salon ensure-user error: {e}")
+        raise HTTPException(status_code=500, detail="操作失败")
+
+
+def _decode_pending_email(user_id: str) -> str | None:
+    if not user_id.startswith("pending_"):
+        return None
+    try:
+        return bytes.fromhex(user_id[8:]).decode("utf-8")
+    except Exception:
+        return None
+
+
+def _query_zpay_paid(order_id: str) -> bool:
+    try:
+        from ..payment import get_payment_provider
+        from ..payment.zpay import ZPayProvider
+        provider = get_payment_provider()
+        if not isinstance(provider, ZPayProvider):
+            return False
+        result = provider.query_order_status(order_id, timeout=10)
+        return bool(result.get("paid"))
+    except Exception as e:
+        logging.warning(f"salon complete-order zpay query failed: {e}")
+        return False
+
+
+@router.post("/complete-order")
+async def complete_order(item: dict):
+    order_id = (item.get("order_id") or "").strip()
+    if not order_id:
+        raise HTTPException(status_code=400, detail="缺少 order_id")
+
+    base = (PB_URL or "").rstrip("/")
+    token, err = _get_admin_token()
+    if not token:
+        raise HTTPException(status_code=500, detail=f"服务配置错误：{err}")
+
+    from ..services.payment import parse_order_id
+
+    user_id, pay_type = parse_order_id(order_id)
+    if not user_id or pay_type != "salon":
+        raise HTTPException(status_code=400, detail="订单格式无效")
+
+    max_retries = 4
+    for attempt in range(max_retries):
+        r = requests.get(
+            f"{base}/api/collections/site_vip/records",
+            params={"filter": f'order_id = "{order_id}" && site_id = "{SITE_ID}"', "perPage": 1},
+            headers={"Authorization": f"Bearer {token}"},
+            timeout=10,
+        )
+        total = (r.json().get("totalItems") or 0) if r.status_code == 200 else 0
+        if total > 0:
+            break
+        if attempt < max_retries - 1:
+            paid_via = None
+            pay_price = ""
+            if _query_zpay_paid(order_id):
+                paid_via = "zpay"
+                try:
+                    from ..payment.zpay import ZPayProvider
+                    from ..payment import get_payment_provider
+                    p = get_payment_provider()
+                    if isinstance(p, ZPayProvider):
+                        r = p.query_order_status(order_id, timeout=8)
+                        if r.get("paid"):
+                            pay_price = str(r.get("pay_price", "") or r.get("money", ""))
+                except Exception:
+                    pass
+            if not paid_via:
+                try:
+                    from ..payment.xorpay import XorPayProvider
+                    from ..payment import get_payment_provider
+                    p = get_payment_provider()
+                    if isinstance(p, XorPayProvider):
+                        r = p.query_order_status(order_id, timeout=8)
+                        if r.get("paid"):
+                            paid_via = "xorpay"
+                            pay_price = str(r.get("pay_price", ""))
+                except Exception:
+                    pass
+            if paid_via:
+                from ..services.payment import handle_payment_success
+                # pay_price 为空时用默认茶歇费 1 元
+                price = pay_price if pay_price and float(pay_price or 0) > 0 else "1"
+                try:
+                    handle_payment_success(
+                        order_id,
+                        price,
+                        f"salon-{paid_via}-poll",
+                        use_memos=False,
+                        site_id=SITE_ID,
+                    )
+                    time.sleep(1)
+                    continue
+                except Exception as e:
+                    logging.warning(f"salon complete-order 手动触发支付处理失败: {e}")
+            if paid_via:
+                time.sleep(2)
+                continue
+        resp_preview = r.text[:200] if r.ok else f"status={r.status_code}"
+        logging.warning(f"salon complete-order 未找到订单: order_id={order_id}, attempt={attempt + 1}, pb_resp={resp_preview}")
+        raise HTTPException(status_code=400, detail="订单不存在或未支付成功")
+
+    if user_id.startswith("pending_"):
+        email = _decode_pending_email(user_id)
+        if not email:
+            raise HTTPException(status_code=400, detail="订单格式无效")
+        login_r = requests.post(
+            f"{base}/api/collections/users/auth-with-password",
+            json={"identity": email, "password": DEFAULT_PASSWORD},
+            timeout=10,
+        )
+        if login_r.status_code != 200:
+            raise HTTPException(status_code=500, detail="登录失败，请稍后重试")
+        auth_data = login_r.json()
+        return {
+            "ok": True,
+            "token": auth_data.get("token"),
+            "record": auth_data.get("record"),
+            "is_new": True,
+        }
+
+    return {"ok": True, "token": None, "record": None, "is_new": False}
+
+
+@router.get("/vip/check")
+async def vip_check(user_id: str = ""):
+    """检查用户是否为 salon VIP"""
+    if not user_id:
+        return {"vip": False}
+    token, _ = _get_admin_token()
+    return {"vip": _check_site_vip(user_id, token)}
+
+
+@router.get("/join/by-wechat")
+async def join_by_wechat(wechat_id: str = ""):
+    """按微信号查询 solanRed 记录，供 salon 前端提交后刷新状态"""
+    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()
+        solan = pb_client.collection("solanRed")
+        safe_wechat = wechat_id.strip().replace("\\", "\\\\").replace('"', '\\"')
+        existing = solan.get_list(1, 1, {"filter": f'wechatId = "{safe_wechat}"', "sort": "-created"})
+        if not existing.items:
+            return {"hasRecord": False, "record": None, "isComplete": False, "isWechatFriend": False, "isApproved": False, "isRejected": False, "vip": False}
+        rec = existing.items[0]
+        amount = rec.get("amount") or 0
+        is_complete = bool(rec.get("order_id")) and (amount or 0) > 0
+        return {
+            "hasRecord": True,
+            "record": rec,
+            "isComplete": is_complete,
+            "isWechatFriend": bool(rec.get("is_wechat_friend")),
+            "isApproved": bool(rec.get("is_approved")),
+            "isRejected": bool(rec.get("is_rejected")),
+            "vip": bool(rec.get("vip")),
+        }
+    except Exception as e:
+        logging.warning("salon join/by-wechat error: %s", e)
+        return {"hasRecord": False, "record": None, "isComplete": False, "isWechatFriend": False, "isApproved": False, "isRejected": False, "vip": False}
+
+
+@router.get("/join/status")
+async def join_status(user_id: str = ""):
+    """检查用户是否已提交 salon 加入申请"""
+    if not user_id:
+        return {"hasRecord": False, "isComplete": False}
+    try:
+        from ..services import get_pb_client
+        pb_client = get_pb_client()
+        solan = pb_client.collection("solanRed")
+        existing = solan.get_list(1, 1, {"filter": f'user_id = "{user_id}"', "sort": "-created"})
+        has_record = len(existing.items) > 0
+        rec = existing.items[0] if existing.items else None
+        is_complete = bool(rec and rec.get("order_id") and rec.get("amount", 0) > 0)
+        return {"hasRecord": has_record, "isComplete": is_complete}
+    except Exception:
+        return {"hasRecord": False, "isComplete": False}
+
+
+def _validate_xiaohongshu_url(url: str) -> dict | None:
+    """校验小红书主页 URL 并尝试爬取用户名/头像/粉丝数/性别。返回 None 表示校验失败。"""
+    url = (url or "").strip()
+    if not url:
+        return None
+    if "xiaohongshu.com" not in url.lower():
+        return None
+    if not url.startswith("http://") and not url.startswith("https://"):
+        return None
+    try:
+        r = requests.get(url, timeout=10, headers={"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36"})
+        if r.status_code != 200:
+            return None
+        html = r.text
+        nickname = ""
+        avatar = ""
+        followers = 0
+        gender = ""
+        import re
+        if '"nickname"' in html or '"user"' in html:
+            m = re.search(r'"nickname"\s*:\s*"([^"]+)"', html)
+            if m:
+                nickname = m.group(1)
+            m = re.search(r'"avatar"\s*:\s*"([^"]+)"', html)
+            if m:
+                avatar = m.group(1).replace("\\u002F", "/")
+            m = re.search(r'"follows"\s*:\s*(\d+)', html)
+            if m:
+                followers = int(m.group(1))
+        if not nickname and '"desc"' in html:
+            m = re.search(r'"desc"\s*:\s*"([^"]+)"', html)
+            if m:
+                nickname = m.group(1)[:20]
+        return {"nickname": nickname or "用户", "avatar": avatar, "followers": followers, "gender": gender, "url": url}
+    except Exception as e:
+        logging.warning(f"xiaohongshu crawl failed: {e}")
+        return {"nickname": "用户", "avatar": "", "followers": 0, "gender": "", "url": url}
+
+
+@router.post("/join")
+async def submit_join(item: dict):
+    """salon 加入社区报名表单，写入 solanRed 集合。需提供 xiaohongshu_url，提交前校验并爬取信息。"""
+    user_id = item.get("user_id")
+    if not user_id:
+        raise HTTPException(status_code=400, detail="missing user_id")
+
+    xiaohongshu_url = (item.get("xiaohongshu_url") or "").strip()
+    if not xiaohongshu_url:
+        raise HTTPException(status_code=400, detail="请填写小红书主页地址")
+
+    profile = _validate_xiaohongshu_url(xiaohongshu_url)
+    if not profile:
+        raise HTTPException(status_code=400, detail="小红书主页地址无效或无法访问，请检查后重试")
+
+    try:
+        from ..services import get_pb_client
+        pb_client = get_pb_client()
+        solan = pb_client.collection("solanRed")
+        wechat_id = (item.get("wechatId") or "").strip()
+        if wechat_id:
+            safe_wechat = wechat_id.replace("\\", "\\\\").replace('"', '\\"')
+            existing = solan.get_list(1, 1, {"filter": f'wechatId = "{safe_wechat}"'})
+            if existing.items:
+                raise HTTPException(status_code=400, detail="该微信号已报名，请勿重复提交")
+        pb_data = {
+            "user_id": user_id,
+            "nickname": item.get("nickname", ""),
+            "occupation": item.get("occupation", ""),
+            "reason": item.get("reason", ""),
+            "wechatId": item.get("wechatId", ""),
+            "gender": profile.get("gender", "") or item.get("gender", ""),
+            "education": item.get("education", ""),
+            "gradYear": item.get("gradYear", ""),
+            "phone": item.get("phone", ""),
+            "single": item.get("single", ""),
+            "media": item.get("media", ""),
+            "calculated_age": item.get("calculated_age", 0),
+            "type": item.get("type", ""),
+            "order_id": "",
+            "amount": 0,
+            "xiaohongshu_url": xiaohongshu_url,
+            "xiaohongshu_nickname": profile.get("nickname", ""),
+            "xiaohongshu_avatar": profile.get("avatar", ""),
+            "xiaohongshu_followers": profile.get("followers", 0),
+        }
+        record = solan.create(pb_data)
+        logging.info(f"salon 申请表单提交成功 - user_id: {user_id}, record_id: {record.id}")
+        return {"ok": True, "user_id": user_id, "record_id": record.id}
+    except Exception as e:
+        logging.error(f"salon join 提交失败: {e}")
+        raise HTTPException(status_code=500, detail="submit failed")