ServiziDigitali24SRL/lead-scraper-agent
GitHub: ServiziDigitali24SRL/lead-scraper-agent
一款开源的 B2B 潜在客户挖掘 Agent，通过 15 阶段的严格流水线实现从客户画像访谈、公开数据源发现到多渠道信息富化与导出的全自动化流程。
Stars: 0 | Forks: 0
# 🕵️ Lead Scraper Agent
**许可证：** MIT · **要求：** Python 3.10+ · **付费设置：** 无
## ⚡ 快速开始
```
git clone https://github.com//lead-scraper-agent.git
cd lead-scraper-agent
python -m venv .venv && source .venv/bin/activate # Windows: .venv\Scripts\activate
# （可编辑）使用 CLI 命令 `lead-scraper` 安装
pip install -e . # runtime; per i test: pip install -e ".[dev]"
# 启动 agent：开始交互式问答，然后是 15 个阶段
lead-scraper # equivalente a: python -m agent
```
无需修改任何配置文件：开箱即用。
`.env` 是**可选的**，仅供想要扩大数据量的用户使用（参见 § 扩展）。
### 想在不接触网络的情况下查看 pipeline 吗？
```
python -m examples.offline_demo
```
它使用本地 fixture（一个目录列表 + 三个
虚假主页）和 mock 的 DNS resolver 执行**全部 15 个阶段**：
**零网络请求**。这有助于理解 discovery → qualification → enrichment → export 的完整流程。
## 🧭 工作原理（15 阶段 pipeline）
pipeline 被划分为 **15 个原子阶段**。每个阶段都有明确的**输入要求**，
在数据库/磁盘上产生**已验证的 artifact** 和一个 **gate**：如果前一阶段未通过其自身的检查，当前阶段就无法开始。
这可以防止“压缩”或跳过步骤（即使是修改该工具的 LLM 也无法绕过）。
| # | 阶段 | 已验证的 artifact | 退出 Gate |
|---|---|---|---|
| 0 | **PREFLIGHT** | `run_id`, 文件夹, DB init | DB 可写，存在 deps |
| 1 | **INTERVISTA** | `icp_profile.json` | 必填字段已填写 |
| 2 | **VALIDAZIONE ICP** | 通过 schema 验证的配置文件 | **显式**选择 `limit` |
| 3 | **QUERY MAPPING** | `queries.json` | ≥1 个 OSM tag + ≥1 个 keyword + seed directory |
| 4 | **PIANO DISCOVERY** | 来源计划 + 配额 | 配额与 `limit` 一致 |
| 5 | **DISCOVERY** | `raw_domains` (DB) | ≥1 个域名，状态已保存 |
| 6 | **DEDUP + CANONICAL** | `domains` (根域名) | 剩余 0 个重复项 |
| 7 | **FILTRO ESCLUSIONI** | 已移除社交/聚合平台 | exclude_list 中无任何域名 |
| 8 | **QUALIFICA (ICP)** | `qualified` (score ≥ min) | 子集已保存，count > 0 |
| 9 | **CODA ENRICHMENT** | `pending` 队列 | 队列 = 合格数量（仅限合格项！） |
| 10 | **FETCH (礼貌)** | 缓存中的 HTML | 已读取 robots，rate-limit 已激活 |
| 11 | **ENRICHMENT 25 METODI** | `candidates` (+ `source`) | 每个 candidate 都有一个 `source` |
| 12 | **VALIDAZIONE** | 已验证 MX + E.164 | `verified` 已填充 |
| 13 | **SCORING** | `role_score` | 每个 contact 都有一个 score |
| 14 | **COMPLIANCE** | 抑制列表 + 来源 | 排除 opt-out |
| 15 | **EXPORT** | `CSV` + `JSON` 位于 `./output/` | 文件已写入，行数 = 预期 |
每个阶段都会写入 **SQLite**，并为每条记录保存状态（`pending/done/failed/blocked`）：
如果运行中断，会从特定阶段和未完成的记录处**恢复**（`--resume`）。
## 🚧 严格的 Guardrail（防捷径，以代码形式实现）
1. **每个阶段在磁盘/数据库上生成一个 artifact**，并带有明确的 `phase_status`。绝对没有仅在内存中进行的数据传递。
2. **入场 Gate**：每个阶段都会调用 `require_phase(N-1, "done")`。如果上游的 artifact 缺失或状态不是 `done` → 触发 `RuntimeError`，停止运行。
3. **禁止合并**：enrichment（阶段 11）**仅**从合格的队列中读取数据。无法对未通过 qualification 的域名进行 enrich（`queue_only_contains_qualified()` 会验证这一点）。
4. **顺序不可变**：`PHASES = [0..15]`。`--from N` / `--only N` 虽然可用于调试，但**仍会**检查上游的 artifact。
5. **由用户决定 limit**：如果 `target.limit` 不明确（来自访谈或 `--limit`），阶段 2 将失败。没有隐藏的默认值。
6. **幂等性**：基于 key 进行 upsert；重新运行不会导致数据重复。
7. 在每个阶段之后进行 **Checkpoint**（在继续之前进行 commit）。
8. **顺序测试**：`tests/test_pipeline_order.py` 证明跳过阶段会引发错误。否则工具无法通过测试（不是 "verde"）。
## ⭐ Discovery — 搜索犬（优先处理站点目录）
最丰富的数据载体不是单个网站：而是**列出了大量站点的页面**（行业协会、商会、B2B 门户网站、专业名录、展会参展商名单、联盟成员名单）。
`discovery/directories.py` 是一个专门的 crawler，它会：
1. 根据您的 ICP **寻找 seed directory**（`query_mapper` 生成类似 `"elenco "`、`"associazione "`、`"albo "`、`"registro imprese "`、`" aziende site:."` 的查询）。
2. **跟踪目录列表**：在每个 seed 上，提取所有指向企业域名的外部链接（排除内部/导航链接）。
3. **受控的递归**，深度可达 `--directory-depth`（默认为 2），并使用已访问 URL 的集合来避免循环。
4. **分页扩展**：处理 `?page=2`、"下一页"、load-more、A-Z 字母索引。
5. **目录识别**：如果一个页面链接到许多不同的外部域名，则认为它是一个列表 → 宝藏矿脉，提取所有内容。
其他来源将补充该数据级联，每个来源都有自己的配额（阶段 4）：
| 来源 | 优势 | 需要的 Key |
|---|---|---|
| **directories.py** ⭐ | 网站列表，针对特定行业极其密集 | 否 |
| **osm.py** (Overpass) | 本地企业，通常已内置 `phone` | 否 |
| **wikidata.py** (SPARQL) | 按行业 + 国家/地区结构化的企业 | 否 |
| **commoncrawl.py** | 针对高数量的原始数据量 | 否 |
| **ddg.py** (DuckDuckGo HTML) | 针对 keyword 的精细化结果 | 否 |
所有数据都被标准化为**根域名**（`tldextract`），并进行全局去重；社交/市场/聚合平台（facebook、instagram、tripadvisor、thefork、justeat、deliveroo、amazon…）均被**剔除**。
## 🎤 阶段 1 — 访谈
启动时，agent 会通过示例逐一询问：
1. **你是谁？** 你卖什么。
2. **Buyer persona？** 决策者角色 + 企业规模。
3. **行业/类别？**
4. **国家？**（必填，ISO 格式）。
5. **城市/地区？**（可选）。
6. **需要多少个 lead？** — **由你决定，没有隐藏的默认值：**
```
[1] ~500 veloce, minuti (prova/test)
[2] ~5.000 ore, gratis e gentile
[3] ~50.000 mezza giornata / giorno
[4] 100k-300k più giorni gratis, o ore con proxy/SERP (.env)
[5] personalizzato -> inserisci il numero
```
如果你不选择，agent 会**要求**你提供该值。输出 → `icp_profile.json`。
你也可以跳过访谈：`python -m agent --profile icp_profile.json`。
## 🔎 阶段 11 — 25 种 enrichment 方法
全部免费，按优先级级联执行；每个 contact 上都会保留 `source` 追踪。
**A · 在已下载的 HTML 中**
`/contatti /impressum /about` · `mailto:` · JSON-LD · Cloudflare cfemail (XOR 解码) ·
去混淆 `[at]/[dot]` + 实体 + unicode · `tel:` · Microdata/RDFa · hCard ·
OpenGraph · 注释/`data-*` · 图片 alt/title
**B · 同一站点的其他页面**
深层页面 `/team /staff /legal` · sitemap.xml + robots.txt（递归 sitemapindex） · security.txt/humans.txt · privacy/DPO · careers · 链接的 PDF/DOCX（提取文本） · JS/CSS bundle · vCard/RSS
**C · 外部公开来源（无需 key）**
Wayback CDX · crt.sh (subdomains) · Overpass（通过 `website` tag 获取电话） ·
Common Crawl index · GitHub（公开 commit 中的 email）
**D · 推断 + 验证**
根据页面上找到的姓名进行 pattern-guess `nome.cognome@`，并进行 **MX** 验证（在 `--enable-guess` 下，**绝不**发送 SMTP）。
## ✅ 验证、scoring、输出
- **Email：** 验证域名的 **MX** 记录（无 SMTP 探测 → 避免黑名单）。
- **电话：** 使用 `phonenumbers` 规范化为 **E.164** 格式。
- **Contact score：** 一个 `mario.rossi@`（决策者）的权重高于通用的 `info@`。
位于 `./output/` 的最终 CSV：
```
domain,company_name,source_discovery,kind,value,source_methods,confidence,verified,
status,exportable,guessed,inferred,role_score,source_url,page_url,found_at,last_seen
```
- **`status`**：`valid` / `role` / `no_mx` / `catch_all` / `invalid` / `disposable`。
- **`exportable`**：仅对已验证且可操作的 lead 为 `true` → 在此字段上进行筛选以用于 CRM 导入。
- **`inferred`**：如果地址是推断出来的（pattern-guess）而不是直接找到的，则为 `true`。
- **`source_url` + `last_seen`**：每个数据的来源均可审计。
找到但无联系信息的企业仍会被记录（`value` 为空的行） → 这样你就可以衡量按来源/行业划分的**真实覆盖率**。也支持导出 JSON。
可选的 CRM 插件（`export/plugins/`）默认为 **OFF**。
## ⚙️ 主要 Flag
| Flag | 默认值 | 功能 |
|---|---|---|
| `--profile FILE` | — | 使用现有的 `icp_profile.json`（跳过访谈） |
| `--limit N` | — | 限定要 enrich 的合格域名数量（必须显式指定！） |
| `--resume` | off | 从 SQLite 状态恢复中断的运行 |
| `--concurrency N` | 5 | 全局并发请求数 |
| `--rate SEC` | 1.0 | 向同一 host 发送请求的最小间隔秒数 |
| `--directory-depth N` | 2 | directory 的递归深度 |
| `--enable-guess` | off | 启用第 25 种方法（带 pattern 推断 + MX 的 pattern-guess） |
| `--smtp` | off | 启用 SMTP 验证（RCPT + catch-all）。⚠️ 端口 25 经常被屏蔽 → `unknown`；有被列入黑名单的风险，建议在代理后使用 |
| `--gdpr` | off | GDPR 模式：强制执行 `robots.txt`，记录合法依据，保证来源可追溯 |
| `--skip GROUP` | — | 跳过一组 enrichment 方法 (A/B/C/D)，可重复使用 |
| `--skip-source SRC` | — | 跳过 discovery 来源，可重复使用 |
| `--from N` / `--only N` | — | 调试：从阶段 N 开始 / 仅执行阶段 N（验证上游 artifact + 配置 hash） |
| `--ignore-robots` | off | （不推荐）忽略 robots.txt（在 `--gdpr` 下被忽略） |
| `--out DIR` | ./output | 输出文件夹 |
### 反馈循环（退信账本）
记录营销活动的真实结果：该工具会学习哪些方法有效，并且不会重新导出已发生退信的地址。
```
lead-scraper-ledger record --db output/state.sqlite \
--value mario.rossi@acme.it --outcome bounce --method D:pattern-guess
lead-scraper-ledger stats --db output/state.sqlite # hit-rate per metodo
```
## 🚀 扩展（可选，`.env`）
默认情况下，agent 是**缓慢且温和的**（尊重目标 host，支持断点续传）。为了扩大数据量，请将 `.env.example` 复制到 `.env`：
```
PROXY_POOL= # aggira i rate-limit degli host
SERP_API_KEY= # discovery più rapida (rimpiazza Bing, ritirata ago-2025)
LLM_API_KEY= # query_mapper adattivo per settori non pre-mappati
CRM_PLUGIN= # es. "example" per attivare uno stub di export
```
这些都不是必需的：如果缺失 = 100% 免费模式。
## ⚖️ 合规性（在欧盟地区上线前请先阅读）
此工具收集的是**公开的联系数据**。在欧盟，将其用于 B2B 的主动营销需要**合法依据（合法利益）**和 **opt-out** 机制。agent 会为你提供帮助：
- 每条记录均保存**来源**（discovery 来源 + 提取方法）。
- **抑制列表**（`compliance/suppression.txt`）：始终排除。
- 默认遵守 `robots.txt`；具有可识别且实的 UA（从不伪造 Googlebot）；每个域名最多抓取 40 页；验证 MX，**不进行 SMTP RCPT**。
⚠️ 使用的法律责任仍由运营者承担。本文不构成法律建议。
## 🧱 仓库结构
```
lead-scraper-agent/
agent/
__main__.py # entrypoint CLI (python -m agent)
interview.py # Fase 1 -> icp_profile.json
query_mapper.py # ICP -> tag OSM + keyword + query + seed directory
core/
orchestrator.py # esegue PHASES[0..15] in ordine, applica i gate
phases.py # enum fasi + require_phase() (GUARDRAIL)
db.py # SQLite: stato per record/fase, resume, upsert idempotente
fetcher.py # httpx async: robots-aware, rate-limit, retry backoff, UA onesto
config.py # .env opzionale (proxy/SERP/LLM -> spenti di default)
discovery/
directories.py ⭐ # crawler di ELENCHI DI SITI
osm.py wikidata.py commoncrawl.py ddg.py base.py
qualify/
scorer.py # dominio vs ICP -> tiene solo i match
enrich/
extractors/ # i 25 metodi: group_a/b/c/d.py + common.py
validate.py # MX + phonenumbers E.164
score.py # decisore vs info@
export/
to_csv.py to_json.py plugins/ # plugin CRM OFF di default
compliance/
suppression.py suppression.txt
examples/
offline_demo.py # run completo su fixture, zero rete
tests/ # test_pipeline_order.py (gate) + cfemail/deobf/sitemap/phones/dedup/e2e
.env.example requirements.txt LICENSE (MIT) README.md
```
## 🤝 贡献
最有用的扩展点是 `agent/query_mapper.py` 中的 **行业 → OSM tag / keyword / seed directory 映射**：映射的垂直领域越多，你挖出的企业就越多。欢迎提交 PR。
标签：B2B销售, Python, Terrascan, URL抓取, 数据增强, 数据抓取, 无后门, 潜在客户挖掘, 自动化代理, 逆向工具