办公文档全能工具箱 office-toolkit：Word 创建/编辑/模板填充、Excel 读写/公式、PDF 合并/拆分/解密/水印、PPT 大纲创建、AI OCR 识别（调平台视觉模型，本机无需装 Tesseract）。一次性覆盖 Word+Excel+PDF+PPT+OCR 五类办公需求，本地执行、确定性计算、按次付费。
---
name: office-toolkit
description: 办公文档全能工具箱 office-toolkit：Word 创建/编辑/查找替换/模板填充、Excel 读写/公式/数据透视、PDF 合并/拆分/提取/压缩/水印/解密、PPT 大纲创建/提取、AI OCR 识别（调 AIMS 平台视觉模型，本机无需装 Tesseract）。一次性覆盖 Word+Excel+PDF+PPT+OCR 五类办公需求，适合行政、人事、财务、销售、学生、运营等多角色日常办公。本地执行、确定性计算、按次付费。
---
本 skill 的真实能力依赖平台后端工具。在网页 / chat 端点下，这些工具可能不可用：
此时**请勿轻信**，请改用 **agent / CLI 路径**运行本 skill 以获取真实结果。任何路径下都**严禁谎称「已调用 / 已搜索完成」而实际未执行**。
处理 Word / Excel / PDF / PPT / OCR 的本地工具集。Word/Excel/PDF/PPT 操作为确定性计算；**OCR 由 AIMS 平台视觉模型（qwen-vl-ocr-latest）完成**，本机无需装 Tesseract。一次安装，五类办公需求全能处理。
**场景 1：行政人员月度汇报** —— Word 写月度总结 → 插入本月销售 Excel 表格 → 加上"内部资料"水印 PDF → 转 PPT 给领导汇报。一次脚本链搞定。
**场景 2：人事批量生成合同** —— 用 `{{占位符}}` 模板批量生成 50 份员工合同（姓名/入职日期/薪资），避免手工复制粘贴。
**场景 3：财务汇总多部门 Excel** —— 12 个部门每月报表合并 → 按部门透视 → 导出汇总表。再用自然语言问"哪些部门金额异常"，AI 自动筛。
| 角色 | 典型用途 |
|------|----------|
| 行政 / 文员 | Word 排版、批量生成通知、PDF 加水印 |
| 人事 / HR | 批量合同、模板填充 |
| 财务 / 会计 | 多表合并、透视汇总、列统计 |
| 销售 / 运营 | 销售数据透视、客户资料整理 |
| 学生 / 教师 | PDF 转 PPT、扫描件 OCR |
| 律师 / 法务 | PDF 解密（自有文档）、合并拆分 |
# 1. 创建 Word 报告
python scripts/docx_ops.py create --output 汇报.docx --title "Q3 季度汇报"
# 3. 追加内容
python scripts/docx_ops.py append --input 汇报.docx --text "本月销售 100 万。"
# 4. 加水印导出 PDF
python scripts/pdf_ops.py watermark --input 汇报.docx --output 汇报_水印.pdf --text "机密"
# 5. 转 PPT
python scripts/pptx_ops.py create --outline outline.json --output 汇报.pptx输出：`{"ok": true, "output": "汇报.pptx", "pdf_pages": 5, "ppt_slides": 5}`
| 对比项 | 传统做法 | office-toolkit |
|--------|---------|----------------|
| 5 类办公需求 | 装 5 个软件 | 一次安装，6 个脚本全包 |
| 模板批量填充 | Word 邮件合并（学习成本高）| CLI 一行命令 |
| PDF 解密 | 买收费工具 / 在线工具（上传风险）| 本地脚本，离线 |
| OCR 识别 | 装 Tesseract（中文差）+ 折腾 | 平台视觉模型，中文友好 |
| AI 问 Excel | 自己写 VBA | 自然语言一句话 |
| 成本 | 多个软件订阅 | 按次付费，单价 ≤ ¥0.30 |
用户要求处理以下任意内容时触发：
pip install -r requirements.txt
# OCR 不再需要本机装 Tesseract，由 AIMS 平台识别
# 需要：环境变量 AIMS_API_KEY=aims_sk_xxx（也可用 --api-key 传入）统一约定：所有脚本输出 JSON，成功 `{"ok": true, ...}`，失败 `{"ok": false, "error": "原因"}`，退出码非 0 即失败。先跑 `check_deps.py` 确认依赖。
| 子命令 | 参数 | 说明 |
|---|---|---|
| create | `--output --title` | 新建文档（可选标题） |
| append | `--input --text [--heading 1-9] [--output]` | 追加段落/标题 |
| table | `--input --rows "a,b\|X,Y" [--delimiter]` | 追加表格（\| 分行为列分隔，首行加粗） |
| replace | `--input --find --replace [--output]` | 全文查找替换（段落+表格） |
| fill | `--input --map "name=张三;date=2026-08-19" [--output]` | 填充 `{{name}}` 占位符 |
示例：
python scripts/docx_ops.py create --output 汇报.docx --title "季度汇报"
python scripts/docx_ops.py append --input 汇报.docx --text "本月销售 100 万。"
python scripts/docx_ops.py table --input 汇报.docx --rows "月份,销售额|1月,100|2月,150"
python scripts/docx_ops.py replace --input 汇报.docx --find "季度" --replace "年度"
python scripts/docx_ops.py fill --input 模板.docx --map "name=张三;date=2026-08-19"| 子命令 | 参数 | 说明 |
|---|---|---|
| create | `--output [--sheet 名称]` | 新建工作簿 |
| write | `--input --rows "a,b\|X,Y"` | 写行（首行可作表头） |
| formula | `--input --cell C3 --expr "=SUM(B2:B3)"` | 单元格写公式 |
| read | `--input [--row-limit]` | 读取为 JSON |
| 子命令 | 参数 | 说明 |
|---|---|---|
| info | `--input [--password]` | 页数/元数据/加密状态 |
| merge | `--inputs f1 f2 ... --output` | 合并多个 PDF |
| split | `--input --pages "1-3,5" --output-dir` | 按页拆分（1 基） |
| extract | `--input [--output] [--page-start] [--page-end]` | 提取文本 |
| compress | `--input --output` | 去重复对象压缩 |
| watermark | `--input --output --text [--font-size] [--opacity] [--angle]` | 加对角水印 |
| decrypt | `--input --output [--password]` | 解密解锁 |
示例：
python scripts/pdf_ops.py merge --inputs a.pdf b.pdf --output 合并.pdf
python scripts/pdf_ops.py split --input a.pdf --pages 1-3,5 --output-dir out/
python scripts/pdf_ops.py watermark --input a.pdf --output w.pdf --text "机密"| 子命令 | 参数 | 说明 |
|---|---|---|
| create | `--outline 大纲.json --output [--title]` | 按大纲创建（含封面/标题/要点/备注） |
| read | `--input` | 提取每页文本 |
| outline | `--input [--output]` | 导出大纲 JSON |
大纲格式：
{"title":"产品发布","cover":{"title":"AI 新品发布会","subtitle":"2026 年秋季"},
"slides":[{"title":"市场背景","bullets":["市场规模 200 亿","年增速 35%"],"notes":"备注"}]}| 子命令 | 参数 | 说明 |
|---|---|---|
| image | `--input [--output] [--api-key]` | 图片 OCR（调 AIMS 平台 qwen-vl-ocr-latest） |
| pdf | `--input [--output] [--force] [--zoom 2.0] [--api-key]` | 扫描版 PDF OCR（无文本层页自动调用 AI；有文本层则直接复用） |
**鉴权**：需 `AIMS_API_KEY` 环境变量或 `--api-key aims_sk_xxx`。**优势**：本机无需装 Tesseract，中英文识别准确率约 100%。
任意 office 文件（docx/xlsx/pptx/pdf）提取文本：`python scripts/extract_office_text.py 文件 [--json]`。
1. **先查依赖**：执行任何操作前跑 `check_deps.py`，缺依赖先安装，不要硬跑。
2. **路径**：输入文件必须存在；输出目录自动创建。中文路径兼容。
3. **输出**：一律 JSON，关键字段含 `output`（生成文件路径）与必要统计（页数/行数/替换次数）。
4. **确定性**：所有转换本地完成，不调用网络、不上传用户文档。
5. **失败处理**：脚本报错即失败（退出码 1），错误信息中文描述原因，把错误原样转达用户并给出修复建议。
<!-- ===== 以下为内嵌脚本代码（agent 安装时按需落盘为 scripts/<name> 并 chmod +x） ===== -->
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""PDF 批量处理：对目录下所有 PDF 批量加水印 / 加密 / 压缩。
用法示例：
python batch_pdf.py watermark --input-dir ./docs/ --output-dir ./out/ --text "机密"
python batch_pdf.py encrypt --input-dir ./docs/ --output-dir ./out/ --password 123456
python batch_pdf.py compress --input-dir ./docs/ --output-dir ./out/
说明：调用 pdf_ops.py 完成单文件操作（复用已验证逻辑），本脚本只做目录遍历与汇总。
依赖：pip install pypdf reportlab（watermark 需要 reportlab）
"""
import argparse
import json
import subprocess
import sys
from pathlib import Path
PDF_OPS = Path(__file__).resolve().parent / "pdf_ops.py"
def _run_pdf_ops(sub_cmd: list[str]) -> dict:
r = subprocess.run([sys.executable, str(PDF_OPS)] + sub_cmd,
capture_output=True, text=True)
try:
return json.loads(r.stdout or '{"ok": false}')
except json.JSONDecodeError:
return {"ok": False, "error": (r.stderr or r.stdout or "unknown")[:200]}
def _collect_pdfs(input_dir: str) -> list[Path]:
d = Path(input_dir)
if not d.is_dir():
raise ValueError(f"目录不存在: {input_dir}")
pdfs = sorted(d.glob("*.pdf"))
if not pdfs:
raise ValueError(f"目录下没有 PDF 文件: {input_dir}")
return pdfs
def cmd_watermark(args) -> dict:
pdfs = _collect_pdfs(args.input_dir)
out_dir = Path(args.output_dir)
out_dir.mkdir(parents=True, exist_ok=True)
results, failed = [], []
for p in pdfs:
out = out_dir / f"{p.stem}_水印.pdf"
r = _run_pdf_ops(["watermark", "--input", str(p), "--output", str(out),
"--text", args.text, "--opacity", str(args.opacity)])
(results if r.get("ok") else failed).append({"file": str(p), **r})
return {"ok": len(failed) == 0, "total": len(pdfs), "succeeded": len(results),
"failed": failed, "output_dir": str(out_dir)}
def cmd_encrypt(args) -> dict:
pdfs = _collect_pdfs(args.input_dir)
out_dir = Path(args.output_dir)
out_dir.mkdir(parents=True, exist_ok=True)
results, failed = [], []
for p in pdfs:
out = out_dir / f"{p.stem}_加密.pdf"
cmd = ["encrypt", "--input", str(p), "--output", str(out), "--password", args.password]
r = _run_pdf_ops(cmd)
(results if r.get("ok") else failed).append({"file": str(p), **r})
return {"ok": len(failed) == 0, "total": len(pdfs), "succeeded": len(results),
"failed": failed, "output_dir": str(out_dir)}
def cmd_compress(args) -> dict:
pdfs = _collect_pdfs(args.input_dir)
out_dir = Path(args.output_dir)
out_dir.mkdir(parents=True, exist_ok=True)
results, failed = [], []
for p in pdfs:
out = out_dir / f"{p.stem}_压缩.pdf"
r = _run_pdf_ops(["compress", "--input", str(p), "--output", str(out)])
(results if r.get("ok") else failed).append({"file": str(p), **r})
return {"ok": len(failed) == 0, "total": len(pdfs), "succeeded": len(results),
"failed": failed, "output_dir": str(out_dir)}
def build_parser() -> argparse.ArgumentParser:
ap = argparse.ArgumentParser(description="PDF 批量处理工具")
sub = ap.add_subparsers(dest="command", required=True)
p = sub.add_parser("watermark", help="批量加水印")
p.add_argument("--input-dir", required=True)
p.add_argument("--output-dir", required=True)
p.add_argument("--text", required=True, help="水印文字")
p.add_argument("--opacity", type=float, default=0.25)
p.set_defaults(func=cmd_watermark)
p = sub.add_parser("encrypt", help="批量加密（打开密码）")
p.add_argument("--input-dir", required=True)
p.add_argument("--output-dir", required=True)
p.add_argument("--password", required=True)
p.set_defaults(func=cmd_encrypt)
p = sub.add_parser("compress", help="批量压缩")
p.add_argument("--input-dir", required=True)
p.add_argument("--output-dir", required=True)
p.set_defaults(func=cmd_compress)
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()#!/usr/bin/env python3
import importlib
import shutil
import subprocess
import sys
PYTHON_MODULES = [
("pypdf", "PDF embedded text extraction"),
("docx", "DOCX extraction via python-docx"),
("openpyxl", "XLSX extraction"),
("pptx", "PPTX extraction via python-pptx"),
]
SYSTEM_TOOLS = [
("pdftoppm", "PDF rasterization for OCR (from poppler-utils)"),
("tesseract", "OCR engine"),
("libreoffice", "Legacy Office conversion (.doc/.xls/.ppt)"),
("antiword", "Legacy .doc extraction fallback"),
("catdoc", "Legacy .doc extraction fallback"),
]
RECOMMENDED_LANGS = ["chi_sim", "eng"]
def check_python_modules():
print("[Python modules]")
missing = []
for mod, desc in PYTHON_MODULES:
try:
importlib.import_module(mod)
print(f"OK {mod:<10} - {desc}")
except Exception as e:
print(f"MISS {mod:<10} - {desc} ({e})")
missing.append(mod)
print()
return missing
def check_system_tools():
print("[System tools]")
missing = []
for tool, desc in SYSTEM_TOOLS:
path = shutil.which(tool)
if path:
print(f"OK {tool:<10} - {desc} [{path}]")
else:
print(f"MISS {tool:<10} - {desc}")
missing.append(tool)
print()
return missing
def check_tesseract_langs():
print("[Tesseract languages]")
if not shutil.which("tesseract"):
print("SKIP tesseract not installed")
print()
return RECOMMENDED_LANGS[:]
try:
r = subprocess.run(
["tesseract", "--list-langs"],
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
text=True,
check=False,
)
except Exception as e:
print(f"ERROR unable to query tesseract languages: {e}")
print()
return RECOMMENDED_LANGS[:]
lines = [x.strip() for x in (r.stdout or "").splitlines() if x.strip()]
if lines and lines[0].lower().startswith("list of available"):
lines = lines[1:]
langs = set(lines)
missing = []
for lang in RECOMMENDED_LANGS:
if lang in langs:
print(f"OK {lang}")
else:
print(f"MISS {lang}")
missing.append(lang)
print()
return missing
def main():
py_missing = check_python_modules()
tool_missing = check_system_tools()
lang_missing = check_tesseract_langs()
print("[Summary]")
if not py_missing and not tool_missing and not lang_missing:
print("All recommended dependencies are available.")
sys.exit(0)
if py_missing:
print("Missing Python modules:", ", ".join(py_missing))
if tool_missing:
print("Missing system tools:", ", ".join(tool_missing))
if lang_missing:
print("Missing OCR languages:", ", ".join(lang_missing))
print("\nMinimum modern-document support requires Python modules only.")
print("Best real-world coverage also needs pdftoppm, tesseract, libreoffice, antiword/catdoc, and chi_sim OCR.")
sys.exit(1)
if __name__ == "__main__":
main()#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""合同/模板批量生成：扫描 {{占位符}} → 提取字段 → 批量填值生成多份 docx。
用法示例：
python contract.py schema --template 合同模板.docx # 列出模板里的占位符字段
python contract.py generate --template 合同模板.docx --data 合同数据.json --output-dir out/
python contract.py generate --template 合同模板.docx --row "甲方=张三;金额=10000" --output 合同_张三.docx
--data JSON 格式：{"items": [{"甲方": "张三", "金额": "10000"}, ...]} 或直接数组 [{...}, {...}]
依赖：pip install python-docx
"""
import argparse
import json
import re
import sys
from pathlib import Path
try:
from docx import Document
except ImportError: # pragma: no cover
sys.stderr.write("缺少依赖 python-docx，请先执行: pip install python-docx\n")
sys.exit(2)
PLACEHOLDER = re.compile(r"\{\{\s*([^}]+?)\s*\}\}")
def _iter_cells(container):
"""迭代段落与表格单元格。"""
for p in container.paragraphs:
yield p
for table in getattr(container, "tables", []):
for row in table.rows:
for cell in row.cells:
yield from _iter_cells(cell)
def _replace_in_run(run, mapping, missing):
text = run.text
for m in PLACEHOLDER.finditer(text):
key = m.group(1)
if key not in mapping:
missing.add(key)
continue
text = text.replace(m.group(0), str(mapping[key]))
run.text = text
def _collect_fields(container) -> set:
fields = set()
for p in _iter_cells(container):
fields.update(PLACEHOLDER.findall(p.text))
return fields
def cmd_schema(args) -> dict:
doc = Document(args.template)
fields = sorted(_collect_fields(doc))
return {"ok": True, "template": args.template, "fields": fields, "count": len(fields)}
def _generate_one(template, mapping, output):
doc = Document(template)
missing = set()
for p in _iter_cells(doc):
for run in p.runs:
_replace_in_run(run, mapping, missing)
out = Path(output)
out.parent.mkdir(parents=True, exist_ok=True)
doc.save(str(out))
return str(out), missing
def cmd_generate(args) -> dict:
if args.data:
payload = json.loads(Path(args.data).read_text(encoding="utf-8"))
elif args.row:
mapping = {}
for pair in args.row.split(";"):
if "=" in pair:
k, v = pair.split("=", 1)
mapping[k.strip()] = v.strip()
payload = [mapping]
else:
raise ValueError("需要 --data 或 --row")
items = payload.get("items", payload) if isinstance(payload, dict) else payload
if not isinstance(items, list) or not items:
raise ValueError("数据应为数组或 {'items': [...]}")
out_dir = Path(args.output_dir) if args.output_dir else None
generated = []
all_missing = set()
for i, item in enumerate(items, 1):
mapping = {str(k): str(v) if v is not None else "" for k, v in item.items()}
if out_dir:
out = out_dir / f"合同_{i:03d}.docx"
elif args.output:
out = Path(args.output)
else:
raise ValueError("需要 --output-dir 或 --output")
path, missing = _generate_one(args.template, mapping, str(out))
generated.append(path)
all_missing.update(missing)
return {
"ok": True,
"count": len(generated),
"files": generated,
"missing_fields": sorted(all_missing) if all_missing else [],
"note": "缺失字段未替换（模板中保留 {{占位符}}）；建议先跑 schema 确认字段" if all_missing else "全部字段已填充",
}
def build_parser() -> argparse.ArgumentParser:
ap = argparse.ArgumentParser(description="合同/模板批量生成工具")
sub = ap.add_subparsers(dest="command", required=True)
p = sub.add_parser("schema", help="扫描模板占位符字段")
p.add_argument("--template", required=True)
p.set_defaults(func=cmd_schema)
p = sub.add_parser("generate", help="批量生成（数据数组→多份 docx）")
p.add_argument("--template", required=True)
p.add_argument("--data", default=None, help="JSON 数据文件")
p.add_argument("--row", default=None, help='单份数据 "甲方=张三;金额=10000"')
p.add_argument("--output-dir", default=None)
p.add_argument("--output", default=None)
p.set_defaults(func=cmd_generate)
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()#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""格式转换：xlsx↔csv、docx→markdown、图片→PDF。
用法示例：
python convert_ops.py xlsx2csv --input a.xlsx --output a.csv
python convert_ops.py csv2xlsx --input a.csv --output a.xlsx [--sheet 数据]
python convert_ops.py docx2md --input 报告.docx --output 报告.md
python convert_ops.py images2pdf --inputs a.png b.jpg --output 扫描件.pdf
依赖：pip install openpyxl python-docx Pillow pypdf
"""
import argparse
import csv
import json
import sys
from pathlib import Path
try:
import openpyxl
from docx import Document
from PIL import Image
except ImportError: # pragma: no cover
sys.stderr.write("缺少依赖，请先执行: pip install openpyxl python-docx Pillow\n")
sys.exit(2)
def cmd_xlsx2csv(args) -> dict:
wb = openpyxl.load_workbook(args.input, data_only=True)
ws = wb[args.sheet] if args.sheet else wb.active
out = Path(args.output)
out.parent.mkdir(parents=True, exist_ok=True)
with out.open("w", encoding="utf-8-sig", newline="") as f:
w = csv.writer(f)
for row in ws.iter_rows(values_only=True):
w.writerow(["" if v is None else v for v in row])
return {"ok": True, "output": str(out), "sheet": ws.title}
def cmd_csv2xlsx(args) -> dict:
wb = openpyxl.Workbook()
ws = wb.active
ws.title = args.sheet
with open(args.input, encoding=args.encoding, newline="") as f:
for row in csv.reader(f):
ws.append(row)
out = Path(args.output)
out.parent.mkdir(parents=True, exist_ok=True)
wb.save(str(out))
return {"ok": True, "output": str(out), "sheet": ws.title}
def cmd_docx2md(args) -> dict:
doc = Document(args.input)
lines = []
for para in doc.paragraphs:
text = para.text.rstrip()
if not text:
continue
style = (para.style.name or "").lower()
if style.startswith("heading 1"):
lines.append(f"# {text}")
elif style.startswith("heading 2"):
lines.append(f"## {text}")
elif style.startswith("heading 3"):
lines.append(f"### {text}")
else:
lines.append(text)
for table in doc.tables:
md_rows = []
for row in table.rows:
md_rows.append("| " + " | ".join(c.text.strip() for c in row.cells) + " |")
if md_rows:
lines.append("")
lines.append(md_rows[0])
n_cols = len(md_rows[0].split("|")) - 2 # 首尾空串
lines.append("|" + "---|" * n_cols)
lines.extend(md_rows[1:])
lines.append("")
out = Path(args.output)
out.parent.mkdir(parents=True, exist_ok=True)
out.write_text("\n\n".join(lines), encoding="utf-8")
return {"ok": True, "output": str(out), "chars": len("\n\n".join(lines))}
def cmd_images2pdf(args) -> dict:
imgs = []
for p in args.inputs:
im = Image.open(p)
if im.mode in ("RGBA", "P", "LA"):
im = im.convert("RGB")
imgs.append(im)
if not imgs:
raise ValueError("没有可转换的图片")
out = Path(args.output)
out.parent.mkdir(parents=True, exist_ok=True)
first, rest = imgs[0], imgs[1:]
first.save(str(out), "PDF", save_all=True, append_images=rest, resolution=args.dpi)
return {"ok": True, "output": str(out), "images": len(imgs), "bytes": out.stat().st_size}
def build_parser() -> argparse.ArgumentParser:
ap = argparse.ArgumentParser(description="格式转换工具")
sub = ap.add_subparsers(dest="command", required=True)
p = sub.add_parser("xlsx2csv")
p.add_argument("--input", required=True)
p.add_argument("--output", required=True)
p.add_argument("--sheet", default=None)
p.set_defaults(func=cmd_xlsx2csv)
p = sub.add_parser("csv2xlsx")
p.add_argument("--input", required=True)
p.add_argument("--output", required=True)
p.add_argument("--sheet", default="Sheet1")
p.add_argument("--encoding", default="utf-8-sig")
p.set_defaults(func=cmd_csv2xlsx)
p = sub.add_parser("docx2md")
p.add_argument("--input", required=True)
p.add_argument("--output", required=True)
p.set_defaults(func=cmd_docx2md)
p = sub.add_parser("images2pdf")
p.add_argument("--inputs", nargs="+", required=True)
p.add_argument("--output", required=True)
p.add_argument("--dpi", type=int, default=150)
p.set_defaults(func=cmd_images2pdf)
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()#!/usr/bin/env python3
"""Word (.docx) create / edit operations for the office-toolkit skill.
Adapted from OpenClaw Office Assistant (MIT, Copyright (c) 2026 Windrunner20);
extended with creation/editing capabilities. Local, deterministic, no network.
"""
import argparse
import csv
import io
import json
import sys
from pathlib import Path
from docx import Document
from docx.shared import Pt, RGBColor
from docx.enum.text import WD_ALIGN_PARAGRAPH
def _load(path):
p = Path(path).expanduser().resolve()
if not p.exists():
raise FileNotFoundError(f"File not found: {p}")
return p
def _save(doc, out):
o = Path(out).expanduser().resolve()
o.parent.mkdir(parents=True, exist_ok=True)
doc.save(str(o))
return str(o)
def cmd_create(args):
doc = Document()
if args.title:
h = doc.add_heading(args.title, level=0)
h.alignment = WD_ALIGN_PARAGRAPH.CENTER
out = _save(doc, args.output)
return {"ok": True, "file": out, "action": "create"}
def cmd_append(args):
doc = Document(str(_load(args.input)))
if args.heading:
doc.add_heading(args.text, level=args.heading)
else:
doc.add_paragraph(args.text)
out = _save(doc, args.output or args.input)
return {"ok": True, "file": out, "action": "append", "chars": len(args.text)}
def cmd_table(args):
doc = Document(str(_load(args.input)))
rows = [r.split(args.delimiter) for r in args.rows.split("|") if r.strip()]
if not rows:
return {"ok": False, "error": "no rows provided"}
n_rows, n_cols = len(rows), max(len(r) for r in rows)
table = doc.add_table(rows=n_rows, cols=n_cols)
table.style = "Light Grid Accent 1"
for i, row in enumerate(rows):
for j in range(n_cols):
cell = table.cell(i, j)
cell.text = row[j] if j < len(row) else ""
if i == 0:
for run in cell.paragraphs[0].runs:
run.bold = True
out = _save(doc, args.output or args.input)
return {"ok": True, "file": out, "action": "table", "rows": n_rows, "cols": n_cols}
def _replace_in_runs(container, find, replace):
count = 0
for p in container.paragraphs:
for run in p.runs:
if find in run.text:
run.text = run.text.replace(find, replace)
count += 1
return count
def cmd_replace(args):
doc = Document(str(_load(args.input)))
total = 0
total += _replace_in_runs(doc, args.find, args.replace)
for table in doc.tables:
for row in table.rows:
for cell in row.cells:
total += _replace_in_runs(cell, args.find, args.replace)
out = _save(doc, args.output or args.input)
return {"ok": True, "file": out, "action": "replace", "replacements": total}
def cmd_fill(args):
doc = Document(str(_load(args.input)))
mapping = {}
for pair in args.map.split(";"):
if "=" in pair:
k, v = pair.split("=", 1)
mapping["{{" + k.strip() + "}}"] = v.strip()
total = 0
for placeholder, value in mapping.items():
total += _replace_in_runs(doc, placeholder, value)
for table in doc.tables:
for row in table.rows:
for cell in row.cells:
total += _replace_in_runs(cell, placeholder, value)
out = _save(doc, args.output or args.input)
return {"ok": True, "file": out, "action": "fill", "replacements": total}
def build_parser():
ap = argparse.ArgumentParser(description="Word (.docx) create/edit toolkit.")
sub = ap.add_subparsers(dest="cmd", required=True)
p = sub.add_parser("create", help="Create a new .docx (optionally with title)")
p.add_argument("--output", required=True)
p.add_argument("--title", default="")
p.set_defaults(func=cmd_create)
p = sub.add_parser("append", help="Append paragraph or heading")
p.add_argument("--input", required=True)
p.add_argument("--output", default="")
p.add_argument("--text", required=True)
p.add_argument("--heading", type=int, default=0, help="0=paragraph, 1-9=heading level")
p.set_defaults(func=cmd_append)
p = sub.add_parser("table", help="Append a table (rows delimited by |, cells by comma)")
p.add_argument("--input", required=True)
p.add_argument("--output", default="")
p.add_argument("--rows", required=True, help="a,b,c|X,Y,Z")
p.add_argument("--delimiter", default=",")
p.set_defaults(func=cmd_table)
p = sub.add_parser("replace", help="Find/replace text in paragraphs and tables")
p.add_argument("--input", required=True)
p.add_argument("--output", default="")
p.add_argument("--find", required=True)
p.add_argument("--replace", default="")
p.set_defaults(func=cmd_replace)
p = sub.add_parser("fill", help="Fill {{key}} placeholders from key=value; pairs")
p.add_argument("--input", required=True)
p.add_argument("--output", default="")
p.add_argument("--map", required=True, help="name=Alice;date=2026-08-19")
p.set_defaults(func=cmd_fill)
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()#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""Word 排版美化：页眉页脚 / 页码 / 标题样式 / 正文字体。
用法示例：
python docx_style.py header --input a.docx --text "内部资料" [--output]
python docx_style.py footer --input a.docx --text "第 1 页" [--output]
python docx_style.py pagenum --input a.docx --align center [--output]
python docx_style.py heading --input a.docx --level 1 --size 22 --color 1F4E79 [--output]
python docx_style.py font --input a.docx --size 12 --name 宋体 [--output]
依赖：pip install python-docx
"""
import argparse
import json
import sys
from pathlib import Path
try:
from docx import Document
from docx.shared import Pt, RGBColor
from docx.enum.text import WD_ALIGN_PARAGRAPH
from docx.oxml.ns import qn
from docx.oxml import OxmlElement
except ImportError: # pragma: no cover
sys.stderr.write("缺少依赖 python-docx，请先执行: pip install python-docx\n")
sys.exit(2)
def _load(path) -> Document:
p = Path(path)
if not p.exists():
raise FileNotFoundError(f"文件不存在: {p}")
return Document(str(p))
def _save(doc, out, input_path):
o = Path(out or input_path)
o.parent.mkdir(parents=True, exist_ok=True)
doc.save(str(o))
return str(o)
def _set_cn_font(run, name: str | None, size: int | None):
if name:
run.font.name = name
r = run._element.rPr.rFonts
r.set(qn("w:eastAsia"), name)
if size:
run.font.size = Pt(size)
def _header_para(doc, align):
sec = doc.sections[0]
p = sec.header.paragraphs[0]
p.alignment = {"left": WD_ALIGN_PARAGRAPH.LEFT, "center": WD_ALIGN_PARAGRAPH.CENTER,
"right": WD_ALIGN_PARAGRAPH.RIGHT}.get(align, WD_ALIGN_PARAGRAPH.LEFT)
return p
def cmd_header(args) -> dict:
doc = _load(args.input)
p = _header_para(doc, args.align)
run = p.add_run(args.text)
if args.size:
run.font.size = Pt(args.size)
out = _save(doc, args.output, args.input)
return {"ok": True, "output": out, "action": "header", "text": args.text}
def cmd_footer(args) -> dict:
doc = _load(args.input)
sec = doc.sections[0]
p = sec.footer.paragraphs[0]
p.alignment = {"left": WD_ALIGN_PARAGRAPH.LEFT, "center": WD_ALIGN_PARAGRAPH.CENTER,
"right": WD_ALIGN_PARAGRAPH.RIGHT}.get(args.align, WD_ALIGN_PARAGRAPH.LEFT)
run = p.add_run(args.text)
if args.size:
run.font.size = Pt(args.size)
out = _save(doc, args.output, args.input)
return {"ok": True, "output": out, "action": "footer", "text": args.text}
def _add_page_field(p):
"""插入 PAGE 域（显示当前页码）。"""
run = p.add_run()
fld = OxmlElement("w:fldSimple")
fld.set(qn("w:instr"), "PAGE")
run._r.append(fld)
def cmd_pagenum(args) -> dict:
doc = _load(args.input)
for sec in doc.sections:
sec.footer.is_linked_to_previous = False
p = sec.footer.paragraphs[0]
p.alignment = {"left": WD_ALIGN_PARAGRAPH.LEFT, "center": WD_ALIGN_PARAGRAPH.CENTER,
"right": WD_ALIGN_PARAGRAPH.RIGHT}.get(args.align, WD_ALIGN_PARAGRAPH.CENTER)
if args.prefix:
p.add_run(args.prefix)
_add_page_field(p)
if args.suffix:
p.add_run(args.suffix)
out = _save(doc, args.output, args.input)
return {"ok": True, "output": out, "action": "pagenum", "note": "已插入页码域（PAGE），打开后自动显示"}
def cmd_heading(args) -> dict:
doc = _load(args.input)
style_name = f"Heading {args.level}"
try:
style = doc.styles[style_name]
except KeyError:
raise ValueError(f"样式不存在: {style_name}")
font = style.font
font.size = Pt(args.size) if args.size else font.size
if args.color:
font.color.rgb = RGBColor.from_string(args.color)
if args.name:
font.name = args.name
style.element.rPr.rFonts.set(qn("w:eastAsia"), args.name)
out = _save(doc, args.output, args.input)
return {"ok": True, "output": out, "action": "heading", "style": style_name}
def cmd_font(args) -> dict:
doc = _load(args.input)
for p in doc.paragraphs:
if not p.text.strip():
continue
for run in p.runs:
_set_cn_font(run, args.name, args.size)
out = _save(doc, args.output, args.input)
return {"ok": True, "output": out, "action": "font", "name": args.name, "size": args.size}
def build_parser() -> argparse.ArgumentParser:
ap = argparse.ArgumentParser(description="Word 排版美化工具")
sub = ap.add_subparsers(dest="command", required=True)
common = {"input": True}
p = sub.add_parser("header", help="设置页眉文本")
p.add_argument("--input", required=True)
p.add_argument("--text", required=True)
p.add_argument("--align", default="left", choices=["left", "center", "right"])
p.add_argument("--size", type=int, default=None)
p.add_argument("--output", default=None)
p.set_defaults(func=cmd_header)
p = sub.add_parser("footer", help="设置页脚文本")
p.add_argument("--input", required=True)
p.add_argument("--text", required=True)
p.add_argument("--align", default="center", choices=["left", "center", "right"])
p.add_argument("--size", type=int, default=None)
p.add_argument("--output", default=None)
p.set_defaults(func=cmd_footer)
p = sub.add_parser("pagenum", help="插入页码域")
p.add_argument("--input", required=True)
p.add_argument("--align", default="center", choices=["left", "center", "right"])
p.add_argument("--prefix", default="第 ")
p.add_argument("--suffix", default=" 页")
p.add_argument("--output", default=None)
p.set_defaults(func=cmd_pagenum)
p = sub.add_parser("heading", help="设置标题样式")
p.add_argument("--input", required=True)
p.add_argument("--level", type=int, default=1, choices=[1, 2, 3])
p.add_argument("--size", type=int, default=None)
p.add_argument("--color", default=None, help="RRGGBB 如 1F4E79")
p.add_argument("--name", default=None, help="字体名（含中文）")
p.add_argument("--output", default=None)
p.set_defaults(func=cmd_heading)
p = sub.add_parser("font", help="设置正文字体/字号")
p.add_argument("--input", required=True)
p.add_argument("--size", type=int, default=None)
p.add_argument("--name", default=None)
p.add_argument("--output", default=None)
p.set_defaults(func=cmd_font)
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()#!/usr/bin/env python3
import argparse
import json
import os
import shutil
import subprocess
import sys
import tempfile
from pathlib import Path
def run(cmd):
return subprocess.run(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True)
def available_tess_langs():
r = run(["tesseract", "--list-langs"])
if r.returncode != 0:
return []
lines = [x.strip() for x in r.stdout.splitlines() if x.strip()]
if lines and lines[0].lower().startswith("list of available"):
lines = lines[1:]
return lines
def extract_pdf(path: Path, page_limit: int):
from pypdf import PdfReader
reader = PdfReader(str(path))
page_count = len(reader.pages)
texts = []
for i, page in enumerate(reader.pages[:page_limit], start=1):
try:
txt = page.extract_text() or ""
except Exception:
txt = ""
txt = txt.strip()
if txt:
texts.append(f"\n## Page {i}\n{txt}\n")
joined = "\n".join(texts).strip()
used_ocr = False
if len(joined) >= 200:
return {
"type": "pdf",
"page_count": page_count,
"extraction": "text",
"text": joined,
}
langs = available_tess_langs()
tess_lang = None
if "chi_sim" in langs and "eng" in langs:
tess_lang = "chi_sim+eng"
elif "chi_sim" in langs:
tess_lang = "chi_sim"
elif "eng" in langs:
tess_lang = "eng"
if not shutil.which("pdftoppm") or not shutil.which("tesseract") or not tess_lang:
return {
"type": "pdf",
"page_count": page_count,
"extraction": "text-empty",
"text": joined,
"warning": "No usable OCR pipeline available; install poppler-utils, tesseract-ocr, and tesseract language packs (prefer chi_sim+eng).",
}
used_ocr = True
with tempfile.TemporaryDirectory(prefix="office-doc-pdf-") as td:
prefix = os.path.join(td, "page")
cmd = ["pdftoppm", "-f", "1", "-l", str(min(page_limit, page_count)), "-r", "180", "-png", str(path), prefix]
r = run(cmd)
if r.returncode != 0:
return {
"type": "pdf",
"page_count": page_count,
"extraction": "ocr-failed",
"text": joined,
"warning": r.stderr.strip()[:1000],
}
page_texts = []
for img in sorted(Path(td).glob("page-*.png")):
rr = run(["tesseract", str(img), "stdout", "-l", tess_lang])
txt = (rr.stdout or "").strip()
if txt:
page_texts.append(f"\n## {img.stem}\n{txt}\n")
ocr_text = "\n".join(page_texts).strip()
combined = ocr_text if len(ocr_text) > len(joined) else joined
return {
"type": "pdf",
"page_count": page_count,
"extraction": "ocr" if used_ocr else "text",
"ocr_lang": tess_lang,
"text": combined,
}
def extract_docx(path: Path):
from docx import Document
doc = Document(str(path))
parts = []
for p in doc.paragraphs:
t = p.text.strip()
if t:
parts.append(t)
for table_idx, table in enumerate(doc.tables, start=1):
rows = []
for row in table.rows:
cells = [c.text.strip().replace("\n", " ") for c in row.cells]
if any(cells):
rows.append(" | ".join(cells))
if rows:
parts.append(f"\n[Table {table_idx}]\n" + "\n".join(rows))
return {"type": "docx", "text": "\n\n".join(parts).strip()}
def extract_doc(path: Path):
for cmd in (["antiword", str(path)], ["catdoc", str(path)]):
if shutil.which(cmd[0]):
r = run(cmd)
if r.returncode == 0 and r.stdout.strip():
return {"type": "doc", "extraction": cmd[0], "text": r.stdout.strip()}
if shutil.which("libreoffice"):
with tempfile.TemporaryDirectory(prefix="office-doc-doc-") as td:
r = run(["libreoffice", "--headless", "--convert-to", "docx", "--outdir", td, str(path)])
out = Path(td) / (path.stem + ".docx")
if out.exists():
data = extract_docx(out)
data["type"] = "doc"
data["extraction"] = "libreoffice->docx"
return data
return {"type": "doc", "warning": (r.stderr or r.stdout).strip()[:1000], "text": ""}
return {"type": "doc", "warning": "No DOC extractor found (antiword/catdoc/libreoffice).", "text": ""}
def sheet_to_lines(ws, row_limit):
rows = []
max_row = min(ws.max_row or 0, row_limit)
max_col = ws.max_column or 0
for r in ws.iter_rows(min_row=1, max_row=max_row, min_col=1, max_col=max_col, values_only=True):
vals = ["" if v is None else str(v).strip() for v in r]
if any(vals):
rows.append(" | ".join(vals))
return rows
def extract_xlsx(path: Path, row_limit: int):
from openpyxl import load_workbook
wb = load_workbook(str(path), read_only=True, data_only=True)
parts = []
for name in wb.sheetnames:
ws = wb[name]
lines = sheet_to_lines(ws, row_limit)
parts.append(f"\n# Sheet: {name}\n")
if lines:
parts.extend(lines)
else:
parts.append("(empty)")
return {"type": "xlsx", "sheets": wb.sheetnames, "text": "\n".join(parts).strip()}
def extract_xls(path: Path, row_limit: int):
if shutil.which("libreoffice"):
with tempfile.TemporaryDirectory(prefix="office-doc-xls-") as td:
r = run(["libreoffice", "--headless", "--convert-to", "xlsx", "--outdir", td, str(path)])
out = Path(td) / (path.stem + ".xlsx")
if out.exists():
data = extract_xlsx(out, row_limit)
data["type"] = "xls"
data["extraction"] = "libreoffice->xlsx"
return data
return {"type": "xls", "warning": (r.stderr or r.stdout).strip()[:1000], "text": ""}
return {"type": "xls", "warning": "No XLS converter found (libreoffice).", "text": ""}
def extract_pptx(path: Path):
from pptx import Presentation
prs = Presentation(str(path))
parts = []
for idx, slide in enumerate(prs.slides, start=1):
lines = []
for shape in slide.shapes:
txt = getattr(shape, "text", "")
txt = (txt or "").strip()
if txt:
lines.append(txt)
if slide.has_notes_slide:
notes_texts = []
for shape in slide.notes_slide.shapes:
txt = getattr(shape, "text", "")
txt = (txt or "").strip()
if txt:
notes_texts.append(txt)
if notes_texts:
lines.append("[Notes]\n" + "\n".join(notes_texts))
if lines:
parts.append(f"\n# Slide {idx}\n" + "\n".join(lines))
else:
parts.append(f"\n# Slide {idx}\n(no extractable text)")
return {"type": "pptx", "slide_count": len(prs.slides), "text": "\n\n".join(parts).strip()}
def extract_ppt(path: Path):
if shutil.which("libreoffice"):
with tempfile.TemporaryDirectory(prefix="office-doc-ppt-") as td:
r = run(["libreoffice", "--headless", "--convert-to", "pptx", "--outdir", td, str(path)])
out = Path(td) / (path.stem + ".pptx")
if out.exists():
data = extract_pptx(out)
data["type"] = "ppt"
data["extraction"] = "libreoffice->pptx"
return data
return {"type": "ppt", "warning": (r.stderr or r.stdout).strip()[:1000], "text": ""}
return {"type": "ppt", "warning": "No PPT converter found (libreoffice).", "text": ""}
def main():
ap = argparse.ArgumentParser(description="Extract text from office documents for OpenClaw skills.")
ap.add_argument("file")
ap.add_argument("--page-limit", type=int, default=20)
ap.add_argument("--row-limit", type=int, default=50)
ap.add_argument("--max-chars", type=int, default=50000)
ap.add_argument("--json", action="store_true")
args = ap.parse_args()
path = Path(args.file).expanduser().resolve()
if not path.exists():
print(json.dumps({"error": f"File not found: {path}"}, ensure_ascii=False))
sys.exit(2)
suffix = path.suffix.lower()
if suffix == ".pdf":
data = extract_pdf(path, args.page_limit)
elif suffix == ".docx":
data = extract_docx(path)
elif suffix == ".doc":
data = extract_doc(path)
elif suffix == ".xlsx":
data = extract_xlsx(path, args.row_limit)
elif suffix == ".xls":
data = extract_xls(path, args.row_limit)
elif suffix == ".pptx":
data = extract_pptx(path)
elif suffix == ".ppt":
data = extract_ppt(path)
else:
data = {"error": f"Unsupported extension: {suffix}"}
data["file"] = str(path)
text = data.get("text", "") or ""
if len(text) > args.max_chars:
data["text"] = text[: args.max_chars] + "\n\n[TRUNCATED]"
data["truncated"] = True
if args.json:
print(json.dumps(data, ensure_ascii=False, indent=2))
else:
meta = {k: v for k, v in data.items() if k != "text"}
print(json.dumps(meta, ensure_ascii=False, indent=2))
print("\n---TEXT---\n")
print(data.get("text", ""))
if __name__ == "__main__":
main()#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""OCR 识别脚本（平台视觉模型版）：图片 / 扫描版 PDF 文字识别。
原理：把图片用 base64 传给 AIMS 平台的视觉模型做 OCR，识别在平台完成，
本地无需安装 Tesseract。
用法示例：
python ocr.py image --input scan.png --output out.txt
python ocr.py pdf --input scan.pdf --output out.txt
依赖：pip install pymupdf（仅 pdf 模式需要）
鉴权：AIMS_API_KEY 环境变量 或 --api-key aims_sk_xxx
"""
import argparse
import json
import sys
from pathlib import Path
# 让脚本既可作为 office-toolkit 子模块运行，也可独立运行
_HERE = Path(__file__).resolve().parent
_ROOT = _HERE.parent.parent # office-skills/
sys.path.insert(0, str(_ROOT))
# _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 存在。"
)
from aims_chat import chat_vision, chat_pdf_page # noqa: E402
OCR_PROMPT = (
"请逐行转录图片里的全部文字（含中文、英文、数字、标点），"
"保持原顺序，只输出识别到的文字，不要解释、不要加编号。"
)
def cmd_image(args) -> dict:
text = chat_vision(args.input, OCR_PROMPT, api_key=args.api_key)
_save(text, args.output)
return {"ok": True, "output": args.output, "chars": len(text), "preview": text[:300]}
def cmd_pdf(args) -> dict:
try:
import fitz
except ImportError:
raise RuntimeError("PDF 模式需要 PyMuPDF：pip install pymupdf")
doc = fitz.open(args.input)
parts = []
chars = 0
for i in range(doc.page_count):
# 优先用文本层
if not args.force:
txt0 = doc[i].get_text().strip()
if txt0:
parts.append(f"---- 第 {i + 1} 页（文本层）----\n{txt0}")
chars += len(txt0)
continue
# 无文本层 → 渲染成图 → 视觉 OCR
text = chat_pdf_page(args.input, i, OCR_PROMPT, zoom=args.zoom, api_key=args.api_key)
parts.append(f"---- 第 {i + 1} 页 ----\n{text}")
chars += len(text)
body = "\n\n".join(parts)
_save(body, args.output)
return {"ok": True, "output": args.output, "pages": doc.page_count, "chars": chars, "preview": body[:300]}
def _save(text: str, output: str | None):
if output:
Path(output).parent.mkdir(parents=True, exist_ok=True)
Path(output).write_text(text, encoding="utf-8")
def build_parser() -> argparse.ArgumentParser:
ap = argparse.ArgumentParser(description="OCR 识别（平台视觉模型）")
ap.add_argument("--api-key", default=None, help="AIMS API Key（也可用环境变量 AIMS_API_KEY）")
sub = ap.add_subparsers(dest="command", required=True)
p = sub.add_parser("image", help="图片 OCR")
p.add_argument("--input", required=True)
p.add_argument("--output", default=None)
p.set_defaults(func=cmd_image)
p = sub.add_parser("pdf", help="扫描版 PDF OCR（无文本层页面）")
p.add_argument("--input", required=True)
p.add_argument("--output", default=None)
p.add_argument("--force", action="store_true", help="忽略文本层，强制 OCR 全部页面")
p.add_argument("--zoom", type=float, default=2.0, help="PDF 渲染倍率，越大越清晰越慢")
p.set_defaults(func=cmd_pdf)
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()#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""PDF 转 PPTX：按页提取 PDF 文本，每页生成一张 PPT（标题+要点）。
用法示例：
python pdf2ppt.py --input 论文.pdf --output 论文.pptx [--max-bullets 6] [--password xxx]
依赖：pip install pypdf python-pptx
说明：纯文本转换，确定性输出；扫描版 PDF（无文本层）请先用 ocr.py 识别出文本再转换。
"""
import argparse
import json
import re
import sys
from pathlib import Path
try:
from pypdf import PdfReader
from pptx import Presentation
from pptx.util import Inches, Pt
except ImportError: # pragma: no cover
sys.stderr.write("缺少依赖 pypdf/python-pptx，请先执行: pip install pypdf python-pptx\n")
sys.exit(2)
def _clean(text: str) -> list[str]:
"""按行清理，返回非空行。"""
lines = []
for line in text.splitlines():
line = line.strip()
if line:
lines.append(line)
return lines
def _pick_title(lines: list[str]) -> tuple[str, list[str]]:
"""取第一行作为标题（截断 40 字），其余为要点。"""
if not lines:
return "", []
title = lines[0]
if len(title) > 40:
title = title[:40] + "…"
return title, lines[1:]
def cmd_convert(args) -> dict:
reader = PdfReader(args.input)
if reader.is_encrypted:
if not args.password:
raise ValueError("PDF 已加密，请提供 --password（仅限您有权处理的文档）")
if not reader.decrypt(args.password):
raise ValueError("密码错误")
prs = Presentation()
max_bullets = args.max_bullets
slide_total = 0
for i, page in enumerate(reader.pages, 1):
raw = page.extract_text() or ""
lines = _clean(raw)
if not lines:
continue # 无文本页跳过
title, rest = _pick_title(lines)
slide = prs.slides.add_slide(prs.slide_layouts[1])
if title:
slide.shapes.title.text = f"第 {i} 页 · {title}"
else:
slide.shapes.title.text = f"第 {i} 页"
body = None
for ph in slide.placeholders:
if ph.placeholder_format.idx == 1:
body = ph
break
bullets = rest[:max_bullets]
if body is not None:
tf = body.text_frame
tf.clear()
for j, b in enumerate(bullets):
para = tf.paragraphs[0] if j == 0 else tf.add_paragraph()
para.text = b
elif bullets:
tb = slide.shapes.add_textbox(Inches(0.6), Inches(1.6), Inches(9), Inches(5))
tf = tb.text_frame
for j, b in enumerate(bullets):
para = tf.paragraphs[0] if j == 0 else tf.add_paragraph()
para.text = b
slide_total += 1
if slide_total == 0:
raise ValueError("PDF 无可提取文本（可能是扫描件）。请先 OCR 识别出文本再转换。")
out = Path(args.output)
out.parent.mkdir(parents=True, exist_ok=True)
prs.save(str(out))
return {
"ok": True,
"output": str(out),
"pdf_pages": len(reader.pages),
"ppt_slides": slide_total,
"bytes": out.stat().st_size,
"note": "每页提取前 max_bullets 个要点；扫描版请先 OCR",
}
def build_parser() -> argparse.ArgumentParser:
ap = argparse.ArgumentParser(description="PDF 转 PPTX（文本转换）")
ap.add_argument("--input", required=True, help="源 PDF")
ap.add_argument("--output", required=True, help="输出 PPTX")
ap.add_argument("--max-bullets", type=int, default=6, help="每页最多要点数，默认 6")
ap.add_argument("--password", default=None, help="加密 PDF 的密码（仅限本人文档）")
return ap
def main():
args = build_parser().parse_args()
try:
result = cmd_convert(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()#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""PDF 表单：查看字段 / 填写表单字段 / 导出已填写副本。
用法示例：
python pdf_form.py fields --input 表单.pdf
python pdf_form.py fill --input 表单.pdf --data '{"name": "张三", "age": "30"}' --output 已填.pdf
依赖：pip install pypdf
说明：填写后保留原始外观；若需锁定不可编辑，需另加 flatten（LibreOffice 或 pdftk）。
"""
import argparse
import json
import sys
from pathlib import Path
try:
from pypdf import PdfReader, PdfWriter
from pypdf.generic import NameObject, TextStringObject, BooleanObject
except ImportError: # pragma: no cover
sys.stderr.write("缺少依赖 pypdf，请先执行: pip install pypdf\n")
sys.exit(2)
def cmd_fields(args) -> dict:
reader = PdfReader(args.input)
if reader.is_encrypted and args.password:
reader.decrypt(args.password)
fields = reader.get_fields() or {}
out = []
for name, f in fields.items():
ft = f.get("/FT", "unknown")
ft_map = {"/Tx": "文本", "/Btn": "按钮/复选", "/Ch": "下拉/列表", "/Sig": "签名"}
out.append({
"name": name,
"type": ft_map.get(str(ft), str(ft)),
"value": f.get("/V", ""),
"default": f.get("/DV", ""),
})
return {"ok": True, "file": args.input, "fields": out, "count": len(out)}
def cmd_fill(args) -> dict:
reader = PdfReader(args.input)
if reader.is_encrypted:
if not args.password:
raise ValueError("PDF 已加密，请提供 --password（仅限本人有权处理的文档）")
if not reader.decrypt(args.password):
raise ValueError("密码错误")
data = json.loads(args.data)
if not isinstance(data, dict):
raise ValueError("--data 应为 JSON 对象，如 {\"name\": \"张三\"}")
# 校验字段是否存在，避免静默漏填
fields = reader.get_fields() or {}
unknown = [k for k in data if k not in fields]
if unknown:
raise ValueError(f"以下字段不存在于表单: {unknown}；可用 fields 命令查看全部字段")
writer = PdfWriter()
writer.append(reader)
# 直接写字段 /V 值（兼容 reportlab / 各类生成器），并开启 NeedAppearances
acro = writer.root_object.get("/AcroForm")
if acro is None:
raise ValueError("该 PDF 没有表单（/AcroForm 不存在）；请用 fields 命令确认")
acro_fields = acro.get_object().get("/Fields") or []
if not acro_fields:
raise ValueError("该 PDF 表单为空（无字段）")
matched = 0
for field_ref in acro_fields:
field = field_ref.get_object()
name = field.get("/T")
if name and name in data:
field[NameObject("/V")] = TextStringObject(str(data[name]))
matched += 1
if matched == 0:
raise ValueError(f"字段名与数据不匹配，已匹配 0 个；可用 fields 命令查看字段名")
acro.get_object()[NameObject("/NeedAppearances")] = BooleanObject(True)
out = Path(args.output)
out.parent.mkdir(parents=True, exist_ok=True)
with out.open("wb") as f:
writer.write(f)
return {
"ok": True,
"output": str(out),
"filled_fields": matched,
"note": "已生成填写副本，原文件未改动；部分阅读器打开时自动刷新显示",
}
def build_parser() -> argparse.ArgumentParser:
ap = argparse.ArgumentParser(description="PDF 表单工具")
ap.add_argument("--password", default=None)
sub = ap.add_subparsers(dest="command", required=True)
p = sub.add_parser("fields", help="列出表单字段")
p.add_argument("--input", required=True)
p.set_defaults(func=cmd_fields)
p = sub.add_parser("fill", help="填写表单字段")
p.add_argument("--input", required=True)
p.add_argument("--data", required=True, help='JSON 对象 {"字段名": "值"}')
p.add_argument("--output", required=True)
p.set_defaults(func=cmd_fill)
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()#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""PDF 处理脚本：合并 / 拆分 / 文本提取 / 压缩 / 加水印 / 解密解锁。
用法示例：
python pdf_ops.py merge --inputs a.pdf b.pdf --output out.pdf
python pdf_ops.py split --input a.pdf --pages 1-3,5 --output-dir out/
python pdf_ops.py extract --input a.pdf --output a.txt
python pdf_ops.py compress --input a.pdf --output a_small.pdf
python pdf_ops.py watermark --input a.pdf --output w.pdf --text "机密"
python pdf_ops.py decrypt --input locked.pdf --output free.pdf --password xxx
python pdf_ops.py info --input a.pdf
依赖：pip install pypdf reportlab
合规：解密仅限处理用户本人拥有或有权限处理的文档，禁止破解他人文件。
"""
import argparse
import json
import sys
from pathlib import Path
try:
from pypdf import PdfReader, PdfWriter
except ImportError: # pragma: no cover
sys.stderr.write("缺少依赖 pypdf，请先执行: pip install pypdf\n")
sys.exit(2)
# ---------- 工具 ----------
def _open_reader(path: str, password: str | None = None) -> PdfReader:
reader = PdfReader(path)
if reader.is_encrypted:
if not password:
raise ValueError(
f"文件已加密，需提供 --password（仅限您本人有权处理的文档）"
)
if not reader.decrypt(password):
raise ValueError("密码错误，解密失败")
return reader
def _write(writer: PdfWriter, output: str) -> dict:
out = Path(output)
out.parent.mkdir(parents=True, exist_ok=True)
with out.open("wb") as f:
writer.write(f)
return {"ok": True, "output": str(out), "size_bytes": out.stat().st_size}
# ---------- 命令 ----------
def cmd_info(args) -> dict:
reader = _open_reader(args.input, args.password)
pages = len(reader.pages)
meta = {}
if reader.metadata:
for k in ("title", "author", "subject", "creator"):
v = getattr(reader.metadata, k, None)
if v:
meta[k] = str(v)
return {
"ok": True,
"file": args.input,
"pages": pages,
"encrypted": reader.is_encrypted,
"metadata": meta,
}
def cmd_merge(args) -> dict:
writer = PdfWriter()
total = 0
for path in args.inputs:
reader = _open_reader(path, args.password)
for page in reader.pages:
writer.add_page(page)
total += 1
writer.append(reader) # 保留书签等结构
writer.add_metadata({"/Creator": "office-toolkit pdf_ops"})
return _write(writer, args.output) | {"pages": total}
def _parse_page_ranges(spec: str, total: int) -> list[int]:
"""'1-3,5,8-9' -> [1,2,3,5,8,9]（1 基）"""
pages: list[int] = []
for part in spec.split(","):
part = part.strip()
if not part:
continue
if "-" in part:
a, b = part.split("-", 1)
lo, hi = int(a), int(b)
else:
lo = hi = int(part)
if lo < 1 or hi > total or lo > hi:
raise ValueError(f"页码范围越界: {part}（共 {total} 页）")
pages.extend(range(lo, hi + 1))
return sorted(set(pages))
def cmd_split(args) -> dict:
reader = _open_reader(args.input, args.password)
total = len(reader.pages)
pages = _parse_page_ranges(args.pages, total)
out_dir = Path(args.output_dir)
out_dir.mkdir(parents=True, exist_ok=True)
written = []
for num in pages:
writer = PdfWriter()
writer.add_page(reader.pages[num - 1])
name = f"{Path(args.input).stem}_p{num:03d}.pdf"
out = out_dir / name
with out.open("wb") as f:
writer.write(f)
written.append(str(out))
return {"ok": True, "output_dir": str(out_dir), "files": written, "count": len(written)}
def cmd_extract(args) -> dict:
reader = _open_reader(args.input, args.password)
start = args.page_start or 1
end = args.page_end or len(reader.pages)
parts = []
chars = 0
for i in range(start - 1, min(end, len(reader.pages))):
text = reader.pages[i].extract_text() or ""
parts.append(f"---- 第 {i + 1} 页 ----\n{text}")
chars += len(text)
body = "\n\n".join(parts)
if args.output:
Path(args.output).write_text(body, encoding="utf-8")
return {"ok": True, "output": args.output, "pages_extracted": end - start + 1, "chars": chars, "preview": body[:300]}
def cmd_compress(args) -> dict:
"""压缩：去掉重复对象 + 重构流。图片级压缩需额外依赖（pikepdf），缺省时提示。"""
reader = _open_reader(args.input, args.password)
writer = PdfWriter()
writer.append(reader)
writer.compress_identical_objects(
remove_duplicates=True, remove_unreferenced=True
)
out = Path(args.output)
out.parent.mkdir(parents=True, exist_ok=True)
with out.open("wb") as f:
writer.write(f)
src_size = Path(args.input).stat().st_size
dst_size = out.stat().st_size
return {
"ok": True,
"output": str(out),
"before_bytes": src_size,
"after_bytes": dst_size,
"saved_pct": round((1 - dst_size / src_size) * 100, 1) if src_size else 0,
"note": "文本型 PDF 压缩有限；含大量图片的 PDF 建议后续接 pikepdf 图片重压缩",
}
def cmd_watermark(args) -> dict:
from reportlab.pdfgen import canvas as rl_canvas
from reportlab.lib.colors import Color
reader = _open_reader(args.input, args.password)
page_w = float(reader.pages[0].mediabox.width)
page_h = float(reader.pages[0].mediabox.height)
# 生成透明水印页
wm_path = Path(args.output).with_suffix(".wm_tmp.pdf")
c = rl_canvas.Canvas(str(wm_path), pagesize=(page_w, page_h))
c.setFont("Helvetica", args.font_size)
c.setFillColor(Color(0.5, 0.5, 0.5, alpha=args.opacity))
c.saveState()
c.translate(page_w / 2, page_h / 2)
c.rotate(args.angle)
c.drawCentredString(0, 0, args.text)
c.restoreState()
c.showPage()
c.save()
wm_reader = PdfReader(str(wm_path))
wm_page = wm_reader.pages[0]
writer = PdfWriter()
for page in reader.pages:
page.merge_page(wm_page)
writer.add_page(page)
result = _write(writer, args.output)
wm_path.unlink(missing_ok=True)
return result | {"watermark_text": args.text}
def cmd_encrypt(args) -> dict:
"""加密：设置打开密码（owner 密码同值）。仅建议用于保护本人文档。"""
reader = _open_reader(args.input, args.password_arg)
writer = PdfWriter()
for page in reader.pages:
writer.add_page(page)
writer.encrypt(user_password=args.password, owner_password=args.owner or args.password,
use_128bit=True, permissions_flag=args.permissions)
return _write(writer, args.output) | {"password_set": True, "note": "请牢记密码，丢失无法找回"}
def cmd_decrypt(args) -> dict:
"""解密解锁：去掉打开密码；无密码且仅带权限限制时，直接重建输出为无限制副本。
仅限用户本人拥有或有明确权限处理的文档。
"""
reader = _open_reader(args.input, args.password)
writer = PdfWriter()
writer.append(reader)
writer.encrypt(user_password="", owner_password="", use_128bit=True)
# 上面的 encrypt 会重设权限；若要完全无加密副本：
# 重新以明文写出（PdfWriter 默认不加密即可）
out = Path(args.output)
out.parent.mkdir(parents=True, exist_ok=True)
writer_plain = PdfWriter()
for page in reader.pages:
writer_plain.add_page(page)
with out.open("wb") as f:
writer_plain.write(f)
return {
"ok": True,
"output": str(out),
"was_encrypted": reader.is_encrypted,
"note": "已生成无打开密码、无权限限制的副本；请确认您对该文档拥有处理权限",
}
# ---------- CLI ----------
def build_parser() -> argparse.ArgumentParser:
ap = argparse.ArgumentParser(description="PDF 处理工具")
sub = ap.add_subparsers(dest="command", required=True)
p = sub.add_parser("info", help="查看页数/元数据/加密状态")
p.add_argument("--input", required=True)
p.add_argument("--password", default=None)
p.set_defaults(func=cmd_info)
p = sub.add_parser("merge", help="合并多个 PDF")
p.add_argument("--inputs", nargs="+", required=True)
p.add_argument("--output", required=True)
p.add_argument("--password", default=None)
p.set_defaults(func=cmd_merge)
p = sub.add_parser("split", help="按页码拆分")
p.add_argument("--input", required=True)
p.add_argument("--pages", required=True, help="如 1-3,5,8-9")
p.add_argument("--output-dir", required=True)
p.add_argument("--password", default=None)
p.set_defaults(func=cmd_split)
p = sub.add_parser("extract", help="提取文本")
p.add_argument("--input", required=True)
p.add_argument("--output", default=None)
p.add_argument("--page-start", type=int, default=None)
p.add_argument("--page-end", type=int, default=None)
p.add_argument("--password", default=None)
p.set_defaults(func=cmd_extract)
p = sub.add_parser("compress", help="压缩（去重复对象）")
p.add_argument("--input", required=True)
p.add_argument("--output", required=True)
p.add_argument("--password", default=None)
p.set_defaults(func=cmd_compress)
p = sub.add_parser("watermark", help="加水印")
p.add_argument("--input", required=True)
p.add_argument("--output", required=True)
p.add_argument("--text", required=True)
p.add_argument("--font-size", type=int, default=48)
p.add_argument("--opacity", type=float, default=0.25)
p.add_argument("--angle", type=int, default=45)
p.add_argument("--password", default=None)
p.set_defaults(func=cmd_watermark)
p = sub.add_parser("encrypt", help="加密（设置打开密码）")
p.add_argument("--input", required=True)
p.add_argument("--output", required=True)
p.add_argument("--password", required=True, help="打开密码")
p.add_argument("--owner", default=None, help="owner 密码（默认同 password）")
p.add_argument("--permissions", type=int, default=-44, help="权限标志位，默认禁止打印/复制")
p.add_argument("--password-arg", default=None) # 源文件本身加密时用
p.set_defaults(func=cmd_encrypt)
p = sub.add_parser("decrypt", help="解密解锁（仅限本人有权处理的文档）")
p.add_argument("--input", required=True)
p.add_argument("--output", required=True)
p.add_argument("--password", default=None)
p.set_defaults(func=cmd_decrypt)
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()#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""PPTX 处理脚本：从大纲创建 / 文本提取 / 大纲导出。
用法示例：
python pptx_ops.py create --title "产品发布会" --outline out.json --output deck.pptx
python pptx_ops.py read --input deck.pptx
python pptx_ops.py outline --input deck.pptx --output outline.json
--outline 为 JSON：{"slides": [{"title": "...", "bullets": ["a", "b"], "notes": "..."}]}
依赖：pip install python-pptx
"""
import argparse
import json
import sys
from pathlib import Path
try:
from pptx import Presentation
from pptx.util import Inches, Pt
except ImportError: # pragma: no cover
sys.stderr.write("缺少依赖 python-pptx，请先执行: pip install python-pptx\n")
sys.exit(2)
def _layout_for(prs, title: str):
"""选择合适版式：含标题+内容的最通用版式。"""
if not title:
return prs.slide_layouts[6] # blank
for idx in (1, 5, 0): # Title+Content / Title Only / Title Slide
try:
return prs.slide_layouts[idx]
except IndexError:
continue
return prs.slide_layouts[0]
def cmd_create(args) -> dict:
spec = json.loads(Path(args.outline).read_text(encoding="utf-8"))
prs = Presentation()
slides = spec.get("slides", [])
if not slides:
raise ValueError("大纲缺少 slides 数组")
# 封面
if spec.get("cover"):
cover = prs.slides.add_slide(prs.slide_layouts[0])
cover.shapes.title.text = spec["cover"].get("title", spec.get("title", ""))
if spec["cover"].get("subtitle") and cover.placeholders[1].has_text_frame:
cover.placeholders[1].text = spec["cover"]["subtitle"]
for item in slides:
slide = prs.slides.add_slide(_layout_for(prs, item.get("title", "")))
if item.get("title"):
slide.shapes.title.text = item["title"]
bullets = item.get("bullets") or []
# 找正文占位符
body = None
for ph in slide.placeholders:
if ph.placeholder_format.idx == 1:
body = ph
break
if body is not None:
tf = body.text_frame
tf.clear()
for i, b in enumerate(bullets):
para = tf.paragraphs[0] if i == 0 else tf.add_paragraph()
para.text = str(b)
para.level = 0
elif bullets:
# 无正文占位符 → 文本框追加
tb = slide.shapes.add_textbox(Inches(0.6), Inches(1.6), Inches(9), Inches(5))
tf = tb.text_frame
for i, b in enumerate(bullets):
para = tf.paragraphs[0] if i == 0 else tf.add_paragraph()
para.text = str(b)
if item.get("notes"):
slide.notes_slide.notes_text_frame.text = item["notes"]
out = Path(args.output)
out.parent.mkdir(parents=True, exist_ok=True)
prs.save(str(out))
return {"ok": True, "output": str(out), "slides": len(slides), "bytes": out.stat().st_size}
def cmd_read(args) -> dict:
prs = Presentation(args.input)
slides = []
for i, slide in enumerate(prs.slides, 1):
texts = []
for shape in slide.shapes:
if shape.has_text_frame:
t = shape.text_frame.text.strip()
if t:
texts.append(t)
slides.append({"page": i, "text": texts})
return {"ok": True, "file": args.input, "slides": len(slides), "content": slides}
def cmd_outline(args) -> dict:
prs = Presentation(args.input)
slides = []
for slide in prs.slides:
title = ""
bullets = []
for shape in slide.shapes:
if not shape.has_text_frame:
continue
txt = shape.text_frame.text
if not txt.strip():
continue
if shape == slide.shapes.title:
title = txt.strip()
else:
bullets.append(txt.strip())
slides.append({"title": title, "bullets": bullets})
data = {"source": args.input, "slides": slides}
if args.output:
Path(args.output).write_text(
json.dumps(data, ensure_ascii=False, indent=2), encoding="utf-8"
)
return {"ok": True, "slides": len(slides), "outline": data}
def build_parser() -> argparse.ArgumentParser:
ap = argparse.ArgumentParser(description="PPTX 处理工具")
sub = ap.add_subparsers(dest="command", required=True)
p = sub.add_parser("create", help="按大纲创建 PPTX")
p.add_argument("--title", default="")
p.add_argument("--outline", required=True, help="JSON 大纲文件")
p.add_argument("--output", required=True)
p.set_defaults(func=cmd_create)
p = sub.add_parser("read", help="提取每页文本")
p.add_argument("--input", required=True)
p.set_defaults(func=cmd_read)
p = sub.add_parser("outline", help="导出大纲 JSON")
p.add_argument("--input", required=True)
p.add_argument("--output", default=None)
p.set_defaults(func=cmd_outline)
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()#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""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 金额,数量
依赖：pip install openpyxl
"""
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)
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}
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)
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()#!/usr/bin/env python3
"""Excel (.xlsx) create / edit / read operations for the office-toolkit skill.
Adapted from OpenClaw Office Assistant (MIT, Copyright (c) 2026 Windrunner20);
extended with creation/editing + formula support. Local, deterministic, no network.
"""
import argparse
import json
import sys
from pathlib import Path
from openpyxl import Workbook, load_workbook
from openpyxl.utils import get_column_letter
def _load(path):
p = Path(path).expanduser().resolve()
if not p.exists():
raise FileNotFoundError(f"File not found: {p}")
return p
def _save(wb, out):
o = Path(out).expanduser().resolve()
o.parent.mkdir(parents=True, exist_ok=True)
wb.save(str(o))
return str(o)
def _parse_rows(spec, delimiter):
return [r.split(delimiter) for r in spec.split("|") if r.strip()]
def cmd_create(args):
wb = Workbook()
ws = wb.active
ws.title = args.sheet or "Sheet1"
out = _save(wb, args.output)
return {"ok": True, "file": out, "action": "create", "sheet": ws.title}
def cmd_write(args):
wb = load_workbook(str(_load(args.input))) if Path(args.input).exists() else Workbook()
ws = wb.active
if args.sheet:
ws = wb[args.sheet] if args.sheet in wb.sheetnames else wb.create_sheet(args.sheet)
rows = _parse_rows(args.rows, args.delimiter)
start = args.start_row
for i, row in enumerate(rows):
for j, val in enumerate(row):
cell = ws.cell(row=start + i, column=args.start_col + j, value=val)
if i == 0 and args.header_bold:
cell.font = cell.font.copy(bold=True)
out = _save(wb, args.output or args.input)
return {"ok": True, "file": out, "action": "write", "rows": len(rows)}
def cmd_formula(args):
wb = load_workbook(str(_load(args.input)))
ws = wb[args.sheet] if args.sheet in wb.sheetnames else wb.active
ws[args.cell] = args.expr
out = _save(wb, args.output or args.input)
return {"ok": True, "file": out, "action": "formula", "cell": args.cell, "expr": args.expr}
def cmd_read(args):
wb = load_workbook(str(_load(args.input)), data_only=True)
data = {}
for name in wb.sheetnames:
ws = wb[name]
max_row = min(ws.max_row or 0, args.row_limit)
max_col = ws.max_column or 0
rows = []
for r in ws.iter_rows(min_row=1, max_row=max_row, min_col=1, max_col=max_col, values_only=True):
vals = ["" if v is None else v for v in r]
if any(str(v) != "" for v in vals):
rows.append(vals)
data[name] = rows
return {"ok": True, "file": str(_load(args.input)), "sheets": wb.sheetnames, "data": data}
def build_parser():
ap = argparse.ArgumentParser(description="Excel (.xlsx) create/edit/read toolkit.")
sub = ap.add_subparsers(dest="cmd", required=True)
p = sub.add_parser("create", help="Create a new .xlsx")
p.add_argument("--output", required=True)
p.add_argument("--sheet", default="Sheet1")
p.set_defaults(func=cmd_create)
p = sub.add_parser("write", help="Write rows (a,b|c,d); first row can be bold header")
p.add_argument("--input", default="")
p.add_argument("--output", default="")
p.add_argument("--sheet", default="")
p.add_argument("--rows", required=True)
p.add_argument("--delimiter", default=",")
p.add_argument("--start-row", type=int, default=1)
p.add_argument("--start-col", type=int, default=1)
p.add_argument("--header-bold", action="store_true")
p.set_defaults(func=cmd_write)
p = sub.add_parser("formula", help="Set a formula in a cell")
p.add_argument("--input", required=True)
p.add_argument("--output", default="")
p.add_argument("--sheet", default="")
p.add_argument("--cell", required=True, help="e.g. C2")
p.add_argument("--expr", required=True, help="e.g. =A2+B2")
p.set_defaults(func=cmd_formula)
p = sub.add_parser("read", help="Read workbook to JSON")
p.add_argument("--input", required=True)
p.add_argument("--row-limit", type=int, default=200)
p.set_defaults(func=cmd_read)
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])