1541 lines
61 KiB
Python
1541 lines
61 KiB
Python
"""
|
||
salon 加入流程：check-user、ensure-user、complete-order
|
||
克隆 meetup 逻辑，使用 site_id=salon
|
||
"""
|
||
import asyncio
|
||
import json
|
||
import logging
|
||
import re
|
||
import time
|
||
from urllib.parse import parse_qs, unquote, urlparse
|
||
|
||
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
|
||
_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",
|
||
}
|
||
|
||
|
||
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,
|
||
},
|
||
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_order(order_id: str) -> dict | None:
|
||
"""查询 ZPAY 订单状态，返回 {paid, money}。使用 get_payment_provider_by_name 确保查询 ZPAY，不依赖默认 provider"""
|
||
try:
|
||
from ..payment import get_payment_provider_by_name
|
||
from ..payment.zpay import ZPayProvider
|
||
provider = get_payment_provider_by_name("zpay")
|
||
if not isinstance(provider, ZPayProvider):
|
||
return None
|
||
result = provider.query_order_status(order_id, timeout=10)
|
||
return result if result.get("paid") else None
|
||
except Exception as e:
|
||
logging.warning(f"salon complete-order zpay query failed: {e}")
|
||
return None
|
||
|
||
|
||
def _query_xorpay_paid(order_id: str) -> tuple[bool, str]:
|
||
"""查询 XorPay 是否已支付，返回 (paid, pay_price)。微信内订单走 XorPay，需显式查 xorpay 不依赖默认 provider"""
|
||
try:
|
||
from ..payment import get_payment_provider_by_name
|
||
from ..payment.xorpay import XorPayProvider
|
||
provider = get_payment_provider_by_name("xorpay")
|
||
if not isinstance(provider, XorPayProvider):
|
||
return False, ""
|
||
result = provider.query_order_status(order_id, timeout=8)
|
||
if result.get("paid"):
|
||
return True, str(result.get("pay_price", "") or "")
|
||
return False, ""
|
||
except Exception as e:
|
||
logging.warning(f"salon complete-order xorpay query failed: {e}")
|
||
return False, ""
|
||
|
||
|
||
def _update_solan_red_on_payment(token: str, user_id: str, order_id: str, amount_cents: int) -> None:
|
||
"""支付成功后兜底更新 solanRed：order_id、amount、vip"""
|
||
base = (PB_URL or "").rstrip("/")
|
||
try:
|
||
r = requests.get(
|
||
f"{base}/api/collections/solanRed/records",
|
||
params={"filter": f'user_id = "{user_id}"', "sort": "-created", "perPage": 1},
|
||
headers={"Authorization": f"Bearer {token}"},
|
||
timeout=(8, 20),
|
||
)
|
||
if r.status_code != 200:
|
||
return
|
||
data = r.json()
|
||
items = data.get("items") or []
|
||
if not items:
|
||
return
|
||
rec_id = items[0].get("id")
|
||
if not rec_id:
|
||
return
|
||
patch = {"order_id": order_id, "amount": amount_cents, "vip": True}
|
||
requests.patch(
|
||
f"{base}/api/collections/solanRed/records/{rec_id}",
|
||
json=patch,
|
||
headers={"Content-Type": "application/json", "Authorization": f"Bearer {token}"},
|
||
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}")
|
||
|
||
|
||
@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="订单格式无效")
|
||
|
||
def _query_solan_red_paid() -> dict | None:
|
||
"""查 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=(8, 20),
|
||
)
|
||
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, (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 and xorpay_paid:
|
||
paid_via = "xorpay"
|
||
pay_price = xorpay_price
|
||
if paid_via:
|
||
money_y = float(pay_price or 0) if pay_price else 0
|
||
cached_amount_cents = int(money_y * 100) if money_y > 0 else 100
|
||
from ..services.payment import handle_payment_success, try_add_processed_order, remove_processed_order
|
||
if try_add_processed_order(order_id):
|
||
price = pay_price if pay_price and float(pay_price or 0) > 0 else "1"
|
||
try:
|
||
await asyncio.to_thread(
|
||
handle_payment_success,
|
||
order_id,
|
||
price,
|
||
f"salon-{paid_via}-poll",
|
||
use_memos=False,
|
||
site_id=SITE_ID,
|
||
)
|
||
await asyncio.sleep(0.3) # 短暂等待 PB 写入完成
|
||
continue
|
||
except Exception as e:
|
||
remove_processed_order(order_id)
|
||
logging.warning(f"salon complete-order 手动触发支付处理失败: {e}")
|
||
await asyncio.sleep(1)
|
||
continue
|
||
logging.warning(f"salon complete-order 未找到订单: order_id={order_id}, attempt={attempt + 1}")
|
||
raise HTTPException(status_code=400, detail="订单不存在或未支付成功")
|
||
|
||
# 兜底更新 solanRed：回调可能未到，确保 order_id、amount、vip 正确（按 user_id 查找，pending_ 用户记录存的是 pending_xxx）
|
||
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
|
||
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=(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="登录失败，请稍后重试")
|
||
return {
|
||
"ok": True,
|
||
"token": auth_data.get("token"),
|
||
"record": auth_data.get("record"),
|
||
"is_new": True,
|
||
}
|
||
|
||
# 非 pending 用户：返回 solanRed 记录供前端显示微信号
|
||
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=(6, 12),
|
||
)
|
||
if r.status_code == 200:
|
||
items = r.json().get("items") or []
|
||
if items:
|
||
rec = items[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 fetch_solan_record 请求超时/失败: {e}")
|
||
return None
|
||
|
||
record = await asyncio.to_thread(_fetch_solan_record)
|
||
return {"ok": True, "token": None, "record": record, "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/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
|
||
_get = lambda r, k, d=None: getattr(r, k, d) if r else d
|
||
is_complete = bool(rec and _get(rec, "order_id") and (_get(rec, "amount", 0) or 0) > 0)
|
||
return {"hasRecord": has_record, "isComplete": is_complete}
|
||
except Exception:
|
||
return {"hasRecord": False, "isComplete": False}
|
||
|
||
|
||
def _query_by_wechat_sync(wechat_id: str) -> dict:
|
||
"""同步按 wechatId 查询 solanRed，供 asyncio.to_thread 调用"""
|
||
try:
|
||
from ..services import get_pb_client
|
||
pb_client = get_pb_client()
|
||
solan = pb_client.collection("solanRed")
|
||
w = wechat_id.strip().replace("\\", "\\\\").replace('"', '\\"')
|
||
existing = solan.get_list(1, 1, {"filter": f'wechatId = "{w}"', "sort": "-created"})
|
||
rec = existing.items[0] if existing.items else None
|
||
has_record = rec is not None
|
||
_get = lambda r, k, d=None: getattr(r, k, d) if r else d
|
||
amount = int(_get(rec, "amount", 0) or 0) if rec else 0
|
||
is_complete = has_record and bool(_get(rec, "order_id")) and amount > 0
|
||
is_wechat_friend = bool(_get(rec, "is_wechat_friend")) if rec else False
|
||
is_approved = bool(_get(rec, "is_approved")) if rec else False
|
||
is_rejected = bool(_get(rec, "is_rejected")) if rec else False
|
||
vip = bool(_get(rec, "vip")) if rec else False
|
||
record_dict = None
|
||
if rec:
|
||
record_dict = {
|
||
"id": _get(rec, "id"),
|
||
"user_id": _get(rec, "user_id"),
|
||
"wechatId": _get(rec, "wechatId"),
|
||
"nickname": _get(rec, "nickname"),
|
||
"media": _get(rec, "media"),
|
||
"amount": amount,
|
||
"order_id": _get(rec, "order_id"),
|
||
"vip": vip,
|
||
"is_volunteer": bool(_get(rec, "is_volunteer")),
|
||
}
|
||
return {
|
||
"hasRecord": has_record,
|
||
"record": record_dict,
|
||
"isComplete": is_complete,
|
||
"isWechatFriend": is_wechat_friend,
|
||
"isApproved": is_approved,
|
||
"isRejected": is_rejected,
|
||
"vip": vip,
|
||
}
|
||
except Exception as e:
|
||
logging.warning(f"salon join by-wechat 查询失败: {e}")
|
||
return {"hasRecord": False, "record": None, "isComplete": False, "isWechatFriend": False, "isApproved": False, "isRejected": False, "vip": False}
|
||
|
||
|
||
@router.get("/join/by-wechat")
|
||
async def join_by_wechat(wechat_id: str = ""):
|
||
"""按 wechatId 查询 solanRed 记录"""
|
||
if not wechat_id or not wechat_id.strip():
|
||
return {"hasRecord": False, "record": None, "isComplete": False, "isWechatFriend": False, "isApproved": False, "isRejected": False, "vip": False}
|
||
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}
|
||
try:
|
||
r = requests.get(
|
||
f"{base}/api/collections/solanRed/records",
|
||
params={
|
||
"filter": 'is_volunteer = true && is_approved = true',
|
||
"sort": "-created",
|
||
"perPage": 1,
|
||
},
|
||
headers={"Authorization": f"Bearer {token}"},
|
||
timeout=10,
|
||
)
|
||
if r.status_code != 200:
|
||
return {"volunteer": None}
|
||
data = r.json()
|
||
items = data.get("items") or []
|
||
volunteer = None
|
||
if items:
|
||
rec = items[0]
|
||
nickname = str(rec.get("xiaohongshu_nickname") or rec.get("nickname") or "").strip() or "志愿者"
|
||
media = str(rec.get("media") or "").strip() or None
|
||
xiaohongshu_url = str(rec.get("xiaohongshu_url") or "").strip() or None
|
||
volunteer = {
|
||
"nickname": nickname,
|
||
"avatar": media if media else None,
|
||
"xiaohongshu_url": xiaohongshu_url if xiaohongshu_url else None,
|
||
}
|
||
return {"volunteer": volunteer}
|
||
except Exception as e:
|
||
logging.warning(f"salon join volunteers 查询失败: {e}")
|
||
return {"volunteer": None}
|
||
|
||
|
||
@router.get("/join/volunteers")
|
||
async def join_volunteers():
|
||
"""获取志愿者列表 - solanRed 中 is_volunteer=true && is_approved=true"""
|
||
token, _ = _get_admin_token()
|
||
if not token:
|
||
return {"volunteer": None}
|
||
return await asyncio.to_thread(_fetch_volunteers_sync, token)
|
||
|
||
|
||
class XiaohongshuProfileRequest(BaseModel):
|
||
url: str | 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:
|
||
if not url or depth > 3:
|
||
return url
|
||
|
||
url = str(url).strip()
|
||
if _XHS_PROFILE_PATTERN.match(url):
|
||
return url
|
||
|
||
try:
|
||
parsed = urlparse(url)
|
||
host = (parsed.netloc or "").lower()
|
||
if host.endswith("xiaohongshu.com") and parsed.path.startswith("/login"):
|
||
query = parse_qs(parsed.query)
|
||
redirect = query.get("redirectPath", [None])[0] or query.get("redirect_path", [None])[0]
|
||
if redirect:
|
||
redirect = unquote(redirect).strip()
|
||
if redirect.startswith("/"):
|
||
redirect = f"{parsed.scheme or 'https'}://{parsed.netloc}{redirect}"
|
||
return _normalize_xiaohongshu_profile_url(redirect, depth + 1)
|
||
except Exception:
|
||
return url
|
||
|
||
return url
|
||
|
||
|
||
def _resolve_xhslink(url: str) -> str:
|
||
"""解析 xhslink.com 短链为真实 profile URL（m/ 为主页短链）"""
|
||
if not _XHS_SHORT_PATTERN.match(url):
|
||
return _normalize_xiaohongshu_profile_url(url)
|
||
try:
|
||
r = requests.get(
|
||
url,
|
||
allow_redirects=True,
|
||
headers=_XHS_REQUEST_HEADERS,
|
||
timeout=10,
|
||
)
|
||
return _normalize_xiaohongshu_profile_url(r.url or url)
|
||
except Exception:
|
||
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")
|
||
if raw is None:
|
||
raw = value.get("_value")
|
||
return raw
|
||
return value
|
||
|
||
|
||
def _sanitize_state_json(raw: str) -> str:
|
||
raw = re.sub(r":\s*undefined\b", ": null", raw)
|
||
raw = re.sub(r":\s*NaN\b", ": null", raw)
|
||
raw = re.sub(r":\s*-?Infinity\b", ": null", raw)
|
||
return raw
|
||
|
||
|
||
def _extract_state_from_html(html: str) -> dict | None:
|
||
if not html:
|
||
return None
|
||
m = re.search(r"__INITIAL_STATE__\s*=\s*(\{.*?\})\s*</script>", html, re.S)
|
||
if not m:
|
||
return None
|
||
try:
|
||
return json.loads(_sanitize_state_json(m.group(1)))
|
||
except Exception as exc:
|
||
logging.debug(f"[xiaohongshu] state parse failed: {exc}")
|
||
return None
|
||
|
||
|
||
def _looks_like_personal_link(value) -> bool:
|
||
if value is None:
|
||
return False
|
||
text = str(value).strip()
|
||
return bool(re.fullmatch(r"[a-zA-Z0-9][-a-zA-Z0-9.]*\.(?:com|cn|net|io|me)", text)) and "xiaohongshu" not in text.lower()
|
||
|
||
|
||
def _extract_avatar_url(value) -> str | None:
|
||
value = _unwrap_ref(value)
|
||
if not isinstance(value, dict):
|
||
return None
|
||
|
||
for key in ("imageb", "images", "avatarUrl", "avatar", "image", "headImage"):
|
||
raw = value.get(key)
|
||
if not raw:
|
||
continue
|
||
text = str(raw).strip()
|
||
if text.startswith("http://") or text.startswith("https://"):
|
||
return text
|
||
|
||
return None
|
||
|
||
|
||
def _parse_count_value(value) -> tuple[int | None, str | None]:
|
||
if value is None:
|
||
return None, None
|
||
if isinstance(value, (int, float)):
|
||
num = int(value)
|
||
return num, str(num)
|
||
|
||
text = str(value).strip()
|
||
if not text or text == "-":
|
||
return None, None
|
||
|
||
cleaned = text.replace(",", "")
|
||
multiplier = 1
|
||
if cleaned.endswith(("万", "w", "W")):
|
||
multiplier = 10000
|
||
cleaned = cleaned[:-1]
|
||
if cleaned.endswith("+"):
|
||
cleaned = cleaned[:-1]
|
||
|
||
m = re.search(r"\d+(?:\.\d+)?", cleaned)
|
||
if not m:
|
||
return None, text
|
||
|
||
num = int(float(m.group()) * multiplier)
|
||
return num, text
|
||
|
||
|
||
def _set_count_fields(out: dict, key: str, raw_value) -> None:
|
||
num, text = _parse_count_value(raw_value)
|
||
if num is not None:
|
||
out[key] = num
|
||
if text:
|
||
out[f"{key}Text"] = text
|
||
|
||
|
||
def _merge_profile(base: dict | None, extra: dict | None) -> dict:
|
||
merged = dict(base or {})
|
||
if not extra:
|
||
return merged
|
||
|
||
count_keys = {"followers", "following", "likesAndCollects", "postCount"}
|
||
|
||
for key, value in extra.items():
|
||
if value is None or value == "":
|
||
continue
|
||
|
||
if key in count_keys:
|
||
current = merged.get(key)
|
||
if key == "postCount":
|
||
if not isinstance(current, int) or value > current:
|
||
merged[key] = value
|
||
continue
|
||
|
||
current_text = merged.get(f"{key}Text")
|
||
extra_text = extra.get(f"{key}Text")
|
||
current_exact = isinstance(current_text, str) and current_text != "" and "+" not in current_text
|
||
extra_exact = isinstance(extra_text, str) and extra_text != "" and "+" not in extra_text
|
||
if current is None or current == 0 or (extra_exact and not current_exact):
|
||
merged[key] = value
|
||
continue
|
||
|
||
if key.endswith("Text"):
|
||
current = merged.get(key)
|
||
current_exact = isinstance(current, str) and current != "" and "+" not in current
|
||
extra_exact = isinstance(value, str) and value != "" and "+" not in value
|
||
if not current or (extra_exact and not current_exact):
|
||
merged[key] = value
|
||
continue
|
||
|
||
if key == "resolvedUrl":
|
||
merged[key] = value
|
||
continue
|
||
|
||
if not merged.get(key):
|
||
merged[key] = value
|
||
|
||
return merged
|
||
|
||
|
||
def _extract_note_stats_from_state(state: dict) -> tuple[int, str | None]:
|
||
user = _unwrap_ref(state.get("user"))
|
||
if not isinstance(user, dict):
|
||
return 0, None
|
||
|
||
notes = _unwrap_ref(user.get("notes"))
|
||
loaded_count = 0
|
||
if isinstance(notes, list):
|
||
for tab in notes:
|
||
tab = _unwrap_ref(tab)
|
||
if isinstance(tab, list):
|
||
loaded_count = max(loaded_count, len(tab))
|
||
|
||
note_queries = _unwrap_ref(user.get("noteQueries"))
|
||
has_more = False
|
||
if isinstance(note_queries, list):
|
||
has_more = any(isinstance(item, dict) and bool(item.get("hasMore")) for item in note_queries)
|
||
|
||
if loaded_count <= 0:
|
||
return 0, None
|
||
return loaded_count, f"{loaded_count}+" if has_more else str(loaded_count)
|
||
|
||
|
||
def _extract_user_from_state(state: dict, target_user_id: str | None) -> dict | None:
|
||
"""从 __INITIAL_STATE__ 提取当前 profile 用户"""
|
||
def check(u):
|
||
u = _unwrap_ref(u)
|
||
if not u or not isinstance(u, dict):
|
||
return None
|
||
uid = str(u.get("user_id") or u.get("userId") or u.get("id") or "").strip()
|
||
if target_user_id and uid and uid != target_user_id:
|
||
return None
|
||
return u
|
||
|
||
user = state.get("user")
|
||
if isinstance(user, dict):
|
||
ui = user.get("userInfo")
|
||
if ui:
|
||
found = check(ui)
|
||
if found:
|
||
return found
|
||
found = check(user)
|
||
if found:
|
||
return found
|
||
|
||
for key in ("userInfo", "userDetail"):
|
||
u = state.get(key)
|
||
found = check(u)
|
||
if found:
|
||
return found
|
||
return None
|
||
|
||
|
||
def _parse_user(user: dict) -> dict | None:
|
||
"""解析用户对象为返回格式"""
|
||
user = _unwrap_ref(user)
|
||
if not isinstance(user, dict):
|
||
return None
|
||
|
||
nickname = str(user.get("nickname") or user.get("nickName") or user.get("name") or user.get("user_name") or "").strip()
|
||
if not nickname:
|
||
return None
|
||
|
||
out = {"nickname": nickname}
|
||
|
||
redbook_id = str(user.get("redbook_id") or user.get("red_id") or user.get("redId") or "").strip()
|
||
if redbook_id:
|
||
out["redbookId"] = redbook_id
|
||
|
||
avatar_url = _extract_avatar_url(user)
|
||
if avatar_url:
|
||
out["avatarUrl"] = avatar_url
|
||
|
||
ip_location = str(user.get("ipLocation") or user.get("ip_location") or "").strip()
|
||
if ip_location:
|
||
out["ipLocation"] = ip_location
|
||
|
||
desc = str(user.get("desc") or user.get("bio") or "").strip()
|
||
if _looks_like_personal_link(desc):
|
||
out["personalLink"] = desc
|
||
|
||
_set_count_fields(out, "followers", user.get("fans_count") or user.get("fansCount") or user.get("followers"))
|
||
_set_count_fields(out, "following", user.get("follow_count") or user.get("followCount"))
|
||
_set_count_fields(out, "likesAndCollects", user.get("collect_count") or user.get("collectCount") or user.get("likes_count") or user.get("likesCount"))
|
||
_set_count_fields(out, "postCount", user.get("notes_count") or user.get("notesCount") or user.get("post_count") or user.get("postCount"))
|
||
return out
|
||
|
||
|
||
def _parse_profile_from_state(state: dict, target_user_id: str | None) -> dict | None:
|
||
if not state or not isinstance(state, dict):
|
||
return None
|
||
|
||
out = {}
|
||
user = _unwrap_ref(state.get("user"))
|
||
user_page_data = None
|
||
if isinstance(user, dict):
|
||
user_page_data = _unwrap_ref(user.get("userPageData"))
|
||
|
||
if isinstance(user_page_data, dict):
|
||
basic_info = _unwrap_ref(user_page_data.get("basicInfo")) or {}
|
||
if isinstance(basic_info, dict):
|
||
nickname = str(basic_info.get("nickname") or basic_info.get("nickName") or "").strip()
|
||
if nickname:
|
||
out["nickname"] = nickname
|
||
|
||
avatar_url = _extract_avatar_url(basic_info)
|
||
if avatar_url:
|
||
out["avatarUrl"] = avatar_url
|
||
|
||
redbook_id = str(basic_info.get("redId") or basic_info.get("redbookId") or "").strip()
|
||
if redbook_id:
|
||
out["redbookId"] = redbook_id
|
||
|
||
ip_location = str(basic_info.get("ipLocation") or "").strip()
|
||
if ip_location:
|
||
out["ipLocation"] = ip_location
|
||
|
||
desc = str(basic_info.get("desc") or "").strip()
|
||
if _looks_like_personal_link(desc):
|
||
out["personalLink"] = desc
|
||
|
||
interactions = _unwrap_ref(user_page_data.get("interactions"))
|
||
if isinstance(interactions, list):
|
||
for item in interactions:
|
||
item = _unwrap_ref(item)
|
||
if not isinstance(item, dict):
|
||
continue
|
||
item_type = str(item.get("type") or "").strip().lower()
|
||
item_name = str(item.get("name") or "").strip()
|
||
target_key = None
|
||
if item_type in {"follows", "follow"} or item_name == "\u5173\u6ce8":
|
||
target_key = "following"
|
||
elif item_type == "fans" or item_name == "\u7c89\u4e1d":
|
||
target_key = "followers"
|
||
elif item_type == "interaction" or item_name == "\u83b7\u8d5e\u4e0e\u6536\u85cf":
|
||
target_key = "likesAndCollects"
|
||
if target_key:
|
||
_set_count_fields(out, target_key, item.get("count"))
|
||
|
||
note_count, note_count_text = _extract_note_stats_from_state(state)
|
||
if note_count > 0:
|
||
out["postCount"] = note_count
|
||
out["postCountText"] = note_count_text or str(note_count)
|
||
|
||
parsed_user = _parse_user(_extract_user_from_state(state, target_user_id) or {})
|
||
out = _merge_profile(out, parsed_user)
|
||
return out or None
|
||
|
||
|
||
def _parse_dom_text(text: str) -> dict:
|
||
"""从页面可见文本解析：关注、粉丝、获赞与收藏、小红书号、IP属地、个人链接"""
|
||
out = {}
|
||
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
|
||
prev = lines[idx - 1]
|
||
if re.fullmatch(r"[-\d.,+wW\u4e07]+", prev):
|
||
_set_count_fields(out, key, prev)
|
||
return
|
||
|
||
pick_count("\u5173\u6ce8", "following")
|
||
pick_count("\u7c89\u4e1d", "followers")
|
||
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("获赞与收藏")
|
||
if i_fen >= 0:
|
||
before = text[max(0, i_fen - 100) : i_fen]
|
||
after = text[i_fen + 2 : i_huo + 80] if i_huo >= 0 else text[i_fen + 2 : i_fen + 80]
|
||
nums_before = [int(m.group()) for m in re.finditer(r"\d+", before)]
|
||
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 and not out.get("following"):
|
||
out["following"] = nums_before[0]
|
||
if nums_after and not out.get("likesAndCollects"):
|
||
out["likesAndCollects"] = nums_after[0]
|
||
|
||
# 笔记/发帖
|
||
m = re.search(r"笔记\s*(\d+)|(\d+)\s*篇|共\s*(\d+)\s*篇", text)
|
||
if m:
|
||
out["postCount"] = int(m.group(1) or m.group(2) or m.group(3) or 0)
|
||
out["postCountText"] = str(out["postCount"])
|
||
|
||
# 小红书号
|
||
m = re.search(r"小红书号[：:]\s*([a-zA-Z0-9_]+)", text)
|
||
if m:
|
||
out["redbookId"] = m.group(1).strip()
|
||
|
||
# IP属地
|
||
m = re.search(r"IP属地[：:]\s*([^\s\n]+)", text)
|
||
if m:
|
||
out["ipLocation"] = m.group(1).strip()
|
||
|
||
# 个人链接
|
||
m = re.search(
|
||
r"(?:小红书号|IP属地)[^\n]*\n\s*([a-zA-Z0-9][-a-zA-Z0-9.]*\.(?:com|cn|net|io|me))\b",
|
||
text,
|
||
re.MULTILINE,
|
||
)
|
||
if m:
|
||
out["personalLink"] = m.group(1).strip()
|
||
else:
|
||
m = re.search(r"\b([a-zA-Z0-9][-a-zA-Z0-9.]*\.(?:com|cn|net|io|me))\b", text[:600])
|
||
if m and "xiaohongshu" not in m.group(1).lower():
|
||
out["personalLink"] = m.group(1).strip()
|
||
|
||
return out
|
||
|
||
|
||
def _extract_nickname_from_body(body_text: str) -> str | None:
|
||
"""从页面正文提取昵称（profile 页通常昵称在首屏靠前）"""
|
||
if not body_text or len(body_text) < 2:
|
||
return None
|
||
lines = [ln.strip() for ln in body_text.split("\n") if ln.strip()]
|
||
skip = {"发现", "发布", "通知", "创作中心", "业务合作", "关注", "笔记", "收藏", "登录", "注册"}
|
||
for ln in lines:
|
||
if len(ln) < 2 or len(ln) > 40:
|
||
continue
|
||
if ln in skip or ln.startswith("http") or ln.isdigit():
|
||
continue
|
||
if "粉丝" in ln or "关注" in ln or "笔记" in ln or "收藏" in ln:
|
||
continue
|
||
if re.match(r"^[\d\s+]+$", ln):
|
||
continue
|
||
return ln
|
||
return None
|
||
|
||
|
||
async def _fetch_xiaohongshu_profile(url: str) -> dict | None:
|
||
"""使用 Playwright 获取小红书用户资料"""
|
||
try:
|
||
from playwright.async_api import async_playwright
|
||
except ImportError:
|
||
logging.warning("playwright 未安装，请执行: pip install playwright && playwright install chromium")
|
||
return None
|
||
|
||
profile_pattern = re.compile(r"/user/profile/([a-zA-Z0-9_-]+)")
|
||
m = profile_pattern.search(url)
|
||
target_user_id = m.group(1) if m else None
|
||
|
||
async with async_playwright() as p:
|
||
browser = await p.chromium.launch(
|
||
headless=True,
|
||
args=["--no-sandbox", "--disable-setuid-sandbox", "--disable-blink-features=AutomationControlled"],
|
||
)
|
||
try:
|
||
context = await browser.new_context(
|
||
viewport={"width": 1280, "height": 800},
|
||
user_agent="Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36",
|
||
locale="zh-CN",
|
||
)
|
||
page = await context.new_page()
|
||
await page.goto(url, wait_until="load", timeout=20000)
|
||
await page.wait_for_timeout(3000)
|
||
|
||
state = await page.evaluate("""() => {
|
||
const w = window;
|
||
return w.__INITIAL_STATE__ || w.__INITIAL_SSR_STATE__ || null;
|
||
}""")
|
||
|
||
body_text = await page.evaluate("""() => document.body ? document.body.innerText : ''""")
|
||
|
||
# 从 profile 区域提取
|
||
dom_stats = await page.evaluate("""() => {
|
||
const nums = (txt) => {
|
||
const arr = [];
|
||
let m;
|
||
const r = /(\\d+)\\s*粉丝|粉丝\\s*(\\d+)/g;
|
||
while ((m = r.exec(txt)) !== null) arr.push(+(m[1]||m[2]||0));
|
||
return arr;
|
||
};
|
||
const body = document.body?.innerText || '';
|
||
const fans = nums(body);
|
||
const follow = (body.match(/(\\d+)\\s*关注|关注\\s*(\\d+)/g) || []).map(s => parseInt(s.replace(/\\D/g,''))).filter(n=>!isNaN(n));
|
||
const likes = (body.match(/(\\d+)\\s*获赞与收藏|获赞与收藏\\s*(\\d+)/g) || []).map(s => parseInt(s.replace(/\\D/g,''))).filter(n=>!isNaN(n));
|
||
return {
|
||
followers: fans.length ? Math.max(...fans) : null,
|
||
following: follow.length ? Math.max(...follow) : null,
|
||
likesAndCollects: likes.length ? Math.max(...likes) : null
|
||
};
|
||
}""")
|
||
dom_stats = dom_stats or {}
|
||
|
||
dom_parsed = _parse_dom_text(body_text or "")
|
||
if dom_stats.get("followers"):
|
||
dom_parsed["followers"] = dom_stats["followers"]
|
||
if dom_stats.get("following") is not None:
|
||
dom_parsed["following"] = dom_stats["following"]
|
||
if dom_stats.get("likesAndCollects"):
|
||
dom_parsed["likesAndCollects"] = dom_stats["likesAndCollects"]
|
||
title = await page.title()
|
||
await browser.close()
|
||
|
||
def _make_result(nick: str) -> dict:
|
||
return {
|
||
"nickname": nick,
|
||
"followers": dom_parsed.get("followers", 0),
|
||
"following": dom_parsed.get("following", 0),
|
||
"likesAndCollects": dom_parsed.get("likesAndCollects", 0),
|
||
"postCount": dom_parsed.get("postCount", 0),
|
||
"gender": None,
|
||
"redbookId": dom_parsed.get("redbookId"),
|
||
"ipLocation": dom_parsed.get("ipLocation"),
|
||
"personalLink": dom_parsed.get("personalLink"),
|
||
}
|
||
|
||
nick_from_title = title.split(" - ")[0].strip() if title and " - " in title else None
|
||
nick_from_body = _extract_nickname_from_body(body_text or "")
|
||
fallback_nick = nick_from_title or nick_from_body
|
||
|
||
if not state or not isinstance(state, dict):
|
||
if fallback_nick and len(fallback_nick) < 50:
|
||
return _make_result(fallback_nick)
|
||
logging.debug(f"[xiaohongshu] no state, title={title!r}, body_preview={body_text[:200] if body_text else ''!r}")
|
||
return None
|
||
|
||
user = _extract_user_from_state(state, target_user_id)
|
||
if not user:
|
||
if fallback_nick and len(fallback_nick) < 50:
|
||
return _make_result(fallback_nick)
|
||
return None
|
||
|
||
parsed = _parse_user(user)
|
||
if not parsed:
|
||
if fallback_nick and len(fallback_nick) < 50:
|
||
return _make_result(fallback_nick)
|
||
return None
|
||
|
||
if parsed.get("followers", 0) == 0 and dom_parsed.get("followers"):
|
||
parsed["followers"] = dom_parsed["followers"]
|
||
if parsed.get("following", 0) == 0 and dom_parsed.get("following") is not None:
|
||
parsed["following"] = dom_parsed["following"]
|
||
if parsed.get("likesAndCollects", 0) == 0 and dom_parsed.get("likesAndCollects") is not None:
|
||
parsed["likesAndCollects"] = dom_parsed["likesAndCollects"]
|
||
if parsed.get("postCount", 0) == 0 and dom_parsed.get("postCount"):
|
||
parsed["postCount"] = dom_parsed["postCount"]
|
||
if not parsed.get("redbookId") and dom_parsed.get("redbookId"):
|
||
parsed["redbookId"] = dom_parsed["redbookId"]
|
||
if not parsed.get("ipLocation") and dom_parsed.get("ipLocation"):
|
||
parsed["ipLocation"] = dom_parsed["ipLocation"]
|
||
if not parsed.get("personalLink") and dom_parsed.get("personalLink"):
|
||
parsed["personalLink"] = dom_parsed["personalLink"]
|
||
return parsed
|
||
except Exception as e:
|
||
logging.warning(f"[xiaohongshu] fetch error: {e}", exc_info=True)
|
||
try:
|
||
await browser.close()
|
||
except Exception:
|
||
pass
|
||
return None
|
||
|
||
def _parse_dom_text(text: str) -> dict:
|
||
out = {}
|
||
if not text:
|
||
return out
|
||
|
||
lines = [line.strip() for line in text.splitlines() if line.strip()]
|
||
|
||
def pick_count(label: str, key: str) -> None:
|
||
for idx, line in enumerate(lines):
|
||
if line != label or idx == 0:
|
||
continue
|
||
prev = lines[idx - 1]
|
||
if re.fullmatch(r"[-\d.,+wW\u4e07]+", prev):
|
||
_set_count_fields(out, key, prev)
|
||
return
|
||
|
||
pick_count("\u5173\u6ce8", "following")
|
||
pick_count("\u7c89\u4e1d", "followers")
|
||
pick_count("\u83b7\u8d5e\u4e0e\u6536\u85cf", "likesAndCollects")
|
||
|
||
m = re.search(r"\u7b14\u8bb0\s*(\d+)|(\d+)\s*\u7bc7|\u5171\s*(\d+)\s*\u7bc7", text)
|
||
if m:
|
||
out["postCount"] = int(m.group(1) or m.group(2) or m.group(3) or 0)
|
||
out["postCountText"] = str(out["postCount"])
|
||
|
||
m = re.search(r"\u5c0f\u7ea2\u4e66\u53f7[：:]\s*([a-zA-Z0-9_]+)", text)
|
||
if m:
|
||
out["redbookId"] = m.group(1).strip()
|
||
|
||
m = re.search(r"IP\u5c5e\u5730[：:]\s*([^\s\n]+)", text)
|
||
if m:
|
||
out["ipLocation"] = m.group(1).strip()
|
||
|
||
m = re.search(
|
||
r"(?:\u5c0f\u7ea2\u4e66\u53f7|IP\u5c5e\u5730)[^\n]*\n\s*([a-zA-Z0-9][-a-zA-Z0-9.]*\.(?:com|cn|net|io|me))\b",
|
||
text,
|
||
re.MULTILINE,
|
||
)
|
||
if m:
|
||
out["personalLink"] = m.group(1).strip()
|
||
else:
|
||
m = re.search(r"\b([a-zA-Z0-9][-a-zA-Z0-9.]*\.(?:com|cn|net|io|me))\b", text[:600])
|
||
if m and "xiaohongshu" not in m.group(1).lower():
|
||
out["personalLink"] = m.group(1).strip()
|
||
|
||
return out
|
||
|
||
|
||
def _extract_nickname_from_body(body_text: str) -> str | None:
|
||
if not body_text or len(body_text) < 2:
|
||
return None
|
||
|
||
lines = [ln.strip() for ln in body_text.split("\n") if ln.strip()]
|
||
skip = {
|
||
"\u53d1\u73b0",
|
||
"\u53d1\u5e03",
|
||
"\u901a\u77e5",
|
||
"\u521b\u4f5c\u4e2d\u5fc3",
|
||
"\u4e1a\u52a1\u5408\u4f5c",
|
||
"\u5173\u6ce8",
|
||
"\u7b14\u8bb0",
|
||
"\u6536\u85cf",
|
||
"\u767b\u5f55",
|
||
"\u6ce8\u518c",
|
||
}
|
||
|
||
for ln in lines:
|
||
if len(ln) < 2 or len(ln) > 40:
|
||
continue
|
||
if ln in skip or ln.startswith("http") or ln.isdigit():
|
||
continue
|
||
if "\u7c89\u4e1d" in ln or "\u5173\u6ce8" in ln or "\u7b14\u8bb0" in ln or "\u6536\u85cf" in ln:
|
||
continue
|
||
if re.match(r"^[\d\s+]+$", ln):
|
||
continue
|
||
return ln
|
||
return None
|
||
|
||
|
||
async def _fetch_xiaohongshu_profile(url: str) -> dict | None:
|
||
profile_pattern = re.compile(r"/user/profile/([a-zA-Z0-9_-]+)")
|
||
m = profile_pattern.search(url)
|
||
target_user_id = m.group(1) if m else None
|
||
|
||
resolved_url = url
|
||
parsed = None
|
||
|
||
try:
|
||
response = requests.get(url, allow_redirects=True, headers=_XHS_REQUEST_HEADERS, timeout=15)
|
||
resolved_url = response.url or url
|
||
state = _extract_state_from_html(response.text)
|
||
parsed = _parse_profile_from_state(state, target_user_id)
|
||
|
||
title_match = re.search(r"<title>(.*?)</title>", response.text, re.I | re.S)
|
||
if title_match:
|
||
title_text = re.sub(r"\s+", " ", title_match.group(1)).strip()
|
||
fallback_nick = title_text.split(" - ")[0].lstrip("@").strip()
|
||
if fallback_nick:
|
||
parsed = _merge_profile(parsed, {"nickname": fallback_nick})
|
||
|
||
if parsed and parsed.get("nickname") and parsed.get("postCount", 0) > 0:
|
||
parsed["resolvedUrl"] = resolved_url
|
||
return parsed
|
||
except Exception as exc:
|
||
logging.debug(f"[xiaohongshu] html fetch fallback failed: {exc}")
|
||
|
||
try:
|
||
from playwright.async_api import async_playwright
|
||
except ImportError:
|
||
logging.warning("playwright is not installed; run `pip install playwright && playwright install chromium`")
|
||
if parsed:
|
||
parsed["resolvedUrl"] = resolved_url
|
||
return parsed
|
||
|
||
async with async_playwright() as p:
|
||
browser = await p.chromium.launch(
|
||
headless=True,
|
||
args=["--no-sandbox", "--disable-setuid-sandbox", "--disable-blink-features=AutomationControlled"],
|
||
)
|
||
try:
|
||
context = await browser.new_context(
|
||
viewport={"width": 1280, "height": 800},
|
||
user_agent="Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36",
|
||
locale="zh-CN",
|
||
)
|
||
page = await context.new_page()
|
||
|
||
async def handle_route(route):
|
||
if route.request.resource_type in {"image", "media", "font"}:
|
||
await route.abort()
|
||
else:
|
||
await route.continue_()
|
||
|
||
await page.route("**/*", handle_route)
|
||
await page.goto(resolved_url, wait_until="domcontentloaded", timeout=20000)
|
||
await page.wait_for_timeout(2500)
|
||
|
||
state = await page.evaluate("""() => {
|
||
const unwrap = (value) => {
|
||
if (!value) return value ?? null;
|
||
if (typeof value !== "object") return value;
|
||
if (value.__v_isRef) return unwrap(value._rawValue ?? value._value);
|
||
if (Array.isArray(value)) return value.map(unwrap);
|
||
const out = {};
|
||
for (const [key, nested] of Object.entries(value)) {
|
||
if (key === "dep") continue;
|
||
out[key] = unwrap(nested);
|
||
}
|
||
return out;
|
||
};
|
||
const w = window;
|
||
return unwrap(w.__INITIAL_STATE__ || w.__INITIAL_SSR_STATE__ || null);
|
||
}""")
|
||
|
||
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
|
||
nick_from_body = _extract_nickname_from_body(body_text or "")
|
||
fallback_nick = nick_from_title or nick_from_body or None
|
||
|
||
parsed = _merge_profile(parsed, _parse_profile_from_state(state, target_user_id))
|
||
parsed = _merge_profile(parsed, dom_parsed)
|
||
if fallback_nick:
|
||
parsed = _merge_profile(parsed, {"nickname": fallback_nick})
|
||
|
||
if not parsed or not parsed.get("nickname"):
|
||
logging.debug(f"[xiaohongshu] no state, title={title!r}, body_preview={body_text[:200] if body_text else ''!r}")
|
||
return None
|
||
|
||
parsed["resolvedUrl"] = page.url or resolved_url
|
||
return parsed
|
||
except Exception as e:
|
||
logging.warning(f"[xiaohongshu] fetch error: {e}", exc_info=True)
|
||
try:
|
||
await browser.close()
|
||
except Exception:
|
||
pass
|
||
if parsed:
|
||
parsed["resolvedUrl"] = resolved_url
|
||
return parsed
|
||
|
||
|
||
@router.post("/xiaohongshu/profile")
|
||
async def xiaohongshu_profile(item: XiaohongshuProfileRequest):
|
||
"""获取小红书用户资料，支持主页链接、xhslink.com/m/ 短链、笔记链接 xhslink.com/o/"""
|
||
logging.info("[xiaohongshu] profile request received")
|
||
url_raw = (item.get_url_or_id() or "").strip()
|
||
if not url_raw:
|
||
logging.warning("[xiaohongshu] 400: 请输入小红书链接")
|
||
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 _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]}")
|
||
raise HTTPException(status_code=400, detail="链接解析失败，请确认链接有效")
|
||
|
||
parsed = await _fetch_xiaohongshu_profile(url)
|
||
if not parsed:
|
||
logging.warning(f"[xiaohongshu] 400: 无法解析用户资料 url={url[:80]}")
|
||
raise HTTPException(status_code=400, detail="无法解析用户资料，请确认链接有效")
|
||
|
||
logging.info(f"[xiaohongshu] 200 ok: nickname={parsed.get('nickname', '')}")
|
||
return {"ok": True, **parsed}
|
||
|
||
|
||
@router.post("/join")
|
||
async def submit_join(item: dict):
|
||
"""salon 加入社区报名表单，写入 solanRed 集合"""
|
||
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 映射为空
|
||
pb_type = raw_type if raw_type in ("session-1", "session-2") else ""
|
||
|
||
qualify = bool(item.get("qualify", False)) # 女性且小红书帖子>10 时自动通过
|
||
user_info_val = (item.get("userInfo") or "").strip()
|
||
pb_data = {
|
||
"user_id": user_id,
|
||
"nickname": item.get("nickname", ""),
|
||
"occupation": item.get("occupation", ""),
|
||
"reason": reason,
|
||
"wechatId": item.get("wechatId", ""),
|
||
"gender": item.get("gender", ""),
|
||
"education": item.get("education", ""),
|
||
"gradYear": item.get("gradYear", ""),
|
||
"single": item.get("single", ""),
|
||
"media": item.get("media", ""),
|
||
"calculated_age": item.get("calculated_age", 0),
|
||
"type": pb_type,
|
||
"order_id": item.get("order_id", ""),
|
||
"amount": item.get("amount", 0),
|
||
"xiaohongshu_url": item.get("xiaohongshu_url", ""),
|
||
"is_volunteer": bool(item.get("is_volunteer", False)),
|
||
"is_wechat_friend": bool(item.get("is_wechat_friend", False)),
|
||
"is_approved": qualify, # 符合条件时自动通过审核
|
||
"age_range": item.get("age_range", ""),
|
||
"first_time": item.get("first_time", ""),
|
||
"userInfo": user_info_val,
|
||
}
|
||
phone_val = item.get("phone", "")
|
||
if phone_val and str(phone_val).strip():
|
||
pb_data["phone"] = str(phone_val).strip()
|
||
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")
|