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