外贸客户挖掘 lead-miner：按行业+城市生成 Google Maps 客户列表，用户确认范围后 Hunter.io 验证邮箱。三步流程可控成本，输出 xlsx 客户名单含公司名+地址+电话+邮箱+置信度。适合外贸拓客、邮件群发、本地客户开发。全程接真实接口。
---
name: lead-miner
description: 外贸客户挖掘（lead-miner）：按城市+行业搜 Google Maps 客户 → 用户确认范围 → Hunter.io 验证邮箱。三步流程可控成本，输出 xlsx 客户名单含公司名+地址+电话+邮箱+联系人+置信度。适合外贸拓客找海外买家、邮件群发客户名单、本地市场调研（杭州咖啡店主邮箱、义乌外贸客户、上海餐厅老板）。全程接平台真实接口（serpapi.engine + hunter.domain-search + hunter.email-verifier），无 mock。
---
本 skill 的真实能力依赖平台后端工具。在网页 / chat 端点下，这些工具可能不可用：
此时**请勿轻信**，请改用 **agent / CLI 路径**运行本 skill 以获取真实结果。任何路径下都**严禁谎称「已调用 / 已搜索完成」而实际未执行**。
按 **用户问询 → Google Maps 搜客户 → 用户确认范围 → Hunter.io 验证邮箱** 的顺序找外贸客户，每步可控、不乱扣费。
用户说"外贸搜客"、"找海外买家"、"按行业+城市搜客户"、"找杭州/上海/纽约的餐厅/咖啡/健身的老板邮箱"、"我要做 XX 卖给 Y 城市的咖啡店，帮我找联系方式"、"导出 Google Maps 客户名单带邮箱"时触发。
[0] 问询用户 智能体先问：找哪个城市？哪些类目？要几个候选？预算上限？
↓
↓ 用户确认后
↓
[1] search 调平台 serpapi.engine 拿 Google Maps POI 列表（手机 / 官网 / 评分齐全）
↓
↓ 输出 candidates.json（含完整商家资料）
↓
[2] confirm 用户从候选里勾选要验证的（first-N / 指定 id / all）
↓
↓ 不调任何 API；只算 cost_estimate + 检查 --budget
↓ 输出 selected.json
↓
[3] verify 仅对 selected 调 hunter.domain-search + hunter.email-verifier 拿邮箱+验证
↓
输出 leads.xlsx（含公司名/地址/电话/官网/邮箱/联系人/职位/置信度/验证状态）→ **关键**：verify 是按用户确认的 select 范围跑；不选 = 不扣 Hunter 任何费用。
1. **控制成本**：城市/类目/数量直接影响 search 扣费（¥0.21/次 serpapi）+ verify 扣费（hunter 1 个域名 ~¥0.30）。
智能体没问清楚就开跑，可能一次 ¥10+ 余额就空了。
2. **避免乱搜**：杭州和"杭州的咖啡店"对应 query 不同（英文命中率更高）；
类目选错就找不到 POI。
3. **预算协商**：用户给 ¥3 的预算，智能体选择 first-3 验证而不是 first-50。
python scripts/lead_miner.py run --interactive智能体会按顺序问：
[1/4] 找哪个城市？（1=杭州，2=上海，3=北京，4=深圳，5=广州，6=成都，7=武汉，8=其他）
[2/4] 找哪些类目？（1=咖啡，2=餐厅，3=健身房；可多选如 1,2）
[3/4] 要几个候选？（1=10 个，2=30 个，3=50 个，4=100 个）
[4/4] 预算上限？【预估成本 ¥X.YY，建议 ¥X.YY+0.5 buffer】
确认后跑 search → confirm → verify。python scripts/lead_miner.py run --description "我要杭州的咖啡店老板邮箱，10个以内"LLM 自动抽取 `{city: 杭州, category: coffee, limit: 10}` 后直接跑完整三步。
python scripts/lead_miner.py run \
--city 杭州 \
--category coffee \
--limit 10 \
--select first-3 \
--budget 1.5 \
--output leads.xlsxecho '{"city": "杭州", "category": "coffee", "limit": 10, "budget": 1.5, "select": "first-3"}' \
| python scripts/lead_miner.py run --stdin支持 **3 种模式**：
python scripts/lead_miner.py search \
--mode serpapi \
--city 杭州 \
--category coffee \
--limit 50 \
--output candidates.jsonexport GOOGLE_MAPS_API_KEY=AIzaSy...
python scripts/lead_miner.py search --mode api --city 杭州 --category coffee --limit 50python scripts/lead_miner.py search --mode browser --city 杭州 --category coffee --limit 30
python scripts/lead_miner.py search --mode browser --headed ... # 加 --headed 弹窗（调试）{
"ok": true,
"source": "platform_serpapi_google_maps",
"mode": "serpapi",
"query": "coffee 杭州",
"returned": 18,
"places": [
{
"id": "ChIJ...",
"name": "西湖龙井咖啡馆",
"category": "coffee",
"address": "浙江省杭州市西湖区...",
"phone": "+86 571-xxxx-xxxx",
"website": "https://example.com",
"domain": "example.com",
"rating": 4.6,
"reviews": 234,
"lat": 30.27,
"lng": 120.13,
"place_id": "ChIJ..."
}
],
"platform_cost_yuan": "0.21",
"note": "phone/website 字段完整；可直接进 verify 步骤"
}# 默认选前 3 个（试水）
python scripts/lead_miner.py confirm \
--input candidates.json \
--select first-3 \
--output selected.json
# 手动指定 id（逗号分隔）
python scripts/lead_miner.py confirm \
--input candidates.json \
--select P001,P005,P009
# 全部验证
python scripts/lead_miner.py confirm --input candidates.json --select all
# 加 --budget 上限（推荐！避免乱扣费）
python scripts/lead_miner.py confirm \
--input candidates.json \
--select all \
--budget 1.00**`--select` 写法**：
**confirm 输出 selected.json 示例**：
{
"ok": true,
"selected_count": 3,
"estimated_emails_found": 5,
"cost_estimate": {
"mode": "serpapi",
"search_count": 18,
"verify_count": 3,
"used_description": false,
"platform_llm_calls": 0,
"search_unit_yuan": 0.21,
"verify_cost_yuan": 0.762,
"platform_cost_yuan": 0.0,
"total_yuan": 4.542,
"breakdown": "search(18×0.21=3.78) + verify(domain_search(3×0.204=0.612) + verifier(4.5×0.10=0.45)=0.912) + llm(0×0.02=0.0) = ¥4.542"
},
"budget_yuan": 5.0,
"budget_exceeded": false,
"next_step": "运行 verify 子命令前，请确认 selected_count 和 cost_estimate 与你的预期一致"
}**`--budget` 拦截**：估算费用超过 `--budget N` 时，confirm 返回 `ok=false`；run 会自动跳过 verify，避免乱扣费。
python scripts/lead_miner.py verify \
--input selected.json \
--output leads.xlsx1. **对每个商家**：
2. **对每个邮箱**：
{
"ok": true,
"output": "leads.xlsx",
"lead_count": 12,
"no_email_count": 2,
"verified_count": 8,
"cost_summary": {
"calls": 18,
"domain_search_yuan": 4.08,
"email_verifier_yuan": 1.50,
"total_yuan": 5.58,
"source": "platform_hunter"
}
}**主 sheet "Leads"**：
| 列 | 含义 |
|----|------|
| 公司名 | 公司名 |
| 类别 | 类目 |
| 地址 | Google Maps 地址 |
| 电话 | Google Maps 电话 |
| 网站 | 官网 URL |
| 域名 | 提取的域名 |
| 邮箱 | Hunter 找到的邮箱 |
| 联系人 | first_name + last_name |
| 职位 | position（CEO / Manager / 等） |
| 部门 | department |
| 置信度 | Hunter confidence（0-100） |
| 验证状态 | ✓ valid / ✗ invalid / ? accept_all / ? webmail |
| 数据源 | platform_hunter |
**副 sheet "未找到邮箱"**：Hunter 找不到的商家（写"无 website"或"Hunter 无数据"原因）。
skill 内部用 `calc_cost()` 实时计算，按**数据源 + 数量**给透明报价。
| 项 | 单价（元/次） | 说明 |
|----|--------------|------|
| search serpapi | 0.21 | 平台 serpapi.engine，调 Google Maps（仅上游成功时扣费） |
| search api | 0.23 | Google Maps Places API（用户自备 KEY） |
| search browser | 0.0 | 本地 playwright，无平台成本（但有反爬风险） |
| verify domain-search | 0.204 | hunter.domain-search，按域名算 |
| verify email-verifier | 0.10 | hunter.email-verifier，按邮箱算（每域名平均 1.5 个邮箱） |
| platform chat | 0.02 | AIMS qwen-flash（仅当用 --description 时） |
| 场景 | 公式 | 实际成本 |
|------|------|---------|
| serpapi 搜 10 + verify 3 | `10×0.21 + 3×0.204 + 3×1.5×0.10` | ¥3.12 |
| serpapi 搜 50 + verify 20 | `50×0.21 + 20×0.204 + 20×1.5×0.10` | ¥16.44 |
| serpapi 搜 100 + verify first-3 | `100×0.21 + 3×0.304` | ¥21.91 |
| description + run + first-3 | `+3×0.21 + 3×0.304 + 1×0.02` | ¥1.57 |
# 1. --budget 上限（推荐）
python lead_miner.py run --city 杭州 --limit 50 --select first-10 --budget 1.00
# 2. --select first-N 只验证前 N 个
python lead_miner.py run --city 杭州 --limit 50 --select first-3
# 3. 不设 budget 会显示 cost 但不拦截
python lead_miner.py run --city 杭州 --limit 50 --select all| skill 调用 | 卖价（分） | 实际成本（serpapi search）| 推荐毛利 |
|-----------|-----------|---------------------------|---------|
| search serpapi | 30 | search 1×0.21 + llm 1×0.02 | ~23% |
| verify 1 个邮箱 | 50 | domain 0.204 + verifier 0.10 | ~40% |
| run 一条龙（first-3）| 150 | 0.21 + 3×0.304 | ~37% |
详细定价策略：见 docs/PRICING.md。
pip install openpyxl
# playwright 仅 browser 模式需要
pip install playwright && python -m playwright install chromium| 错误 | 原因 | 解决 |
|------|------|------|
| `search` 返回空 | 城市/类目无匹配 | 改 category 或放宽 city / 用英文 query（"coffee shop Hangzhou"） |
| `verify` 找不到邮箱 | 域名 Hunter 无数据 | 写"未找到邮箱" sheet，可手动用其他渠道找 |
| `leads.xlsx` 文件被占用 | Excel 打开状态 | 关闭 Excel 后重跑 |
| `serpapi.engine` 余额不足 | Token 用完 | 充值或换 agent/model |
| `platform_token_plan` 429 | 平台限流 | 加 `--limit 5` 或等几分钟后重试 |
| `openpyxl 未安装` | 缺依赖 | `pip install openpyxl` 或接受 JSON 输出 |
| `AIMS_API_KEY 缺` | 环境变量未设 | `export AIMS_API_KEY=...` |
A: 分三步让用户**控制成本**。search 0.21 元算便宜，但 verify 1 个域名 0.30 元 ×50 就 ¥15 了。
用户看到 cost_estimate 后可以决定要不要花这个钱。
A: 可以。删 selected.json 重跑 confirm 即可，没扣任何费用。
A: 1 个类目 50 个候选 verify 全部可能 ¥15+；先问预算就能选合适的 select 范围。
A: 直接在 `--city 城市名 --category 类别` 指定英文 query（如 `coffee shop Shenzhen`），
serpapi 会按 Google Maps 数据返回。
A: 在 leads.xlsx 的"未找到邮箱" sheet 里有说明。可手动去公司官网 About 页找，或换其他工具（apollo.io / snov.io）。
---
**问题**：小李做咖啡豆出口，想找海外咖啡店进口商。
**步骤**：
1. 问询 → 选城市"义乌"，类目"咖啡"，数量 30 个，预算 ¥5
2. search → 拿到 20 个义乌咖啡店 + 10 个相关企业（星巴克、两岸咖啡、上岛咖啡等）
3. confirm → 选 first-10（先试水），看到 cost_estimate = ¥3.4，能接受
4. verify → hunter 真接口，10 个域名里 6 个找到邮箱
5. 拿到 xlsx → "Leads" sheet 6 行 + "未找到邮箱" sheet 4 行
6. 拿着邮箱写开发信，按行业/规模分组发
**实际成本**：¥3.4 左右（search 0.21 + verify 6×0.304 + llm 0.02 + 一些失败的搜索不计费）
**问题**：王经理想摸杭州餐厅老板的联系方式，做新品推广。
**步骤**：
1. 问询 → 选"杭州"，类目"餐厅"，数量 50，预算 ¥10
2. search → 50 个杭州餐厅（含地址、电话、官网、评分）
3. confirm → 选 first-20（先验证 20 个）
4. verify → hunter 对餐厅官网有数据，平均 1.5 个邮箱/域
5. 拿到 xlsx → 30 个邮箱 + 联系人 + 职位
**实际成本**：search 0.21 + verify 20×0.304 = ¥6.3
**问题**：运营需要给"上海所有咖啡店"发新品介绍邮件，需要完整名单。
**步骤**：
1. 描述 → "上海咖啡店主邮箱，100 个以内，预算 ¥10"
2. LLM 自动抽 city=上海/category=coffee/limit=100/budget=10
3. search → 拿到 100 个上海咖啡店
4. confirm → 选 first-30（先 ¥10 预算内能验证多少）
5. verify → 30 个域名，hunter 命中 18 个
6. xlsx 导给邮件群发工具
**实际成本**：search 0.21 + verify 18×0.304 + LLM 0.02 = ¥5.7
---
| 人群 | 用法 | 价值 |
|------|------|------|
| **外贸业务员** | 找海外进口商，写开发信 | 一周节省 20+ 小时手动找客户 |
| **本地服务商** | 找本地餐厅/咖啡店/健身房推广 | 1 小时内拿到完整联系方式名单 |
| **市场调研员** | 摸某个城市/行业的客户密度 | 数据来源真实（Google Maps）而非 mock |
| **邮件营销人员** | 批量拿邮箱，配合 mailchimp/自己群发工具 | 邮箱经过 Hunter.io 验证，有效率 > 70% |
---
输入：城市=义乌，类目=coffee，limit=20，预算=¥2.0，select=first-2（有 domain 的）
| # | 名称 | domain | phone |
|---|------|--------|-------|
| 1 | 星巴克咖啡 | www.starbucks.com.cn | 0579 8527 0900 |
| 2 | 星巴克咖啡 | (空) | 0579 8502 0896 |
| 3 | 瓶子Coffee | (空) | (空) |
| 4 | **两岸咖啡义乌总店** | **www.liangan.cn** | (空) |
| 5 | 迪欧咖啡 | (空) | 0579 8525 2525 |
| 6 | 迪欧咖啡 | (空) | 0579 8559 0778 |
| 7 | 艾坦咖啡馆 | (空) | (空) |
| 8 | M-box Restaurant & Bar | (空) | 0579 8533 0199 |
| 9 | 两岸咖啡义乌兴中店 | (空) | 0579 8535 2777 |
| 10 | 米朗奇咖啡义乌店 | (空) | 0579 8562 2228 |
| 11 | **上岛咖啡** | **www.ubccn.com** | (空) |
| 12 | 蓝桥咖啡 | (空) | 0579 8553 3990 |
| 13 | 麦当劳 | www.mcdonalds.com.cn | 0579 8526 8108 |
| 14 | TOM'S COFFEE | (空) | 156 5792 6732 |
| 15 | 彼岸咖啡No.4 | (空) | 0579 8527 7004 |
| 16 | costa | (空) | (空) |
| 17 | ｐｅｒｒｙ＇ｓ ｃａｆｅ西餐吧 | (空) | 0579 8551 7695 |
| 18 | 迷失咖啡 | (空) | (空) |
| 19 | Bosphorus | (空) | 0579 8556 2463 |
| 20 | 拉丁餐厅 | (空) | (空) |
20 个里 4 个有 domain（20% 命中率，符合预期：大多小店没官网）。
**selected.json**：
{
"ok": true,
"selected": [
{"name": "两岸咖啡义乌总店", "domain": "www.liangan.cn", "phone": null, ...},
{"name": "上岛咖啡", "domain": "www.ubccn.com", "phone": null, ...}
],
"selected_count": 2
}**cost_estimate**（confirm 阶段）：
**verify 输出 leads.xlsx（5.7 KB，2 sheet）**：
**结论**：Hunter 中文小公司数据库覆盖有限（liangan.cn / ubccn.com 都无员工邮箱）。下次试外贸大客户（百度、字节、海尔等）应该有数据。
---
| 方式 | 耗时 | 成本 | 效果 |
|------|------|------|------|
| **手动 Google Maps + Hunter 注册** | 4-8 小时/50 个 | Hunter $49/月起 | 易遗漏，重复劳动 |
| **lead-miner 一键** | 30 秒/100 个 | ¥3-6/100 个 | 自动串联，全流程可控 |
| **第三方数据公司** | 等 1-3 天 | ¥1000+/份 | 数据可能过时 |
**lead-miner 的 5 个独特价值**：
1. **完全可控**：每步独立，不满意可中断不扣费
2. **透明计费**：cost_estimate 在扣费前告诉用户
3. **无需自己配 KEY**：平台代理所有上游 API（serpapi、hunter）
4. **可重复**：同一城市同一类目每次结果相似，便于对比
5. **可扩展**：本地模式 browser 模式完全离线，serpapi/api 模式在线
---
---
| 搭配 skill | 用法 |
|-----------|------|
| **contract-generator**（合同生成）| 找到客户邮箱 → 用合同模板生成合作合同 → 邮件发送 |
| **call-analyzer**（通话分析） | 客户回电录音 → AI 提炼痛点 → 跟进 |
| **format-converter**（格式转换） | xlsx → CSV → 导入 mailchimp / SendGrid |
| **excel-analyze**（数据分析）| 多次搜索结果汇总 → 分析客户密度分布 |
<!-- ===== 以下为内嵌脚本代码（agent 安装时按需落盘为 scripts/<name> 并 chmod +x） ===== -->
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""外贸搜客：从 Google Maps 找客户 → 用户确认范围 → 验证邮箱。
三步流程（每个步骤独立，避免乱扣费）：
1. search 调平台 serpapi.engine 拿候选客户列表
2. confirm 用户从列表中勾选要验证的（读 JSON 文件）
3. verify 仅对勾选部分调 hunter.domain-search + hunter.email-verifier
用法示例：
# 默认：交互式问询（城市/类目/预算）→ 直接执行三步流程
python lead_miner.py run
# 非交互（agent 调）：从 stdin 读 JSON 一次性给参数
python lead_miner.py run --stdin <<EOF
{"city": "杭州", "category": "coffee", "limit": 10, "budget": 1.0}
EOF
# 自然语言描述（让 qwen-flash 抽取 city/category/limit）
python lead_miner.py run --description "我要杭州的咖啡店老板邮箱，10个以内"
# 三步走完整流程（手动指定参数）
python lead_miner.py search --query "coffee shop Hangzhou" --limit 10 --output candidates.json
python lead_miner.py confirm --input candidates.json --select first-3 --output selected.json
python lead_miner.py verify --input selected.json --output leads.xlsx
依赖：
- AIMS_API_KEY（必填，平台认证）
- openpyxl（输出 xlsx）
当前状态：全部接真接口（平台 serpapi.engine + hunter.domain-search + hunter.email-verifier）。
真实成本：search 0.21 元/次（serpapi），domain 0.204 元/次 + verifier 0.10 元/次（hunter）。
预算可由 --budget 上限；超出则 confirm 自动跳过 verify。
"""
import argparse
import json
import os
import re
import sys
from pathlib import Path
_HERE = Path(__file__).resolve().parent
_ROOT = _HERE.parent # lead-miner/
_DATA = _ROOT / "data"
# AIMS 平台 MCP 端点（与 install.md §2 一致）
AIMS_URL = "https://aimsgateway.cn/mcp/"
# 让脚本可作为 lead-miner 独立 skill 运行
# _lib 查找（2026-09-22 修复）：优先包内 _lib/（发布包已随附），
# 退回开发布局 office-skills/_lib —— 否则装到用户侧会 ModuleNotFoundError。
for _lib_cand in (Path(__file__).resolve().parent.parent / "_lib",
Path(__file__).resolve().parent.parent.parent / "_lib"):
if _lib_cand.is_dir():
sys.path.insert(0, str(_lib_cand))
break
else:
raise RuntimeError(
"找不到 _lib/（需要 aims_functional_api.py）。"
"发布包应自带 _lib/；开发环境下请确认 office-skills/_lib 存在。"
)
try:
from aims_chat import chat_text_json # noqa: E402
_AIMS_OK = True
except Exception:
_AIMS_OK = False
# 确认协议（2026-09-22 platform-paid-confirm-protocol）
# 之前本脚本 _rpc_call 不传 confirm，也没有 --approved，导致 serpapi/hunter 的
# 付费调用永远被平台 -32006 拦下（不误扣费，但也交付不了任何真数据）。
try:
from aims_functional_api import ( # noqa: E402
NeedsConfirmation,
raise_if_needs_confirmation,
estimate_cost,
)
_CONFIRM_OK = True
except Exception:
class NeedsConfirmation(RuntimeError): # type: ignore[no-redef]
"""_lib 不可用时的占位（保证 except 子句始终有可用的类）。"""
def to_dict(self):
return {"ok": False, "needs_confirmation": True, "error": str(self)}
_CONFIRM_OK = False
def raise_if_needs_confirmation(resource_id, parsed): # type: ignore[misc]
return parsed
def estimate_cost(resource_id, units=1, api_key=None): # type: ignore[misc]
return {"ok": False, "error": "_lib/aims_functional_api 不可用"}
# ---------- 第一步：search ----------
def cmd_search(args):
"""按 mode 分发：serpapi（推荐）/ api（Google Places API）/ browser（playwright）"""
# ---- 自然语言抽取 ----
if args.description and (not args.city or args.city == "杭州"):
# 用户只给了 description，没给 city → 用 LLM 抽
params = _llm_extract_search_params(args.description)
if params:
args.city = params.get("city") or args.city
args.category = params.get("category") or args.category
args.limit = params.get("limit") or args.limit
args._llm_extracted = params
if args.mode == "api":
return _search_api(args)
elif args.mode == "browser":
return _search_browser(args)
elif args.mode == "serpapi":
return _search_serpapi(args)
else:
raise ValueError(f"--mode 必须是 api|browser|serpapi，当前 {args.mode}")
def _search_api(args):
"""调 Google Maps Places API Text Search（用户需配 GOOGLE_MAPS_API_KEY 环境变量）。"""
api_key = os.environ.get("GOOGLE_MAPS_API_KEY")
if not api_key:
raise RuntimeError(
"mode=api 需要 GOOGLE_MAPS_API_KEY 环境变量。"
"如果你没有 key，请用 --mode serpapi（用平台 serpapi.engine，无需自备 key）。"
)
import urllib.request
query = args.query or f"{args.category} {args.city}"
url = (
"https://maps.googleapis.com/maps/api/place/textsearch/json"
f"?query={query}&key={api_key}®ion=us"
)
req = urllib.request.Request(url)
with urllib.request.urlopen(req, timeout=30) as r:
data = json.loads(r.read())
if data.get("status") != "OK":
raise RuntimeError(f"Google Maps API 错误: {data.get('status')} {data.get('error_message')}")
places = []
for i, r in enumerate(data.get("results", []), 1):
places.append({
"id": r.get("place_id") or f"A{i:03d}",
"name": r.get("name", ""),
"category": args.category or "unknown",
"address": r.get("formatted_address", ""),
"phone": "", # Places API textsearch 不返回电话，需另调 place details
"website": "",
"rating": r.get("rating", 0),
"lat": r.get("geometry", {}).get("location", {}).get("lat"),
"lng": r.get("geometry", {}).get("location", {}).get("lng"),
})
if args.limit:
places = places[: args.limit]
result = {
"ok": True,
"source": "google_maps_places_api",
"mode": "api",
"query": query,
"returned": len(places),
"places": places,
"next_step": "phone 和 website 字段为空——需要再调 place details 拿全",
}
if args.output:
Path(args.output).write_text(json.dumps(result, ensure_ascii=False, indent=2), encoding="utf-8")
result["output"] = args.output
return result
def _search_browser(args):
"""打开 Chrome（playwright）+ 全自动抓 POI 数据。用户零介入。
流程（headless=True 时无窗口，agent 跑）：
1. 启动 chromium（默认 headless，用户不可见）
2. 导航到 https://www.google.com/maps/search/{query}
3. 等列表加载完成（智能等直到结果稳定）
4. 自动滚动触发懒加载（按用户 --limit 估算滚动次数）
5. 抓所有 POI 卡片
6. 关闭浏览器，输出 JSON
失败 fallback：检测到 CAPTCHA 时返回 needs_user_help=True，让用户选 SerpApi / Outscraper。
警告：headless 模式 Google Maps 反爬更严，命中率约 50-70%。如频繁被拦，
建议切到 mode=api（Google Maps Places API 需 key）或接 SerpApi。
"""
try:
from playwright.sync_api import sync_playwright, TimeoutError as PWTimeoutError
except ImportError:
raise RuntimeError(
"browser 模式需要 playwright：pip install playwright && playwright install chromium"
)
query = args.query or f"{args.category} {args.city}"
headless = getattr(args, "headless", True)
headful = not headless # 反过来，方便读
print(f"▶ [agent 自动] {'headless' if headless else 'headed'} 模式启动 Chrome")
print(f" 查询: {query}")
print(f" 限制: {args.limit or 10} 个结果")
print(f" 💡 用户无需操作——skill 自动滚动、自动抓取")
places = []
captcha_detected = False
try:
with sync_playwright() as p:
# headless=True 不开窗口；False 开窗口（调试用）
launch_args = ["--disable-blink-features=AutomationControlled"] if headless else []
browser = p.chromium.launch(headless=headless, args=launch_args)
# 用真实 UA，避免被识别为机器人
ctx = browser.new_context(
viewport={"width": 1280, "height": 800},
user_agent=(
"Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
"AppleWebKit/537.36 (KHTML, like Gecko) "
"Chrome/120.0.0.0 Safari/537.36"
),
locale="zh-CN",
)
page = ctx.new_page()
print(f" ⏳ 导航 Google Maps ...")
page.goto(
f"https://www.google.com/maps/search/{query.replace(' ', '+')}",
wait_until="domcontentloaded",
timeout=60000,
)
# 智能等列表加载（最多等 15 秒）
print(f" ⏳ 等结果列表加载 ...")
for attempt in range(15):
page.wait_for_timeout(1000)
cards = page.query_selector_all('[role="article"]')
if cards and len(cards) >= 3:
print(f" ✓ 检测到 {len(cards)} 张卡片（{attempt + 1}s 后）")
break
# 检测 CAPTCHA
if page.query_selector('iframe[src*="recaptcha"]') or "sorry" in page.url.lower():
captcha_detected = True
break
if captcha_detected:
print(f"⚠️ 检测到 CAPTCHA 反爬拦截")
browser.close()
return _search_browser_captcha_fallback(args, query, "Google Maps 检测到 CAPTCHA")
# 自动滚动触发懒加载（按 limit 估算：每屏约 5 个）
target_cards = max(args.limit or 10, 5)
scroll_rounds = (target_cards // 5) + 2
print(f" ⏳ 自动滚动 {scroll_rounds} 次以加载更多结果 ...")
for i in range(scroll_rounds):
page.mouse.wheel(0, 2000)
page.wait_for_timeout(800)
cur = len(page.query_selector_all('[role="article"]'))
print(f" 滚动 {i + 1}/{scroll_rounds}: 当前 {cur} 张")
# 抓所有 POI 卡片
cards = page.query_selector_all('[role="article"]')
if not cards:
cards = page.query_selector_all(".Nv2PK")
if not cards:
cards = page.query_selector_all(".qBF1Pd")
print(f" ✓ 共抓到 {len(cards)} 张卡片")
for i, card in enumerate(cards, 1):
try:
text = card.inner_text(timeout=3000)
except Exception:
continue
lines = [l.strip() for l in text.splitlines() if l.strip()]
if not lines:
continue
name = lines[0]
address = lines[1] if len(lines) > 1 else ""
rating = 0
for l in lines[1:5]:
m = re.search(r"(\d\.\d)\s*★?", l)
if m:
rating = float(m.group(1))
break
# 尝试提取域名（卡片里可能带）
domain = ""
m = re.search(r"(https?://[^/\s]+)", text)
if m:
domain = m.group(1).replace("https://", "").replace("http://", "")
# 尝试提取电话
phone = ""
m = re.search(r"(\+?\d[\d\s\-\(\)]{6,}\d)", text)
if m:
phone = m.group(1).strip()
places.append({
"id": f"B{i:03d}",
"name": name,
"category": args.category or "unknown",
"address": address,
"phone": phone,
"website": domain,
"domain": domain,
"rating": rating,
"lat": None,
"lng": None,
})
browser.close()
except PWTimeoutError as e:
print(f"⚠️ 浏览器超时: {e}")
return _search_browser_captcha_fallback(args, query, f"浏览器超时（{e}）")
except Exception as e:
print(f"⚠️ browser 模式错误: {e}")
return _search_browser_captcha_fallback(args, query, str(e))
if args.limit:
places = places[: args.limit]
result = {
"ok": True,
"source": "google_maps_browser",
"mode": "browser",
"headless": headless,
"query": query,
"returned": len(places),
"places": places,
"note": "phone/website 字段可能不完整——Google Maps 列表卡片只显示部分信息",
}
if args.output:
Path(args.output).write_text(json.dumps(result, ensure_ascii=False, indent=2), encoding="utf-8")
result["output"] = args.output
return result
def _search_serpapi(args):
"""通过 AIMS 平台的 serpapi.engine 一级工具调 Google Maps（无需自备 SERPAPI_KEY）。
平台价格：¥0.21/次（仅上游成功时扣费）
返回数据：local_results 数组，含 title/address/phone/website/rating/gps/place_id
注意：现在不再需要 SERPAPI_KEY 环境变量——平台代理调用 + 扣费
"""
query = args.query or f"{args.category} {args.city}"
print(f"▶ [平台 serpapi.engine] 搜索: {query}")
print(f" 限制: {args.limit or 10} 个结果")
print(f" 价格: ¥0.21/次（平台计费，仅上游成功时扣费）")
# 调平台 serpapi.engine（google_maps engine = POI 列表）
# 平台返回：平铺的 SerpApi 响应 + _aimschina 元数据（无 data 包裹层）
resp = _rpc_call(
"serpapi.engine",
{"engine": "google_maps", "q": query, "num": str(args.limit or 10),
"hl": "zh-CN", "gl": "cn"},
confirm=getattr(args, "approved", False),
)
local_results = resp.get("local_results") or []
if not local_results:
return {
"ok": True,
"source": "platform_serpapi_google_maps",
"mode": "serpapi",
"query": query,
"returned": 0,
"places": [],
"note": "SerpApi 返回空结果——换个 query 试试",
}
places = []
for i, r in enumerate(local_results, 1):
gps = r.get("gps_coordinates", {})
places.append({
"id": r.get("place_id") or f"S{i:03d}",
"name": r.get("title", ""),
"category": args.category or "unknown",
"address": r.get("address", ""),
"phone": r.get("phone", ""),
"website": r.get("website", ""),
"domain": _extract_domain({"website": r.get("website", "")}),
"rating": r.get("rating", 0),
"reviews": r.get("reviews", 0),
"lat": gps.get("latitude"),
"lng": gps.get("longitude"),
"place_id": r.get("place_id"),
})
if args.limit:
places = places[: args.limit]
result = {
"ok": True,
"source": "platform_serpapi_google_maps",
"mode": "serpapi",
"query": query,
"returned": len(places),
"places": places,
"platform_cost_yuan": resp.get("_aimschina", {}).get("cost_yuan", "0"),
"platform_resource_id": resp.get("_aimschina", {}).get("resource_id", "serpapi/google_maps"),
"note": "phone/website 字段完整；可直接进 verify 步骤",
}
if args.output:
Path(args.output).write_text(json.dumps(result, ensure_ascii=False, indent=2), encoding="utf-8")
result["output"] = args.output
return result
# 平台返回 {"data": {...SerpApi 完整响应...}}
data = resp.get("data", {})
if not data:
return {
"ok": True,
"source": "platform_serpapi_google_maps",
"mode": "serpapi",
"query": query,
"returned": 0,
"places": [],
"note": "SerpApi 返回空结果——换个 query 试试",
}
local_results = data.get("local_results") or []
places = []
for i, r in enumerate(local_results, 1):
gps = r.get("gps_coordinates", {})
places.append({
"id": r.get("place_id") or f"S{i:03d}",
"name": r.get("title", ""),
"category": args.category or "unknown",
"address": r.get("address", ""),
"phone": r.get("phone", ""),
"website": r.get("website", ""),
"domain": _extract_domain({"website": r.get("website", "")}),
"rating": r.get("rating", 0),
"reviews": r.get("reviews", 0),
"lat": gps.get("latitude"),
"lng": gps.get("longitude"),
"place_id": r.get("place_id"),
})
if args.limit:
places = places[: args.limit]
result = {
"ok": True,
"source": "platform_serpapi_google_maps",
"mode": "serpapi",
"query": query,
"returned": len(places),
"places": places,
"platform_cost_yuan": resp.get("_aimschina", {}).get("cost_yuan", "0"),
"platform_resource_id": resp.get("_aimschina", {}).get("resource_id", "serpapi/google"),
"note": "phone/website 字段完整；可直接进 verify 步骤",
}
if args.output:
Path(args.output).write_text(json.dumps(result, ensure_ascii=False, indent=2), encoding="utf-8")
result["output"] = args.output
return result
def _rpc_call(tool_name, args, confirm=False):
"""调 AIMS 平台一级 MCP 工具（自命名空间工具 + aims.*）。
confirm: 用户已确认费用时才传 True（对应 CLI 的 --approved）。默认 False —
平台会返回 -32006 requires_confirmation 且**不扣费**，本函数把它抛成
NeedsConfirmation（带真实报价），调用方据此让用户确认。
返回内部数据。兼容性：
- hunter.* 系列：返 {"data": {...}, "meta": ..., "_aimschina": ...}（data 包裹）
- serpapi.* 系列：返平铺 {...search_metadata..., "local_results": ...}
这里保留原始结构，由调用方按需取 resp["data"] 或 resp。
抛出 RuntimeError 当 isError=True。
"""
api_key = os.environ.get("AIMS_API_KEY")
if not api_key:
raise RuntimeError("缺少 AIMS_API_KEY 环境变量")
import urllib.request
rpc_args = dict(args or {})
if confirm:
rpc_args["confirm"] = True
body = json.dumps({
"jsonrpc": "2.0", "id": 11, "method": "tools/call",
"params": {"name": tool_name, "arguments": rpc_args},
}).encode()
headers = {"Content-Type": "application/json", "Authorization": f"Bearer {api_key}"}
req = urllib.request.Request(AIMS_URL, data=body, method="POST", headers=headers)
with urllib.request.urlopen(req, timeout=120) as r:
resp = json.loads(r.read())
is_err = resp.get("result", {}).get("isError")
text = resp.get("result", {}).get("content", [{}])[0].get("text", "{}")
parsed = json.loads(text)
if is_err:
# -32006 → NeedsConfirmation（结构化报价）；其它错误照旧 RuntimeError
raise_if_needs_confirmation(tool_name.replace(".", "/"), parsed)
raise RuntimeError(f"{tool_name} 错误: {parsed}")
return parsed
def _unwrap(resp):
"""兼容工具返回：hunter 系列有 data 包裹，serpapi 系列平铺。
返回内部数据 dict；如果有 data 包裹则剥掉；否则返回原 dict。
"""
if isinstance(resp, dict) and isinstance(resp.get("data"), dict):
return resp["data"]
return resp
def _search_browser_captcha_fallback(args, query, reason):
"""browser 模式失败时的 fallback——返回 needs_user_help 让用户选下一步。"""
return {
"ok": False,
"needs_user_help": True,
"error": f"Google Maps 自动化失败: {reason}",
"fallback_options": {
"option_a_use_serpapi": {
"name": "切到 SerpApi（推荐）",
"description": "用平台 serpapi.engine（Google Maps 接口），0.21 元/次，无需浏览器，无需 CAPTCHA 风险",
"requires": "AIMS_API_KEY 环境变量（已是平台用户，无额外 key）",
"next_step": "用 --mode serpapi",
},
"option_b_use_api": {
"name": "切到 Google Maps Places API",
"description": "官方 API，需 GCP 账号 + 启用 Maps JavaScript API + 配 key",
"requires": "GOOGLE_MAPS_API_KEY 环境变量",
"next_step": "用 --mode api",
},
"option_c_rewrite_query": {
"name": "换 query 重试（英文）",
"description": "Google Maps 对英文 query 命中率更高；改成 'coffee shop Hangzhou' 而非 '杭州的咖啡店'",
"next_step": "用 --mode browser --query 'coffee shop Hangzhou'",
},
"option_d_headed_debug": {
"name": "用 headed Chrome 试一次",
"description": "弹窗让你看到具体哪里失败（CAPTCHA 按钮/地区限制），适合调试",
"next_step": "加 --headed 重跑",
},
},
}
def _llm_extract_search_params(description):
"""用 AIMS 平台 qwen-flash 从自然语言抽取 city/category/limit。
输入示例："我要杭州的咖啡店老板邮箱，10个以内"
输出示例：{"city": "杭州", "category": "coffee", "limit": 10, "intent": "找老板邮箱"}
"""
if not _AIMS_OK:
raise RuntimeError(
"自然语言抽取需要 AIMS helper。检查 office-skills/_lib/aims_chat.py 是否存在"
)
prompt = (
"你是一个外贸客户搜索助手。从用户的中文描述里抽取搜索参数。\n"
"严格返回 JSON（不要解释、不要 Markdown 代码块）：\n"
'{"city": "城市（中文，如 杭州；不知道就 null）", '
'"category": "类别（coffee/restaurant/fitness 三选一；不知道就 coffee）", '
'"limit": 数字（用户说要多少；不知道就 10）, '
'"intent": "用户意图的简短总结"}\n'
f"--- 用户描述 ---\n{description}"
)
try:
return chat_text_json(prompt, api_key=os.environ.get("AIMS_API_KEY"), no_rag=True)
except Exception as e:
print(f"⚠️ LLM 抽取失败（{e}），继续用默认参数", file=sys.stderr)
return None
# ---------- 第二步：confirm ----------
def cmd_confirm(args):
"""用户从 candidates.json 里挑选要验证邮箱的 id，写入 selected.json。"""
in_path = Path(args.input)
if not in_path.exists():
raise FileNotFoundError(f"--input {args.input} 不存在，请先跑 search")
data = json.loads(in_path.read_text(encoding="utf-8"))
places = data.get("places", [])
# 解析 --select，支持 3 种写法
selected_ids = []
if args.select:
sel = args.select
if sel == "all":
selected_ids = [p["id"] for p in places]
elif sel.startswith("first-"):
try:
n = int(sel.split("-")[1])
selected_ids = [p["id"] for p in places[:n]]
except Exception:
raise ValueError(f"--select first-N 格式错（应为 first-3）")
elif Path(sel).exists():
ids_text = Path(sel).read_text(encoding="utf-8").strip()
selected_ids = [x.strip() for x in re.split(r"[,\n\r\s]+", ids_text) if x.strip()]
else:
selected_ids = [x.strip() for x in sel.split(",") if x.strip()]
if not selected_ids:
raise ValueError("--select 不能为空（用 id 列表、first-N、all，或文件路径）")
selected_places = [p for p in places if p["id"] in selected_ids]
missing = [i for i in selected_ids if i not in {p["id"] for p in selected_places}]
if missing:
print(f"⚠️ 以下 id 在 search 结果中找不到：{missing}", file=sys.stderr)
# 估算邮箱数（按 Hunter 经验值 1.5 个/域名）
estimated_found = int(len(selected_places) * 1.5)
# 动态算成本（按数据源 + 数量）
search_mode = data.get("mode", "serpapi")
used_desc = data.get("llm_extracted_from_description") is not None
cost = calc_cost(
mode=search_mode,
search_count=len(places),
verify_count=len(selected_places),
used_description=used_desc,
)
cost["estimated_emails_found"] = estimated_found
# budget 上限检查
budget = getattr(args, "budget", None)
budget_exceeded = False
if budget is not None and cost["total_yuan"] > budget:
budget_exceeded = True
print(
f"⚠️ 估算 ¥{cost['total_yuan']} 超过 --budget ¥{budget}；请缩小 --select 范围",
file=sys.stderr,
)
result = {
"ok": not budget_exceeded,
"source": "user_confirmed",
"selected": selected_places,
"selected_count": len(selected_places),
"estimated_emails_found": estimated_found,
"cost_estimate": cost,
"budget_yuan": budget,
"budget_exceeded": budget_exceeded,
"next_step": (
"运行 verify 子命令前，请确认 selected_count 和 cost_estimate 与你的预期一致；"
"不满意可调整 --select 或 --budget 重跑 confirm"
if not budget_exceeded else
f"❌ 估算 ¥{cost['total_yuan']} 超过 budget ¥{budget}，不继续 verify。请减少 selected_count 或加 budget。"
),
}
if args.output:
out_path = Path(args.output)
out_path.parent.mkdir(parents=True, exist_ok=True)
out_path.write_text(json.dumps(result, ensure_ascii=False, indent=2), encoding="utf-8")
result["output"] = args.output
return result
# ---------- 第三步：verify ----------
def cmd_verify(args):
"""对 selected.json 里的每个商家，调 AIMS 平台 hunter.domain-search 拿邮箱，
再调 hunter.email-verifier 验证邮箱有效性。
平台价格：domain-search ¥0.204/次 + email-verifier ¥0.10/次
"""
in_path = Path(args.input)
if not in_path.exists():
raise FileNotFoundError(f"--input {args.input} 不存在，请先跑 confirm")
data = json.loads(in_path.read_text(encoding="utf-8"))
selected = data.get("selected", [])
leads = []
no_email = []
cost_breakdown = {"domain_search": 0.0, "email_verifier": 0.0, "calls": 0}
for place in selected:
domain = _extract_domain(place)
if not domain:
no_email.append({"place_id": place["id"], "name": place["name"], "reason": "无 website"})
continue
# 1) 调 hunter.domain-search 拿邮箱列表
emails = []
try:
resp = _rpc_call("hunter.domain-search", {"domain": domain, "limit": "5"},
confirm=getattr(args, "approved", False))
inner = _unwrap(resp)
cost_breakdown["domain_search"] += float(inner.get("_aimschina", {}).get("cost_yuan", "0") or 0)
cost_breakdown["calls"] += 1
# hunter 内部数据（已剥 data 包裹）：含 emails 数组
emails = inner.get("emails", []) or []
except Exception as e:
no_email.append({"place_id": place["id"], "name": place["name"],
"domain": domain, "reason": f"hunter 错误: {e}"})
continue
if not emails:
no_email.append({"place_id": place["id"], "name": place["name"],
"domain": domain, "reason": "Hunter 无数据"})
continue
# 2) 对每个邮箱调 hunter.email-verifier 验证
for e in emails:
email = e.get("value", "")
verified_status = "unknown"
if email:
try:
v_resp = _rpc_call("hunter.email-verifier", {"email": email},
confirm=getattr(args, "approved", False))
v_inner = _unwrap(v_resp)
cost_breakdown["email_verifier"] += float(v_inner.get("_aimschina", {}).get("cost_yuan", "0") or 0)
cost_breakdown["calls"] += 1
# Hunter verifier status: "valid" / "invalid" / "accept_all" / "webmail" / "disposable" / "unknown"
verified_status = v_inner.get("status", "unknown")
except Exception:
verified_status = "verify_failed"
leads.append({
"place_id": place["id"],
"company_name": place["name"],
"category": place["category"],
"address": place["address"],
"phone": place["phone"],
"website": place["website"],
"domain": domain,
"email": email,
"position": e.get("position", ""),
"department": e.get("department", ""),
"seniority": e.get("seniority", ""),
"contact_name": f"{e.get('first_name', '')}{e.get('last_name', '')}".strip(),
"confidence": e.get("confidence", 0),
"hunter_type": e.get("type", ""),
"verified_status": verified_status,
"verified": verified_status == "valid",
"source": "platform_hunter", # 全部接真接口
})
# 输出 xlsx
xlsx_path = Path(args.output)
xlsx_path.parent.mkdir(parents=True, exist_ok=True)
try:
from openpyxl import Workbook
from openpyxl.styles import Font, PatternFill, Alignment
wb = Workbook()
ws = wb.active
ws.title = "Leads"
headers = ["公司名", "类别", "地址", "电话", "网站", "域名", "邮箱",
"联系人", "职位", "部门", "置信度", "验证状态", "数据源"]
ws.append(headers)
for c in ws[1]:
c.font = Font(bold=True)
c.fill = PatternFill("solid", fgColor="DDDDDD")
for lead in leads:
status_text = {
"valid": "✓ valid",
"invalid": "✗ invalid",
"accept_all": "? accept_all",
"webmail": "? webmail",
"disposable": "✗ disposable",
"unknown": "? unknown",
"verify_failed": "✗ 验证失败",
}.get(lead["verified_status"], lead["verified_status"])
ws.append([
lead["company_name"], lead["category"], lead["address"], lead["phone"],
lead["website"], lead["domain"], lead["email"], lead["contact_name"],
lead["position"], lead.get("department", ""), lead["confidence"],
status_text, lead["source"],
])
# 未找到邮箱的单独 sheet
if no_email:
ws2 = wb.create_sheet("未找到邮箱")
ws2.append(["公司名", "地址", "原因"])
for c in ws2[1]:
c.font = Font(bold=True)
for n in no_email:
ws2.append([n["name"], n.get("address") or n.get("domain") or "", n["reason"]])
wb.save(str(xlsx_path))
except ImportError:
# 没 openpyxl 降级到 json
xlsx_path = xlsx_path.with_suffix(".json")
xlsx_path.write_text(json.dumps({"leads": leads, "no_email": no_email},
ensure_ascii=False, indent=2), encoding="utf-8")
return {
"ok": True,
"warning": "openpyxl 未安装，已降级输出 JSON（请 pip install openpyxl）",
"output": str(xlsx_path),
"lead_count": len(leads),
"no_email_count": len(no_email),
}
return {
"ok": True,
"output": str(xlsx_path),
"lead_count": len(leads),
"no_email_count": len(no_email),
"verified_count": sum(1 for l in leads if l["verified"]),
"cost_summary": {
"calls": cost_breakdown["calls"],
"domain_search_yuan": round(cost_breakdown["domain_search"], 4),
"email_verifier_yuan": round(cost_breakdown["email_verifier"], 4),
"total_yuan": round(cost_breakdown["domain_search"] + cost_breakdown["email_verifier"], 4),
"source": "platform_hunter",
},
}
# ---------- 报价：estimate（免费，不调上游） ----------
def cmd_estimate(args):
"""只报费用：调**免费**的 aims.list_functional_apis 读平台实时单价。
覆盖三步全链路，0 上游花费、0 扣费：
search leg: serpapi/google_maps × 1
verify leg: hunter/domain-search × 选中数
+ hunter/email-verifier × 预估邮箱数
"""
mode = getattr(args, "mode", "serpapi")
limit = getattr(args, "limit", None) or 10
sel = getattr(args, "select", None) or "first-3"
verify_count = _parse_select_count(sel, limit)
api_key = getattr(args, "api_key", None)
legs = {}
if mode == "serpapi":
legs["search_serpapi_google_maps"] = estimate_cost(
"serpapi/google_maps", units=1, api_key=api_key)
elif mode == "browser":
legs["search_browser"] = {
"ok": True, "resource_id": "google/maps (本地浏览器)", "units": 1,
"unit_cost_yuan": 0.0, "total_yuan": 0.0, "cost_basis": "local",
"upstream_called": False, "note": "本地 playwright，无平台费用（有反爬风险）",
}
else:
legs["search_google_places_api"] = {
"ok": True, "resource_id": "google/maps (自备 key)", "units": 1,
"unit_cost_yuan": 0.0, "total_yuan": 0.0, "cost_basis": "user_key",
"upstream_called": False, "note": "走你自己的 GOOGLE_MAPS_API_KEY，平台不扣费",
}
emails_est = verify_count * PRICING["verify"]["estimated_emails_per_domain"]
legs["verify_domain_search"] = estimate_cost(
"hunter/domain-search", units=verify_count, api_key=api_key)
legs["verify_email_verifier"] = estimate_cost(
"hunter/email-verifier", units=emails_est, api_key=api_key)
unknown = [k for k, v in legs.items() if not v.get("ok")]
total = round(sum(float(v.get("total_yuan") or 0) for v in legs.values()), 4)
return {
"ok": not unknown,
"upstream_called": False,
"mode": mode,
"limit": limit,
"select": sel,
"verify_count": verify_count,
"estimated_emails": emails_est,
"legs": legs,
"total_yuan": total,
"pricing_source": "平台实时单价 (aims.list_functional_apis)，非硬编码价表",
"unknown_legs": unknown,
"note": ("本报价未调用上游、未扣费。确认后加 --approved 真实执行；"
"不带 --approved 时平台返回 -32006 且不扣费。"),
}
def _parse_select_count(sel, limit):
"""把 --select 的各种写法折算成"要验证几个商家"。"""
if not sel:
return min(3, limit)
if sel == "all":
return limit
if isinstance(sel, str) and sel.startswith("first-"):
try:
return min(int(sel.split("-")[1]), limit)
except (ValueError, IndexError):
return min(3, limit)
if isinstance(sel, str):
try:
if Path(sel).exists():
txt = Path(sel).read_text(encoding="utf-8").strip()
return len([x for x in re.split(r"[,\n\r\s]+", txt) if x.strip()])
except OSError:
pass
if "," in sel:
return len([x for x in sel.split(",") if x.strip()])
return min(3, limit)
# ---------- 第四步：run（一条命令） ----------
def cmd_run(args):
"""一键：search → confirm → verify。适合 agent 调用。
三种入口：
1. --interactive：阻塞问询用户（city/category/budget），CLI 模式
2. --stdin：从 stdin 读 JSON 一次性给参数（agent 自动化）
3. 默认参数全给：直接跑（agent 已知参数时用）
"""
import tempfile
# 1) 处理参数输入模式
if args.interactive and args.stdin:
raise ValueError("--interactive 和 --stdin 不能同时用")
if args.interactive:
params = _interactive_ask()
elif args.stdin:
params = _read_stdin_params()
else:
params = {}
# 合并参数：命令行 > stdin/interactive > default
args.city = params.get("city") or args.city
args.category = params.get("category") or args.category
args.limit = params.get("limit") or args.limit
args.budget = params.get("budget") or args.budget
args.select = params.get("select") or args.select
# 必填检查
if not (args.city or args.query or args.description):
raise ValueError("需要 city / query / description 至少一个——用 --interactive 启用问询")
if not (args.category or args.query or args.description):
raise ValueError("需要 category / query / description 至少一个——用 --interactive 启用问询")
tmpdir = Path(tempfile.mkdtemp(prefix="lead_miner_"))
cand_path = tmpdir / "candidates.json"
sel_path = tmpdir / "selected.json"
# search（带 mode / description / query / headed 透传）
search_args = argparse.Namespace(
mode=args.mode, city=args.city, category=args.category, limit=args.limit,
query=args.query, description=args.description,
headed=getattr(args, "headed", False),
approved=getattr(args, "approved", False),
output=str(cand_path),
)
search_result = cmd_search(search_args)
# confirm（默认 first-N 或 all）
sel = args.select or "all"
confirm_args = argparse.Namespace(
input=str(cand_path), select=sel, output=str(sel_path),
budget=getattr(args, "budget", None),
)
confirm_result = cmd_confirm(confirm_args)
# verify（如果 budget 超了 confirm 返回 ok=false，跳过 verify）
out_xlsx = args.output or "leads.xlsx"
if confirm_result.get("ok") is False and confirm_result.get("budget_exceeded"):
print(f"❌ budget 超过，不执行 verify")
verify_result = {
"ok": False,
"skipped": True,
"reason": "confirm 阶段 budget 超限",
}
else:
verify_args = argparse.Namespace(input=str(sel_path), output=out_xlsx,
approved=getattr(args, "approved", False))
verify_result = cmd_verify(verify_args)
return {
"ok": True,
"search": {
"total_matched": search_result.get("total_matched", search_result.get("returned", 0)),
"returned": search_result["returned"],
},
"confirm": {
"selected_count": confirm_result["selected_count"],
"estimated_emails": confirm_result["estimated_emails_found"],
"estimated_cost": confirm_result["cost_estimate"],
},
"verify": verify_result,
"tmp_dir": str(tmpdir),
}
# ---------- 工具函数 ----------
def _extract_domain(place):
"""从 website 提取域名（去协议和路径）。"""
url = place.get("website") or ""
m = re.search(r"https?://([^/]+)", url)
return m.group(1) if m else None
# ---------- 定价（按数据源 + 数量动态算） ----------
# AIMS 平台官方价格（实测 2026-09-17，详见 install.md §17.4）
PRICING = {
"search": {
"serpapi": 0.21, # 平台 serpapi.engine ¥0.21/次（仅上游成功时扣费）
"api": 0.23, # Google Maps Places $32/1000 ≈ ¥0.23/次（需自备 KEY）
"browser": 0.0, # 本地浏览器，无平台成本（但有反爬风险）
},
"verify": {
"domain_search_unit": 0.204, # hunter.domain-search ¥0.204/次
"email_verifier_unit": 0.10, # hunter.email-verifier ¥0.10/次
"estimated_emails_per_domain": 1.5, # Hunter 平均每个域名找到 1.5 个邮箱
},
"platform_chat": 0.02, # AIMS qwen-flash 1 次约 0.02 元
}
def _estimate_total_cost(limit, verify_count, mode="serpapi"):
"""估算用户输入参数的预估总成本（用于交互问询时给预算建议）。"""
if not limit or not verify_count:
return 0
# 用 calc_cost 的同样逻辑
cost = calc_cost(mode=mode, search_count=limit, verify_count=verify_count, used_description=False)
return cost["total_yuan"]
def _interactive_ask():
"""阻塞问询用户：城市 / 类目 / 数量 / 预算。返回 dict。
流程：
1. 问"找哪个城市？"——接受默认（按 Enter）或输入
2. 问"找哪些类目？"——列出 3 选 1 + 多选
3. 问"要几个候选？"——10/20/50/100
4. 显示预估成本，问"预算上限？"——给建议预算
5. 用户确认后跑
"""
print("=" * 60)
print("📍 lead-miner 配置（必填 4 项）")
print("=" * 60)
# 1) 城市
cities = ["杭州", "上海", "北京", "深圳", "广州", "成都", "武汉", "其他"]
print("\n[1/4] 找哪个城市？")
for i, c in enumerate(cities, 1):
print(f" {i}. {c}")
while True:
sel = input(" 输入数字 (默认 杭州 = 1): ").strip() or "1"
try:
idx = int(sel) - 1
if 0 <= idx < len(cities):
if cities[idx] == "其他":
city = input(" 输入城市名: ").strip()
else:
city = cities[idx]
break
print(f" ❌ 范围 1-{len(cities)}")
except ValueError:
print(f" ❌ 请输入数字 1-{len(cities)}")
# 2) 类目
categories = [
("coffee", "咖啡店"),
("restaurant", "餐厅"),
("fitness", "健身房"),
]
print("\n[2/4] 找哪些类目？（可多选，逗号分隔，如 1,2）")
for i, (k, n) in enumerate(categories, 1):
print(f" {i}. {n} ({k})")
while True:
sel = input(" 输入数字 (默认 1 = 咖啡): ").strip() or "1"
try:
ids = [int(x.strip()) - 1 for x in sel.split(",")]
if all(0 <= i < len(categories) for i in ids):
cats = [categories[i][0] for i in ids]
break
print(f" ❌ 范围 1-{len(categories)}")
except ValueError:
print(f" ❌ 请输入数字")
# 3) 数量
print("\n[3/4] 要几个候选？")
print(" 1. 10 个（试水 ¥1.5）")
print(" 2. 30 个（小批量 ¥4.5）")
print(" 3. 50 个（中批量 ¥7.5）")
print(" 4. 100 个（大批量 ¥15）")
limit_map = {"1": 10, "2": 30, "3": 50, "4": 100}
while True:
sel = input(" 输入数字 (默认 1 = 10): ").strip() or "1"
if sel in limit_map:
limit = limit_map[sel]
break
print(" ❌ 1-4")
# 4) 预算（基于预估给建议）
estimated = _estimate_total_cost(limit, limit, mode="serpapi") # 假设全验证
print(f"\n[4/4] 预算上限？预估成本 ¥{estimated:.2f}（如果你 verify 全部）")
print(f" 建议：先 ¥{estimated + 0.5:.2f}（多给点 buffer）")
budget_str = input(f" 输入金额 (默认 ¥{estimated + 0.5:.2f}, 0 = 不限): ").strip()
try:
budget = float(budget_str) if budget_str else estimated + 0.5
except ValueError:
budget = estimated + 0.5
# 5) 确认
print("\n" + "=" * 60)
print("✅ 配置确认：")
print(f" 城市: {city}")
print(f" 类目: {', '.join(cats)}")
print(f" 候选: {limit} 个")
print(f" 预算: ¥{budget:.2f}")
print(f" 预估全验证成本: ¥{estimated:.2f}")
print("=" * 60)
while True:
ok = input("开始搜？[回车确认 / n 取消]: ").strip().lower()
if ok == "" or ok == "y":
break
if ok == "n":
raise RuntimeError("用户取消")
print(" ❌ 回车 = yes, n = no")
return {
"city": city,
"category": cats[0], # 主类目（多选时取第一个；用户后续可 --select 调整）
"categories": cats,
"limit": limit,
"budget": budget,
}
def _read_stdin_params():
"""从 stdin 读 JSON 一次性给参数（agent 自动化用）。
期望格式：
{"city": "杭州", "category": "coffee", "limit": 10, "budget": 1.5}
"""
try:
raw = sys.stdin.read()
if not raw.strip():
return {}
return json.loads(raw)
except json.JSONDecodeError as e:
raise ValueError(f"--stdin 模式下 stdin 不是合法 JSON: {e}")
def calc_cost(mode: str, search_count: int, verify_count: int,
used_description: bool = False, search_calls: int = 1) -> dict:
"""根据数据源 + 数量算总价（元）。全接真接口（无 mock）。
Args:
mode: search 数据源（serpapi/api/browser）
search_count: search 返回的候选数（仅用于显示；不参与计费）
verify_count: confirm 选中的数量
used_description: 是否用了 --description 触发 LLM 抽取
search_calls: 实际 search 调用次数（默认 1；多批搜时 > 1）
Returns:
dict 含 search_cost_yuan / verify_cost_yuan / platform_cost_yuan / total_yuan / breakdown
"""
# search 费用按"调用次数"算（serpapi 每次调 Google Maps 计一次费，与 limit 无关）
search_unit = PRICING["search"].get(mode, 0)
search_cost = search_calls * search_unit
# Verify 成本（domain_search + email_verifier 估算）
domain_cost = verify_count * PRICING["verify"]["domain_search_unit"]
emails_count = verify_count * PRICING["verify"]["estimated_emails_per_domain"]
verifier_cost = emails_count * PRICING["verify"]["email_verifier_unit"]
verify_cost = domain_cost + verifier_cost
# 平台 LLM 调用次数（仅 --description 触发时算 1 次；verify 阶段不再计）
platform_calls = 1 if used_description else 0
platform_cost = platform_calls * PRICING["platform_chat"]
total = search_cost + verify_cost + platform_cost
return {
"mode": mode,
"search_count": search_count,
"search_calls": search_calls,
"verify_count": verify_count,
"used_description": used_description,
"platform_llm_calls": platform_calls,
"search_unit_yuan": search_unit,
"verify_cost_yuan": round(verify_cost, 4),
"platform_cost_yuan": round(platform_cost, 4),
"total_yuan": round(total, 4),
"breakdown": (
f"search({search_calls}次×{search_unit}={round(search_cost, 3)}) "
f"+ verify(domain_search({verify_count}×0.204={round(domain_cost, 3)}) + "
f"verifier({emails_count}×0.10={round(verifier_cost, 3)})={round(verify_cost, 3)}) "
f"+ llm({platform_calls}×{PRICING['platform_chat']}={round(platform_cost, 3)}) "
f"= ¥{round(total, 3)}"
),
}
# ---------- CLI ----------
def build_parser():
ap = argparse.ArgumentParser(description="外贸搜客：Google Maps → 用户确认 → Hunter.io 验证邮箱")
ap.add_argument("--data-dir", default=str(_DATA), help="兼容字段；当前已无 mock 数据，仅留作兼容")
sub = ap.add_subparsers(dest="command", required=True)
# search
p = sub.add_parser("search", help="第一步：搜客户（serpapi/api/browser 三模式）")
p.add_argument("--mode", default="serpapi", choices=["serpapi", "api", "browser"],
help="serpapi=平台 serpapi.engine（推荐生产，无需 key）；api=Google Places API（需自备 KEY）；browser=playwright（无需 key，可能反爬）")
p.add_argument("--city", default=None, help="城市（必填，除非 --query 或 --description）")
p.add_argument("--category", default=None, help="类别：coffee/restaurant/fitness")
p.add_argument("--limit", type=int, default=10, help="返回数量上限")
p.add_argument("--query", default="", help="自然语言查询（如 'coffee shop Hangzhou'）")
p.add_argument("--description", default="",
help="中文自然语言描述（如 '我要杭州的咖啡店'）；会自动让 LLM 抽取 city/category/limit")
p.add_argument("--headed", action="store_true",
help="browser 模式默认 headless（无窗口）；加 --headed 会弹窗口（调试用）")
p.add_argument("--output", help="输出 candidates.json")
p.add_argument("--approved", action="store_true",
help="用户已确认费用后才允许调用付费接口（search 会扣 serpapi 费）")
p.set_defaults(func=cmd_search)
# confirm
p = sub.add_parser("confirm", help="第二步：用户确认要验证的范围")
p.add_argument("--input", required=True, help="search 输出的 JSON")
p.add_argument("--select", required=True,
help="选哪些 id：逗号分隔（P001,P003）/ first-N（first-3）/ all / id 文件路径")
p.add_argument("--budget", type=float, default=None,
help="成本上限（元）；估算超过时 confirm 返回 ok=false，不继续 verify")
p.add_argument("--output", help="输出 selected.json")
p.set_defaults(func=cmd_confirm)
# verify
p = sub.add_parser("verify", help="第三步：对 selected 里的商家调平台 hunter 验证邮箱")
p.add_argument("--input", required=True, help="confirm 输出的 JSON")
p.add_argument("--output", default="leads.xlsx", help="输出 xlsx 文件")
p.add_argument("--approved", action="store_true",
help="用户已确认费用后才允许调用付费接口（verify 会扣 hunter 费）")
p.set_defaults(func=cmd_verify)
# run
p = sub.add_parser("run", help="一条命令跑完整三步（agent 用）")
p.add_argument("--mode", default="serpapi", choices=["serpapi", "api", "browser"])
p.add_argument("--city", default=None)
p.add_argument("--category", default=None)
p.add_argument("--limit", type=int, default=10)
p.add_argument("--query", default="")
p.add_argument("--description", default="",
help="自然语言描述（如 '我要杭州的咖啡店老板邮箱，10个'）；LLM 自动抽取 city/category/limit")
p.add_argument("--headed", action="store_true")
p.add_argument("--select", default="first-3", help="默认 first-N；推荐 first-3 试水")
p.add_argument("--budget", type=float, default=None, help="成本上限（元）")
p.add_argument("--interactive", action="store_true",
help="交互式问询（阻塞问 city/category/budget，再工作）")
p.add_argument("--stdin", action="store_true",
help="从 stdin 读 JSON 参数（agent 自动化用；与 --interactive 互斥）")
p.add_argument("--output", default="leads.xlsx")
p.add_argument("--approved", action="store_true",
help="用户已确认费用后才允许调用付费接口（run 会扣 serpapi + hunter 费）")
p.set_defaults(func=cmd_run)
# estimate（免费报价）
p = sub.add_parser("estimate", help="只报费用：不调上游、不扣费（免费）")
p.add_argument("--mode", default="serpapi", choices=["serpapi", "api", "browser"])
p.add_argument("--limit", type=int, default=10, help="预计候选数")
p.add_argument("--select", default="first-3", help="预计验证范围（first-N / all / id 列表）")
p.add_argument("--api-key", default=None)
p.set_defaults(func=cmd_estimate)
return ap
def main():
args = build_parser().parse_args()
# 允许 --data-dir 覆盖
global _DATA
if hasattr(args, "data_dir") and args.data_dir:
_DATA = Path(args.data_dir)
try:
result = args.func(args)
except NeedsConfirmation as e:
# 付费工具未确认 —— 打印结构化报价，退出码 3（区别于 1=真错误），
# 让 agent 能把 estimated_cost_yuan 展示给用户而不是当成失败重试。
print(json.dumps(e.to_dict(), ensure_ascii=False, indent=2))
sys.exit(3)
except Exception as e:
print(json.dumps({"ok": False, "error": str(e)}, ensure_ascii=False))
sys.exit(1)
print(json.dumps(result, ensure_ascii=False, indent=2))
if __name__ == "__main__":
main()<!-- ===== 以下为共享依赖 _lib（落盘为 _lib/<name>，与 scripts/ 同级上层） ===== -->
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""AIMS 平台 AI 调用公共 helper（6 个 skill 复用）。
能力：
- chat_text(prompt, model="qwen-flash") 纯文本对话
- chat_vision(image_path_or_url, prompt, ...) 图片+文本多模态
- chat_pdf_pages(pdf_path, prompt, zoom=2.0) 扫描版 PDF OCR
- chat_with_usage(...) 返回文本+用量
环境：依赖 AIMS_API_KEY 环境变量或 --api-key
"""
import argparse
import base64
import json
import os
import sys
import time
import urllib.request
from pathlib import Path
AIMS_URL = "https://aimsgateway.cn/mcp/"
TEXT_MODEL = "qwen-flash" # 通用文本（最便宜）
VISION_MODEL = "qwen-vl-ocr-latest" # 视觉/OCR
# ---------- 鉴权 & 协议 ----------
def _get_key(api_key=None):
key = api_key or os.environ.get("AIMS_API_KEY")
if not key:
raise RuntimeError(
"缺少 AIMS API Key：请设置环境变量 AIMS_API_KEY，或调用方传 --api-key"
)
return key
def _rpc(url, key, method, params, mid, tries=3, timeout=180):
body = json.dumps({"jsonrpc": "2.0", "id": mid, "method": method, "params": params}).encode()
headers = {"Content-Type": "application/json", "Authorization": f"Bearer {key}"}
last = None
for _ in range(tries):
try:
req = urllib.request.Request(url, data=body, method="POST", headers=headers)
with urllib.request.urlopen(req, timeout=timeout) as r:
return json.loads(r.read().decode())
except Exception as e: # noqa: BLE001
last = e
time.sleep(2)
raise RuntimeError(f"MCP 调用失败: {last}")
def _handshake(key):
_rpc(AIMS_URL, key, "initialize",
{"protocolVersion": "2025-03-26", "capabilities": {},
"clientInfo": {"name": "office-skills", "version": "1.0"}}, 1)
try:
_rpc(AIMS_URL, key, "notifications/initialized", {}, 2)
except Exception:
pass
# ---------- 图片编码 ----------
def _file_to_data_url(path):
p = Path(path)
if not p.exists():
raise FileNotFoundError(path)
b64 = base64.b64encode(p.read_bytes()).decode()
ext = p.suffix.lower()
mime = {
"png": "image/png", "jpg": "image/jpeg", "jpeg": "image/jpeg",
"gif": "image/gif", "webp": "image/webp", "bmp": "image/bmp",
}.get(ext, "image/png")
return f"data:{mime};base64,{b64}"
def _pdf_page_to_data_url(pdf_path, page_index=0, zoom=2.0):
try:
import fitz # PyMuPDF
except ImportError:
raise RuntimeError("PDF 多模态需要 PyMuPDF：pip install pymupdf")
doc = fitz.open(pdf_path)
if page_index < 0 or page_index >= doc.page_count:
raise IndexError(f"页码 {page_index} 越界，共 {doc.page_count} 页")
pix = doc[page_index].get_pixmap(matrix=fitz.Matrix(zoom, zoom))
raw = pix.tobytes("png")
b64 = base64.b64encode(raw).decode()
return f"data:image/png;base64,{b64}"
# ---------- 聊天封装 ----------
def _chat_call(key, model, content, max_tokens=4000, temperature=0, mid=7, no_rag=True):
"""调 aims.chat。
关键参数 `no_rag` 默认 True（2026-09-16 平台确认）：
- 平台默认给 chat 注入知识库（KB）内容，视觉模型/OCR 收到 RAG 文本会跑偏
- 加上 `no_rag: True` 后，模型纯粹根据 prompt + 图片回答
- 纯文本调用也建议加，避免被 KB 干扰
"""
arguments = {
"model": model,
"messages": [{"role": "user", "content": content}],
"max_tokens": max_tokens, "temperature": temperature,
"no_rag": no_rag,
}
resp = _rpc(AIMS_URL, key, "tools/call",
{"name": "aims.chat", "arguments": arguments}, mid)
txt = resp.get("result", {}).get("content", [{}])[0].get("text", "")
if not txt:
raise RuntimeError("平台返回为空: " + json.dumps(resp, ensure_ascii=False)[:300])
try:
obj = json.loads(txt)
return obj["choices"][0]["message"]["content"], obj.get("usage", {})
except Exception:
return txt, {}
def chat_text(prompt, model=None, max_tokens=4000, temperature=0, api_key=None, no_rag=True):
"""纯文本对话：返回文本。"""
key = _get_key(api_key)
_handshake(key)
model = model or TEXT_MODEL
text, _ = _chat_call(key, model, prompt, max_tokens, temperature, no_rag=no_rag)
return text.strip()
def chat_text_with_usage(prompt, model=None, max_tokens=4000, temperature=0, api_key=None, no_rag=True):
key = _get_key(api_key)
_handshake(key)
model = model or TEXT_MODEL
text, usage = _chat_call(key, model, prompt, max_tokens, temperature, no_rag=no_rag)
return text.strip(), usage
def chat_vision(image_input, prompt, model=None, max_tokens=4000, api_key=None, no_rag=True):
"""图片对话：image_input 可为本地路径或 data: URI。"""
key = _get_key(api_key)
_handshake(key)
model = model or VISION_MODEL
if image_input.startswith("data:"):
data_url = image_input
else:
data_url = _file_to_data_url(image_input)
content = [
{"type": "image_url", "image_url": {"url": data_url}},
{"type": "text", "text": prompt},
]
text, _ = _chat_call(key, model, content, max_tokens, 0, no_rag=no_rag)
return text.strip()
def chat_pdf_page(pdf_path, page_index, prompt, model=None, zoom=2.0, max_tokens=4000, api_key=None, no_rag=True):
"""扫描版 PDF 单页：先渲染成图，再视觉 OCR。"""
data_url = _pdf_page_to_data_url(pdf_path, page_index, zoom)
return chat_vision(data_url, prompt, model=model, max_tokens=max_tokens, api_key=api_key, no_rag=no_rag)
# ---------- JSON 输出辅助 ----------
def _strip_code_fence(s):
s = s.strip()
if s.startswith("```"):
lines = s.splitlines()
if lines and lines[0].startswith("```"):
lines = lines[1:]
if lines and lines[-1].strip().startswith("```"):
lines = lines[:-1]
s = "\n".join(lines)
return s.strip()
def chat_text_json(prompt, model=None, max_tokens=4000, api_key=None, no_rag=True):
"""纯文本对话 → 强制 JSON 输出（解析失败抛错）。"""
text = chat_text(prompt, model=model, max_tokens=max_tokens, api_key=api_key, no_rag=no_rag)
return json.loads(_strip_code_fence(text))
if __name__ == "__main__":
ap = argparse.ArgumentParser(description="AIMS 平台 AI 调用 helper 自检")
ap.add_argument("--api-key", default=None)
ap.add_argument("--text", default="用一句话介绍 Python。")
args = ap.parse_args()
print("=== chat_text ===")
print(chat_text(args.text, api_key=args.api_key))#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""AIMS 平台功能型 API 公共调用 helper（付费 skill 共用）。
能力：
- call_functional_api(resource_id, arguments) 统一调用平台托管功能型 API
- needs_confirm → 抛出 NeedsConfirmation（结构化，带真实报价）
- estimate_cost(resource_id, units) 免费报价（不调上游，0 花费）
协议（2026-09-22 platform-paid-confirm-protocol）：
付费工具默认 **confirm=False**（安全默认）。平台返回 -32006 时不扣费，
本模块把它翻译成 NeedsConfirmation 异常（带 estimated_cost_yuan），
而不是笼统的 RuntimeError——这样 agent 能把真实报价展示给用户，
用户确认后加 --approved 重试。
环境：依赖 AIMS_API_KEY 环境变量或 --api-key
"""
import argparse
import json
import os
import time
import urllib.request
AIMS_URL = "https://aimsgateway.cn/mcp/"
# 内部源名 → 对外目录名（与平台 application/services/public_naming.py 的
# PUBLIC_SOURCES 对齐；未登记的源原样透出）。estimate 用它把内部 resource_id
# 映射成 list_functional_apis 返回的对外 id。
_EXTERNALIZE = {"serpapi": "web", "hunter": "email", "jev": "decision", "apis": "public-api"}
class NeedsConfirmation(RuntimeError):
"""付费工具未确认 —— 平台返回 -32006 / requires_confirmation。
仍未扣费。调用方应把 estimated_cost_yuan 展示给用户，
用户同意后带 confirm=True 重试。
继承 RuntimeError 是为了向后兼容既有 ``except RuntimeError`` 调用方。
"""
def __init__(self, resource_id, estimated_cost_yuan=0.0, basis="", raw=None, hint=""):
self.resource_id = resource_id
self.estimated_cost_yuan = float(estimated_cost_yuan or 0.0)
self.basis = basis or ""
self.raw = raw if isinstance(raw, dict) else {}
self.hint = hint or ""
super().__init__(
f"付费工具需确认: {resource_id} 约 ¥{self.estimated_cost_yuan:.2f} "
f"(价格依据: {self.basis or 'unknown'})。"
f"请把该报价展示给用户；确认后加 --approved 重试。未确认不会扣费。"
)
def to_dict(self):
"""给 CLI 输出的结构化载荷（agent 可直接机读）。"""
return {
"ok": False,
"needs_confirmation": True,
"resource_id": self.resource_id,
"estimated_cost_yuan": self.estimated_cost_yuan,
"cost_basis": self.basis or "unknown",
"hint": self.hint or "把费用展示给用户；用户确认后加 --approved 重试",
}
def _external_rid(resource_id):
"""内部 resource_id → 对外目录 id。serpapi/google → web/google"""
src, _, rest = (resource_id or "").strip().partition("/")
ext = _EXTERNALIZE.get(src, src)
return f"{ext}/{rest}" if rest else ext
def _maybe_needs_confirmation(resource_id, parsed):
"""把平台的 -32006 载荷翻译成 NeedsConfirmation；不是则返回 None。
兼容三种形态：
A {"error": {"code": -32006, "data": {...}}} 平台 JSON-RPC 包装
B {"code": -32006, "data": {...}} 已解包的 error 对象
C {"requires_confirmation": true, ...} 直接是 data
"""
if not isinstance(parsed, dict):
return None
err = parsed.get("error")
if isinstance(err, dict) and err.get("code") == -32006:
d = err.get("data") if isinstance(err.get("data"), dict) else {}
return NeedsConfirmation(resource_id, d.get("estimated_cost_yuan"),
d.get("cost_basis", ""), err, d.get("hint", ""))
if parsed.get("code") == -32006:
d = parsed.get("data") if isinstance(parsed.get("data"), dict) else {}
return NeedsConfirmation(resource_id, d.get("estimated_cost_yuan"),
d.get("cost_basis", ""), parsed, d.get("hint", ""))
if parsed.get("requires_confirmation") is True:
return NeedsConfirmation(resource_id, parsed.get("estimated_cost_yuan"),
parsed.get("cost_basis", ""), parsed, parsed.get("hint", ""))
return None
def raise_if_needs_confirmation(resource_id, parsed):
"""载荷是 -32006 时抛 NeedsConfirmation，否则原样返回 parsed。
给自建 _rpc_call 的 skill 复用（它们拿到 isError 载荷后调这个，
就能和走本模块的 skill 有同一套结构化确认语义）。
"""
nc = _maybe_needs_confirmation(resource_id, parsed)
if nc is not None:
raise nc
return parsed
# ---------- 鉴权 & 协议 ----------
def _get_key(api_key=None):
key = api_key or os.environ.get("AIMS_API_KEY")
if not key:
raise RuntimeError(
"缺少 AIMS API Key：请设置环境变量 AIMS_API_KEY，或调用方传 --api-key"
)
return key
def _rpc(url, key, method, params, mid, tries=3, timeout=120):
body = json.dumps({"jsonrpc": "2.0", "id": mid, "method": method, "params": params}).encode()
headers = {"Content-Type": "application/json", "Authorization": f"Bearer {key}"}
last = None
for _ in range(tries):
try:
req = urllib.request.Request(url, data=body, method="POST", headers=headers)
with urllib.request.urlopen(req, timeout=timeout) as r:
return json.loads(r.read().decode())
except Exception as e: # noqa: BLE001
last = e
time.sleep(2)
raise RuntimeError(f"MCP 调用失败: {last}")
def _handshake(key):
_rpc(AIMS_URL, key, "initialize",
{"protocolVersion": "2025-03-26", "capabilities": {},
"clientInfo": {"name": "office-skills-func-api", "version": "1.0"}}, 1)
try:
_rpc(AIMS_URL, key, "notifications/initialized", {}, 2)
except Exception:
pass
# ---------- 功能型 API 调用 ----------
def call_functional_api(resource_id, arguments=None, api_key=None, mid=3, confirm=False):
"""调 AIMS 平台的功能型 API。
Args:
resource_id: 资源 ID，9 个可选值：
- hunter/account
- hunter/domain-search
- hunter/email-finder
- hunter/email-verifier
- serpapi/account
- serpapi/{engine}
- apis/catalog
- apis/{slug}
- apis/{slug}/{path}
arguments: dict，参数键值对（按上游 API 的 params 字段说明）
api_key: 可选 Bearer Token（默认读 env）
confirm: True 时附 confirm:true，平台放行付费工具；False 时平台返回
-32006 / requires_confirmation，未付费不扣费（默认 False，安全默认）
返回 dict（平台直接返回上游响应 + 末尾 `_aimschina` 元数据）
抛出 RuntimeError 当 isError=True
"""
key = _get_key(api_key)
_handshake(key)
args = dict(arguments or {})
if confirm:
args["confirm"] = True
resp = _rpc(AIMS_URL, key, "tools/call",
{"name": "aims.call_functional_api",
"arguments": {"resource_id": resource_id, "arguments": args}}, mid)
is_err = resp.get("result", {}).get("isError")
text = resp.get("result", {}).get("content", [{}])[0].get("text", "{}")
parsed = json.loads(text)
if is_err:
_nc = _maybe_needs_confirmation(resource_id, parsed)
if _nc is not None:
raise _nc
raise RuntimeError(f"{resource_id} 错误: {parsed}")
return parsed
# ---------- 9 个快捷封装 ----------
def hunter_account(api_key=None, confirm=False):
"""hunter 账户与配额（免费）。"""
return call_functional_api("hunter/account", {}, api_key=api_key, confirm=confirm)
def hunter_domain_search(domain, limit=10, offset=0, type=None,
seniority=None, department=None, api_key=None, confirm=False):
"""按域名搜邮箱。返回 {"domain":..., "emails":[{value,type,confidence,position,...}], "_aimschina":{cost_yuan,...}}"""
args = {"domain": domain, "limit": str(limit)}
if offset:
args["offset"] = str(offset)
if type:
args["type"] = type
if seniority:
args["seniority"] = seniority
if department:
args["department"] = department
return call_functional_api("hunter/domain-search", args, api_key=api_key, confirm=confirm)
def hunter_email_finder(domain, full_name=None, first_name=None, last_name=None,
max_duration=10, api_key=None, confirm=False):
"""按人名 + 域名推断邮箱。"""
args = {"domain": domain, "max_duration": str(max_duration)}
if full_name:
args["full_name"] = full_name
elif first_name and last_name:
args["first_name"] = first_name
args["last_name"] = last_name
return call_functional_api("hunter/email-finder", args, api_key=api_key, confirm=confirm)
def hunter_email_verifier(email, api_key=None, confirm=False):
"""验证邮箱是否可投递。返回 {"status": "valid"|"invalid"|"accept_all"|"webmail"|"disposable"|"unknown", ...}"""
return call_functional_api("hunter/email-verifier", {"email": email}, api_key=api_key, confirm=confirm)
def serpapi_account(api_key=None, confirm=False):
"""serpapi 账户与配额（免费）。"""
return call_functional_api("serpapi/account", {}, api_key=api_key, confirm=confirm)
def serpapi_search(engine, q=None, num=10, page=1, gl=None, hl=None,
location=None, api_key=None, confirm=False):
"""透传到任意 serpapi 引擎（google / google_maps / google_shopping 等）。
engine 必填，例如 "google_maps" 拿 POI、"google" 拿网页搜索结果、"google_shopping" 拿商品。
实际调 serpapi.engine 工具（不是走 functional_api 路由）。
confirm=True 时附 confirm:true 放行付费工具；默认 False 平台返回 -32006 不扣费。
"""
args = {"engine": engine, "num": str(num), "page": str(page)}
if q:
args["q"] = q
if gl:
args["gl"] = gl
if hl:
args["hl"] = hl
if location:
args["location"] = location
if confirm:
args["confirm"] = True
key = _get_key(api_key)
_handshake(key)
resp = _rpc(AIMS_URL, key, "tools/call",
{"name": "serpapi.engine", "arguments": args}, 3)
is_err = resp.get("result", {}).get("isError")
text = resp.get("result", {}).get("content", [{}])[0].get("text", "{}")
parsed = json.loads(text)
if is_err:
_nc = _maybe_needs_confirmation("serpapi/" + str(engine or ""), parsed)
if _nc is not None:
raise _nc
raise RuntimeError(f"serpapi.engine({engine}) 错误: {parsed}")
return parsed
def apis_catalog(api_key=None):
"""列出所有 apis 目录（450+ 免密钥公共 API）。"""
return call_functional_api("apis/catalog", {}, api_key=api_key)
def apis_slug(slug, api_key=None):
"""调 apis/{slug} 顶层端点（GET）。"""
return call_functional_api(f"apis/{slug}", {}, api_key=api_key)
def apis_slug_path(slug, path, method="GET", body=None, params=None, api_key=None):
"""调 apis/{slug}/{path} 任意端点。
Args:
slug: 例如 "openweather"
path: 例如 "data/2.5/weather"
method: GET / POST / PUT / DELETE
body: POST/PUT body
params: query 参数
"""
args = {"method": method}
if body is not None:
args["body"] = body
if params is not None:
args["params"] = params
return call_functional_api(f"apis/{slug}/{path}", args, api_key=api_key)
# ---------- 免费报价 estimate（0 上游花费） ----------
# 本地兜底价：**只登记实测过的资源**（来源：lead-miner install.md §17.4，
# 实测日期 2026-09-17）。平台实时价取不到时才用它，且 cost_basis 会明确标
# local_fallback + price_note 提示可能有偏差——绝不拿它冒充权威报价。
# 没实测过的资源**一律不给兜底价**（宁可 ok=False，也不猜）。
LOCAL_FALLBACK_PRICE_YUAN = {
"serpapi/google_maps": 0.21,
"hunter/domain-search": 0.204,
"hunter/email-verifier": 0.10,
}
def estimate_cost(resource_id, units=1, api_key=None):
"""纯报价：走**免费**的 aims.list_functional_apis 读实时单价，不调上游。
这是唯一可靠的报价途径——catalog 声明价与实收价实测差 ~18%，
而 list_functional_apis 返回的是平台已折算的 effective 单价
（observed 实测价优先，缺失才回落 declared）。
Args:
resource_id: 内部 resource_id，如 "serpapi/google" / "hunter/domain-search"
units: 预计调用次数
api_key: 可选 Bearer Token（默认读 env）
Returns:
dict: {ok, resource_id, external_resource_id, units, unit_cost_yuan,
cost_basis, total_yuan, upstream_called: False, note}
查不到价时返回 ok=False（不抛异常——报价失败不该让调用方崩）
"""
ext = _external_rid(resource_id)
base = {
"resource_id": resource_id,
"external_resource_id": ext,
"units": int(units or 1),
"upstream_called": False,
}
def _fallback(note):
"""平台价拿不到时用本地实测兜底；没登记的资源保持 ok=False（不猜价）。"""
unit = LOCAL_FALLBACK_PRICE_YUAN.get(resource_id)
if unit is None:
return False
base.update({
"ok": True,
"unit_cost_yuan": round(unit, 4),
"cost_basis": "local_fallback",
"billing_unit": "per_call",
"total_yuan": round(unit * base["units"], 4),
"price_note": ("本地登记的【实测兜底价】(2026-09-17)，非平台实时价，"
"实收可能偏差约 ±18%；要准数请配好 AIMS_API_KEY 重跑 estimate。"),
"note": note,
})
return True
try:
key = _get_key(api_key)
except RuntimeError as e:
if not _fallback("用本地实测兜底价（未触达平台）"):
base.update({"ok": False, "error": str(e),
"note": "estimate 需要 AIMS_API_KEY 才能读到平台实时单价"})
return base
try:
_handshake(key)
resp = _rpc(AIMS_URL, key, "tools/call",
{"name": "aims.list_functional_apis",
"arguments": {"limit": 200}}, 7)
text = resp.get("result", {}).get("content", [{}])[0].get("text", "{}")
payload = json.loads(text)
except Exception as e: # noqa: BLE001
if not _fallback("用本地实测兜底价（平台单价查询失败）"):
base.update({"ok": False, "error": f"读取平台单价失败: {e}"})
return base
items = payload.get("apis") or []
hit = None
for it in items:
if isinstance(it, dict) and it.get("resource_id") in (ext, resource_id):
hit = it
break
if hit is None:
if _fallback("平台目录里没有该资源，改用本地实测兜底价"):
base["not_in_catalog"] = True
return base
base.update({
"ok": False,
"error": f"平台目录里没找到 {ext}（内部 id: {resource_id}）",
"note": "单价未知——不要凭猜测报给用户；请确认该资源是否已上架/被 chat 策略隐藏",
"available_count": len(items),
})
return base
unit = float(hit.get("unit_cost_yuan") or 0.0)
base.update({
"ok": True,
"name": hit.get("name") or "",
"unit_cost_yuan": round(unit, 4),
"cost_basis": hit.get("price_basis") or "declared",
"billing_unit": hit.get("billing_unit") or "per_call",
"total_yuan": round(unit * base["units"], 4),
"note": "本报价未调用上游、未扣费；用户确认后加 --approved 真实执行",
})
return base
def emit_estimate(resource_id, units=1, api_key=None):
"""CLI 用：打印报价 JSON 并以退出码结束（0=有报价 / 1=查不到）。
在 skill 里这样用（parse_args 之后、真调付费接口之前）：
if args.estimate:
emit_estimate("serpapi/google")
"""
est = estimate_cost(resource_id, units=units, api_key=api_key)
print(json.dumps(est, ensure_ascii=False, indent=2))
raise SystemExit(0 if est.get("ok") else 1)
def add_estimate_flag(parser):
"""给 skill 的 argparse 统一加 --estimate（免费报价，不调上游）。"""
parser.add_argument(
"--estimate", action="store_true",
help="只报预估费用，不调用上游、不扣费（免费）",
)
return parser
# ---------- 列出当前所有可用 resource_id ----------
KNOWN_RESOURCES = [
"hunter/account",
"hunter/domain-search",
"hunter/email-finder",
"hunter/email-verifier",
"serpapi/account",
"serpapi/{engine}",
"apis/catalog",
"apis/{slug}",
"apis/{slug}/{path}",
]
if __name__ == "__main__":
import argparse
ap = argparse.ArgumentParser(description="AIMS 功能型 API helper 自检")
ap.add_argument("--api-key", default=None)
ap.add_argument("--resource", default="hunter/account", help=f"资源 ID，已知: {KNOWN_RESOURCES}")
ap.add_argument("--args", default="{}", help="JSON 字符串参数")
args = ap.parse_args()
print(f"=== call_functional_api({args.resource}) ===")
parsed_args = json.loads(args.args)
result = call_functional_api(args.resource, parsed_args, api_key=args.api_key)
print(json.dumps(result, ensure_ascii=False, indent=2)[:2000])A: 分三步让用户**控制成本**。search 0.21 元算便宜，但 verify 1 个域名 0.30 元 ×50 就 ¥15 了。
用户看到 cost_estimate 后可以决定要不要花这个钱。
A: 可以。删 selected.json 重跑 confirm 即可，没扣任何费用。
A: 1 个类目 50 个候选 verify 全部可能 ¥15+；先问预算就能选合适的 select 范围。
A: 直接在 `--city 城市名 --category 类别` 指定英文 query（如 `coffee shop Shenzhen`），
serpapi 会按 Google Maps 数据返回。
A: 在 leads.xlsx 的"未找到邮箱" sheet 里有说明。可手动去公司官网 About 页找，或换其他工具（apollo.io / snov.io）。
---