503 lines
19 KiB
Python
503 lines
19 KiB
Python
"""
|
||
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")
|