Initial commit: XHS crawler with dedup and content extraction

This commit is contained in:
mzl
2026-01-09 16:07:07 +08:00
commit 83472e3af1
16 changed files with 3652 additions and 0 deletions
+9
View File
@@ -0,0 +1,9 @@
{
"permissions": {
"allow": [
"Bash(git init:*)",
"Bash(git add:*)",
"Bash(git commit:*)"
]
}
}
+50
View File
@@ -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
+17
View File
@@ -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"]
+24
View File
@@ -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"]
+84
View File
@@ -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 <service>`,否则容器还在用旧 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
```
+805
View File
@@ -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 登录窗口,完成扫码/验证后再回来点击“重新检查登录状态”。"
}
File diff suppressed because it is too large Load Diff
+1
View File
@@ -0,0 +1 @@
[]
+32
View File
@@ -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
+17
View File
@@ -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
+45
View File
@@ -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:-<cache>}"
# 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:-<YOUR_IP>}:28000/"
echo "[xhs-cloud] noVNC Auto: http://${IP_HINT:-<YOUR_IP>}: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."
+7
View File
@@ -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
+49
View File
@@ -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 - <<PY > "$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
+34
View File
@@ -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
+24
View File
@@ -0,0 +1,24 @@
<!DOCTYPE html>
<html lang="zh-CN">
<head>
<meta charset="utf-8" />
<meta name="viewport" content="width=device-width,initial-scale=1" />
<title>Auto VNC</title>
<style>
body{font-family:system-ui,-apple-system,Segoe UI,Roboto,Helvetica,Arial; padding:18px;}
.hint{color:#666;}
code{background:#f6f6f6; padding:2px 6px; border-radius:6px;}
</style>
</head>
<body>
<h3>正在进入登录桌面…</h3>
<p class="hint">如果没有自动跳转,请点击下面链接。</p>
<p><a id="go" href="vnc.html?autoconnect=1&resize=remote">进入 noVNC</a></p>
<script>
// Auto redirect in 300ms
setTimeout(function(){
location.href = document.getElementById('go').href;
}, 300);
</script>
</body>
</html>
+966
View File
@@ -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