Universal Extractor

专门对付那些别人抓不到的网页。 Canvas 绑图、字体加密、滑动验证码——Firecrawl 和 Jina Reader 返回空壳的地方，这里才刚开始工作。

⚠️ 使用前必读：本项目包含验证码破解、字体解密、MITM 抓包等高风险功能。不当使用可能导致违法。请先阅读 🛡️ 法律风险警告。

别人做不到的，我们做了

大多数工具在 DOM 里直接有文字时才有效。真实世界远不止如此。以下三个场景，目前只有 Universal Extractor 支持：

独有能力	场景	原理
Canvas Hook	WPS / 飞书 / 腾讯文档——DOM 里只有 `<canvas>`，文字在渲染层	注入 JS 拦截 `fillText` / `strokeText`，捕获 Canvas 上每一段文字
字体解密	起点 / 天眼查 / 大众点评——HTML 里是 `` 乱码	fontTools 解析 cmap 表 + ddddocr 字模 OCR，PUA 码点 → 真实汉字
验证码本地求解	GeeTest / DataDome 滑动验证码 + 文字顺序验证码	OpenCV Canny + 模板匹配 + 贝塞尔轨迹，三通道加权投票，本地免费 90%+ 识别率

别人能做的，我们更快

普通网页（博客、文档、SPA）自动走最优路径：

场景	策略	速度
静态 HTML	Jina Reader 快扫	~500ms
SPA（React / Vue）	StealthyFetcher 浏览器 DOM	~2s
GBK 老站	curl_cffi TLS 指纹	~300ms
Cloudflare 保护	自动 Turnstile 求解	~5s

同一域名第二次抓取直接命中最优层——越用越快。

架构

Pipeline: Search → Verify → Extract (9-layer fallback) → Validate → Return

Layer ① Jina Reader        ← 静态页最快（SPA / Canvas 自动跳过）
Layer ② curl_cffi HTTP     ← GBK 编码站优先
Layer ③ Browser DOM        ← SPA 必用（预收集 Canvas + CDP 数据复用）
Layer ④ Canvas Hook        ← 拦截 Canvas fillText / strokeText
Layer ⑤ CDP Heap Scan      ← 扫 __NEXT_DATA__ / __NUXT__ 隐藏数据
Layer ⑥ Screenshot OCR     ← Canvas 应用主力（ddddocr + Tesseract）
Layer ⑦ Vision LLM         ← 终极兜底（GPT-4o / Claude / Qwen-VL）
Layer ⑧ Captcha Slider     ← 检测到滑动验证码时自动求解
Layer ⑨ Captcha Font       ← 检测到自定义字体时自动解密

自适应调度：每层不再盲目串行。URL 分类后自动算分排序——比如 kdocs.cn 这样的 Canvas 应用，Jina / curl 自动跳过，OCR 排到最前。

快速开始

安装

# 基础安装
pip install universal-extractor

# 验证码本地求解（推荐）
pip install universal-extractor[captcha]

# 字体解密（对付起点 / 天眼查等）
pip install universal-extractor[captcha-font]

# 文件提取（PDF / Office / EPUB）
pip install universal-extractor[files]

# REST API 服务
pip install universal-extractor[api]

# 全部可选依赖
pip install universal-extractor[all]

🐳 Docker（推荐部署方式）

# API 服务一键启动
docker compose up -d

# 验证
curl http://localhost:11235/health
open http://localhost:11235/docs      # Swagger 交互式文档

# CLI 单次提取
docker compose run --rm ue extract https://example.com

构建本地镜像：

docker build -t universal-extractor .
docker run --rm universal-extractor extract https://example.com

配置 Vision 后端（可选）

至少设置一个环境变量以启用 Layer ⑥⑦ 的云端 OCR：

变量	后端	默认模型	模型覆盖变量
`OPENAI_API_KEY`	GPT-4o-mini	`gpt-4o-mini`	`VISION_OPENAI_MODEL`
`ANTHROPIC_API_KEY`	Claude Sonnet	`claude-sonnet-4-6`	`VISION_ANTHROPIC_MODEL`
`DASHSCOPE_API_KEY`	Qwen-VL-Max	`qwen-vl-max`	`VISION_QWEN_MODEL`
`DEEPSEEK_API_KEY`	DeepSeek	`deepseek-chat`	`VISION_DEEPSEEK_MODEL`

不配 Vision 后端也能用——前 5 层不需要任何 API Key。

最简用法

from universal_extractor import UniversalExtractor

ue = UniversalExtractor(headless=True)
text = ue.extract("https://example.com/article")
print(text)

功能全景

🌐 网页提取

from universal_extractor import Pipeline, PipelineConfig

pipeline = Pipeline(PipelineConfig(
    headless=True,
    min_completeness=0.5,
    search_backends=["duckduckgo", "brave"],
    enable_cross_validation=True,
))
result = pipeline.run("三体 小说 全文")
print(result.text)          # 提取的正文
print(result.score)         # 质量评分 (0.0 ~ 1.0)
print(result.winning_stage) # 命中的阶段名称

📄 文件提取

from universal_extractor import UniversalExtractor

ue = UniversalExtractor()

# PDF / Word / Excel / PPT / EPUB / 图片 / 音频 → Markdown
text = ue.extract_file("report.pdf")
text = ue.extract_auto("presentation.pptx")  # 自动判断文件 vs URL

🔍 搜索 + 提取

from universal_extractor import search_urls, search_compare

# 多引擎搜索（DuckDuckGo / Brave / Exa / SearXNG）
urls = search_urls("机器学习入门", max_results=10)

# 搜索 + 对比交叉验证
results = search_compare("新能源汽车政策", backends=["duckduckgo", "brave"])

🕷️ 深度爬取

from universal_extractor import Crawler, CrawlConfig

c = Crawler(headless=True)

# BFS 广度优先（文档站）
c.crawl("https://docs.example.com/", "docs.md", config=CrawlConfig(
    strategy="bfs", max_depth=2, max_pages=200,
))

# 链接评分模式（按查询词相关性排序）
c.crawl("https://example.com/", "out.md", config=CrawlConfig(
    strategy="best_first",
    score_links=True,
    score_query="machine learning",
))

# SQLite 持久化队列（进程重启不丢进度）
c.crawl_with_queue("https://example.com/", "out.md")

# 异步流式
async for label, url, data in c.crawl_stream(start_url):
    print(f"[{label}] → {len(data['text'])} chars")

📊 结构化提取

from universal_extractor import (
    StructuredExtractor,
    SchemaExtractionStrategy,
    LLMExtractionStrategy,
    extract_tables,
)

# Schema 定义 → 结构化输出
extractor = StructuredExtractor()
result = extractor.extract(
    url="https://books.toscrape.com/",
    strategy=SchemaExtractionStrategy(schema={"title": "str", "price": "float"}),
)

# LLM 智能提取
result = extractor.extract(
    url="https://example.com/article",
    strategy=LLMExtractionStrategy(
        instruction="提取文章中提到的所有人名和职位",
        api_key="sk-...",
    ),
)

# 表格提取（支持分块 LLM 处理超大表格）
tables = extract_tables("https://example.com/data-table")
for t in tables:
    print(t.headers, len(t.rows))

🧩 示例驱动提取

from universal_extractor import extract_by_example

# 给几个例子，自动找出规律批量爬取
results = extract_by_example(
    url="https://books.toscrape.com/",
    examples=["A Light in the Attic", "£51.77"],
)
# → [{"title": "...", "price": "..."}, ...]

📱 移动端提取

from universal_extractor import MobileExtractor

m = MobileExtractor()

# 当前屏幕截图 + OCR
text = m.extract_screen()

# 打开 App → 自动滚动 → 逐屏 OCR → 去重合并
text = m.extract_app(
    package="com.example.news",
    scroll_count=5,
    wait_after_scroll=1.5,
)

# 自定义交互脚本
m.extract_interactive([
    {"action": "start_app", "package": "com.example.app"},
    {"action": "wait", "seconds": 2},
    {"action": "tap", "x": 500, "y": 800},
    {"action": "screenshot"},
])

前置条件：Android 设备 + USB 调试 + ADB。不需要 root。

💬 微信小程序 & 公众号

微信公众号（标准 H5，直接提取）：

from universal_extractor import WechatMPExtractor

wx = WechatMPExtractor()
article = wx.extract_url("https://mp.weixin.qq.com/s/abc123")
print(article.title, article.author, article.content)

MITM 代理捕获（拦截小程序 API 请求）：

pip install universal-extractor[mitm]
ue mitm --ca-export              # 导出 CA 证书
ue mitm --setup-guide            # 打印代理配置指南
ue mitm --timeout 60             # 捕获微信相关域名

wxapkg 解包（从 PC 微信缓存提取小程序包）：

ue wxapkg list                         # 列出缓存的小程序包
ue wxapkg analyze <path>               # 分析 app.json 元数据（无需解包工具）
ue wxapkg unpack <path> -o output/     # 完整解包

🧠 验证码求解

from universal_extractor.captcha import SliderCaptchaSolver, FontCaptchaSolver

# 滑动验证码（OpenCV Canny 边缘检测 + 模板匹配）
slider = SliderCaptchaSolver(captcha_type="geetest")
result = slider.solve(
    background_bytes=open("bg.png", "rb").read(),
    slider_bytes=open("slider.png", "rb").read(),
)
if result.solved:
    print(f"距离: {result.distance}px, 轨迹: {len(result.track)} 点")

# 字体解密（fontTools cmap 表解析 + ddddocr 字模 OCR）
font = FontCaptchaSolver()
mapping = font.crack_from_page(html, page_url="https://...")
decoded = mapping.decode(pua_encrypted_text)  # &#xe001; → "第"

# 文字顺序验证码
result = solver.solve_text_order(
    image_bytes=open("captcha.png", "rb").read(),
    prompt_text="请依次点击：请 点 击 文 字",
)
# result.clicks → [(120, 45), (310, 80), ...] 按正确顺序排列

方案	技术栈	特点
滑动验证码	OpenCV Canny + TM_CCOEFF_NORMED 模板匹配 + 4 阶段贝塞尔轨迹	本地免费
字体解密	fontTools cmap 表 + ddddocr 字模 OCR + PUA 反向映射	支持起点 / 天眼查等
文字顺序验证码	ddddocr ONNX 目标检测 + 坐标排序	本地免费
云端降级	CapSolver API	保底备用

🔊 认证 & 登录态

from universal_extractor import AuthManager, Credential

am = AuthManager()
am.save_credential("example.com", Credential(
    username="user@example.com",
    password="s3cret",
))

# 自动登录 + 保持会话 + 过期自动续期
if not am.session_valid("example.com"):
    am.refresh_session("example.com")

🔌 中间件 & 信号

from universal_extractor import Pipeline, signal

# 注册自定义中间件
Pipeline.register_middleware(MyRetryMiddleware())
Pipeline.register_pipeline(MyExportPipeline())

# 事件监听
signal.on("extraction_completed", lambda result: print(result.score))
signal.on("captcha_detected", lambda url, t: send_alert(f"Captcha at {url}"))

# 开发模式（首次缓存 HTTP 响应，后续回放）
pipeline = Pipeline(PipelineConfig(dev_mode=True))

🌍 内容过滤

from universal_extractor import (
    PruningContentFilter,
    BM25ContentFilter,
    LLMContentFilter,
    filter_chain,
)

# 三层过滤链：Pruning（去噪）→ BM25（相关性排序）→ LLM（精筛）
filtered = filter_chain(raw_text, query="机器学习", threshold=0.3)

CLI

# 网页提取
ue extract https://example.com/article
ue extract https://example.com --format json

# 文件提取
ue file report.pdf
ue file presentation.pptx --format md -o output.md

# 搜索 + 提取
ue search "机器学习入门"
ue run "三体 小说"

# 深度爬取（BFS / DFS / BestFirst）
ue crawl https://docs.example.com/ -o docs.md --strategy bfs --max-depth 2
ue crawl https://novel.com/catalog -o novel.md --score-links --score-query "主角"

# 批量处理
ue batch urls.txt --format json -o results.json

# 启动 REST API
ue serve                          # http://127.0.0.1:11235
ue serve --port 8080 --reload

# 移动端提取
ue mobile                         # 当前屏幕截图 + OCR
ue mobile --package com.example   # 打开 App + 滚动提取

# MITM 代理捕获
ue mitm --pattern "api.example.com" --timeout 30
ue mitm --ca-export               # 导出 CA 证书
ue mitm --setup-guide             # 打印代理配置指南

# 微信小程序
ue wechat "https://mp.weixin.qq.com/s/abc123"
ue wxapkg list
ue wxapkg unpack <path> -o output/

# 通用标志
ue extract URL --dev-mode         # 开发模式（缓存 HTTP 响应）
ue extract URL --format json      # JSON 输出
ue extract URL --verbose          # 详细日志

REST API Server

pip install universal-extractor[api]
ue serve

POST /v1/extract        # 提取 URL 或文件
POST /v1/crawl          # 深度爬取
GET  /v1/task/{id}      # 查询任务状态
GET  /v1/health         # 健康检查

浏览器打开 http://127.0.0.1:11235/docs 查看交互式 Swagger 文档。

MCP Server

让 Claude / Cursor 直接调用提取能力：

{
  "mcpServers": {
    "universal-extractor": {
      "command": "python",
      "args": ["-m", "UniversalExtractor.mcp_server"]
    }
  }
}

Plugin 系统

第三方包只需在 pyproject.toml 声明 entry point 即可注册自定义提取阶段：

[project.entry-points."universal_extractor.stages"]
my_stage = "my_package.extraction:MyCustomStage"

自定义 Stage 实现：

from UniversalExtractor.pipeline import ExtractionStage, StageContext, PipelineStageResult

class MyCustomStage(ExtractionStage):
    stage_name = "my_custom_stage"
    stage_index = 9
    description = "My custom extraction logic"

    def can_handle(self, url: str, context: StageContext) -> bool:
        return True

    def quality_score(self, url: str, context: StageContext) -> int:
        return 50  # 0–100，高分优先执行

    def extract(self, url: str, context: StageContext) -> PipelineStageResult:
        result = PipelineStageResult(stage_name=self.stage_name, stage_index=self.stage_index)
        # ... 你的提取逻辑 ...
        return result

安装你的包后，Pipeline 启动时自动发现并注册。

模块结构

UniversalExtractor/
├── __init__.py              # 懒加载公开 API（100+ 导出符号）
├── extractor.py             # URL + 文件自动提取入口
├── pipeline/                # Pipeline 编排引擎
│   ├── pipeline.py          #   9 层降级编排 + 中间件 + 信号
│   ├── config.py            #   全局配置
│   ├── stage.py             #   ExtractionStage ABC + StageRegistry
│   ├── middleware.py        #   下载中间件链 + 后处理管道
│   ├── context.py           #   Stage 共享上下文
│   ├── result.py            #   Pipeline 结果数据类
│   └── stages/              #   提取阶段实现
│       ├── jina.py          #   ① Jina Reader 快扫
│       ├── curl.py          #   ② curl_cffi HTTP
│       ├── browser.py       #   ③ Playwright 浏览器 DOM
│       ├── canvas.py        #   ④ Canvas fillText Hook
│       ├── cdp.py           #   ⑤ CDP Heap Scan
│       ├── ocr.py           #   ⑥ 截图 OCR
│       ├── vision.py        #   ⑦ Vision LLM
│       ├── captcha_slider.py #  ⑧ 滑动验证码求解
│       ├── captcha_font.py  #   ⑨ 字体解密
│       └── browser_agent.py #   ⑩ AI 代理兜底
├── captcha/                 # 验证码求解子包
│   ├── slider_solver.py     #   滑动验证码（OpenCV + ddddocr）
│   └── font_solver.py       #   字体解密（fontTools + ddddocr）
├── extraction/              # 结构化提取
│   ├── structured.py        #   Schema / Regex / LLM 策略
│   ├── table_extraction.py  #   表格提取（含分块 LLM）
│   ├── llm_strategy.py      #   LLM 提取策略
│   └── schema_strategy.py   #   Schema 提取策略
├── ocr_providers.py         # Vision 后端（GPT-4o / Claude / Qwen / DeepSeek / Tesseract）
├── canvas_hook.py           # Canvas fillText / strokeText 拦截 JS
├── screenshot.py            # 截帧 / 感知哈希去重 / 垂直拼接
├── scrolling.py             # 虚拟滚动（4 层降级 + VirtualScrollConfig）
├── classifier.py            # URL 分类 + 7 种能力标签
├── strategy.py              # 域名策略缓存
├── adaptive.py              # SQLite 自适应学习
├── completeness.py          # 7 因子连续完整性评分
├── cross_validator.py       # 多源交叉验证
├── search.py                # 多引擎搜索（DuckDuckGo / Brave / Exa / SearXNG）
├── filter.py                # 3 层内容过滤器（Pruning / BM25 / LLM）
├── crawler.py               # 深度爬取（BFS / DFS / BestFirst + 链接评分）
├── request_queue.py         # SQLite 持久化请求队列
├── signals.py               # 事件总线（发布/订阅）
├── file_ingest.py           # 文件 → Markdown（PDF / Word / Excel / PPT / EPUB）
├── auth.py                  # 登录态管理（加密存储 + 自动续期）
├── mobile.py                # 移动端提取（ADB 截图 + OCR）
├── mitm_capture.py          # MITM 代理 API 捕获
├── wechat_mp.py             # 微信小程序 / 公众号提取
├── wxapkg.py                # 微信小程序包 (.wxapkg) 扫描与解包
├── example_api.py           # 示例驱动自动提取
├── api_server.py            # REST API Server（FastAPI）
├── mcp_server.py            # MCP Server
├── http_client.py           # HTTP 客户端（curl_cffi + dev mode）
├── rate_limiter.py          # 并发安全限速器
├── proxy_manager.py         # 代理管理器
├── session_manager.py       # 会话管理器
├── observability.py         # 性能指标收集
├── weblens.py               # 搜 + 筛 + 抓 编排引擎
├── jd_engine.py             # 全平台 JD 结构化引擎
└── cli.py                   # 命令行接口（11 个子命令）

提取能力与覆盖率

网站类型	生效层	提取率
博客 / 新闻 / 文档	① DOM	~98%
SPA（React / Vue）	③ Browser DOM	~95%
反爬保护站点	③ StealthyFetcher + Cloudflare 求解	~95%
富文本编辑器	③ DOM（6 种编辑器选择器）	~95%
Shadow DOM / iframe	③ deepText 递归	~95%
自定义字体加密（起点 / 天眼查等）	⑨ FontSolver → ③ DOM 重新提取	~90%
普通 Canvas 页面	④ Canvas Hook（rAF 轮询）	~90%
GeeTest / DataDome 验证码	⑧ CaptchaSlider 求解 → ③ 重新提取	~85%
Canvas 流式文档（WPS / 飞书 / 腾讯文档）	⑤⑥⑦ OCR + Vision LLM	~70%

技术栈

组件	用途
Scrapling 0.4.9	浏览器自动化 + TLS 指纹伪装 + Cloudflare Turnstile 求解
Playwright CDP	Chrome DevTools 底层注入 / 内存扫描 / 滚动控制
OpenCV	Canny 边缘检测 + 模板匹配（滑动验证码求解）
ddddocr	ONNX 中文 OCR + 目标检测（字体解密 + 顺序验证码）
fontTools	TrueType / OpenType cmap 表解析（字体解密）
Multi-Vision-LLM	GPT-4o / Claude / Qwen-VL / DeepSeek 云端 OCR
Tesseract	本地 OCR 保底
curl_cffi	TLS 指纹伪装 HTTP 请求
FastAPI	REST API Server
SQLite	策略缓存 + 请求队列持久化

使用伦理与责任声明

本项目定位为本地网页内容提取研究工具，设计初衷面向：

🟢 合法合规场景：个人学习研究、自有网站内容管理、已获授权的数据采集、学术研究、SEO 分析
🟢 合理使用：遵循目标网站 robots.txt，控制请求频率（内置 RateLimiter），不绕过付费墙（paywall）
🟢 本地优先：所有提取默认在本地完成，不上传用户数据到第三方（除非用户显式配置 Vision LLM 后端）

责任边界

本项目不是反爬虫对抗工具。Captcha 求解和 TLS 伪装能力的存在是为了在合法授权场景下保持提取可用性
使用者应遵守目标网站的**服务条款（ToS）**和所在地法律法规
使用者对自身的使用行为承担全部责任，本项目作者不对任何滥用行为负责
如果你不确定自己的使用场景是否合规，请咨询法律专业人士

技术中性声明

Canvas Hook、字体解密、滑动验证码求解等技术属于通用的浏览器自动化和图像处理技术，在以下合法场景中均有应用：

无障碍访问（将 Canvas 渲染的文字转为屏幕阅读器可读文本）
网页存档与数字保存
自动化测试与质量保证
内容聚合与搜索引擎索引

技术本身是中性的——如何使用取决于使用者。

License

MIT

Name		Name	Last commit message	Last commit date
Latest commit History 91 Commits
.github/workflows		.github/workflows
UniversalExtractor		UniversalExtractor
adr		adr
docs		docs
tests		tests
.dockerignore		.dockerignore
.gitignore		.gitignore
.pre-commit-config.yaml		.pre-commit-config.yaml
CHANGELOG.md		CHANGELOG.md
Dockerfile		Dockerfile
LEGAL_WARNING.md		LEGAL_WARNING.md
LICENSE		LICENSE
README.md		README.md
README_EN.md		README_EN.md
docker-compose.yml		docker-compose.yml
mkdocs.yml		mkdocs.yml
pyproject.toml		pyproject.toml
requirements.txt		requirements.txt

	Firecrawl	Scrapling	Crawl4AI	Universal Extractor
许可协议	AGPL-3.0	BSD-3	Apache-2.0	MIT
Canvas 绑图页面	❌	❌	❌	✅
字体加密解密	❌	❌	❌	✅
滑动验证码本地求解	❌	❌	❌	✅
文字顺序验证码	❌	❌	❌	✅
文件提取（PDF / Office）	❌	❌	❌	✅
深度爬取（BFS / DFS / BestFirst）	✅	❌	✅	✅
链接相关性评分	❌	❌	✅	✅
移动端 App 提取	❌	❌	❌	✅
微信小程序提取	❌	❌	❌	✅
MITM API 捕获	❌	❌	❌	✅
登录态管理	❌	❌	❌	✅
示例驱动自动提取	❌	❌	❌	✅
中间件链 / 事件系统	❌	❌	❌	✅
REST API Server	✅	❌	✅	✅
MCP Server	✅	✅	❌	✅
策略自适应学习	❌	❌	❌	✅
部署方式	Docker / 云	pip install	pip install	pip install
结构化提取（Schema / LLM）	✅	❌	✅	✅
表格提取（分块 LLM）	❌	❌	❌	✅
多引擎搜索 + 交叉验证	❌	❌	❌	✅

Folders and files

Latest commit

History

Repository files navigation

Universal Extractor

别人做不到的，我们做了

别人能做的，我们更快

架构

快速开始

安装

🐳 Docker（推荐部署方式）

配置 Vision 后端（可选）

最简用法

功能全景

🌐 网页提取

📄 文件提取

🔍 搜索 + 提取

🕷️ 深度爬取

📊 结构化提取

🧩 示例驱动提取

📱 移动端提取

💬 微信小程序 & 公众号

🧠 验证码求解

🔊 认证 & 登录态

🔌 中间件 & 信号

🌍 内容过滤

CLI

REST API Server

MCP Server

Plugin 系统

模块结构

提取能力与覆盖率

同类工具对比

技术栈

使用伦理与责任声明

责任边界

技术中性声明

License

About

Topics

Resources

License

Contributing

Uh oh!

Stars

Watchers

Forks

Releases 4

Packages 0

Uh oh!

Contributors

Uh oh!

Languages

Packages