'小红书解析'

app/main.py

@@ -97,7 +97,7 @@ def _safe_stderr_write(message: str) -> None:
 
 def _should_trace_request(path: str) -> bool:
     return any(
-        segment in path for segment in ("/nomadvip", "/digital", "/cnomadcna", "/nomadlms")
+        segment in path for segment in ("/nomadvip", "/digital", "/cnomadcna", "/nomadlms", "/salon")
     )
 
 # 启动时校验 BASE_URL（线上部署必须为公网可访问地址，否则 ZPAY 无法回调）

app/routers/salon_meetup.py

@@ -2,8 +2,11 @@
 salon 加入流程：check-user、ensure-user、complete-order
 克隆 meetup 逻辑，使用 site_id=salon
 """
+import json
 import logging
+import re
 import time
+from urllib.parse import parse_qs, unquote, urlparse
 
 import requests
 from fastapi import APIRouter, HTTPException
@@ -17,6 +20,12 @@ 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_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]:
@@ -335,6 +344,755 @@ async def join_status(user_id: str = ""):
         return {"hasRecord": False, "isComplete": False}
 
 
+class XiaohongshuProfileRequest(BaseModel):
+    url: str
+
+
+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"""
+    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 _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
+
+    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")
+
+    # 格式：0 关注 559 粉丝 536 获赞与收藏。找「关注」之后第一个「粉丝」，避免匹配到推荐区的 10+粉丝
+    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:
+            out["following"] = nums_before[0]
+        if nums_after:
+            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=30000)
+            await page.wait_for_timeout(5000)
+
+            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 "")
+            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 短链"""
+    logging.info("[xiaohongshu] profile request received")
+    url = (item.url or "").strip()
+    if not url:
+        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 not xh_profile.match(url) and not xh_short.match(url):
+        logging.warning(f"[xiaohongshu] 400: 链接格式异常 url={url[:80]}")
+        raise HTTPException(status_code=400, detail="小红书link异常")
+
+    url = _resolve_xhslink(url)
+    if 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 集合"""

requirements.txt

@@ -6,4 +6,5 @@ pydantic
 python-multipart
 pocketbase
 requests
-minio
+minio
+playwright