feat: pdf skill
This commit is contained in:
parent
885e7c401c
commit
ea5a187b2f
111
skills/pdf/SKILL.md
Normal file
111
skills/pdf/SKILL.md
Normal file
@ -0,0 +1,111 @@
|
||||
---
|
||||
name: pdf
|
||||
description: "处理本地 PDF 文件或远程 HTTPS PDF 链接，包括下载、读取、文本与表格提取、页面渲染与视觉审阅、创建、修改、合并、拆分、旋转、表单处理和最终质量校验。当用户提供 .pdf 文件或 HTTPS PDF 地址，或要求生成、编辑、总结、审阅版式重要的 PDF 时使用。"
|
||||
---
|
||||
|
||||
# PDF 处理
|
||||
|
||||
## 核心原则
|
||||
|
||||
- 优先进行视觉检查。PDF 的文本提取结果不能代表真实版式，必须把页面渲染为 PNG 后检查。
|
||||
- 收到远程 HTTPS PDF 链接时，先下载为本地 PDF，再执行任何读取、编辑或渲染操作。
|
||||
- 创建 PDF 时优先使用 `reportlab`；提取文本或表格时使用 `pdfplumber`；读取元数据、合并、拆分、旋转或处理页面结构时使用 `pypdf`。
|
||||
- 每次完成影响内容或版式的修改后，重新渲染并检查最新版本。
|
||||
- 不覆盖用户提供的源文件。编辑时写入新的输出文件。
|
||||
|
||||
## 环境约定
|
||||
|
||||
环境已预置 Poppler（`pdfinfo`、`pdftoppm`）以及 `reportlab`、`pdfplumber`、`pypdf`。直接使用环境提供的可执行文件和 Python 运行时；不要安装依赖，也不要提示用户安装依赖。如果默认 `python3` 未加载预置模块，先定位并改用环境提供的 Python 运行时。如果 Poppler 报告字体缓存目录不可写，把 `XDG_CACHE_HOME` 临时设置为本次任务目录下的可写缓存目录后重试。
|
||||
|
||||
## 标准流程
|
||||
|
||||
1. 获取输入：识别本地 PDF 路径或 HTTPS PDF 链接；远程链接先按“远程 PDF 下载”流程保存到本地。
|
||||
2. 校验文件：确认文件存在、大小大于 0、可被 PDF 工具解析，并用 `pdfinfo` 获取页数、页面尺寸、加密状态等信息。
|
||||
3. 建立工作目录：把中间文件放在 `tmp/pdfs/<任务名>/`，使用稳定且可读的文件名。
|
||||
4. 读取和处理：根据任务使用 `pdfplumber`、`pypdf` 或 `reportlab`，但不要仅依赖文本提取判断页面内容。
|
||||
5. 渲染页面：使用 Poppler 把待审阅的 PDF 转成 PNG。
|
||||
6. 视觉检查：逐页检查文字、图片、表格、页眉页脚、页码、分页和留白。
|
||||
7. 输出结果：把最终 PDF 写入 `output/pdf/`，文件名保持稳定且能表达内容。
|
||||
8. 清理中间文件：交付完成后仅删除本次任务对应的 `tmp/pdfs/<任务名>/`，保留最终产物。
|
||||
|
||||
## 远程 PDF 下载
|
||||
|
||||
只接受 HTTPS 地址。必须使用本 Skill 自带的 `scripts/download_pdf.py` 下载，不要使用 `curl`、shell 脚本或临时编写的下载逻辑。脚本会自动创建目标目录、流式下载、限制重定向协议、控制文件大小，并使用 `pypdf` 校验下载内容。
|
||||
|
||||
```text
|
||||
python3 scripts/download_pdf.py \
|
||||
--url 'https://example.com/document.pdf' \
|
||||
--output 'tmp/pdfs/<任务名>/source.pdf'
|
||||
```
|
||||
|
||||
保留完整 URL，包括查询参数，并把它作为一个完整参数传入。目标文件已存在时，脚本默认拒绝覆盖；仅在确认该文件是本次任务生成的下载缓存时使用 `--overwrite`。可选参数：
|
||||
|
||||
- `--timeout <秒>`：连接和读取超时，默认 `60`。
|
||||
- `--max-bytes <字节数>`：最大下载大小，默认 `104857600`（100 MiB）。
|
||||
- `--overwrite`：覆盖已存在的目标文件。
|
||||
|
||||
脚本成功时退出码为 `0`，并输出：
|
||||
|
||||
```json
|
||||
{
|
||||
"ok": true,
|
||||
"path": "/absolute/path/to/source.pdf",
|
||||
"size_bytes": 123456,
|
||||
"page_count": 10,
|
||||
"encrypted": false
|
||||
}
|
||||
```
|
||||
|
||||
脚本失败时退出码为非 `0`，并输出 `{"ok": false, "error": "具体错误"}`。只有收到 `ok: true` 后才能继续读取或渲染该文件。
|
||||
|
||||
遵守以下规则：
|
||||
|
||||
- 跟随 HTTPS 重定向，但不允许降级到 HTTP。
|
||||
- 不根据 URL 后缀或响应的 `Content-Type` 单独判断文件类型，以实际 PDF 解析结果为准。
|
||||
- 如果下载结果是 HTML、登录页、错误页或无法解析的内容，停止后续处理并明确说明链接需要授权或真实 PDF 下载地址。
|
||||
- 如果链接包含签名、令牌或其他敏感查询参数，不在最终回复、日志摘要、错误信息或生成的文件名中复述这些参数；下载脚本也不会在结果中回显 URL。
|
||||
- 如果 PDF 已加密且任务需要密码，先向用户索取密码；不得尝试绕过加密。
|
||||
|
||||
## 读取与审阅
|
||||
|
||||
- 使用 `pdfplumber` 提取正文、页级文本和表格，用于搜索、总结和结构化分析。
|
||||
- 使用 `pypdf` 读取元数据、书签、页数和页面对象，或执行合并、拆分、旋转等结构操作。
|
||||
- 如果提取出的文本很少或为空，将文件视为可能的扫描件，转而检查渲染后的页面；不要把空文本误判为空白 PDF。
|
||||
- 对表格、图表、公式、双栏排版、批注、印章和扫描内容，以渲染图为主要依据。
|
||||
- 回答问题或生成摘要时，区分 PDF 中明确写出的内容和根据版式、图表得出的推断。
|
||||
|
||||
## 创建与修改
|
||||
|
||||
- 使用 `reportlab` 创建新 PDF；复杂文档优先使用 Platypus 的文档流、段落、表格和分页组件。
|
||||
- 使用 `pypdf` 完成合并、拆分、旋转、页面重排、元数据和表单等操作。
|
||||
- 处理中文或其他非 ASCII 文字时，选择环境中可用且覆盖所需字符的 Unicode 字体，并嵌入 PDF；渲染后重点检查缺字、黑方块和字体回退。
|
||||
- 为正文、标题、表格、图注和页眉页脚建立一致的字号、行距、边距与层级。
|
||||
- 表格跨页时重复表头，避免行被不自然截断；图片和图表保持清晰、等比缩放并与说明文字对齐。
|
||||
- 只使用 ASCII 连字符 `-`，避免 U+2011 等 Unicode 横线字符引发字体或换行问题。
|
||||
|
||||
## 渲染与视觉检查
|
||||
|
||||
使用 `pdftoppm` 将 PDF 渲染到本次任务的工作目录：
|
||||
|
||||
```bash
|
||||
pdftoppm -png -r 150 \
|
||||
'output/pdf/<最终文件名>.pdf' \
|
||||
'tmp/pdfs/<任务名>/page'
|
||||
```
|
||||
|
||||
对于文字较小、图表密集或 150 DPI 下无法确认的页面，提高分辨率并重新渲染。创建或修改 PDF 时必须检查全部页面；只读超长文档时可先生成低分辨率预览定位相关页面，再以足够分辨率检查相关页。
|
||||
|
||||
逐页确认：
|
||||
|
||||
- 没有裁切、重叠、溢出、异常空白页或错误分页。
|
||||
- 没有黑方块、乱码、缺字或不可读的小字号。
|
||||
- 标题层级、段落间距、页边距和对齐方式一致。
|
||||
- 表格、图表和图片清晰、完整、标注正确。
|
||||
- 页眉、页脚、页码和章节衔接正确。
|
||||
- 引用和参考文献可读，不含工具令牌、占位符或临时路径。
|
||||
|
||||
## 交付要求
|
||||
|
||||
- 只有在最新渲染结果不存在可见的内容或格式缺陷后，才交付创建或修改后的 PDF。
|
||||
- 最终文件必须位于 `output/pdf/`，不得把临时 PNG 或下载缓存当作最终产物。
|
||||
- 回复时给出最终文件路径，并简要说明完成的处理和验证；失败时说明具体原因，不提供依赖安装提示。
|
||||
4
skills/pdf/agents/openai.yaml
Normal file
4
skills/pdf/agents/openai.yaml
Normal file
@ -0,0 +1,4 @@
|
||||
interface:
|
||||
display_name: "PDF 处理"
|
||||
short_description: "读取、创建、审阅与可视化检查本地或远程 PDF 文件"
|
||||
default_prompt: "使用 $pdf 下载并审阅这个 PDF，提取要点并检查页面布局。"
|
||||
248
skills/pdf/scripts/download_pdf.py
Normal file
248
skills/pdf/scripts/download_pdf.py
Normal file
@ -0,0 +1,248 @@
|
||||
#!/usr/bin/env python3
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import json
|
||||
import logging
|
||||
import os
|
||||
import socket
|
||||
import sys
|
||||
import tempfile
|
||||
import urllib.error
|
||||
import urllib.parse
|
||||
import urllib.request
|
||||
from pathlib import Path
|
||||
from typing import NoReturn, Optional
|
||||
|
||||
DEFAULT_TIMEOUT_SECONDS = 60
|
||||
DEFAULT_MAX_BYTES = 100 * 1024 * 1024
|
||||
CHUNK_SIZE = 1024 * 1024
|
||||
USER_AGENT = "wechat-robot-pdf-skill/1.0"
|
||||
|
||||
|
||||
class SkillArgumentParser(argparse.ArgumentParser):
|
||||
def error(self, message: str) -> NoReturn:
|
||||
raise ValueError(f"参数错误：{message}")
|
||||
|
||||
|
||||
def _emit(payload: dict) -> None:
|
||||
sys.stdout.write(json.dumps(payload, ensure_ascii=False) + "\n")
|
||||
|
||||
|
||||
def _validate_https_url(value: str) -> str:
|
||||
url = value.strip()
|
||||
if not url:
|
||||
raise ValueError("PDF URL 不能为空")
|
||||
if any(character.isspace() or ord(character) < 32 for character in url):
|
||||
raise ValueError("PDF URL 不能包含空白字符或控制字符")
|
||||
|
||||
parsed = urllib.parse.urlsplit(url)
|
||||
if parsed.scheme.lower() != "https" or not parsed.hostname:
|
||||
raise ValueError("PDF URL 必须是有效的 HTTPS 地址")
|
||||
if parsed.username is not None or parsed.password is not None:
|
||||
raise ValueError("PDF URL 不允许包含用户名或密码")
|
||||
try:
|
||||
parsed.port
|
||||
except ValueError as exc:
|
||||
raise ValueError("PDF URL 端口格式不正确") from exc
|
||||
return url
|
||||
|
||||
|
||||
class HTTPSOnlyRedirectHandler(urllib.request.HTTPRedirectHandler):
|
||||
def redirect_request(self, req, fp, code, msg, headers, newurl):
|
||||
safe_url = _validate_https_url(newurl)
|
||||
return super().redirect_request(req, fp, code, msg, headers, safe_url)
|
||||
|
||||
|
||||
def _parse_args(argv: list[str]) -> argparse.Namespace:
|
||||
parser = SkillArgumentParser(description="下载并校验远程 HTTPS PDF")
|
||||
parser.add_argument(
|
||||
"--url",
|
||||
"--pdf-url",
|
||||
"--pdf_url",
|
||||
dest="url",
|
||||
required=True,
|
||||
help="远程 HTTPS PDF 地址",
|
||||
)
|
||||
parser.add_argument(
|
||||
"--output",
|
||||
required=True,
|
||||
help="本地输出路径，必须以 .pdf 结尾；父目录不存在时会自动创建",
|
||||
)
|
||||
parser.add_argument(
|
||||
"--timeout",
|
||||
type=int,
|
||||
default=DEFAULT_TIMEOUT_SECONDS,
|
||||
help=f"连接和读取超时秒数，默认 {DEFAULT_TIMEOUT_SECONDS}",
|
||||
)
|
||||
parser.add_argument(
|
||||
"--max-bytes",
|
||||
"--max_bytes",
|
||||
dest="max_bytes",
|
||||
type=int,
|
||||
default=DEFAULT_MAX_BYTES,
|
||||
help=f"最大下载字节数，默认 {DEFAULT_MAX_BYTES}",
|
||||
)
|
||||
parser.add_argument(
|
||||
"--overwrite",
|
||||
action="store_true",
|
||||
help="允许覆盖已存在的目标文件",
|
||||
)
|
||||
args = parser.parse_args(argv)
|
||||
|
||||
args.url = _validate_https_url(args.url)
|
||||
if args.timeout <= 0:
|
||||
raise ValueError("timeout 必须大于 0")
|
||||
if args.max_bytes <= 0:
|
||||
raise ValueError("max-bytes 必须大于 0")
|
||||
|
||||
output = Path(args.output).expanduser()
|
||||
if output.suffix.lower() != ".pdf":
|
||||
raise ValueError("output 必须以 .pdf 结尾")
|
||||
args.output = output.resolve()
|
||||
return args
|
||||
|
||||
|
||||
def _validate_pdf(path: Path) -> tuple[Optional[int], bool]:
|
||||
with path.open("rb") as file:
|
||||
prefix = file.read(1024)
|
||||
if b"%PDF-" not in prefix:
|
||||
raise ValueError("下载内容不是 PDF 文件")
|
||||
|
||||
try:
|
||||
from pypdf import PdfReader
|
||||
except ImportError as exc:
|
||||
raise RuntimeError("当前 Python 未加载环境预置的 pypdf 模块") from exc
|
||||
|
||||
logging.getLogger("pypdf").setLevel(logging.ERROR)
|
||||
reader = None
|
||||
try:
|
||||
reader = PdfReader(str(path), strict=False)
|
||||
encrypted = bool(reader.is_encrypted)
|
||||
page_count = None if encrypted else len(reader.pages)
|
||||
except Exception as exc:
|
||||
raise ValueError("下载内容不是可解析的 PDF 文件") from exc
|
||||
finally:
|
||||
stream = getattr(reader, "stream", None)
|
||||
if stream is not None and hasattr(stream, "close"):
|
||||
stream.close()
|
||||
return page_count, encrypted
|
||||
|
||||
|
||||
def _download(args: argparse.Namespace) -> dict:
|
||||
output: Path = args.output
|
||||
if output.exists() and not args.overwrite:
|
||||
raise FileExistsError(f"目标文件已存在：{output}")
|
||||
|
||||
output.parent.mkdir(parents=True, exist_ok=True)
|
||||
request = urllib.request.Request(
|
||||
args.url,
|
||||
headers={
|
||||
"Accept": "application/pdf,application/octet-stream;q=0.9,*/*;q=0.1",
|
||||
"Accept-Encoding": "identity",
|
||||
"User-Agent": USER_AGENT,
|
||||
},
|
||||
method="GET",
|
||||
)
|
||||
opener = urllib.request.build_opener(HTTPSOnlyRedirectHandler())
|
||||
temp_path: Optional[Path] = None
|
||||
|
||||
try:
|
||||
with tempfile.NamedTemporaryFile(
|
||||
mode="wb",
|
||||
prefix=f".{output.name}.",
|
||||
suffix=".part",
|
||||
dir=str(output.parent),
|
||||
delete=False,
|
||||
) as temp_file:
|
||||
temp_path = Path(temp_file.name)
|
||||
with opener.open(request, timeout=args.timeout) as response:
|
||||
_validate_https_url(response.geturl())
|
||||
content_length = response.headers.get("Content-Length")
|
||||
if content_length:
|
||||
try:
|
||||
expected_bytes = int(content_length)
|
||||
except ValueError:
|
||||
expected_bytes = 0
|
||||
if expected_bytes > args.max_bytes:
|
||||
raise ValueError(
|
||||
f"远程文件超过大小限制：最多允许 {args.max_bytes} 字节"
|
||||
)
|
||||
|
||||
downloaded_bytes = 0
|
||||
while True:
|
||||
chunk = response.read(CHUNK_SIZE)
|
||||
if not chunk:
|
||||
break
|
||||
downloaded_bytes += len(chunk)
|
||||
if downloaded_bytes > args.max_bytes:
|
||||
raise ValueError(
|
||||
f"远程文件超过大小限制：最多允许 {args.max_bytes} 字节"
|
||||
)
|
||||
temp_file.write(chunk)
|
||||
|
||||
temp_file.flush()
|
||||
os.fsync(temp_file.fileno())
|
||||
|
||||
if downloaded_bytes == 0:
|
||||
raise ValueError("远程服务器返回了空文件")
|
||||
|
||||
page_count, encrypted = _validate_pdf(temp_path)
|
||||
if args.overwrite:
|
||||
os.replace(temp_path, output)
|
||||
else:
|
||||
try:
|
||||
os.link(temp_path, output)
|
||||
except FileExistsError as exc:
|
||||
raise FileExistsError(f"目标文件已存在：{output}") from exc
|
||||
temp_path.unlink()
|
||||
temp_path = None
|
||||
|
||||
return {
|
||||
"ok": True,
|
||||
"path": str(output),
|
||||
"size_bytes": downloaded_bytes,
|
||||
"page_count": page_count,
|
||||
"encrypted": encrypted,
|
||||
}
|
||||
finally:
|
||||
if temp_path is not None:
|
||||
try:
|
||||
temp_path.unlink(missing_ok=True)
|
||||
except OSError:
|
||||
pass
|
||||
|
||||
|
||||
def _failure_message(exc: Exception) -> str:
|
||||
if isinstance(exc, urllib.error.HTTPError):
|
||||
return f"下载失败：远程服务器返回 HTTP {exc.code}"
|
||||
if isinstance(exc, (TimeoutError, socket.timeout)):
|
||||
return "下载失败：连接或读取超时"
|
||||
if isinstance(exc, urllib.error.URLError):
|
||||
if isinstance(exc.reason, (TimeoutError, socket.timeout)):
|
||||
return "下载失败：连接或读取超时"
|
||||
return "下载失败：无法访问远程服务器"
|
||||
if isinstance(exc, FileExistsError):
|
||||
return str(exc)
|
||||
if isinstance(exc, PermissionError):
|
||||
return "写入失败：没有目标路径的写入权限"
|
||||
if isinstance(exc, OSError):
|
||||
return f"文件处理失败：{exc}"
|
||||
return str(exc)
|
||||
|
||||
|
||||
def main(argv: Optional[list[str]] = None) -> int:
|
||||
try:
|
||||
args = _parse_args(sys.argv[1:] if argv is None else argv)
|
||||
result = _download(args)
|
||||
except Exception as exc:
|
||||
_emit({"ok": False, "error": _failure_message(exc)})
|
||||
return 1
|
||||
|
||||
_emit(result)
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main())
|
||||
Loading…
Reference in New Issue
Block a user