外贸客户挖掘增强版 lead-miner-pro：先与客户沟通需求 → 预估费用 → 调 Google Maps + Hunter.io 验证邮箱，输出 xlsx 客户名单。比基础版 lead-miner 强调预算透明、强制费用预估、用户确认后才执行。适合谨慎型客户、大金额场景、销售/商务代理、新手首次尝试。支持统一处理流程。
---
name: lead-miner-pro
description: 外贸客户挖掘（lead-miner-pro）增强版：先免费预估费用 → 等用户明确确认 → 再调用 Google Maps + Hunter.io 验证邮箱，输出 xlsx 客户名单。脚本分步 + agent 编排，结构性护栏保证"未确认绝不花一分钱"，适合需要预算透明、怕被乱扣费的外贸拓客、本地市场调研、邮件群发场景。
---
本 skill 的真实能力依赖平台后端付费工具。在网页 / chat 端点下，这些工具可能不可用：
此时**请勿轻信**，请改用 **agent / CLI 路径**运行本 skill 以获取真实结果。任何路径下都**严禁谎称「已调用 / 已搜索完成」而实际未执行**。
> 费用预估必须基于脚本 `estimate` 子命令的真实返回，**禁止编造金额**。
**核心承诺：用户完全确认之前，本 skill 不调用任何付费接口（一分钱都不花）。**
本 skill 采用 **「脚本分步 + agent 编排」** 架构：所有付费动作（Google Maps 搜索、Hunter 邮箱验证）都被代码层 `--approved` 开关锁死，**只有拿到用户明确确认后，agent 才传 `--approved` 放行**。即使 LLM 自己"想"提前搜，脚本也会直接拒绝——成本可控是结构性的，不靠 LLM 自律。
用户说"外贸搜客"、"找海外买家"、"按行业+城市搜客户"、"先告诉我多少钱"、"先预估一下费用"、"再开始搜"、"怕被乱扣费"时触发。**当用户要求"先告知价格"或"先沟通需求"时**，优先使用本 skill。
脚本 `scripts/lead_miner_pro.py` 在每一处付费调用前都强制检查 `--approved`：
| 子命令 | 是否付费 | 无 `--approved` 时 |
|--------|----------|---------------------|
| `estimate` | 否（纯本地计算） | 正常返回预估，不调任何接口 |
| `search` | 是（serpapi ¥0.255/次） | **拒绝**，提示先 estimate |
| `verify` | 是（hunter ¥0.204+¥0.10/个） | **拒绝**，提示先确认范围 |
| `run` | 是（全链路） | **拒绝**，提示 estimate→确认→`run --approved` |
| `confirm` | 否（读文件算成本） | 正常，写入 selected.json |
**agent 的硬责任**：**绝不允许凭自己的判断传 `--approved`**。只有当终端用户明确打字确认（如"确认执行"/"同意"/"go"+ 看到预估金额）后，才能在下一步命令加 `--approved`。
[0] estimate（免费） agent 调 estimate → 把 cost_estimate 展示给用户
↓
[1] 等用户确认（必须） agent 问"预估总费用 ¥X.XX，确认执行吗？"；用户打字确认才继续
↓ ← 没这一步，后面全是拒绝
[2] search --approved 拿到候选 POI（candidates.json）
↓
[3] confirm（免费） agent 让用户选范围（first-N / 具体 id），展示二次成本
↓ （可选：再要一次确认）
[4] verify --approved 仅对确认范围调 hunter 拿邮箱+验证 → leads.xlsx# 第 0 步：免费预估，不花一分钱
python scripts/lead_miner_pro.py estimate --city 杭州 --category coffee --limit 10 --verify_count 3
# 第 2 步：用户确认后，才加 --approved 搜
python scripts/lead_miner_pro.py search --city 杭州 --category coffee --limit 2 --approved --output candidates.json
# 第 3 步：选范围（免费）
python scripts/lead_miner_pro.py confirm --input candidates.json --select first-3 --output selected.json
# 第 4 步：用户再次确认后，才加 --approved 验证
python scripts/lead_miner_pro.py verify --input selected.json --approved --output leads.xlsx
# 一键全链路（同样必须 --approved，且已获用户确认）：
python scripts/lead_miner_pro.py run --city 杭州 --category coffee --limit 2 --select first-3 --approved --output leads.xlsx> **CLI 人机模式**：终端用户也可直接 `--interactive`（脚本内联逐步问询，输 y 才执行）。但 **agent 模式禁止用 `--interactive`**（会阻塞），必须走"estimate→确认→`--approved`"显式编排。
1. **禁止**在用户看到预估费用前传 `--approved`。
2. **禁止**跳过 confirm 直接 verify（verify 只认 selected.json，且需 `--approved`）。
3. **禁止**把 `--approved` 当成默认值写进模板——它必须是"用户确认"这一动作之后的显式标记。
4. **禁止**自己编预估费用——`estimate` 的 `cost_estimate` 才是真实值（之前 chat 模式乱报 ¥198/¥180 已废弃）。
| 项 | 单价（元） | 公式 |
|----|-----------|------|
| search serpapi | 0.21 | 1 次搜索（与 limit 无关） |
| verify domain-search | 0.204 | N × 0.204（N=验证数量） |
| verify email-verifier | 0.10 | N × 1.5 × 0.10（每域约 1.5 邮箱） |
典型场景（用户侧看到的"本 skill 卖价"另算，见下）：
| 场景 | 上游总成本 | 说明 |
|------|-----------|------|
| 只 search 不要邮箱 | ¥0.21 | 拿 POI 名单 |
| search + verify first-3 | ¥1.27 | 拿邮箱+验证 |
| search + verify first-10 | ¥3.54 | 大批量 |
**谁付钱（已实测澄清）**：终端用户调用本 skill 时，平台在其会话内注入用户自己的 key，**用户既付 skill 卖价（¥0.20），也付上游（serpapi/hunter）**。开发者（发布者）不承担上游成本、不亏本。所以"用户完全确认才工作"既保护用户预算，也符合平台计费模型。
| 维度 | lead-miner | lead-miner-pro |
|------|-----------|----------------|
| 计费模型 | 卖价 ¥1.20/次 | 卖价 ¥0.20/次（让利） |
| 成本控制 | 靠 agent 自觉 | **代码层 `--approved` 硬锁**，未确认零花费 |
| 流程 | 标准 3 步 | 强制 estimate→确认→执行 |
| 适用 | 熟手大批量 | 谨慎型 / 新手 / 怕乱扣费 |
| 错误 | 原因 | 解决 |
|------|------|------|
| "缺少 --approved" | agent 未等确认就调付费命令 | 回到 estimate，先让用户确认 |
| 余额不足 | 用户平台余额不够 | 提示用户充值 |
| hunter 找不到邮箱 | 小公司数据库覆盖差 | 写"未找到邮箱"sheet，可手动查 |
<!-- ===== 以下为内嵌脚本代码（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
# ---------- 第一步：search ----------
def cmd_search(args):
"""按 mode 分发：serpapi（推荐）/ api（Google Places API）/ browser（playwright）"""
# ===== 硬护栏：未获用户明确确认，绝不调用任何搜索/外部接口（含 LLM 抽取）=====
if not getattr(args, "approved", False):
raise RuntimeError(
"⛔ 未获用户明确确认（缺少 --approved），拒绝调用搜索接口。\n"
"正确流程：先用 `estimate` 向用户展示预估费用 → 用户确认 → 再加 --approved 执行。"
)
# ---- 自然语言抽取 ----
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 包裹层）
# 用户已 --approved 时把 confirm:true 一起带上，平台付费工具确认机制放行
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.*）。
返回内部数据。兼容性：
- hunter.* 系列：返 {"data": {...}, "meta": ..., "_aimschina": ...}（data 包裹）
- serpapi.* 系列：返平铺 {...search_metadata..., "local_results": ...}
这里保留原始结构，由调用方按需取 resp["data"] 或 resp。
抛出 RuntimeError 当 isError=True。
confirm=True 时，把 args 里加 `confirm: true`（平台付费工具用户确认机制）。
"""
api_key = os.environ.get("AIMS_API_KEY")
if not api_key:
raise RuntimeError("缺少 AIMS_API_KEY 环境变量")
import urllib.request
if confirm and isinstance(args, dict):
args = dict(args) # 拷贝，避免污染调用方
args["confirm"] = True
body = json.dumps({
"jsonrpc": "2.0", "id": 11, "method": "tools/call",
"params": {"name": tool_name, "arguments": 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:
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/次
"""
# ===== 硬护栏：未获用户明确确认，绝不调用 hunter 付费接口 =====
if not getattr(args, "approved", False):
raise RuntimeError(
"⛔ 未获用户明确确认（缺少 --approved），拒绝调用 hunter 付费接口。\n"
"请先让用户确认 verify 范围与费用，再加 --approved 重试。"
)
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",
},
}
# ---------- 第四步：run（一条命令） ----------
def cmd_run(args):
"""一键：search → confirm → verify。适合 agent 调用。
三种入口：
1. --interactive：阻塞问询用户（city/category/budget），CLI 模式（脚本内联确认）
2. --stdin：从 stdin 读 JSON 一次性给参数（agent 自动化）
3. 默认参数全给：直接跑（agent 已知参数时用）
硬护栏：必须 --approved（agent 在用户确认后传入）或 --interactive（脚本内联问询），
否则拒绝执行任何付费步骤。
"""
import tempfile
# ===== 硬护栏：必须用户确认（--approved）或交互问询（--interactive）=====
if not (getattr(args, "approved", False) or args.interactive):
raise RuntimeError(
"⛔ 缺少 --approved 且非 --interactive：拒绝执行任何付费步骤。\n"
"正确流程：estimate 展示预估 → 用户确认 → run --approved ..."
)
# 交互模式下，用户在脚本内已确认，视为已批准；否则用 --approved 标记
effective_approved = bool(getattr(args, "approved", False) or args.interactive)
# 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=effective_approved,
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=effective_approved)
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 cmd_estimate(args):
"""仅做费用预估，绝不调用任何付费接口。
agent 第一步必须调用它，把 cost_estimate 展示给用户；等用户明确确认后，
再调用 search / verify / run 并带 --approved。这是"用户完全确认才工作"的结构性保障。
"""
cost = calc_cost(
mode=args.mode,
search_count=args.limit,
verify_count=args.verify_count,
used_description=False,
)
result = {
"ok": True,
"free": True,
"no_api_called": True,
"mode": args.mode,
"params": {
"city": args.city, "category": args.category,
"limit": args.limit, "verify_count": args.verify_count,
},
"cost_estimate": cost,
"message": (
"以上为预估费用，本次未调用任何付费接口。"
"请向用户展示并等待其明确确认；确认后执行时务必加 --approved。"
),
}
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 _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("--approved", action="store_true",
help="用户已明确确认费用后才允许调用付费搜索接口（serpapi/api/browser）")
p.add_argument("--output", help="输出 candidates.json")
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("--approved", action="store_true",
help="用户已明确确认 verify 范围与费用后才允许调用 hunter 付费接口")
p.add_argument("--output", default="leads.xlsx", help="输出 xlsx 文件")
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("--approved", action="store_true",
help="用户已明确确认后才执行全链路（search→confirm→verify）付费步骤")
p.add_argument("--output", default="leads.xlsx")
p.set_defaults(func=cmd_run)
# estimate（免费预估，不调用任何付费接口）
p = sub.add_parser("estimate", help="仅预估费用，不调用任何付费接口（agent 先给用户看报价）")
p.add_argument("--mode", default="serpapi", choices=["serpapi", "api", "browser"])
p.add_argument("--city", default=None, help="城市（仅用于展示）")
p.add_argument("--category", default=None, help="类别（仅用于展示）")
p.add_argument("--limit", type=int, default=10, help="拟搜索候选数")
p.add_argument("--verify_count", type=int, default=3, help="拟验证数量（用于预估 verify 成本）")
p.add_argument("--output", help="可选，输出预估 JSON")
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 Exception as e:
# needs-confirmation:structured-payload
# 平台未确认付费调用（-32006）→ NeedsConfirmation。
# 必须原样吐结构化载荷：needs_confirmation / estimated_cost_yuan
# / cost_basis 要能被上层机读，不能拍平成一句错误文本。
# exit 3 = 待用户确认（0=成功 / 1=失败 / 3=待确认）。
if hasattr(e, "to_dict") and hasattr(e, "estimated_cost_yuan"):
print(json.dumps(e.to_dict(), ensure_ascii=False, indent=2))
sys.exit(3)
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])