解析账号
This commit is contained in:
+26
-7
@@ -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))
|
||||
|
||||
@@ -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,
|
||||
|
||||
Reference in New Issue
Block a user