使用 Playwright 自动化浏览器抓取多个平台（微博、小红书等）热搜话题的 Python 工具。
- 多平台支持: 微博、小红书、百度、B站、GitHub、Hacker News 等
- 统一数据模型: 所有平台数据格式统一，支持扩展字段
- 多种输出方式: 控制台、JSON、CSV、SQLite 数据库
- Web API 服务: FastAPI 提供的 RESTful API，支持查询、搜索、统计
- 插件化架构: 新增平台只需添加一个文件
- 自动登录管理: 支持 Cookie 持久化和自动重新登录
- 并发抓取: 同时抓取多个平台提高效率
- 数据缓存: 文件缓存系统，API 响应缓存 5 分钟
# 安装 Python 依赖
pip install -r requirements.txt
# 安装 Chromium 浏览器（只需执行一次）
python3 -m playwright install chromium# 列出所有可用平台
python topic_hub.py --list
# 抓取微博热搜
python topic_hub.py weibo
# 抓取小红书热搜（需登录）
python topic_hub.py xiaohongshu
# 同时抓取多个平台
python topic_hub.py weibo xiaohongshu
# 只显示前10条
python topic_hub.py weibo -n 10
# 保存为 JSON 文件
python topic_hub.py weibo -o hot_topics.json
# 保存为 CSV 文件
python topic_hub.py weibo -o hot_topics.csv
# 保存到 SQLite 数据库
python topic_hub.py weibo --db
# 显示浏览器窗口（调试用）
python topic_hub.py weibo --visible
# 强制重新登录（用于需要登录的平台）
python topic_hub.py xiaohongshu --reloginTopic Hub 提供 FastAPI 驱动的 Web 服务，支持通过 RESTful API 访问热搜数据。
# 使用启动脚本（推荐）
./start.sh
# 或使用 uvicorn 直接启动
uvicorn web.backend.app.main:app --host 0.0.0.0 --port 8000 --reload| 端点 | 方法 | 说明 |
|---|---|---|
/ |
GET | API 信息 |
/health |
GET | 健康检查 |
/api/v1/topics |
GET | 获取话题列表（支持分页、筛选） |
/api/v1/topics/{id} |
GET | 获取话题详情 |
/api/v1/topics/search/{query} |
GET | 搜索话题 |
/api/v1/platforms |
GET | 获取平台列表 |
/api/v1/stats |
GET | 获取统计数据 |
启动服务后访问：
- Swagger UI: http://localhost:8000/docs
- ReDoc: http://localhost:8000/redoc
# 获取所有话题
curl http://localhost:8000/api/v1/topics/
# 获取特定平台的话题
curl http://localhost:8000/api/v1/topics/?platform=weibo
# 搜索话题
curl http://localhost:8000/api/v1/topics/search/科技
# 获取统计数据
curl http://localhost:8000/api/v1/stats项目包含基于 Next.js 14 的前端界面，使用 React Query 进行数据获取，Tailwind CSS 进行样式设计。
前端遵循 DESIGN.md 中定义的设计系统：
- 美学方向: 编辑数据感 (Editorial Data-Driven)
- 配色: 深色模式优先，主色调 Indigo (#4F46E5)
- 字体: Plus Jakarta Sans (标题) + Inter/Noto Sans SC (正文) + JetBrains Mono (数字)
- 热力条: 垂直设计，高/中/低三档颜色区分
- 框架: Next.js 14 (App Router)
- 样式: Tailwind CSS
- 数据获取: TanStack Query (React Query)
- 组件: 自定义组件库
cd web/frontend
# 安装依赖
npm install
# 开发模式
npm run dev
# 构建生产版本
npm run build# 启动服务
docker-compose up -d
# 查看日志
docker-compose logs -f
# 停止服务
docker-compose down
# 同时启动 CLI 定时抓取服务
docker-compose --profile cli up -d# 构建镜像
docker build -t topic-hub .
# 运行容器
docker run -d \
-p 8000:8000 \
-v $(pwd)/data:/app/data \
-e DATABASE_URL=sqlite:///data/topic_hub.db \
--name topic-hub \
topic-hub| 变量 | 默认值 | 说明 |
|---|---|---|
DATABASE_URL |
sqlite:///data/topic_hub.db | 主数据库路径 |
QUEUE_URL |
sqlite:///data/queue/task_queue.db | 任务队列数据库 |
CACHE_DIR |
data/cache | 缓存目录 |
SCHEDULER_ENABLED |
true | 启用定时任务 |
FETCH_INTERVAL_MINUTES |
5 | 抓取间隔（分钟） |
DATA_RETENTION_DAYS |
30 | 数据保留天数 |
| 参数 | 说明 | 示例 |
|---|---|---|
platforms |
要抓取的平台（可多个） | weibo, xiaohongshu |
-n, --number |
限制显示数量 | -n 20 |
-o, --output |
输出文件路径 | -o data.json |
--db |
保存到 SQLite 数据库 | --db |
--list |
列出所有可用平台 | --list |
--visible |
显示浏览器窗口 | --visible |
--relogin |
强制重新登录 | --relogin |
import asyncio
from platforms.discovery import get_spider_class
async def main():
# 获取微博爬虫类
WeiboSpider = get_spider_class("weibo")
spider = WeiboSpider()
# 抓取数据
topics = await spider.fetch()
for topic in topics[:10]:
print(f"{topic.rank}. {topic.title} - {topic.hot_value}")
asyncio.run(main())============================================================
🔥 微博 (50 条话题)
============================================================
🥇 话题一标题 - 123万热度
🥈 话题二标题 - 98万热度
🥉 话题三标题 - 87万热度
4. 话题四标题 - 76万热度
5. 话题五标题 - 65万热度
...
============================================================
{
"fetch_time": "2026-03-21T14:30:25",
"count": 50,
"topics": [
{
"id": "weibo_1",
"platform": "weibo",
"rank": 1,
"title": "话题标题",
"hot_value": "123万热度",
"url": "https://s.weibo.com/weibo?q=...",
"category": null,
"created_at": "2026-03-21T14:30:25",
"extra": {}
}
]
}.
├── topic_hub.py # 主入口
├── cli/ # CLI 模块
│ ├── commands/ # 子命令实现
│ ├── formatters.py # 输出格式化
│ └── main.py # CLI 入口
├── core/ # 核心框架
│ ├── base.py # BaseSpider 抽象基类
│ ├── models.py # HotTopic 数据模型 (SQLAlchemy)
│ ├── database.py # 数据库配置
│ ├── storage.py # StorageManager 存储管理
│ ├── auth.py # AuthManager 认证管理
│ ├── cache.py # FileCache 文件缓存
│ ├── queue.py # SQLiteQueue 任务队列
│ ├── browser_pool.py # 浏览器连接池
│ └── context.py # PlaywrightContext 浏览器上下文
├── platforms/ # 平台实现
│ ├── weibo.py # 微博爬虫
│ ├── xiaohongshu.py # 小红书爬虫
│ ├── baidu.py # 百度热搜爬虫
│ ├── bilibili.py # B站热门爬虫
│ ├── github.py # GitHub Trending 爬虫
│ ├── hackernews.py # Hacker News 爬虫
│ └── discovery.py # 平台自动发现
├── web/ # Web 服务
│ ├── backend/ # FastAPI 后端
│ │ └── app/
│ │ ├── main.py # FastAPI 应用入口
│ │ ├── config.py # 配置管理
│ │ ├── dependencies.py # 依赖注入
│ │ ├── scheduler.py # 定时任务
│ │ ├── schemas.py # Pydantic 模型
│ │ └── api/v1/ # API 路由
│ └── frontend/ # Next.js 前端
│ ├── src/ # 源代码
│ │ ├── app/ # 页面组件
│ │ ├── components/ # UI 组件
│ │ └── lib/ # 工具函数
│ └── public/ # 静态资源
├── data/ # 数据目录
│ ├── topic_hub.db # SQLite 数据库
│ ├── queue/ # 任务队列数据
│ ├── cache/ # 缓存文件
│ └── cookies/ # Cookie 存储
├── tests/ # 测试目录
├── requirements.txt # 依赖列表
├── Dockerfile # Docker 镜像
├── docker-compose.yml # Docker Compose 配置
├── start.sh # 启动脚本
├── README.md # 使用说明
├── CLAUDE.md # 开发指南
├── DESIGN.md # 设计系统
└── TODOS.md # 待办事项
要添加新的平台支持，只需在 platforms/ 目录下创建一个新文件：
# platforms/douyin.py
"""Douyin hot topics spider"""
from typing import List, Dict, Any
from playwright.async_api import Page
from core.base import BaseSpider
from core.models import HotTopic
class DouyinSpider(BaseSpider):
name = "douyin"
display_name = "抖音"
requires_login = False
default_url = "https://www.douyin.com/..."
async def extract(self, page: Page) -> List[Dict[str, Any]]:
# 提取原始数据
pass
def parse(self, raw_items: List[Dict[str, Any]]) -> List[HotTopic]:
# 解析为 HotTopic
pass新平台会自动被发现，无需修改核心代码。
- 网络要求：需要能够访问目标平台网站
- 反爬机制：平台可能有反爬虫机制，如遇到失败可稍后重试
- 频率限制：建议每次调用间隔至少 30 秒
- 登录要求：小红书等平台需要登录，首次使用会弹出浏览器窗口
# 运行所有单元测试
python -m pytest tests/ -v
# 运行单元测试（跳过集成测试）
python -m pytest tests/ -v -m "not integration"
# 生成代码覆盖率报告
python -m pytest tests/ -m "not integration" --cov=core --cov=platforms本工具仅供学习研究使用，请勿用于商业用途或高频抓取。请遵守各平台的相关使用条款和 robots.txt 规定。