Excel 数据分析 excel-analyze：多表合并、按列分组透视聚合（和/均值/计数/极值）、数值列统计；支持自然语言问 Excel（AI 转 JSON 操作计划并执行）。适合财务汇总、销售统计、库存盘点、考勤统计等多表数据处理场景。
---
name: excel-analyze
description: Excel 数据分析 excel-analyze：多表合并、按列分组透视聚合（求和/均值/计数/最大/最小）、数值列统计（和/均值/极值）；支持自然语言问 Excel（AI 自动转 JSON 操作计划并执行）。适合财务汇总、销售统计、库存盘点、考勤统计、多部门数据合并、不会写公式也能分析 Excel。确定性计算、本地执行、数据不上传。
---
本 skill 的真实能力依赖平台后端工具。在网页 / chat 端点下，这些工具可能不可用：
此时**请勿轻信**，请改用 **agent / CLI 路径**运行本 skill 以获取真实结果。任何路径下都**严禁谎称「已调用 / 已搜索完成」而实际未执行**。
处理 Excel 表格数据：合并多个同结构工作簿、数据透视聚合、数值列统计。支持**自然语言提问**（AI 自动转成 JSON 操作计划并执行）。本地执行，数据不外传。
**场景 1：财务月度汇总** —— 12 个部门的销售表（结构一致、字段相同）合并成一个总表，再按部门透视金额总和，3 行命令搞定。
**场景 2：销售不会写公式** —— 直接问 "金额最高的前 5 个客户是谁"，AI 自动识别操作 + 执行 + 给你表格，不用学 VLOOKUP。
**场景 3：HR 考勤统计** —— 多个子公司考勤表合并，按部门统计出勤天数平均值，定位异常部门。
| 角色 | 典型用途 |
|------|----------|
| 财务 / 会计 | 多部门月报合并、按科目汇总 |
| 销售 / 运营 | 销售数据透视、按区域统计、客户排名 |
| HR / 人事 | 考勤统计、工资汇总、绩效排名 |
| 仓储 / 库存 | 多仓库库存盘点、按品类汇总 |
| 学生 / 教师 | 调研数据统计、问卷分析 |
| 数据分析师 | 快速探查、Top N、过滤异常值 |
# 1. 多表合并
python scripts/xlsx_analyze.py merge --inputs 1月.xlsx 2月.xlsx 3月.xlsx --output Q1.xlsx
# 2. 数据透视
python scripts/xlsx_analyze.py pivot --input 销售.xlsx --group 区域 --value 金额 --agg sum
# 3. 自然语言问 Excel
python scripts/xlsx_analyze.py nlq --input 销售表.xlsx --question "按区域统计金额总和"
python scripts/xlsx_analyze.py nlq --input 销售表.xlsx --question "金额最高的前 5 个客户"
python scripts/xlsx_analyze.py nlq --input 销售表.xlsx --question "金额大于 5000 的有哪些"输出示例：
{"ok": true, "operation": "group_agg", "groups": [
{"区域": "华东", "金额总和": 1234567, "count": 45},
{"区域": "华南", "金额总和": 987654, "count": 32}
]}| 对比项 | 传统做法 | excel-analyze |
|--------|---------|----------------|
| 多表合并 | 复制粘贴 / 邮件合并 | CLI 一行命令 |
| 数据透视 | 装 Excel + 学透视向导 | CLI `--group --value --agg` |
| 数值统计 | 写 SUM/AVERAGE 公式 | `--summary` 直接输出 |
| 自然语言问 | 自己琢磨公式 | "金额最高的前 5 个" 一句话 |
| 不会写公式 | 求助同事 | AI 自动生成 |
| 数据安全 | 上传在线 AI 工具（隐私风险）| 本地计算，只把表头 + 5 行样本给 AI |
| 成本 | Excel 订阅 | 按次付费，¥0.10/次 |
用户要求"合并 Excel / 多表合并 / 数据透视 / 按 XX 汇总 / 统计某列 / Excel 求和平均 / 分析表格数据 / 帮我看看这个表里哪个区域销售额最高"时触发。
用户要求"合并 Excel / 多表合并 / 数据透视 / 按 XX 汇总 / 统计某列 / Excel 求和平均 / 分析表格数据 / 帮我看看这个表里哪个区域销售额最高"时触发。
pip install openpyxl
# 自然语言问 Excel（nlq）：依赖 AIMS_API_KEYpython scripts/xlsx_analyze.py merge --inputs a.xlsx b.xlsx c.xlsx --output 全部.xlsxpython scripts/xlsx_analyze.py pivot --input 销售.xlsx --group 区域 --value 金额 --agg sumpython scripts/xlsx_analyze.py summary --input 销售.xlsx --columns 金额 数量输出每列的 数量 / 总和 / 均值 / 最小值 / 最大值。
python scripts/xlsx_analyze.py nlq --input 销售表.xlsx --question "按区域统计金额总和"
python scripts/xlsx_analyze.py nlq --input 销售表.xlsx --question "金额最高的前 3 行"
python scripts/xlsx_analyze.py nlq --input 销售表.xlsx --question "金额大于 5000 的有哪些"1. 列名取第一行表头；列名不存在时报错并列出已有列
2. 空值与文本自动跳过数值计算（非空非数值列会标注 non_empty）
3. 输出 JSON：`{"ok": true, "rows"/"stats": [...]}`，失败 `{"ok": false, "error": "原因"}`
| 情况 | 处理 |
|---|---|
| 文件打不开/表头为空 | 明确报错，提示检查文件 |
| 分组列/值列不存在 | 报错并列出现有列名 |
| 合并文件不足 2 个 | 报错提示至少 2 个 |
<!-- ===== 以下为内嵌脚本代码（agent 安装时按需落盘为 scripts/<name> 并 chmod +x） ===== -->
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""Excel 进阶分析：多表合并 / 数据透视 / 列统计 / 自然语言问 Excel。
用法示例：
python xlsx_analyze.py merge --inputs a.xlsx b.xlsx c.xlsx --output all.xlsx
python xlsx_analyze.py merge --inputs *.xlsx --output all.xlsx --dedup
python xlsx_analyze.py pivot --input sales.xlsx --group 区域 --value 金额 --agg sum
python xlsx_analyze.py summary --input sales.xlsx --columns 金额,数量
python xlsx_analyze.py nlq --input sales.xlsx --question "哪个区域销售总额最高？" # 自然语言问 Excel
依赖：pip install openpyxl
AI 增强（nlq）：依赖 AIMS_API_KEY
"""
import argparse
import csv
import json
import sys
from pathlib import Path
try:
import openpyxl
except ImportError: # pragma: no cover
sys.stderr.write("缺少依赖 openpyxl，请先执行: pip install openpyxl\n")
sys.exit(2)
_HERE = Path(__file__).resolve().parent
_ROOT = _HERE.parent.parent
# _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
_AIMS_OK = True
except Exception: # pragma: no cover
_AIMS_OK = False
def _load_wb(path):
return openpyxl.load_workbook(path, data_only=True)
def _header_map(ws, header_row=1):
"""返回 {列名: 列号}。"""
mapping = {}
for cell in ws[header_row]:
if cell.value is not None:
mapping[str(cell.value).strip()] = cell.column
return mapping
# ---------- 合并 ----------
def cmd_merge(args) -> dict:
"""多个 xlsx 同结构合并：取第一个文件的表头，追加其余文件数据行（去重可选）。"""
if len(args.inputs) < 2:
raise ValueError("至少需要 2 个文件")
first = Path(args.inputs[0])
wb = openpyxl.load_workbook(first, data_only=False)
ws = wb.active
headers = {c.value: c.column for c in ws[1] if c.value is not None}
if not headers:
raise ValueError(f"第一个文件表头为空: {first}")
total_rows = ws.max_row - 1 # 已有数据行
for path in args.inputs[1:]:
wb2 = _load_wb(path)
ws2 = wb2.active
h2 = {str(c.value).strip(): c.column for c in ws2[1] if c.value is not None}
for row in ws2.iter_rows(min_row=2, values_only=True):
# 按第一个文件的列名取对应值
values = []
for name, col in headers.items():
src_col = h2.get(name)
idx = src_col - 1 if src_col else None
values.append(row[idx] if idx is not None and idx < len(row) else None)
ws.append(values)
total_rows += 1
out = Path(args.output)
out.parent.mkdir(parents=True, exist_ok=True)
wb.save(str(out))
return {"ok": True, "output": str(out), "files": len(args.inputs), "data_rows": total_rows}
# ---------- 透视 ----------
def cmd_pivot(args) -> dict:
wb = _load_wb(args.input)
ws = wb.active
hmap = _header_map(ws)
if args.group not in hmap:
raise ValueError(f"分组列不存在: {args.group}（现有: {list(hmap)}）")
if args.value not in hmap:
raise ValueError(f"值列不存在: {args.value}（现有: {list(hmap)}）")
g_col = hmap[args.group]
v_col = hmap[args.value]
agg = args.agg
groups: dict[str, list] = {}
for row in ws.iter_rows(min_row=2, values_only=True):
g = row[g_col - 1]
v = row[v_col - 1]
key = str(g) if g is not None else "(空)"
if v is None:
continue
groups.setdefault(key, []).append(float(v))
result = []
for key, vals in groups.items():
n = len(vals)
if agg == "sum":
r = sum(vals)
elif agg == "mean":
r = sum(vals) / n
elif agg == "count":
r = n
elif agg == "max":
r = max(vals)
elif agg == "min":
r = min(vals)
else:
raise ValueError(f"不支持的聚合: {agg}")
result.append({args.group: key, f"{args.value}_{agg}": round(r, 4), "count": n})
if args.output:
wbo = openpyxl.Workbook()
wso = wbo.active
wso.title = "透视结果"
wso.append([args.group, f"{args.value}_{agg}", "count"])
for item in result:
wso.append([item[args.group], item[f"{args.value}_{agg}"], item["count"]])
wbo.save(args.output)
return {"ok": True, "output": args.output, "agg": agg, "groups": len(result), "rows": result}
# ---------- 列统计 ----------
def cmd_summary(args) -> dict:
wb = _load_wb(args.input)
ws = wb.active
hmap = _header_map(ws)
out = []
for name in args.columns:
if name not in hmap:
out.append({"column": name, "error": "列不存在"})
continue
col = hmap[name]
vals = []
for row in ws.iter_rows(min_row=2, values_only=True):
v = row[col - 1]
if isinstance(v, (int, float)) and v == v: # 非 NaN
vals.append(float(v))
if not vals:
out.append({"column": name, "numeric": False, "non_empty": sum(1 for r in ws.iter_rows(min_row=2, values_only=True) if r[col-1] is not None)})
continue
out.append({
"column": name,
"numeric": True,
"count": len(vals),
"sum": round(sum(vals), 4),
"mean": round(sum(vals) / len(vals), 4),
"min": round(min(vals), 4),
"max": round(max(vals), 4),
})
return {"ok": True, "file": args.input, "stats": out}
# ---------- 自然语言问 Excel ----------
_ALLOWED_AGGS = {"sum", "mean", "count", "max", "min"}
def _summarize_xlsx(path: str, sample_rows: int = 5) -> dict:
"""读 xlsx 表结构 + 样本行，喂给 AI。"""
wb = _load_wb(path)
ws = wb.active
hmap = _header_map(ws)
sample = []
for i, row in enumerate(ws.iter_rows(min_row=2, max_row=sample_rows + 1, values_only=True), 1):
sample.append({h: row[hmap[h] - 1] if hmap[h] - 1 < len(row) else None for h in hmap})
return {
"sheet": ws.title,
"headers": list(hmap.keys()),
"total_rows": ws.max_row - 1,
"sample": sample,
}
def _exec_plan(df_ops: dict, xlsx_path: str) -> dict:
"""根据 AI 输出的 JSON 操作计划，跑出答案。"""
import openpyxl as _op
wb = _op.load_workbook(xlsx_path, data_only=True)
ws = wb.active
hmap = _header_map(ws)
op = df_ops.get("op", "describe")
if op == "describe":
return _summarize_xlsx(xlsx_path)
if op == "filter":
col = df_ops["column"]
op_ = df_ops.get("op_", "==")
value = df_ops.get("value")
if col not in hmap:
raise ValueError(f"列不存在: {col}")
c = hmap[col] - 1
rows = []
# 把字符串数字转 float 用于比较
def _to_num(x):
if isinstance(x, (int, float)):
return float(x)
try:
return float(str(x))
except (TypeError, ValueError):
return None
for row in ws.iter_rows(min_row=2, values_only=True):
v = row[c] if c < len(row) else None
if op_ in (">", "<", ">=", "<="):
vn = _to_num(v)
cn = _to_num(value)
if vn is None or cn is None:
continue
if op_ == ">" and vn > cn:
rows.append(row)
elif op_ == "<" and vn < cn:
rows.append(row)
elif op_ == ">=" and vn >= cn:
rows.append(row)
elif op_ == "<=" and vn <= cn:
rows.append(row)
else:
if op_ == "==" and str(v) == str(value):
rows.append(row)
elif op_ == "!=" and str(v) != str(value):
rows.append(row)
return {"op": "filter", "column": col, "op_": op_, "value": value, "matched": len(rows),
"preview": [list(r) for r in rows[:10]]}
if op == "group_agg":
g_col = df_ops.get("group") or df_ops.get("group_column") or df_ops.get("column")
v_col = df_ops.get("value") or df_ops.get("value_column") or df_ops.get("value_col")
agg = df_ops.get("agg", "sum")
if not g_col or not v_col:
raise ValueError(f"group_agg 需要 group 和 value 字段；plan={df_ops}")
if g_col not in hmap or v_col not in hmap:
raise ValueError(f"列不存在: group={g_col} value={v_col}")
if agg not in _ALLOWED_AGGS:
raise ValueError(f"聚合不支持: {agg}")
gc = hmap[g_col] - 1
vc = hmap[v_col] - 1
buckets: dict[str, list] = {}
for row in ws.iter_rows(min_row=2, values_only=True):
g = row[gc] if gc < len(row) else None
v = row[vc] if vc < len(row) else None
if v is None:
continue
try:
buckets.setdefault(str(g), []).append(float(v))
except (TypeError, ValueError):
continue
buckets.setdefault(str(g), []).append(float(v))
out = []
for k, vs in buckets.items():
n = len(vs)
if agg == "sum":
r = sum(vs)
elif agg == "mean":
r = sum(vs) / n
elif agg == "count":
r = n
elif agg == "max":
r = max(vs)
else:
r = min(vs)
out.append({g_col: k, f"{v_col}_{agg}": round(r, 4), "count": n})
out.sort(key=lambda x: x.get(f"{v_col}_{agg}", 0), reverse=(agg != "count"))
return {"op": "group_agg", "group": g_col, "value": v_col, "agg": agg,
"groups": len(out), "rows": out}
if op == "top_n":
col = df_ops["column"]
n = int(df_ops.get("n", 5))
order = df_ops.get("order", "desc")
if col not in hmap:
raise ValueError(f"列不存在: {col}")
c = hmap[col] - 1
rows = []
for row in ws.iter_rows(min_row=2, values_only=True):
rows.append(list(row))
rows.sort(key=lambda r: (r[c] if isinstance(r[c], (int, float)) else 0), reverse=(order == "desc"))
return {"op": "top_n", "column": col, "n": n, "order": order, "preview": rows[:n]}
raise ValueError(f"不支持的 op: {op}")
def cmd_nlq(args) -> dict:
"""自然语言问 Excel：AI 转 JSON 操作计划 → 本地执行。"""
if not _AIMS_OK:
raise RuntimeError("nlq 需要 AIMS helper（_lib/aims_chat.py）")
summary = _summarize_xlsx(args.input)
if summary["total_rows"] > args.max_rows:
return {
"ok": False,
"error": f"行数 {summary['total_rows']} 超过限制 {args.max_rows}；请先用 merge/summary 命令缩小数据范围",
}
prompt = (
"你是一个数据分析助手。下面是一份 Excel 的表头、样本行和数据量，请把用户的问题转成 JSON 操作计划。\n"
"严格返回 JSON（不要解释、不要 Markdown 代码块），格式：\n"
"{\"op\": \"describe|filter|group_agg|top_n\", \"column\": \"...\", \"op_\": \">|>=|<|<=|==|!=\", "
"\"value\": ..., \"group\": \"...\", \"value_col\": \"...\", \"agg\": \"sum|mean|count|max|min\", "
"\"n\": 5, \"order\": \"asc|desc\"}\n"
f"--- 表结构 ---\n{json.dumps(summary, ensure_ascii=False, default=str)}\n"
f"--- 用户问题 ---\n{args.question}"
)
plan = chat_text_json(prompt, api_key=args.api_key, max_tokens=2000)
result = _exec_plan(plan, args.input)
return {
"ok": True,
"file": args.input,
"question": args.question,
"plan": plan,
"answer": result,
}
def build_parser() -> argparse.ArgumentParser:
ap = argparse.ArgumentParser(description="Excel 进阶分析工具")
sub = ap.add_subparsers(dest="command", required=True)
p = sub.add_parser("merge", help="合并多个同结构 xlsx")
p.add_argument("--inputs", nargs="+", required=True)
p.add_argument("--output", required=True)
p.set_defaults(func=cmd_merge)
p = sub.add_parser("pivot", help="按列分组聚合")
p.add_argument("--input", required=True)
p.add_argument("--group", required=True, help="分组列名")
p.add_argument("--value", required=True, help="数值列名")
p.add_argument("--agg", default="sum", choices=["sum", "mean", "count", "max", "min"])
p.add_argument("--output", default=None, help="结果另存 xlsx（缺省仅打印）")
p.set_defaults(func=cmd_pivot)
p = sub.add_parser("summary", help="列统计")
p.add_argument("--input", required=True)
p.add_argument("--columns", nargs="+", required=True)
p.set_defaults(func=cmd_summary)
p = sub.add_parser("nlq", help="自然语言问 Excel（依赖 AIMS_API_KEY）")
p.add_argument("--input", required=True)
p.add_argument("--question", required=True, help="中文/英文提问，如「哪个区域销售总额最高」")
p.add_argument("--max-rows", type=int, default=5000, help="数据行数上限，超过拒绝（避免 token 太贵）")
p.add_argument("--api-key", default=None)
p.set_defaults(func=cmd_nlq)
return ap
def main():
args = build_parser().parse_args()
try:
result = args.func(args)
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])