xhs_crawler/xhs_crawler_core.py

# xhs_crawler_core.py
# 核心功能：
# - login_mode_auto(): 打开浏览器手动登录，检测到已登录后自动保存 storage_state.json
# - check_login_state_with_storage(): 用 storage_state.json 检查登录是否有效
# - resolve_profile_by_search(): 通过“搜索页-用户”解析出个人主页 url + 昵称 + 小红书号（需要登录态）
# - crawl_detail(): 详情弹框采集（标题/时间地点/赞藏评），防漏采滚动策略（需要登录态）

import re
import json
import random
import urllib.parse
import os
import sys
import shutil

# Excel 导出（可选）
try:
    from openpyxl import Workbook
except Exception:
    Workbook = None
from pathlib import Path
from datetime import datetime
from typing import Callable, Optional, Tuple, Dict, List

from playwright.sync_api import sync_playwright, TimeoutError as PWTimeoutError
from urllib.parse import urlsplit, urlunsplit
import time


# ========= 列表页选择器 =========
NOTE_CONTAINER_SELECTOR = ".feeds-container"
NOTE_ITEM_SELECTOR = "section.note-item"
OPEN_DETAIL_SELECTOR = "a.cover.mask.ld"

LIST_TITLE_SELECTOR = "a.title"
LIST_LIKE_SELECTOR = ".author-wrapper .like-wrapper .count"

# ========= 详情弹框选择器 =========
MODAL_ROOT_SELECTOR = ".note-detail-mask"
DETAIL_TITLE_SELECTOR = "#detail-title"
DETAIL_DATE_LOC_SELECTOR = ".bottom-container .date"
# 笔记正文内容选择器（多种选择器适配不同版本）
CONTENT_SELECTORS = [
    ".note-detail-mask .desc",
    ".note-detail-mask .content-text",
    ".note-detail-mask .post-content",
    ".note-detail-mask .note-text",
    "#detail-desc",
    ".desc",
    ".content-text",
    ".post-content",
    ".note-text"
]

LIKE_COUNT_SELECTOR = ".buttons.engage-bar-style .like-wrapper .count, .engage-bar-container .like-wrapper .count"
COLLECT_COUNT_SELECTOR = ".buttons.engage-bar-style .collect-wrapper .count, .engage-bar-container .collect-wrapper .count"
COMMENT_COUNT_SELECTOR = ".buttons.engage-bar-style .chat-wrapper .count, .engage-bar-container .chat-wrapper .count"
COMMENT_TOTAL_SELECTOR = ".comments-container .total, .comment-wrapper .total, .comment-count"

CLOSE_SELECTOR = (
    ".note-detail-mask .close-circle .close, "
    ".note-detail-mask .close-box, "
    ".close-circle .close, "
    ".close-box"
)

# ========= 搜索页-用户结果 选择器 =========
USER_TAB_SELECTOR = "#channel-container #user.channel"
USER_ITEM_SELECTOR = ".user-list-item"
USER_ITEM_LINK_SELECTOR = ".user-list-item a[href^='/user/profile/']"
USER_NAME_SELECTOR = ".user-name"
USER_DESC_SELECTOR = ".user-desc"

# 未登录/拦截时常见文本（兜底判断）
BLOCK_TEXTS_STRONG = ["扫码登录", "安全验证", "请先登录", "登录后查看"]

def _winreg_app_path(exe_name: str) -> str:
    """Windows: 从注册表 App Paths 获取浏览器路径（支持非 C 盘安装）"""
    if sys.platform != "win32":
        return ""
    try:
        import winreg
    except Exception:
        return ""
    subkey = fr"SOFTWARE\Microsoft\Windows\CurrentVersion\App Paths\{exe_name}"
    for root in (getattr(winreg, "HKEY_CURRENT_USER", None), getattr(winreg, "HKEY_LOCAL_MACHINE", None)):
        if root is None:
            continue
        try:
            k = winreg.OpenKey(root, subkey)
            try:
                val, _ = winreg.QueryValueEx(k, "")
                val = (val or "").strip()
                if val and Path(val).exists():
                    return val
            finally:
                try:
                    winreg.CloseKey(k)
                except Exception:
                    pass
        except Exception:
            pass
    return ""


def _which(cmd: str) -> str:
    try:
        return shutil.which(cmd) or ""
    except Exception:
        return ""


def find_system_browser_executable() -> str:
    """自动探测系统 Chrome/Edge 可执行文件路径"""
    chrome = _winreg_app_path("chrome.exe")
    if chrome:
        return chrome
    edge = _winreg_app_path("msedge.exe")
    if edge:
        return edge

    candidates = [
        r"C:\Program Files\Google\Chrome\Application\chrome.exe",
        r"C:\Program Files (x86)\Google\Chrome\Application\chrome.exe",
        os.path.expandvars(r"%LOCALAPPDATA%\Google\Chrome\Application\chrome.exe"),
        r"C:\Program Files\Microsoft\Edge\Application\msedge.exe",
        r"C:\Program Files (x86)\Microsoft\Edge\Application\msedge.exe",
        os.path.expandvars(r"%LOCALAPPDATA%\Microsoft\Edge\Application\msedge.exe"),
    ]
    for p in candidates:
        try:
            if p and Path(p).exists():
                return p
        except Exception:
            pass

    for cmd in ("chrome", "chrome.exe", "msedge", "msedge.exe"):
        p = _which(cmd)
        if p:
            try:
                if Path(p).exists():
                    return p
            except Exception:
                pass

    return ""


def pick_browser_executable(manual_path: str = "") -> str:
    """优先使用手动路径；否则自动探测系统 Chrome/Edge；都没有则返回空字符串"""
    mp = (manual_path or "").strip().strip('"')
    if mp:
        try:
            p = Path(mp).expanduser()
            if p.exists() and p.is_file():
                return str(p)
        except Exception:
            pass
    return find_system_browser_executable()


def clean_profile_url(url: str) -> str:
    """
    清洗小红书个人主页 URL：去掉 ?xsec_token=...&xsec_source=... 等 query，只保留 path。
    例如：
      https://www.xiaohongshu.com/user/profile/xxx?xsec_token=... -> https://www.xiaohongshu.com/user/profile/xxx
    """
    if not url:
        return url

    u = url.strip()

    # 兼容相对路径：/user/profile/xxx?...
    if u.startswith("/"):
        u = "https://www.xiaohongshu.com" + u

    parts = urlsplit(u)
    # 去掉 query + fragment，仅保留 scheme/netloc/path
    return urlunsplit((parts.scheme, parts.netloc, parts.path, "", ""))

def _log(log_cb: Optional[Callable[[str], None]], msg: str):
    if log_cb:
        log_cb(msg)


def _to_int(text: str) -> int:
    if not text:
        return 0
    t = str(text).strip()
    if not re.search(r"\d", t):
        return 0
    t = t.replace(",", "").lower()
    m = re.match(r"^(\d+(?:\.\d+)?)(万|w|k)?$", t, re.I)
    if m:
        num = float(m.group(1))
        unit = (m.group(2) or "").lower()
        if unit in ("万", "w"):
            return int(num * 10000)
        if unit == "k":
            return int(num * 1000)
        return int(num)
    digits = re.sub(r"[^\d]", "", t)
    return int(digits) if digits else 0


def _safe_text(locator) -> str:
    try:
        if locator.count() == 0:
            return ""
        return locator.first.inner_text().strip()
    except Exception:
        return ""


def _extract_profile_id_from_href(href: str) -> str:
    if not href:
        return ""
    m = re.search(r"/user/profile/([^/?#]+)", href)
    return m.group(1) if m else ""


def _wait_ms(page, rng: Tuple[int, int]):
    a, b = rng
    page.wait_for_timeout(random.randint(a, b))


def check_login_state_with_storage(
    storage_state_path: str,
    *,
    headless: bool = True,
    slow_mo: int = 0,
    log: Optional[Callable[[str], None]] = None,
    browser_exe_path: str = "",
) -> Tuple[bool, str]:
    """
    使用 storage_state.json 打开小红书首页，做一个“是否被登录/风控拦截”的判断。
    """
    pth = Path(storage_state_path)
    if not pth.exists():
        return False, f"未找到登录状态文件：{pth.name}"

    try:
        with sync_playwright() as p:
            exe_path = pick_browser_executable(browser_exe_path)
            if exe_path:
                _log(log, f"[浏览器] 使用：{exe_path}")
            launch_kwargs = dict(
                headless=headless,
                slow_mo=slow_mo,
                args=["--no-sandbox", "--disable-setuid-sandbox"],
            )
            if exe_path:
                launch_kwargs["executable_path"] = exe_path
            browser = p.chromium.launch(**launch_kwargs)
            context = browser.new_context(
                storage_state=str(pth),
                locale="zh-CN",
                viewport={"width": 1200, "height": 800},
            )
            page = context.new_page()
            page.goto("https://www.xiaohongshu.com", wait_until="domcontentloaded", timeout=60000)
            page.wait_for_timeout(1200)

            # 强拦截关键词
            for t in BLOCK_TEXTS_STRONG:
                try:
                    if page.get_by_text(t).first.is_visible():
                        context.close(); browser.close()
                        return False, f"登录状态无效或被拦截：页面出现「{t}」"
                except Exception:
                    pass

            # 弱判断：若明显出现“登录”且可见，可能未登录（但不绝对）
            try:
                if page.get_by_text("登录").first.is_visible():
                    # 这里不强判失败，给个提示
                    context.close(); browser.close()
                    return False, "疑似未登录（页面出现明显「登录」入口），建议重新登录保存状态"
            except Exception:
                pass

            context.close()
            browser.close()

        return True, "登录状态正常"
    except Exception as e:
        return False, f"登录状态检查失败：{e}"


def login_mode_auto(
    storage_state_path: str,
    *,
    log: Optional[Callable[[str], None]] = None,
    headless: bool = False,
    slow_mo: int = 80,
    browser_exe_path: str = "",
):
    """
    更稳的自动保存登录态：
    - 打开浏览器让你手动扫码/验证码
    - 程序只在“登录弹窗/扫码提示消失”且“连续多次判定已登录”后才保存并关闭
    - 避免刚打开就因为基础 cookies 误判
    """
    import time

    pth = Path(storage_state_path)
    _log(log, f"[登录] 将打开浏览器，请完成扫码/验证码登录。成功后会自动保存：{pth.name}")

    # 这些选择器/文本用于判断“仍处于登录/扫码/拦截状态”
    LOGIN_MODAL_SELECTORS = [
        ".login-container",
        ".reds-login-mask",
        ".login-modal",
        "[class*='login']",           # 兜底：class 包含 login
        "[class*='Login']",
        "[data-testid*='login']",
    ]
    STRONG_BLOCK_TEXTS = ["扫码登录", "安全验证", "请先登录", "登录后查看"]

    def is_login_block_visible(page) -> bool:
        # 1) 强文本（最可靠）
        for t in STRONG_BLOCK_TEXTS:
            try:
                if page.get_by_text(t).first.is_visible():
                    return True
            except Exception:
                pass

        # 2) 登录弹窗类 DOM（可能会误伤，但宁可保守）
        for sel in LOGIN_MODAL_SELECTORS:
            try:
                loc = page.locator(sel)
                if loc.count() > 0 and loc.first.is_visible():
                    # class*='login' 可能会误命中页面其他元素，所以再加一层：同时出现“登录”文本更可信
                    try:
                        if page.get_by_text("登录").first.is_visible():
                            return True
                    except Exception:
                        # 没有“登录”文字就不强判
                        pass
            except Exception:
                pass
        return False

    def has_auth_like_cookie(context) -> bool:
        """
        只用 cookies 是否为空会误判，所以这里找更像“登录态”的 cookie 名称特征（保守+兜底）。
        不保证 100% 命中，但配合“登录弹窗消失 + 连续多次确认”会明显稳。
        """
        try:
            cookies = context.cookies()
        except Exception:
            cookies = []

        if not cookies:
            return False

        names = [c.get("name", "").lower() for c in cookies]
        # 常见登录态 cookie 命名特征：session/token/auth 等
        for n in names:
            if any(k in n for k in ["session", "token", "auth", "passport", "sid"]):
                return True
        return False

    with sync_playwright() as p:
        exe_path = pick_browser_executable(browser_exe_path)
        if exe_path:
            _log(log, f"[浏览器] 使用：{exe_path}")
        launch_kwargs = dict(
            headless=headless,
            slow_mo=slow_mo,
            args=["--window-size=1366,820", "--no-sandbox", "--disable-setuid-sandbox"],
        )
        if exe_path:
            launch_kwargs["executable_path"] = exe_path
        browser = p.chromium.launch(**launch_kwargs)
        context = browser.new_context(locale="zh-CN", viewport={"width": 1366, "height": 820})
        page = context.new_page()

        page.goto("https://www.xiaohongshu.com", wait_until="domcontentloaded", timeout=60000)

        start = time.monotonic()
        stable_ok = 0

        # 关键：最短等待时间，避免刚打开就误判
        MIN_SECONDS_BEFORE_SAVE = 10
        NEED_STABLE_OK_TIMES = 3  # 连续 3 次判定 OK 才保存
        CHECK_INTERVAL_MS = 1500

        while True:
            if page.is_closed():
                context.close(); browser.close()
                raise RuntimeError("浏览器已关闭，未能保存登录态")

            page.wait_for_timeout(CHECK_INTERVAL_MS)

            blocked = is_login_block_visible(page)

            # 如果仍然在扫码/登录弹窗，就清空稳定计数，继续等
            if blocked:
                stable_ok = 0
                continue

            # 不在登录弹窗了，再做“更像登录态”的 cookie 特征判断
            auth_cookie = has_auth_like_cookie(context)

            # 兜底：页面不再出现明显“登录”入口，也算一个弱信号
            login_text_visible = False
            try:
                login_text_visible = page.get_by_text("登录").first.is_visible()
            except Exception:
                login_text_visible = False

            ok_now = auth_cookie or (not login_text_visible)

            if ok_now:
                stable_ok += 1
            else:
                stable_ok = 0

            elapsed = time.monotonic() - start

            if elapsed >= MIN_SECONDS_BEFORE_SAVE and stable_ok >= NEED_STABLE_OK_TIMES:
                context.storage_state(path=str(pth))
                _log(log, f"[登录] ✅ 已保存登录态：{pth}")
                break

        context.close()
        browser.close()

def resolve_profile_by_search(
    keyword: str,
    storage_state_path: str,
    *,
    headless: bool = True,
    slow_mo: int = 0,
    log: Optional[Callable[[str], None]] = None,
    browser_exe_path: str = "",
) -> Dict[str, str]:
    kw = (keyword or "").strip()
    if not kw:
        raise ValueError("关键词不能为空")

    pth = Path(storage_state_path)
    if not pth.exists():
        raise FileNotFoundError(f"未找到 {pth.name}，请先登录保存状态")

    search_url = "https://www.xiaohongshu.com/search_result?keyword=" + urllib.parse.quote(kw, safe="")
    _log(log, f"[账号] 打开搜索页：{search_url}")

    with sync_playwright() as p:
        exe_path = pick_browser_executable(browser_exe_path)
        if exe_path:
            _log(log, f"[浏览器] 使用：{exe_path}")
        launch_kwargs = dict(
            headless=headless,
            slow_mo=slow_mo,
            args=["--no-sandbox", "--disable-setuid-sandbox"],
        )
        if exe_path:
            launch_kwargs["executable_path"] = exe_path
        browser = p.chromium.launch(**launch_kwargs)
        context = browser.new_context(
            storage_state=str(pth),
            locale="zh-CN",
            viewport={"width": 1200, "height": 800},
        )
        page = context.new_page()

        page.goto(search_url, wait_until="domcontentloaded", timeout=60000)
        page.wait_for_timeout(1200)

        # 检查当前 URL，防止被重定向到个人主页
        current_url = page.url
        _log(log, f"[账号] 当前页面：{current_url}")
        if "/user/profile/" in current_url and current_url != search_url:
            # 被重定向到某个用户主页，这可能是搜索机制导致的
            _log(log, f"[账号] 检测到重定向到用户主页，尝试返回搜索页")
            page.goto(search_url, wait_until="domcontentloaded", timeout=60000)
            page.wait_for_timeout(1500)

        # ✅ 不再嵌套调用 check_login_state_with_storage（避免二次启动 playwright）
        for t in ["扫码登录", "安全验证", "请先登录", "登录后查看"]:
            try:
                if page.get_by_text(t).first.is_visible():
                    context.close(); browser.close()
                    raise RuntimeError(f"无法解析账号：页面出现「{t}」，说明登录态无效/被拦截，请重新登录保存")
            except Exception:
                pass

        def _try_click_user_tab() -> bool:
            selectors = [
                USER_TAB_SELECTOR,
                "#channel-container .channel:has-text('用户')",
                "#channel-container [data-testid*='user']",
                "div.channel:has-text('用户')",
            ]
            for sel in selectors:
                try:
                    loc = page.locator(sel)
                    if loc.count() > 0:
                        _log(log, f"[账号] 找到用户标签：{sel}")
                        loc.first.click(timeout=3000)
                        page.wait_for_timeout(500)
                        return True
                except Exception:
                    pass
            try:
                _log(log, f"[账号] 尝试通过角色定位用户标签")
                page.get_by_role("tab", name="用户").click(timeout=3000)
                page.wait_for_timeout(500)
                return True
            except Exception:
                _log(log, f"[账号] 未找到用户标签，可能已经是用户搜索结果")
                return False

        def _wait_user_items() -> bool:
            selectors = [USER_ITEM_LINK_SELECTOR, USER_ITEM_SELECTOR, "a[href^='/user/profile/']", ".search-user-card", ".user-item", ".search-user-item"]
            for sel in selectors:
                try:
                    page.wait_for_selector(sel, state="attached", timeout=25000)
                    if page.locator(sel).count() > 0:
                        return True
                except Exception:
                    pass
            return False

        # 切到“用户”Tab（不同版本/AB 测试兼容）
        def _find_user_items():
            selectors = [
                USER_ITEM_SELECTOR,
                ".search-user-card",
                ".user-item",
                ".search-user-item",
            ]
            for sel in selectors:
                try:
                    loc = page.locator(sel)
                    if loc.count() > 0:
                        return loc, False
                except Exception:
                    pass
            return page.locator("a[href^='/user/profile/']"), True


        _try_click_user_tab()
        page.wait_for_timeout(800)

        if not _wait_user_items():
            # 常见兜底提示：无结果/被拦截/需要登录
            for t in ["没有找到", "暂无", "没有相关内容", "需要登录", "请先登录", "扫码登录"]:
                try:
                    if page.get_by_text(t).first.is_visible():
                        context.close(); browser.close()
                        raise RuntimeError(f"未找到用户结果：页面提示「{t}」")
                except Exception:
                    pass
            context.close(); browser.close()
            raise RuntimeError("未找到用户结果（页面结构变化或被拦截），请确认登录态并重试")

        items, use_anchor_only = _find_user_items()
        n = items.count()
        _log(log, f"[账号] 找到 {n} 个用户结果，使用模式：{'锚点链接' if use_anchor_only else '结构化元素'}")
        if n == 0:
            context.close(); browser.close()
            raise RuntimeError("未找到用户结果（可能被拦截/需要更完整登录态）")

        best = None
        best_score = -1

        for i in range(n):
            it = items.nth(i)
            _log(log, f"[账号] 分析第 {i+1}/{n} 个结果...")
            if use_anchor_only:
                href = (it.get_attribute("href") or "").strip()
                nickname = ""
                xhs_no = ""
                try:
                    full_text = (it.inner_text() or "").strip()
                    # 解析昵称和账号：从完整文本中提取
                    # 示例：青岛西门子家电规划师-海燕呀 1天前更新 小红书号：V13863940939 粉丝・52 笔记・217 关注
                    # 提取昵称：从开头到 "小红书号" 之前，去除时间信息
                    lines = full_text.split('\n')
                    main_text = lines[0] if lines else full_text

                    # 提取小红书号
                    m = re.search(r'小红书号[：:]\s*([A-Za-z0-9]+)', main_text)
                    if m:
                        xhs_no = m.group(1).strip()

                    # 提取昵称：去除时间信息和账号信息
                    # 先移除小红书号部分
                    nickname_part = re.sub(r'小红书号[：:]\s*[A-Za-z0-9]+.*', '', main_text)
                    # 移除时间信息（如 "1天前更新"）
                    nickname_part = re.sub(r'\d+[天小时分秒]*前更新', '', nickname_part)
                    # 移除粉丝、笔记、关注信息
                    nickname_part = re.sub(r'粉丝[・・]\d+\s*笔记[・・]\d+\s*关注.*', '', nickname_part)
                    nickname = nickname_part.strip()
                except Exception:
                    nickname = ""
                    xhs_no = ""
            else:
                href = (it.locator("a[href^='/user/profile/']").first.get_attribute("href") or "").strip()
                nickname = ""
                try:
                    nickname = (it.locator(USER_NAME_SELECTOR).first.inner_text() or "").strip()
                except Exception:
                    nickname = ""
                xhs_no = ""
                try:
                    desc = (it.locator(USER_DESC_SELECTOR).first.inner_text() or "").strip()
                    m = re.search(r"\u5c0f\u7ea2\u4e66\u53f7\uff1a\s*(\S+)", desc)
                    if m:
                        xhs_no = m.group(1).strip()
                except Exception:
                    xhs_no = ""
            if not href:
                continue

            score = 0
            score_reasons = []

            # 精确匹配小红书号：最高优先级
            if xhs_no and xhs_no.lower() == kw.lower():
                score += 1000
                score_reasons.append("小红书号精确匹配")

            # 精确匹配昵称：高优先级
            if nickname and nickname.lower() == kw.lower():
                score += 500
                score_reasons.append("昵称精确匹配")

            # 昵称包含关键词：中优先级（不使用 elif，允许累加）
            if nickname and kw.lower() in nickname.lower() and nickname.lower() != kw.lower():
                score += 50
                score_reasons.append("昵称包含关键词")

            # URL 中包含关键词：低优先级
            if kw.lower() in href.lower():
                score += 20
                score_reasons.append("URL包含关键词")

            # 有有效的 profile ID：基础分
            if _extract_profile_id_from_href(href):
                score += 5
                score_reasons.append("有效Profile ID")

            reasons_str = ", ".join(score_reasons) if score_reasons else "无匹配"
            _log(log, f"[账号]   - 昵称: {nickname[:30] if len(nickname) > 30 else nickname}, 账号: {xhs_no}, 得分: {score} ({reasons_str})")

            if score > best_score:
                best_score = score
                url = "https://www.xiaohongshu.com" + href if href.startswith("/") else href
                best = {"xhs_no": xhs_no, "nickname": nickname or kw, "url": clean_profile_url(url)}

        context.close()
        browser.close()

    if not best:
        raise RuntimeError("解析失败：没有可用的用户条目")

    _log(log, f"[账号] ✅ 已解析：{best['nickname']}（{best['xhs_no'] or '无小红书号'}） -> {best['url']}")
    return best


def resolve_profile_by_direct_url(
    xhs_id: str,
    storage_state_path: str,
    *,
    headless: bool = True,
    slow_mo: int = 0,
    log: Optional[Callable[[str], None]] = None,
    browser_exe_path: str = "",
) -> Dict[str, str]:
    """
    直接通过小红书号访问用户主页，避免搜索的干扰
    适用于已知小红书号的场景
    """
    xhs_id = (xhs_id or "").strip()
    if not xhs_id:
        raise ValueError("小红书号不能为空")

    pth = Path(storage_state_path)
    if not pth.exists():
        raise FileNotFoundError(f"未找到 {pth.name}，请先登录保存状态")

    # 直接构造用户主页 URL
    profile_url = f"https://www.xiaohongshu.com/user/profile/{xhs_id}"
    _log(log, f"[账号] 直接访问主页：{profile_url}")

    with sync_playwright() as p:
        exe_path = pick_browser_executable(browser_exe_path)
        if exe_path:
            _log(log, f"[浏览器] 使用：{exe_path}")
        launch_kwargs = dict(
            headless=headless,
            slow_mo=slow_mo,
            args=["--no-sandbox", "--disable-setuid-sandbox"],
        )
        if exe_path:
            launch_kwargs["executable_path"] = exe_path
        browser = p.chromium.launch(**launch_kwargs)
        context = browser.new_context(
            storage_state=str(pth),
            locale="zh-CN",
            viewport={"width": 1200, "height": 800},
        )
        page = context.new_page()

        # 访问用户主页
        page.goto(profile_url, wait_until="domcontentloaded", timeout=60000)
        page.wait_for_timeout(2000)

        # 检查当前 URL 是否被重定向
        current_url = page.url
        _log(log, f"[账号] 当前 URL：{current_url}")

        # 检查是否出现错误页面
        page_text = page.inner_text("body") or ""
        for t in ["不存在", "内容暂不可见", "页面不存在", "404", "访问受限"]:
            if t in page_text:
                context.close(); browser.close()
                raise RuntimeError(f"用户不存在或访问受限：{t}")

        # 提取用户信息
        nickname = ""
        xhs_no = ""

        try:
            # 尝试多种选择器获取昵称
            nickname_selectors = [
                ".user-info .user-name",
                ".user-name",
                ".username",
                ".user-title",
                "h1.user-name",
                ".profile-user-name",
            ]
            for sel in nickname_selectors:
                try:
                    elem = page.locator(sel)
                    if elem.count() > 0:
                        nickname = elem.first.inner_text().strip()
                        if nickname:
                            break
                except Exception:
                    continue

            # 尝试获取小红书号
            desc_selectors = [
                ".user-info .user-desc",
                ".user-desc",
                ".user-id",
                ".user-no",
                ".xiaohongshu-id",
            ]
            for sel in desc_selectors:
                try:
                    elem = page.locator(sel)
                    if elem.count() > 0:
                        desc = elem.first.inner_text() or ""
                        m = re.search(r'小红书号[：:]\s*([A-Za-z0-9]+)', desc)
                        if m:
                            xhs_no = m.group(1).strip()
                            break
                except Exception:
                    continue

        except Exception as e:
            _log(log, f"[账号] 提取用户信息失败：{e}")

        # 清理 URL（移除查询参数）
        clean_url = current_url.split("?")[0].strip()
        if not clean_url.startswith("http"):
            clean_url = profile_url

        context.close()
        browser.close()

        result = {
            "xhs_no": xhs_no or xhs_id,
            "nickname": nickname or xhs_id,
            "url": clean_url
        }

        _log(log, f"[账号] ✅ 已解析：{result['nickname']}（{result['xhs_no']}） -> {result['url']}")
        return result


def crawl_detail(
    profile_url: str,
    limit: int = 20,
    storage_state_path: str = "storage_state.json",
    headless: bool = False,
    slow_mo: int = 80,
    batch_min: int = 3,
    batch_max: int = 7,
    open_wait: Tuple[int, int] = (1800, 3200),
    between_wait: Tuple[int, int] = (1500, 3200),
    scroll_wait: Tuple[int, int] = (1600, 2800),
    long_break: Tuple[int, int] = (6000, 12000),
    scroll_target_new: int = 2,
    scroll_max_attempts: int = 10,
    out_file: str = "out/result_modal_details.json",
    stop_event=None,
    log: Optional[Callable[[str], None]] = None,
    browser_exe_path: str = "",
    dedup_check: Optional[Callable[[str], bool]] = None,  # 去重检查回调，返回True表示已采集
):
    pth = Path(storage_state_path)
    if not pth.exists():
        raise FileNotFoundError(f"未找到 {pth.name}，请先登录保存状态")

    out_path = Path(out_file)
    out_path.parent.mkdir(parents=True, exist_ok=True)

    def _stopped() -> bool:
        return bool(stop_event and stop_event.is_set())

    def get_visible_cards(page):
        container = page.locator(NOTE_CONTAINER_SELECTOR)
        return container.locator(NOTE_ITEM_SELECTOR)

    def get_card_key(item) -> str:
        opener = item.locator(OPEN_DETAIL_SELECTOR)
        if opener.count() == 0:
            return ""
        href = opener.first.get_attribute("href") or ""
        return href.split("?")[0].strip()

    def count_visible_unprocessed(page, processed: set) -> int:
        items = get_visible_cards(page)
        n = items.count()
        keys = set()
        for i in range(n):
            key = get_card_key(items.nth(i))
            if key and key not in processed:
                keys.add(key)
        return len(keys)

    def find_next_visible_unprocessed(page, processed: set):
        items = get_visible_cards(page)
        n = items.count()
        for i in range(n):
            item = items.nth(i)
            opener = item.locator(OPEN_DETAIL_SELECTOR)
            if opener.count() == 0:
                continue
            href = opener.first.get_attribute("href") or ""
            key = href.split("?")[0].strip()
            if not key or key in processed:
                continue
            ts = time.time()
            if key in blocked_until and (ts - blocked_until[key]) < 60:
                continue
            list_title = _safe_text(item.locator(LIST_TITLE_SELECTOR))
            list_like = _to_int(_safe_text(item.locator(LIST_LIKE_SELECTOR)))
            return item, opener.first, href, key, list_title, list_like
        return None

    def close_modal(page):
        try:
            if page.locator(CLOSE_SELECTOR).count() > 0:
                page.locator(CLOSE_SELECTOR).first.click(timeout=5000)
                page.wait_for_selector(MODAL_ROOT_SELECTOR, state="hidden", timeout=8000)
                return
        except Exception:
            pass
        try:
            page.keyboard.press("Escape")
            page.wait_for_selector(MODAL_ROOT_SELECTOR, state="hidden", timeout=8000)
            return
        except Exception:
            pass
        try:
            page.mouse.click(20, 20)
            page.wait_for_selector(MODAL_ROOT_SELECTOR, state="hidden", timeout=8000)
        except Exception:
            pass

    
    def _get_modal_root(page):
        """优先使用真正的弹框遮罩 .note-detail-mask；如缺失则回退到 #noteContainer。"""
        try:
            m = page.locator(".note-detail-mask")
            if m.count() > 0 and m.first.is_visible():
                return m.first
        except Exception:
            pass
        return page.locator("#noteContainer").first

    def _safe_text_scoped(page, root, selector: str) -> str:
        """先在 root 内找，找不到再全局找（适配部分结构把按钮区放在 noteContainer 外）。"""
        try:
            t = _safe_text(root.locator(selector))
            if t:
                return t
        except Exception:
            pass
        return _safe_text(page.locator(selector))

    def extract_detail_from_modal(page) -> dict:
        modal = _get_modal_root(page)
        title = _safe_text_scoped(page, modal, DETAIL_TITLE_SELECTOR)
        date_loc = _safe_text_scoped(page, modal, DETAIL_DATE_LOC_SELECTOR)
        like_cnt = _to_int(_safe_text_scoped(page, modal, LIKE_COUNT_SELECTOR))
        collect_cnt = _to_int(_safe_text_scoped(page, modal, COLLECT_COUNT_SELECTOR))
        comment_cnt = _to_int(_safe_text_scoped(page, modal, COMMENT_COUNT_SELECTOR))
        if comment_cnt == 0:
            comment_cnt = _to_int(_safe_text_scoped(page, modal, COMMENT_TOTAL_SELECTOR))

        # 提取笔记正文内容（前200字）
        content = ""
        for sel in CONTENT_SELECTORS:
            try:
                elem = modal.locator(sel)
                if elem.count() > 0:
                    full_text = elem.first.inner_text().strip()
                    if full_text:
                        # 取前200个字符
                        content = full_text[:200] + ("..." if len(full_text) > 200 else "")
                        break
            except Exception:
                continue

        return {
            "detail_title": title,
            "detail_date_loc": date_loc,
            "detail_content": content,
            "like": like_cnt,
            "collect": collect_cnt,
            "comment": comment_cnt,
        }

    def overlap_scroll(page, stagnant_times: int = 0):
        try:
            h = page.evaluate("() => window.innerHeight || 800")
        except Exception:
            h = 800
        base = random.uniform(0.55, 0.85)
        boost = 1.0 + min(stagnant_times * 0.10, 0.35)
        dist = int(h * base * boost)
        page.evaluate("(d) => window.scrollBy(0, d)", dist)

    def scroll_to_load_more(page, processed: set) -> bool:
        before = count_visible_unprocessed(page, processed)
        stagnant = 0
        for _ in range(scroll_max_attempts):
            if _stopped():
                return False
            overlap_scroll(page, stagnant_times=stagnant)
            extra = stagnant * 500
            a, b = scroll_wait
            page.wait_for_timeout(random.randint(a + extra, b + extra))

            after = count_visible_unprocessed(page, processed)
            gained = after - before
            if gained >= scroll_target_new:
                return True
            if gained > 0:
                stagnant = 0
                before = after
            else:
                stagnant += 1

            if stagnant >= 3:
                try:
                    page.evaluate("""
                    () => {
                        const y = document.body.scrollHeight - window.innerHeight;
                        window.scrollTo(0, Math.max(0, y - Math.floor(Math.random() * 600)));
                    }
                    """)
                except Exception:
                    pass
                page.wait_for_timeout(random.randint(2000, 3500))
                after2 = count_visible_unprocessed(page, processed)
                if after2 > before:
                    return True
        return False

    _log(log, f"[采集] 打开：{profile_url}")
    _log(log, f"[采集] limit={limit} headless={headless} slow_mo={slow_mo}")

    results = []
    processed = set()
    fail_counts: Dict[str, int] = {}
    blocked_until: Dict[str, float] = {}  # key -> last fail timestamp (avoid immediate re-click)
    batch_target = random.randint(batch_min, batch_max)
    batch_count = 0
    _log(log, f"[采集] 长休息批次：每采集 {batch_min}~{batch_max} 条随机一次（当前目标={batch_target}）")

    with sync_playwright() as p:
        exe_path = pick_browser_executable(browser_exe_path)
        if exe_path:
            _log(log, f"[浏览器] 使用：{exe_path}")
        launch_kwargs = dict(
            headless=headless,
            slow_mo=slow_mo,
            args=["--window-size=1366,820", "--no-sandbox", "--disable-setuid-sandbox"],
        )
        if exe_path:
            launch_kwargs["executable_path"] = exe_path
        browser = p.chromium.launch(**launch_kwargs)
        context = browser.new_context(
            storage_state=str(pth),
            viewport={"width": 1366, "height": 820},
            locale="zh-CN",
            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",
        )
        context.set_default_timeout(60000)
        page = context.new_page()

        page.goto(profile_url, wait_until="domcontentloaded", timeout=60000)
        page.wait_for_selector(NOTE_CONTAINER_SELECTOR, timeout=30000)
        _wait_ms(page, (1200, 2200))

        safety_round = 0
        while len(results) < limit and safety_round < 2000:
            safety_round += 1
            if _stopped():
                _log(log, "[采集] 收到停止请求，正在退出...")
                break

            found = find_next_visible_unprocessed(page, processed)
            if found:
                item, opener, href, key, list_title, list_like = found

                # 去重检查：在点击之前先检查是否已采集
                if dedup_check and key:
                    note_id = key.split("?")[0].strip()  # 提取纯净的笔记ID
                    if dedup_check(note_id):
                        _log(log, f"[去重] 笔记已采集，跳过：{note_id}")
                        processed.add(key)
                        continue

                try:
                    item.scroll_into_view_if_needed()
                except Exception:
                    pass
                _wait_ms(page, (700, 1300))

                try:
                    opener.click(timeout=6000)
                except Exception:
                    try:
                        item.click(timeout=6000)
                    except Exception:
                        continue

                try:
                    # 关键：不要用 #noteContainer 这种“永远可见”的元素判断弹框是否打开
                    page.wait_for_selector(DETAIL_TITLE_SELECTOR, state="visible", timeout=25000)
                except PWTimeoutError:
                    # 先做一次“强制点击+更长等待”的快速重试，尽量别直接跳过
                    retry_ok = False
                    try:
                        opener.click(timeout=6000, force=True)
                        page.wait_for_selector(DETAIL_TITLE_SELECTOR, state="visible", timeout=30000)
                        retry_ok = True
                    except Exception:
                        retry_ok = False
                    if not retry_ok:

                        # 关键增强：避免一直点第一个笔记
                        fail_counts[key] = fail_counts.get(key, 0) + 1
                        blocked_until[key] = time.time()
                        _log(log, f"[警告] 弹框打开超时：{fail_counts[key]} 次（将先跳过 60 秒）")
                        close_modal(page)
                        # 额外兜底：ESC 关闭潜在遮罩
                        try:
                            page.keyboard.press("Escape")
                        except Exception:
                            pass
                        # 轻微滚动，让“下一个卡片”进入可见区域，避免一直选中第一个
                        try:
                            overlap_scroll(page, stagnant_times=min(fail_counts[key], 3))
                            _wait_ms(page, scroll_wait)
                        except Exception:
                            pass
                        # 多次失败则临时跳过该笔记，避免卡死（后续你可以再跑一次补采）
                        if fail_counts[key] >= 3:
                            processed.add(key)
                            _log(log, "[跳过] 该笔记连续 3 次打开失败，已临时跳过以保证任务继续")
                        continue

                _wait_ms(page, open_wait)
                detail = extract_detail_from_modal(page)

                # 兜底：部分笔记标题在弹框里可能为空（或加载慢），用列表标题补一下
                if not (detail.get("detail_title") or "").strip():
                    detail["detail_title"] = (list_title or "").strip()

                results.append({
                    "idx": len(results) + 1,
                    "profile_url": profile_url,
                    "href": href,
                    "list_title": list_title,
                    "list_like": list_like,
                    **detail,
                    "crawled_at": datetime.now().isoformat(timespec="seconds"),
                })
                processed.add(key)

                _log(log, f"[成功] {len(results)}/{limit} 赞={detail['like']} 藏={detail['collect']} 评={detail['comment']} | {detail['detail_title'][:25]}")

                close_modal(page)
                _wait_ms(page, between_wait)

                batch_count += 1
                if batch_count >= batch_target and len(results) < limit:
                    _log(log, f"[休息] 已采集 {batch_count} 条 -> 长休息")
                    _wait_ms(page, long_break)
                    batch_target = random.randint(batch_min, batch_max)
                    batch_count = 0
                    _log(log, f"[采集] 新的长休息目标={batch_target}")

                continue

            ok = scroll_to_load_more(page, processed)
            if not ok:
                _log(log, "[警告] 多次滚动仍无新卡片：可能到底或受限。等待后再判断退出")
                page.wait_for_timeout(random.randint(4000, 8000))
                if count_visible_unprocessed(page, processed) == 0:
                    break
        if str(out_path).lower().endswith((".xlsx", ".xls")):
            if Workbook is None:
                _log(log, "[错误] 需要 openpyxl 才能导出 Excel，请先安装：pip install openpyxl")
            else:
                wb = Workbook()
                ws = wb.active
                ws.title = "data"
                # Excel 导出字段（不包含 href 和 detail_content）
                headers = ["detail_title", "like", "collect", "comment", "detail_date_loc", "crawled_at"]
                ws.append(headers)
                for r in results:
                    ws.append([r.get(h, "") for h in headers])
                out_path.parent.mkdir(parents=True, exist_ok=True)
                wb.save(str(out_path))
                _log(log, f"[保存] {out_path}（Excel，共 {len(results)} 条）")
        else:
            out_path.parent.mkdir(parents=True, exist_ok=True)
            out_path.write_text(json.dumps(results, ensure_ascii=False, indent=2), encoding="utf-8")
            _log(log, f"[保存] {out_path}（JSON，共 {len(results)} 条）")

        context.close()
        browser.close()

    return results