diff --git a/app/__pycache__/__init__.cpython-311.pyc b/app/__pycache__/__init__.cpython-311.pyc new file mode 100644 index 0000000..4468a8d Binary files /dev/null and b/app/__pycache__/__init__.cpython-311.pyc differ diff --git a/app/__pycache__/auth_manager.cpython-311.pyc b/app/__pycache__/auth_manager.cpython-311.pyc new file mode 100644 index 0000000..132e7df Binary files /dev/null and b/app/__pycache__/auth_manager.cpython-311.pyc differ diff --git a/app/__pycache__/config.cpython-311.pyc b/app/__pycache__/config.cpython-311.pyc new file mode 100644 index 0000000..f833756 Binary files /dev/null and b/app/__pycache__/config.cpython-311.pyc differ diff --git a/app/__pycache__/log_manager.cpython-311.pyc b/app/__pycache__/log_manager.cpython-311.pyc new file mode 100644 index 0000000..86e880c Binary files /dev/null and b/app/__pycache__/log_manager.cpython-311.pyc differ diff --git a/app/__pycache__/scraper.cpython-311.pyc b/app/__pycache__/scraper.cpython-311.pyc new file mode 100644 index 0000000..7fc1d18 Binary files /dev/null and b/app/__pycache__/scraper.cpython-311.pyc differ diff --git a/app/__pycache__/user_resolver.cpython-311.pyc b/app/__pycache__/user_resolver.cpython-311.pyc new file mode 100644 index 0000000..d3f764a Binary files /dev/null and b/app/__pycache__/user_resolver.cpython-311.pyc differ diff --git a/app/__pycache__/web.cpython-311.pyc b/app/__pycache__/web.cpython-311.pyc new file mode 100644 index 0000000..cb14ebf Binary files /dev/null and b/app/__pycache__/web.cpython-311.pyc differ diff --git a/app/auth_manager.py b/app/auth_manager.py new file mode 100644 index 0000000..d182202 --- /dev/null +++ b/app/auth_manager.py @@ -0,0 +1,171 @@ + +"""auth_manager.py + +登录态管理（单账号） +- 使用 Playwright storage_state 保存到 data/auth_state.json +- 额外保存元数据到 data/auth_meta.json（登录时间/最后验证时间等） +- 提供 verify_login / get_login_status 等方法 + +设计目标： +1) 不依赖“文件存在”判断登录，而是用 Playwright 打开页面做轻量验证 +2) 单账号长期复用：只要不掉线，就一直用 +""" + +from __future__ import annotations + +import json +import os +from dataclasses import dataclass +from datetime import datetime, timedelta +from typing import Optional, Tuple + +from playwright.async_api import Browser, BrowserContext, TimeoutError as PlaywrightTimeoutError + +from app import config + + +AUTH_META_FILE = os.path.join(config.DATA_DIR, "auth_meta.json") + + +@dataclass +class LoginStatus: + logged_in: bool + login_time: Optional[str] = None + last_verified_time: Optional[str] = None + message: Optional[str] = None + + +def _utc_iso() -> str: + # 使用本地时间 ISO 字符串，方便 UI 展示 + return datetime.now().isoformat(timespec="seconds") + + +def load_auth_meta() -> dict: + if not os.path.exists(AUTH_META_FILE): + return {} + try: + with open(AUTH_META_FILE, "r", encoding="utf-8") as f: + return json.load(f) or {} + except Exception: + return {} + + +def save_auth_meta(meta: dict) -> None: + tmp = AUTH_META_FILE + ".tmp" + with open(tmp, "w", encoding="utf-8") as f: + json.dump(meta, f, ensure_ascii=False, indent=2) + os.replace(tmp, AUTH_META_FILE) + + +async def _new_context(browser: Browser) -> BrowserContext: + """基于已保存的 storage_state 创建 context。""" + if os.path.exists(config.AUTH_STATE_FILE): + return await browser.new_context(storage_state=config.AUTH_STATE_FILE) + return await browser.new_context() + + +async def open_login_page(browser: Browser) -> None: + """打开登录页面（通常需要用户在 VNC/有界面环境里扫码）。""" + context = await _new_context(browser) + page = await context.new_page() + await page.goto(config.LOGIN_URL, wait_until="domcontentloaded") + # 不关闭页面：让用户扫码 + # 注意：context/page 生命周期由浏览器进程维护；任务结束时统一关闭 browser 即可 + + +async def verify_login(browser: Browser, timeout_sec: int = 20) -> Tuple[bool, str]: + """验证当前 storage_state 是否仍然有效。 + + 策略（尽量稳，避免依赖某个易变 selector）： + 1) 打开首页； + 2) 若跳转到 /login 或出现明显登录按钮/文案，则判定未登录； + 3) 若未跳登录页，并且存在一定数量 cookie，则判定登录有效（启发式）。 + """ + if not os.path.exists(config.AUTH_STATE_FILE): + return False, "未找到登录态文件，请先登录" + + context = await _new_context(browser) + page = await context.new_page() + + try: + await page.goto(config.XIAOHONGSHU_BASE_URL + "/", wait_until="domcontentloaded", timeout=timeout_sec * 1000) + await page.wait_for_timeout(800) + + url = page.url or "" + if "/login" in url: + return False, "已跳转到登录页（登录态可能已失效）" + + # 启发式：页面上是否出现明显“登录”入口 + login_text_hits = await page.locator("text=登录").count() + if login_text_hits > 0: + # 有些已登录页面也会出现“登录”字样（比如弹窗），再做一次 cookie 辅助判断 + cookies = await context.cookies() + if len(cookies) < 3: + return False, "检测到登录入口且 cookie 数量过少，疑似未登录" + + cookies = await context.cookies() + if len(cookies) < 3: + return False, "cookie 数量过少，疑似未登录" + + # 通过：写入 meta + meta = load_auth_meta() + if not meta.get("login_time"): + meta["login_time"] = _utc_iso() + meta["last_verified_time"] = _utc_iso() + save_auth_meta(meta) + + # 刷新保存一次 state（让 cookie 续期信息尽可能落盘） + try: + await context.storage_state(path=config.AUTH_STATE_FILE) + except Exception: + # 不要因为保存失败影响验证结果 + pass + + return True, "ok" + except PlaywrightTimeoutError: + return False, "验证登录超时（网络或风控），请稍后重试" + except Exception as e: + return False, f"验证登录异常: {e}" + finally: + try: + await page.close() + except Exception: + pass + try: + await context.close() + except Exception: + pass + + +async def get_login_status(browser: Optional[Browser]) -> LoginStatus: + """获取登录状态（如果 browser 可用则做轻量验证，否则仅返回本地 meta 状态）。""" + meta = load_auth_meta() + if not os.path.exists(config.AUTH_STATE_FILE): + return LoginStatus(logged_in=False, message="未登录") + + if browser is None: + return LoginStatus( + logged_in=True if meta.get("login_time") else True, + login_time=meta.get("login_time"), + last_verified_time=meta.get("last_verified_time"), + message="仅本地状态（未进行在线验证）" + ) + + ok, msg = await verify_login(browser) + meta = load_auth_meta() + return LoginStatus( + logged_in=ok, + login_time=meta.get("login_time"), + last_verified_time=meta.get("last_verified_time"), + message=None if ok else msg + ) + + +def clear_login_state() -> None: + """清理登录态。""" + for p in [config.AUTH_STATE_FILE, AUTH_META_FILE]: + try: + if os.path.exists(p): + os.remove(p) + except Exception: + pass diff --git a/app/log_manager.py b/app/log_manager.py index dadd79e..f754bd7 100644 --- a/app/log_manager.py +++ b/app/log_manager.py @@ -1,153 +1,77 @@ + +"""log_manager.py + +简单的线程安全日志管理器： +- 追加日志（level/message/thread_id） +- 持久化到 data/logs.json +- 提供读取最近 N 条、清空 + +说明：采集器是异步运行，但日志写入使用 threading.Lock 足够。 """ -Log Manager - 持久化日志管理器 -支持日志加载、保存、追加、清空 -""" + +from __future__ import annotations + import json import os import threading from datetime import datetime from typing import List, Dict, Optional + from app import config class LogManager: - """日志管理器 - 线程安全""" - - def __init__(self, log_file: str = None, max_logs: int = None): - self.log_file = log_file or config.LOG_FILE - self.max_logs = max_logs or config.MAX_LOGS + def __init__(self, log_file: Optional[str] = None, max_logs: int = 2000): + self.log_file = log_file or os.path.join(config.DATA_DIR, "logs.json") + self.max_logs = max_logs self._lock = threading.Lock() self.logs: List[Dict] = [] - self._load_logs() + self._load() - def _load_logs(self): - """启动时加载历史日志""" - with self._lock: - if os.path.exists(self.log_file): - try: - with open(self.log_file, 'r', encoding='utf-8') as f: - data = json.load(f) - self.logs = data.get('logs', []) - except (json.JSONDecodeError, IOError) as e: - print(f"[LogManager] 加载日志文件失败: {e}") - self.logs = [] - else: - self.logs = [] + def _load(self) -> None: + if not os.path.exists(self.log_file): + self.logs = [] + return + try: + with open(self.log_file, "r", encoding="utf-8") as f: + data = json.load(f) or {} + self.logs = data.get("logs", []) or [] + except Exception: + self.logs = [] + + def _save_locked(self) -> None: + os.makedirs(os.path.dirname(self.log_file), exist_ok=True) + tmp = self.log_file + ".tmp" + with open(tmp, "w", encoding="utf-8") as f: + json.dump( + {"logs": self.logs, "last_updated": datetime.now().isoformat(timespec="seconds")}, + f, + ensure_ascii=False, + indent=2, + ) + os.replace(tmp, self.log_file) def append(self, level: str, message: str, thread_id: Optional[str] = None) -> Dict: - """ - 追加新日志 - 返回日志条目 - """ - with self._lock: - log_entry = { - "timestamp": datetime.now().isoformat(), - "level": level, - "message": message, - "thread_id": thread_id - } - self.logs.append(log_entry) + entry = { + "time": datetime.now().isoformat(timespec="seconds"), + "level": level.upper(), + "message": message, + } + if thread_id: + entry["thread_id"] = thread_id - # 限制日志数量（保留最新的 max_logs 条） + with self._lock: + self.logs.append(entry) if len(self.logs) > self.max_logs: - self.logs = self.logs[-self.max_logs:] + self.logs = self.logs[-self.max_logs :] + self._save_locked() + return entry - # 异步保存到文件 - self._save() - - return log_entry - - def _save(self): - """保存到文件（需在锁内调用）""" - try: - with open(self.log_file, 'w', encoding='utf-8') as f: - json.dump({ - "logs": self.logs, - "last_updated": datetime.now().isoformat() - }, f, ensure_ascii=False, indent=2) - except IOError as e: - print(f"[LogManager] 保存日志文件失败: {e}") - - def get_logs(self, level: Optional[str] = None, limit: Optional[int] = None) -> List[Dict]: - """ - 获取日志 - :param level: 过滤级别 (None/INFO/SUCCESS/WARNING/ERROR/DEBUG) - :param limit: 限制返回数量 - """ - with self._lock: - if level: - filtered = [log for log in self.logs if log['level'] == level] - else: - filtered = self.logs - - if limit: - return filtered[-limit:] - return filtered - - def search_logs(self, keyword: str, limit: Optional[int] = None) -> List[Dict]: - """ - 搜索日志 - :param keyword: 搜索关键词 - :param limit: 限制返回数量 - """ - with self._lock: - filtered = [ - log for log in self.logs - if keyword.lower() in log['message'].lower() - ] - - if limit: - return filtered[-limit:] - return filtered - - def clear(self): - """清空日志""" + def clear(self) -> None: with self._lock: self.logs = [] - self._save() + self._save_locked() - def get_stats(self) -> Dict: - """获取日志统计信息""" + def get_latest(self, limit: int = 200) -> List[Dict]: with self._lock: - stats = { - "total": len(self.logs), - "by_level": {} - } - - for log in self.logs: - level = log['level'] - stats["by_level"][level] = stats["by_level"].get(level, 0) + 1 - - return stats - - def export_to_text(self) -> str: - """导出为文本格式""" - with self._lock: - lines = [] - lines.append("=" * 80) - lines.append("小红书采集器日志") - lines.append(f"导出时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}") - lines.append(f"总日志数: {len(self.logs)}") - lines.append("=" * 80) - lines.append("") - - for log in self.logs: - timestamp = log['timestamp'][:19] # 去掉微秒 - level = log['level'].ljust(8) - thread = f"[{log['thread_id']}]" if log.get('thread_id') else "" - message = log['message'] - lines.append(f"{timestamp} | {level} | {thread} {message}") - - return "\n".join(lines) - - -# 全局日志管理器实例 -_log_manager_instance = None - - -def get_log_manager() -> LogManager: - """获取全局日志管理器实例""" - global _log_manager_instance - if _log_manager_instance is None: - _log_manager_instance = LogManager() - return _log_manager_instance + return list(self.logs[-max(1, int(limit)) :]) diff --git a/app/scraper.py b/app/scraper.py index 81e698c..d7dddd0 100644 --- a/app/scraper.py +++ b/app/scraper.py @@ -467,16 +467,20 @@ class XiaoHongShuScraper: await self.emit_log("INFO", "启动单线程采集模式") # 创建 Context 并登录 - context = await self.browser.new_context( - viewport={'width': config.VIEWPORT_WIDTH, 'height': config.VIEWPORT_HEIGHT}, - user_agent=config.USER_AGENT - ) - - # 检查是否有保存的登录状态 + + # 创建 Context（优先加载已保存的 storage_state） if os.path.exists(config.AUTH_STATE_FILE): - await self.emit_log("INFO", "加载已保存的登录状态") - await context.add_cookies([json.load(open(config.AUTH_STATE_FILE))]) + await self.emit_log("INFO", "加载已保存的登录状态 (storage_state)") + context = await self.browser.new_context( + storage_state=config.AUTH_STATE_FILE, + viewport={'width': config.VIEWPORT_WIDTH, 'height': config.VIEWPORT_HEIGHT}, + user_agent=config.USER_AGENT + ) else: + context = await self.browser.new_context( + viewport={'width': config.VIEWPORT_WIDTH, 'height': config.VIEWPORT_HEIGHT}, + user_agent=config.USER_AGENT + ) # 需要登录 login_success = await self.handle_login(context) if not login_success: @@ -559,29 +563,31 @@ class XiaoHongShuScraper: """多线程模式""" await self.emit_log("INFO", f"启动多线程采集模式，线程数: {max_threads}") - # 创建多个 Context + + # 创建多个 Context（同一账号并发会增加风控风险，请合理控制线程数） contexts = [] - for i in range(max_threads): - ctx = await self.browser.new_context( + + # 确保已有登录态；若没有，先用第一个 context 登录并保存 + if not os.path.exists(config.AUTH_STATE_FILE): + first = await self.browser.new_context( viewport={'width': config.VIEWPORT_WIDTH, 'height': config.VIEWPORT_HEIGHT}, user_agent=config.USER_AGENT ) + login_success = await self.handle_login(first) + if not login_success: + await first.close() + return + await first.close() - # 第一个 Context 登录，其他复制 Cookies - if i == 0: - if os.path.exists(config.AUTH_STATE_FILE): - await self.emit_log("INFO", "加载已保存的登录状态") - await ctx.add_cookies([json.load(open(config.AUTH_STATE_FILE))]) - else: - login_success = await self.handle_login(ctx) - if not login_success: - await ctx.close() - return - else: - await self.copy_cookies(ctx, contexts[0]) - await self.emit_log("INFO", f"Thread-{i+1} 已复制登录状态") - + # 基于 storage_state 为每个线程创建独立 context + for i in range(max_threads): + ctx = await self.browser.new_context( + storage_state=config.AUTH_STATE_FILE, + viewport={'width': config.VIEWPORT_WIDTH, 'height': config.VIEWPORT_HEIGHT}, + user_agent=config.USER_AGENT + ) contexts.append(ctx) + await self.emit_log("INFO", f"Thread-{i+1} context ready") # 创建任务队列 task_queue = asyncio.Queue() diff --git a/app/user_resolver.py b/app/user_resolver.py new file mode 100644 index 0000000..e51027c --- /dev/null +++ b/app/user_resolver.py @@ -0,0 +1,153 @@ + +"""user_resolver.py + +将“数字小红书号（例如 11432840954）”解析成 profile_id（/user/profile/{profile_id} 里的那串）。 +- 使用 Playwright 打开搜索结果页（更接近真实浏览器，风险更低） +- 解析 HTML：根据 '小红书号：' 定位，再提取附近的 profile href +- 强制缓存：data/profile_map.json，避免重复搜索造成风控压力 +""" + +from __future__ import annotations + +import json +import os +import random +import re +from html import unescape +from typing import Dict, Optional, Tuple + +from playwright.async_api import Browser + +from app import config + +PROFILE_MAP_FILE = os.path.join(config.DATA_DIR, "profile_map.json") + + +def _load_map() -> Dict[str, str]: + if not os.path.exists(PROFILE_MAP_FILE): + return {} + try: + with open(PROFILE_MAP_FILE, "r", encoding="utf-8") as f: + return json.load(f) or {} + except Exception: + return {} + + +def _save_map(m: Dict[str, str]) -> None: + tmp = PROFILE_MAP_FILE + ".tmp" + with open(tmp, "w", encoding="utf-8") as f: + json.dump(m, f, ensure_ascii=False, indent=2) + os.replace(tmp, PROFILE_MAP_FILE) + + +def extract_profile_id_from_search_html(html: str, keyword: str) -> Optional[str]: + """从 search_result HTML 中解析出对应 keyword 的 profile_id。""" + if not html: + return None + + html = unescape(html) + marker = f"小红书号：{keyword}" + idx = html.find(marker) + if idx == -1: + return None + + start = max(0, idx - 5000) + end = min(len(html), idx + 5000) + chunk = html[start:end] + + m = re.search(r'href="\/user\/profile\/([0-9a-fA-F]+)', chunk) + if not m: + return None + return m.group(1) + + +def parse_line_to_keyword_or_profile(line: str) -> Tuple[Optional[str], Optional[str], Optional[str]]: + """解析一行 txt，返回 (keyword数字, profile_id, label备注) 三者之一。 + + 允许格式： + - 11432840954 + - 11432840954 张三 + - https://www.xiaohongshu.com/user/profile/ ... + - /user/profile/ ... + - profile_id(十六进制串) + - 支持行尾 # 备注 + """ + raw = line.strip() + if not raw: + return None, None, None + + # 整行注释 + if raw.startswith("#") or raw.startswith("//"): + return None, None, None + + # 拆备注：支持 # 以及连续空格 + label = None + if "#" in raw: + raw, label = raw.split("#", 1) + raw = raw.strip() + label = label.strip() or None + + parts = raw.split() + core = parts[0].strip() if parts else "" + if len(parts) > 1 and not label: + label = " ".join(parts[1:]).strip() or None + + # URL/相对路径中直接提取 profile_id + m = re.search(r"/user/profile/([0-9a-fA-F]+)", core) + if m: + return None, m.group(1), label + + # 可能直接就是 profile_id（十六进制串，长度>=8） + if re.fullmatch(r"[0-9a-fA-F]{8,}", core): + return None, core, label + + # 纯数字：小红书号 + if re.fullmatch(r"\d{5,20}", core): + return core, None, label + + return None, None, label + + +async def resolve_keyword_to_profile_id(browser: Browser, keyword: str, *, slow_mode: bool = True) -> Tuple[Optional[str], str]: + """将数字 keyword 解析成 profile_id，并写入缓存。返回 (profile_id, message)。""" + m = _load_map() + if keyword in m: + return m[keyword], "cache" + + # 使用新的 context（不强制登录） + context = await browser.new_context() + page = await context.new_page() + + try: + url = f"{config.XIAOHONGSHU_BASE_URL}/search_result?keyword={keyword}&source=unknown" + await page.goto(url, wait_until="domcontentloaded", timeout=20000) + + # 轻微等待/滚动（更像真人浏览） + if slow_mode: + await page.wait_for_timeout(int(random.uniform(800, 1600))) + # 小幅滚动一下 + try: + await page.mouse.wheel(0, random.randint(200, 600)) + except Exception: + pass + await page.wait_for_timeout(int(random.uniform(400, 1000))) + + html = await page.content() + pid = extract_profile_id_from_search_html(html, keyword) + if not pid: + return None, "not_found" + + m[keyword] = pid + _save_map(m) + return pid, "ok" + except Exception as e: + return None, f"error:{e}" + finally: + try: + await page.close() + except Exception: + pass + try: + await context.close() + except Exception: + pass diff --git a/app/web.py b/app/web.py index 58af4bd..b5ff2c6 100644 --- a/app/web.py +++ b/app/web.py @@ -1,1096 +1,478 @@ + +"""web.py - FastAPI 后端 + +目标： +- 提供基本的 Web UI + API（启动/停止采集、导入用户、登录/验证、日志、SSE） +- 单账号登录态长期复用（storage_state 保存） +- txt 导入：支持“数字小红书号”自动解析为 profile_id（并缓存，低频，减少风控） +- 移除无用/冗余的调试/自更新接口，代码保持清晰可维护 """ -Web Backend - FastAPI 应用 -提供 API 和 SSE 事件推送 -""" + +from __future__ import annotations + import asyncio import json import os -import subprocess -from datetime import datetime, timedelta -from typing import Dict, List, Optional +from dataclasses import dataclass +from typing import List, Optional, Dict, Any + from fastapi import FastAPI, HTTPException, BackgroundTasks -from fastapi.responses import HTMLResponse, JSONResponse, StreamingResponse +from fastapi.responses import HTMLResponse, StreamingResponse, PlainTextResponse from fastapi.staticfiles import StaticFiles from fastapi.templating import Jinja2Templates from fastapi.requests import Request -from pydantic import BaseModel -from typing import List as ListType from fastapi.middleware.cors import CORSMiddleware +from pydantic import BaseModel + +from playwright.async_api import async_playwright, Browser from app import config -from app.log_manager import get_log_manager +from app.log_manager import LogManager from app.scraper import XiaoHongShuScraper +from app import auth_manager +from app import user_resolver -# ========== Pydantic 模型 ========== +# ========================= +# App State +# ========================= + +@dataclass +class AppState: + is_running: bool = False + scraper: Optional[XiaoHongShuScraper] = None + event_queue: Optional[asyncio.Queue] = None + playwright: Any = None + browser: Optional[Browser] = None + + +app_state = AppState(event_queue=asyncio.Queue()) +log_manager = LogManager() + + +# ========================= +# FastAPI init +# ========================= + +app = FastAPI(title="小红书数据采集器", version="1.1.0") + +app.add_middleware( + CORSMiddleware, + allow_origins=["*"], + allow_credentials=True, + allow_methods=["*"], + allow_headers=["*"], + expose_headers=["*"], +) + +# 静态资源 + 模板 +app.mount("/static", StaticFiles(directory=os.path.join(config.BASE_DIR, "static")), name="static") +templates = Jinja2Templates(directory=os.path.join(config.BASE_DIR, "templates")) + + +@app.on_event("startup") +async def on_startup(): + # 启动 Playwright + Browser（长期复用，减少反复启动带来的抖动） + app_state.playwright = await async_playwright().start() + app_state.browser = await app_state.playwright.chromium.launch( + headless=config.HEADLESS, + args=config.BROWSER_ARGS, + ) + log_manager.append("INFO", f"Browser launched (headless={config.HEADLESS})") + + +@app.on_event("shutdown") +async def on_shutdown(): + # 安全关闭 + try: + if app_state.browser: + await app_state.browser.close() + finally: + app_state.browser = None + + try: + if app_state.playwright: + await app_state.playwright.stop() + finally: + app_state.playwright = None + + +# ========================= +# Models +# ========================= + class StartRequest(BaseModel): mode: str = "single" # single / multi max_threads: int = 3 class StatusResponse(BaseModel): - is_running: bool - mode: str - max_threads: int - total_users: int - completed_users: int - failed_users: int - active_threads: int - users_status: Dict[str, Dict] + running: bool + mode: Optional[str] = None + user_count: int = 0 + + +class UsersResponse(BaseModel): + users: List[str] + count: int + + +class ImportResponse(BaseModel): + users: List[str] + count: int + total_lines: int + imported: int + skipped: int + resolved: int + failed_resolve: int + duplicates: int + invalid: int + message: str + # 可选返回一些失败明细（避免过大，只给前 20 条） + failed_examples: List[str] = [] -# ========== 全局状态 ========== -class AppState: - def __init__(self): - self.is_running = False - self.mode = "single" - self.max_threads = 1 - self.total_users = 0 - self.completed_users = 0 - self.failed_users = 0 - self.active_threads = 0 - self.users_status: Dict[str, Dict] = {} - self.thread_status: Dict[str, Dict] = {} - self.event_queue: asyncio.Queue = None - self.scraper: Optional[XiaoHongShuScraper] = None - self.scraper_task: Optional[asyncio.Task] = None - self.browser = None - self.playwright = None - - -app_state = AppState() -log_manager = get_log_manager() - -# ========== FastAPI 应用 ========== -app = FastAPI(title="小红书数据采集器", version="1.0.0") - -# 添加 CORS 中间件 -app.add_middleware( - CORSMiddleware, - allow_origins=["*"], # 生产环境应该指定具体域名 - allow_credentials=True, - allow_methods=["*"], - allow_headers=["*"], - expose_headers=["*"] -) - -# 挂载静态文件和模板 -app.mount("/static", StaticFiles(directory="static"), name="static") -templates = Jinja2Templates(directory="templates") - - -# ========== 路由 ========== - -@app.get("/", response_class=HTMLResponse) -async def index(request: Request): - """首页""" - return templates.TemplateResponse("index.html", {"request": request}) - - - - -@app.get("/api/status", response_model=StatusResponse) -async def get_status(): - """获取当前状态""" - return StatusResponse( - is_running=app_state.is_running, - mode=app_state.mode, - max_threads=app_state.max_threads, - total_users=app_state.total_users, - completed_users=app_state.completed_users, - failed_users=app_state.failed_users, - active_threads=app_state.active_threads, - users_status=app_state.users_status - ) - - -@app.get("/api/debug/browser") -async def debug_browser(): - """调试浏览器状态""" - return { - "browser_exists": app_state.browser is not None, - "contexts_count": len(app_state.browser.contexts) if app_state.browser else 0, - "contexts": [str(type(ctx)) for ctx in app_state.browser.contexts] if app_state.browser else [], - "data_dir": config.DATA_DIR, - "auth_file_exists": os.path.exists(os.path.join(config.DATA_DIR, "auth_state.json")), - "storage_file_exists": os.path.exists(os.path.join(config.DATA_DIR, "storage_state.json")) - } - - -@app.post("/api/start") -async def start_scraping(request: StartRequest): - """启动采集""" - if app_state.is_running: - raise HTTPException(status_code=400, detail="采集任务正在运行中") - - # 读取用户 ID 列表 - users_file = os.path.join(config.DATA_DIR, "users.txt") - if not os.path.exists(users_file): - raise HTTPException(status_code=400, detail="用户文件不存在，请先创建 data/users.txt") - - with open(users_file, 'r', encoding='utf-8') as f: - user_ids = [line.strip() for line in f if line.strip()] - - if not user_ids: - raise HTTPException(status_code=400, detail="用户 ID 列表为空") - - # 验证模式 - if request.mode not in ["single", "multi"]: - raise HTTPException(status_code=400, detail="无效的模式，必须是 single 或 multi") - - # 验证线程数 - if request.mode == "multi": - if request.max_threads < 1 or request.max_threads > config.MAX_CONCURRENT_THREADS: - raise HTTPException(status_code=400, detail=f"线程数必须在 1-{config.MAX_CONCURRENT_THREADS} 之间") - - # 重置状态 - app_state.is_running = True - app_state.mode = request.mode - app_state.max_threads = request.max_threads if request.mode == "multi" else 1 - app_state.total_users = len(user_ids) - app_state.completed_users = 0 - app_state.failed_users = 0 - app_state.active_threads = 0 - app_state.users_status = {} - app_state.thread_status = {} - - # 初始化用户状态 - for user_id in user_ids: - app_state.users_status[user_id] = { - "status": "pending", - "note_count": 0, - "thread_id": None - } - - # 创建事件队列 - app_state.event_queue = asyncio.Queue() - - # 创建采集器（传入已初始化的浏览器实例） - app_state.scraper = XiaoHongShuScraper(app_state.event_queue, log_manager, app_state.browser) - - # 启动采集任务 - async def run_scraper(): - try: - await app_state.scraper.run( - user_ids=user_ids, - mode=request.mode, - max_threads=request.max_threads - ) - finally: - app_state.is_running = False - await app_state.event_queue.put({ - "type": "finished" - }) - - app_state.scraper_task = asyncio.create_task(run_scraper()) - - return { - "success": True, - "message": f"采集任务已启动，模式: {request.mode}, 用户数: {len(user_ids)}", - "total_users": len(user_ids), - "mode": request.mode, - "max_threads": app_state.max_threads - } - - -@app.post("/api/stop") -async def stop_scraping(): - """停止采集""" - if not app_state.is_running: - raise HTTPException(status_code=400, detail="没有正在运行的采集任务") - - if app_state.scraper: - app_state.scraper.stop() - - # 取消任务 - if app_state.scraper_task: - app_state.scraper_task.cancel() - try: - await app_state.scraper_task - except asyncio.CancelledError: - pass - - app_state.is_running = False - - return { - "success": True, - "message": "采集任务已停止" - } - - -@app.post("/api/logs/clear") -async def clear_logs(): - """清空日志""" - log_manager.clear() - return { - "success": True, - "message": "日志已清空" - } - - -@app.get("/api/logs") -async def get_logs(level: Optional[str] = None, limit: Optional[int] = None): - """获取日志""" - logs = log_manager.get_logs(level=level, limit=limit) - return { - "logs": logs, - "stats": log_manager.get_stats() - } - - -@app.get("/api/logs/search") -async def search_logs(keyword: str, limit: Optional[int] = None): - """搜索日志""" - logs = log_manager.search_logs(keyword, limit) - return { - "logs": logs, - "count": len(logs) - } - - -@app.get("/api/logs/download") -async def download_logs(): - """下载日志文件""" - text = log_manager.export_to_text() - - return StreamingResponse( - iter([text]), - media_type="text/plain", - headers={ - "Content-Disposition": f"attachment; filename=xiaohongshu_logs_{datetime.now().strftime('%Y%m%d_%H%M%S')}.txt" - } - ) - - -@app.get("/api/users") -async def get_users(): - """获取用户列表""" - users_file = os.path.join(config.DATA_DIR, "users.txt") - - if not os.path.exists(users_file): - return {"users": []} - - with open(users_file, 'r', encoding='utf-8') as f: - users = [{"id": line.strip()} for line in f if line.strip()] - - # 添加采集状态 - for user in users: - user_id = user["id"] - if user_id in app_state.users_status: - user.update(app_state.users_status[user_id]) - else: - user["status"] = "pending" - user["note_count"] = 0 - - return {"users": users} - - -@app.post("/api/users") -async def add_user(request: dict): - """添加用户""" - user_id = request.get("user_id", "").strip() - - if not user_id: - raise HTTPException(status_code=400, detail="用户ID不能为空") - - # 验证用户ID格式 - if not user_id or len(user_id) < 10: - raise HTTPException(status_code=400, detail="用户ID格式不正确") - - users_file = os.path.join(config.DATA_DIR, "users.txt") - - # 检查是否已存在 - if os.path.exists(users_file): - with open(users_file, 'r', encoding='utf-8') as f: - existing_users = [line.strip() for line in f if line.strip()] - if user_id in existing_users: - raise HTTPException(status_code=400, detail="用户ID已存在") - - # 添加到文件 - with open(users_file, 'a', encoding='utf-8') as f: - f.write(f"{user_id}\n") - - return { - "success": True, - "message": f"用户 {user_id} 已添加", - "user_id": user_id - } - - -@app.delete("/api/users/{user_id}") -async def remove_user(user_id: str): - """删除用户""" - users_file = os.path.join(config.DATA_DIR, "users.txt") - - if not os.path.exists(users_file): - raise HTTPException(status_code=404, detail="用户文件不存在") - - # 读取所有用户 - with open(users_file, 'r', encoding='utf-8') as f: - users = [line.strip() for line in f if line.strip()] - - # 检查用户是否存在 - if user_id not in users: - raise HTTPException(status_code=404, detail="用户不存在") - - # 删除用户 - users.remove(user_id) - - # 写回文件 - with open(users_file, 'w', encoding='utf-8') as f: - for user in users: - f.write(f"{user}\n") - - # 删除采集数据文件 - data_file = os.path.join(config.DATA_DIR, f"{user_id}.json") - if os.path.exists(data_file): - os.remove(data_file) - - return { - "success": True, - "message": f"用户 {user_id} 已删除" - } - - -@app.delete("/api/users") -async def clear_all_users(): - """清空所有用户""" - users_file = os.path.join(config.DATA_DIR, "users.txt") - - if not os.path.exists(users_file): - raise HTTPException(status_code=404, detail="用户文件不存在") - - # 清空文件 - with open(users_file, 'w', encoding='utf-8') as f: - f.write("") - - # 删除所有采集数据 - data_dir = config.DATA_DIR - for filename in os.listdir(data_dir): - if filename.endswith('.json') and filename != 'auth_state.json': - file_path = os.path.join(data_dir, filename) - os.remove(file_path) - - # 重置状态 - app_state.users_status = {} - - return { - "success": True, - "message": "所有用户已清空" - } - - -@app.post("/api/users/import/txt") -async def import_users_from_txt(): - """从users.txt导入用户（模拟）""" - users_file = os.path.join(config.DATA_DIR, "users.txt") - - if not os.path.exists(users_file): - raise HTTPException(status_code=404, detail="users.txt文件不存在") - - with open(users_file, 'r', encoding='utf-8') as f: - user_ids = [line.strip() for line in f if line.strip()] - - return { - "success": True, - "count": len(user_ids), - "users": user_ids - } - - -@app.get("/api/users/export/txt") -async def export_users_to_txt(): - """导出用户列表到txt文件""" - users_file = os.path.join(config.DATA_DIR, "users.txt") - - if not os.path.exists(users_file): - raise HTTPException(status_code=404, detail="用户文件不存在") - - with open(users_file, 'r', encoding='utf-8') as f: - content = f.read() - - return StreamingResponse( - iter([content]), - media_type="text/plain", - headers={ - "Content-Disposition": f"attachment; filename=users_{datetime.now().strftime('%Y%m%d_%H%M%S')}.txt" - } - ) - - -# ========== 登录相关 API ========== - class LoginStatusResponse(BaseModel): logged_in: bool login_time: Optional[str] = None - expire_time: Optional[str] = None + last_verified_time: Optional[str] = None + message: Optional[str] = None +# ========================= +# Pages +# ========================= + +@app.get("/", response_class=HTMLResponse) +async def index(request: Request): + return templates.TemplateResponse("index.html", {"request": request}) + + +# ========================= +# Helpers +# ========================= + +def _users_file() -> str: + return os.path.join(config.DATA_DIR, "users.txt") + + +def read_users_txt() -> List[str]: + path = _users_file() + if not os.path.exists(path): + return [] + users: List[str] = [] + with open(path, "r", encoding="utf-8") as f: + for line in f: + line = line.strip() + if not line: + continue + users.append(line) + # 去重但保持顺序 + seen = set() + out = [] + for u in users: + if u in seen: + continue + seen.add(u) + out.append(u) + return out + + +def write_users_txt(users: List[str]) -> None: + path = _users_file() + tmp = path + ".tmp" + with open(tmp, "w", encoding="utf-8") as f: + for u in users: + f.write(u + "\n") + os.replace(tmp, path) + + +def require_browser() -> Browser: + if app_state.browser is None: + raise HTTPException(status_code=500, detail="Browser not ready") + return app_state.browser + + +# ========================= +# Status / Logs +# ========================= + +@app.get("/api/status", response_model=StatusResponse) +async def api_status(): + return StatusResponse( + running=app_state.is_running, + mode=getattr(app_state.scraper, "mode", None) if app_state.scraper else None, + user_count=len(read_users_txt()), + ) + + +@app.post("/api/logs/clear") +async def clear_logs(): + log_manager.clear() + return {"success": True} + + +@app.get("/api/logs") +async def get_logs(limit: int = 200): + return {"logs": log_manager.get_latest(limit=limit)} + + +# ========================= +# Users +# ========================= + +@app.get("/api/users", response_model=UsersResponse) +async def get_users(): + users = read_users_txt() + return UsersResponse(users=users, count=len(users)) + + +@app.post("/api/users") +async def add_user(user_id: str): + user_id = (user_id or "").strip() + if not user_id: + raise HTTPException(status_code=400, detail="user_id 不能为空") + + users = read_users_txt() + if user_id in users: + return {"success": True, "message": "已存在"} + + users.append(user_id) + write_users_txt(users) + return {"success": True} + + +@app.delete("/api/users/{user_id}") +async def delete_user(user_id: str): + users = read_users_txt() + users = [u for u in users if u != user_id] + write_users_txt(users) + return {"success": True} + + +@app.delete("/api/users") +async def delete_all_users(): + write_users_txt([]) + return {"success": True} + + +@app.get("/api/users/export/txt") +async def export_users_txt(): + return PlainTextResponse("\\n".join(read_users_txt())) + + +@app.get("/api/users/import/txt", response_model=ImportResponse) +async def import_users_from_txt(): + """从 data/users.txt 读取并规范化： + - 若是 profile_id / profile URL：直接收录 + - 若是纯数字小红书号：通过搜索页解析成 profile_id（Playwright，低频、缓存） + - 输出：users.txt 最终仅保存 profile_id（稳定可采集） + """ + path = _users_file() + if not os.path.exists(path): + raise HTTPException(status_code=400, detail="data/users.txt 不存在") + + with open(path, "r", encoding="utf-8") as f: + raw_lines = [l.rstrip("\n") for l in f] + + total_lines = len(raw_lines) + + browser = require_browser() + + imported_ids: List[str] = [] + duplicates = 0 + invalid = 0 + skipped = 0 + resolved = 0 + failed_resolve = 0 + failed_examples: List[str] = [] + + seen = set() + + # 第一轮：先把能直接解析的收集，数字 keyword 单独列表 + keywords: List[tuple[str, str]] = [] # (keyword, raw_line) + for line in raw_lines: + kw, pid, _label = user_resolver.parse_line_to_keyword_or_profile(line) + if pid: + pid = pid.lower() + if pid in seen: + duplicates += 1 + continue + seen.add(pid) + imported_ids.append(pid) + continue + if kw: + keywords.append((kw, line)) + else: + # 空行/注释等视为 skipped；其它无法识别为 invalid + if not line.strip() or line.strip().startswith("#") or line.strip().startswith("//"): + skipped += 1 + else: + invalid += 1 + if len(failed_examples) < 20: + failed_examples.append(f"invalid: {line}") + continue + + # 第二轮：逐个解析 keyword（单线程+慢一点，减少风控） + for kw, raw in keywords: + pid, msg = await user_resolver.resolve_keyword_to_profile_id(browser, kw, slow_mode=True) + if not pid: + failed_resolve += 1 + if len(failed_examples) < 20: + failed_examples.append(f"resolve_failed({kw}): {raw} ({msg})") + continue + pid = pid.lower() + if pid in seen: + duplicates += 1 + continue + seen.add(pid) + imported_ids.append(pid) + resolved += 1 + + # 每解析一个稍微间隔一下（更像真人） + await asyncio.sleep(0.8) + + # 写回：只保存 profile_id + write_users_txt(imported_ids) + + imported = len(imported_ids) + message = f"导入完成：{imported} 个 profile_id（解析成功 {resolved}）" + + log_manager.append("INFO", message) + + return ImportResponse( + users=imported_ids, + count=imported, + total_lines=total_lines, + imported=imported, + skipped=skipped, + resolved=resolved, + failed_resolve=failed_resolve, + duplicates=duplicates, + invalid=invalid, + message=message, + failed_examples=failed_examples, + ) + + +# ========================= +# Login +# ========================= + @app.get("/api/login/status", response_model=LoginStatusResponse) async def get_login_status(): - """获取登录状态""" - auth_file = os.path.join(config.DATA_DIR, "auth_state.json") - - if not os.path.exists(auth_file): - return LoginStatusResponse(logged_in=False) - - try: - with open(auth_file, 'r', encoding='utf-8') as f: - auth_data = json.load(f) - - # 检查是否过期 - login_time = auth_data.get("login_time") - expire_time = auth_data.get("expire_time") - current_time = datetime.now().isoformat() - - if expire_time and current_time > expire_time: - # 登录已过期，清除状态 - os.remove(auth_file) - return LoginStatusResponse(logged_in=False) - - return LoginStatusResponse( - logged_in=auth_data.get("logged_in", False), - login_time=login_time, - expire_time=expire_time - ) - except Exception: - return LoginStatusResponse(logged_in=False) + status = await auth_manager.get_login_status(app_state.browser) + return LoginStatusResponse( + logged_in=status.logged_in, + login_time=status.login_time, + last_verified_time=status.last_verified_time, + message=status.message, + ) @app.post("/api/login/qrcode") -async def generate_qrcode(): - """重定向到小红书官方登录页面""" - # 返回小红书登录页面 URL - login_url = "https://www.xiaohongshu.com/login" +async def open_login(): + """兼容旧前端：返回 login_url，同时在服务端打开 Playwright 登录页。 + 说明：前端 window.open 打开的页面不会影响服务端 Playwright 的登录态。 + 正确使用方式：在部署环境（Docker + VNC）里打开 VNC 浏览器扫码登录。 + """ + browser = require_browser() + await auth_manager.open_login_page(browser) return { - "login_url": login_url, - "message": "请在打开的小红书页面完成扫码登录", - "instruction": "登录成功后，返回此页面点击'验证登录'按钮" + "login_url": config.LOGIN_URL, + "hint": "已在服务端打开登录页。请在 VNC/有界面环境里扫码登录，然后点击“验证登录”。" } -@app.get("/api/check-login-status") -async def check_login_status(): - """检查小红书登录状态（直接验证）""" - try: - print("[DEBUG] === 检查浏览器状态 ===") - print(f"[DEBUG] 浏览器实例存在: {app_state.browser is not None}") - - # 如果浏览器可用，直接使用浏览器验证 - if app_state.browser: - print("[DEBUG] 浏览器可用，尝试获取context...") - if app_state.browser.contexts: - print(f"[DEBUG] 已有 {len(app_state.browser.contexts)} 个contexts") - context = app_state.browser.contexts[0] - print("[DEBUG] 使用现有context") - else: - print("[DEBUG] 创建新context") - context = await app_state.browser.new_context() - - try: - page = await context.new_page() - await page.goto("https://www.xiaohongshu.com/", timeout=10000) - await asyncio.sleep(2) - - # 获取页面cookies - cookies = await context.cookies() - print(f"[DEBUG] 找到 {len(cookies)} 个cookies") - - # 打印所有cookies - for cookie in cookies: - cookie_name = cookie.get('name', '') - cookie_value = cookie.get('value', '')[:20] if cookie.get('value') else '' - print(f"[DEBUG] Cookie: {cookie_name} = {cookie_value}...") - - # 检查是否有小红书相关的session cookie - session_cookies = [] - for cookie in cookies: - cookie_name = cookie.get('name', '').lower() - if any(keyword in cookie_name for keyword in ['session', 'token', 'login', 'auth']): - session_cookies.append(cookie) - print(f"[DEBUG] ✓ 找到登录cookie: {cookie_name}") - - if session_cookies: - return { - "success": True, - "message": "已登录小红书", - "logged_in": True, - "cookies_count": len(session_cookies) - } - else: - print("[DEBUG] ✗ 未找到登录相关的 cookies") - return { - "success": False, - "message": "未找到登录状态，请先登录" - } - - except Exception as e: - print(f"[DEBUG] 浏览器页面操作失败: {str(e)}") - pass - finally: - # 确保页面关闭 - if 'page' in locals(): - await page.close() - else: - print("[DEBUG] ✗ 浏览器实例不存在") - - # 检查本地是否有登录状态文件 - print("[DEBUG] 检查本地登录状态文件...") - auth_file = os.path.join(config.DATA_DIR, "auth_state.json") - if os.path.exists(auth_file): - try: - with open(auth_file, 'r', encoding='utf-8') as f: - auth_data = json.load(f) - print(f"[DEBUG] ✓ 找到本地auth_state: {auth_data}") - return { - "success": True, - "message": "本地登录状态有效", - "logged_in": True, - "login_time": auth_data.get("login_time"), - "expire_time": auth_data.get("expire_time") - } - except Exception as e: - print(f"[DEBUG] ✗ 读取auth_state失败: {str(e)}") - - print("[DEBUG] ✗ 未找到任何登录状态") - return { - "success": False, - "message": "未找到登录状态，请先登录" - } - - except Exception as e: - print(f"[DEBUG] ✗ 检查登录状态异常: {str(e)}") - return { - "success": False, - "message": f"检查登录状态失败: {str(e)}" - } - - @app.post("/api/login/verify") async def verify_login(): - """验证登录状态""" - try: - print("[DEBUG] 开始验证登录状态...") - - # 首先尝试保存 storage_state - if app_state.browser and app_state.browser.context: - try: - # 创建 storage_state 文件 - storage_file = os.path.join(config.DATA_DIR, "storage_state.json") - print(f"[DEBUG] 尝试保存 storage_state 到: {storage_file}") - await app_state.browser.context.storage_state(path=storage_file) - print(f"[DEBUG] ✓ 成功保存 storage_state: {storage_file}") - - # 检查文件是否存在和大小 - if os.path.exists(storage_file): - size = os.path.getsize(storage_file) - print(f"[DEBUG] storage_state 文件大小: {size} bytes") - if size > 0: - # 使用 storage_state 验证登录 - print("[DEBUG] 使用 storage_state 验证登录...") - return await verify_login_with_storage(storage_file) - else: - print("[DEBUG] storage_state 文件为空") - else: - print("[DEBUG] storage_state 文件未创建") - - except Exception as e: - print(f"[DEBUG] ✗ 保存 storage_state 失败: {str(e)}") - pass - - # 如果保存失败，尝试直接检查 cookies - print("[DEBUG] 尝试直接检查 cookies...") - if app_state.browser: - try: - cookies = await app_state.browser.context.cookies() - print(f"[DEBUG] 找到 {len(cookies)} 个cookies") - - # 打印所有 cookies 用于调试 - for cookie in cookies: - cookie_name = cookie.get('name', '') - cookie_value = cookie.get('value', '')[:20] if cookie.get('value') else '' - print(f"[DEBUG] Cookie: {cookie_name} = {cookie_value}...") - - # 检查是否有登录相关的 cookie - session_cookies = [] - for cookie in cookies: - cookie_name = cookie.get('name', '').lower() - if any(keyword in cookie_name for keyword in ['session', 'token', 'auth', 'passport', 'sid']): - session_cookies.append(cookie) - print(f"[DEBUG] ✓ 找到登录cookie: {cookie_name}") - - # 创建 auth_state 文件 - auth_file = os.path.join(config.DATA_DIR, "auth_state.json") - auth_data = { - "login_time": datetime.now().isoformat(), - "expire_time": (datetime.now() + timedelta(hours=24)).isoformat() - } - with open(auth_file, 'w', encoding='utf-8') as f: - json.dump(auth_data, f, ensure_ascii=False, indent=2) - - print("[DEBUG] ✓ 创建 auth_state 文件成功") - return { - "success": True, - "message": "登录成功", - "logged_in": True, - "login_time": auth_data["login_time"], - "expire_time": auth_data["expire_time"], - "expire_in_hours": 24 - } - - if not session_cookies: - print("[DEBUG] ✗ 未找到登录相关的 cookies") - - except Exception as e: - print(f"[DEBUG] ✗ 检查cookies失败: {str(e)}") - pass - - # 检查本地登录状态文件 - print("[DEBUG] 检查本地登录状态文件...") - auth_file = os.path.join(config.DATA_DIR, "auth_state.json") - if os.path.exists(auth_file): - try: - with open(auth_file, 'r', encoding='utf-8') as f: - auth_data = json.load(f) - print(f"[DEBUG] ✓ 找到本地 auth_state 文件: {auth_data}") - - # 检查登录是否过期 - login_time = auth_data.get("login_time") - expire_time = auth_data.get("expire_time") - current_time = datetime.now().isoformat() - - if expire_time and current_time > expire_time: - # 登录已过期，清除状态 - print("[DEBUG] ✗ 登录已过期") - os.remove(auth_file) - return { - "success": False, - "message": "登录已过期，请重新登录" - } - - print("[DEBUG] ✓ 本地登录状态有效") - return { - "success": True, - "message": "登录成功", - "logged_in": True, - "login_time": login_time, - "expire_time": expire_time, - "expire_in_hours": 24 - } - except Exception as e: - print(f"[DEBUG] ✗ 读取 auth_state 文件失败: {str(e)}") - - print("[DEBUG] ✗ 未找到任何登录状态") - return { - "success": False, - "message": "未找到登录状态，请先登录" - } - - except Exception as e: - print(f"[DEBUG] ✗ 验证登录异常: {str(e)}") - return { - "success": False, - "message": f"验证失败: {str(e)}" - } - - -async def verify_login_with_storage(storage_file: str): - """使用 storage_state 文件验证登录状态""" - try: - from pathlib import Path - from playwright.async_api import async_playwright - - pth = Path(storage_file) - if not pth.exists(): - return { - "success": False, - "message": "未找到登录状态文件" - } - - # 使用 async_playwright 创建临时浏览器实例验证 - async with async_playwright() as p: - # 使用与主浏览器相同的配置 - browser = await p.chromium.launch( - headless=True, # 无头模式验证 - args=["--no-sandbox", "--disable-setuid-sandbox"] - ) - - # 使用 storage_state 创建新 context - context = await browser.new_context( - storage_state=str(pth), - locale="zh-CN", - viewport={"width": 1200, "height": 800} - ) - - page = await context.new_page() - - # 访问小红书主页 - await page.goto("https://www.xiaohongshu.com", wait_until="domcontentloaded", timeout=60000) - await asyncio.sleep(1.2) # 等待页面加载 - - # 强拦截关键词 - 使用更可靠的元素检查方式 - block_texts = ["扫码登录", "安全验证", "请先登录", "登录后查看"] - for text in block_texts: - try: - if await page.get_by_text(text).first.is_visible(): - await browser.close() - return { - "success": False, - "message": f"登录状态无效或被拦截：页面出现「{text}」" - } - except Exception: - pass - - # 弱判断：若明显出现"登录"且可见，可能未登录 - try: - if await page.get_by_text("登录").first.is_visible(): - await browser.close() - return { - "success": False, - "message": "疑似未登录（页面出现明显「登录」入口），建议重新登录保存状态" - } - except Exception: - pass - - # 检查 URL 是否被重定向 - current_url = page.url - if "login" in current_url.lower() or "404" in current_url: - await browser.close() - return { - "success": False, - "message": "被重定向到登录页面，登录状态无效" - } - - # 检查页面标题 - title = await page.title() - if "登录" in title or "404" in title: - await browser.close() - return { - "success": False, - "message": "页面标题显示登录或错误，登录状态无效" - } - - # 所有检查都通过，认为登录有效 - await browser.close() - - # 创建 auth_state 文件作为备份 - auth_file = os.path.join(config.DATA_DIR, "auth_state.json") - auth_data = { - "login_time": datetime.now().isoformat(), - "expire_time": (datetime.now() + timedelta(hours=24)).isoformat() - } - with open(auth_file, 'w', encoding='utf-8') as f: - json.dump(auth_data, f, ensure_ascii=False, indent=2) - - print(f"[DEBUG] 登录验证成功") - return { - "success": True, - "message": "登录成功", - "logged_in": True, - "login_time": auth_data["login_time"], - "expire_time": auth_data["expire_time"], - "expire_in_hours": 24 - } - - except Exception as e: - print(f"[DEBUG] storage_state 验证失败: {str(e)}") - return { - "success": False, - "message": f"登录验证失败: {str(e)}" - } + browser = require_browser() + ok, msg = await auth_manager.verify_login(browser) + meta = auth_manager.load_auth_meta() + return { + "success": ok, + "message": None if ok else msg, + "login_time": meta.get("login_time"), + "expire_time": None, # 单账号长期复用：不强制本地过期时间 + "last_verified_time": meta.get("last_verified_time"), + } @app.post("/api/logout") async def logout(): - """退出登录""" - auth_file = os.path.join(config.DATA_DIR, "auth_state.json") + auth_manager.clear_login_state() + return {"success": True} - try: - if os.path.exists(auth_file): - os.remove(auth_file) - return { - "success": True, - "message": "已退出登录" - } - except Exception as e: - raise HTTPException(status_code=500, detail=f"退出登录失败: {str(e)}") +# ========================= +# Scrape control +# ========================= +@app.post("/api/start") +async def start_scraping(request: StartRequest, background_tasks: BackgroundTasks): + if app_state.is_running: + raise HTTPException(status_code=400, detail="采集任务正在运行中") + + users = read_users_txt() + if not users: + raise HTTPException(status_code=400, detail="用户列表为空，请先导入 users.txt") + + # 启动前校验登录态（单账号） + browser = require_browser() + ok, msg = await auth_manager.verify_login(browser) + if not ok: + raise HTTPException(status_code=400, detail=f"未登录或登录失效：{msg}") + + app_state.is_running = True + app_state.scraper = XiaoHongShuScraper(app_state.event_queue, log_manager, browser) + + async def run_scraper(): + try: + await app_state.scraper.run(user_ids=users, mode=request.mode, max_threads=request.max_threads) + finally: + app_state.is_running = False + + background_tasks.add_task(run_scraper) + return {"success": True} + + +@app.post("/api/stop") +async def stop_scraping(): + if not app_state.is_running or not app_state.scraper: + return {"success": True, "message": "未在运行"} + + await app_state.scraper.stop() + app_state.is_running = False + return {"success": True} + + +# ========================= +# Data +# ========================= @app.get("/api/data/{user_id}") async def get_user_data(user_id: str): - """获取用户采集数据""" + # 读取该用户数据文件（若存在） file_path = os.path.join(config.DATA_DIR, f"{user_id}.json") - if not os.path.exists(file_path): - raise HTTPException(status_code=404, detail="用户数据不存在") + raise HTTPException(status_code=404, detail="数据不存在") + try: + with open(file_path, "r", encoding="utf-8") as f: + return json.load(f) + except Exception as e: + raise HTTPException(status_code=500, detail=str(e)) - with open(file_path, 'r', encoding='utf-8') as f: - data = json.load(f) - - return data +# ========================= +# SSE events +# ========================= @app.get("/api/events") -async def event_stream(): - """SSE 事件流""" +async def sse_events(): async def event_generator(): while True: - try: - # 等待事件，超时 1 秒发送心跳 - event = await asyncio.wait_for( - app_state.event_queue.get(), - timeout=1.0 - ) + event = await app_state.event_queue.get() + yield f"data: {json.dumps(event, ensure_ascii=False)}\n\n" - # 更新状态 - if event["type"] == "log": - # 日志不需要更新状态 - pass - elif event["type"] == "progress": - app_state.completed_users = event.get("completed", app_state.completed_users) - app_state.failed_users = event.get("failed", app_state.failed_users) - app_state.active_threads = event.get("active_threads", app_state.active_threads) - elif event["type"] == "user_status": - if event["user_id"] in app_state.users_status: - app_state.users_status[event["user_id"]]["status"] = event["status"] - app_state.users_status[event["user_id"]]["note_count"] = event.get("note_count", 0) - app_state.users_status[event["user_id"]]["thread_id"] = event.get("thread_id") - elif event["type"] == "thread_status": - app_state.thread_status[event["thread_id"]] = { - "status": event["status"], - "current_user": event.get("current_user") - } - # 更新活跃线程数 - app_state.active_threads = sum( - 1 for s in app_state.thread_status.values() if s.get("status") == "busy" - ) - - # 发送事件 - yield f"data: {json.dumps(event)}\n\n" - - except asyncio.TimeoutError: - # 发送心跳 - yield ": heartbeat\n\n" - - except Exception as e: - # 错误事件 - yield f"data: {json.dumps({'type': 'error', 'message': str(e)})}\n\n" - - return StreamingResponse( - event_generator(), - media_type="text/event-stream", - headers={ - "Cache-Control": "no-cache", - "Connection": "keep-alive", - "Access-Control-Allow-Origin": "*" - } - ) - - -# ========== Git 相关 API ========== - -@app.get("/api/git/status") -async def git_status(): - """获取 Git 状态""" - if not os.path.exists("/app/.git"): - return { - "enabled": False, - "message": "不是 Git 仓库" - } - - try: - # 获取当前分支 - result = subprocess.run( - ["git", "branch", "--show-current"], - capture_output=True, - text=True, - cwd="/app" - ) - current_branch = result.stdout.strip() - - # 获取最新提交 - result = subprocess.run( - ["git", "log", "-1", "--format=%h %s %cr"], - capture_output=True, - text=True, - cwd="/app" - ) - latest_commit = result.stdout.strip() - - # 检查是否有未提交的更改 - result = subprocess.run( - ["git", "status", "--porcelain"], - capture_output=True, - text=True, - cwd="/app" - ) - has_changes = len(result.stdout.strip()) > 0 - - return { - "enabled": True, - "branch": current_branch, - "latest_commit": latest_commit, - "has_changes": has_changes, - "auto_pull_enabled": config.ENABLE_GIT_PULL, - "auto_pull_interval": config.AUTO_PULL_INTERVAL - } - - except Exception as e: - return { - "enabled": False, - "error": str(e) - } - - -@app.post("/api/git/pull") -async def git_pull(): - """手动拉取最新代码""" - if not os.path.exists("/app/.git"): - raise HTTPException(status_code=400, detail="不是 Git 仓库") - - try: - # 获取当前分支 - result = subprocess.run( - ["git", "branch", "--show-current"], - capture_output=True, - text=True, - cwd="/app" - ) - current_branch = result.stdout.strip() - - # 拉取最新代码 - result = subprocess.run( - ["git", "fetch", "origin", current_branch], - capture_output=True, - text=True, - cwd="/app" - ) - - if result.returncode != 0: - raise HTTPException(status_code=500, detail=f"拉取失败: {result.stderr}") - - # 检查是否有更新 - result = subprocess.run( - ["git", "rev-list", "HEAD..origin/" + current_branch, "--count"], - capture_output=True, - text=True, - cwd="/app" - ) - commits_behind = int(result.stdout.strip()) - - if commits_behind == 0: - return { - "success": True, - "message": "已是最新代码", - "needs_restart": False - } - - # 重置到最新代码 - result = subprocess.run( - ["git", "reset", "--hard", "origin/" + current_branch], - capture_output=True, - text=True, - cwd="/app" - ) - - if result.returncode != 0: - raise HTTPException(status_code=500, detail=f"重置失败: {result.stderr}") - - # 检查 Python 文件是否有变化 - result = subprocess.run( - ["git", "diff", "--name-only", "HEAD@{1}", "HEAD"], - capture_output=True, - text=True, - cwd="/app" - ) - changed_files = result.stdout.strip().split('\n') - python_changed = any(f.endswith('.py') for f in changed_files if f) - - return { - "success": True, - "message": f"已拉取 {commits_behind} 个新提交", - "changed_files": changed_files, - "needs_restart": python_changed, - "restart_hint": "Python 代码有变化，建议重启容器以应用更新" - } - - except HTTPException: - raise - except Exception as e: - raise HTTPException(status_code=500, detail=f"拉取失败: {str(e)}") - - -@app.post("/api/git/restart") -async def restart_container(): - """ - 请求重启容器 - 注意：这需要在容器外配置重启策略 - """ - return { - "success": True, - "message": "重启请求已接收", - "methods": [ - "docker restart xiaohongshu-scraper", - "docker-compose restart" - ] - } - - -# ========== 启动事件 ========== -@app.on_event("startup") -async def startup_event(): - """启动时初始化""" - # 确保数据目录存在 - os.makedirs(config.DATA_DIR, exist_ok=True) - os.makedirs(config.SCREENSHOT_DIR, exist_ok=True) - - # 初始化事件队列 - app_state.event_queue = asyncio.Queue() - - # 初始化浏览器 - try: - from playwright.async_api import async_playwright - app_state.playwright = await async_playwright().start() - app_state.browser = await app_state.playwright.chromium.launch( - headless=config.HEADLESS, - args=config.BROWSER_ARGS - ) - print("[INFO] 浏览器初始化完成") - except Exception as e: - print(f"[ERROR] 浏览器初始化失败: {str(e)}") - - print("=" * 60) - print("小红书数据采集器 Web 服务启动") - print(f"访问地址: http://{config.WEB_HOST}:{config.WEB_PORT}") - print("=" * 60) - - -@app.on_event("shutdown") -async def shutdown_event(): - """关闭时清理""" - if app_state.scraper: - app_state.scraper.stop() - - if app_state.scraper_task: - app_state.scraper_task.cancel() - - # 关闭浏览器 - if app_state.browser: - await app_state.browser.close() - if app_state.playwright: - await app_state.playwright.stop() - print("[INFO] 浏览器已关闭") - - print("Web 服务已关闭") - - -if __name__ == "__main__": - import uvicorn - uvicorn.run( - "app.web:app", - host=config.WEB_HOST, - port=config.WEB_PORT, - reload=True - ) + return StreamingResponse(event_generator(), media_type="text/event-stream") diff --git a/小红书数据采集器使用指南.md b/小红书数据采集器使用指南.md new file mode 100644 index 0000000..25f1166 --- /dev/null +++ b/小红书数据采集器使用指南.md @@ -0,0 +1,353 @@ +# 小红书数据采集器使用指南 + +## 系统概述 + +这是一个基于 FastAPI + Playwright 的小红书数据采集器，具有 Web 可视化监控界面，支持多线程采集、实时进度展示和持久化日志记录。 + +## 系统特点 + +- 🌐 **Web 可视化界面**：实时监控采集进度和状态 +- 🔐 **真实扫码登录**：打开小红书官方登录页面，扫码登录后验证状态 +- 👥 **账号管理**：支持添加、删除用户账号 +- 📊 **实时日志**：分类显示操作日志，支持过滤和搜索 +- 🔄 **多线程采集**：支持单线程/多线程模式切换 +- 🐳 **Docker 部署**：容器化部署，支持自动更新 +- 📁 **数据持久化**：所有数据本地存储，重启不丢失 + +## 快速开始 + +### 1. Docker 部署（推荐） + +```bash +# 克隆或下载项目 +cd /path/to/demo + +# 构建并启动容器（使用18085端口） +docker-compose up -d --build + +# 查看容器状态 +docker-compose ps + +# 查看日志 +docker-compose logs -f +``` + +### 2. 本地运行 + +```bash +# 安装依赖 +pip install -r requirements.txt + +# 安装浏览器 +playwright install chromium + +# 创建数据目录 +mkdir -p data data/screenshots + +# 运行程序 +python main.py +``` + +### 3. 访问系统 + +打开浏览器访问：`http://localhost:18085` + +## 系统界面说明 + +### 1. 账号管理区域 + +![账号管理](./images/account-management.png) + +- **添加账号**：点击添加账号按钮，输入小红书用户 ID +- **刷新账号**：重新加载账号列表和状态 +- **登录状态**： + - 🔴 红色圆点：未登录 + - 🟢 绿色圆点：已登录 + - 显示剩余登录时间（小时:分钟） + +### 2. 控制面板 + +![控制面板](./images/control-panel.png) + +- **启动采集**：开始数据采集任务 +- **停止**：停止正在进行的采集任务 +- **清空日志**：清空所有历史日志 +- **导出数据**：下载采集到的数据文件 +- **模式选择**： + - 单线程：逐个处理账号 + - 多线程：同时处理多个账号（可配置线程数） +- **线程数**：多线程模式下的并发线程数（1-10） + +### 3. 进度监控 + +![进度监控](./images/progress.png) + +- **整体进度**：显示采集进度条和百分比 +- **活跃线程**：多线程模式下显示正在运行的线程数 +- **用户列表**：显示每个用户的采集状态 + - ✅ 已完成：显示采集的笔记数量 + - 🔄 进行中：显示当前处理的笔记数量 + - ⏳ 待处理：尚未开始处理的账号 + - ❌ 失败：采集失败的账号 + +### 4. 日志面板 + +![日志面板](./images/logs.png) + +- **日志级别过滤**：可选择查看 ALL/INFO/SUCCESS/WARNING/ERROR/DEBUG 级别的日志 +- **搜索功能**：搜索特定关键词的日志 +- **自动滚动**：开启后自动滚动到最新日志 +- **时间戳**：显示每条日志的具体时间 +- **线程标识**：多线程模式下显示日志来源线程 + +### 5. 错误截图 + +![错误截图](./images/screenshots.png) + +如果采集过程中遇到错误（如风控检测），系统会自动截图并保存： +- 截图保存在：`data/screenshots/` 目录 +- 在界面中显示错误截图缩略图 +- 可点击查看大图或下载所有截图 + +## 操作流程 + +### 1. 首次使用 + +```bash +# 1. 启动系统 +docker-compose up -d --build + +# 2. 访问 Web 界面 +# 打开 http://localhost:18080 + +# 3. 添加要采集的用户 ID +# 点击"添加账号"按钮，输入小红书用户 ID +# 格式：长字符串（如 5ef74296000000000100003a） +# 可从浏览器地址栏获取：https://www.xiaohongshu.com/user/profile/用户ID + +# 4. 真实扫码登录 +# 点击"打开登录页"按钮，会在新窗口打开小红书官方登录页面 +# 在打开的页面中使用小红书App扫描二维码完成登录 +# 登录成功后，返回本页面点击"验证登录"按钮 +# 登录状态有效期：24小时 +``` + +### 2. 配置采集参数 + +1. 选择采集模式： + - **单线程**：稳定但速度慢，适合少量账号 + - **多线程**：速度快，适合大量账号（推荐 3-5 个线程） + +2. 设置线程数（多线程模式）： + - 建议：3-5 个线程 + - 过多线程可能触发风控 + +### 3. 开始采集 + +1. 确保已添加至少一个用户账号 +2. 确保已登录（显示绿色状态） +3. 点击"启动采集" +4. 系统会： + - 检查登录状态 + - 读取用户列表 + - 开始采集并实时显示进度 + +### 4. 监控采集 + +采集过程中可以： +- 查看实时进度和状态 +- 监控日志输出 +- 观察各线程工作状态 +- 随时停止采集（点击"停止"） + +### 5. 数据导出 + +采集完成后可以： +- 查看每个用户的数据文件：`data/{用户ID}.json` +- 下载所有日志：点击"导出数据" +- 下载错误截图（如有） + +## 用户 ID 获取方法 + +### 方法1：浏览器地址栏 + +1. 打开小红书网页版 +2. 访问目标用户主页：https://www.xiaohongshu.com/user/profile/用户名 +3. 从浏览器地址栏复制用户 ID： + ``` + https://www.xiaohongshu.com/user/profile/5ef74296000000000100003a + ``` + 复制 `5ef74296000000000100003a` 部分 + +### 方法2：开发者工具 + +1. 在用户页面按 F12 打开开发者工具 +2. 切换到 Network（网络）标签 +3. 刷新页面 +4. 查找包含 "user" 或 "profile" 的请求 +5. 从请求 URL 中提取用户 ID + +## 数据文件说明 + +### 输入文件 + +``` +data/users.txt # 用户 ID 列表，每行一个 +``` + +### 输出文件 + +``` +data/ +├── auth_state.json # 登录状态缓存 +├── logs.json # 操作日志历史 +├── users.txt # 用户列表 +├── 5ef74296000000000100003a.json # 用户采集数据 +├── 5b6c1f9e0000000001000388.json # 用户采集数据 +└── screenshots/ # 错误截图目录 + ├── error_xxx1.png + └── error_xxx2.png +``` + +### 数据格式示例 + +```json +// data/{用户ID}.json +{ + "user_id": "5ef74296000000000100003a", + "last_updated": "2025-01-12T10:30:00", + "notes": [ + { + "title": "笔记标题", + "publish_time": "2025-01-10 14:30", + "location": "上海", + "comment_count": 128, + "like_count": 5234, + "collect_count": 892, + "note_url": "https://www.xiaohongshu.com/explore/xxx" + } + ] +} +``` + +## 常见问题 + +### Q1: 为什么提示"未登录"？ + +A: 登录状态 24 小时后会自动过期，请： +1. 点击"模拟登录"重新登录 +2. 检查 `data/auth_state.json` 文件是否存在 + +### Q2: 采集速度很慢怎么办？ + +A: 可以尝试： +1. 使用多线程模式（建议 3-5 线程） +2. 检查网络连接 +3. 避免在高峰期采集 + +### Q3: 如何避免被封禁？ + +A: 系统已内置防护措施： +1. 拟人化延迟（随机 2-60 秒） +2. 平滑滚动加载 +3. 多线程错开启动 +4. 风控检测自动截图 +5. 建议不要设置过多线程 + +### Q4: 数据采集不全怎么办？ + +A: 可能原因： +1. 网络不稳定导致加载失败 +2. 笔记数量过多，滚动加载不完整 +3. 小红书页面结构变化 + +### Q5: Docker 容器无法启动？ + +A: 检查： +1. Docker 是否运行 +2. 端口 18080 是否被占用 +3. Docker Compose 配置是否正确 + +## 配置说明 + +### 环境变量 + +```env +# Web 服务器配置 +PORT=18080 +HOST=0.0.0.0 + +# 浏览器配置 +HEADLESS=false # 是否无头模式 +SCREENSHOT_ON_ERROR=true # 错误时截图 + +# 多线程配置 +MAX_CONCURRENT_THREADS=5 # 最大并发线程数 +THREAD_MIN_DELAY=2 # 线程间最小间隔（秒） +``` + +### 配置文件位置 + +```python +# app/config.py +DATA_DIR = "data" # 数据目录 +SCREENSHOT_DIR = "data/screenshots" # 截图目录 +MAX_CONCURRENT_THREADS = 5 # 最大并发线程数 +``` + +## 更新和维护 + +### 1. 自动更新（Docker） + +系统支持 Git 自动更新： +1. 代码提交到远程仓库 +2. 系统自动拉取最新代码 +3. 检测到 Python 代码变化时建议重启容器 + +### 2. 手动更新 + +```bash +# 拉取最新代码 +git pull origin master + +# 重新构建容器 +docker-compose build + +# 重启服务 +docker-compose up -d +``` + +### 3. 数据备份 + +重要数据请定期备份： +```bash +# 备份数据目录 +tar -czf backup-$(date +%Y%m%d).tar.gz data/ + +# 备份特定用户数据 +cp data/*.json backup-$(date +%Y%m%d)/ +``` + +## 技术支持 + +如果遇到问题： +1. 查看 Web 界面的日志面板 +2. 检查 Docker 容器日志：`docker-compose logs` +3. 查看 `data/logs.json` 文件 +4. 确保网络连接正常 + +## 版本更新日志 + +### v1.0.0 (2025-01-12) +- ✅ 实现基础采集功能 +- ✅ Web 可视化界面 +- ✅ 多线程支持 +- ✅ 实时日志系统 +- ✅ 账号管理功能 +- ✅ 模拟登录系统 +- ✅ Docker 部署支持 +- ✅ 数据导出功能 + +--- + +**注意**：本工具仅供学习和研究使用，请遵守小红书的使用条款和相关法律法规。 \ No newline at end of file