diff --git a/app/server.py b/app/server.py index be3d5a7..e98cbae 100644 --- a/app/server.py +++ b/app/server.py @@ -22,6 +22,7 @@ from xhs_crawler_core import ( check_login_state_with_storage, login_mode_auto, resolve_profile_by_search, + resolve_profile_by_direct_url, ) APP_ROOT = pathlib.Path(__file__).resolve().parent.parent @@ -541,14 +542,32 @@ def resolve_profile(payload: Dict[str, Any], token: Optional[str] = None): raise HTTPException(status_code=400, detail="keyword required") headless = bool(payload.get("headless", True)) slow_mo = int(payload.get("slow_mo", 0)) + try: - profile = resolve_profile_by_search( - keyword=keyword, - storage_state_path=storage_path, - headless=headless, - slow_mo=slow_mo, - log=loghub.emit, - ) + # 判断是否为小红书号(纯字母数字,不含空格和特殊字符) + import re + is_xhs_id = bool(re.match(r'^[A-Za-z0-9]+$', keyword)) + + if is_xhs_id: + # 对于小红书号,使用直接访问方式避免搜索跳转问题 + loghub.emit(f"[解析] 检测到小红书号格式,使用直接访问方式:{keyword}") + profile = resolve_profile_by_direct_url( + xhs_id=keyword, + storage_state_path=storage_path, + headless=headless, + slow_mo=slow_mo, + log=loghub.emit, + ) + else: + # 对于昵称或其他关键词,使用搜索方式 + loghub.emit(f"[解析] 使用搜索方式解析:{keyword}") + profile = resolve_profile_by_search( + keyword=keyword, + storage_state_path=storage_path, + headless=headless, + slow_mo=slow_mo, + log=loghub.emit, + ) except Exception as e: loghub.emit(f"[解析][错误] {e}") raise HTTPException(status_code=400, detail=str(e)) diff --git a/xhs_crawler_core.py b/xhs_crawler_core.py index 8c0bbf6..b676680 100644 --- a/xhs_crawler_core.py +++ b/xhs_crawler_core.py @@ -651,6 +651,130 @@ def resolve_profile_by_search( 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,