967 lines
36 KiB
Python
967 lines
36 KiB
Python
# 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)
|
||
|
||
# ✅ 不再嵌套调用 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:
|
||
loc.first.click(timeout=3000)
|
||
return True
|
||
except Exception:
|
||
pass
|
||
try:
|
||
page.get_by_role("tab", name="用户").click(timeout=3000)
|
||
return True
|
||
except Exception:
|
||
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()
|
||
if n == 0:
|
||
context.close(); browser.close()
|
||
raise RuntimeError("未找到用户结果（可能被拦截/需要更完整登录态）")
|
||
|
||
best = None
|
||
best_score = -1
|
||
|
||
for i in range(n):
|
||
it = items.nth(i)
|
||
if use_anchor_only:
|
||
href = (it.get_attribute("href") or "").strip()
|
||
nickname = ""
|
||
try:
|
||
nickname = (it.inner_text() or "").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
|
||
if xhs_no and xhs_no.lower() == kw.lower():
|
||
score += 100
|
||
if nickname and kw in nickname:
|
||
score += 10
|
||
if _extract_profile_id_from_href(href):
|
||
score += 1
|
||
|
||
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 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"
|
||
# 按需求文档的字段顺序
|
||
headers = ["href", "detail_title", "detail_content", "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
|