From 83472e3af1e80ac1156b0b6ec5fe56cc0b116170 Mon Sep 17 00:00:00 2001 From: mzl Date: Fri, 9 Jan 2026 16:07:07 +0800 Subject: [PATCH] Initial commit: XHS crawler with dedup and content extraction --- .claude/settings.local.json | 9 + .gitignore | 50 ++ Dockerfile | 17 + Dockerfile.novnc | 24 + README_CN.md | 84 ++ app/server.py | 805 +++++++++++++++++++ app/static/index.html | 1488 +++++++++++++++++++++++++++++++++++ data/profiles.json | 1 + docker-compose.yml | 32 + quick_hotpatch.sh | 17 + quick_start.sh | 45 ++ requirements.txt | 7 + scripts/login_watcher.sh | 49 ++ scripts/start_novnc.sh | 34 + scripts/vnc_auto.html | 24 + xhs_crawler_core.py | 966 +++++++++++++++++++++++ 16 files changed, 3652 insertions(+) create mode 100644 .claude/settings.local.json create mode 100644 .gitignore create mode 100644 Dockerfile create mode 100644 Dockerfile.novnc create mode 100644 README_CN.md create mode 100644 app/server.py create mode 100644 app/static/index.html create mode 100644 data/profiles.json create mode 100644 docker-compose.yml create mode 100644 quick_hotpatch.sh create mode 100644 quick_start.sh create mode 100644 requirements.txt create mode 100644 scripts/login_watcher.sh create mode 100644 scripts/start_novnc.sh create mode 100644 scripts/vnc_auto.html create mode 100644 xhs_crawler_core.py diff --git a/.claude/settings.local.json b/.claude/settings.local.json new file mode 100644 index 0000000..daf7da7 --- /dev/null +++ b/.claude/settings.local.json @@ -0,0 +1,9 @@ +{ + "permissions": { + "allow": [ + "Bash(git init:*)", + "Bash(git add:*)", + "Bash(git commit:*)" + ] + } +} diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..2d755ac --- /dev/null +++ b/.gitignore @@ -0,0 +1,50 @@ +# Python +__pycache__/ +*.py[cod] +*$py.class +*.so +.Python +env/ +venv/ +ENV/ +.venv + +# 小红书采集器运行时生成 +data/crawled_ids.json +data/storage_state*.json +data/*.xlsx +data/uploads/ + +# 采集输出 +output/ +out/ + +# 日志文件 +logs/*.log +logs/app.log +*.log + +# 临时文件 +*.tmp +*.bak +*.swp +*~ + +# 操作系统 +.DS_Store +Thumbs.db +desktop.ini + +# IDE +.vscode/ +.idea/ +*.iml +.pyc +.pyo + +# Docker volumes（如果使用） +data volumes/ + +# 环境变量（可能包含敏感信息） +.env +.env.local diff --git a/Dockerfile b/Dockerfile new file mode 100644 index 0000000..c89ea0b --- /dev/null +++ b/Dockerfile @@ -0,0 +1,17 @@ +FROM mcr.microsoft.com/playwright/python:v1.57.0-jammy + +ENV DEBIAN_FRONTEND=noninteractive +ENV TZ=Asia/Shanghai + +WORKDIR /app + +COPY requirements.txt /app/requirements.txt +RUN pip install --no-cache-dir -r /app/requirements.txt + +# Playwright browsers are already included in base image +COPY xhs_crawler_core.py /app/xhs_crawler_core.py +COPY app /app/app +COPY data /app/data + +EXPOSE 8000 +CMD ["uvicorn", "app.server:app", "--host", "0.0.0.0", "--port", "8000"] diff --git a/Dockerfile.novnc b/Dockerfile.novnc new file mode 100644 index 0000000..8acac16 --- /dev/null +++ b/Dockerfile.novnc @@ -0,0 +1,24 @@ +FROM mcr.microsoft.com/playwright/python:v1.57.0-jammy + +ENV DEBIAN_FRONTEND=noninteractive +ENV TZ=Asia/Shanghai + +WORKDIR /app + +COPY requirements.txt /app/requirements.txt +RUN pip install --no-cache-dir -r /app/requirements.txt + +RUN apt-get update && apt-get install -y --no-install-recommends \ + tzdata xvfb x11vnc fluxbox novnc websockify \ + && rm -rf /var/lib/apt/lists/* + + +COPY scripts/vnc_auto.html /usr/share/novnc/vnc_auto.html +COPY xhs_crawler_core.py /app/xhs_crawler_core.py +COPY data /app/data +COPY scripts/start_novnc.sh /app/start_novnc.sh +COPY scripts/login_watcher.sh /app/login_watcher.sh +RUN chmod +x /app/start_novnc.sh /app/login_watcher.sh + +EXPOSE 6080 +CMD ["/app/start_novnc.sh"] diff --git a/README_CN.md b/README_CN.md new file mode 100644 index 0000000..cb4df88 --- /dev/null +++ b/README_CN.md @@ -0,0 +1,84 @@ +# XHS Cloud 稳定版（Docker，自建服务器） + +> 适用：自建服务器固定 IP、一个采集账号；每周拉取 500~600 个发布账号的笔记/数据；需要 Web 界面 + 实时日志 + 一键触发登录（noVNC）。 + +## 端口约定（可改） +- Web：28000（容器 8000） +- noVNC：26080（容器 6080） +> 你们只开放 10000-50000 端口的话，把这两个端口改到范围内即可。 + +## 一、首次部署（非 1Panel） +```bash +cd /opt +sudo mkdir -p xhs_cloud && sudo chown -R $USER:$USER xhs_cloud +# 上传并解压本 zip 到 /opt/xhs_cloud +cd /opt/xhs_cloud +``` + +### 1）构建镜像（首次） +> 如遇到 Docker Hub 网络超时，先保证 build 已完成；运行 compose 时加 `--pull never` 避免再拉取。 +```bash +sudo docker build --provenance=false --no-cache -t xhs-cloud-web:stable -f Dockerfile . +sudo docker build --provenance=false --no-cache -t xhs-cloud-novnc:stable -f Dockerfile.novnc . +``` + +### 2）启动 +```bash +sudo docker compose up -d --pull never +sudo docker ps +``` + +### 3）访问 +- Web 控制台：`http://<服务器IP>:28000` +- noVNC：`http://<服务器IP>:26080/vnc_auto.html` + +## 二、登录（Web 一键触发） +1. 打开 Web 控制台 → 点击【启动登录】 +2. Web 会提示打开 noVNC（或自动打开） +3. 进入 noVNC 桌面后，会自动弹出 Chromium → 扫码登录小红书 +4. 登录成功后会写入：`data/storage_state.json`，浏览器会自动关闭 + +> 登录流程状态文件：`data/.login_status.json` + +## 三、采集（正常情况无需 noVNC） +- 直接在 Web 控制台点【开始采集】即可 +- 日志通过 WebSocket 推送到页面 + +## 四、验证命令（排错时用） +### 1）确认端口映射 +```bash +sudo docker port xhs-web +sudo docker port xhs-novnc +sudo ss -lntp | egrep "(:28000|:26080)\b" +``` + +### 2）确认 login watcher 在跑 +```bash +sudo docker exec -it xhs-novnc bash -lc "ps aux | grep login_watcher | grep -v grep" +``` + +### 3）手工触发一次登录（等价于 Web 按钮） +```bash +echo '{"ts":'$(date +%s)'}' | sudo tee ./data/.login_request.json > /dev/null +``` + +## 五、我们踩过的坑（必读） +1. **Dockerfile 不能以 `ENV` 开头**：必须先 `FROM ...`，否则报 `no build stage` / `unknown instruction`。 +2. **把命令行记录误写进 Dockerfile**：例如 `cd /opt/xhs_cloud` 会导致 `unknown instruction: cd`。 +3. **Docker Hub 拉取超时**：`docker compose up` 会尝试 pull；本地已 build 时用 `--pull never`。 +4. **Playwright 缺失**：web/novnc 镜像要基于 `mcr.microsoft.com/playwright/python:*` 或在镜像内 `pip install playwright` + `playwright install`。 +5. **headed 浏览器需要 XServer**：在容器里 `headless=False` 必须配 `Xvfb`；noVNC 方案需要 `DISPLAY=:99`。 +6. **`xvfb-run -a` 的 display 不是 :99**：会导致你在 noVNC 里看不到浏览器；正确做法是直接 `export DISPLAY=:99`。 +7. **noVNC 访问根路径只会看到目录列表**：要访问 `/vnc.html` 或 `/vnc_auto.html`。 +8. **watcher 放在 `tail -f /dev/null` 后面永远不会执行**：必须在 tail 之前启动。 +9. **镜像更新但容器没更新**：`docker build` 后要 `docker compose up -d --force-recreate `，否则容器还在用旧 sha。 +10. **端口混淆导致 curl 000**：容器监听 8000，但宿主机映射可能是 28000；curl 要打宿主机端口。 +11. **静态文件路径不对**：页面实际在 `/app/app/static/index.html`，不是 `/app/static`。 + +## 六、升级/更新 +- 更新代码后： +```bash +sudo docker build --provenance=false --no-cache -t xhs-cloud-web:stable -f Dockerfile . +sudo docker build --provenance=false --no-cache -t xhs-cloud-novnc:stable -f Dockerfile.novnc . +sudo docker compose up -d --pull never --force-recreate xhs-web xhs-novnc +``` diff --git a/app/server.py b/app/server.py new file mode 100644 index 0000000..f49ce87 --- /dev/null +++ b/app/server.py @@ -0,0 +1,805 @@ + +import os +import json +import time +import threading +import queue +import pathlib +import zipfile +import re +from datetime import datetime +from typing import Any, Dict, List, Optional, Tuple + +from openpyxl import load_workbook + +from fastapi import FastAPI, WebSocket, WebSocketDisconnect, HTTPException, UploadFile, File, Body +from fastapi.responses import HTMLResponse, FileResponse, JSONResponse +from fastapi.staticfiles import StaticFiles + +# Import your existing crawler core +from xhs_crawler_core import ( + crawl_detail, + check_login_state_with_storage, + login_mode_auto, + resolve_profile_by_search, +) + +APP_ROOT = pathlib.Path(__file__).resolve().parent.parent +DATA_DIR = APP_ROOT / "data" +UPLOAD_DIR = DATA_DIR / "uploads" +OUT_DIR = APP_ROOT / "output" # 修改为 ./output/ 目录 +LOG_DIR = APP_ROOT / "logs" +CRAWLED_IDS_PATH = DATA_DIR / "crawled_ids.json" # 已采集笔记ID记录 + +CONFIG_PATH = DATA_DIR / "config.json" +PROFILES_PATH = DATA_DIR / "profiles.json" +STORAGE_STATE_PATH_DEFAULT = DATA_DIR / "storage_state.json" + +LOGIN_REQ_PATH = DATA_DIR / ".login_request.json" +LOGIN_STATUS_PATH = DATA_DIR / ".login_status.json" +LOGIN_LOCK_PATH = DATA_DIR / ".login_lock" + +OUT_DIR.mkdir(parents=True, exist_ok=True) +LOG_DIR.mkdir(parents=True, exist_ok=True) +DATA_DIR.mkdir(parents=True, exist_ok=True) +UPLOAD_DIR.mkdir(parents=True, exist_ok=True) + +# ---- simple token auth (optional, lightweight) ---- +ADMIN_TOKEN = os.getenv("ADMIN_TOKEN", "").strip() + +def require_token(token: Optional[str]): + if ADMIN_TOKEN and token != ADMIN_TOKEN: + raise HTTPException(status_code=401, detail="Unauthorized") + +# ---- logging hub (queue + websocket broadcast) ---- +class LogHub: + def __init__(self): + self.q: "queue.Queue[str]" = queue.Queue() + self.clients: List[WebSocket] = [] + self.lock = threading.Lock() + self._stop = threading.Event() + self._thread = threading.Thread(target=self._pump, daemon=True) + self._thread.start() + + def emit(self, msg: str): + ts = time.strftime("%Y-%m-%d %H:%M:%S") + line = f"[{ts}] {msg}" + self.q.put(line) + # also append to file + try: + with open(LOG_DIR / "app.log", "a", encoding="utf-8") as f: + f.write(line + "\n") + except Exception: + pass + + def attach(self, ws: WebSocket): + with self.lock: + self.clients.append(ws) + + def detach(self, ws: WebSocket): + with self.lock: + if ws in self.clients: + self.clients.remove(ws) + + def _pump(self): + while not self._stop.is_set(): + try: + line = self.q.get(timeout=0.5) + except queue.Empty: + continue + # broadcast + dead = [] + with self.lock: + clients = list(self.clients) + for ws in clients: + try: + # schedule send in event loop thread via ws.send_text (FastAPI handles) + # Here we just call it; if disconnected it will raise. + import anyio + anyio.from_thread.run(ws.send_text, line) + except Exception: + dead.append(ws) + for ws in dead: + self.detach(ws) + +loghub = LogHub() + +# ---- login state cache ---- +LOGIN_CACHE_TTL = 60.0 +_login_cache = {"ts": 0.0, "ok": False, "msg": "", "path": ""} +_login_lock = threading.Lock() + +def _resolve_storage_path(cfg: Dict[str, Any]) -> str: + storage_path = cfg.get("storage_state_path") or str(STORAGE_STATE_PATH_DEFAULT) + if isinstance(storage_path, str) and storage_path.startswith("data/"): + storage_path = str(APP_ROOT / storage_path) + return storage_path + +def _normalize_wait_range(value: Any, default_range: List[float]) -> tuple: + if isinstance(value, (list, tuple)) and len(value) == 2: + rng = value + else: + rng = default_range + try: + a = float(rng[0]) + b = float(rng[1]) + except Exception: + a = float(default_range[0]) + b = float(default_range[1]) + if a > b: + a, b = b, a + # Heuristic: treat small values as seconds and convert to ms. + if max(a, b) <= 60: + a *= 1000.0 + b *= 1000.0 + return (int(a), int(b)) + +def _check_login_cached(storage_path: str) -> Tuple[bool, str]: + now = time.time() + with _login_lock: + if _login_cache["path"] == storage_path and (now - _login_cache["ts"]) < LOGIN_CACHE_TTL: + return _login_cache["ok"], _login_cache["msg"] + ok, msg = check_login_state_with_storage(storage_path) + with _login_lock: + _login_cache.update({"ts": now, "ok": ok, "msg": msg, "path": storage_path}) + return ok, msg + +# ---- crawled IDs manager (去重管理) ---- +class CrawledIDsManager: + """管理已采集的笔记ID，用于去重""" + def __init__(self): + self.lock = threading.Lock() + self.ids: set = set() + self._load() + + def _load(self): + """从文件加载已采集ID""" + if CRAWLED_IDS_PATH.exists(): + try: + with open(CRAWLED_IDS_PATH, "r", encoding="utf-8") as f: + data = json.load(f) + self.ids = set(data.get("ids", [])) + loghub.emit(f"[去重] 已加载 {len(self.ids)} 条已采集笔记ID") + except Exception as e: + loghub.emit(f"[去重] 加载失败: {e}") + self.ids = set() + else: + self.ids = set() + self._save() + + def _save(self): + """保存已采集ID到文件""" + try: + CRAWLED_IDS_PATH.parent.mkdir(parents=True, exist_ok=True) + with open(CRAWLED_IDS_PATH, "w", encoding="utf-8") as f: + json.dump({"ids": list(self.ids)}, f, ensure_ascii=False, indent=2) + except Exception as e: + loghub.emit(f"[去重] 保存失败: {e}") + + def add(self, note_id: str): + """添加已采集ID""" + with self.lock: + if note_id and note_id not in self.ids: + self.ids.add(note_id) + self._save() + + def add_batch(self, note_ids: List[str]): + """批量添加已采集ID""" + with self.lock: + for note_id in note_ids: + if note_id and note_id not in self.ids: + self.ids.add(note_id) + self._save() + + def contains(self, note_id: str) -> bool: + """检查是否已采集""" + with self.lock: + return note_id in self.ids + + def clear(self): + """清空所有记录""" + with self.lock: + count = len(self.ids) + self.ids.clear() + self._save() + loghub.emit(f"[去重] 已清空 {count} 条记录") + + def count(self) -> int: + """返回已采集ID数量""" + with self.lock: + return len(self.ids) + +# 全局去重管理器 +crawled_manager = CrawledIDsManager() + +# ---- worker manager ---- +class TaskManager: + def __init__(self): + self.running = False + self.stop_event = threading.Event() + self.thread: Optional[threading.Thread] = None + self.progress = { + "total": 0, + "done": 0, + "current": "", + "status": "idle", # idle/running/stopped/failed/done + "last_error": "" + } + + def load_config(self) -> Dict[str, Any]: + if CONFIG_PATH.exists(): + with open(CONFIG_PATH, "r", encoding="utf-8") as f: + return json.load(f) + return {} + + def load_profiles(self) -> List[Dict[str, Any]]: + if PROFILES_PATH.exists(): + with open(PROFILES_PATH, "r", encoding="utf-8") as f: + data = json.load(f) + if isinstance(data, list): + return data + return [] + + def _log(self, msg: str): + loghub.emit(msg) + + def _run_one(self, profile: Dict[str, Any], cfg: Dict[str, Any]): + url = (profile.get("url") or "").strip() + xhs_no = (profile.get("xhs_no") or "").strip() + nickname = (profile.get("nickname") or "").strip() + who = xhs_no or nickname or url + if not url: + self._log(f"[跳过] 缺少 url: {profile}") + return + + # 修改文件命名规则：账号ID_时间戳.xlsx + # 从 URL 中提取账号 ID + account_id = who + if "/user/profile/" in url: + account_id = url.split("/user/profile/")[-1].split("?")[0].strip() + elif who and who != url: + account_id = who + + # 生成时间戳 + timestamp = datetime.now().strftime("%Y%m%d%H%M%S") + out_name = f"{account_id}_{timestamp}.xlsx".replace("/", "_").replace("\\", "_") + out_path = str(OUT_DIR / out_name) + + self._log(f"[开始] {who} -> {url}") + results = crawl_detail( + profile_url=url, + out_file=out_path, + limit=int(cfg.get("limit", 30)), + storage_state_path=_resolve_storage_path(cfg), + headless=bool(cfg.get("headless", True)), + slow_mo=int(cfg.get("slow_mo", 0)), + batch_min=int(cfg.get("batch_min", 12)), + batch_max=int(cfg.get("batch_max", 24)), + open_wait=_normalize_wait_range(cfg.get("open_wait"), [0.8, 2.0]), + between_wait=_normalize_wait_range(cfg.get("between_wait"), [0.4, 1.2]), + scroll_wait=_normalize_wait_range(cfg.get("scroll_wait"), [0.4, 1.1]), + long_break=_normalize_wait_range(cfg.get("long_break"), [8, 18]), + stop_event=self.stop_event, + log=lambda m: self._log(f"[{who}] {m}"), + dedup_check=crawled_manager.contains if cfg.get("enable_dedup", True) else None, + ) + + # 将采集成功的笔记ID添加到去重列表 + if results and isinstance(results, list): + note_ids = [] + for r in results: + href = r.get("href", "") + if href: + # 提取笔记ID（去除query参数） + note_id = href.split("?")[0].strip() + note_ids.append(note_id) + if note_ids: + crawled_manager.add_batch(note_ids) + self._log(f"[去重] 已添加 {len(note_ids)} 条笔记ID到去重库") + + self._log(f"[完成] {who} 输出: {out_name}") + + def _worker_loop(self, idx: int, q: "queue.Queue[Dict[str, Any]]", cfg: Dict[str, Any]): + while not self.stop_event.is_set(): + try: + profile = q.get_nowait() + except queue.Empty: + return + try: + self.progress["current"] = profile.get("xhs_no") or profile.get("nickname") or profile.get("url") or "" + self._log(f"[W{idx}] 处理: {self.progress['current']}") + self._run_one(profile, cfg) + self.progress["done"] += 1 + except Exception as e: + self.progress["last_error"] = str(e) + self._log(f"[W{idx}][错误] {e}") + finally: + q.task_done() + + def start(self, profiles_override: Optional[List[Dict[str, Any]]] = None): + if self.running: + raise HTTPException(status_code=409, detail="Task already running") + cfg = self.load_config() + if profiles_override is not None: + if not isinstance(profiles_override, list): + raise HTTPException(status_code=400, detail="profiles must be a list") + profiles = list(profiles_override) + else: + profiles = self.load_profiles() + if not profiles: + raise HTTPException(status_code=400, detail="profiles.json 为空，请先导入账号列表") + # login check first + storage_path = _resolve_storage_path(cfg) + ok, msg = _check_login_cached(storage_path) + if not ok: + raise HTTPException(status_code=412, detail=msg or "登录态无效：请先在 noVNC 中完成登录并保存 storage_state.json") + + self.stop_event.clear() + self.running = True + self.progress.update({"total": len(profiles), "done": 0, "current": "", "status": "running", "last_error": ""}) + + def run(): + try: + self._log(f"[任务] 启动采集，总计 {len(profiles)} 个账号，并发={int(cfg.get('concurrency',2))}") + q: "queue.Queue[Dict[str, Any]]" = queue.Queue() + for p in profiles: + q.put(p) + + concurrency = int(cfg.get("concurrency", 2)) + workers = [] + for i in range(concurrency): + t = threading.Thread(target=self._worker_loop, args=(i+1, q, cfg), daemon=True) + workers.append(t) + t.start() + + for t in workers: + t.join() + + if self.stop_event.is_set(): + self.progress["status"] = "stopped" + self._log("[任务] 已停止") + else: + self.progress["status"] = "done" + self._log("[任务] 全部完成") + except Exception as e: + self.progress["status"] = "failed" + self.progress["last_error"] = str(e) + self._log(f"[任务][致命错误] {e}") + finally: + self.running = False + + self.thread = threading.Thread(target=run, daemon=True) + self.thread.start() + + def stop(self): + if not self.running: + return + self._log("[任务] 收到停止请求") + self.stop_event.set() + +taskmgr = TaskManager() + +def require_login_state() -> str: + cfg = taskmgr.load_config() + storage_path = _resolve_storage_path(cfg) + ok, msg = _check_login_cached(storage_path) + if not ok: + raise HTTPException(status_code=412, detail=msg or "登录状态无效，请先登录") + return storage_path + +app = FastAPI(title="XHS Crawler Cloud") + +app.mount("/static", StaticFiles(directory=str(APP_ROOT / "app" / "static")), name="static") + +@app.get("/", response_class=HTMLResponse) +def index(): + html_path = APP_ROOT / "app" / "static" / "index.html" + return HTMLResponse(html_path.read_text(encoding="utf-8")) + +@app.get("/api/ping") +def ping(): + return {"ok": True, "time": int(time.time())} + +@app.get("/api/status") +def status(token: Optional[str] = None): + require_token(token) + cfg = taskmgr.load_config() + storage_path = _resolve_storage_path(cfg) + login_ok = False + login_msg = "" + try: + login_ok, login_msg = _check_login_cached(storage_path) + except Exception as e: + loghub.emit(f"[状态] 检查登录态异常: {e}") + login_msg = str(e) + + return { + "running": taskmgr.running, + "progress": taskmgr.progress, + "login_ok": login_ok, + "login_msg": login_msg, + "storage_state_path": storage_path, + } + +@app.get("/api/config") +def get_config(token: Optional[str] = None): + require_token(token) + require_login_state() + return taskmgr.load_config() + +@app.post("/api/config") +def set_config(payload: Dict[str, Any], token: Optional[str] = None): + require_token(token) + require_login_state() + with open(CONFIG_PATH, "w", encoding="utf-8") as f: + json.dump(payload, f, ensure_ascii=False, indent=2) + loghub.emit("[配置] 已保存 config.json") + return {"ok": True} + +@app.get("/api/profiles") +def get_profiles(token: Optional[str] = None): + require_token(token) + require_login_state() + return taskmgr.load_profiles() + +@app.post("/api/profiles") +def set_profiles(payload: List[Dict[str, Any]], token: Optional[str] = None): + require_token(token) + require_login_state() + with open(PROFILES_PATH, "w", encoding="utf-8") as f: + json.dump(payload, f, ensure_ascii=False, indent=2) + loghub.emit(f"[Profiles] 已保存 profiles.json，共 {len(payload)} 条") + return {"ok": True} + +def _merge_profile(existing: List[Dict[str, Any]], profile: Dict[str, Any]) -> Tuple[List[Dict[str, Any]], int, int]: + url = (profile.get("url") or "").strip() + if not url: + return existing, 0, 0 + for p in existing: + if (p.get("url") or "").strip() == url: + updated = False + for k in ("xhs_no", "nickname", "url"): + v = (profile.get(k) or "").strip() + if v and v != (p.get(k) or ""): + p[k] = v + updated = True + return existing, 0, 1 if updated else 0 + existing.append(profile) + return existing, 1, 0 + +@app.post("/api/profiles/resolve") +def resolve_profile(payload: Dict[str, Any], token: Optional[str] = None): + require_token(token) + storage_path = require_login_state() + keyword = str(payload.get("keyword", "")).strip() + if not keyword: + 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, + ) + except Exception as e: + loghub.emit(f"[解析][错误] {e}") + raise HTTPException(status_code=400, detail=str(e)) + existing = taskmgr.load_profiles() + merged, added, updated = _merge_profile(existing, profile) + with open(PROFILES_PATH, "w", encoding="utf-8") as f: + json.dump(merged, f, ensure_ascii=False, indent=2) + if added: + loghub.emit(f"[Profiles] 解析新增账号：{profile.get('xhs_no') or profile.get('url')}") + elif updated: + loghub.emit(f"[Profiles] 更新账号信息：{profile.get('xhs_no') or profile.get('url')}") + return {"ok": True, "profile": profile, "added": added, "updated": updated, "profiles": merged} +# ---- Excel import helpers (profiles) ---- +def _excel_headers(path: pathlib.Path, sheet: str) -> List[str]: + wb = load_workbook(filename=str(path), read_only=True, data_only=True) + if sheet not in wb.sheetnames: + raise HTTPException(status_code=400, detail=f"sheet not found: {sheet}") + ws = wb[sheet] + # find first non-empty row as header + header = [] + for row in ws.iter_rows(min_row=1, max_row=20, values_only=True): + if row and any(v is not None and str(v).strip() for v in row): + header = [str(v).strip() if v is not None else "" for v in row] + break + wb.close() + # sanitize + header = [h for h in header if h] + return header + +def _excel_column_values(path: pathlib.Path, sheet: str, column_key: str) -> List[str]: + wb = load_workbook(filename=str(path), read_only=True, data_only=True) + if sheet not in wb.sheetnames: + raise HTTPException(status_code=400, detail=f"sheet not found: {sheet}") + ws = wb[sheet] + # header row + header_row = None + header_vals = None + for ridx, row in enumerate(ws.iter_rows(min_row=1, max_row=50, values_only=True), start=1): + if row and any(v is not None and str(v).strip() for v in row): + header_row = ridx + header_vals = [str(v).strip() if v is not None else "" for v in row] + break + if not header_row or not header_vals: + wb.close() + return [] + # locate column index by header text OR excel col letter + col_idx = None + if re.fullmatch(r"[A-Za-z]{1,3}", column_key.strip() or ""): + # letter + letters = column_key.strip().upper() + col_idx = 0 + for ch in letters: + col_idx = col_idx * 26 + (ord(ch) - ord('A') + 1) + else: + for i, h in enumerate(header_vals, start=1): + if h == column_key: + col_idx = i + break + if not col_idx: + wb.close() + raise HTTPException(status_code=400, detail=f"column not found: {column_key}") + values: List[str] = [] + for row in ws.iter_rows(min_row=header_row+1, values_only=True): + if not row: + continue + v = row[col_idx-1] if col_idx-1 < len(row) else None + if v is None: + continue + s = str(v).strip() + if not s: + continue + values.append(s) + wb.close() + # unique keep order + seen = set() + out = [] + for s in values: + if s in seen: + continue + seen.add(s) + out.append(s) + return out + +# Keep uploaded excel file records in memory (simple) +_excel_files: Dict[str, str] = {} +_excel_lock = threading.Lock() + +@app.post("/api/excel/upload") +def excel_upload(file: UploadFile = File(...), token: Optional[str] = None): + require_token(token) + require_login_state() + if not file.filename.lower().endswith((".xlsx", ".xlsm", ".xltx", ".xltm")): + raise HTTPException(status_code=400, detail="Only .xlsx/.xlsm supported") + ts = int(time.time()) + fid = f"{ts}_{abs(hash(file.filename))%1000000}" + dst = UPLOAD_DIR / f"{fid}.xlsx" + content = file.file.read() + dst.write_bytes(content) + wb = load_workbook(filename=str(dst), read_only=True, data_only=True) + sheets = list(wb.sheetnames) + wb.close() + with _excel_lock: + _excel_files[fid] = str(dst) + # return columns for first sheet as convenience + cols = _excel_headers(dst, sheets[0]) if sheets else [] + loghub.emit(f"[Excel] 上传 {file.filename} -> {dst.name}") + return {"file_id": fid, "sheets": sheets, "columns": cols, "default_sheet": sheets[0] if sheets else ""} + +@app.get("/api/excel/meta") +def excel_meta(file_id: str, sheet: str, token: Optional[str] = None): + require_token(token) + require_login_state() + with _excel_lock: + path_str = _excel_files.get(file_id) + if not path_str or not os.path.exists(path_str): + raise HTTPException(status_code=404, detail="excel file_id not found") + cols = _excel_headers(pathlib.Path(path_str), sheet) + return {"file_id": file_id, "sheet": sheet, "columns": cols} + +@app.get("/api/excel/column_values") +def excel_column_values(file_id: str, sheet: str, column: str, token: Optional[str] = None): + require_token(token) + require_login_state() + with _excel_lock: + path_str = _excel_files.get(file_id) + if not path_str or not os.path.exists(path_str): + raise HTTPException(status_code=404, detail="excel file_id not found") + vals = _excel_column_values(pathlib.Path(path_str), sheet, column) + loghub.emit(f"[Excel] 读取列 {column} -> {len(vals)} 条") + return {"file_id": file_id, "sheet": sheet, "column": column, "values": vals} + +def _value_to_profile(v: str) -> Dict[str, Any]: + s = (v or "").strip() + if not s: + return {} + if s.startswith("http://") or s.startswith("https://"): + return {"xhs_no": "", "nickname": "", "url": s} + # treat as XHS user id / short id + url = f"https://www.xiaohongshu.com/user/profile/{s}" + return {"xhs_no": s, "nickname": "", "url": url} + +@app.post("/api/profiles/import_excel") +def profiles_import_excel(payload: Dict[str, Any], token: Optional[str] = None): + """ + payload: {file_id, sheet, column} + column can be header text or Excel column letter (A/B/C...) + """ + require_token(token) + require_login_state() + file_id = str(payload.get("file_id","")).strip() + sheet = str(payload.get("sheet","")).strip() + column = str(payload.get("column","")).strip() + if not (file_id and sheet and column): + raise HTTPException(status_code=400, detail="file_id/sheet/column required") + with _excel_lock: + path_str = _excel_files.get(file_id) + if not path_str or not os.path.exists(path_str): + raise HTTPException(status_code=404, detail="excel file_id not found") + vals = _excel_column_values(pathlib.Path(path_str), sheet, column) + new_profiles = [] + for v in vals: + p = _value_to_profile(v) + if p: + new_profiles.append(p) + # merge by url + existing = taskmgr.load_profiles() + seen = set((p.get("url") or "").strip() for p in existing if (p.get("url") or "").strip()) + merged = list(existing) + add_count = 0 + for p in new_profiles: + u = (p.get("url") or "").strip() + if not u or u in seen: + continue + merged.append(p) + seen.add(u) + add_count += 1 + with open(PROFILES_PATH, "w", encoding="utf-8") as f: + json.dump(merged, f, ensure_ascii=False, indent=2) + loghub.emit(f"[Profiles] Excel 导入完成：新增 {add_count} 条（总计 {len(merged)}）") + return {"ok": True, "added": add_count, "total": len(merged), "profiles": merged} + +# ---- multiple storage_state accounts (optional) ---- +@app.get("/api/accounts") +def list_accounts(token: Optional[str] = None): + require_token(token) + require_login_state() + items = [] + for p in sorted(DATA_DIR.glob("storage_state*.json")): + items.append({"name": p.name, "path": str(p), "mtime": int(p.stat().st_mtime)}) + return items + +@app.post("/api/accounts/select") +def select_account(payload: Dict[str, Any], token: Optional[str] = None): + require_token(token) + require_login_state() + name = str(payload.get("name","")).strip() + if not name: + raise HTTPException(status_code=400, detail="name required") + p = DATA_DIR / name + if not p.exists(): + raise HTTPException(status_code=404, detail="storage_state file not found") + cfg = taskmgr.load_config() + cfg["storage_state_path"] = f"data/{name}" + with open(CONFIG_PATH, "w", encoding="utf-8") as f: + json.dump(cfg, f, ensure_ascii=False, indent=2) + loghub.emit(f"[登录] 已切换采集账号：{name}") + return {"ok": True, "storage_state_path": cfg["storage_state_path"]} + + +@app.post("/api/start") +def start(payload: Optional[Dict[str, Any]] = Body(default=None), token: Optional[str] = None): + require_token(token) + require_login_state() + profiles_override = None + if isinstance(payload, dict) and payload.get("profiles") is not None: + if not isinstance(payload.get("profiles"), list): + raise HTTPException(status_code=400, detail="profiles must be a list") + profiles_override = payload.get("profiles") + taskmgr.start(profiles_override=profiles_override) + return {"ok": True} + +@app.post("/api/stop") +def stop(token: Optional[str] = None): + require_token(token) + require_login_state() + taskmgr.stop() + return {"ok": True} + +@app.get("/api/files") +def list_files(token: Optional[str] = None): + require_token(token) + require_login_state() + items = [] + for p in sorted(OUT_DIR.glob("*")): + if p.is_file(): + items.append({"name": p.name, "size": p.stat().st_size, "mtime": int(p.stat().st_mtime)}) + return items + +@app.get("/api/files/{name}") +def download_file(name: str, token: Optional[str] = None): + require_token(token) + require_login_state() + p = OUT_DIR / name + if not p.exists() or not p.is_file(): + raise HTTPException(status_code=404, detail="file not found") + return FileResponse(str(p), filename=p.name) + +@app.get("/api/files_zip") +def download_zip(token: Optional[str] = None): + require_token(token) + require_login_state() + zip_path = OUT_DIR / "out_bundle.zip" + with zipfile.ZipFile(zip_path, "w", zipfile.ZIP_DEFLATED) as z: + for p in OUT_DIR.glob("*.xlsx"): + z.write(p, arcname=p.name) + return FileResponse(str(zip_path), filename="out_bundle.zip") + +@app.post("/api/dedup/clear") +def clear_dedup(token: Optional[str] = None): + """清空去重记录""" + require_token(token) + require_login_state() + count = crawled_manager.count() + crawled_manager.clear() + loghub.emit(f"[去重] 已通过接口清空 {count} 条记录") + return {"ok": True, "cleared": count} + +@app.get("/api/dedup/status") +def dedup_status(token: Optional[str] = None): + """查询去重状态""" + require_token(token) + require_login_state() + return {"ok": True, "count": crawled_manager.count()} + + +@app.post("/api/login/start") +def login_start(token: Optional[str] = None): + require_token(token) + if LOGIN_LOCK_PATH.exists(): + raise HTTPException(status_code=409, detail="login already running, please open noVNC to finish it") + DATA_DIR.mkdir(parents=True, exist_ok=True) + UPLOAD_DIR.mkdir(parents=True, exist_ok=True) + LOGIN_REQ_PATH.write_text(json.dumps({"ts": int(time.time())}, ensure_ascii=False), encoding="utf-8") + return {"ok": True, "msg": "login requested. Open noVNC and scan QR code."} + +@app.get("/api/login/status") +def login_status(token: Optional[str] = None): + require_token(token) + if LOGIN_STATUS_PATH.exists(): + try: + return json.loads(LOGIN_STATUS_PATH.read_text(encoding="utf-8")) + except Exception: + return {"state": "unknown", "msg": "status parse error"} + if LOGIN_LOCK_PATH.exists(): + return {"state": "running", "msg": "login running, open noVNC"} + return {"state": "idle", "msg": "no login yet"} + + +@app.websocket("/ws/logs") +async def ws_logs(ws: WebSocket): + await ws.accept() + loghub.attach(ws) + loghub.emit("[WS] 日志连接已建立") + try: + while True: + # keepalive / receive (client may send ping) + await ws.receive_text() + except WebSocketDisconnect: + loghub.emit("[WS] 日志连接已断开") + finally: + loghub.detach(ws) + +@app.get("/api/login_hint") +def login_hint(token: Optional[str] = None): + require_token(token) + return { + "novnc_url": os.getenv("NOVNC_URL", "http://219.157.255.213:26080/vnc_auto.html"), + "note": "如果登录态无效，请打开 noVNC 登录窗口，完成扫码/验证后再回来点击“重新检查登录状态”。" + } diff --git a/app/static/index.html b/app/static/index.html new file mode 100644 index 0000000..1a64053 --- /dev/null +++ b/app/static/index.html @@ -0,0 +1,1488 @@ + + + + + + XHS 采集器（云端） + + + + +
+ +
+
+ +
+

XHS 采集器（云端）

+

自适应页面 · 绿色主调 · 登录后可用

+
+
+ +
+ + +
+ +
+
+ + 状态：未登录 +
+ + + +
+
+ + +
+
+
+

请先登录

+

当前为 未登录 状态。登录成功后，才能进行账号解析、导入与采集等操作。

+
+ + + +
+
+

提示：页面不会刷新，登录状态由 JS 自动检测更新。

+
+
+ + +
+
+
账号管理 解析 / 导入 / 维护
+ +
+ +
+ + +
+ +
+ + + + + + + +
+
+ + +
+
+
已保存账号 本地存储
+
+ +
+
+ +
+
+
+ + +
+ +
+ +
+ +
+ + + + + + + + + + + + + + + + + +
小红书号昵称URL操作
（占位）xhs_user_001昵称示例https://www.xiaohongshu.com/user/profile/xxxx移除
+
+
+ 分页： + 123 + … +
+
共 1 页（占位）
+
+
+
+
+ + + + + +
+
+
运行日志 全部输出在此
+
+ + +
+
+
+ +
+
+ +
+ +
+ +
+ + + + + + diff --git a/data/profiles.json b/data/profiles.json new file mode 100644 index 0000000..fe51488 --- /dev/null +++ b/data/profiles.json @@ -0,0 +1 @@ +[] diff --git a/docker-compose.yml b/docker-compose.yml new file mode 100644 index 0000000..1225d33 --- /dev/null +++ b/docker-compose.yml @@ -0,0 +1,32 @@ +services: + xhs-web: + image: xhs-cloud-web:stable + build: + context: . + dockerfile: Dockerfile + container_name: xhs-web + restart: unless-stopped + ports: + - "28000:8000" + volumes: + - ./data:/app/data + - ./out:/app/out + - ./logs:/app/logs + environment: + - TZ=Asia/Shanghai + + xhs-novnc: + image: xhs-cloud-novnc:stable + build: + context: . + dockerfile: Dockerfile.novnc + container_name: xhs-novnc + restart: unless-stopped + ports: + - "26080:6080" + volumes: + - ./data:/app/data + - ./out:/app/out + - ./logs:/app/logs + environment: + - TZ=Asia/Shanghai diff --git a/quick_hotpatch.sh b/quick_hotpatch.sh new file mode 100644 index 0000000..2ae804f --- /dev/null +++ b/quick_hotpatch.sh @@ -0,0 +1,17 @@ +#!/usr/bin/env bash +set -euo pipefail +# Fast hot-patch without rebuild: copy host app/ into container then restart. +# Usage: ./quick_hotpatch.sh + +ROOT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)" +cd "$ROOT_DIR" + +echo "[hotpatch] Copying app/ into xhs-web..." +docker cp ./app/server.py xhs-web:/app/app/server.py +docker cp ./app/static/index.html xhs-web:/app/app/static/index.html + +echo "[hotpatch] Restarting xhs-web..." +docker restart xhs-web >/dev/null + +echo "[hotpatch] Tail logs..." +docker logs xhs-web --tail=50 diff --git a/quick_start.sh b/quick_start.sh new file mode 100644 index 0000000..299df25 --- /dev/null +++ b/quick_start.sh @@ -0,0 +1,45 @@ +#!/usr/bin/env bash +set -euo pipefail + +ROOT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)" +cd "$ROOT_DIR" + +# Optional: NOCACHE=1 ./quick_start.sh +BUILD_FLAGS="" +if [[ "${NOCACHE:-0}" == "1" ]]; then + BUILD_FLAGS="--no-cache" +fi + +echo "[xhs-cloud] Working dir: $ROOT_DIR" +echo "[xhs-cloud] Docker: $(docker -v || true)" +echo "[xhs-cloud] Compose: $(docker compose version || true)" +echo "[xhs-cloud] Build flags: ${BUILD_FLAGS:-}" + +# Stop old containers +docker compose down || true + +# Build images locally (avoid pulling from Docker Hub) +docker compose build ${BUILD_FLAGS} --pull=false + +# Start (never pull) +docker compose up -d --pull never + +echo +echo "[xhs-cloud] Containers:" +docker ps --format "table {{.Names}}\t{{.Status}}\t{{.Ports}}" | sed -n '1,10p' + +echo +echo "[xhs-cloud] Listening ports (expect 28000 and 26080):" +ss -lntp 2>/dev/null | egrep "(:28000|:26080)" || true + +IP_HINT="$(hostname -I 2>/dev/null | awk '{print $1}' || true)" +echo +echo "[xhs-cloud] Web console: http://${IP_HINT:-}:28000/" +echo "[xhs-cloud] noVNC Auto: http://${IP_HINT:-}:26080/vnc_auto.html" +echo +echo "[xhs-cloud] Health check (local):" +curl -s -o /dev/null -w "web http_code=%{http_code}\n" http://127.0.0.1:28000/ || true +curl -s -o /dev/null -w "api http_code=%{http_code}\n" http://127.0.0.1:28000/api/login/status || true + +echo +echo "[xhs-cloud] Done." diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..99adaff --- /dev/null +++ b/requirements.txt @@ -0,0 +1,7 @@ +fastapi==0.115.6 +uvicorn[standard]==0.30.6 +pydantic==2.10.3 +python-multipart==0.0.12 +jinja2==3.1.4 +openpyxl==3.1.5 +playwright==1.57.0 diff --git a/scripts/login_watcher.sh b/scripts/login_watcher.sh new file mode 100644 index 0000000..abad1e1 --- /dev/null +++ b/scripts/login_watcher.sh @@ -0,0 +1,49 @@ +#!/usr/bin/env bash +set -euo pipefail + +export DISPLAY=:99 + +DATA_DIR="/app/data" +REQ="${DATA_DIR}/.login_request.json" +STATUS="${DATA_DIR}/.login_status.json" +LOCK="${DATA_DIR}/.login_lock" + +write_status () { + local state="$1" + local msg="$2" + local ts + ts="$(date -Iseconds)" + python - < "$STATUS" +import json +print(json.dumps({"state":"$state","msg":"$msg","ts":"$ts"}, ensure_ascii=False)) +PY +} + +echo "[login] watcher running, data_dir=${DATA_DIR}" + +while true; do + if [ -f "$REQ" ]; then + if [ -f "$LOCK" ]; then + sleep 1 + continue + fi + + touch "$LOCK" + write_status "running" "login started. Open noVNC and finish XHS login." + + set +e + python -c "from xhs_crawler_core import login_mode_auto; login_mode_auto('data/storage_state.json', headless=False)" + rc=$? + set -e + + if [ $rc -eq 0 ]; then + write_status "success" "login finished. storage_state.json updated." + else + write_status "failed" "login failed (rc=${rc}). Check xhs-novnc logs." + fi + + rm -f "$REQ" + rm -f "$LOCK" + fi + sleep 1 +done diff --git a/scripts/start_novnc.sh b/scripts/start_novnc.sh new file mode 100644 index 0000000..d8948b9 --- /dev/null +++ b/scripts/start_novnc.sh @@ -0,0 +1,34 @@ +#!/usr/bin/env bash +set -e + +export DISPLAY=:99 +mkdir -p /tmp/fluxbox + +# Start Xvfb +Xvfb :99 -screen 0 1280x720x24 -nolisten tcp & + +# Wait for X socket +for i in {1..50}; do + [ -S /tmp/.X11-unix/X99 ] && break + sleep 0.1 +done + +# Start window manager +fluxbox -display :99 & + +# Start VNC server +x11vnc -display :99 -nopw -forever -shared -rfbport 5900 & + +# Start noVNC web +websockify --web=/usr/share/novnc/ 0.0.0.0:6080 localhost:5900 & + +# Start login watcher (MUST be before tail) +if [ -x /app/login_watcher.sh ]; then + /app/login_watcher.sh & + echo "[login] watcher started" +else + echo "[login] watcher missing: /app/login_watcher.sh" +fi + +echo "[noVNC] Ready on :6080 (open /vnc_auto.html)" +tail -f /dev/null diff --git a/scripts/vnc_auto.html b/scripts/vnc_auto.html new file mode 100644 index 0000000..f49c2d1 --- /dev/null +++ b/scripts/vnc_auto.html @@ -0,0 +1,24 @@ + + + + + + Auto VNC + + + +

正在进入登录桌面…

+

如果没有自动跳转，请点击下面链接。

+

进入 noVNC

+ + + diff --git a/xhs_crawler_core.py b/xhs_crawler_core.py new file mode 100644 index 0000000..04d1312 --- /dev/null +++ b/xhs_crawler_core.py @@ -0,0 +1,966 @@ +# 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