Web Crawler

工程级多平台内容爬取框架，基于 Python 异步架构。灵感来源于 Java SuperWorker 项目，结合了更优的设计方案和最佳实践。

特性

• 多平台支持：微信公众号、微博、今日头条、新浪、东方财富、RSS、通用 HTML
• 统一 Playwright：摒弃 Selenium 双工具链，全面使用 Playwright async API
• 模板方法 + 装饰器注册：新增平台只需 4 个选择器
• 可插拔架构：存储（SQLite/MySQL）、队列（内存/Redis）均可替换
• 反检测中间件：UA 轮换、stealth 注入、验证码检测、自适应限速
• 异步优先：asyncio + Playwright async API + aiosqlite
• 零外设依赖：SQLite + 内存队列开箱即用

快速开始

# 安装
pip install -e ".[dev]"

# 安装浏览器
playwright install chromium

# 查看支持的平台
python -m crawler platforms

# 爬取单个 URL
python -m crawler crawl "https://mp.weixin.qq.com/s/xxxxx"

# 从文件批量爬取
python -m crawler crawl --file urls.txt --concurrent 5

# 指定平台
python -m crawler crawl --platform rss "https://example.com/feed.xml"

# 导出为 JSON
python -m crawler crawl "https://weibo.com/xxx" --output result.json

# 启动 Web 管理界面
python -m crawler serve --port 8000

项目结构

crawler/
├── core/           # 核心抽象 (BaseCrawler, PlatformRegistry, CrawlEngine)
├── platforms/      # 7个平台实现 (weixin/weibo/toutiao/sina/eastmoney/rss/generic)
├── browser/        # Playwright 封装 + 反检测 + 页面操作
├── parser/         # HTML解析/日期解析/内容清洗/RSS解析/Markdown转换
├── storage/        # 存储抽象层 (SQLiteBackend, 可替换)
├── queue/          # 队列抽象层 (MemoryQueue, 可替换)
├── middleware/      # 中间件 (反检测/重试/限速)
├── config/         # Pydantic 配置 + YAML 加载
├── cli/            # Click 命令行
└── api/            # FastAPI REST API

架构设计

URL → Queue → Middleware → Browser/HTTP → Platform Crawler → Parse → Clean → Store
                                              ↑
                              PlatformRegistry.resolve_by_url() 自动识别平台

新增平台只需三步

# 1. 创建 crawler/platforms/newplatform.py
from crawler.core.base import BaseCrawler
from crawler.core.registry import PlatformRegistry

@PlatformRegistry.register
class NewPlatformCrawler(BaseCrawler):
    @property
    def name(self): return "newplatform"
    @property
    def url_patterns(self): return [r"newplatform\.com"]
    @property
    def title_selector(self): return "h1"
    @property
    def date_selector(self): return ".date"
    @property
    def source_selector(self): return ".author"
    @property
    def content_selector(self): return ".content"

# 2. 在 platforms/__init__.py 中添加 import
# 3. 完成！

可插拔接口

组件	接口	默认实现	生产替代
存储	StorageBackend	SQLite + SQLAlchemy	MySQL/PostgreSQL
队列	QueueBackend	asyncio.Queue	Redis/RabbitMQ
浏览器	BrowserManager	Playwright async	-
中间件	Middleware	反检测+重试+限速	自定义

测试

python -m pytest tests/ -v
# 107 tests passed

技术栈

用途	技术
浏览器自动化	Playwright (async)
HTML 解析	BeautifulSoup4 + lxml
HTTP 客户端	httpx (async)
数据库	SQLAlchemy 2.0 + aiosqlite
配置	Pydantic Settings + YAML
CLI	Click
Web API	FastAPI + Uvicorn
RSS 解析	feedparser
日志	structlog